Sử dụng liên tục từ bên trong một Conduit

Question

Đầu tiên, một phiên bản đơn giản của tác vụ tôi muốn thực hiện: Tôi có một số tệp lớn (lên đến 30 GB) mà tôi muốn cắt tỉa cho các mục trùng lặp. Để kết thúc này, tôi thiết lập một cơ sở dữ liệu của băm dữ liệu và mở từng tệp một, băm nhỏ từng mục và ghi nó vào cơ sở dữ liệu và tệp đầu ra iff băm của nó chưa có trong cơ sở dữ liệu.

Tôi biết cách thực hiện điều này với lặp lại, điều tra viên và tôi muốn thử ống dẫn. Tôi cũng biết làm thế nào để làm điều đó với ống dẫn, nhưng bây giờ tôi muốn sử dụng ống dẫn & dai dẳng. Tôi đang gặp sự cố với các loại và có thể với toàn bộ khái niệm là ResourceT.

Dưới đây là một số mã giả để minh họa cho vấn đề:

withSqlConn "foo.db" $ runSqlConn $ runResourceT $ 
    sourceFile "in" $= parseBytes $= dbAction $= serialize $$ sinkFile "out" 

Vấn đề nằm ở chỗ dbAction chức năng. Tôi muốn truy cập vào cơ sở dữ liệu ở đây, một cách tự nhiên. Kể từ khi hoạt động nó làm là về cơ bản chỉ là một bộ lọc, đầu tiên tôi nghĩ để viết nó như thế:

dbAction = CL.mapMaybeM p 
    where p :: (MonadIO m, MonadBaseControl IO (SqlPersist m)) => DataType -> m (Maybe DataType) 
      p = lift $ putStrLn "foo" -- fine 
      insert $ undefined -- type error! 
      return undefined 

Các lỗi cụ thể tôi nhận được là:

Could not deduce (m ~ b0 m0) 
from the context (MonadIO m, MonadBaseControl IO (SqlPersist m)) 
    bound by the type signature for 
      p :: (MonadIO m, MonadBaseControl IO (SqlPersist m)) => 
          DataType -> m (Maybe DataType) 
    at tools/clean-wac.hs:(33,1)-(34,34) 
    `m' is a rigid type variable bound by 
     the type signature for 
     p :: (MonadIO m, MonadBaseControl IO (SqlPersist m)) => 
         DataType -> m (Maybe (DataType)) 
     at tools/clean-wac.hs:33:1 
Expected type: m (Key b0 val0) 
    Actual type: b0 m0 (Key b0 val0) 

Lưu ý rằng điều này có thể là do những giả định sai Tôi đã thiết kế chữ ký kiểu. Nếu tôi nhận xét ra chữ ký kiểu và cũng loại bỏ các tuyên bố lift, thông báo lỗi biến thành:

No instance for (PersistStore ResourceT (SqlPersist IO)) 
    arising from a use of `p' 
Possible fix: 
    add an instance declaration for 
    (PersistStore ResourceT (SqlPersist IO)) 
In the first argument of `CL.mapMaybeM', namely `p' 

Vì vậy, điều này có nghĩa rằng chúng ta không thể truy cập vào PersistStore ở tất cả qua ResourceT?

tôi không thể viết Conduit của riêng tôi, hoặc, mà không sử dụng CL.mapMaybeM:

dbAction = filterP 
filterP :: (MonadIO m, MonadBaseControl IO (SqlPersist m)) => Conduit DataType m DataType 
filterP = loop 
    where loop = awaitE >>= either return go 
      go s = do lift $ insert $ undefined -- again, type error 
        loop 

Điều này dẫn đến thêm một lỗi loại I không hoàn toàn hiểu.

Could not deduce (m ~ b0 m0) 
from the context (MonadIO m, MonadBaseControl IO (SqlPersist m)) 
    bound by the type signature for 
      filterP :: (MonadIO m, 
           MonadBaseControl IO (SqlPersist m)) => 
           Conduit DataType m DataType 
    `m' is a rigid type variable bound by 
     the type signature for 
     filterP :: (MonadIO m, 
          MonadBaseControl IO (SqlPersist m)) => 
          Conduit DataType m DataType 
Expected type: Conduit DataType m DataType 
    Actual type: Pipe 
       DataType DataType DataType() (b0 m0)() 
In the expression: loop 
In an equation for `filterP' 

Vì vậy, câu hỏi của tôi là: có thể sử dụng liên tục như tôi định dùng trong ống dẫn không? Và nếu, làm thế nào? Tôi biết rằng vì tôi có thể sử dụng liftIO bên trong ống dẫn, tôi chỉ có thể sử dụng và nói, HDBC, nhưng tôi muốn sử dụng liên tục một cách rõ ràng để hiểu cách hoạt động và vì tôi thích chủ nghĩa bất khả tri của chương trình phụ trợ db.

Answer 1

Mã bên dưới biên dịch tốt cho tôi. Có thể các khung công tác đã chuyển sang thời gian chờ đợi và mọi thứ bây giờ chỉ hoạt động?

Tuy nhiên, lưu ý những thay đổi sau tôi phải thực hiện khi thế giới đã thay đổi một chút hoặc tôi không có tất cả mã của bạn. Tôi đã sử dụng ống dẫn-1.0.9.3 và liên tục-1.3.0 với GHC 7.6.3.

• Bỏ qua parseBytes và serialise như tôi không có định nghĩa của bạn và xác định DataType = ByteString để thay thế.

• Giới thiệu thông số Proxy và chữ ký loại rõ ràng cho giá trị undefined để tránh các vấn đề với loại tiêm gia đình. Những khả năng này không xuất hiện trong mã thực của bạn bởi vì nó sẽ có loại được xác định cụ thể hoặc bên ngoài cho val.

• Dùng await hơn awaitE và chỉ được sử dụng () như các loại để thay thế đối với trường hợp Left, như awaitE đã được nghỉ hưu.

• Đã chuyển một giả Connection chức năng tạo thành withSqlConn - có lẽ tôi đã sử dụng một số chức năng cụ thể của Sqlite?

Dưới đây là các mã:

{-# LANGUAGE FlexibleContexts, NoMonomorphismRestriction, 
      TypeFamilies, ScopedTypeVariables #-} 

module So133331988 where 

import Control.Monad.Trans 
import Database.Persist.Sql 
import Data.ByteString 
import Data.Conduit 
import Data.Conduit.Binary 
import Data.Proxy 

test proxy = 
    withSqlConn (return (undefined "foo.db")) $ runSqlConn $ runResourceT $ 
     sourceFile "in" $= dbAction proxy $$ sinkFile "out" 

dbAction = filterP 

type DataType = ByteString 

filterP 
    :: forall m val 
    . (MonadIO m, MonadBaseControl IO (SqlPersist m) 
     , PersistStore m, PersistEntity val 
     , PersistEntityBackend val ~ PersistMonadBackend m) 
    => Proxy val 
    -> Conduit DataType m DataType 
filterP Proxy = loop 
    where loop = await >>= maybe (return()) go 
      go s = do lift $ insert (undefined :: val) 
        loop