stm-conduit 2.1.4 → 2.2
raw patch · 4 files changed
+63/−23 lines, 4 files
Files
- Data/Conduit/TMChan.hs +13/−8
- Data/Conduit/TQueue.hs +47/−12
- stm-conduit.cabal +1/−1
- test/Test.hs +2/−2
Data/Conduit/TMChan.hs view
@@ -111,17 +111,21 @@ -- | A simple wrapper around a TBMChan. As data is pushed into the sink, it -- will magically begin to appear in the channel. If the channel is full,--- the sink will block until space frees up. When the sink is closed, the--- channel will close too.-sinkTBMChan :: MonadIO m => TBMChan a -> Sink a m ()-sinkTBMChan ch = chanSink ch writeTBMChan closeTBMChan+-- 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) {-# INLINE sinkTBMChan #-} -- | A simple wrapper around a TMChan. As data is pushed into this sink, it--- will magically begin to appear in the channel. When the sink is closed,--- the channel will close too.-sinkTMChan :: MonadIO m => TMChan a -> Sink a m ()-sinkTMChan ch = chanSink ch writeTMChan closeTMChan+-- 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) {-# INLINE sinkTMChan #-} infixl 5 >=<@@ -131,6 +135,7 @@ 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
Data/Conduit/TQueue.hs view
@@ -16,8 +16,11 @@ -- Here is short description of data structures: -- -- * TQueue - unbounded infinite queue+-- -- * TBQueue - bounded infinite queue+-- -- * TMQueue - unbounded finite (closable) queue+-- -- * TBMQueue - bounded finite (closable) queue -- -- Caveats@@ -29,16 +32,19 @@ module Data.Conduit.TQueue ( -- * Connectors- -- ** TQueue connectors + -- ** Infinite queues+ -- $inifinite+ -- *** TQueue connectors sourceTQueue , sinkTQueue- -- ** TBQueue connectors+ -- *** TBQueue connectors , sourceTBQueue , sinkTBQueue- -- ** TMQueue connectors+ -- ** Closable queues+ -- *** TMQueue connectors , sourceTMQueue , sinkTMQueue- -- ** TBMQueue connectors+ -- *** TBMQueue connectors , sourceTBMQueue , sinkTBMQueue , module Control.Concurrent.STM.TQueue@@ -48,12 +54,13 @@ import Control.Concurrent.STM.TQueue import Control.Concurrent.STM.TBMQueue import Control.Concurrent.STM.TMQueue+import Control.Monad import Control.Monad.IO.Class import Data.Conduit import Data.Conduit.Internal --+-- | 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 q = ConduitM src where src = PipeM pull@@ -61,6 +68,8 @@ return $ HaveOutput src close x close = return () +-- | 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 q = ConduitM src where src = sink@@ -69,6 +78,8 @@ >> (return $ NeedInput push close)) close _ = return () +-- | 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 q = ConduitM src where src = PipeM pull@@ -76,6 +87,9 @@ return $ HaveOutput src close x close = return () +-- | 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 () sinkTBQueue q = ConduitM src where src = sink@@ -84,6 +98,9 @@ >> (return $ NeedInput push close)) close _ = return () +-- | 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 q = ConduitM src where src = PipeM pull@@ -94,15 +111,23 @@ close = do liftSTM $ closeTMQueue q return () -sinkTMQueue :: MonadIO m => TMQueue a -> Sink a m ()-sinkTMQueue q = ConduitM src+-- | 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 = ConduitM src where src = sink sink = NeedInput push close push input = PipeM ((liftSTM $ writeTMQueue q input) >> (return $ NeedInput push close))- close _ = do liftSTM $ closeTMQueue q+ close _ = do when shouldClose (liftSTM $ closeTMQueue q) return () +-- | 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 q = ConduitM src where src = PipeM pull@@ -113,15 +138,25 @@ close = do liftSTM $ closeTBMQueue q return () -sinkTBMQueue :: MonadIO m => TBMQueue a -> Sink a m ()-sinkTBMQueue q = ConduitM src+-- | 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 = ConduitM src where src = sink sink = NeedInput push close push input = PipeM ((liftSTM $ writeTBMQueue q input) >> (return $ NeedInput push close))- close _ = do liftSTM $ closeTBMQueue q+ close _ = do when shouldClose (liftSTM $ closeTBMQueue q) return () liftSTM :: forall (m :: * -> *) a. MonadIO m => STM a -> m a liftSTM = liftIO . atomically++-- $infinite+-- It's impossible to close infinite queues but they work slightly faster,+-- so it's reasonable to use them inside infinite computations for+-- performance reasons.
stm-conduit.cabal view
@@ -1,5 +1,5 @@ Name: stm-conduit-Version: 2.1.4+Version: 2.2 Synopsis: Introduces conduits to channels, and promotes using conduits concurrently. Description: Provides two simple conduit wrappers around STM
test/Test.hs view
@@ -41,7 +41,7 @@ ] test_simpleList = do chan <- atomically $ newTMChan- forkIO . runResourceT $ sourceList testList $$ sinkTMChan chan+ forkIO . runResourceT $ sourceList testList $$ sinkTMChan chan True lst' <- runResourceT $ sourceTMChan chan $$ consume assertEqual "for the numbers [1..10000]," testList lst' closed <- atomically $ isClosedTMChan chan@@ -57,7 +57,7 @@ testList = [1..10000] test_simpleMQueue = do q <- atomically $ newTMQueue- forkIO . runResourceT $ sourceList testList $$ sinkTMQueue q+ forkIO . runResourceT $ sourceList testList $$ sinkTMQueue q True lst' <- runResourceT $ sourceTMQueue q $$ consume assertEqual "for the numbers [1..10000]," testList lst' closed <- atomically $ isClosedTMQueue q