Создание источника потоковой передачи с помощью 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 () это может быть проще...

Answer 2

Вот модификация Алека выше, которая компилируется и запускается. mkPgSource это общая функция, на которую Алек ссылался в конце своего поста.

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