diff --git a/CHANGELOG.md b/CHANGELOG.md
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,8 @@
+## 0.0.8.0
+
+Add `Consume`, `await`, `consumeStream`, `zipCoroutines`,
+`consumeEach`, and associated examples.
+
 ## 0.0.7.0
 
 Add `withEffToIO'`, `race`, `connectCoroutines`, `receiveStream`,
diff --git a/bluefin-internal.cabal b/bluefin-internal.cabal
--- a/bluefin-internal.cabal
+++ b/bluefin-internal.cabal
@@ -1,6 +1,6 @@
 cabal-version:      3.0
 name:               bluefin-internal
-version:            0.0.7.0
+version:            0.0.8.0
 license:            MIT
 license-file:       LICENSE
 author:             Tom Ellis
diff --git a/src/Bluefin/Internal.hs b/src/Bluefin/Internal.hs
--- a/src/Bluefin/Internal.hs
+++ b/src/Bluefin/Internal.hs
@@ -13,6 +13,7 @@
 import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)
 import Control.Exception (throwIO, tryJust)
 import qualified Control.Exception
+import Control.Monad (forever)
 import Control.Monad.Base (MonadBase (liftBase))
 import Control.Monad.IO.Class (MonadIO, liftIO)
 import Control.Monad.IO.Unlift (MonadUnliftIO, withRunInIO)
@@ -112,7 +113,7 @@
 
 -- | Connect two coroutines.  Their execution is interleaved by
 -- exchanging @a@s and @b@s. When the first yields its first @a@ it
--- starts the second (which is waiting to receive an @a@).
+-- starts the second (which is awaiting an @a@).
 connectCoroutines ::
   forall es a b r.
   (forall e. Coroutine a b e -> Eff (e :& es) r) ->
@@ -137,12 +138,40 @@
 
   race (useImplWithin t1) (useImplWithin t2) io
 
+-- | Old name for 'consumeStream'.  @receiveStream@ will be deprecated
+-- in a future version.
 receiveStream ::
-  (forall e. Coroutine () a e -> Eff (e :& es) r) ->
+  (forall e. Consume a e -> Eff (e :& es) r) ->
   (forall e. Stream a e -> Eff (e :& es) r) ->
   Eff es r
-receiveStream r s = connectCoroutines r (\() -> s)
+receiveStream = consumeStream
 
+consumeStream ::
+  -- | Each 'await' from the @Consume@ ...
+  (forall e. Consume a e -> Eff (e :& es) r) ->
+  -- | ... receives the value 'yield'ed from the @Stream@
+  (forall e. Stream a e -> Eff (e :& es) r) ->
+  Eff es r
+consumeStream r s = connectCoroutines r (\() -> s)
+
+zipCoroutines ::
+  (e1 :> es) =>
+  Coroutine (a1, a2) b e1 ->
+  (forall e. Coroutine a1 b e -> Eff (e :& es) r) ->
+  (forall e. Coroutine a2 b e -> Eff (e :& es) r) ->
+  -- | ͘
+  Eff es r
+zipCoroutines c m1 m2 = do
+  connectCoroutines m1 $ \a1 c1 -> do
+    connectCoroutines (useImplWithin m2) $ \a2 c2 -> do
+      evalState (a1, a2) $ \ass -> do
+        forever $ do
+          as <- get ass
+          b' <- yieldCoroutine c as
+          a1' <- yieldCoroutine c1 b'
+          a2' <- yieldCoroutine c2 b'
+          put ass (a1', a2')
+
 instance (e :> es) => MonadBase IO (EffReader (IOE e) es) where
   liftBase = liftIO
 
@@ -271,6 +300,8 @@
 -- @a@ and then expects values of type @()@.
 type Stream a = Coroutine a ()
 
+type Consume a = Coroutine () a
+
 -- | You can define a @Handle@ instance for your compound handles.  As
 -- an example, an "application" handle with a dynamic effect for
 -- database queries, a concrete effect for application state and a
@@ -400,10 +431,10 @@
 -- Right "No exception thrown"
 -- @
 throw ::
-  (ex :> es) =>
-  Exception e ex ->
+  (e :> es) =>
+  Exception ex e ->
   -- | Value to throw
-  e ->
+  ex ->
   Eff es a
 throw (UnsafeMkException throw_) e = UnsafeMkEff (throw_ e)
 
@@ -488,7 +519,6 @@
   (a -> Eff es ()) ->
   -- | Run the body
   (a -> Eff es b) ->
-  -- |
   Eff es b
 bracket before after body =
   UnsafeMkEff $
@@ -722,6 +752,23 @@
     ii <- get i
     yield st (ii, s)
     put i (ii + 1)
+
+-- | A version of 'forEach' specialized to @Consume@.  Every time the
+-- @Consume@ is used to 'await' a @b@, feed it the one created by the
+-- handler.
+consumeEach ::
+  -- | Body
+  ( forall e.
+    Consume b e ->
+    Eff (e :& es) r
+  ) ->
+  -- | Value to send to each @await@ in the body.
+  Eff es b ->
+  Eff es r
+consumeEach k e = forEach k (\() -> e)
+
+await :: (e :> es) => Consume a e -> Eff es a
+await r = yieldCoroutine r ()
 
 type EarlyReturn = Exception
 
diff --git a/src/Bluefin/Internal/Examples.hs b/src/Bluefin/Internal/Examples.hs
--- a/src/Bluefin/Internal/Examples.hs
+++ b/src/Bluefin/Internal/Examples.hs
@@ -15,7 +15,7 @@
 import qualified Bluefin.Internal.Pipes as P
 import Control.Exception (IOException)
 import qualified Control.Exception
-import Control.Monad (forever, unless, when)
+import Control.Monad (forever, replicateM_, unless, when)
 import Control.Monad.IO.Class (liftIO)
 import Data.Foldable (for_)
 import Data.Monoid (Any (Any, getAny))
@@ -169,6 +169,189 @@
 
   forEach formattedLines $ \line -> effIO io (putStrLn line)
 
+awaitList ::
+  (e :> es) =>
+  [a] ->
+  IOE e ->
+  (forall e1. Consume a e1 -> Eff (e1 :& es) ()) ->
+  Eff es ()
+awaitList l io k = evalState l $ \s -> do
+  withJump $ \done ->
+    bracket
+      (pure ())
+      (\() -> effIO io (putStrLn "Released"))
+      $ \() -> do
+        consumeEach (useImplWithin k) $ do
+          (x, xs) <-
+            get s >>= \case
+              [] -> jumpTo done
+              x : xs -> pure (x, xs)
+          put s xs
+          pure x
+
+takeRec ::
+  (e3 :> es) =>
+  Int ->
+  (forall e. Consume a e -> Eff (e :& es) ()) ->
+  Consume a e3 ->
+  Eff es ()
+takeRec n k rec =
+  withJump $ \done -> evalState n $ \s -> consumeEach (useImplWithin k) $ do
+    s' <- get s
+    if s' <= 0
+      then jumpTo done
+      else do
+        modify s (subtract 1)
+        await rec
+
+mapRec ::
+  (e :> es) =>
+  (a -> b) ->
+  (forall e1. Consume b e1 -> Eff (e1 :& es) ()) ->
+  Consume a e ->
+  Eff es ()
+mapRec f = traverseRec (pure . f)
+
+traverseRec ::
+  (e :> es) =>
+  (a -> Eff es b) ->
+  (forall e1. Consume b e1 -> Eff (e1 :& es) ()) ->
+  Consume a e ->
+  Eff es ()
+traverseRec f k rec = forEach k $ \() -> do
+  r <- await rec
+  f r
+
+awaitUsage ::
+  (e1 :> es, e2 :> es) =>
+  IOE e1 ->
+  (forall e. Consume () e -> Eff (e :& es) ()) ->
+  Consume Int e2 ->
+  Eff es ()
+awaitUsage io x = do
+  mapRec (* 11) $
+    mapRec (subtract 1) $
+      takeRec 3 $
+        traverseRec (effIO io . print) $
+          useImplWithin x
+
+awaitExample :: IO ()
+awaitExample = runEff $ \io -> do
+  awaitList [1 :: Int ..] io $ awaitUsage io $ \rec -> do
+    replicateM_ 5 (await rec)
+
+consumeStreamExample :: IO (Either String String)
+consumeStreamExample = runEff $ \io -> do
+  try $ \ex -> do
+    consumeStream
+      ( \r ->
+          bracket
+            (effIO io (putStrLn "Starting 2"))
+            (\_ -> effIO io (putStrLn "Leaving 2"))
+            $ \_ -> do
+              for_ [1 :: Int .. 100] $ \n -> do
+                b' <- await r
+                effIO
+                  io
+                  ( putStrLn
+                      ("Consumed body " ++ show b' ++ " at time " ++ show n)
+                  )
+              pure "Consumer finished first"
+      )
+      ( \y -> bracket
+          (effIO io (putStrLn "Starting 1"))
+          (\_ -> effIO io (putStrLn "Leaving 1"))
+          $ \_ -> do
+            for_ [1 :: Int .. 10] $ \n -> do
+              effIO io (putStrLn ("Sending " ++ show n))
+              yield y n
+              when (n > 5) $ do
+                effIO io (putStrLn "Aborting...")
+                throw ex "Aborted"
+
+            pure "Yielder finished first"
+      )
+
+consumeStreamExample2 :: IO ()
+consumeStreamExample2 = runEff $ \io -> do
+  let counter yeven yodd = for_ [0 :: Int .. 10] $ \i -> do
+        if even i
+          then yield yeven i
+          else yield yodd i
+
+  let foo yeven =
+        consumeStream
+          ( \r -> forever $ do
+              i <- await r
+              effIO io (putStrLn ("Odd: " ++ show i))
+          )
+          (counter yeven)
+
+  let bar =
+        consumeStream
+          ( \r -> forever $ do
+              i <- await r
+              effIO io (putStrLn ("Even: " ++ show i))
+          )
+          foo
+
+  bar
+
+connectExample :: IO (Either String String)
+connectExample = runEff $ \io -> do
+  try $ \ex -> do
+    connectCoroutines
+      ( \y -> bracket
+          (effIO io (putStrLn "Starting 1"))
+          (\_ -> effIO io (putStrLn "Leaving 1"))
+          $ \_ -> do
+            for_ [1 :: Int .. 10] $ \n -> do
+              effIO io (putStrLn ("Sending " ++ show n))
+              yield y n
+              when (n > 5) $ do
+                effIO io (putStrLn "Aborting...")
+                throw ex "Aborted"
+
+            pure "Yielder finished first"
+      )
+      ( \binit r ->
+          bracket
+            (effIO io (putStrLn "Starting 2"))
+            (\_ -> effIO io (putStrLn "Leaving 2"))
+            $ \_ -> do
+              effIO io (putStrLn ("Consumed intial " ++ show binit))
+              for_ [1 :: Int .. 100] $ \n -> do
+                b' <- await r
+                effIO
+                  io
+                  ( putStrLn
+                      ("Consumed body " ++ show b' ++ " at time " ++ show n)
+                  )
+              pure "Consumer finished first"
+      )
+
+zipCoroutinesExample :: IO ()
+zipCoroutinesExample = runEff $ \io -> do
+  let m1 y = do
+        r <- yieldCoroutine y 1
+        evalState r $ \rs -> do
+          for_ [1 .. 10 :: Int] $ \i -> do
+            r' <- get rs
+            r'' <- yieldCoroutine y (r' + i)
+            put rs r''
+
+  let m2 y = do
+        r <- yieldCoroutine y 1
+        evalState r $ \rs -> do
+          for_ [1 .. 5 :: Int] $ \i -> do
+            r' <- get rs
+            r'' <- yieldCoroutine y (r' - i)
+            put rs r''
+
+  forEach (\c -> zipCoroutines c m1 m2) $ \i@(i1, i2) -> do
+    effIO io (print i)
+    pure (i1 + i2)
+
 -- Count the number of (strictly) positives and (strictly) negatives
 -- in a list, unless we see a zero, in which case we bail with an
 -- error message.
@@ -665,3 +848,32 @@
 pipesExample2 :: IO String
 pipesExample2 = runEff $ \io -> runEffect $ do
   stdinLn io >-> takeWhile' (/= "quit") >-> stdoutLn io
+
+-- Acquiring resource
+-- 1
+-- 2
+-- 3
+-- 4
+-- 5
+-- Releasing resource
+-- Finishing
+promptCoroutine :: IO ()
+promptCoroutine = runEff $ \io -> do
+  -- consumeStream connects a consumer to a producer
+  consumeStream
+    -- Like a pipes Consumer.  Prints the first five elements it
+    -- awaits.
+    ( \r -> for_ [1 :: Int .. 5] $ \_ -> do
+        v <- await r
+        effIO io (print v)
+    )
+    -- Like a pipes Producer. Yields successive integers indefinitely.
+    -- Unlike in pipes, we can simply use Bluefin's standard bracket
+    -- for prompt release of a resource
+    ( \y ->
+        bracket
+          (effIO io (putStrLn "Acquiring resource"))
+          (\_ -> effIO io (putStrLn "Releasing resource"))
+          (\_ -> for_ [1 :: Int ..] $ \i -> yield y i)
+    )
+  effIO io (putStrLn "Finishing")
diff --git a/src/Bluefin/Internal/Pipes.hs b/src/Bluefin/Internal/Pipes.hs
--- a/src/Bluefin/Internal/Pipes.hs
+++ b/src/Bluefin/Internal/Pipes.hs
@@ -1,6 +1,6 @@
 module Bluefin.Internal.Pipes where
 
-import Bluefin.Internal hiding (yield)
+import Bluefin.Internal hiding (yield, await)
 import qualified Bluefin.Internal
 import Control.Monad (forever)
 import Data.Foldable (for_)