pipes-concurrency 2.0.1 → 2.0.2
raw patch · 5 files changed
+62/−17 lines, 5 filesdep ~pipesdep ~pipes-concurrency
Dependency ranges changed: pipes, pipes-concurrency
Files
- LICENSE +1/−1
- pipes-concurrency.cabal +5/−5
- src/Pipes/Concurrent.hs +20/−10
- src/Pipes/Concurrent/Tutorial.hs +33/−0
- tests/tests-main.hs +3/−1
LICENSE view
@@ -1,4 +1,4 @@-Copyright (c) 2013 Gabriel Gonzalez+Copyright (c) 2014 Gabriel Gonzalez All rights reserved. Redistribution and use in source and binary forms, with or without modification,
pipes-concurrency.cabal view
@@ -1,10 +1,10 @@ Name: pipes-concurrency-Version: 2.0.1+Version: 2.0.2 Cabal-Version: >=1.8.0.2 Build-Type: Simple License: BSD3 License-File: LICENSE-Copyright: 2013 Gabriel Gonzalez+Copyright: 2013, 2014 Gabriel Gonzalez Author: Gabriel Gonzalez Maintainer: Gabriel439@gmail.com Bug-Reports: https://github.com/Gabriel439/Haskell-Pipes-Concurrency-Library/issues@@ -32,7 +32,7 @@ Hs-Source-Dirs: src Build-Depends: base >= 4 && < 5 ,- pipes >= 4.0 && < 4.1,+ pipes >= 4.0 && < 4.2, stm >= 2.4 && < 2.5 Exposed-Modules: Pipes.Concurrent,@@ -45,7 +45,7 @@ HS-Source-Dirs: tests . Build-Depends: base >= 4 && < 5 ,- pipes >= 4.0.0 && < 4.1,- pipes-concurrency >= 2.0.0 && < 4.1,+ pipes >= 4.0.0 && < 4.2,+ pipes-concurrency >= 2.0.0 && < 2.1, stm >= 2.4 && < 2.5, async >= 2.0 && < 2.1
src/Pipes/Concurrent.hs view
@@ -1,10 +1,7 @@ -- | Asynchronous communication between pipes -{-# LANGUAGE CPP, RankNTypes#-}+{-# LANGUAGE RankNTypes, Trustworthy #-} -#if __GLASGOW_HASKELL__ >= 702-{-# LANGUAGE Trustworthy #-}-#endif {- 'unsafeIOToSTM' requires the Trustworthy annotation. I use 'unsafeIOToSTM' to touch IORefs to mark them as still alive. This@@ -47,7 +44,7 @@ ) where import Control.Applicative (- Alternative(empty, (<|>)), Applicative(pure, (<*>)), (<*), (<$>) )+ Alternative(empty, (<|>)), Applicative(pure, (*>), (<*>)), (<*), (<$>) ) import Control.Concurrent (forkIO) import Control.Concurrent.STM (atomically, STM) import qualified Control.Concurrent.STM as S@@ -171,19 +168,26 @@ -} spawn' :: Buffer a -> IO (Output a, Input a, STM ()) spawn' buffer = do- (read, write) <- case buffer of+ (write, read) <- case buffer of Bounded n -> do q <- S.newTBQueueIO n- return (S.readTBQueue q, S.writeTBQueue q)+ return (S.writeTBQueue q, S.readTBQueue q) Unbounded -> do q <- S.newTQueueIO- return (S.readTQueue q, S.writeTQueue q)+ return (S.writeTQueue q, S.readTQueue q) Single -> do m <- S.newEmptyTMVarIO- return (S.takeTMVar m, S.putTMVar m)+ return (S.putTMVar m, S.takeTMVar m) Latest a -> do t <- S.newTVarIO a- return (S.readTVar t, S.writeTVar t)+ return (S.writeTVar t, S.readTVar t)+ New -> do+ m <- S.newEmptyTMVarIO+ return (\x -> S.tryTakeTMVar m *> S.putTMVar m x, S.takeTMVar m)+ Newest n -> do+ q <- S.newTBQueueIO n+ let write x = S.writeTBQueue q x <|> (S.tryReadTBQueue q *> write x)+ return (write, S.readTBQueue q) sealed <- S.newTVarIO False let seal = S.writeTVar sealed True@@ -226,6 +230,12 @@ 'Latest' is never empty nor full. -} | Latest a+ {-| Like @Bounded@, but 'send' never fails (the buffer is never full).+ Instead, old elements are discard to make room for new elements+ -}+ | Newest Int+ -- | Like @Newest 1@, but more efficient+ | New {- $reexport @Control.Concurrent@ re-exports 'forkIO', although I recommend using the
src/Pipes/Concurrent/Tutorial.hs view
@@ -572,6 +572,39 @@ 'recv' never removes the value from the mailbox. A 'Latest' mailbox is also never full because 'send' always succeeds, overwriting the previously stored value.++ Another alternative is to use the 'Newest' mailbox, which is like a+ 'Bounded' mailbox, except 'send' never blocks (the mailbox is never full).+ Instead, if there is no room 'send' will remove the oldest message from the+ mailbox to make room for a new message.++ The 'New' mailbox is like the 'Newest' mailbox, except optimized for the+ special case where you want to store a single message. You can use 'New' to+ read from a source that might potentially update rapidly, but still sleep if+ the source has no new values:++> inputDevice :: Producer Integer IO ()+> inputDevice = do+> each [1..100] -- Rapid updates+> lift $ threadDelay 4000000 -- Source goes quiet for 4 seconds+> each [101..] -- More rapid updates+>+> main = do+> (output, input) <- spawn New+> ...++ When the source goes quiet, the 'Input' will now block and wait, and will+ never read the same value twice:++> $ ./peek+> 7+> 100+> <Longer pause>+> 16793+> 5239440+> 10474439+> $+ -} {- $callback
tests/tests-main.hs view
@@ -11,7 +11,7 @@ import System.Timeout defaultTimeout :: Int-defaultTimeout = 100000 -- 0.1 s+defaultTimeout = 200000 -- 0.2 s labelPrint :: (Show a) => String -> Consumer a IO r labelPrint label = forever $ do@@ -114,9 +114,11 @@ runTest (testSenderClose $ Bounded 7) "BoundedNotFilledSenderClose" runTest (testSenderClose Single) "SingleSenderClose" runTestExpectTimeout (testSenderCloseDelayedSend $ Latest 42) "LatestSenderClose"+ runTest (testSenderCloseDelayedSend New) "NewSenderClose" -- runTest (testReceiverClose Unbounded) "UnboundedReceiverClose" runTest (testReceiverClose $ Bounded 3) "BoundedFilledReceiverClose" runTest (testReceiverClose $ Bounded 7) "BoundedNotFilledReceiverClose" runTest (testReceiverClose Single) "SingleReceiverClose" runTest (testReceiverCloseDelayedReceive $ Latest 42) "LatestReceiverClose"+ runTest (testReceiverClose New) "NewReceiverClose"