stm-conduit 2.8.0 → 3.0.0
raw patch · 3 files changed
+48/−20 lines, 3 filesdep ~conduit-combinatorsdep ~conduit-extradep ~transformers
Dependency ranges changed: conduit-combinators, conduit-extra, transformers
Files
- Data/Conduit/TMChan.hs +41/−14
- stm-conduit.cabal +2/−2
- test/Test.hs +5/−4
Data/Conduit/TMChan.hs view
@@ -65,10 +65,13 @@ import Control.Monad.IO.Class ( liftIO, MonadIO ) import Control.Monad.Trans.Class import Control.Monad.Trans.Resource+import Control.Concurrent (killThread) import Control.Concurrent.STM import Control.Concurrent.STM.TBMChan import Control.Concurrent.STM.TMChan +import qualified Control.Exception.Lifted as Lifted+ import Data.Conduit import qualified Data.Conduit.List as CL @@ -76,7 +79,7 @@ :: MonadIO m => chan -- ^ The channel. -> (chan -> STM (Maybe a)) -- ^ The 'read' function.- -> (chan -> STM ()) -- ^ The 'close' function.+ -> (chan -> IO ()) -- ^ The 'close' function. -> Source m a chanSource ch reader closer = loop@@ -86,7 +89,7 @@ case a of Just x -> yieldOr x close >> loop Nothing -> return ()- close = liftSTM $ closer ch+ close = liftIO $ closer ch {-# INLINE chanSource #-} chanSink@@ -106,14 +109,14 @@ -- -- 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 ch = chanSource ch readTBMChan (atomically . closeTBMChan) {-# 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 ch = chanSource ch readTMChan (atomically . closeTMChan) {-# INLINE sourceTMChan #-} -- | A simple wrapper around a TBMChan. As data is pushed into the sink, it@@ -170,21 +173,34 @@ -- 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@).+--+-- @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, MonadBaseControl IO mi) => [Source mi a] -- ^ 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+mergeSources sx bound = do+ c <- liftSTM $ newTBMChan bound+ refcount <- liftSTM . newTVar $ length sx+ regs <- forM (map (transPipe lift) sx) $ \s -> Lifted.mask_ $ do+ register . killThread <=< runResourceT $ resourceForkIO $ s $$ chanSink c writeTBMChan $ decRefcount refcount+ return $ chanSource c readTBMChan+ (\chan -> do liftSTM $ closeTBMChan chan+ mapM_ 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)+(<=>) :: (MonadIO mi, MonadThrow mi, MonadIO mo, MonadBaseControl IO mi) => Show i => Conduit i (ResourceT mi) i -> Conduit i (ResourceT mi) i@@ -199,7 +215,16 @@ -- 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)+--+-- @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.+mergeConduits :: (MonadIO mi, MonadIO mi, MonadThrow mi, MonadIO mo, MonadBaseControl IO mi) => [Conduit i (ResourceT mi) o] -- ^ The conduits to merge. -> Int -- ^ The bound for the channels. -> ResourceT mi (Conduit i mo o)@@ -208,13 +233,15 @@ refcount <- liftSTM $ newTVar len iChannels <- replicateM len $ liftSTM $ newTBMChan bound oChannel <- liftSTM $ newTBMChan bound- forM_ (zip iChannels conduits)- $ \(iChannel, conduit) -> resourceForkIO+ regs <- forM (zip iChannels conduits)+ $ \(iChannel, conduit) -> Lifted.mask_ $+ register . killThread <=< resourceForkIO $ sourceTBMChan iChannel $$ conduit- =$ chanSink oChannel (writeTBMChans . (:[])) (decRefcount refcount)+ =$ chanSink oChannel writeTBMChan (decRefcount refcount) return $ toConsumer (chanSink iChannels writeTBMChans (mapM_ closeTBMChan))- >> toProducer (sourceTBMChan oChannel)+ >> toProducer (chanSource oChannel readTBMChan (\c -> do atomically $ closeTBMChan c+ mapM_ release regs)) where writeTBMChans channels a = forM_ channels $ \c -> writeTBMChan c a
stm-conduit.cabal view
@@ -1,5 +1,5 @@ Name: stm-conduit-Version: 2.8.0+Version: 3.0.0 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/cgaebel/stm-conduit@@ -28,7 +28,7 @@ build-depends: base == 4.*- , transformers >= 0.2 && < 0.5+ , transformers >= 0.2 && < 0.6 , stm == 2.4.* , stm-chans >= 2.0 && < 3.1 , cereal >= 0.4.0.1
test/Test.hs view
@@ -73,10 +73,11 @@ where testList = [1..10000] -test_multipleWriters = do ms <- runResourceT $ mergeSources [ sourceList ([1..10]::[Integer])- , sourceList ([11..20]::[Integer])- ] 3- xs <- ms $$ consume+test_multipleWriters = do xs <- runResourceT $ do+ ms <- mergeSources [ sourceList ([1..10]::[Integer])+ , sourceList ([11..20]::[Integer])+ ] 3+ 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