stm-conduit 2.2.2 → 2.3.0

raw patch · 3 files changed

+58/−7 lines, 3 files

Files

• Data/Conduit/TMChan.hs +41/−1
• stm-conduit.cabal +1/−1
• test/Test.hs +16/−5

@@ -49,6 +49,8 @@                            -- * Parallel Combinators                            , (>=<)                            , mergeSources+                           , (<=>)+                           , mergeConduits                            ) where  import Control.Applicative@@ -60,7 +62,7 @@ import Control.Concurrent.STM.TMChan  import Data.Conduit-import Data.Conduit.Internal+import Data.Conduit.Internal (Pipe (..), ConduitM (..))  chanSource      :: MonadIO m@@ -171,3 +173,41 @@                            mapM_ (\s -> resourceForkIO $ s $$ chanSink c writeTBMChan $ decRefcount refcount) sx                            return $ sourceTBMChan c +-- | 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 m, MonadBaseControl IO m)+      => Show i+      => Conduit i (ResourceT m) i+      -> Conduit i (ResourceT m) i+      -> ResourceT m (Conduit i (ResourceT m) i)+sa <=> sb = mergeConduits [ sa, sb ] 16+{-# INLINE (<=>) #-}++-- | Provide an input across several conduits, putting them all into a bounded+--   channel. Returns a conduit which can be pulled from to pull from all the+--   given conduits in a first-come-first-serve basis.+--+--   The order of the new conduits's outputs is undefined, but it will be some+--   combination of the given conduits.+mergeConduits :: (MonadIO m, MonadBaseControl IO m)+              => [Conduit i (ResourceT m) o] -- ^ The conduits to merge.+              -> Int -- ^ The bound for the channels.+              -> ResourceT m (Conduit i (ResourceT m) o)+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)+        return+            $  toConsumer (chanSink iChannels writeTBMChans (mapM_ closeTBMChan))+            >> toProducer (sourceTBMChan oChannel)+  where+    writeTBMChans channels a = forM_ channels $ \c -> writeTBMChan c a

@@ -1,5 +1,5 @@ Name:                stm-conduit-Version:             2.2.2+Version:             2.3.0 Synopsis:            Introduces conduits to channels, and promotes using                      conduits concurrently. Description:         Provides two simple conduit wrappers around STM

@@ -1,6 +1,6 @@ module Main ( main ) where -import Data.List+import Data.List (sort)  import Test.Framework (defaultMain, testGroup) import Test.Framework.Providers.HUnit@@ -10,10 +10,8 @@ import Test.HUnit  import qualified Control.Monad as Monad-import Control.Monad.IO.Class (liftIO)-import Control.Monad.Trans.Class (lift) import Control.Monad.Trans.Resource (runResourceT)-import Control.Concurrent+import Control.Concurrent (forkIO) import Control.Concurrent.STM import Control.Concurrent.STM.TMQueue import Data.Conduit@@ -36,6 +34,7 @@                 , testCase "bufferToFile" test_bufferToFile                 , testCase "gatherFrom" test_gatherFrom                 , testCase "drainTo" test_drainTo+                , testCase "mergeConduits" test_mergeConduits             ],         testGroup "Bug fixes" [                   testCase "multipleWriters" test_multipleWriters@@ -72,7 +71,7 @@                                                             , sourceList ([11..20]::[Integer])                                                             ] 3                           xs <- runResourceT $ ms $$ consume-                          liftIO $ assertEqual "for the numbers [1..10] and [11..20]," [1..20] $ sort xs+                          assertEqual "for the numbers [1..10] and [11..20]," [1..20] $ sort xs  test_asyncOperator = do sum'  <- CL.sourceList [1..n] $$ CL.fold (+) 0                         assertEqual ("for the sum of 1 to " ++ show n) sum sum'@@ -110,3 +109,15 @@         case mres of             Nothing  -> return acc             Just res -> go (acc + res) queue++test_mergeConduits = do merged <- runResourceT $ mergeConduits+                                                    [ CL.map (* 2)+                                                    , scanlConduit (+) 0+                                                    ] 16+                        let+                          input = [1..10]+                          expected = Prelude.map (2 *) input ++ Prelude.scanl (+) 0 input+                        xs <- runResourceT $ sourceList ([1..10] :: [Integer]) $$ merged =$ consume+                        assertEqual "merged results" (sort expected) (sort xs)+  where+    scanlConduit f b = yield b >> CL.scanl (\a -> (\x -> (x, x)) . f a) b