stm-conduit 1.0.0 → 1.1.0

raw patch · 3 files changed

+92/−4 lines, 3 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

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

@@ -0,0 +1,79 @@+{-# LANGUAGE Rank2Types, KindSignatures #-}+-- | Contains a simple source and sink linking together conduits in+--   different threads. For extended examples of usage and bottlenecks+--   see 'Data.Conduit.TMChan'.+--+--   TQueue is an amoritized FIFO queue behaves like TChan, with two+--   important differences:+--+--     * it's faster (but amortized thus the cost of individual operations+--     may vary a lot)+--+--     * it doesn't provide equivalent of the dupTChan and cloneTChan+--     operations+--+--+--   Here is short description of data structures:+--     +--     * TQueue   - unbounded infinite queue+--     * TBMQueue - bounded infinite queue+--+-- Caveats+-- +--   Infinite operations means that source doesn't know when stream is+--   ended so one need to use other methods of finishing stream like+--   sending an exception or finish conduit in downstream.+--++module Data.Conduit.TQueue+  ( -- * Connectors+    -- ** TQueue connectors +    sourceTQueue+  , sinkTQueue+    -- ** TBQueue connectors+  , sourceTBQueue+  , sinkTBQueue+  , module Control.Concurrent.STM.TQueue+  ) where++import Control.Concurrent.STM+import Control.Concurrent.STM.TQueue+import Control.Monad.IO.Class+import Data.Conduit+import Data.Conduit.Internal++++sourceTQueue :: MonadIO m => TQueue a -> Source m a+sourceTQueue q = ConduitM src+  where src = PipeM pull+        pull = do x <- liftSTM $ readTQueue q+                  return $ HaveOutput src close x+        close = return ()++sinkTQueue :: MonadIO m => TQueue a -> Sink a m ()+sinkTQueue q = ConduitM src+  where src        = sink+        sink       = NeedInput push close+        push input = PipeM ((liftSTM $ writeTQueue q input)+                            >> (return $ NeedInput push close))+        close _    = return ()++sourceTBQueue :: MonadIO m => TBQueue a -> Source m a+sourceTBQueue q = ConduitM src+  where src = PipeM pull+        pull = do x <- liftSTM $ readTBQueue q+                  return $ HaveOutput src close x+        close = return ()++sinkTBQueue :: MonadIO m => TBQueue a -> Sink a m ()+sinkTBQueue q = ConduitM src+  where src        = sink+        sink       = NeedInput push close+        push input = PipeM ((liftSTM $ writeTBQueue q input)+                            >> (return $ NeedInput push close))+        close _    = return ()++liftSTM :: forall (m :: * -> *) a. MonadIO m => STM a -> m a+liftSTM = liftIO . atomically+

@@ -1,5 +1,5 @@ Name:                stm-conduit-Version:             1.0.0+Version:             1.1.0 Synopsis:            Introduces conduits to channels, and promotes using                      conduits concurrently. Description:         Provides two simple conduit wrappers around STM@@ -8,7 +8,7 @@ License:             BSD3 License-file:        LICENSE Author:              Clark Gaebel-Maintainer:          cgaebel@csclub.uwaterloo.ca+Maintainer:          cgaebel@uwaterloo.ca Category:            Concurrency, Conduit  Build-type:          Simple@@ -18,6 +18,7 @@ Library     exposed-modules:         Data.Conduit.TMChan+        Data.Conduit.TQueue          build-depends:         base         == 4.*

@@ -14,14 +14,16 @@ import Control.Concurrent import Control.Concurrent.STM import Data.Conduit-import Data.Conduit.List+import Data.Conduit.List as CL import Data.Conduit.TMChan+import Data.Conduit.TQueue  main = defaultMain tests  tests = [         testGroup "Behaves to spec" [-                testCase "simpleList" test_simpleList+                  testCase "simpleList using TMChan" test_simpleList+                , testCase "simpleList using TQueue" test_simpleQueue             ],         testGroup "Bug fixes" [                 testCase "multipleWriters" test_multipleWriters@@ -37,6 +39,12 @@     where         testList = [1..10000] +test_simpleQueue = do q <- atomically $ newTQueue+                      forkIO . runResourceT $ sourceList testList $$ sinkTQueue q+                      lst'  <- runResourceT $ sourceTQueue q $$ CL.take (length testList)+                      assertEqual "for the numbers [1..10000]," testList lst'+    where+        testList = [1..10000] test_multipleWriters = do ms <- runResourceT $ mergeSources [ sourceList ([1..10]::[Integer])                                                             , sourceList ([11..20]::[Integer])                                                             ] 3