postgresql-simpleを使用したスト​​リーミングコンジットソースの作成

Question

postgresql-simpleは、ストリーミングクエリの機能を提供します。

fold 
    :: (FromRow row, ToRow params) 
    => Connection -> Query -> params -> a -> (a -> row -> IO a) -> IO a 

ストリーミングを最大限活用するコンジットソースを作成したいと考えています。 IOはfoldに（と思う？）反変な位置に表示されますので、

mySource :: (FromRow row, Monad m) => Source m row 

残念ながら、私は本当に種類に苦しんでいます。次の型チェックでは、値を生成する前にストリーム全体を折りたたみます。

getConduit :: Connection -> IO (C.ConduitM() Event IO()) 
getConduit conn = fold_ conn queryEventRecord CL.sourceNull foo 
    where 
    foo :: C.ConduitM() Event IO() -> Event -> IO (C.ConduitM() Event IO()) 
    foo cond evt = pure (cond >> C.yield evt) 

これを実装する方法についてのあらゆる指針は非常に高く評価されます。ありがとう！最終的にするためにstm-conduitを使用

Answer 1

これについて移動するための一つの（とても素敵ではない）方法でそれ

に

• は
• はちょうどこのチャンネル
• に行をダンプするforeach_を設定した行を受け取るために新しいTMChanを作りますチャネル外発信元

私はこのオフハンドをテストする手段はありませんが、次のように動作するはずです

import Conduit 
import Database.PostgreSQL.Simple (foreach_) 
import Data.Conduit.TMChan (sourceTMChan) 
import Control.Concurrent.STM.TMChan (newTMChanIO, writeTMChan, atomically) 

mySource :: (FromRow row, MonadIO m) => Connection -> Query -> IO (Source m row) 
mySource connection query = do 
    chan <- newTMChanIO 
    forEach_ connection query (atomically . writeTMChan chan) 
    pure (sourceTMChan chan) 

だけ私たちは、これがより簡単かもしれないforEach_ :: (MonadIO m, FromRow r) => Connection -> Query -> (r -> m()) -> m()を持っていた場合...

Answer 2

ここではそのコンパイルし、実行します上記のアレックさんの変更があります。 mkPgSourceは、彼の記事の最後にアレックが言及した一般的な機能です。永続-PostgreSQLの

import Database.PostgreSQL.Simple 
import Database.PostgreSQL.Simple.FromRow 
import Database.PostgreSQL.Simple.ToRow 
import Control.Monad.IO.Class (MonadIO) 
import Data.Conduit.TMChan (sourceTMChan) 
import Control.Concurrent.STM.TMChan (newTMChanIO, writeTMChan, 
closeTMChan, TMChan) 
import GHC.Conc (atomically, forkIO) 
import Conduit 

--closes the channel after action is done to terminate the source 
mkPgSource :: (MonadIO m, FromRow r) => ((r -> IO()) -> IO()) -> IO (Source m r) 
mkPgSource action = do 
    chan <- newTMChanIO 
    _ <- forkIO $ do action $ atomically . (writeTMChan chan) 
       atomically $ closeTMChan chan 
    pure $ sourceTMChan chan 

sourceQuery :: (ToRow params, FromRow r, MonadIO m) => 
    Connection -> Query -> params -> IO (Source m r) 
sourceQuery conn q params = mkPgSource $ forEach conn q params 

sourceQuery_ :: (FromRow r, MonadIO m) => Connection -> Query -> IO 
(Source m r) 
sourceQuery_ conn q = mkPgSource $ forEach_ conn q