stm-conduit 2.7.0 → 4.0.1
raw patch · 8 files changed
Files
- Data/Conduit/Async.hs +11/−17
- Data/Conduit/Async/Composition.hs +46/−56
- Data/Conduit/TMChan.hs +111/−60
- Data/Conduit/TQueue.hs +20/−26
- Data/Conduit/Utils.hs +23/−23
- stm-conduit.cabal +15/−19
- test/DocTests.hs +7/−7
- test/Test.hs +76/−37
Data/Conduit/Async.hs view
@@ -15,39 +15,33 @@ , drainTo ) where -#if __GLASGOW_HASKELL__ < 710-import Control.Applicative-#endif-import Control.Concurrent.Async.Lifted-import Control.Concurrent.STM-import Control.Exception.Lifted import Control.Monad.IO.Class import Control.Monad.Loops import Control.Monad.Trans.Class-import Control.Monad.Trans.Control import Data.Conduit- import Data.Conduit.Async.Composition+import Data.Foldable+import UnliftIO -- | Gather output values asynchronously from an action in the base monad and -- then yield them downstream. This provides a means of working around the -- restriction that 'ConduitM' cannot be an instance of 'MonadBaseControl' -- in order to, for example, yield values from within a Haskell callback -- function called from a C library.-gatherFrom :: (MonadIO m, MonadBaseControl IO m)+gatherFrom :: (MonadIO m, MonadUnliftIO m) => Int -- ^ Size of the queue to create -> (TBQueue o -> m ()) -- ^ Action that generates output values- -> Producer m o+ -> ConduitT () o m () gatherFrom size scatter = do- chan <- liftIO $ newTBQueueIO size+ chan <- liftIO $ newTBQueueIO (fromIntegral size) worker <- lift $ async (scatter chan)- lift . restoreM =<< gather worker chan+ gather worker chan where gather worker chan = do (xs, mres) <- liftIO $ atomically $ do xs <- whileM (not <$> isEmptyTBQueue chan) (readTBQueue chan) (xs,) <$> pollSTM worker- Prelude.mapM_ yield xs+ traverse_ yield xs case mres of Just (Left e) -> liftIO $ throwIO (e :: SomeException) Just (Right r) -> return r@@ -55,14 +49,14 @@ -- | Drain input values into an asynchronous action in the base monad via a -- bounded 'TBQueue'. This is effectively the dual of 'gatherFrom'.-drainTo :: (MonadIO m, MonadBaseControl IO m)+drainTo :: (MonadIO m, MonadUnliftIO m) => Int -- ^ Size of the queue to create -> (TBQueue (Maybe i) -> m r) -- ^ Action to consume input values- -> Consumer i m r+ -> ConduitT i Void m r drainTo size gather = do- chan <- liftIO $ newTBQueueIO size+ chan <- liftIO $ newTBQueueIO (fromIntegral size) worker <- lift $ async (gather chan)- lift . restoreM =<< scatter worker chan+ scatter worker chan where scatter worker chan = do mval <- await
Data/Conduit/Async/Composition.hs view
@@ -24,14 +24,8 @@ , runCConduit ) where -#if __GLASGOW_HASKELL__ < 710-import Control.Applicative-#endif import Conduit-import qualified "async" Control.Concurrent.Async as A-import Control.Concurrent.Async.Lifted hiding (link2)-import Control.Concurrent.STM-import Control.Exception (finally)+import Control.Concurrent.STM (orElse, check) import Control.Monad hiding (forM_) import Control.Monad.Loops import Control.Monad.Trans.Resource@@ -40,10 +34,10 @@ import qualified Data.Conduit.List as CL import Data.Foldable (forM_) import Data.Serialize-import Data.Void-import GHC.Prim+import GHC.Exts (Constraint) import System.Directory (removeFile)-import System.IO+import System.IO (openBinaryTempFile)+import UnliftIO -- | Concurrently join the producer and consumer, using a bounded queue of the -- given size. The producer will block when the queue is full, if it is@@ -158,7 +152,7 @@ -- -- >>> runResourceT $ bufferToFile 1 Nothing "/tmp" (CL.sourceList [1,2,3]) CL.consume -- [1,2,3]-bufferToFile :: (CFConduitLike c1, CFConduitLike c2, Serialize x, MonadBaseControl IO m, MonadIO m, MonadResource m)+bufferToFile :: (CFConduitLike c1, CFConduitLike c2, Serialize x, MonadUnliftIO m, MonadResource m, MonadThrow m) => Int -- ^ Size of the bounded queue in memory -> Maybe Int -- ^ Max elements to keep on disk at one time -> FilePath -- ^ Directory to write temp files to@@ -209,20 +203,20 @@ -- equivalent of 'runConduit'. -- -- The underlying monad must always be an instance of- -- 'MonadBaseControl IO'. If the conduits is a 'CFConduit', it must+ -- 'MonadUnliftIO'. If the conduits is a 'CFConduit', it must -- additionally be a in instance of 'MonadResource'. runCConduit :: (RunConstraints c m) => c () Void m r -> m r -instance CCatable ConduitM ConduitM CConduit where+instance CCatable ConduitT ConduitT CConduit where buffer' i a b = buffer' i (Single a) (Single b) -instance CCatable ConduitM CConduit CConduit where+instance CCatable ConduitT CConduit CConduit where buffer' i a b = buffer' i (Single a) b -instance CCatable ConduitM CFConduit CFConduit where+instance CCatable ConduitT CFConduit CFConduit where buffer' i a b = buffer' i (asCFConduit a) b -instance CCatable CConduit ConduitM CConduit where+instance CCatable CConduit ConduitT CConduit where buffer' i a b = buffer' i a (Single b) instance CCatable CConduit CConduit CConduit where@@ -232,7 +226,7 @@ instance CCatable CConduit CFConduit CFConduit where buffer' i a b = buffer' i (asCFConduit a) b -instance CCatable CFConduit ConduitM CFConduit where+instance CCatable CFConduit ConduitT CFConduit where buffer' i a b = buffer' i a (asCFConduit b) instance CCatable CFConduit CConduit CFConduit where@@ -243,27 +237,27 @@ buffer' i (FMultiple i' a as) b = FMultiple i' a (buffer' i as b) buffer' i (FMultipleF bufsz dsksz tmpDir a as) b = FMultipleF bufsz dsksz tmpDir a (buffer' i as b) -instance CRunnable ConduitM where- type RunConstraints ConduitM m = (Monad m)+instance CRunnable ConduitT where+ type RunConstraints ConduitT m = (MonadUnliftIO m, Monad m) runCConduit = runConduit instance CRunnable CConduit where- type RunConstraints CConduit m = (MonadBaseControl IO m, MonadIO m)+ type RunConstraints CConduit m = (MonadUnliftIO m, MonadIO m) runCConduit (Single c) = runConduit c runCConduit (Multiple bufsz c cs) = do- chan <- liftIO $ newTBQueueIO bufsz+ chan <- liftIO $ newTBQueueIO (fromIntegral bufsz) withAsync (sender chan c) $ \c' -> stage chan c' cs instance CRunnable CFConduit where- type RunConstraints CFConduit m = (MonadBaseControl IO m, MonadIO m, MonadResource m)+ type RunConstraints CFConduit m = (MonadThrow m, MonadUnliftIO m, MonadIO m, MonadResource m) runCConduit (FSingle c) = runConduit c runCConduit (FMultiple bufsz c cs) = do- chan <- liftIO $ newTBQueueIO bufsz+ chan <- liftIO $ newTBQueueIO (fromIntegral bufsz) withAsync (sender chan c) $ \c' -> fstage (receiver chan) c' cs runCConduit (FMultipleF bufsz filemax tempDir c cs) = do- context <- liftIO $ BufferContext <$> newTBQueueIO bufsz+ context <- liftIO $ BufferContext <$> newTBQueueIO (fromIntegral bufsz) <*> newTQueueIO <*> newTVarIO filemax <*> newTVarIO False@@ -274,40 +268,36 @@ -- | A "concurrent conduit", in which the stages run in parallel with -- a buffering queue between them. data CConduit i o m r where- Single :: ConduitM i o m r -> CConduit i o m r- Multiple :: Int -> ConduitM i x m () -> CConduit x o m r -> CConduit i o m r---- C.C.A.L's link2 has the wrong type: https://github.com/maoe/lifted-async/issues/16-link2 :: MonadBase IO m => Async a -> Async b -> m ()-link2 = (liftBase .) . A.link2+ Single :: ConduitT i o m r -> CConduit i o m r+ Multiple :: Int -> ConduitT i x m () -> CConduit x o m r -> CConduit i o m r -- Combines a producer with a queue, sending it everything the -- producer produces.-sender :: (MonadIO m) => TBQueue (Maybe o) -> ConduitM () o m () -> m ()+sender :: (MonadIO m) => TBQueue (Maybe o) -> ConduitT () o m () -> m () sender chan input = do- input $$ mapM_C (send chan . Just)+ runConduit $ input .| mapM_C (send chan . Just) send chan Nothing -- One "layer" of withAsync in a CConduit run.-stage :: (MonadBaseControl IO m, MonadIO m) => TBQueue (Maybe i) -> Async x -> CConduit i Void m r -> m r+stage :: (MonadUnliftIO m) => TBQueue (Maybe i) -> Async x -> CConduit i Void m r -> m r stage chan prevAsync (Single c) = -- The last layer; feed the output of "chan" into the conduit and -- wait for the result.- withAsync (receiver chan $$ c) $ \c' -> do+ withAsync (runConduit $ receiver chan .| c) $ \c' -> do link2 prevAsync c' wait c' stage chan prevAsync (Multiple bufsz c cs) = do -- not the last layer, so take the input from "chan", have this -- layer's conduit process it, and send the conduit's output to the -- next layer.- chan' <- liftIO $ newTBQueueIO bufsz- withAsync (sender chan' $ receiver chan =$= c) $ \c' -> do+ chan' <- liftIO $ newTBQueueIO (fromIntegral bufsz)+ withAsync (sender chan' $ receiver chan .| c) $ \c' -> do link2 prevAsync c' stage chan' c' cs -- A Producer which produces the values of the given channel until -- Nothing is received. This is the other half of "sender".-receiver :: (MonadIO m) => TBQueue (Maybe o) -> ConduitM () o m ()+receiver :: (MonadIO m) => TBQueue (Maybe o) -> ConduitT () o m () receiver chan = do mx <- recv chan case mx of@@ -317,14 +307,14 @@ -- | A "concurrent conduit", in which the stages run in parallel with -- a buffering queue and possibly a disk file between them. data CFConduit i o m r where- FSingle :: ConduitM i o m r -> CFConduit i o m r- FMultiple :: Int -> ConduitM i x m () -> CFConduit x o m r -> CFConduit i o m r- FMultipleF :: (Serialize x) => Int -> Maybe Int -> FilePath -> ConduitM i x m () -> CFConduit x o m r -> CFConduit i o m r+ FSingle :: ConduitT i o m r -> CFConduit i o m r+ FMultiple :: Int -> ConduitT i x m () -> CFConduit x o m r -> CFConduit i o m r+ FMultipleF :: (Serialize x) => Int -> Maybe Int -> FilePath -> ConduitT i x m () -> CFConduit x o m r -> CFConduit i o m r class CFConduitLike a where asCFConduit :: a i o m r -> CFConduit i o m r -instance CFConduitLike ConduitM where+instance CFConduitLike ConduitT where asCFConduit = FSingle instance CFConduitLike CConduit where@@ -335,7 +325,7 @@ asCFConduit = id data BufferContext m a = BufferContext { chan :: TBQueue a- , restore :: TQueue (Source m a)+ , restore :: TQueue (ConduitT () a m ()) , slotsFree :: TVar (Maybe Int) , done :: TVar Bool , tempDir :: FilePath@@ -344,47 +334,47 @@ -- The file-backed equivlent of "sender". This sends the values -- generated by "input" to the "chan" in the BufferContext until it -- gets full, then flushes it to disk via "persistChan".-fsender :: (MonadIO m, MonadResource m, Serialize x) => BufferContext m x -> ConduitM () x m () -> m ()+fsender :: (MonadThrow m, MonadResource m, Serialize x) => BufferContext m x -> ConduitT () x m () -> m () fsender bc@BufferContext{..} input = do- input $$ mapM_C $ \x -> join $ liftIO $ atomically $ do+ runConduit $ input .| (mapM_C $ \x -> join $ liftIO $ atomically $ do (writeTBQueue chan x >> return (return ())) `orElse` do action <- persistChan bc writeTBQueue chan x- return action+ return action) liftIO $ atomically $ writeTVar done True -- Connect a stage to another stage via either an in-memory queue or a -- disk buffer. This is the file-backed equivalent of "stage".-fstage :: (MonadBaseControl IO m, MonadIO m, MonadResource m) => ConduitM () i m () -> Async x -> CFConduit i Void m r -> m r+fstage :: (MonadThrow m, MonadUnliftIO m, MonadResource m) => ConduitT () i m () -> Async x -> CFConduit i Void m r -> m r fstage prevStage prevAsync (FSingle c) = -- The final conduit in the chain; just accept everything from -- the previous stage and wait for the result.- withAsync (prevStage $$ c) $ \c' -> do+ withAsync (runConduit $ prevStage .| c) $ \c' -> do link2 prevAsync c' wait c' fstage prevStage prevAsync (FMultiple bufsz c cs) = do -- This stage is connected to the next via a non-file-backed -- channel, so it just uses "sender" and "reciever" in the same way -- "stage" does.- chan' <- liftIO $ newTBQueueIO bufsz- withAsync (sender chan' $ prevStage =$= c) $ \c' -> do+ chan' <- liftIO $ newTBQueueIO (fromIntegral bufsz)+ withAsync (sender chan' $ prevStage .| c) $ \c' -> do link2 prevAsync c' fstage (receiver chan') c' cs fstage prevStage prevAsync (FMultipleF bufsz dsksz tempDir c cs) = do -- This potentially needs to write its output to a file, so it uses -- "fsender" send and tells the next stage to use "freceiver" to read.- bc <- liftIO $ BufferContext <$> newTBQueueIO bufsz+ bc <- liftIO $ BufferContext <$> newTBQueueIO (fromIntegral bufsz) <*> newTQueueIO <*> newTVarIO dsksz <*> newTVarIO False <*> pure tempDir- withAsync (fsender bc $ prevStage =$= c) $ \c' -> do+ withAsync (fsender bc $ prevStage .| c) $ \c' -> do link2 prevAsync c' fstage (freceiver bc) c' cs -- Receives from disk files or the in-memory queue if no spill-to-disk -- has occurred.-freceiver :: (MonadIO m) => BufferContext m o -> ConduitM () o m ()+freceiver :: (MonadIO m) => BufferContext m o -> ConduitT () o m () freceiver BufferContext{..} = loop where loop = do (src, exit) <- liftIO $ atomically $ do@@ -397,7 +387,7 @@ -- The channel is full, so (return an action which will) spill it to disk, unless too -- many items are there already.-persistChan :: (MonadIO m, MonadResource m, Serialize o) => BufferContext m o -> STM (m ())+persistChan :: (MonadThrow m, MonadResource m, Serialize o) => BufferContext m o -> STM (m ()) persistChan BufferContext{..} = do xs <- exhaust chan mslots <- readTVar slotsFree@@ -406,8 +396,8 @@ filePath <- newEmptyTMVar writeTQueue restore $ do (path, key) <- liftIO $ atomically $ takeTMVar filePath- CB.sourceFile path $= do- C.conduitGet get+ CB.sourceFile path .| do+ C.conduitGet2 get liftIO $ atomically $ modifyTVar slotsFree (fmap (+ len)) release key case xs of@@ -417,7 +407,7 @@ return $ do (key, (path, h)) <- allocate (openBinaryTempFile tempDir "conduit.bin") (\(path, h) -> hClose h `finally` removeFile path) liftIO $ do- CL.sourceList xs $= C.conduitPut put $$ CB.sinkHandle h+ runConduit $ CL.sourceList xs .| C.conduitPut put .| CB.sinkHandle h hClose h atomically $ putTMVar filePath (path, key)
Data/Conduit/TMChan.hs view
@@ -1,4 +1,4 @@-{-# LANGUAGE CPP #-}+{-# LANGUAGE BangPatterns #-} {-# LANGUAGE NoMonomorphismRestriction #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE RankNTypes #-}@@ -18,8 +18,14 @@ -- data (pictures, in this case) will be streamed down the channel to whatever -- is on the other side. ----- > _ <- forkIO . runResourceT $ loadTextures lotsOfPictures $$ sinkTBMChan chan+-- > _ <- forkIO . runResourceT $ do+-- > _ <- register $ atomically $ closeTBMChan chan+-- > loadTextures lotsOfPictures $$ sinkTBMChan chan --+-- We register closing function explicitly, because starting with version+-- @1.3.0@ @conduits@ library no longer maintain resources, so this is the+-- only way to safely close channel in case of exceptions.+-- -- Finally, we connect something to the other end of the channel. In this -- case, we connect a sink which uploads the textures one by one to the -- graphics card.@@ -58,46 +64,48 @@ , mergeConduits ) where -#if __GLASGOW_HASKELL__ < 710-import Control.Applicative-#endif import Control.Monad import Control.Monad.IO.Class ( liftIO, MonadIO ) import Control.Monad.Trans.Class import Control.Monad.Trans.Resource-import Control.Concurrent.STM+import Control.Concurrent (killThread, forkIOWithUnmask)+import Control.Concurrent.STM hiding (atomically) import Control.Concurrent.STM.TBMChan import Control.Concurrent.STM.TMChan import Data.Conduit+import Data.Foldable import qualified Data.Conduit.List as CL+import UnliftIO as Lifted +-- | Convert channel into the source.+--+-- *N.B* Since version 4.0 this function does not close the+-- channel if downstream is closed. chanSource :: MonadIO m => chan -- ^ The channel. -> (chan -> STM (Maybe a)) -- ^ The 'read' function.- -> (chan -> STM ()) -- ^ The 'close' function.- -> Source m a-chanSource ch reader closer =+ -> ConduitT z a m ()+chanSource ch reader = loop where loop = do a <- liftSTM $ reader ch case a of- Just x -> yieldOr x close >> loop+ Just x -> yield x >> loop Nothing -> return ()- close = liftSTM $ closer ch {-# INLINE chanSource #-} +-- | Convert channel into the consumer.+--+-- *N.B* chanSink :: MonadIO m => chan -- ^ The channel. -> (chan -> a -> STM ()) -- ^ The 'write' function.- -> (chan -> STM ()) -- ^ The 'close' function.- -> Sink a m ()-chanSink ch writer closer = do- CL.mapM_ $ liftIO . atomically . writer ch- liftSTM $ closer ch+ -> ConduitT a z m ()+chanSink ch writer = CL.mapM_ $ liftIO . atomically . writer ch {-# INLINE chanSink #-} -- | A simple wrapper around a TBMChan. As data is pushed into the channel, the@@ -105,15 +113,15 @@ -- channel is closed, the source will close also. -- -- If the channel fills up, the pipeline will stall until values are read.-sourceTBMChan :: MonadIO m => TBMChan a -> Source m a-sourceTBMChan ch = chanSource ch readTBMChan closeTBMChan+sourceTBMChan :: MonadIO m => TBMChan a -> ConduitT () a m ()+sourceTBMChan ch = chanSource ch readTBMChan {-# INLINE sourceTBMChan #-} -- | A simple wrapper around a TMChan. As data is pushed into the channel, the -- source will read it and pass it down the conduit pipeline. When the -- channel is closed, the source will close also.-sourceTMChan :: MonadIO m => TMChan a -> Source m a-sourceTMChan ch = chanSource ch readTMChan closeTMChan+sourceTMChan :: MonadIO m => TMChan a -> ConduitT () a m ()+sourceTMChan ch = chanSource ch readTMChan {-# INLINE sourceTMChan #-} -- | A simple wrapper around a TBMChan. As data is pushed into the sink, it@@ -121,27 +129,26 @@ -- the sink will block until space frees up. sinkTBMChan :: MonadIO m => TBMChan a- -> Bool -- ^ Should the channel be closed when the sink is closed?- -> Sink a m ()-sinkTBMChan ch close = chanSink ch writeTBMChan (when close . closeTBMChan)+ -> ConduitT a z m ()+sinkTBMChan ch = chanSink ch writeTBMChan {-# INLINE sinkTBMChan #-} -- | A simple wrapper around a TMChan. As data is pushed into this sink, it -- will magically begin to appear in the channel. sinkTMChan :: MonadIO m => TMChan a- -> Bool -- ^ Should the channel be closed when the sink is closed?- -> Sink a m ()-sinkTMChan ch close = chanSink ch writeTMChan (when close . closeTMChan)+ -> ConduitT a z m ()+sinkTMChan ch = chanSink ch writeTMChan {-# INLINE sinkTMChan #-} infixl 5 >=< -- | Modifies a TVar, returning its new value. modifyTVar'' :: TVar a -> (a -> a) -> STM a-modifyTVar'' tv f = do x <- f <$> readTVar tv- writeTVar tv x- return x+modifyTVar'' tv f = do+ !x <- f <$> readTVar tv+ writeTVar tv x+ return x liftSTM :: forall (m :: * -> *) a. MonadIO m => STM a -> m a liftSTM = liftIO . atomically@@ -151,17 +158,18 @@ -- -- The order of the new source's data is undefined, but it will be some -- combination of the two given sources.-(>=<) :: (MonadResource mi, MonadIO mo, MonadBaseControl IO mi)- => Source mi a- -> Source mi a- -> mi (Source mo a)+(>=<) :: (MonadResource mi, MonadIO mo, MonadUnliftIO mi)+ => ConduitT () a mi ()+ -> ConduitT () a mi ()+ -> mo (ConduitT () a mi ()) sa >=< sb = mergeSources [ sa, sb ] 16 {-# INLINE (>=<) #-} decRefcount :: TVar Int -> TBMChan a -> STM ()-decRefcount tv chan = do n <- modifyTVar'' tv (subtract 1)- when (n == 0) $- closeTBMChan chan+decRefcount tv chan = do+ n <- modifyTVar'' tv (subtract 1)+ when (n == 0) $+ closeTBMChan chan -- | Merges a list of sources, putting them all into a bounded channel, and -- returns a source which can be pulled from to pull from all the given@@ -170,25 +178,44 @@ -- The order of the new source's data is undefined, but it will be some -- combination of the given sources. The monad of the resultant source -- (@mo@) is independent of the monads of the input sources (@mi@).-mergeSources :: (MonadResource mi, MonadIO mo, MonadBaseControl IO mi)- => [Source mi a] -- ^ The sources to merge.+--+-- @since 3.0+-- All spawned threads will be removed when source is closed or upon an+-- exit from 'ResourceT' region. This means that result can only be used+-- within a 'runResourceT' scope.+--+-- @before 3.0+-- Spawned threads are not guaranteed to be closed. This may happen if+-- Source was closed before all it's input were closed.+mergeSources :: (MonadResource mi, MonadIO mo, MonadUnliftIO mi)+ => [ConduitT () a mi ()] -- ^ The sources to merge. -> Int -- ^ The bound of the intermediate channel.- -> mi (Source mo a)-mergeSources sx bound = do c <- liftSTM $ newTBMChan bound- refcount <- liftSTM . newTVar $ length sx- mapM_ (\s -> runResourceT $ resourceForkIO $ s $$ chanSink c writeTBMChan $ decRefcount refcount) (map (transPipe lift) sx)- return $ sourceTBMChan c+ -> mo (ConduitT () a mi ())+mergeSources sx bound = do+ return $ do + (chkey, c) <- allocate (liftSTM $ newTBMChan bound)+ (liftSTM . closeTBMChan)+ refcount <- liftSTM . newTVar $ length sx+ st <- lift $ askUnliftIO+ regs <- forM sx $ \s ->+ register . killThread =<<+ (liftIO $ forkIOWithUnmask $ \unmask ->+ (unmask $ unliftIO st $+ runConduit $ s .| chanSink c writeTBMChan)+ `Lifted.finally` (liftSTM $ decRefcount refcount c))+ chanSource c readTBMChan+ release chkey+ traverse_ release regs -- | Combines two conduits with unbounded channels, creating a new conduit -- which pulls data from a mix of the two: whichever produces first. -- -- The order of the new conduit's output is undefined, but it will be some -- combination of the two given conduits.-(<=>) :: (MonadIO mi, MonadIO mo, MonadBaseControl IO mi)- => Show i- => Conduit i (ResourceT mi) i- -> Conduit i (ResourceT mi) i- -> ResourceT mi (Conduit i mo i)+(<=>) :: (MonadThrow mi, MonadIO mo, MonadUnliftIO mi)+ => ConduitT i i (ResourceT mi) ()+ -> ConduitT i i (ResourceT mi) ()+ -> ResourceT mi (ConduitT i i mo ()) sa <=> sb = mergeConduits [ sa, sb ] 16 {-# INLINE (<=>) #-} @@ -199,22 +226,46 @@ -- The order of the new conduits's outputs is undefined, but it will be some -- combination of the given conduits. The monad of the resultant conduit -- (@mo@) is independent of the monads of the input conduits (@mi@).-mergeConduits :: (MonadIO mi, MonadIO mo, MonadBaseControl IO mi)- => [Conduit i (ResourceT mi) o] -- ^ The conduits to merge.+--+-- @since 3.0+-- Closes all worker processes when resulting conduit is closed or when execution+-- leaves ResourceT context. This means that conduit is only valid inside+-- 'runResouceT' scope.+--+-- @before 3.0+-- Spawned threads are not guaranteed to be closed, This may happen if threads+-- Conduit was closed before all threads have finished execution.+{-# DEPRECATED mergeConduits "This method will dissapear in the next version." #-}+mergeConduits :: (MonadIO mo, MonadUnliftIO mi)+ => [ConduitT i o (ResourceT mi) ()] -- ^ The conduits to merge. -> Int -- ^ The bound for the channels.- -> ResourceT mi (Conduit i mo o)+ -> ResourceT mi (ConduitT i o mo ()) mergeConduits conduits bound = do let len = length conduits refcount <- liftSTM $ newTVar len- iChannels <- replicateM len $ liftSTM $ newTBMChan bound- oChannel <- liftSTM $ newTBMChan bound- forM_ (zip iChannels conduits)- $ \(iChannel, conduit) -> resourceForkIO- $ sourceTBMChan iChannel- $$ conduit- =$ chanSink oChannel (writeTBMChans . (:[])) (decRefcount refcount)+ (iregs, iChannels) <- fmap unzip + $ replicateM len $ allocate (liftSTM $ newTBMChan bound)+ (liftSTM . closeTBMChan)+ (oreg, oChannel) <- allocate (liftSTM $ newTBMChan bound)+ (liftSTM . closeTBMChan)+ regs <- forM (zip iChannels conduits)+ $ \(iChannel, conduit) -> Lifted.mask_ $+ register . killThread <=< resourceForkIO+ $ (runConduit $ sourceTBMChan iChannel+ .| conduit+ .| chanSink oChannel writeTBMChan)+ `finally` + (liftIO $ atomically $ decRefcount refcount oChannel >> closeTBMChan iChannel)+ treg <- register $ do + traverse_ release regs+ traverse_ release iregs+ release oreg return- $ toConsumer (chanSink iChannels writeTBMChans (mapM_ closeTBMChan))- >> toProducer (sourceTBMChan oChannel)+ $ do chanSink iChannels writeTBMChans+ -- release internal channels effiniently closing input channels+ traverse_ release iregs+ chanSource oChannel readTBMChan+ -- release internals everything+ release treg where writeTBMChans channels a = forM_ channels $ \c -> writeTBMChan c a
Data/Conduit/TQueue.hs view
@@ -14,7 +14,7 @@ -- -- -- Here is short description of data structures:--- +-- -- * TQueue - unbounded infinite queue -- -- * TBQueue - bounded infinite queue@@ -24,7 +24,7 @@ -- * TBMQueue - bounded finite (closable) queue -- -- Caveats--- +-- -- Infinite operations means that source doesn't know when stream is -- ended so one need to use other methods of finishing stream like -- sending an exception or finish conduit in downstream.@@ -62,35 +62,37 @@ -- | A simple wrapper around a "TQueue". As data is pushed into the queue, the -- source will read it and pass it down the conduit pipeline.-sourceTQueue :: MonadIO m => TQueue a -> Source m a+sourceTQueue :: MonadIO m => TQueue a -> ConduitT z a m () sourceTQueue q = forever $ liftSTM (readTQueue q) >>= yield -- | A simple wrapper around a "TQueue". As data is pushed into this sink, it -- will magically begin to appear in the queue.-sinkTQueue :: MonadIO m => TQueue a -> Sink a m ()+sinkTQueue :: MonadIO m => TQueue a -> ConduitT a z m () sinkTQueue q = CL.mapM_ (liftSTM . writeTQueue q) -- | A simple wrapper around a "TBQueue". As data is pushed into the queue, the -- source will read it and pass it down the conduit pipeline.-sourceTBQueue :: MonadIO m => TBQueue a -> Source m a+sourceTBQueue :: MonadIO m => TBQueue a -> ConduitT z a m () sourceTBQueue q = forever $ liftSTM (readTBQueue q) >>= yield -- | A simple wrapper around a "TBQueue". As data is pushed into this sink, it--- will magically begin to appear in the queue. Boolean argument is used--- to specify if queue should be closed when the sink is closed.-sinkTBQueue :: MonadIO m => TBQueue a -> Sink a m ()+-- will magically begin to appear in the queue.+sinkTBQueue :: MonadIO m => TBQueue a -> ConduitT a z m () sinkTBQueue q = CL.mapM_ (liftSTM . writeTBQueue q) -- | A convenience wrapper for creating a source and sink TBQueue of the given -- size at once, without exposing the underlying queue.-entangledPair :: MonadIO m => Int -> m (Source m a, Sink a m ())+--+-- Returns release key that can be used for premature close of the communication+-- channel, otherwise channel will be closed when the ResourceT scope will be closed.+entangledPair :: MonadIO m => Int -> m (ConduitT z a m (), ConduitT a l m ()) entangledPair size = liftM (liftM2 (,) sourceTBQueue sinkTBQueue) $- liftIO $ atomically $ newTBQueue size+ liftIO $ atomically $ newTBQueue (fromIntegral size) -- | A simple wrapper around a "TMQueue". As data is pushed into the queue, the -- source will read it and pass it down the conduit pipeline. When the -- queue is closed, the source will close also.-sourceTMQueue :: MonadIO m => TMQueue a -> Source m a+sourceTMQueue :: MonadIO m => TMQueue a -> ConduitT z a m () sourceTMQueue q = loop where@@ -98,23 +100,19 @@ mx <- liftSTM $ readTMQueue q case mx of Nothing -> return ()- Just x -> yieldOr x close >> loop- close = liftSTM $ closeTMQueue q+ Just x -> yield x >> loop -- | A simple wrapper around a "TMQueue". As data is pushed into this sink, it -- will magically begin to appear in the queue. sinkTMQueue :: MonadIO m => TMQueue a- -> Bool -- ^ Should the queue be closed when the sink is closed?- -> Sink a m ()-sinkTMQueue q shouldClose = do- CL.mapM_ (liftSTM . writeTMQueue q)- when shouldClose (liftSTM $ closeTMQueue q)+ -> ConduitT a z m ()+sinkTMQueue q = CL.mapM_ (liftSTM . writeTMQueue q) -- | A simple wrapper around a "TBMQueue". As data is pushed into the queue, the -- source will read it and pass it down the conduit pipeline. When the -- queue is closed, the source will close also.-sourceTBMQueue :: MonadIO m => TBMQueue a -> Source m a+sourceTBMQueue :: MonadIO m => TBMQueue a -> ConduitT z a m () sourceTBMQueue q = loop where@@ -122,18 +120,14 @@ mx <- liftSTM $ readTBMQueue q case mx of Nothing -> return ()- Just x -> yieldOr x close >> loop- close = liftSTM $ closeTBMQueue q+ Just x -> yield x >> loop -- | A simple wrapper around a "TBMQueue". As data is pushed into this sink, it -- will magically begin to appear in the queue. sinkTBMQueue :: MonadIO m => TBMQueue a- -> Bool -- ^ Should the queue be closed when the sink is closed?- -> Sink a m ()-sinkTBMQueue q shouldClose = do- CL.mapM_ (liftSTM . writeTBMQueue q)- when shouldClose (liftSTM $ closeTBMQueue q)+ -> ConduitT a z m ()+sinkTBMQueue q = CL.mapM_ (liftSTM . writeTBMQueue q) liftSTM :: forall (m :: * -> *) a. MonadIO m => STM a -> m a
Data/Conduit/Utils.hs view
@@ -1,8 +1,8 @@ {-# LANGUAGE FunctionalDependencies #-} {-# LANGUAGE KindSignatures #-} {-# LANGUAGE FlexibleInstances #-}-{-| -This module provide different utility functions that allow to use safe higher +{-|+This module provide different utility functions that allow to use safe higher level usage. Conduit pairs allow creation of an internal datastructure that acts as a bridge,@@ -25,8 +25,8 @@ > mkSource = sourceTBMQueue > mkSink = flip sinkTBMQueue True - * Use "pair" or "pairBounded" to create a bridge. Because bridge data structure - is hidden and not seen in parameters, we need proxy type to help compiler to + * Use "pair" or "pairBounded" to create a bridge. Because bridge data structure+ is hidden and not seen in parameters, we need proxy type to help compiler to choose type, we use "Proxy2" for that. > pairTBMQueue = pairBounded (proxy2 :: Proxy2 TBMQueue a)@@ -40,15 +40,15 @@ conduit code. This package provides predefined pairs for all STM types that are used-in the package. +in the package. -} module Data.Conduit.Utils- ( + ( -- * Conduit pairs -- ** Low level functions pairBounded -- MonadIO m => m (Source m a, Sink m a ())- , pair -- MonadIO m => Int -> m (Source m a, Sink m a ()) + , pair -- MonadIO m => Int -> m (Source m a, Sink m a ()) -- ** Classes , UnboundedStream(..) , BoundedStream(..)@@ -90,14 +90,14 @@ -- | Class for structures that can handle unbounded stream of values. -- Such streams break conduit assumptions that constant memory will be--- used, because if receiver is slower then sender than values will +-- used, because if receiver is slower then sender than values will -- be accumulated. class UnboundedStream i o | i -> o where mkUStream :: i a -> IO (o a) -- | Class for structures that can handle bounded stream of values i.e.--- there is exists 'Int' value that sets an upper limit on the number --- of values that can be handled by structure. Exact meaning of this +-- there is exists 'Int' value that sets an upper limit on the number+-- of values that can be handled by structure. Exact meaning of this -- limit may depend on the carrier type. class BoundedStream i o | i -> o where mkBStream :: i a -> Int -> IO (o a)@@ -105,14 +105,14 @@ -- | Class that describes how we can make conduit out of the carrier -- value. class MonadIO m => IsConduit m (x :: * -> *) where- mkSink :: x a -> Sink a m ()- mkSource :: x a -> Source m a+ mkSink :: x a -> ConduitT a Void m ()+ mkSource :: x a -> ConduitT () a m () -- | Create bounded conduit pair, see "BoundedStream" class description. pairBounded :: (MonadIO m, IsConduit m o, BoundedStream i o) => i a -- ^ Type description. -> Int -- ^ Conduit size.- -> m (Source m a, Sink a m ())+ -> m (ConduitT () a m (), ConduitT a Void m ()) pairBounded p s = do q <- liftIO $ mkBStream p s return (mkSource q, mkSink q)@@ -120,7 +120,7 @@ -- | Create unbounded pair, see "UnboundedStream" class description. pair :: (MonadIO m, IsConduit m o, UnboundedStream i o) => i a -- ^ Type description.- -> m (Source m a, Sink a m ())+ -> m (ConduitT () a m (), ConduitT a Void m ()) pair p = do q <- liftIO $ mkUStream p return (mkSource q, mkSink q)@@ -129,13 +129,13 @@ -- Instances ------------------------------------------------------------------------------- instance BoundedStream (Proxy2 TBQueue) TBQueue where- mkBStream _ i = atomically $ newTBQueue i+ mkBStream _ i = atomically $ newTBQueue (fromIntegral i) instance BoundedStream (Proxy2 TBMQueue) TBMQueue where- mkBStream _ i = atomically $ newTBMQueue i+ mkBStream _ i = atomically $ newTBMQueue (fromIntegral i) instance BoundedStream (Proxy2 TBMChan) TBMChan where- mkBStream _ i = atomically $ newTBMChan i+ mkBStream _ i = atomically $ newTBMChan (fromIntegral i) instance UnboundedStream (Proxy2 TMQueue) TMQueue where mkUStream _ = atomically $ newTMQueue@@ -152,11 +152,11 @@ instance MonadIO m => IsConduit m TBMQueue where mkSource = sourceTBMQueue- mkSink = flip sinkTBMQueue True+ mkSink = sinkTBMQueue instance MonadIO m => IsConduit m TMQueue where mkSource = sourceTMQueue- mkSink = flip sinkTMQueue True+ mkSink = sinkTMQueue instance MonadIO m => IsConduit m TQueue where mkSource = sourceTQueue@@ -164,11 +164,11 @@ instance MonadIO m => IsConduit m TBMChan where mkSource = sourceTBMChan- mkSink = flip sinkTBMChan True+ mkSink = sinkTBMChan instance MonadIO m => IsConduit m TMChan where mkSource = sourceTMChan- mkSink = flip sinkTMChan True+ mkSink = sinkTMChan ------------------------------------------------------------------------------- -- Specialized functions@@ -179,12 +179,12 @@ -- is not closable then there is no way to notify receiver side that bridge -- is closed, so it's possible to use it only in infinite streams of when -- some other mechanism of notification is used.-pairTQueue, pairTMQueue, pairTMChan :: MonadIO m => m (Source m a, Sink a m ())+pairTQueue, pairTMQueue, pairTMChan :: MonadIO m => m (ConduitT () a m (), ConduitT a Void m ()) pairTQueue = pair (proxy2 :: Proxy2 TQueue a) pairTMQueue = pair (proxy2 :: Proxy2 TMQueue a) pairTMChan = pair (proxy2 :: Proxy2 TMChan a) -pairTBQueue, pairTBMQueue, pairTBMChan :: MonadIO m => Int -> m (Source m a, Sink a m ())+pairTBQueue, pairTBMQueue, pairTBMChan :: MonadIO m => Int -> m (ConduitT () a m (), ConduitT a Void m ()) pairTBQueue = pairBounded (proxy2 :: Proxy2 TBQueue a) pairTBMQueue = pairBounded (proxy2 :: Proxy2 TBMQueue a) pairTBMChan = pairBounded (proxy2 :: Proxy2 TBMChan a)
stm-conduit.cabal view
@@ -1,18 +1,18 @@ Name: stm-conduit-Version: 2.7.0+Version: 4.0.1 Synopsis: Introduces conduits to channels, and promotes using conduits concurrently. Description: Provides two simple conduit wrappers around STM channels - a source and a sink.-Homepage: https://github.com/wowus/stm-conduit+ +Homepage: https://github.com/cgaebel/stm-conduit License: BSD3 License-file: LICENSE Author: Clark Gaebel-Maintainer: cgaebel@uwaterloo.ca+Maintainer: cg.wowus.cg@gmail.com Category: Concurrency, Conduit Build-type: Simple -tested-with: GHC >= 7.6 && < 7.11--- selects GHC 7.4.1, 7.4.2, 7.6.1, 7.6.2, and 7.6.3+tested-with: GHC == 8.0.1, GHC == 8.2.1 Cabal-version: >=1.8 @@ -28,23 +28,19 @@ build-depends: base == 4.*- , transformers >= 0.2 && < 0.5- , stm == 2.4.*+ , transformers >= 0.2 && < 0.6+ , stm >= 2.4 && < 2.6 , stm-chans >= 2.0 && < 3.1 , cereal >= 0.4.0.1- , cereal-conduit >= 0.7.2- , conduit >= 1.0 && < 1.3- , conduit-combinators >= 0.3- , conduit-extra >= 1.0 && < 1.2+ , cereal-conduit >= 0.8+ , conduit >= 1.0 && < 1.4+ , conduit-extra >= 1.0 && < 1.4 , directory >= 1.1- , resourcet >= 0.3 && < 1.2+ , exceptions+ , resourcet >= 0.3 && < 1.3 , async >= 2.0.1- , monad-control >= 0.3.2 , monad-loops >= 0.4.2- , lifted-base >= 0.2.1- , lifted-async >= 0.1- , void >= 0.7- , ghc-prim+ , unliftio >= 0.2.0 && < 0.3.0 ghc-options: -Wall -fwarn-tabs -fwarn-unused-imports @@ -74,12 +70,12 @@ , stm , stm-conduit , conduit- , conduit-combinators >= 0.3 , transformers , stm-chans , resourcet , directory+ , unliftio source-repository head type: git- location: git://github.com/wowus/stm-conduit.git+ location: git://github.com/cgaebel/stm-conduit.git
test/DocTests.hs view
@@ -4,11 +4,11 @@ main :: IO () main = doctest [- "-packageghc"- , "-isrc"- , "-idist/build/autogen/"- , "-optP-include"- , "-optPdist/build/autogen/cabal_macros.h"- , "-cpp"- , "Data/Conduit/Async/Composition.hs"+-- "-packageghc"+-- , "-isrc"+-- , "-idist/build/autogen/"+-- , "-optP-include"+-- , "-optPdist/build/autogen/cabal_macros.h"+-- , "-cpp"+-- , "Data/Conduit/Async/Composition.hs" ]
test/Test.hs view
@@ -1,4 +1,4 @@-{-# LANGUAGE ScopedTypeVariables, RankNTypes, FlexibleContexts #-}+{-# LANGUAGE ScopedTypeVariables, RankNTypes, FlexibleContexts, BangPatterns #-} module Main ( main ) where @@ -12,7 +12,7 @@ import Test.HUnit import qualified Control.Monad as Monad-import Control.Monad.Trans.Resource (runResourceT)+import Control.Monad.Trans.Resource (runResourceT, allocate, release) import Control.Concurrent (forkIO, threadDelay) import Control.Concurrent.STM import Control.Concurrent.STM.TMQueue@@ -23,12 +23,13 @@ import Data.Conduit.TQueue import System.Directory import Conduit+import qualified UnliftIO as Lifted main = defaultMain tests tests = [ testGroup "Behaves to spec" [- testCase "simpleList using TMChan" test_simpleList+ testCase "simpleList using TMChan" test_simpleList , testCase "simpleList using TQueue" test_simpleQueue , testCase "simpleList using TMQueue" test_simpleMQueue ],@@ -48,38 +49,65 @@ ] ] -test_simpleList = do chan <- atomically $ newTMChan- forkIO . runResourceT $ sourceList testList $$ sinkTMChan chan True- lst' <- runResourceT $ sourceTMChan chan $$ consume- assertEqual "for the numbers [1..10000]," testList lst'- closed <- atomically $ isClosedTMChan chan- assertBool "channel is closed after running" closed+test_simpleList :: IO ()+test_simpleList =+ -- We start global resource region because we want to protect+ -- the channel. This is not strictly required under some cases+ -- but is needed in order to have exception safety. We need+ -- to cleanup properly in case of exception, but it depending+ -- on the use case it may be implemented by the other means.+ runResourceT $ do+ (reg, chan) <- allocate (newTMChanIO)+ (atomically . closeTMChan)+ _ <- Lifted.async $ do+ runConduit $ sourceList testList .| sinkTMChan chan+ -- We have to release resource explicitly in order to+ -- show that no data will be sent here ever.+ release reg+ + lst' <- runConduit $ sourceTMChan chan .| consume+ liftIO $ do + assertEqual "for the numbers [1..10000]," testList lst'+ closed <- atomically $ isClosedTMChan chan+ assertBool "channel is closed after running" closed where testList = [1..10000] -test_simpleQueue = do q <- atomically $ newTQueue- forkIO . runResourceT $ sourceList testList $$ sinkTQueue q- lst' <- runResourceT $ sourceTQueue q $$ CL.take (length testList)- assertEqual "for the numbers [1..10000]," testList lst'+test_simpleQueue :: IO ()+test_simpleQueue = runResourceT $ do+ q <- liftIO $ atomically $ newTQueue+ liftIO $ forkIO . runConduit $ sourceList testList .| sinkTQueue q+ lst' <- runConduit $ sourceTQueue q .| CL.take (length testList)+ liftIO $ assertEqual "for the numbers [1..10000]," testList lst' where testList = [1..10000] -test_simpleMQueue = do q <- atomically $ newTMQueue- forkIO . runResourceT $ sourceList testList $$ sinkTMQueue q True- lst' <- runResourceT $ sourceTMQueue q $$ consume- assertEqual "for the numbers [1..10000]," testList lst'- closed <- atomically $ isClosedTMQueue q- assertBool "channel is closed after running" closed- where- testList = [1..10000]+test_simpleMQueue :: IO ()+test_simpleMQueue = runResourceT $ do+ (reg, q) <- allocate (newTMQueueIO) (atomically . closeTMQueue)+ _ <- Lifted.async $ do+ runConduit $ sourceList testList .| sinkTMQueue q+ release reg+ + lst' <- runConduit $ sourceTMQueue q .| consume+ liftIO $ do+ assertEqual "for the numbers [1..10000]," testList lst'+ closed <- atomically $ isClosedTMQueue q+ assertBool "channel is closed after running" closed+ where+ testList = [1..10000] -test_multipleWriters = do ms <- runResourceT $ mergeSources [ sourceList ([1..10]::[Integer])- , sourceList ([11..20]::[Integer])- ] 3- xs <- ms $$ consume- assertEqual "for the numbers [1..10] and [11..20]," [1..20] $ sort xs+test_multipleWriters :: IO ()+test_multipleWriters = do+ xs <- runResourceT $ do+ ms <- mergeSources [ sourceList ([1..10]::[Integer])+ , sourceList ([11..20]::[Integer])+ ] 3+ runConduit $ ms .| consume+ assertEqual "for the numbers [1..10] and [11..20]," [1..20] $ sort xs -test_asyncOperator = do sum' <- CL.sourceList [1..n] $$ CL.fold (+) 0+test_asyncOperator :: IO ()+test_asyncOperator = do sum' <- runConduit $ CL.sourceList [1..n] .| CL.fold (+) 0 assertEqual ("for the sum of 1 to " ++ show n) sum sum' sum'' <- CL.sourceList [1..n] $$& CL.fold (+) 0 assertEqual "for the sum computed with the $$ and the $$&" sum' sum''@@ -88,10 +116,12 @@ n = 100 sum = (n * (n+1)) `div` 2 +test_buffer :: IO () test_buffer = do sum' <- buffer 128 (CL.sourceList [1..100]) (CL.fold (+) 0) assertEqual "sum computed using buffer" sum' (5050 :: Integer) +test_multi_buffer :: IO () test_multi_buffer = do sumDoubles <- CL.sourceList [1..100] $$& mapC (* 2) $=& CL.fold (+) 0 assertEqual "sum of doubles computed using two buffers" sumDoubles (10100 :: Integer)@@ -99,36 +129,43 @@ -- When we're testing file-buffering, we have to make sure to consume -- slowly enough to ensure the incoming data piles up enough to be -- flushed to disk..-slowDown :: (MonadIO m) => Int -> Conduit x m x+slowDown :: (MonadIO m) => Int -> ConduitT x x m () slowDown delay = awaitForever $ \x -> do liftIO $ threadDelay delay yield x +aLot, aLittle :: Int aLot = 10000 aLittle = 5000 +test_bufferToFile :: IO () test_bufferToFile = do tempDir <- getTemporaryDirectory- sum' <- runResourceT $ bufferToFile 16 (Just 25) tempDir (CL.sourceList [1 :: Int .. 100]) (slowDown aLittle $= CL.fold (+) 0)+ sum' <- runResourceT $ bufferToFile 16 (Just 25) tempDir (CL.sourceList [1 :: Int .. 100]) (slowDown aLittle .| CL.fold (+) 0) assertEqual "sum computed using bufferToFile" sum' 5050 +test_multi_bufferToFile :: IO () test_multi_bufferToFile = do tempDir <- getTemporaryDirectory sumDoubles <- let buf c = bufferToFile' 16 (Just 25) tempDir c -- "c" avoids monomorphism restriction- in runResourceT $ runCConduit $ CL.sourceList [1 :: Int .. 100] `buf` (slowDown aLittle $= mapC (* 2)) `buf` (slowDown aLot $= CL.fold (+) 0)+ in runResourceT $ runCConduit $ CL.sourceList [1 :: Int .. 100] `buf` (slowDown aLittle .| mapC (* 2)) `buf` (slowDown aLot .| CL.fold (+) 0) assertEqual "sum of doubles computed using bufferToFile" sumDoubles 10100 +test_mixed_buffer :: IO () test_mixed_buffer = do tempDir <- getTemporaryDirectory- sumDoubles <- let buf = bufferToFile' 16 (Just 25) tempDir- in runResourceT $ CL.sourceList [1 :: Int .. 100] $$& mapC (* 2) `buf` (slowDown aLittle $= CL.fold (+) 0)+ sumDoubles <-+ let buf = bufferToFile' 16 (Just 25) tempDir+ in runResourceT $+ CL.sourceList [1 :: Int .. 100] $$& mapC (* 2) `buf` (slowDown aLittle .| CL.fold (+) 0) assertEqual "sum of doubles computed using mixed buffers" sumDoubles 10100 sumTriples <- let buf = bufferToFile' 16 (Just 25) tempDir- in runResourceT $ CL.sourceList [1 :: Int .. 100] `buf` (slowDown aLittle $= mapC (* 3)) $$& CL.fold (+) 0+ in runResourceT $ CL.sourceList [1 :: Int .. 100] `buf` (slowDown aLittle .| mapC (* 3)) $$& CL.fold (+) 0 assertEqual "sum of triples computed using mixed buffers" sumTriples 15150 +test_gatherFrom :: IO () test_gatherFrom = do- sum' <- gatherFrom 128 gen $$ CL.fold (+) 0+ sum' <- runConduit $ gatherFrom 128 gen .| CL.fold (+) 0 assertEqual "sum computed using gatherFrom" sum' 5050 where gen queue = Monad.void $ Monad.foldM f queue [1..100]@@ -137,16 +174,18 @@ atomically $ writeTBQueue q x return q +test_drainTo :: IO () test_drainTo = do- sum' <- CL.sourceList [1..100] $$ drainTo 128 (go 0)+ sum' <- runConduit $ CL.sourceList [1..100] .| drainTo 128 (go 0) assertEqual "sum computed using drainTo" sum' 5050 where- go acc queue = do+ go !acc queue = do mres <- atomically $ readTBQueue queue case mres of Nothing -> return acc Just res -> go (acc + res) queue +test_mergeConduits :: IO () test_mergeConduits = do merged <- runResourceT $ mergeConduits [ CL.map (* 2)@@ -155,5 +194,5 @@ let input = [1..10] expected = Prelude.map (2 *) input ++ tail (Prelude.scanl (+) 0 input)- xs <- sourceList ([1..10] :: [Integer]) $$ merged =$ consume+ xs <- runConduit $ sourceList ([1..10] :: [Integer]) .| merged .| consume assertEqual "merged results" (sort expected) (sort xs)