diff --git a/Control/Proxy/Concurrent.hs b/Control/Proxy/Concurrent.hs
--- a/Control/Proxy/Concurrent.hs
+++ b/Control/Proxy/Concurrent.hs
@@ -1,205 +1,249 @@
--- | Asynchronous communication between proxies
-
-{-# LANGUAGE CPP #-}
-
-#if __GLASGOW_HASKELL__ >= 702
-{-# LANGUAGE Trustworthy #-}
-#endif
-{- 'unsafeIOToSTM' requires the Trustworthy annotation.
-
-    I use 'unsafeIOToSTM' to touch an IORef to mark it 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,
-    Size(..),
-    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 ((<|>), (<*), pure)
-import Control.Concurrent (forkIO)
-import Control.Concurrent.STM (atomically, STM)
-import Control.Monad.Trans.Class (lift)
-import qualified Control.Concurrent.STM as S
-import qualified Control.Proxy as P
-import Data.IORef (newIORef, readIORef, mkWeakIORef)
-import GHC.Conc.Sync (unsafeIOToSTM)
-import System.Mem (performGC)
-
-{-| Spawn a mailbox of the specified 'Size' that has an 'Input' and 'Output' end
--}
-spawn :: Size -> IO (Input a, Output a)
-spawn size = do
-    (read, write) <- case size of
-        Bounded n -> do
-            q <- S.newTBQueueIO n
-            let read = do
-                    ma <- S.readTBQueue q
-                    case ma of
-                        Nothing -> S.unGetTBQueue q ma
-                        _       -> return ()
-                    return ma
-            return (read, S.writeTBQueue q)
-        Unbounded -> do
-            q <- S.newTQueueIO
-            let read = do
-                    ma <- S.readTQueue q
-                    case ma of
-                        Nothing -> S.unGetTQueue q ma
-                        _       -> return ()
-                    return ma
-            return (read, S.writeTQueue q)
-        Single    -> do
-            m <- S.newEmptyTMVarIO
-            let read = do
-                    ma <- S.takeTMVar m
-                    case ma of
-                        Nothing -> S.putTMVar m ma
-                        _       -> return ()
-                    return ma
-            return (read, S.putTMVar m)
-
-    {- Use an IORef to keep track of whether the 'Input' end has been garbage
-       collected and run a finalizer when the collection occurs
-
-       The finalizer cannot anticipate how many listeners there are, so it only
-       writes a single 'Nothing' and trusts that the supplied 'read' action
-       will not consume the 'Nothing'.
-
-       The 'write' must be protected with the "pure ()" fallback so that it does
-       not deadlock if the 'Output' end has also been garbage collected.
-    -}
-    rUp  <- newIORef ()
-    mkWeakIORef rUp (S.atomically $ write Nothing <|> pure ())
-
-    {- Use an IORef to keep track of whether the 'Output' end has been garbage
-       collected and run a finalizer when the collection occurs
-    -}
-    rDn  <- newIORef ()
-    done <- S.newTVarIO False
-    mkWeakIORef rDn (S.atomically $ S.writeTVar done True)
-
-    let quit = do
-            b <- S.readTVar done
-            S.check b
-            return False
-        continue a = do
-            write (Just a)
-            return True
-        {- The '_send' action aborts 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.
-        -}
-        _send a = (quit <|> continue a) <* unsafeIOToSTM (readIORef rUp)
-        _recv = read <* unsafeIOToSTM (readIORef rDn)
-    return (Input _send , Output _recv)
-
-{-| 'Size' specifies how many messages to store in the mailbox before 'send'
-    blocks.
--}
-data Size
-    -- | Store an 'Unbounded' number of messages
-    = Unbounded
-    -- | Store a 'Bounded' number of messages specified by the 'Int' argument
-    | Bounded Int
-    -- | Store only a 'Single' message (like @Bounded 1@, but more efficient)
-    | Single
-
--- | 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 }
-
--- | 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) }
-
-{-| Writes all messages flowing \'@D@\'ownstream to the given 'Input'
-
-    'sendD' terminates when the corresponding 'Output' is garbage collected.
--}
-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 <- lift $ S.atomically $ send input a
-        if alive
-            then do
-                x2 <- P.respond a
-                loop x2
-            else return ()
-
-{-| Convert an 'Output' to a 'P.Producer'
-
-    'recvS' terminates when the corresponding 'Input' is garbage collected.
--}
-recvS :: (P.Proxy p) => Output a -> () -> P.Producer p a IO ()
-recvS output () = P.runIdentityP go
-  where
-    go = do
-        ma <- lift $ S.atomically $ recv output
-        case ma of
-            Nothing -> return ()
-            Just a  -> do
-                P.respond a
-                go
-
-{- $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'.
--}
+-- | Asynchronous communication between proxies
+
+{-# LANGUAGE CPP #-}
+
+#if __GLASGOW_HASKELL__ >= 702
+{-# LANGUAGE Trustworthy #-}
+#endif
+{- 'unsafeIOToSTM' requires the Trustworthy annotation.
+
+    I use 'unsafeIOToSTM' to touch an IORef to mark it 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
+            let read = do
+                    ma <- S.readTBQueue q
+                    case ma of
+                        Nothing -> S.unGetTBQueue q ma
+                        _       -> return ()
+                    return ma
+            return (read, S.writeTBQueue q)
+        Unbounded -> do
+            q <- S.newTQueueIO
+            let read = do
+                    ma <- S.readTQueue q
+                    case ma of
+                        Nothing -> S.unGetTQueue q ma
+                        _       -> return ()
+                    return ma
+            return (read, S.writeTQueue q)
+        Single    -> do
+            m <- S.newEmptyTMVarIO
+            let read = do
+                    ma <- S.takeTMVar m
+                    case ma of
+                        Nothing -> S.putTMVar m ma
+                        _       -> return ()
+                    return ma
+            return (read, S.putTMVar m)
+        Latest a  -> do
+            t <- S.newTVarIO a
+            let write ma = case ma of
+                    Nothing -> return ()
+                    Just a  -> S.writeTVar t a
+            return (fmap Just (S.readTVar t), write)
+
+    {- Use an IORef to keep track of whether the 'Input' end has been garbage
+       collected and run a finalizer when the collection occurs
+
+       The finalizer cannot anticipate how many listeners there are, so it only
+       writes a single 'Nothing' and trusts that the supplied 'read' action
+       will not consume the 'Nothing'.
+
+       The 'write' must be protected with the "pure ()" fallback so that it does
+       not deadlock if the 'Output' end has also been garbage collected.
+    -}
+    rUp  <- newIORef ()
+    mkWeakIORef rUp (S.atomically $ write Nothing <|> pure ())
+
+    {- Use an IORef to keep track of whether the 'Output' end has been garbage
+       collected and run a finalizer when the collection occurs
+    -}
+    rDn  <- newIORef ()
+    done <- S.newTVarIO False
+    mkWeakIORef rDn (S.atomically $ S.writeTVar done True)
+
+    let quit = do
+            b <- S.readTVar done
+            S.check b
+            return False
+        continue a = do
+            write (Just a)
+            return True
+        {- The '_send' action aborts 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.
+        -}
+        _send a = (quit <|> continue a) <* unsafeIOToSTM (readIORef rUp)
+        _recv = read <* unsafeIOToSTM (readIORef rDn)
+    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)
+
+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 :: (P.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
--- a/Control/Proxy/Concurrent/Tutorial.hs
+++ b/Control/Proxy/Concurrent/Tutorial.hs
@@ -1,573 +1,772 @@
-{-| 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
-
-    -- * Callbacks
-    -- $callback
-
-    -- * Safety
-    -- $safety
-
-    -- * Conclusion
-    -- $conclusion
-
-    -- * Appendix
-    -- $appendix
-    ) where
-
-import Control.Proxy
-import Control.Proxy.Concurrent
-
-{- $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,
-
-    * implement 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 ()
-> 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 :: Size -> IO (Input a, Output a)
-
-    'spawn' takes a mailbox 'Size' as an argument, and we specify that we want
-    our mailbox to store an 'Unbounded' number of message.  '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
-
-> import Control.Proxy.Concurrent
->
-> main = do
->     (input, output) <- spawn Unbounded
->     ...
-
-    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 -> x -> p x a x 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?
-
-> -- Each worker refuses to process more than two values
-> 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
-
-> $ ./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 'Size' 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 >-> 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
-> $
--}
-
-{- $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.
--}
-
-{- $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.
--}
-
-{- $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 ()
-> 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
->                      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)
-
-> -- 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
-
--}
+{-| 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,40 +1,39 @@
-Name: pipes-concurrency
-Version: 1.1.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 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,
-        transformers >= 0.2.0.0 && < 0.4
-    Exposed-Modules:
-        Control.Proxy.Concurrent,
-        Control.Proxy.Concurrent.Tutorial
-    GHC-Options: -O2
+Name: pipes-concurrency
+Version: 1.2.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 "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