stm-conduit 2.1.3 → 2.1.4

raw patch · 3 files changed

+41/−1 lines, 3 files

Files

• Data/Conduit/Async.hs +29/−0
• stm-conduit.cabal +1/−1
• test/Test.hs +11/−0

@@ -10,6 +10,7 @@ module Data.Conduit.Async ( buffer                           , ($$&)                           , gatherFrom+                          , drainTo                           ) where  import Control.Applicative@@ -83,3 +84,31 @@             Just (Left e)  -> liftIO $ throwIO (e :: SomeException)             Just (Right r) -> return r             Nothing        -> gather worker chan++-- | Drain input values into an asynchronous action in the base monad via a+--   bounded 'TBQueue'.  This is effectively the dual of 'gatherFrom'.+drainTo :: (MonadIO m, MonadBaseControl IO m)+        => Int                        -- ^ Size of the queue to create+        -> (TBQueue (Maybe i) -> m r)  -- ^ Action to consume input values+        -> Consumer i m r+drainTo size gather = do+    chan   <- liftIO $ newTBQueueIO size+    worker <- lift $ async (gather chan)+    lift . restoreM =<< scatter worker chan+  where+    scatter worker chan = do+        mval <- await+        (mx, action) <- liftIO $ atomically $ do+            mres <- pollSTM worker+            case mres of+                Just (Left e)  ->+                    return (Nothing, liftIO $ throwIO (e :: SomeException))+                Just (Right r) ->+                    return (Just r, return ())+                Nothing        -> do+                    writeTBQueue chan mval+                    return (Nothing, return ())+        action+        case mx of+            Just x  -> return x+            Nothing -> scatter worker chan

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

@@ -32,6 +32,7 @@         testGroup "Async functions" [                   testCase "buffer" test_buffer                 , testCase "gatherFrom" test_gatherFrom+                , testCase "drainTo" test_drainTo             ],         testGroup "Bug fixes" [                   testCase "multipleWriters" test_multipleWriters@@ -91,3 +92,13 @@         f q x = do             atomically $ writeTBQueue q x             return q++test_drainTo = do+    sum' <- CL.sourceList [1..100] $$ drainTo 128 (go 0)+    assertEqual "sum computed using drainTo" sum' 5050+  where+    go acc queue = do+        mres <- atomically $ readTBQueue queue+        case mres of+            Nothing  -> return acc+            Just res -> go (acc + res) queue