diff --git a/Control/Proxy/Concurrent.hs b/Control/Proxy/Concurrent.hs
deleted file mode 100644
--- a/Control/Proxy/Concurrent.hs
+++ /dev/null
@@ -1,228 +0,0 @@
--- | Asynchronous communication between proxies
-
-{-# LANGUAGE CPP #-}
-
-#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
-    action satisfies the necessary safety requirements because:
-
-    * You can safely repeat it if the transaction rolls back
-
-    * It does not acquire any resources
-
-    * It does not leak any inconsistent view of memory to the outside world
-
-    It appears to be unnecessary to read the IORef to keep it from being garbage
-    collected, but I wanted to be absolutely certain since I cannot be sure that
-    GHC won't optimize away the reference to the IORef.
-
-    The other alternative was to make 'send' and 'recv' use the 'IO' monad
-    instead of 'STM', but I felt that it was important to preserve the ability
-    to combine them into larger transactions.
--}
-
-module Control.Proxy.Concurrent (
-    -- * Spawn mailboxes
-    spawn,
-    Buffer(..),
-    Input,
-    Output,
-
-    -- * Send and receive messages
-    send,
-    recv,
-
-    -- * Proxy utilities
-    sendD,
-    recvS,
-
-    -- * Re-exports
-    -- $reexport
-    module Control.Concurrent,
-    module Control.Concurrent.STM,
-    module System.Mem
-    ) where
-
-import Control.Applicative (
-    Alternative(empty, (<|>)), Applicative(pure, (<*>)), (<*), (<$>) )
-import Control.Concurrent (forkIO)
-import Control.Concurrent.STM (atomically, STM)
-import qualified Control.Concurrent.STM as S
-import qualified Control.Proxy as P
-import Data.IORef (newIORef, readIORef, mkWeakIORef)
-import Data.Monoid (Monoid(mempty, mappend))
-import GHC.Conc.Sync (unsafeIOToSTM)
-import System.Mem (performGC)
-
-{-| Spawn a mailbox that has an 'Input' and 'Output' end, using the specified
-    'Buffer' to store messages
--}
-spawn :: Buffer a -> IO (Input a, Output a)
-spawn buffer = do
-    (read, write) <- case buffer of
-        Bounded n -> do
-            q <- S.newTBQueueIO n
-            return (S.readTBQueue q, S.writeTBQueue q)
-        Unbounded -> do
-            q <- S.newTQueueIO
-            return (S.readTQueue q, S.writeTQueue q)
-        Single    -> do
-            m <- S.newEmptyTMVarIO
-            return (S.takeTMVar m, S.putTMVar m)
-        Latest a  -> do
-            t <- S.newTVarIO a
-            return (S.readTVar t, S.writeTVar t)
-
-    {- Use an IORef to keep track of whether the 'Input' end has been garbage
-       collected and run a finalizer when the collection occurs
-    -}
-    rSend    <- newIORef ()
-    doneSend <- S.newTVarIO False
-    mkWeakIORef rSend (S.atomically $ S.writeTVar doneSend True)
-
-    {- Use an IORef to keep track of whether the 'Output' end has been garbage
-       collected and run a finalizer when the collection occurs
-    -}
-    rRecv    <- newIORef ()
-    doneRecv <- S.newTVarIO False
-    mkWeakIORef rRecv (S.atomically $ S.writeTVar doneRecv True)
-
-    let sendOrEnd a = do
-            b <- S.readTVar doneRecv
-            if b
-                then return False
-                else do
-                    write a
-                    return True
-        {- The '_send' action aborts without writing a value to the 'Buffer' if
-           the 'Output' has been garbage collected, since there is no point
-           wasting memory if nothing can empty the mailbox.  This protects
-           against careless users not checking send's return value, especially
-           if they use a mailbox of 'Unbounded' size.
-        -}
-        readOrEnd = (Just <$> read) <|> (do
-            b <- S.readTVar doneSend
-            S.check b
-            return Nothing )
-        _send a = sendOrEnd a <* unsafeIOToSTM (readIORef rSend)
-        _recv   = readOrEnd   <* unsafeIOToSTM (readIORef rRecv)
-    return (Input _send, Output _recv)
-{-# INLINABLE spawn #-}
-
-{-| 'Buffer' specifies how to store messages sent to the 'Input' end until the
-    'Output' receives them.
--}
-data Buffer a
-    -- | Store an 'Unbounded' number of messages in a FIFO queue
-    = Unbounded
-    -- | Store a 'Bounded' number of messages, specified by the 'Int' argument
-    | Bounded Int
-    -- | Store a 'Single' message (like @Bounded 1@, but more efficient)
-    | Single
-    {-| Store the 'Latest' message, beginning with an initial value
-
-        'Latest' is never empty nor full.
-    -}
-    | Latest a
-
--- | Accepts messages for the mailbox
-newtype Input a = Input {
-    {-| Send a message to the mailbox
-
-        * Fails and returns 'False' if the mailbox's 'Output' has been garbage
-          collected (even if the mailbox is not full), otherwise it:
-
-        * Retries if the mailbox is full, or:
-
-        * Succeeds if the mailbox is not full and returns 'True'.
-    -}
-    send :: a -> S.STM Bool }
-
-instance Monoid (Input a) where
-    mempty  = Input (\_ -> return False)
-    mappend i1 i2 = Input (\a -> (||) <$> send i1 a <*> send i2 a)
-
--- | Retrieves messages from the mailbox
-newtype Output a = Output {
-    {-| Receive a message from the mailbox
-
-        * Succeeds and returns a 'Just' if the mailbox is not empty, otherwise
-          it:
-
-        * Retries if mailbox's 'Input' has not been garbage collected, or:
-
-        * Fails if the mailbox's 'Input' has been garbage collected and returns
-          'Nothing'.
-    -}
-    recv :: S.STM (Maybe a) }
-
-instance Functor Output where
-    fmap f m = Output (fmap (fmap f) (recv m))
-
-instance Applicative Output where
-    pure r    = Output (pure (pure r))
-    mf <*> mx = Output ((<*>) <$> recv mf <*> recv mx)
-
-instance Monad Output where
-    return r = Output (return (return r))
-    m >>= f  = Output $ do
-        ma <- recv m
-        case ma of
-            Nothing -> return Nothing
-            Just a  -> recv (f a)
-
--- Deriving 'Alternative'
-instance Alternative Output where
-    empty   = Output empty
-    x <|> y = Output (recv x <|> recv y)
-
-{-| Writes all messages flowing \'@D@\'ownstream to the given 'Input'
-
-    'sendD' terminates when the corresponding 'Output' is garbage collected.
-
-> sendD :: (Proxy p) => Input a -> () -> Pipe p a a IO ()
--}
-sendD :: (P.Proxy p) => Input a -> x -> p x a x a IO ()
-sendD input = P.runIdentityK loop
-  where
-    loop x = do
-        a <- P.request x
-        alive <- P.lift $ S.atomically $ send input a
-        if alive
-            then do
-                x2 <- P.respond a
-                loop x2
-            else return ()
-{-# INLINABLE sendD #-}
-
-{-| Convert an 'Output' to a 'P.Producer'
-
-    'recvS' terminates when the 'Buffer' is empty and the corresponding 'Input'
-    is garbage collected.
-
-> recvS :: (Proxy p) => Output a -> () -> Producer p a IO ()
--}
-recvS :: (P.Proxy p) => Output a -> r -> p x' x y' a IO r
-recvS output r = P.runIdentityP go
-  where
-    go = do
-        ma <- P.lift $ S.atomically $ recv output
-        case ma of
-            Nothing -> return r
-            Just a  -> do
-                P.respond a
-                go
-{-# INLINABLE recvS #-}
-
-{- $reexport
-    @Control.Concurrent@ re-exports 'forkIO', although I recommend using the
-    @async@ library instead.
-
-    @Control.Concurrent.STM@ re-exports 'atomically' and 'STM'.
-
-    @System.Mem@ re-exports 'performGC'.
--}
diff --git a/Control/Proxy/Concurrent/Tutorial.hs b/Control/Proxy/Concurrent/Tutorial.hs
deleted file mode 100644
--- a/Control/Proxy/Concurrent/Tutorial.hs
+++ /dev/null
@@ -1,772 +0,0 @@
-{-| This module provides a tutorial for the @pipes-concurrency@ library.
-
-    This tutorial assumes that you have read the @pipes@ tutorial in
-    @Control.Proxy.Tutorial@.
-
-    I've condensed all the code examples into self-contained code listings in
-    the Appendix section that you can use to follow along.
--}
-
-module Control.Proxy.Concurrent.Tutorial (
-    -- * Introduction
-    -- $intro
-
-    -- * Work Stealing
-    -- $steal
-
-    -- * Termination
-    -- $termination
-
-    -- * Mailbox Sizes
-    -- $mailbox
-
-    -- * Broadcasts
-    -- $broadcast
-
-    -- * Updates
-    -- $updates
-
-    -- * Callbacks
-    -- $callback
-
-    -- * Safety
-    -- $safety
-
-    -- * Conclusion
-    -- $conclusion
-
-    -- * Appendix
-    -- $appendix
-    ) where
-
-import Control.Proxy
-import Control.Proxy.Concurrent
-import Data.Monoid
-
-{- $intro
-    The @pipes-concurrency@ library provides a simple interface for
-    communicating between concurrent pipelines.  Use this library if you want
-    to:
-
-    * merge multiple streams into a single stream,
-
-    * stream data from a callback \/ continuation,
-
-    * broadcast data,
-
-    * build a work-stealing setup, or
-
-    * implement basic functional reactive programming (FRP).
-
-    For example, let's say that we design a simple game with a single unit's
-    health as the global state.  We'll define an event handler that modifies the
-    unit's health in response to events:
-
-> import Control.Monad
-> import Control.Proxy
-> import Control.Proxy.Trans.Maybe
-> import Control.Proxy.Trans.State
-> 
-> -- The game events
-> data Event = Harm Integer | Heal Integer | Quit
-> 
-> -- The game state
-> type Health = Integer
-> 
-> handler :: (Proxy p) => () -> Consumer (StateP Health (MaybeP p)) Event IO r
-> handler () = forever $ do
->     event <- request ()
->     case event of
->         Harm n -> modify (subtract n)
->         Heal n -> modify (+        n)
->         Quit   -> mzero
->     health <- get
->     lift $ putStrLn $ "Health = " ++ show health
-
-    However, we have two concurrent event sources that we wish to hook up to our
-    event handler.  One translates user input to game events:
-
-> user :: (Proxy p) => () -> Producer p Event IO r
-> user () = runIdentityP $ forever $ do
->     command <- lift getLine
->     case command of
->         "potion" -> respond (Heal 10)
->         "quit"   -> respond  Quit
->         _        -> lift $ putStrLn "Invalid command"
-
-    ... while the other creates inclement weather:
-
-> import Control.Concurrent
->
-> acidRain :: (Proxy p) => () -> Producer p Event IO r
-> acidRain () = runIdentityP $ forever $ do
->     respond (Harm 1)
->     lift $ threadDelay 2000000
-
-    To merge these sources, we 'spawn' a new FIFO mailbox which we will use to
-    merge the two streams of asynchronous events:
-
-> spawn :: Buffer a -> IO (Input a, Output a)
-
-    'spawn' takes a mailbox 'Buffer' as an argument, and we will specify that we
-    want our mailbox to store an 'Unbounded' number of messages:
-
-> import Control.Proxy.Concurrent
->
-> main = do
->     (input, output) <- spawn Unbounded
->     ...
-
-   'spawn' creates this mailbox in the background and then returns two values:
-
-    * an @(Input a)@ that we use to add messages of type @a@ to the mailbox
-
-    * an @(Output a)@ that we use to consume messages of type @a@ from the
-      mailbox
-
-    We will be streaming @Event@s through our mailbox, so our @input@ has type
-    @(Input Event)@ and our @output@ has type @(Output Event)@.
-
-    To stream @Event@s into the mailbox , we use 'sendD', which writes values to
-    the mailbox's 'Input' end:
-
-> sendD :: (Proxy p) => Input a -> () -> Pipe p a a IO ()
-
-    We can concurrently forward multiple streams to the same 'Input', which
-    asynchronously merges their messages into the same mailbox:
-
->     ...
->     forkIO $ do runProxy $ acidRain >-> sendD input
->                 performGC  -- I'll explain 'performGC' below
->     forkIO $ do runProxy $ user     >-> sendD input
->                 performGC
->     ...
-
-    To stream @Event@s out of the mailbox, we use 'recvS', which reads values
-    from the mailbox's 'Output' end:
-
-> recvS :: (Proxy p) => Output a -> () -> Producer p a IO ()
-
-    We will forward our merged stream to our @handler@ so that it can listen to
-    both @Event@ sources:
-
->     ...
->     runProxy $ runMaybeK $ evalStateK 100 $ recvS output >-> handler
-
-    Our final @main@ becomes:
-
-> main = do
->     (input, output) <- spawn Unbounded
->     forkIO $ do runProxy $ acidRain >-> sendD input
->                 performGC
->     forkIO $ do runProxy $ user     >-> sendD input
->                 performGC
->     runProxy $ runMaybeK $ evalStateK 100 $ recvS output >-> handler
-
-    ... and when we run it we get the desired concurrent behavior:
-
-> $ ./game
-> Health = 99
-> Health = 98
-> potion<Enter>
-> Health = 108
-> Health = 107
-> Health = 106
-> potion<Enter>
-> Health = 116
-> Health = 115
-> quit<Enter>
-> $
--}
-
-{- $steal
-    You can also have multiple pipes reading from the same mailbox.  Messages
-    get split between listening pipes on a first-come first-serve basis.
-
-    For example, we'll define a \"worker\" that takes a one-second break each
-    time it receives a new job:
-
-> import Control.Concurrent
-> import Control.Monad
-> import Control.Proxy
-> 
-> worker :: (Proxy p, Show a) => Int -> () -> Consumer p a IO r
-> worker i () = runIdentityP $ forever $ do
->     a <- request ()
->     lift $ threadDelay 1000000  -- 1 second
->     lift $ putStrLn $ "Worker #" ++ show i ++ ": Processed " ++ show a
-
-    Fortunately, these workers are cheap, so we can assign several of them to
-    the same job:
-
-> import Control.Concurrent.Async
-> import Control.Proxy.Concurrent
-> 
-> main = do
->     (input, output) <- spawn Unbounded
->     as <- forM [1..3] $ \i ->
->           async $ do runProxy $ recvS output >-> worker i
->                      performGC
->     a  <- async $ do runProxy $ fromListS [1..10] >-> sendD input
->                      performGC
->     mapM_ wait (a:as)
-
-    The above example uses @Control.Concurrent.Async@ from the @async@ package
-    to fork each thread and wait for all of them to terminate:
-
-> $ ./work
-> Worker #2: Processed 3
-> Worker #1: Processed 2
-> Worker #3: Processed 1
-> Worker #3: Processed 6
-> Worker #1: Processed 5
-> Worker #2: Processed 4
-> Worker #2: Processed 9
-> Worker #1: Processed 8
-> Worker #3: Processed 7
-> Worker #2: Processed 10
-> $
-
-    What if we replace 'fromListS' with a different source that reads lines from
-    user input until the user types \"quit\":
-
-> user :: (Proxy p) => () -> Producer p String IO ()
-> user = stdinS >-> takeWhileD (/= "quit")
-> 
-> main = do
->     (input, output) <- spawn Unbounded
->     as <- forM [1..3] $ \i ->
->           async $ do runProxy $ recvS output >-> worker i
->                      performGC
->     a  <- async $ do runProxy $ user >-> sendD input
->                      performGC
->     mapM_ wait (a:as)
-
-    This still produces the correct behavior:
-
-> $ ./work
-> Test<Enter>
-> Worker #1: Processed "Test"
-> Apple<Enter>
-> Worker #2: Processed "Apple"
-> 42<Enter>
-> Worker #3: Processed "42"
-> A<Enter>
-> B<Enter>
-> C<Enter>
-> Worker #1: Processed "A"
-> Worker #2: Processed "B"
-> Worker #3: Processed "C"
-> quit<Enter>
-> $
--}
-
-{- $termination
-
-    Wait...  How do the workers know when to stop listening for data?  After
-    all, anything that has a reference to 'Input' could potentially add more
-    data to the mailbox.
-
-    It turns out that 'recvS' is smart and only terminates when the upstream
-    'Input' is garbage collected.  'recvS' builds on top of the more primitive
-    'recv' command, which returns a 'Nothing' when the 'Input' is garbage
-    collected:
-
-> recv :: Output a -> STM (Maybe a)
-
-    Otherwise, 'recv' blocks if the mailbox is empty since it assumes that if
-    the 'Input' has not been garbage collected then somebody might still produce
-    more data.
-
-    Does it work the other way around?  What happens if the workers go on strike
-    before processing the entire data set?
-
->     ...
->     as <- forM [1..3] $ \i ->
->           -- Each worker refuses to process more than two values
->           async $ do runProxy $ recvS output >-> takeB_ 2 >-> worker i
->                      performGC
->     ...
-
-    Let's find out:
-
-> $ ./work
-> How<Enter>
-> Worker #1: Processed "How"
-> many<Enter>
-> roads<Enter>
-> Worker #2: Processed "many"
-> Worker #3: Processed "roads"
-> must<Enter>
-> a<Enter>
-> man<Enter>
-> Worker #1: Processed "must"
-> Worker #2: Processed "a"
-> Worker #3: Processed "man"
-> walk<Enter>
-> $
-
-    'sendD' similarly shuts down when the 'Output' is garbage collected,
-    preventing the user from submitting new values.  'sendD' builds on top of
-    the more primitive 'send' command, which returns a 'False' when the 'Output'
-    is garbage collected:
-
-> send :: Input a -> a -> STM Bool
-
-    Otherwise, 'send' blocks if the mailbox is full, since it assumes that if
-    the 'Output' has not been garbage collected then somebody could still
-    consume a value from the mailbox, making room for a new value.
-
-    This is why we have to insert 'performGC' calls whenever we release a
-    reference to either the 'Input' or 'Output'.  Without these calls we cannot
-    guarantee that the garbage collector will trigger and notify the opposing
-    end if the last reference was released.  If you forget to insert a
-    'performGC' call then termination will delay until the next garbage
-    collection cycle.
--}
-
-{- $mailbox
-    So far we haven't observed 'send' blocking because we only 'spawn'ed
-    'Unbounded' mailboxes.  However, we can control the size of the mailbox to
-    tune the coupling between the 'Input' and the 'Output' ends.
-
-    If we set the mailbox 'Buffer' to 'Single', then the mailbox holds exactly
-    one message, forcing synchronization between 'send's and 'recv's.  Let's
-    observe this by sending an infinite stream of values, logging all values to
-    'stdout':
-
-> main = do
->     (input, output) <- spawn Single
->     as <- forM [1..3] $ \i ->
->           async $ do runProxy $ recvS output >-> takeB_ 2 >-> worker i
->                      performGC
->     a  <- async $ do runProxy $ enumFromS 1 >-> printD >-> sendD input
->                      performGC
->     mapM_ wait (a:as)
-
-    The 7th value gets stuck in the mailbox, and the 8th value blocks because
-    the mailbox never clears the 7th value:
-
-> $ ./work
-> 1
-> 2
-> 3
-> 4
-> 5
-> Worker #3: Processed 3
-> Worker #2: Processed 2
-> Worker #1: Processed 1
-> 6
-> 7
-> 8
-> Worker #1: Processed 6
-> Worker #2: Processed 5
-> Worker #3: Processed 4
-> $
-
-    Contrast this with an 'Unbounded' mailbox for the same program, which keeps
-    accepting values until downstream finishes processing the first six values:
-
-> $ ./work
-> 1
-> 2
-> 3
-> 4
-> 5
-> 6
-> 7
-> 8
-> 9
-> ...
-> 487887
-> 487888
-> Worker #3: Processed 3
-> Worker #2: Processed 2
-> Worker #1: Processed 1
-> 487889
-> 487890
-> ...
-> 969188
-> 969189
-> Worker #1: Processed 6
-> Worker #2: Processed 5
-> Worker #3: Processed 4
-> 969190
-> 969191
-> $
-
-    You can also choose something in between by using a 'Bounded' mailbox which
-    caps the mailbox size to a fixed value.  Use 'Bounded' when you want mostly
-    loose coupling but still want to guarantee bounded memory usage:
-
-> main = do
->     (input, output) <- spawn (Bounded 100)
->     ...
-
-> $ ./work
-> ...
-> 103
-> 104
-> Worker #3: Processed 3
-> Worker #2: Processed 2
-> Worker #1: Processed 1
-> 105
-> 106
-> 107
-> Worker #1: Processed 6
-> Worker #2: Processed 5
-> Worker #3: Processed 4
-> $
--}
-
-{- $broadcast
-    You can also broadcast data to multiple listeners instead of dividing up the
-    data.  Just use the 'Monoid' instance for 'Input' to combine multiple
-    'Input' ends together into a single broadcast 'Input':
-
-> import Control.Monad
-> import Control.Concurrent.Async
-> import Control.Proxy
-> import Control.Proxy.Concurrent
-> import Data.Monoid
-> 
-> main = do
->     (input1, output1) <- spawn Unbounded
->     (input2, output2) <- spawn Unbounded
->     a1 <- async $ do
->         runProxy $ stdinS >-> sendD (input1 <> input2)
->         performGC
->     as <- forM [output1, output2] $ \output -> async $ do
->         runProxy $ recvS output >-> takeB_ 2 >-> stdoutD
->         performGC
->     mapM_ wait (a1:as)
-
-    In the above example, 'stdinS' will broadcast user input to both mailboxes,
-    and each mailbox forwards its values to 'stdoutD', echoing the message to
-    standard output:
-
-> $ ./broadcast
-> ABC<Enter>
-> ABC
-> ABC
-> DEF<Enter>
-> DEF
-> DEF
-> GHI<Enter>
-> $ 
-
-    The combined 'Input' stays alive as long as any of the original 'Input's
-    remains alive.  In the above example, 'sendD' terminates on the third 'send'
-    attempt because it detects that both listeners died after receiving two
-    messages.
-
-    Use 'mconcat' to broadcast to a list of 'Input's, but keep in mind that you
-    will incur a performance price if you combine thousands of 'Input's or more
-    because they will create a very large 'STM' transaction.  You can improve
-    performance for very large broadcasts if you sacrifice atomicity and
-    manually combine multiple 'send' actions in 'IO' instead of 'STM'.
--}
-
-{- $updates
-    Sometimes you don't want to handle every single event.  For example, you
-    might have an input and output device (like a mouse and a monitor) where the
-    input device updates at a different pace than the output device
-
-> import Control.Concurrent
-> import Control.Proxy
-> 
-> -- Fast input updates
-> inputDevice :: (Monad m, Proxy p) => () -> Producer p Integer m r
-> inputDevice = enumFromS 1
-> 
-> -- Slow output updates
-> outputDevice :: (Proxy p) => () -> Consumer p Integer IO r
-> outputDevice () = runIdentityP $ forever $ do
->     n <- request ()
->     lift $ do
->         print n
->         threadDelay 1000000
-
-    In this scenario you don't want to enforce a one-to-one correspondence
-    between input device updates and output device updates because you don't
-    want either end to block waiting for the other end.  Instead, you just need
-    the output device to consult the 'Latest' value received from the 'Input':
-
-> import Control.Concurrent.Async
-> import Control.Proxy.Concurrent
->
-> main = do
->     (input, output) <- spawn (Latest 0)
->     a1 <- async $ do
->         runProxy $ inputDevice >-> sendD input
->         performGC
->     a2 <- async $ do
->         runProxy $ recvS output >-> takeB_ 5 >-> outputDevice
->         performGC
->     mapM_ wait [a1, a2]
-
-    'Latest' selects a mailbox that always stores exactly one value.  The
-    'Latest' constructor takes a single argument (@0@, in the above example)
-    specifying the starting value to store in the mailbox.  'send' overrides the
-    currently stored value and 'recv' peeks at the latest stored value without
-    consuming it.  In the above example the @outputDevice@ periodically peeks at    the latest value stashed inside the mailbox:
-
-> $ ./peek
-> 5
-> 752452
-> 1502636
-> 2248278
-> 2997705
-> $
-
-    A 'Latest' mailbox is never empty because it begins with a default value and
-    'recv' never removes the value from the mailbox.  A 'Latest' mailbox is also
-    never full because 'send' always succeeds, overwriting the previously stored
-    value.
--}
-
-{- $callback
-    @pipes-concurrency@ also solves the common problem of getting data out of a
-    callback-based framework into @pipes@.
-
-    For example, suppose that we have the following callback-based function:
-
-> import Control.Monad
-> 
-> onLines :: (String -> IO a) -> IO b
-> onLines callback = forever $ do
->     str <- getLine
->     callback str
-
-    We can use 'send' to free the data from the callback and then we can
-    retrieve the data on the outside using 'recvS':
-
-> import Control.Proxy
-> import Control.Proxy.Concurrent
-> 
-> onLines' :: (Proxy p) => () -> Producer p String IO ()
-> onLines' () = runIdentityP $ do
->     (input, output) <- lift $ spawn Single
->     lift $ forkIO $ onLines (\str -> atomically $ send input str)
->     recvS output ()
-> 
-> main = runProxy $ onLines' >-> takeWhileD (/= "quit") >-> stdoutD
-
-    Now we can stream from the callback as if it were an ordinary 'Producer':
-
-> $ ./callback
-> Test<Enter>
-> Test
-> Apple<Enter>
-> Apple
-> quit<Enter>
-> $
-
--}
-
-{- $safety
-    @pipes-concurrency@ avoids deadlocks because 'send' and 'recv' always
-    cleanly return before triggering a deadlock.  This behavior works even in
-    complicated scenarios like:
-
-    * cyclic graphs of connected mailboxes,
-
-    * multiple readers and multiple writers to the same mailbox, and
-
-    * dynamically adding or garbage collecting mailboxes.
-
-    The following example shows how @pipes-concurrency@ will do the right thing
-    even in the case of cycles:
-
-> import Control.Concurrent.Async
-> import Control.Proxy
-> import Control.Proxy.Concurrent
-> 
-> main = do
->     (in1, out1) <- spawn Unbounded
->     (in2, out2) <- spawn Unbounded
->     a1 <- async $ do runProxy $ (fromListS [1,2] >=> recvS out1) >-> sendD in2
->                      performGC
->     a2 <- async $ do runProxy $ recvS out2 >-> printD >-> takeB_ 6 >-> sendD in1
->                      performGC
->     mapM_ wait [a1, a2]
-
-    The above program jump-starts a cyclic chain with two input values and
-    terminates one branch of the cycle after six values flow through.  Both
-    branches correctly terminate and get garbage collected without triggering
-    deadlocks when 'takeB_' finishes:
-
-> $ ./cycle
-> 1
-> 2
-> 1
-> 2
-> 1
-> 2
-> $
-
--}
-
-{- $conclusion
-    @pipes-concurrency@ adds an asynchronous dimension to @pipes@.  This
-    promotes a natural division of labor for concurrent programs:
-
-    * Fork one pipeline per deterministic behavior
-
-    * Communicate between concurrent pipelines using @pipes-concurrency@
-
-    This promotes an actor-style approach to concurrent programming where
-    pipelines behave like processes and mailboxes behave like ... mailboxes.
-
-    You can ask questions about @pipes-concurrency@ and other @pipes@ libraries
-    on the official @pipes@ mailing list at
-    <mailto:haskell-pipes@googlegroups.com>.
--}
-
-{- $appendix
-    I've provided the full code for the above examples here so you can easily
-    try them out:
-
-> -- game.hs
->
-> import Control.Concurrent
-> import Control.Monad
-> import Control.Proxy
-> import Control.Proxy.Concurrent
-> import Control.Proxy.Trans.Maybe
-> import Control.Proxy.Trans.State
-> 
-> -- The game events
-> data Event = Harm Integer | Heal Integer | Quit
-> 
-> -- The game state
-> type Health = Integer
-> 
-> handler :: (Proxy p) => () -> Consumer (StateP Health (MaybeP p)) Event IO r
-> handler () = forever $ do
->     event <- request ()
->     case event of
->         Harm n -> modify (subtract n)
->         Heal n -> modify (+        n)
->         Quit   -> mzero
->     health <- get
->     lift $ putStrLn $ "Health = " ++ show health
->
-> user :: (Proxy p) => () -> Producer p Event IO r
-> user () = runIdentityP $ forever $ do
->     command <- lift getLine
->     case command of
->         "potion" -> respond (Heal 10)
->         "quit"   -> respond  Quit
->         _        -> lift $ putStrLn "Invalid command"
->
-> acidRain :: (Proxy p) => () -> Producer p Event IO r
-> acidRain () = runIdentityP $ forever $ do
->     respond (Harm 1)
->     lift $ threadDelay 2000000
->
-> main = do
->     (input, output) <- spawn Unbounded
->     forkIO $ do runProxy $ acidRain >-> sendD input
->                 performGC  -- I'll explain 'performGC' below
->     forkIO $ do runProxy $ user     >-> sendD input
->                 performGC
->     runProxy $ runMaybeK $ evalStateK 100 $ recvS output >-> handler
-
-> -- work.hs
-> 
-> import Control.Concurrent
-> import Control.Monad
-> import Control.Proxy
-> import Control.Concurrent.Async
-> import Control.Proxy.Concurrent
-> 
-> worker :: (Proxy p, Show a) => Int -> () -> Consumer p a IO r
-> worker i () = runIdentityP $ forever $ do
->     a <- request ()
->     lift $ threadDelay 1000000  -- 1 second
->     lift $ putStrLn $ "Worker #" ++ show i ++ ": Processed " ++ show a
-> {-
-> worker :: (Proxy p, Show a) => Int -> () -> Consumer p a IO ()
-> worker i () = runIdentityP $ replicateM_ 2 $ do
->     a <- request ()
->     lift $ threadDelay 1000000
->     lift $ putStrLn $ "Worker #" ++ show i ++ ": Processed " ++ show a
-> -}
->
-> user :: (Proxy p) => () -> Producer p String IO ()
-> user = stdinS >-> takeWhileD (/= "quit")
-> 
-> main = do
->     (input, output) <- spawn Unbounded
-> --  (input, output) <- spawn Single
-> --  (input, output) <- spawn (Bounded 100)
->     as <- forM [1..3] $ \i ->
->           async $ do runProxy $ recvS output >-> worker i
-> --        async $ do runProxy $ recvS output >-> takeB_ 2 >-> worker i
->                      performGC
->     a  <- async $ do runProxy $ fromListS [1..10]      >-> sendD input
-> --  a  <- async $ do runProxy $ user                   >-> sendD input
-> --  a  <- async $ do runProxy $ enumFromS 1 >-> printD >-> sendD input
->                      performGC
->     mapM_ wait (a:as)
-
-> -- broadcast.hs
->
-> import Control.Monad
-> import Control.Concurrent.Async
-> import Control.Proxy
-> import Control.Proxy.Concurrent
-> import Data.Monoid
-> 
-> main = do
->     (input1, output1) <- spawn Unbounded
->     (input2, output2) <- spawn Unbounded
->     a1 <- async $ do
->         runProxy $ stdinS >-> sendD (input1 <> input2)
->         performGC
->     as <- forM [output1, output2] $ \output -> async $ do
->         runProxy $ recvS output >-> takeB_ 2 >-> stdoutD
->         performGC
->     mapM_ wait (a1:as)
-
-> -- peek.hs
-> 
-> import Control.Concurrent
-> import Control.Concurrent.Async
-> import Control.Proxy
-> import Control.Proxy.Concurrent
-> 
-> inputDevice :: (Monad m, Proxy p) => () -> Producer p Integer m r
-> inputDevice = enumFromS 1
-> 
-> outputDevice :: (Proxy p) => () -> Consumer p Integer IO r
-> outputDevice () = runIdentityP $ forever $ do
->     n <- request ()
->     lift $ do
->         print n
->         threadDelay 1000000
->
-> main = do
->     (input, output) <- spawn (Latest 0)
->     a1 <- async $ do
->         runProxy $ inputDevice >-> sendD input
->         performGC
->     a2 <- async $ do
->         runProxy $ recvS output >-> takeB_ 5 >-> outputDevice
->         performGC
->     mapM_ wait [a1, a2]
-
-> -- callback.hs
-> 
-> import Control.Proxy
-> import Control.Proxy.Concurrent
-> 
-> onLines' :: (Proxy p) => () -> Producer p String IO ()
-> onLines' () = runIdentityP $ do
->     (input, output) <- lift $ spawn Single
->     lift $ forkIO $ onLines (\str -> atomically $ send input str)
->     recvS output ()
-> 
-> main = runProxy $ onLines' >-> takeWhileD (/= "quit) >-> stdoutD
--}
diff --git a/LICENSE b/LICENSE
--- a/LICENSE
+++ b/LICENSE
@@ -1,24 +1,24 @@
-Copyright (c) 2013 Gabriel Gonzalez
-All rights reserved.
-
-Redistribution and use in source and binary forms, with or without modification,
-are permitted provided that the following conditions are met:
-    * Redistributions of source code must retain the above copyright notice,
-      this list of conditions and the following disclaimer.
-    * Redistributions in binary form must reproduce the above copyright notice,
-      this list of conditions and the following disclaimer in the documentation
-      and/or other materials provided with the distribution.
-    * Neither the name of Gabriel Gonzalez nor the names of other contributors
-      may be used to endorse or promote products derived from this software
-      without specific prior written permission.
-
-THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
-ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
-WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
-DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
-ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
-(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
-LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
-ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
-(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
-SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+Copyright (c) 2013 Gabriel Gonzalez
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without modification,
+are permitted provided that the following conditions are met:
+    * Redistributions of source code must retain the above copyright notice,
+      this list of conditions and the following disclaimer.
+    * Redistributions in binary form must reproduce the above copyright notice,
+      this list of conditions and the following disclaimer in the documentation
+      and/or other materials provided with the distribution.
+    * Neither the name of Gabriel Gonzalez nor the names of other contributors
+      may be used to endorse or promote products derived from this software
+      without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
+ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
+ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
diff --git a/Setup.hs b/Setup.hs
--- a/Setup.hs
+++ b/Setup.hs
@@ -1,2 +1,2 @@
-import Distribution.Simple
-main = defaultMain
+import Distribution.Simple
+main = defaultMain
diff --git a/pipes-concurrency.cabal b/pipes-concurrency.cabal
--- a/pipes-concurrency.cabal
+++ b/pipes-concurrency.cabal
@@ -1,39 +1,51 @@
-Name: pipes-concurrency
-Version: 1.2.1
-Cabal-Version: >=1.8.0.2
-Build-Type: Simple
-License: BSD3
-License-File: LICENSE
-Copyright: 2013 Gabriel Gonzalez
-Author: Gabriel Gonzalez
-Maintainer: Gabriel439@gmail.com
-Bug-Reports: https://github.com/Gabriel439/Haskell-Pipes-Concurrency-Library/issues
-Synopsis: Concurrency for the pipes ecosystem
-Description: This library provides light-weight concurrency primitives for
-  pipes, with the following features:
-  .
-  * /Simple API/: Use only five functions
-  .
-  * /Deadlock Safety/: Automatically avoid concurrency deadlocks
-  .
-  * /Flexibility/: Build many-to-many and cyclic communication topologies
-  .
-  * /Dynamic Graphs/: Add or remove readers and writers at any time
-  .
-  Import "Control.Proxy.Concurrent" to use the library.
-  .
-  Read "Control.Proxy.Concurrent.Tutorial" for an tutorial.
-Category: Control, Pipes, Proxies, Concurrency
-Source-Repository head
-    Type: git
-    Location: https://github.com/Gabriel439/Haskell-Pipes-Concurrency-Library
-
-Library
-    Build-Depends:
-        base         >= 4       && < 5  ,
-        pipes        >= 3.0     && < 3.4,
-        stm          >= 2.4     && < 2.5
-    Exposed-Modules:
-        Control.Proxy.Concurrent,
-        Control.Proxy.Concurrent.Tutorial
-    GHC-Options: -O2
+Name: pipes-concurrency
+Version: 2.0.0
+Cabal-Version: >=1.8.0.2
+Build-Type: Simple
+License: BSD3
+License-File: LICENSE
+Copyright: 2013 Gabriel Gonzalez
+Author: Gabriel Gonzalez
+Maintainer: Gabriel439@gmail.com
+Bug-Reports: https://github.com/Gabriel439/Haskell-Pipes-Concurrency-Library/issues
+Synopsis: Concurrency for the pipes ecosystem
+Description: This library provides light-weight concurrency primitives for
+  pipes, with the following features:
+  .
+  * /Simple API/: Use only five functions
+  .
+  * /Deadlock Safety/: Automatically avoid concurrency deadlocks
+  .
+  * /Flexibility/: Build many-to-many and cyclic communication topologies
+  .
+  * /Dynamic Graphs/: Add or remove readers and writers at any time
+  .
+  Import "Pipes.Concurrent" to use the library.
+  .
+  Read "Pipes.Concurrent.Tutorial" for a tutorial.
+Category: Control, Pipes, Concurrency
+Source-Repository head
+    Type: git
+    Location: https://github.com/Gabriel439/Haskell-Pipes-Concurrency-Library
+
+Library
+    Hs-Source-Dirs: src
+    Build-Depends:
+        base         >= 4   && < 5  ,
+        pipes        >= 4.0 && < 4.1,
+        stm          >= 2.4 && < 2.5
+    Exposed-Modules:
+        Pipes.Concurrent,
+        Pipes.Concurrent.Tutorial
+    GHC-Options: -O2
+
+Test-Suite tests
+    Type: exitcode-stdio-1.0
+    Main-Is: tests-main.hs
+    HS-Source-Dirs: tests .
+    Build-Depends:
+        base              >= 4     && < 5  ,
+        pipes             >= 4.0.0 && < 4.1,
+        pipes-concurrency >= 2.0.0 && < 4.1,
+        stm               >= 2.4   && < 2.5,
+        async             >= 2.0   && < 2.1
diff --git a/src/Pipes/Concurrent.hs b/src/Pipes/Concurrent.hs
new file mode 100644
--- /dev/null
+++ b/src/Pipes/Concurrent.hs
@@ -0,0 +1,234 @@
+-- | Asynchronous communication between pipes
+
+{-# LANGUAGE CPP, RankNTypes#-}
+
+#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
+    action satisfies the necessary safety requirements because:
+
+    * You can safely repeat it if the transaction rolls back
+
+    * It does not acquire any resources
+
+    * It does not leak any inconsistent view of memory to the outside world
+
+    It appears to be unnecessary to read the IORef to keep it from being garbage
+    collected, but I wanted to be absolutely certain since I cannot be sure that
+    GHC won't optimize away the reference to the IORef.
+
+    The other alternative was to make 'send' and 'recv' use the 'IO' monad
+    instead of 'STM', but I felt that it was important to preserve the ability
+    to combine them into larger transactions.
+-}
+
+module Pipes.Concurrent (
+    -- * Inputs and Outputs
+    Input(..),
+    Output(..),
+
+    -- * Pipe utilities
+    fromInput,
+    toOutput,
+
+    -- * Actors
+    spawn,
+    spawn',
+    Buffer(..),
+
+    -- * Re-exports
+    -- $reexport
+    module Control.Concurrent,
+    module Control.Concurrent.STM,
+    module System.Mem
+    ) where
+
+import Control.Applicative (
+    Alternative(empty, (<|>)), Applicative(pure, (<*>)), (<*), (<$>) )
+import Control.Concurrent (forkIO)
+import Control.Concurrent.STM (atomically, STM)
+import qualified Control.Concurrent.STM as S
+import Control.Monad (when)
+import Data.IORef (newIORef, readIORef, mkWeakIORef)
+import Data.Monoid (Monoid(mempty, mappend))
+import GHC.Conc.Sync (unsafeIOToSTM)
+import Pipes (MonadIO(liftIO), yield, await, Producer', Consumer')
+import System.Mem (performGC)
+
+{-| An exhaustible source of values
+
+    'recv' returns 'Nothing' if the source is exhausted
+-}
+newtype Input a = Input {
+    recv :: S.STM (Maybe a) }
+
+instance Functor Input where
+    fmap f m = Input (fmap (fmap f) (recv m))
+
+instance Applicative Input where
+    pure r    = Input (pure (pure r))
+    mf <*> mx = Input ((<*>) <$> recv mf <*> recv mx)
+
+instance Monad Input where
+    return r = Input (return (return r))
+    m >>= f  = Input $ do
+        ma <- recv m
+        case ma of
+            Nothing -> return Nothing
+            Just a  -> recv (f a)
+
+-- Deriving 'Alternative'
+instance Alternative Input where
+    empty   = Input empty
+    x <|> y = Input (recv x <|> recv y)
+
+instance Monoid (Input a) where
+    mempty = empty
+    mappend = (<|>)
+
+{-| An exhaustible sink of values
+
+    'send' returns 'False' if the sink is exhausted
+-}
+newtype Output a = Output {
+    send :: a -> S.STM Bool }
+
+instance Monoid (Output a) where
+    mempty  = Output (\_ -> return False)
+    mappend i1 i2 = Output (\a -> (||) <$> send i1 a <*> send i2 a)
+
+{-| Convert an 'Output' to a 'Pipes.Consumer'
+
+    'toOutput' terminates when the 'Output' is exhausted.
+-}
+toOutput :: (MonadIO m) => Output a -> Consumer' a m ()
+toOutput output = loop
+  where
+    loop = do
+        a     <- await
+        alive <- liftIO $ S.atomically $ send output a
+        when alive loop
+{-# INLINABLE toOutput #-}
+
+{-| Convert an 'Input' to a 'Pipes.Producer'
+
+    'fromInput' terminates when the 'Input' is exhausted.
+-}
+fromInput :: (MonadIO m) => Input a -> Producer' a m ()
+fromInput input = loop
+  where
+    loop = do
+        ma <- liftIO $ S.atomically $ recv input
+        case ma of
+            Nothing -> return ()
+            Just a  -> do
+                yield a
+                loop
+{-# INLINABLE fromInput #-}
+
+{-| Spawn a mailbox using the specified 'Buffer' to store messages
+
+    Using 'send' on the 'Output'
+
+        * fails and returns 'False' if the mailbox is sealed, otherwise it:
+
+        * retries if the mailbox is full, or:
+
+        * adds a message to the mailbox and returns 'True'.
+
+    Using 'recv' on the 'Input':
+
+        * retrieves a message from the mailbox wrapped in 'Just' if the mailbox
+          is not empty, otherwise it:
+
+        * retries if the mailbox is not sealed, or:
+
+        * fails and returns 'Nothing'.
+
+    If either the 'Input' or 'Output' is garbage collected the mailbox will
+    become sealed.
+-}
+spawn :: Buffer a -> IO (Output a, Input a)
+spawn buffer = fmap simplify (spawn' buffer)
+  where
+    simplify (output, input, _) = (output, input)
+{-# INLINABLE spawn #-}
+
+{-| Like 'spawn', but also returns an action to manually @seal@ the mailbox
+    early:
+
+> (output, input, seal) <- spawn' buffer
+> ...
+
+    Use the @seal@ action to allow early cleanup of readers and writers to the
+    mailbox without waiting for the next garbage collection cycle.
+-}
+spawn' :: Buffer a -> IO (Output a, Input a, STM ())
+spawn' buffer = do
+    (read, write) <- case buffer of
+        Bounded n -> do
+            q <- S.newTBQueueIO n
+            return (S.readTBQueue q, S.writeTBQueue q)
+        Unbounded -> do
+            q <- S.newTQueueIO
+            return (S.readTQueue q, S.writeTQueue q)
+        Single    -> do
+            m <- S.newEmptyTMVarIO
+            return (S.takeTMVar m, S.putTMVar m)
+        Latest a  -> do
+            t <- S.newTVarIO a
+            return (S.readTVar t, S.writeTVar t)
+
+    sealed <- S.newTVarIO False
+    let seal = S.writeTVar sealed True
+
+    {- Use IORefs to keep track of whether the 'Input' or 'Output' has been
+       garbage collected.  Seal the mailbox when either of them becomes garbage
+       collected.
+    -}
+    rSend <- newIORef ()
+    mkWeakIORef rSend (S.atomically seal)
+    rRecv <- newIORef ()
+    mkWeakIORef rRecv (S.atomically seal)
+
+    let sendOrEnd a = do
+            b <- S.readTVar sealed
+            if b
+                then return False
+                else do
+                    write a
+                    return True
+        readOrEnd = (Just <$> read) <|> (do
+            b <- S.readTVar sealed
+            S.check b
+            return Nothing )
+        _send a = sendOrEnd a <* unsafeIOToSTM (readIORef rSend)
+        _recv   = readOrEnd   <* unsafeIOToSTM (readIORef rRecv)
+    return (Output _send, Input _recv, seal)
+{-# INLINABLE spawn' #-}
+
+-- | 'Buffer' specifies how to buffer messages stored within the mailbox
+data Buffer a
+    -- | Store an 'Unbounded' number of messages in a FIFO queue
+    = Unbounded
+    -- | Store a 'Bounded' number of messages, specified by the 'Int' argument
+    | Bounded Int
+    -- | Store a 'Single' message (like @Bounded 1@, but more efficient)
+    | Single
+    {-| Only store the 'Latest' message, beginning with an initial value
+
+        'Latest' is never empty nor full.
+    -}
+    | Latest a
+
+{- $reexport
+    @Control.Concurrent@ re-exports 'forkIO', although I recommend using the
+    @async@ library instead.
+
+    @Control.Concurrent.STM@ re-exports 'atomically' and 'STM'.
+
+    @System.Mem@ re-exports 'performGC'.
+-}
diff --git a/src/Pipes/Concurrent/Tutorial.hs b/src/Pipes/Concurrent/Tutorial.hs
new file mode 100644
--- /dev/null
+++ b/src/Pipes/Concurrent/Tutorial.hs
@@ -0,0 +1,812 @@
+{-| This module provides a tutorial for the @pipes-concurrency@ library.
+
+    This tutorial assumes that you have read the @pipes@ tutorial in
+    @Pipes.Tutorial@.
+
+    I've condensed all the code examples into self-contained code listings in
+    the Appendix section that you can use to follow along.
+-}
+
+module Pipes.Concurrent.Tutorial (
+    -- * Introduction
+    -- $intro
+
+    -- * Work Stealing
+    -- $steal
+
+    -- * Termination
+    -- $termination
+
+    -- * Mailbox Sizes
+    -- $mailbox
+
+    -- * Broadcasts
+    -- $broadcast
+
+    -- * Updates
+    -- $updates
+
+    -- * Callbacks
+    -- $callback
+
+    -- * Safety
+    -- $safety
+
+    -- * Conclusion
+    -- $conclusion
+
+    -- * Appendix
+    -- $appendix
+    ) where
+
+import Control.Concurrent
+import Control.Monad
+import Pipes
+import Pipes.Concurrent
+import qualified Pipes.Prelude as P
+import Data.Monoid
+
+{- $intro
+    The @pipes-concurrency@ library provides a simple interface for
+    communicating between concurrent pipelines.  Use this library if you want
+    to:
+
+    * merge multiple streams into a single stream,
+
+    * stream data from a callback \/ continuation,
+
+    * broadcast data,
+
+    * build a work-stealing setup, or
+
+    * implement basic functional reactive programming (FRP).
+
+    For example, let's say that we want to design a simple game with two
+    concurrent sources of game @Event@s.
+
+    One source translates user input to game events:
+
+> -- The game events
+> data Event = Harm Integer | Heal Integer | Quit deriving (Show)
+>
+> user :: IO Event
+> user = do
+>     command <- getLine
+>     case command of
+>         "potion" -> return (Heal 10)
+>         "quit"   -> return  Quit
+>         _        -> do
+>             putStrLn "Invalid command"
+>             user  -- Try again
+
+    ... while the other creates inclement weather:
+
+> import Control.Concurrent (threadDelay)
+> import Control.Monad (forever)
+> import Pipes
+>
+> acidRain :: Producer Event IO r
+> acidRain = forever $ do
+>     lift $ threadDelay 2000000  -- Wait 2 seconds
+>     yield (Harm 1)
+
+    We can asynchronously merge these two separate sources of @Event@s into a
+    single stream by 'spawn'ing a first-in-first-out (FIFO) mailbox:
+
+@
+ 'spawn' :: 'Buffer' a -> 'IO' ('Output' a, 'Input' a)
+@
+
+    'spawn' takes a 'Buffer' as an argument which specifies how many messages to
+    store.  In this case we want our mailbox to store an 'Unbounded' number of
+    messages:
+
+> import Pipes.Concurrent
+> 
+> main = do
+>     (output, input) <- spawn Unbounded
+>     ...
+
+   'spawn' creates this mailbox in the background and then returns two values:
+
+    * an @(Output a)@ that we use to add messages of type @a@ to the mailbox
+
+    * an @(Input a)@ that we use to consume messages of type @a@ from the
+      mailbox
+
+    We will be streaming @Event@s through our mailbox, so our @output@ has type
+    @(Output Event)@ and our @input@ has type @(Input Event)@.
+
+    To stream @Event@s into the mailbox , we use 'toOutput', which writes values
+    to the mailbox's 'Output' end:
+
+@
+ 'toOutput' :: ('MonadIO' m) => 'Output' a -> 'Consumer' a m ()
+@
+
+    We can concurrently forward multiple streams to the same 'Output', which
+    asynchronously merges their messages into the same mailbox:
+
+>     ...
+>     forkIO $ do runEffect $ lift user >~  toOutput output
+>                 performGC  -- I'll explain 'performGC' below
+> 
+>     forkIO $ do runEffect $ acidRain  >-> toOutput output
+>                 performGC
+>     ...
+
+    To stream @Event@s out of the mailbox, we use 'fromInput', which streams
+    values from the mailbox's 'Input' end using a 'Producer':
+
+@
+ 'fromInput' :: ('MonadIO' m) => 'Input' a -> 'Producer' a m ()
+@
+
+    For this example we'll build a 'Consumer' to handle this stream of @Event@s,
+    that either harms or heals our intrepid adventurer depending on which
+    @Event@ we receive:
+
+> handler :: Consumer Event IO ()
+> handler = loop 100
+>   where
+>     loop health = do
+>         lift $ putStrLn $ "Health = " ++ show health
+>         event <- await
+>         case event of
+>             Harm n -> loop (health - n)
+>             Heal n -> loop (health + n)
+>             Quit   -> return ()
+
+    Now we can just connect our @Event@ 'Producer' to our @Event@ 'Consumer'
+    using ('>->'):
+
+>     ...
+>     runEffect $ fromInput input >-> handler
+
+    Our final @main@ looks like this:
+
+> main = do
+>     (output, input) <- spawn Unbounded
+>
+>     forkIO $ do runEffect $ lift user >~  toOutput output
+>                 performGC  
+>
+>     forkIO $ do runEffect $ acidRain  >-> toOutput output
+>                 performGC
+>
+>     runEffect $ fromInput input >-> handler
+
+    ... and when we run it we get the desired concurrent behavior:
+
+> $ ./game
+> Health = 100
+> Health = 99
+> Health = 98
+> potion<Enter>
+> Health = 108
+> Health = 107
+> Health = 106
+> potion<Enter>
+> Health = 116
+> Health = 115
+> quit<Enter>
+> $
+-}
+
+{- $steal
+    You can also have multiple pipes reading from the same mailbox.  Messages
+    get split between listening pipes on a first-come first-serve basis.
+
+    For example, we'll define a \"worker\" that takes a one-second break each
+    time it receives a new job:
+
+> import Control.Concurrent (threadDelay)
+> import Control.Monad
+> import Pipes
+> 
+> worker :: (Show a) => Int -> Consumer a IO r
+> worker i = forever $ do
+>     a <- await
+>     lift $ threadDelay 1000000  -- 1 second
+>     lift $ putStrLn $ "Worker #" ++ show i ++ ": Processed " ++ show a
+
+    Fortunately, these workers are cheap, so we can assign several of them to
+    the same job:
+
+> import Control.Concurrent.Async
+> import qualified Pipes.Prelude as P
+> import Pipes.Concurrent
+> 
+> main = do
+>     (output, input) <- spawn Unbounded
+>     as <- forM [1..3] $ \i ->
+>           async $ do runEffect $ fromInput input  >-> worker i
+>                      performGC
+>     a  <- async $ do runEffect $ each [1..10] >-> toOutput output
+>                      performGC
+>     mapM_ wait (a:as)
+
+    The above example uses @Control.Concurrent.Async@ from the @async@ package
+    to fork each thread and wait for all of them to terminate:
+
+> $ ./work
+> Worker #2: Processed 3
+> Worker #1: Processed 2
+> Worker #3: Processed 1
+> Worker #3: Processed 6
+> Worker #1: Processed 5
+> Worker #2: Processed 4
+> Worker #2: Processed 9
+> Worker #1: Processed 8
+> Worker #3: Processed 7
+> Worker #2: Processed 10
+> $
+
+    What if we replace 'each' with a different source that reads lines from user
+    input until the user types \"quit\":
+
+> user :: Producer String IO ()
+> user = P.stdinLn >-> P.takeWhile (/= "quit")
+> 
+> main = do
+>     (output, input) <- spawn Unbounded
+>     as <- forM [1..3] $ \i ->
+>           async $ do runEffect $ fromInput input >-> worker i
+>                      performGC
+>     a  <- async $ do runEffect $ user >-> toOutput output
+>                      performGC
+>     mapM_ wait (a:as)
+
+    This still produces the correct behavior:
+
+> $ ./work
+> Test<Enter>
+> Worker #1: Processed "Test"
+> Apple<Enter>
+> Worker #2: Processed "Apple"
+> 42<Enter>
+> Worker #3: Processed "42"
+> A<Enter>
+> B<Enter>
+> C<Enter>
+> Worker #1: Processed "A"
+> Worker #2: Processed "B"
+> Worker #3: Processed "C"
+> quit<Enter>
+> $
+-}
+
+{- $termination
+
+    Wait...  How do the workers know when to stop listening for data?  After
+    all, anything that has a reference to 'Output' could potentially add more
+    data to the mailbox.
+
+    It turns out that 'spawn' is smart and instruments the 'Input' to
+    terminate when the 'Output' is garbage collected.  'fromInput' builds on top
+    of the more primitive 'recv' command, which returns a 'Nothing' when the
+    'Input' terminates:
+
+@
+ 'recv' :: 'Input' a -> 'STM' ('Maybe' a)
+@
+
+    Otherwise, 'recv' will block if the mailbox is empty since if the 'Output'
+    has not been garbage collected then somebody might still produce more data.
+
+    Does it work the other way around?  What happens if the workers go on strike
+    before processing the entire data set?
+
+>     ...
+>     as <- forM [1..3] $ \i ->
+>           -- Each worker refuses to process more than two values
+>           async $ do runEffect $ fromInput input >-> P.take 2 >-> worker i
+>                      performGC
+>     ...
+
+    Let's find out:
+
+> $ ./work
+> How<Enter>
+> Worker #1: Processed "How"
+> many<Enter>
+> roads<Enter>
+> Worker #2: Processed "many"
+> Worker #3: Processed "roads"
+> must<Enter>
+> a<Enter>
+> man<Enter>
+> Worker #1: Processed "must"
+> Worker #2: Processed "a"
+> Worker #3: Processed "man"
+> walk<Enter>
+> $
+
+    'spawn' tells the 'Output' to similarly terminate when the 'Input' is
+    garbage collected, preventing the user from submitting new values.
+    'toOutput' builds on top of the more primitive 'send' command, which returns
+    a 'False' when the 'Output' terminates:
+
+@
+ 'send' :: 'Output' a -> a -> 'STM' 'Bool'
+@
+
+    Otherwise, 'send' will blocks if the mailbox is full, since if the 'Input'
+    has not been garbage collected then somebody could still consume a value
+    from the mailbox, making room for a new value.
+
+    This is why we have to insert 'performGC' calls whenever we release a
+    reference to either the 'Output' or 'Input'.  Without these calls we cannot
+    guarantee that the garbage collector will trigger and notify the opposing
+    end if the last reference was released.
+
+    There are two ways to avoid using 'performGC'.  First, you can omit the
+    'performGC' call, which is safe and preferable for long-running programs.
+    This simply delays garbage collecting mailboxes until the next garbage
+    collection cycle.
+
+    Second, you can use the 'spawn'' command, which returns a third @seal@
+    action:
+
+> (output, input, seal) <- spawn' buffer
+> ...
+
+    Use this to @seal@ the mailbox so that it cannot receive new messages.  This
+    allows both readers and writers to shut down early without relying on
+    garbage collection:
+
+    * writers will shut down immediately because they can no longer write to the
+      mailbox
+
+    * readers will shut down when the mailbox goes empty because they know that
+      no new data will arrive
+
+    For simplicity, this tutorial will continue to use `performGC` since all
+    the examples are short-lived programs that do not build up a large heap.
+    However, when the heap grows large you want to avoid `performGC` and
+    consider using one of the above two alternatives instead.
+
+    Note only 'Input's and 'Output's specifically built using 'spawn' or
+    'spawn'' make use of the garbage collector.  If you build your own custom
+    'Input's and 'Output's then you do not need to use 'performGC' at all.
+-}
+
+{- $mailbox
+    So far we haven't observed 'send' blocking because we only 'spawn'ed
+    'Unbounded' mailboxes.  However, we can control the size of the mailbox to
+    tune the coupling between the 'Output' and the 'Input' ends.
+
+    If we set the mailbox 'Buffer' to 'Single', then the mailbox holds exactly
+    one message, forcing synchronization between 'send's and 'recv's.  Let's
+    observe this by sending an infinite stream of values, logging all values to
+    the console:
+
+> main = do
+>     (output, input) <- spawn Single
+>     as <- forM [1..3] $ \i ->
+>           async $ do runEffect $ fromInput input >-> P.take 2 >-> worker i
+>                      performGC
+>     a  <- async $ do runEffect $ each [1..] >-> P.chain print >-> toOutput output
+>                      performGC
+>     mapM_ wait (a:as)
+
+    The 7th value gets stuck in the mailbox, and the 8th value blocks because
+    the mailbox never clears the 7th value:
+
+> $ ./work
+> 1
+> 2
+> 3
+> 4
+> 5
+> Worker #3: Processed 3
+> Worker #2: Processed 2
+> Worker #1: Processed 1
+> 6
+> 7
+> 8
+> Worker #1: Processed 6
+> Worker #2: Processed 5
+> Worker #3: Processed 4
+> $
+
+    Contrast this with an 'Unbounded' mailbox for the same program, which keeps
+    accepting values until downstream finishes processing the first six values:
+
+> $ ./work
+> 1
+> 2
+> 3
+> 4
+> 5
+> 6
+> 7
+> 8
+> 9
+> ...
+> 487887
+> 487888
+> Worker #3: Processed 3
+> Worker #2: Processed 2
+> Worker #1: Processed 1
+> 487889
+> 487890
+> ...
+> 969188
+> 969189
+> Worker #1: Processed 6
+> Worker #2: Processed 5
+> Worker #3: Processed 4
+> 969190
+> 969191
+> $
+
+    You can also choose something in between by using a 'Bounded' mailbox which
+    caps the mailbox size to a fixed value.  Use 'Bounded' when you want mostly
+    loose coupling but still want to guarantee bounded memory usage:
+
+> main = do
+>     (output, input) <- spawn (Bounded 100)
+>     ...
+
+> $ ./work
+> ...
+> 103
+> 104
+> Worker #3: Processed 3
+> Worker #2: Processed 2
+> Worker #1: Processed 1
+> 105
+> 106
+> 107
+> Worker #1: Processed 6
+> Worker #2: Processed 5
+> Worker #3: Processed 4
+> $
+-}
+
+{- $broadcast
+    You can also broadcast data to multiple listeners instead of dividing up the
+    data.  Just use the 'Monoid' instance for 'Output' to combine multiple
+    'Output' ends together into a single broadcast 'Output':
+
+> -- broadcast.hs
+>
+> import Control.Monad
+> import Control.Concurrent.Async
+> import Pipes
+> import Pipes.Concurrent
+> import qualified Pipes.Prelude as P
+> import Data.Monoid
+> 
+> main = do
+>     (output1, input1) <- spawn Unbounded
+>     (output2, input2) <- spawn Unbounded
+>     a1 <- async $ do
+>         runEffect $ P.stdinLn >-> toOutput (output1 <> output2)
+>         performGC
+>     as <- forM [input1, input2] $ \input -> async $ do
+>         runEffect $ fromInput input >-> P.take 2 >-> P.stdoutLn
+>         performGC
+>     mapM_ wait (a1:as)
+
+    In the above example, 'P.stdinLn' will broadcast user input to both
+    mailboxes, and each mailbox forwards its values to 'P.stdoutLn', echoing the
+    message to standard output:
+
+> $ ./broadcast
+> ABC<Enter>
+> ABC
+> ABC
+> DEF<Enter>
+> DEF
+> DEF
+> GHI<Enter>
+> $ 
+
+    The combined 'Output' stays alive as long as any of the original 'Output's
+    remains alive.  In the above example, 'toOutput' terminates on the third
+    'send' attempt because it detects that both listeners died after receiving
+    two messages.
+
+    Use 'mconcat' to broadcast to a list of 'Output's, but keep in mind that you
+    will incur a performance price if you combine thousands of 'Output's or more
+    because they will create a very large 'STM' transaction.  You can improve
+    performance for very large broadcasts if you sacrifice atomicity and
+    manually combine multiple 'send' actions in 'IO' instead of 'STM'.
+-}
+
+{- $updates
+    Sometimes you don't want to handle every single event.  For example, you
+    might have an input and output device (like a mouse and a monitor) where the
+    input device updates at a different pace than the output device
+
+> import Control.Concurrent (threadDelay)
+> import Control.Monad
+> import Pipes
+> import qualified Pipes.Prelude as P
+> 
+> -- Fast input updates
+> inputDevice :: (Monad m) => Producer Integer m ()
+> inputDevice = each [1..]
+> 
+> -- Slow output updates
+> outputDevice :: Consumer Integer IO r
+> outputDevice = forever $ do
+>     n <- await
+>     lift $ do
+>         print n
+>         threadDelay 1000000
+
+    In this scenario you don't want to enforce a one-to-one correspondence
+    between input device updates and output device updates because you don't
+    want either end to block waiting for the other end.  Instead, you just need
+    the output device to consult the 'Latest' value received from the 'Input':
+
+> import Control.Concurrent.Async
+> import Pipes.Concurrent
+> 
+> main = do
+>     (output, input) <- spawn (Latest 0)
+>     a1 <- async $ do runEffect $ inputDevice >-> toOutput output
+>                      performGC
+>     a2 <- async $ do runEffect $ fromInput input >-> P.take 5 >-> outputDevice
+>                      performGC
+>     mapM_ wait [a1, a2]
+
+    'Latest' selects a mailbox that always stores exactly one value.  The
+    'Latest' constructor takes a single argument (@0@, in the above example)
+    specifying the starting value to store in the mailbox.  'send' overrides the
+    currently stored value and 'recv' peeks at the latest stored value without
+    consuming it.  In the above example the @outputDevice@ periodically peeks at    the latest value stashed inside the mailbox:
+
+> $ ./peek
+> 7
+> 2626943
+> 5303844
+> 7983519
+> 10604940
+> $
+
+    A 'Latest' mailbox is never empty because it begins with a default value and
+    'recv' never removes the value from the mailbox.  A 'Latest' mailbox is also
+    never full because 'send' always succeeds, overwriting the previously stored
+    value.
+-}
+
+{- $callback
+    @pipes-concurrency@ also solves the common problem of getting data out of a
+    callback-based framework into @pipes@.
+
+    For example, suppose that we have the following callback-based function:
+
+> import Control.Monad
+> 
+> onLines :: (String -> IO a) -> IO b
+> onLines callback = forever $ do
+>     str <- getLine
+>     callback str
+
+    We can use 'send' to free the data from the callback and then we can
+    retrieve the data on the outside using 'fromInput':
+
+> import Pipes
+> import Pipes.Concurrent
+> import qualified Pipes.Prelude as P
+> 
+> onLines' :: Producer String IO ()
+> onLines' = do
+>     (output, input) <- lift $ spawn Single
+>     lift $ forkIO $ onLines (\str -> atomically $ send output str)
+>     fromInput input
+> 
+> main = runEffect $ onLines' >-> P.takeWhile (/= "quit") >-> P.stdoutLn
+
+    Now we can stream from the callback as if it were an ordinary 'Producer':
+
+> $ ./callback
+> Test<Enter>
+> Test
+> Apple<Enter>
+> Apple
+> quit<Enter>
+> $
+
+-}
+
+{- $safety
+    @pipes-concurrency@ avoids deadlocks because 'send' and 'recv' always
+    cleanly return before triggering a deadlock.  This behavior works even in
+    complicated scenarios like:
+
+    * cyclic graphs of connected mailboxes,
+
+    * multiple readers and multiple writers to the same mailbox, and
+
+    * dynamically adding or garbage collecting mailboxes.
+
+    The following example shows how @pipes-concurrency@ will do the right thing
+    even in the case of cycles:
+
+> -- cycle.hs
+>
+> import Control.Concurrent.Async
+> import Pipes
+> import Pipes.Concurrent
+> import qualified Pipes.Prelude as P
+> 
+> main = do
+>     (out1, in1) <- spawn Unbounded
+>     (out2, in2) <- spawn Unbounded
+>     a1 <- async $ do
+>         runEffect $ (each [1,2] >> fromInput in1) >-> toOutput out2
+>         performGC
+>     a2 <- async $ do
+>         runEffect $ fromInput in2 >-> P.chain print >-> P.take 6 >-> toOutput out1
+>         performGC
+>     mapM_ wait [a1, a2]
+
+    The above program jump-starts a cyclic chain with two input values and
+    terminates one branch of the cycle after six values flow through.  Both
+    branches correctly terminate and get garbage collected without triggering
+    deadlocks when 'takeB_' finishes:
+
+> $ ./cycle
+> 1
+> 2
+> 1
+> 2
+> 1
+> 2
+> $
+
+-}
+
+{- $conclusion
+    @pipes-concurrency@ adds an asynchronous dimension to @pipes@.  This
+    promotes a natural division of labor for concurrent programs:
+
+    * Fork one pipeline per deterministic behavior
+
+    * Communicate between concurrent pipelines using @pipes-concurrency@
+
+    This promotes an actor-style approach to concurrent programming where
+    pipelines behave like processes and mailboxes behave like ... mailboxes.
+
+    You can ask questions about @pipes-concurrency@ and other @pipes@ libraries
+    on the official @pipes@ mailing list at
+    <mailto:haskell-pipes@googlegroups.com>.
+-}
+
+{- $appendix
+    I've provided the full code for the above examples here so you can easily
+    try them out:
+
+>-- game.hs
+>
+>import Control.Concurrent (threadDelay)
+>import Control.Monad (forever)
+>import Pipes
+>import Pipes.Concurrent
+>
+>data Event = Harm Integer | Heal Integer | Quit deriving (Show)
+>
+>user :: IO Event
+>user = do
+>    command <- getLine
+>    case command of
+>        "potion" -> return (Heal 10)
+>        "quit"   -> return  Quit
+>        _        -> do
+>            putStrLn "Invalid command"
+>            user
+>
+>acidRain :: Producer Event IO r
+>acidRain = forever $ do
+>    lift $ threadDelay 2000000  -- Wait 2 seconds
+>    yield (Harm 1)
+>
+>handler :: Consumer Event IO ()
+>handler = loop 100
+>  where
+>    loop health = do
+>        lift $ putStrLn $ "Health = " ++ show health
+>        event <- await
+>        case event of
+>            Harm n -> loop (health - n)
+>            Heal n -> loop (health + n)
+>            Quit   -> return ()
+>
+>main = do
+>    (output, input) <- spawn Unbounded
+>
+>    forkIO $ do runEffect $ lift user >~  toOutput output
+>                performGC
+>
+>    forkIO $ do runEffect $ acidRain  >-> toOutput output
+>                performGC
+>
+>    runEffect $ fromInput input >-> handler
+
+>-- work.hs
+>
+>import Control.Concurrent (threadDelay)
+>import Control.Concurrent.Async
+>import Control.Monad
+>import Pipes
+>import Pipes.Concurrent
+>import qualified Pipes.Prelude as P
+>
+>worker :: (Show a) => Int -> Consumer a IO r
+>worker i = forever $ do
+>    a <- await
+>    lift $ threadDelay 1000000  -- 1 second
+>    lift $ putStrLn $ "Worker #" ++ show i ++ ": Processed " ++ show a
+>
+>user :: Producer String IO ()
+>user = P.stdinLn >-> P.takeWhile (/= "quit")
+>
+>main = do
+>--  (output, input) <- spawn Unbounded
+>--  (output, input) <- spawn Single
+>    (output, input) <- spawn (Bounded 100)
+>
+>    as <- forM [1..3] $ \i ->
+>--        async $ do runEffect $ fromInput input  >-> worker i
+>          async $ do runEffect $ fromInput input  >-> P.take 2 >-> worker i
+>                     performGC
+>
+>--  a  <- async $ do runEffect $ each [1..10]                 >-> toOutput output
+>--  a  <- async $ do runEffect $ user                         >-> toOutput output
+>    a  <- async $ do runEffect $ each [1..] >-> P.chain print >-> toOutput output
+>                     performGC
+>
+>    mapM_ wait (a:as)
+
+>-- peek.hs
+>
+>import Control.Concurrent (threadDelay)
+>import Control.Concurrent.Async
+>import Control.Monad
+>import Pipes
+>import Pipes.Concurrent
+>import qualified Pipes.Prelude as P
+>
+>inputDevice :: (Monad m) => Producer Integer m ()
+>inputDevice = each [1..]
+>
+>outputDevice :: Consumer Integer IO r
+>outputDevice = forever $ do
+>    n <- await
+>    lift $ do
+>        print n
+>        threadDelay 1000000
+>
+>main = do
+>    (output, input) <- spawn (Latest 0)
+>    a1 <- async $ do runEffect $ inputDevice >-> toOutput output
+>                     performGC
+>    a2 <- async $ do runEffect $ fromInput input >-> P.take 5 >-> outputDevice
+>                     performGC
+>    mapM_ wait [a1, a2]
+
+>-- callback.hs
+>
+>import Control.Monad
+>import Pipes
+>import Pipes.Concurrent
+>import qualified Pipes.Prelude as P
+>
+>onLines :: (String -> IO a) -> IO b
+>onLines callback = forever $ do
+>    str <- getLine
+>    callback str
+>
+>onLines' :: Producer String IO ()
+>onLines' = do
+>    (output, input) <- lift $ spawn Single
+>    lift $ forkIO $ onLines (\str -> atomically $ send output str)
+>    fromInput input
+>
+>main = runEffect $ onLines' >-> P.takeWhile (/= "quit") >-> P.stdoutLn
+-}
diff --git a/tests/tests-main.hs b/tests/tests-main.hs
new file mode 100644
--- /dev/null
+++ b/tests/tests-main.hs
@@ -0,0 +1,122 @@
+module Main ( main ) where
+
+import Control.Concurrent hiding (yield)
+import Control.Concurrent.Async
+import Control.Monad (forever)
+import Pipes
+import Pipes.Concurrent
+import qualified Pipes.Prelude as P
+import System.Exit
+import System.IO
+import System.Timeout
+
+defaultTimeout :: Int
+defaultTimeout = 100000         -- 0.1 s
+
+labelPrint :: (Show a) => String -> Consumer a IO r
+labelPrint label = forever $ do
+  a <- await
+  lift $ putStrLn $ label ++ ": " ++ show a
+
+testSenderClose :: Buffer Int -> IO ()
+testSenderClose buffer = do
+    (output, input) <- spawn buffer
+    t1 <- async $ do
+        runEffect $ each [1..5] >-> toOutput output
+        performGC
+    t2 <- async $ do
+        runEffect $   fromInput input
+                  >-> P.chain (\_ -> threadDelay 1000)
+                  >-> P.print
+        performGC
+    wait t1
+    wait t2
+
+testSenderCloseDelayedSend :: Buffer Int -> IO ()
+testSenderCloseDelayedSend buffer = do
+    (output, input) <- spawn buffer
+    t1 <- async $ do
+        runEffect $   each [1..5]
+                  >-> P.tee (toOutput output)
+                  >-> for cat (\_ -> lift $ threadDelay 2000)
+        performGC
+    t2 <- async $ do
+        runEffect $   fromInput input
+                  >-> P.chain (\_ -> threadDelay 1000)
+                  >-> P.print
+        performGC
+    wait t1
+    wait t2
+
+testReceiverClose :: Buffer Int -> IO ()
+testReceiverClose buffer = do
+    (output, input) <- spawn buffer
+    t1 <- async $ do
+        runEffect $   each [1..]
+                  >-> P.tee (toOutput output)
+                  >-> P.chain (\_ -> threadDelay 1000)
+                  >-> P.print
+        performGC
+    t2 <- async $ do
+        runEffect $ for (fromInput input >-> P.take 10) discard
+        performGC
+    wait t1
+    wait t2
+
+testReceiverCloseDelayedReceive :: Buffer Int -> IO ()
+testReceiverCloseDelayedReceive buffer = do
+    (output, input) <- spawn buffer
+    t1 <- async $ do
+        runEffect $   each [1..]
+                  >-> P.tee (toOutput output)
+                  >-> P.chain (\_ -> threadDelay 1000)
+                  >-> labelPrint "Send"
+        performGC
+    t2 <- async $ do
+        runEffect $   fromInput input
+                  >-> P.take 10
+                  >-> P.chain (\_ -> threadDelay 800)
+                  >-> labelPrint "Recv"
+        performGC
+    wait t1
+    wait t2
+
+runTest :: IO () -> String -> IO ()
+runTest test name = do
+    putStrLn $ "Starting test: " ++ name
+    hFlush stdout
+    result <- timeout defaultTimeout test
+    case result of
+        Nothing -> do putStrLn $ "Test " ++ name ++ " timed out. Aborting."
+                      exitFailure
+        Just _  -> do putStrLn $ "Test " ++ name ++ " finished."
+    hFlush stdout
+
+runTestExpectTimeout :: IO () -> String -> IO ()
+runTestExpectTimeout test name = do
+    putStrLn $ "Starting test: " ++ name
+    hFlush stdout
+    result <- timeout defaultTimeout test
+    case result of
+        Nothing -> putStrLn $ "Test " ++ name ++ " timed out as expected."
+        Just _  -> do
+            putStrLn $
+                   "Test "
+                ++ name
+                ++ " finished, but a timeout was expected. Aborting."
+            exitFailure
+    hFlush stdout
+
+main :: IO ()
+main = do
+    runTest (testSenderClose Unbounded) "UnboundedSenderClose"
+    runTest (testSenderClose $ Bounded 3) "BoundedFilledSenderClose"
+    runTest (testSenderClose $ Bounded 7) "BoundedNotFilledSenderClose"
+    runTest (testSenderClose Single) "SingleSenderClose"
+    runTestExpectTimeout (testSenderCloseDelayedSend $ Latest 42) "LatestSenderClose"
+    --
+    runTest (testReceiverClose Unbounded) "UnboundedReceiverClose"
+    runTest (testReceiverClose $ Bounded 3) "BoundedFilledReceiverClose"
+    runTest (testReceiverClose $ Bounded 7) "BoundedNotFilledReceiverClose"
+    runTest (testReceiverClose Single) "SingleReceiverClose"
+    runTest (testReceiverCloseDelayedReceive $ Latest 42) "LatestReceiverClose"