stm-conduit 0.2.4 → 0.2.4.1
raw patch · 3 files changed
+66/−9 lines, 3 filesPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
Files
- Data/Conduit/TMChan.hs +22/−7
- stm-conduit.cabal +2/−2
- test/Test.hs +42/−0
Data/Conduit/TMChan.hs view
@@ -50,6 +50,8 @@ , mergeSources ) where +import Control.Applicative+import Control.Monad import Control.Monad.IO.Class ( liftIO ) import Control.Monad.Trans.Resource import Control.Concurrent.STM@@ -67,7 +69,7 @@ chanSource ch reader closer = src where src = Source pull close- pull = do a <- liftIO . atomically $ reader ch+ pull = do a <- liftSTM $ reader ch case a of Just x -> return $ Open src x Nothing -> return Closed@@ -83,7 +85,7 @@ chanSink ch writer closer = sink where sink = SinkData push close- push input = do liftIO . atomically $ writer ch input+ push input = do liftSTM $ writer ch input return $ Processing push close close = liftIO . atomically $ closer ch {-# INLINE chanSink #-}@@ -121,6 +123,15 @@ 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++liftSTM :: ResourceIO m => STM a -> ResourceT m a+liftSTM = liftIO . atomically+ -- | Combines two sources with an unbounded channel, creating a new source -- which pulls data from a mix of the two sources: whichever produces first. --@@ -130,11 +141,14 @@ => Source m a -> Source m a -> ResourceT m (Source m a)-sa >=< sb = do c <- liftIO . atomically $ newTMChan - _ <- resourceForkIO $ sa $$ sinkTMChan c- _ <- resourceForkIO $ sb $$ sinkTMChan c- return $ sourceTMChan c+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+ -- | 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 -- sources in a first-come-first-serve basis.@@ -146,5 +160,6 @@ -> Int -- ^ The bound of the intermediate channel. -> ResourceT m (Source m a) mergeSources sx bound = do c <- liftIO . atomically $ newTBMChan bound- mapM_ (\s -> resourceForkIO $ s $$ sinkTBMChan c) sx+ refcount <- liftSTM . newTVar $ length sx+ mapM_ (\s -> resourceForkIO $ s $$ chanSink c writeTBMChan $ decRefcount refcount) sx return $ sourceTBMChan c
stm-conduit.cabal view
@@ -1,5 +1,5 @@ Name: stm-conduit-Version: 0.2.4+Version: 0.2.4.1 Synopsis: Introduces conduits to channels, and promotes using conduits concurrently. Description: Provides two simple conduit wrappers around STM@@ -32,7 +32,7 @@ main-is: Test.hs hs-source-dirs: test/ - ghc-options: -rtsopts=all+ ghc-options: -rtsopts=all -threaded Build-Depends: base ==4.*, QuickCheck >= 2,
+ test/Test.hs view
@@ -0,0 +1,42 @@+module Main ( main ) where++import Data.List++import Test.Framework (defaultMain, testGroup)+import Test.Framework.Providers.HUnit+import Test.Framework.Providers.QuickCheck2 (testProperty)++import Test.QuickCheck+import Test.HUnit++import Control.Concurrent+import Control.Concurrent.STM+import Data.Conduit+import Data.Conduit.List+import Data.Conduit.TMChan++main = defaultMain tests++tests = [+ testGroup "Behaves to spec" [+ testCase "simpleList" test_simpleList+ ],+ testGroup "Bug fixes" [+ testCase "multipleWriters" test_multipleWriters+ ]+ ]++test_simpleList = do chan <- atomically $ newTMChan+ forkIO . runResourceT $ sourceList testList $$ sinkTMChan chan+ lst' <- runResourceT $ sourceTMChan chan $$ consume+ assertEqual "for the numbers [1..10000]," testList lst'+ closed <- atomically $ isClosedTMChan chan+ assertBool "channel is closed after running" closed+ where+ testList = [1..10000]++test_multipleWriters = do ms <- runResourceT $ mergeSources [ sourceList [1..10]+ , sourceList [11..20]+ ] 3+ xs <- runResourceT $ ms $$ consume+ assertEqual "for the numbers [1..10] and [11..20]," [1..20] $ sort xs