使用postgresql-simple创建流媒体管道源

Question

postgresql-simple 提供用于流式查询的功能,例如

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

我想创建一个充分利用流媒体的管道源.

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

不幸的是,因为IO出现在一个逆变的位置(我认为？)fold,我真的很难与这些类型斗争.以下类型检查,但在产生值之前折叠整个流.

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)

关于如何实现这一点的任何指针将不胜感激!谢谢!

Answer 1

一个(不太好)的方式来解决这个问题

• 创建一个新TMChan的接收行
• 设置foreach_为只将行转储到此通道中
• 最后stm-conduit用来制作一个来自频道的资源

我没有办法对此进行测试,但以下情况应该有效

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 ()这样做可能会更容易......