stm-conduit 1.1.0 → 2.0.0

raw patch · 3 files changed

+65/−5 lines, 3 filesdep ~stm-chansPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: stm-chans

API changes (from Hackage documentation)

Files

• Data/Conduit/TQueue.hs +50/−2
• stm-conduit.cabal +3/−3
• test/Test.hs +12/−0

@@ -16,7 +16,9 @@ --   Here is short description of data structures: --      --     * TQueue   - unbounded infinite queue---     * TBMQueue - bounded infinite queue+--     * TBQueue  - bounded infinite queue+--     * TMQueue  - unbounded finite (closable) queue+--     * TBMQueue - bounded finite (closable) queue -- -- Caveats -- @@ -33,11 +35,19 @@     -- ** TBQueue connectors   , sourceTBQueue   , sinkTBQueue+    -- ** TMQueue connectors+  , sourceTMQueue+  , sinkTMQueue+    -- ** TBMQueue connectors+  , sourceTBMQueue+  , sinkTBMQueue   , module Control.Concurrent.STM.TQueue   ) where  import Control.Concurrent.STM import Control.Concurrent.STM.TQueue+import Control.Concurrent.STM.TBMQueue+import Control.Concurrent.STM.TMQueue import Control.Monad.IO.Class import Data.Conduit import Data.Conduit.Internal@@ -74,6 +84,44 @@                             >> (return $ NeedInput push close))         close _    = return () +sourceTMQueue :: MonadIO m => TMQueue a -> Source m a+sourceTMQueue q = ConduitM src+  where src = PipeM pull+        pull = do mx <- liftSTM $ readTMQueue q+                  case mx of+                      Nothing -> return $ Done ()+                      Just x -> return $ HaveOutput src close x+        close = do liftSTM $ closeTMQueue q+                   return ()++sinkTMQueue :: MonadIO m => TMQueue a -> Sink a m ()+sinkTMQueue q = 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+                     return ()++sourceTBMQueue :: MonadIO m => TBMQueue a -> Source m a+sourceTBMQueue q = ConduitM src+  where src = PipeM pull+        pull = do mx <- liftSTM $ readTBMQueue q+                  case mx of+                      Nothing -> return $ Done ()+                      Just x -> return $ HaveOutput src close x+        close = do liftSTM $ closeTBMQueue q+                   return ()++sinkTBMQueue :: MonadIO m => TBMQueue a -> Sink a m ()+sinkTBMQueue q = 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+                     return ()++ liftSTM :: forall (m :: * -> *) a. MonadIO m => STM a -> m a liftSTM = liftIO . atomically-

@@ -1,5 +1,5 @@ Name:                stm-conduit-Version:             1.1.0+Version:             2.0.0 Synopsis:            Introduces conduits to channels, and promotes using                      conduits concurrently. Description:         Provides two simple conduit wrappers around STM@@ -24,7 +24,7 @@         base         == 4.*       , transformers >= 0.2 && <= 0.4       , stm          == 2.4.*-      , stm-chans    == 1.3.*+      , stm-chans    == 2.0.*       , conduit      == 1.0.*       , resourcet    >= 0.3 && < 0.5 @@ -48,7 +48,7 @@       , stm-conduit       , conduit       , transformers-      , stm-chans+      , stm-chans == 2.0.*    source-repository head     type:     git

@@ -13,6 +13,7 @@ import Control.Monad.Trans.Class (lift) import Control.Concurrent import Control.Concurrent.STM+import Control.Concurrent.STM.TMQueue import Data.Conduit import Data.Conduit.List as CL import Data.Conduit.TMChan@@ -24,6 +25,7 @@         testGroup "Behaves to spec" [                   testCase "simpleList using TMChan" test_simpleList                 , testCase "simpleList using TQueue" test_simpleQueue+                , testCase "simpleList using TMQueue" test_simpleMQueue             ],         testGroup "Bug fixes" [                 testCase "multipleWriters" test_multipleWriters@@ -45,6 +47,16 @@                       assertEqual "for the numbers [1..10000]," testList lst'     where         testList = [1..10000]++test_simpleMQueue = do q <- atomically $ newTMQueue+                       forkIO . runResourceT $ sourceList testList $$ sinkTMQueue q+                       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_multipleWriters = do ms <- runResourceT $ mergeSources [ sourceList ([1..10]::[Integer])                                                             , sourceList ([11..20]::[Integer])                                                             ] 3