diff --git a/Data/Conduit.hs b/Data/Conduit.hs
--- a/Data/Conduit.hs
+++ b/Data/Conduit.hs
@@ -7,6 +7,20 @@
 -- operators.
 module Data.Conduit
     ( -- * Types
+      -- | The three core types to this package are 'Source' (the data
+      -- producer), 'Sink' (the data consumer), and 'Conduit' (the data
+      -- transformer). For all three types, a result will provide the next
+      -- value to be used. For example, the @Open@ constructor includes a new
+      -- @Source@ in it. This leads to the main invariant for all conduit code:
+      -- these three types may /never/ be reused.  While some specific values
+      -- may work fine with reuse, the result is generally unpredictable and
+      -- should no be relied upon.
+      --
+      -- The user-facing API provided by the connect and fuse operators
+      -- automatically addresses the low level details of pulling, pushing, and
+      -- closing, and there should rarely be need to perform these actions in
+      -- user code.
+
       -- ** Source
       module Data.Conduit.Types.Source
       -- *** Buffering
@@ -32,6 +46,8 @@
     , module Data.Conduit.Util.Sink
       -- ** Conduit
     , module Data.Conduit.Util.Conduit
+      -- * Flushing
+    , Flush (..)
       -- * Convenience re-exports
     , ResourceT
     , Resource (..)
@@ -41,6 +57,8 @@
     , ResourceThrow (..)
     ) where
 
+import Control.Applicative ((<$>))
+import Control.Monad (liftM)
 import Control.Monad.Trans.Resource
 import Data.Conduit.Types.Source
 import Data.Conduit.Util.Source
@@ -49,6 +67,8 @@
 import Data.Conduit.Types.Conduit
 import Data.Conduit.Util.Conduit
 
+-- $typeOverview
+
 infixr 0 $$
 
 -- | The connect operator, which pulls data from a source and pushes to a sink.
@@ -57,21 +77,24 @@
 -- 1. In the case of a @SinkNoData@ constructor, the source is not opened at
 -- all, and the output value is returned immediately.
 --
--- 2. The sink returns @Done@, in which case any leftover input is returned via
--- @bsourceUnpull@ the source is closed.
+-- 2. The sink returns @Done@. If the input was a @BufferedSource@, any
+-- leftover input is put in the buffer. For a normal @Source@, the leftover
+-- value is discarded, and the source is closed.
 --
 -- 3. The source return @Closed@, in which case the sink is closed.
 --
--- Note that this function will automatically close any 'Source's, but will not
--- close any 'BufferedSource's, allowing them to be reused.
+-- Note that this function will automatically close any @Source@s, but will not
+-- close any @BufferedSource@s, allowing them to be reused.
 --
--- Since 0.0.0
+-- Since 0.2.0
 ($$) :: (IsSource src, Resource m) => src m a -> Sink a m b -> ResourceT m b
 ($$) = connect
 {-# INLINE ($$) #-}
 
 -- | A typeclass allowing us to unify operators for 'Source' and
 -- 'BufferedSource'.
+--
+-- Since 0.2.0
 class IsSource src where
     connect :: Resource m => src m a -> Sink a m b -> ResourceT m b
     fuseLeft :: Resource m => src m a -> Conduit a m b -> Source m b
@@ -89,38 +112,37 @@
     {-# INLINE fuseLeft #-}
 
 normalConnect :: Resource m => Source m a -> Sink a m b -> ResourceT m b
-normalConnect (Source msrc) (Sink msink) = do
-    sinkI <- msink
-    case sinkI of
-        SinkNoData output -> return output
-        SinkData push close -> do
-            src <- msrc
-            connect' src push close
+normalConnect _ (SinkNoData output) = return output
+normalConnect src0 (SinkLift msink) = msink >>= normalConnect src0
+normalConnect src0 (SinkData push0 close0) =
+    connect' src0 push0 close0
   where
-    connect' src push close =
-        loop
-      where
-        loop = do
-            res <- sourcePull src
-            case res of
-                Closed -> do
-                    res' <- close
-                    return res'
-                Open a -> do
-                    mres <- push a
-                    case mres of
-                        Done _leftover res' -> do
-                            sourceClose src
-                            return res'
-                        Processing -> loop
+    connect' src push close = do
+        res <- sourcePull src
+        case res of
+            Closed -> do
+                res' <- close
+                return res'
+            Open src' a -> do
+                mres <- push a
+                case mres of
+                    Done _leftover res' -> do
+                        sourceClose src'
+                        return res'
+                    Processing push' close' -> connect' src' push' close'
 
-data FuseLeftState a = FLClosed [a] | FLOpen [a]
+data FuseLeftState src conduit output =
+    FLClosed [output]
+  | FLOpen src conduit [output]
 
 infixl 1 $=
 
 -- | Left fuse, combining a source and a conduit together into a new source.
 --
--- Since 0.0.0
+-- Note that any @Source@ passed in will be automatically closed, while a
+-- @BufferedSource@ will be left open.
+--
+-- Since 0.2.0
 ($=) :: (IsSource src, Resource m)
      => src m a
      -> Conduit a m b
@@ -129,185 +151,182 @@
 {-# INLINE ($=) #-}
 
 normalFuseLeft :: Resource m => Source m a -> Conduit a m b -> Source m b
-normalFuseLeft (Source msrc) (Conduit mc) = Source $ do
-    istate <- newRef $ FLOpen [] -- still open, no buffer
-    src <- msrc
-    c <- mc
-    return $ PreparedSource
-        (pull istate src c)
-        (close istate src c)
+normalFuseLeft src0 conduit0 = Source
+    { sourcePull = pull $ FLOpen src0 conduit0 []
+    , sourceClose = return ()
+    }
   where
-    pull istate src c = do
-        state' <- readRef istate
+    mkSrc state = Source (pull state) (close state)
+    pull state' =
         case state' of
             FLClosed [] -> return Closed
-            FLClosed (x:xs) -> do
-                writeRef istate $ FLClosed xs
-                return $ Open x
-            FLOpen (x:xs) -> do
-                writeRef istate $ FLOpen xs
-                return $ Open x
-            FLOpen [] -> do
+            FLClosed (x:xs) -> return $ Open
+                (mkSrc (FLClosed xs))
+                x
+            FLOpen src conduit (x:xs) -> return $ Open
+                (mkSrc (FLOpen src conduit xs))
+                x
+            FLOpen src conduit [] -> do
                 mres <- sourcePull src
                 case mres of
                     Closed -> do
-                        res <- conduitClose c
+                        res <- conduitClose conduit
                         case res of
-                            [] -> do
-                                writeRef istate $ FLClosed []
-                                return Closed
-                            x:xs -> do
-                                writeRef istate $ FLClosed xs
-                                return $ Open x
-                    Open input -> do
-                        res' <- conduitPush c input
+                            [] -> return Closed
+                            x:xs -> return $ Open
+                                (mkSrc (FLClosed xs))
+                                x
+                    Open src'' input -> do
+                        res' <- conduitPush conduit input
                         case res' of
-                            Producing [] -> pull istate src c
-                            Producing (x:xs) -> do
-                                writeRef istate $ FLOpen xs
-                                return $ Open x
+                            Producing conduit' [] ->
+                                pull $ FLOpen src'' conduit' []
+                            Producing conduit' (x:xs) -> return $ Open
+                                (mkSrc (FLOpen src'' conduit' xs))
+                                x
                             Finished _leftover output -> do
-                                sourceClose src
+                                sourceClose src''
                                 case output of
-                                    [] -> do
-                                        writeRef istate $ FLClosed []
-                                        return Closed
-                                    x:xs -> do
-                                        writeRef istate $ FLClosed xs
-                                        return $ Open x
-    close istate src c = do
+                                    [] -> return Closed
+                                    x:xs -> return $ Open
+                                        (mkSrc (FLClosed xs))
+                                        x
+    close state = do
         -- See comment on bufferedFuseLeft for why we need to have the
         -- following check
-        state <- readRef istate
         case state of
             FLClosed _ -> return ()
-            FLOpen _ -> do
-                _ignored <- conduitClose c
-                sourceClose src
+            FLOpen src' (Conduit _ closeC) _ -> do
+                _ignored <- closeC
+                sourceClose src'
 
 infixr 0 =$
 
 -- | Right fuse, combining a conduit and a sink together into a new sink.
 --
--- Since 0.0.0
+-- Since 0.2.0
 (=$) :: Resource m => Conduit a m b -> Sink b m c -> Sink a m c
-Conduit mc =$ Sink ms = Sink $ do
-    s <- ms
-    case s of
-        SinkData pushI closeI -> mc >>= go pushI closeI
-        SinkNoData mres -> return $ SinkNoData mres
+_ =$ SinkNoData res = SinkNoData res
+conduit =$ SinkLift msink = SinkLift (liftM (conduit =$) msink)
+conduitOrig =$ SinkData pushI0 closeI0 = SinkData
+    { sinkPush = push pushI0 closeI0 conduitOrig
+    , sinkClose = close pushI0 closeI0 conduitOrig
+    }
   where
-    go pushI closeI c = do
-        return SinkData
-            { sinkPush = \cinput -> do
-                res <- conduitPush c cinput
-                case res of
-                    Producing sinput -> do
-                        let push [] = return Processing
-                            push (i:is) = do
-                                mres <- pushI i
-                                case mres of
-                                    Processing -> push is
-                                    Done _sleftover res' -> do
-                                        _ <- conduitClose c
-                                        return $ Done Nothing res'
-                        push sinput
-                    Finished cleftover sinput -> do
-                        let push [] = closeI
-                            push (i:is) = do
-                                mres <- pushI i
-                                case mres of
-                                    Processing -> push is
-                                    Done _sleftover res' -> return res'
-                        res' <- push sinput
-                        return $ Done cleftover res'
-            , sinkClose = do
-                sinput <- conduitClose c
-                let push [] = closeI
-                    push (i:is) = do
-                        mres <- pushI i
+    push pushI closeI conduit0 cinput = do
+        res <- conduitPush conduit0 cinput
+        case res of
+            Producing conduit' sinput -> do
+                let loop p c [] = return (Processing (push p c conduit') (close p c conduit'))
+                    loop p _ (i:is) = do
+                        mres <- p i
                         case mres of
-                            Processing -> push is
+                            Processing p' c' -> loop p' c' is
+                            Done _sleftover res' -> do
+                                _ <- conduitClose conduit'
+                                return $ Done Nothing res'
+                loop pushI closeI sinput
+            Finished cleftover sinput -> do
+                let loop _ c [] = c
+                    loop p _ (i:is) = do
+                        mres <- p i
+                        case mres of
+                            Processing p' c' -> loop p' c' is
                             Done _sleftover res' -> return res'
-                push sinput
-            }
+                res' <- loop pushI closeI sinput
+                return $ Done cleftover res'
+    close pushI closeI conduit = do
+        sinput <- conduitClose conduit
+        let loop _ c [] = c
+            loop p _ (i:is) = do
+                mres <- p i
+                case mres of
+                    Processing p' c' -> loop p' c' is
+                    Done _sleftover res' -> return res'
+        loop pushI closeI sinput
 
 infixr 0 =$=
 
 -- | Middle fuse, combining two conduits together into a new conduit.
 --
--- Since 0.0.0
+-- Since 0.2.0
 (=$=) :: Resource m => Conduit a m b -> Conduit b m c -> Conduit a m c
-Conduit outerM =$= Conduit innerM = Conduit $ do
-    outer <- outerM
-    inner <- innerM
-    return PreparedConduit
-        { conduitPush = \inputO -> do
-            res <- conduitPush outer inputO
-            case res of
-                Producing inputI -> do
-                    let push [] front = return $ Producing $ front []
-                        push (i:is) front = do
-                            resI <- conduitPush inner i
-                            case resI of
-                                Producing c -> push is (front . (c ++))
-                                Finished _leftover c -> do
-                                    _ <- conduitClose outer
-                                    return $ Finished Nothing $ front c
-                    push inputI id
-                Finished leftoverO inputI -> do
-                    c <- conduitPushClose inner inputI
-                    return $ Finished leftoverO c
-        , conduitClose = do
-            b <- conduitClose outer
-            c <- conduitPushClose inner b
-            return c
-        }
+outerOrig =$= innerOrig = Conduit
+    (pushF outerOrig innerOrig)
+    (closeF outerOrig innerOrig)
+  where
+    pushF outer0 inner0 inputO = do
+        res <- conduitPush outer0 inputO
+        case res of
+            Producing outer inputI -> do
+                let loop inner [] front = return $ Producing
+                        (Conduit (pushF outer inner) (closeF outer inner))
+                        (front [])
+                    loop inner (i:is) front = do
+                        resI <- conduitPush inner i
+                        case resI of
+                            Producing conduit c -> loop
+                                conduit
+                                is
+                                (front . (c ++))
+                            Finished _leftover c -> do
+                                _ <- conduitClose outer
+                                return $ Finished Nothing $ front c
+                loop inner0 inputI id
+            Finished leftoverO inputI -> do
+                c <- conduitPushClose inner0 inputI
+                return $ Finished leftoverO c
+    closeF outer inner = do
+        b <- conduitClose outer
+        c <- conduitPushClose inner b
+        return c
 
 -- | Push some data to a conduit, then close it if necessary.
-conduitPushClose :: Monad m => PreparedConduit a m b -> [a] -> ResourceT m [b]
+conduitPushClose :: Monad m => Conduit a m b -> [a] -> ResourceT m [b]
 conduitPushClose c [] = conduitClose c
 conduitPushClose c (input:rest) = do
     res <- conduitPush c input
     case res of
         Finished _ b -> return b
-        Producing b -> do
-            b' <- conduitPushClose c rest
+        Producing conduit b -> do
+            b' <- conduitPushClose conduit rest
             return $ b ++ b'
 
--- | When actually interacting with 'Source's, we usually want to be able to
+-- | When actually interacting with @Source@s, we sometimes want to be able to
 -- buffer the output, in case any intermediate steps return leftover data. A
--- 'BufferedSource' allows for such buffering.
+-- @BufferedSource@ allows for such buffering.
 --
--- A 'BufferedSource', unlike a 'Source', is resumable, meaning it can be passed to
--- multiple 'Sink's without restarting.
+-- A @BufferedSource@, unlike a @Source@, is resumable, meaning it can be
+-- passed to multiple @Sink@s without restarting. Therefore, a @BufferedSource@
+-- relaxes the main invariant of this package: the same value may be used
+-- multiple times.
 --
--- Finally, a 'BufferedSource' relaxes one of the invariants of a 'Source':
--- pulling after an the source is closed is allowed.
+-- The intention of a @BufferedSource@ is to be used internally by an
+-- application or library, not to be part of its user-facing API. For example,
+-- the Warp webserver uses a @BufferedSource@ internally for parsing the
+-- request headers, but then passes a normal @Source@ to the web application
+-- for reading the request body.
 --
--- A @BufferedSource@ is also known as a /resumable source/, in that it can be
--- called multiple times, and each time will provide new data. One caveat:
--- while the types will allow you to use the buffered source in multiple
--- threads, there is no guarantee that all @BufferedSource@s will handle this
--- correctly.
+-- One caveat: while the types will allow you to use the buffered source in
+-- multiple threads, there is no guarantee that all @BufferedSource@s will
+-- handle this correctly.
 --
--- Since 0.0.0
-data BufferedSource m a = BufferedSource
-    { bsSource :: PreparedSource m a
-    , bsBuffer :: Ref (Base m) (BSState a)
-    }
+-- Since 0.2.0
+data BufferedSource m a = BufferedSource (Ref (Base m) (BSState m a))
 
-data BSState a = ClosedEmpty | OpenEmpty | ClosedFull a | OpenFull a
+data BSState m a =
+    ClosedEmpty
+  | OpenEmpty (Source m a)
+  | ClosedFull a
+  | OpenFull (Source m a) a
 
--- | Prepare a 'Source' and initialize a buffer. Note that you should manually
--- call 'bsourceClose' when the 'BufferedSource' is no longer in use.
+-- | Places the given @Source@ and a buffer into a mutable variable. Note that
+-- you should manually call 'bsourceClose' when the 'BufferedSource' is no
+-- longer in use.
 --
--- Since 0.0.0
+-- Since 0.2.0
 bufferSource :: Resource m => Source m a -> ResourceT m (BufferedSource m a)
-bufferSource (Source msrc) = do
-    src <- msrc
-    buf <- newRef OpenEmpty
-    return $ BufferedSource src buf
+bufferSource src = BufferedSource <$> newRef (OpenEmpty src)
 
 -- | Turn a 'BufferedSource' into a 'Source'. Note that in general this will
 -- mean your original 'BufferedSource' will be closed. Additionally, all
@@ -316,180 +335,176 @@
 --
 -- Note: @bufferSource@ . @unbufferSource@ is /not/ the identity function.
 --
--- Since 0.0.1
+-- Since 0.2.0
 unbufferSource :: Resource m
                => BufferedSource m a
                -> Source m a
-unbufferSource (BufferedSource src bufRef) = Source $ do
-    buf <- readRef bufRef
-    case buf of
-        OpenEmpty -> return src
-        OpenFull a -> do
-            isUsedRef <- newRef False
-            return PreparedSource
-                { sourcePull = do
-                    isUsed <- readRef isUsedRef
-                    if isUsed
-                        then sourcePull src
-                        else do
-                            writeRef isUsedRef True
-                            return $ Open a
+unbufferSource (BufferedSource bs) = Source
+    { sourcePull = msrc >>= sourcePull
+    , sourceClose = msrc >>= sourceClose
+    }
+  where
+    msrc = do
+        buf <- readRef bs
+        case buf of
+            OpenEmpty src -> return src
+            OpenFull src a -> return Source
+                { sourcePull = return $ Open src a
                 , sourceClose = sourceClose src
                 }
-        ClosedEmpty -> return PreparedSource
-            -- Note: we could put some invariant checking in here if we wanted
-            { sourcePull = return Closed
-            , sourceClose = return ()
-            }
-        ClosedFull a -> do
-            isUsedRef <- newRef False
-            return PreparedSource
-                { sourcePull = do
-                    isUsed <- readRef isUsedRef
-                    if isUsed
-                        then return Closed
-                        else do
-                            writeRef isUsedRef True
-                            return $ Open a
-                , sourceClose = sourceClose src
+            ClosedEmpty -> return Source
+                -- Note: we could put some invariant checking in here if we wanted
+                { sourcePull = return Closed
+                , sourceClose = return ()
                 }
+            ClosedFull a -> return Source
+                { sourcePull = return $ Open
+                    (Source (return Closed) (return ()))
+                    a
+                , sourceClose = return ()
+                }
 
 bufferedConnect :: Resource m => BufferedSource m a -> Sink a m b -> ResourceT m b
-bufferedConnect bs (Sink msink) = do
-    sinkI <- msink
-    case sinkI of
-        SinkNoData output -> return output
-        SinkData push close -> do
-            bsState <- readRef $ bsBuffer bs
-            case bsState of
-                ClosedEmpty -> close
-                OpenEmpty -> connect' push close
-                ClosedFull a -> do
-                    res <- push a
-                    case res of
-                        Done mleftover res' -> do
-                            writeRef (bsBuffer bs) $ maybe ClosedEmpty ClosedFull mleftover
-                            return res'
-                        Processing -> do
-                            writeRef (bsBuffer bs) ClosedEmpty
-                            close
-                OpenFull a -> push a >>= onRes (connect' push close)
-  where
-    connect' push close =
-        loop
-      where
-        loop = do
-            res <- sourcePull $ bsSource bs
+bufferedConnect _ (SinkNoData output) = return output
+bufferedConnect bsrc (SinkLift msink) = msink >>= bufferedConnect bsrc
+bufferedConnect (BufferedSource bs) (SinkData push0 close0) = do
+    bsState <- readRef bs
+    case bsState of
+        ClosedEmpty -> close0
+        OpenEmpty src -> connect' src push0 close0
+        ClosedFull a -> do
+            res <- push0 a
             case res of
-                Closed -> do
-                    writeRef (bsBuffer bs) ClosedEmpty
-                    res' <- close
+                Done mleftover res' -> do
+                    writeRef bs $ maybe ClosedEmpty ClosedFull mleftover
                     return res'
-                Open a -> push a >>= onRes loop
-    onRes _ (Done mleftover res) = do
-        writeRef (bsBuffer bs) (maybe OpenEmpty OpenFull mleftover)
+                Processing _ close' -> do
+                    writeRef bs ClosedEmpty
+                    close'
+        OpenFull src a -> push0 a >>= onRes src
+  where
+    connect' src push close = do
+        res <- sourcePull src
+        case res of
+            Closed -> do
+                writeRef bs ClosedEmpty
+                res' <- close
+                return res'
+            Open src' a -> push a >>= onRes src'
+    onRes src (Done mleftover res) = do
+        writeRef bs $ maybe (OpenEmpty src) (OpenFull src) mleftover
         return res
-    onRes loop Processing = loop
+    onRes src (Processing push close) = connect' src push close
 
 bufferedFuseLeft
     :: Resource m
     => BufferedSource m a
     -> Conduit a m b
     -> Source m b
-bufferedFuseLeft bsrc (Conduit mc) = Source $ do
-    istate <- newRef $ FLOpen [] -- still open, no buffer
-    c <- mc
-    return $ PreparedSource
-        (pull istate c)
-        (close istate c)
+bufferedFuseLeft bsrc conduit0 = Source
+    { sourcePull = pullF $ FLOpen () conduit0 [] -- still open, no buffer
+    , sourceClose = return ()
+    }
   where
-    pull istate c = do
-        state' <- readRef istate
+    mkSrc state = Source
+        (pullF state)
+        (closeF state)
+    pullF state' =
         case state' of
             FLClosed [] -> return Closed
-            FLClosed (x:xs) -> do
-                writeRef istate $ FLClosed xs
-                return $ Open x
-            FLOpen (x:xs) -> do
-                writeRef istate $ FLOpen xs
-                return $ Open x
-            FLOpen [] -> do
+            FLClosed (x:xs) -> return $ Open
+                (mkSrc (FLClosed xs))
+                x
+            FLOpen () conduit (x:xs) -> return $ Open
+                (mkSrc (FLOpen () conduit xs))
+                x
+            FLOpen () conduit [] -> do
                 mres <- bsourcePull bsrc
                 case mres of
-                    Closed -> do
-                        res <- conduitClose c
+                    Nothing -> do
+                        res <- conduitClose conduit
                         case res of
-                            [] -> do
-                                writeRef istate $ FLClosed []
-                                return Closed
-                            x:xs -> do
-                                writeRef istate $ FLClosed xs
-                                return $ Open x
-                    Open input -> do
-                        res' <- conduitPush c input
+                            [] -> return Closed
+                            x:xs -> return $ Open
+                                (mkSrc (FLClosed xs))
+                                x
+                    Just input -> do
+                        res' <- conduitPush conduit input
                         case res' of
-                            Producing [] -> pull istate c
-                            Producing (x:xs) -> do
-                                writeRef istate $ FLOpen xs
-                                return $ Open x
+                            Producing conduit' [] ->
+                                pullF (FLOpen () conduit' [])
+                            Producing conduit' (x:xs) -> return $ Open
+                                (mkSrc (FLOpen () conduit' xs))
+                                x
                             Finished leftover output -> do
                                 bsourceUnpull bsrc leftover
                                 case output of
-                                    [] -> do
-                                        writeRef istate $ FLClosed []
-                                        return Closed
-                                    x:xs -> do
-                                        writeRef istate $ FLClosed xs
-                                        return $ Open x
-    close istate c = do
+                                    [] -> return Closed
+                                    x:xs -> return $ Open
+                                        (mkSrc (FLClosed xs))
+                                        x
+    closeF state = do
         -- Normally we don't have to worry about double closing, as the
         -- invariant of a source is that close is never called twice. However,
         -- here, if the Conduit returned Finished with some data, the overall
         -- Source will return an Open while the Conduit will be Closed.
         -- Therefore, we have to do a check.
-        state <- readRef istate
         case state of
             FLClosed _ -> return ()
-            FLOpen _ -> do
-                _ignored <- conduitClose c
+            FLOpen () (Conduit _ close) _ -> do
+                _ignored <- close
                 return ()
 
-bsourcePull :: Resource m => BufferedSource m a -> ResourceT m (SourceResult a)
-bsourcePull (BufferedSource src bufRef) = do
-    buf <- readRef bufRef
+bsourcePull :: Resource m => BufferedSource m a -> ResourceT m (Maybe a)
+bsourcePull (BufferedSource bs) = do
+    buf <- readRef bs
     case buf of
-        OpenEmpty -> do
+        OpenEmpty src -> do
             res <- sourcePull src
             case res of
-                Open _ -> return res
-                Closed -> writeRef bufRef ClosedEmpty >> return Closed
-        ClosedEmpty -> return Closed
-        OpenFull a -> do
-            writeRef bufRef OpenEmpty
-            return $ Open a
+                Open src' a -> do
+                    writeRef bs $ OpenEmpty src'
+                    return $ Just a
+                Closed -> writeRef bs ClosedEmpty >> return Nothing
+        ClosedEmpty -> return Nothing
+        OpenFull src a -> do
+            writeRef bs (OpenEmpty src)
+            return $ Just a
         ClosedFull a -> do
-            writeRef bufRef ClosedEmpty
-            return $ Open a
+            writeRef bs ClosedEmpty
+            return $ Just a
 
 bsourceUnpull :: Resource m => BufferedSource m a -> Maybe a -> ResourceT m ()
 bsourceUnpull _ Nothing = return ()
-bsourceUnpull (BufferedSource _ bufRef) (Just a) = do
-    buf <- readRef bufRef
+bsourceUnpull (BufferedSource ref) (Just a) = do
+    buf <- readRef ref
     case buf of
-        OpenEmpty -> writeRef bufRef $ OpenFull a
-        ClosedEmpty -> writeRef bufRef $ ClosedFull a
+        OpenEmpty src -> writeRef ref (OpenFull src a)
+        ClosedEmpty -> writeRef ref (ClosedFull a)
         _ -> error $ "Invariant violated: bsourceUnpull called on full data"
 
--- | Close the underlying 'PreparedSource' for the given 'BufferedSource'. Note
+-- | Close the underlying 'Source' for the given 'BufferedSource'. Note
 -- that this function can safely be called multiple times, as it will first
--- check if the 'PreparedSource' was previously closed.
+-- check if the 'Source' was previously closed.
 --
--- Since 0.0.0
+-- Since 0.2.0
 bsourceClose :: Resource m => BufferedSource m a -> ResourceT m ()
-bsourceClose (BufferedSource src bufRef) = do
-    buf <- readRef bufRef
+bsourceClose (BufferedSource ref) = do
+    buf <- readRef ref
     case buf of
-        OpenEmpty -> sourceClose src
-        OpenFull _ -> sourceClose src
+        OpenEmpty src -> sourceClose src
+        OpenFull src _ -> sourceClose src
         ClosedEmpty -> return ()
         ClosedFull _ -> return ()
+-- | Provide for a stream of data that can be flushed.
+--
+-- A number of @Conduit@s (e.g., zlib compression) need the ability to flush
+-- the stream at some point. This provides a single wrapper datatype to be used
+-- in all such circumstances.
+--
+-- Since 0.2.0
+data Flush a = Chunk a | Flush
+    deriving (Show, Eq, Ord)
+instance Functor Flush where
+    fmap _ Flush = Flush
+    fmap f (Chunk a) = Chunk (f a)
diff --git a/Data/Conduit/Binary.hs b/Data/Conduit/Binary.hs
--- a/Data/Conduit/Binary.hs
+++ b/Data/Conduit/Binary.hs
@@ -16,6 +16,7 @@
     , takeWhile
     , dropWhile
     , take
+    , Data.Conduit.Binary.lines
     ) where
 
 import Prelude hiding (head, take, takeWhile, dropWhile)
@@ -27,9 +28,7 @@
 import Control.Monad (liftM)
 import Control.Monad.IO.Class (liftIO)
 import qualified System.IO as IO
-import Control.Monad.Trans.Resource
-    ( withIO, release, newRef, readRef, writeRef
-    )
+import Control.Monad.Trans.Resource (withIO, release)
 import Data.Word (Word8)
 #if CABAL_OS_WINDOWS
 import qualified System.Win32File as F
@@ -43,7 +42,7 @@
 -- While you are not required to call @hClose@ on the resulting handle, you
 -- should do so as early as possible to free scarce resources.
 --
--- Since 0.0.2
+-- Since 0.2.0
 openFile :: ResourceIO m
          => FilePath
          -> IO.IOMode
@@ -52,7 +51,7 @@
 
 -- | Stream the contents of a file as binary data.
 --
--- Since 0.0.0
+-- Since 0.2.0
 sourceFile :: ResourceIO m
            => FilePath
            -> Source m S.ByteString
@@ -69,25 +68,29 @@
 -- function will /not/ automatically close the @Handle@ when processing
 -- completes, since it did not acquire the @Handle@ in the first place.
 --
--- Since 0.0.2.
+-- Since 0.2.0
 sourceHandle :: ResourceIO m
              => IO.Handle
              -> Source m S.ByteString
-sourceHandle h = Source $ return $ PreparedSource
-    { sourcePull = do
+sourceHandle h =
+    src
+  where
+    src = Source pull close
+
+    pull = do
         bs <- liftIO (S.hGetSome h 4096)
         if S.null bs
             then return Closed
-            else return (Open bs)
-    , sourceClose = return ()
-    }
+            else return $ Open src bs
 
+    close = return ()
+
 -- | An alternative to 'sourceHandle'.
 -- Instead of taking a pre-opened 'IO.Handle', it takes an action that opens
 -- a 'IO.Handle' (in read mode), so that it can open it only when needed
 -- and close it as soon as possible.
 --
--- Since 0.1.1
+-- Since 0.2.0
 sourceIOHandle :: ResourceIO m
                => IO IO.Handle
                -> Source m S.ByteString
@@ -95,58 +98,55 @@
     (\handle -> do
         bs <- liftIO (S.hGetSome handle 4096)
         if S.null bs
-            then return Closed
-            else return $ Open bs)
+            then return IOClosed
+            else return $ IOOpen bs)
 
 -- | Stream all incoming data to the given 'IO.Handle'. Note that this function
 -- will /not/ automatically close the @Handle@ when processing completes.
 --
--- Since 0.0.2.
+-- Since 0.2.0
 sinkHandle :: ResourceIO m
            => IO.Handle
            -> Sink S.ByteString m ()
-sinkHandle h = Sink $ return $ SinkData
-    { sinkPush = \input -> liftIO (S.hPut h input) >> return Processing
-    , sinkClose = return ()
-    }
+sinkHandle h =
+    SinkData push close
+  where
+    push input = liftIO (S.hPut h input) >> return (Processing push close)
+    close = return ()
 
 -- | An alternative to 'sinkHandle'.
 -- Instead of taking a pre-opened 'IO.Handle', it takes an action that opens
 -- a 'IO.Handle' (in write mode), so that it can open it only when needed
 -- and close it as soon as possible.
 --
--- Since 0.1.1
+-- Since 0.2.0
 sinkIOHandle :: ResourceIO m
              => IO IO.Handle
              -> Sink S.ByteString m ()
 sinkIOHandle alloc = sinkIO alloc IO.hClose
-    (\handle bs -> liftIO (S.hPut handle bs) >> return Processing)
+    (\handle bs -> liftIO (S.hPut handle bs) >> return IOProcessing)
     (const $ return ())
 
 -- | Stream the contents of a file as binary data, starting from a certain
 -- offset and only consuming up to a certain number of bytes.
 --
--- Since 0.0.0
+-- Since 0.2.0
 sourceFileRange :: ResourceIO m
                 => FilePath
                 -> Maybe Integer -- ^ Offset
                 -> Maybe Integer -- ^ Maximum count
                 -> Source m S.ByteString
-sourceFileRange fp offset count = Source $ do
-    (key, handle) <- withIO (IO.openBinaryFile fp IO.ReadMode) IO.hClose
-    case offset of
-        Nothing -> return ()
-        Just off -> liftIO $ IO.hSeek handle IO.AbsoluteSeek off
-    pull <-
+sourceFileRange fp offset count = Source
+    { sourcePull = do
+        (key, handle) <- withIO (IO.openBinaryFile fp IO.ReadMode) IO.hClose
+        case offset of
+            Nothing -> return ()
+            Just off -> liftIO $ IO.hSeek handle IO.AbsoluteSeek off
         case count of
-            Nothing -> return $ pullUnlimited handle key
-            Just c -> do
-                ic <- newRef c
-                return $ pullLimited ic handle key
-    return PreparedSource
-        { sourcePull = pull
-        , sourceClose = release key
-        }
+            Nothing -> pullUnlimited handle key
+            Just c -> pullLimited c handle key
+    , sourceClose = return ()
+    }
   where
     pullUnlimited handle key = do
         bs <- liftIO $ S.hGetSome handle 4096
@@ -154,9 +154,15 @@
             then do
                 release key
                 return Closed
-            else return $ Open bs
-    pullLimited ic handle key = do
-        c <- fmap fromInteger $ readRef ic
+            else do
+                let src = Source
+                        { sourcePull = pullUnlimited handle key
+                        , sourceClose = release key
+                        }
+                return $ Open src bs
+
+    pullLimited c0 handle key = do
+        let c = fromInteger c0
         bs <- liftIO $ S.hGetSome handle (min c 4096)
         let c' = c - S.length bs
         assert (c' >= 0) $
@@ -165,12 +171,15 @@
                     release key
                     return Closed
                 else do
-                    writeRef ic $ toInteger c'
-                    return $ Open bs
+                    let src = Source
+                            { sourcePull = pullLimited (toInteger c') handle key
+                            , sourceClose = release key
+                            }
+                    return $ Open src bs
 
 -- | Stream all incoming data to the given file.
 --
--- Since 0.0.0
+-- Since 0.2.0
 sinkFile :: ResourceIO m
          => FilePath
          -> Sink S.ByteString m ()
@@ -179,7 +188,7 @@
 -- | Stream the contents of the input to a file, and also send it along the
 -- pipeline. Similar in concept to the Unix command @tee@.
 --
--- Since 0.0.0
+-- Since 0.2.0
 conduitFile :: ResourceIO m
             => FilePath
             -> Conduit S.ByteString m S.ByteString
@@ -188,14 +197,14 @@
     IO.hClose
     (\handle bs -> do
         liftIO $ S.hPut handle bs
-        return $ Producing [bs])
+        return $ IOProducing [bs])
     (const $ return [])
 
 -- | Ensure that only up to the given number of bytes are consume by the inner
 -- sink. Note that this does /not/ ensure that all of those bytes are in fact
 -- consumed.
 --
--- Since 0.0.0
+-- Since 0.2.0
 isolate :: Resource m
         => Int
         -> Conduit S.ByteString m S.ByteString
@@ -204,60 +213,90 @@
     push
     close
   where
-    push 0 bs = return (0, Finished (Just bs) [])
+    push 0 bs = return $ StateFinished (Just bs) []
     push count bs = do
         let (a, b) = S.splitAt count bs
         let count' = count - S.length a
-        return (count',
+        return $
             if count' == 0
-                then Finished (if S.null b then Nothing else Just b) (if S.null a then [] else [a])
-                else assert (S.null b) $ Producing [a])
+                then StateFinished (if S.null b then Nothing else Just b) (if S.null a then [] else [a])
+                else assert (S.null b) $ StateProducing count' [a]
     close _ = return []
 
 -- | Return the next byte from the stream, if available.
 --
--- Since 0.0.2
+-- Since 0.2.0
 head :: Resource m => Sink S.ByteString m (Maybe Word8)
-head = Sink $ return $ SinkData
-    { sinkPush = \bs ->
+head =
+    SinkData push close
+  where
+    push bs =
         case S.uncons bs of
-            Nothing -> return Processing
+            Nothing -> return $ Processing push close
             Just (w, bs') -> do
                 let lo = if S.null bs' then Nothing else Just bs'
                 return $ Done lo (Just w)
-    , sinkClose = return Nothing
-    }
+    close = return Nothing
 
 -- | Return all bytes while the predicate returns @True@.
 --
--- Since 0.0.2
+-- Since 0.2.0
 takeWhile :: Resource m => (Word8 -> Bool) -> Conduit S.ByteString m S.ByteString
-takeWhile p = Conduit $ return $ PreparedConduit
-    { conduitPush = \bs -> do
+takeWhile p =
+    conduit
+  where
+    conduit = Conduit push close
+    push bs = do
         let (x, y) = S.span p bs
         return $
             if S.null y
-                then Producing [x]
+                then Producing conduit [x]
                 else Finished (Just y) (if S.null x then [] else [x])
-    , conduitClose = return []
-    }
+    close = return []
 
 -- | Ignore all bytes while the predicate returns @True@.
 --
--- Since 0.0.2
+-- Since 0.2.0
 dropWhile :: Resource m => (Word8 -> Bool) -> Sink S.ByteString m ()
-dropWhile p = Sink $ return $ SinkData
-    { sinkPush = \bs -> do
+dropWhile p =
+    SinkData push close
+  where
+    push bs = do
         let bs' = S.dropWhile p bs
         return $
             if S.null bs'
-                then Processing
+                then Processing push close
                 else Done (Just bs') ()
-    , sinkClose = return ()
-    }
+    close = return ()
 
 -- | Take the given number of bytes, if available.
 --
--- Since 0.0.3
+-- Since 0.2.0
 take :: Resource m => Int -> Sink S.ByteString m L.ByteString
 take n = L.fromChunks `liftM` (isolate n =$ CL.consume)
+
+-- | Split the input bytes into lines. In other words, split on the LF byte
+-- (10), and strip it from the output.
+--
+-- Since 0.2.0
+lines :: Resource m => Conduit S.ByteString m S.ByteString
+lines =
+    conduitState id push close
+  where
+    push front bs' = return $ StateProducing leftover ls
+      where
+        bs = front bs'
+        (leftover, ls) = getLines id bs
+
+    getLines front bs
+        | S.null bs = (id, front [])
+        | S.null y = (S.append x, front [])
+        | otherwise = getLines (front . (x:)) (S.drop 1 y)
+      where
+        (x, y) = S.breakByte 10 bs
+
+    close front
+        | S.null bs = return []
+        | otherwise = return [bs]
+      where
+        bs = front S.empty
diff --git a/Data/Conduit/Lazy.hs b/Data/Conduit/Lazy.hs
--- a/Data/Conduit/Lazy.hs
+++ b/Data/Conduit/Lazy.hs
@@ -12,17 +12,16 @@
 
 -- | Use lazy I\/O to consume all elements from a @Source@.
 --
--- Since 0.0.0
+-- Since 0.2.0
 lazyConsume :: MonadBaseControl IO m => Source m a -> ResourceT m [a]
-lazyConsume (Source msrc) = do
-    src <- msrc
-    go src
+lazyConsume src0 = do
+    go src0
   where
 
     go src = liftBaseOp_ unsafeInterleaveIO $ do
         res <- sourcePull src
         case res of
             Closed -> return []
-            Open x -> do
-                y <- go src
+            Open src' x -> do
+                y <- go src'
                 return $ x : y
diff --git a/Data/Conduit/List.hs b/Data/Conduit/List.hs
--- a/Data/Conduit/List.hs
+++ b/Data/Conduit/List.hs
@@ -28,12 +28,14 @@
       -- ** Pure
     , map
     , concatMap
+    , concatMapAccum
     , groupBy
     , isolate
     , filter
       -- ** Monadic
     , mapM
     , concatMapM
+    , concatMapAccumM
     ) where
 
 import Prelude
@@ -49,19 +51,19 @@
 
 -- | A strict left fold.
 --
--- Since 0.0.0
+-- Since 0.2.0
 fold :: Resource m
      => (b -> a -> b)
      -> b
      -> Sink a m b
 fold f accum0 = sinkState
     accum0
-    (\accum input -> return (f accum input, Processing))
+    (\accum input -> return (StateProcessing $ f accum input))
     return
 
 -- | A monadic strict left fold.
 --
--- Since 0.0.0
+-- Since 0.2.0
 foldM :: Resource m
       => (b -> a -> m b)
       -> b
@@ -70,29 +72,31 @@
     accum0
     (\accum input -> do
         accum' <- lift $ f accum input
-        return (accum', Processing)
+        return $ StateProcessing accum'
     )
     return
 
 -- | Apply the action to all values in the stream.
 --
--- Since 0.0.0
+-- Since 0.2.0
 mapM_ :: Resource m
       => (a -> m ())
       -> Sink a m ()
-mapM_ f = Sink $ return $ SinkData
-    (\input -> lift (f input) >> return Processing)
-    (return ())
+mapM_ f =
+    SinkData push close
+  where
+    push input = lift (f input) >> return (Processing push close)
+    close = return ()
 
 -- | Convert a list into a source.
 --
--- Since 0.0.0
+-- Since 0.2.0
 sourceList :: Resource m => [a] -> Source m a
 sourceList l0 =
     sourceState l0 go
   where
-    go [] = return ([], Closed)
-    go (x:xs) = return (xs, Open x)
+    go [] = return StateClosed
+    go (x:xs) = return $ StateOpen xs x
 
 -- | Ignore a certain number of values in the stream. This function is
 -- semantically equivalent to:
@@ -102,7 +106,7 @@
 -- However, @drop@ is more efficient as it does not need to hold values in
 -- memory.
 --
--- Since 0.0.0
+-- Since 0.2.0
 drop :: Resource m
      => Int
      -> Sink a m ()
@@ -111,12 +115,12 @@
     push
     close
   where
-    push 0 x = return (0, Done (Just x) ())
+    push 0 x = return $ StateDone (Just x) ()
     push count _ = do
         let count' = count - 1
-        return (count', if count' == 0
-                            then Done Nothing ()
-                            else Processing)
+        return $ if count' == 0
+            then StateDone Nothing ()
+            else StateProcessing count'
     close _ = return ()
 
 -- | Take some values from the stream and return as a list. If you want to
@@ -125,7 +129,7 @@
 --
 -- > take i = isolate i =$ consume
 --
--- Since 0.0.0
+-- Since 0.2.0
 take :: Resource m
      => Int
      -> Sink a m [a]
@@ -134,22 +138,21 @@
     push
     close
   where
-    push (0, front) x = return ((0, front), Done (Just x) (front []))
+    push (0, front) x = return (StateDone (Just x) (front []))
     push (count, front) x = do
         let count' = count - 1
             front' = front . (x:)
-            res = if count' == 0
-                    then Done Nothing (front' [])
-                    else Processing
-        return ((count', front'), res)
+        return $ if count' == 0
+                    then StateDone Nothing (front' [])
+                    else StateProcessing (count', front')
     close (_, front) = return $ front []
 
 -- | Take a single value from the stream, if available.
 --
--- Since 0.0.0
+-- Since 0.2.0
 head :: Resource m => Sink a m (Maybe a)
 head =
-    Sink $ return $ SinkData push close
+    SinkData push close
   where
     push x = return $ Done Nothing (Just x)
     close = return Nothing
@@ -157,81 +160,109 @@
 -- | Look at the next value in the stream, if available. This function will not
 -- change the state of the stream.
 --
--- Since 0.0.0
+-- Since 0.2.0
 peek :: Resource m => Sink a m (Maybe a)
 peek =
-    Sink $ return $ SinkData push close
+    SinkData push close
   where
     push x = return $ Done (Just x) (Just x)
     close = return Nothing
 
 -- | Apply a transformation to all values in a stream.
 --
--- Since 0.0.0
+-- Since 0.2.0
 map :: Monad m => (a -> b) -> Conduit a m b
-map f = Conduit $ return $ PreparedConduit
-    { conduitPush = return . Producing . return . f
-    , conduitClose = return []
-    }
+map f =
+    conduit
+  where
+    conduit = Conduit push close
+    push = return . Producing conduit . return . f
+    close = return []
 
 -- | Apply a monadic transformation to all values in a stream.
 --
 -- If you do not need the transformed values, and instead just want the monadic
 -- side-effects of running the action, see 'mapM_'.
 --
--- Since 0.0.0
+-- Since 0.2.0
 mapM :: Monad m => (a -> m b) -> Conduit a m b
-mapM f = Conduit $ return $ PreparedConduit
-    { conduitPush = fmap (Producing . return) . lift . f
-    , conduitClose = return []
-    }
+mapM f =
+    conduit
+  where
+    conduit = Conduit push close
+    push = fmap (Producing conduit . return) . lift . f
+    close = return []
 
 -- | Apply a transformation to all values in a stream, concatenating the output
 -- values.
 --
--- Since 0.0.0
+-- Since 0.2.0
 concatMap :: Monad m => (a -> [b]) -> Conduit a m b
-concatMap f = Conduit $ return $ PreparedConduit
-    { conduitPush = return . Producing . f
-    , conduitClose = return []
-    }
+concatMap f =
+    conduit
+  where
+    conduit = Conduit push close
+    push = return . Producing conduit . f
+    close = return []
 
 -- | Apply a monadic transformation to all values in a stream, concatenating
 -- the output values.
 --
--- Since 0.0.0
+-- Since 0.2.0
 concatMapM :: Monad m => (a -> m [b]) -> Conduit a m b
-concatMapM f = Conduit $ return $ PreparedConduit
-    { conduitPush = fmap Producing . lift . f
-    , conduitClose = return []
-    }
+concatMapM f =
+    conduit
+  where
+    conduit = Conduit push close
+    push = fmap (Producing conduit) . lift . f
+    close = return []
 
+-- | 'concatMap' with accumerator.
+--
+-- Since 0.2.0
+concatMapAccum :: Resource m => (a -> accum -> (accum, [b])) -> accum -> Conduit a m b
+concatMapAccum f accum = conduitState accum push close
+  where
+    push state input = let (state', result) = f input state
+                       in return $ StateProducing state' result
+    close _ = return []
+
+-- | 'concatMapM' with accumerator.
+--
+-- Since 0.2.0
+concatMapAccumM :: Resource m => (a -> accum -> m (accum, [b])) -> accum -> Conduit a m b
+concatMapAccumM f accum = conduitState accum push close
+  where
+    push state input = do (state', result) <- lift (f input state)
+                          return $ StateProducing state' result
+    close _ = return []
+
 -- | Consume all values from the stream and return as a list. Note that this
 -- will pull all values into memory. For a lazy variant, see
 -- "Data.Conduit.Lazy".
 --
--- Since 0.0.0
+-- Since 0.2.0
 consume :: Resource m => Sink a m [a]
 consume = sinkState
     id
-    (\front input -> return (front . (input :), Processing))
+    (\front input -> return (StateProcessing $ front . (input :)))
     (\front -> return $ front [])
 
 -- | Grouping input according to an equality function.
 --
--- Since 0.0.2
+-- Since 0.2.0
 groupBy :: Resource m => (a -> a -> Bool) -> Conduit a m [a]
 groupBy f = conduitState
     []
     push
     close
   where
-    push []      v = return ([v], Producing [])
+    push []      v = return $ StateProducing [v] []
     push s@(x:_) v =
       if f x v then
-        return (v:s, Producing [])
+        return $ StateProducing (v:s) []
       else
-        return ([v], Producing [s])
+        return $ StateProducing [v] [s]
     close s = return [s]
 
 -- | Ensure that the inner sink consumes no more than the given number of
@@ -246,7 +277,7 @@
 -- >     someOtherSink
 -- >     ...
 --
--- Since 0.0.0
+-- Since 0.2.0
 isolate :: Resource m => Int -> Conduit a m a
 isolate count0 = conduitState
     count0
@@ -256,35 +287,38 @@
     close _ = return []
     push count x = do
         if count == 0
-            then return (count, Finished (Just x) [])
+            then return $ StateFinished (Just x) []
             else do
                 let count' = count - 1
-                return (count',
-                    if count' == 0
-                        then Finished Nothing [x]
-                        else Producing [x])
+                return $ if count' == 0
+                    then StateFinished Nothing [x]
+                    else StateProducing count' [x]
 
 -- | Keep only values in the stream passing a given predicate.
 --
--- Since 0.0.0
+-- Since 0.2.0
 filter :: Resource m => (a -> Bool) -> Conduit a m a
-filter f = Conduit $ return $ PreparedConduit
-    { conduitPush = return . Producing . Prelude.filter f . return
-    , conduitClose = return []
-    }
+filter f =
+    conduit
+  where
+    conduit = Conduit push close
+    push = return . Producing conduit . Prelude.filter f . return
+    close = return []
 
 -- | Ignore the remainder of values in the source. Particularly useful when
 -- combined with 'isolate'.
 --
--- Since 0.0.0
+-- Since 0.2.0
 sinkNull :: Resource m => Sink a m ()
-sinkNull = Sink $ return $ SinkData
-    (\_ -> return Processing)
-    (return ())
+sinkNull =
+    SinkData push close
+  where
+    push _ = return $ Processing push close
+    close = return ()
 
 -- | A source that returns nothing. Note that this is just a type-restricted
 -- synonym for 'mempty'.
 --
--- Since 0.0.4
+-- Since 0.2.0
 sourceNull :: Resource m => Source m a
 sourceNull = mempty
diff --git a/Data/Conduit/Text.hs b/Data/Conduit/Text.hs
--- a/Data/Conduit/Text.hs
+++ b/Data/Conduit/Text.hs
@@ -47,7 +47,7 @@
 
 -- | A specific character encoding.
 --
--- Since 0.0.0
+-- Since 0.2.0
 data Codec = Codec
     { codecName :: T.Text
     , codecEncode
@@ -67,7 +67,7 @@
 -- | Convert text into bytes, using the provided codec. If the codec is
 -- not capable of representing an input character, an exception will be thrown.
 --
--- Since 0.0.0
+-- Since 0.2.0
 encode :: ResourceThrow m => Codec -> C.Conduit T.Text m B.ByteString
 encode codec = CL.mapM $ \t -> do
     let (bs, mexc) = codecEncode codec t
@@ -77,7 +77,7 @@
 -- | Convert bytes into text, using the provided codec. If the codec is
 -- not capable of decoding an input byte sequence, an exception will be thrown.
 --
--- Since 0.0.0
+-- Since 0.2.0
 decode :: ResourceThrow m => Codec -> C.Conduit B.ByteString m T.Text
 decode codec = C.conduitState
     Nothing
@@ -86,7 +86,7 @@
   where
     push mb input = do
         (mb', ts) <- go' mb input
-        return $ (mb', C.Producing ts)
+        return $ C.StateProducing mb' ts
     close mb =
         case mb of
             Nothing -> return []
@@ -115,7 +115,7 @@
         front' = front . (text:)
 
 -- |
--- Since 0.0.0
+-- Since 0.2.0
 data TextException = DecodeException Codec Word8
                    | EncodeException Codec Char
     deriving (Show, Typeable)
@@ -148,7 +148,7 @@
             Right _ -> Right B.empty)
 
 -- |
--- Since 0.0.0
+-- Since 0.2.0
 utf8 :: Codec
 utf8 = Codec name enc dec where
     name = T.pack "UTF-8"
@@ -181,7 +181,7 @@
                     else decodeMore
 
 -- |
--- Since 0.0.0
+-- Since 0.2.0
 utf16_le :: Codec
 utf16_le = Codec name enc dec where
     name = T.pack "UTF-16-LE"
@@ -208,7 +208,7 @@
         decodeAll = (TE.decodeUtf16LE bytes, B.empty)
 
 -- |
--- Since 0.0.0
+-- Since 0.2.0
 utf16_be :: Codec
 utf16_be = Codec name enc dec where
     name = T.pack "UTF-16-BE"
@@ -243,7 +243,7 @@
     x = (fromIntegral x1 `shiftL` 8) .|. fromIntegral x0
 
 -- |
--- Since 0.0.0
+-- Since 0.2.0
 utf32_le :: Codec
 utf32_le = Codec name enc dec where
     name = T.pack "UTF-32-LE"
@@ -253,7 +253,7 @@
         Nothing -> splitSlowly TE.decodeUtf32LE bs
 
 -- |
--- Since 0.0.0
+-- Since 0.2.0
 utf32_be :: Codec
 utf32_be = Codec name enc dec where
     name = T.pack "UTF-32-BE"
@@ -276,7 +276,7 @@
         else B.splitAt lenToDecode bytes
 
 -- |
--- Since 0.0.0
+-- Since 0.2.0
 ascii :: Codec
 ascii = Codec name enc dec where
     name = T.pack "ASCII"
@@ -295,7 +295,7 @@
             else Left (DecodeException ascii (B.head unsafe), unsafe)
 
 -- |
--- Since 0.0.0
+-- Since 0.2.0
 iso8859_1 :: Codec
 iso8859_1 = Codec name enc dec where
     name = T.pack "ISO-8859-1"
diff --git a/Data/Conduit/Types/Conduit.hs b/Data/Conduit/Types/Conduit.hs
--- a/Data/Conduit/Types/Conduit.hs
+++ b/Data/Conduit/Types/Conduit.hs
@@ -2,51 +2,48 @@
 -- is almost always connected either left (to a source) or right (to a sink).
 module Data.Conduit.Types.Conduit
     ( ConduitResult (..)
-    , PreparedConduit (..)
     , Conduit (..)
+    , ConduitPush
+    , ConduitClose
     ) where
 
 import Control.Monad.Trans.Resource (ResourceT)
 import Control.Monad (liftM)
 
+-- | The value of the @conduitPush@ record.
+type ConduitPush input m output = input -> ResourceT m (ConduitResult input m output)
+
+-- | The value of the @conduitClose@ record.
+type ConduitClose m output = ResourceT m [output]
+
 -- | When data is pushed to a @Conduit@, it may either indicate that it is
 -- still producing output and provide some, or indicate that it is finished
 -- producing output, in which case it returns optional leftover input and some
 -- final output.
 --
--- Since 0.0.0
-data ConduitResult input output = Producing [output] | Finished (Maybe input) [output]
+-- The @Producing@ constructor provides a new @Conduit@ to be used in place of
+-- the previous one.
+--
+-- Since 0.2.0
+data ConduitResult input m output =
+    Producing (Conduit input m output) [output]
+  | Finished (Maybe input) [output]
 
-instance Functor (ConduitResult input) where
-    fmap f (Producing o) = Producing (fmap f o)
+instance Monad m => Functor (ConduitResult input m) where
+    fmap f (Producing c o) = Producing (fmap f c) (fmap f o)
     fmap f (Finished i o) = Finished i (fmap f o)
 
 -- | A conduit has two operations: it can receive new input (a push), and can
 -- be closed.
 --
--- Invariants:
---
--- * Neither a push nor close may be performed after a conduit returns a
--- 'Finished' from a push, or after a close is performed.
---
--- Since 0.0.0
-data PreparedConduit input m output = PreparedConduit
-    { conduitPush :: input -> ResourceT m (ConduitResult input output)
-    , conduitClose :: ResourceT m [output]
+-- Since 0.2.0
+data Conduit input m output = Conduit
+    { conduitPush :: ConduitPush input m output
+    , conduitClose :: ConduitClose m output
     }
 
-instance Monad m => Functor (PreparedConduit input m) where
+instance Monad m => Functor (Conduit input m) where
     fmap f c = c
         { conduitPush = liftM (fmap f) . conduitPush c
         , conduitClose = liftM (fmap f) (conduitClose c)
         }
-
--- | A monadic action generating a 'PreparedConduit'. See @Source@ and @Sink@
--- for more motivation.
---
--- Since 0.0.0
-newtype Conduit input m output =
-    Conduit { prepareConduit :: ResourceT m (PreparedConduit input m output) }
-
-instance Monad m => Functor (Conduit input m) where
-    fmap f (Conduit mc) = Conduit (liftM (fmap f) mc)
diff --git a/Data/Conduit/Types/Sink.hs b/Data/Conduit/Types/Sink.hs
--- a/Data/Conduit/Types/Sink.hs
+++ b/Data/Conduit/Types/Sink.hs
@@ -6,28 +6,66 @@
 -- | Defines the types for a sink, which is a consumer of data.
 module Data.Conduit.Types.Sink
     ( SinkResult (..)
-    , PreparedSink (..)
     , Sink (..)
+    , SinkPush
+    , SinkClose
     ) where
 
 import Control.Monad.Trans.Resource
 import Control.Monad.Trans.Class (MonadTrans (lift))
 import Control.Monad.IO.Class (MonadIO (liftIO))
-import Control.Monad (liftM)
+import Control.Monad (liftM, ap)
 import Control.Applicative (Applicative (..))
 import Control.Monad.Base (MonadBase (liftBase))
 
+-- | The value of the @sinkPush@ record.
+type SinkPush input m output = input -> ResourceT m (SinkResult input m output)
+
+-- | The value of the @sinkClose@ record.
+type SinkClose m output = ResourceT m output
+
 -- | A @Sink@ ultimately returns a single output value. Each time data is
 -- pushed to it, a @Sink@ may indicate that it is still processing data, or
 -- that it is done, in which case it returns some optional leftover input and
 -- an output value.
 --
--- Since 0.0.0
-data SinkResult input output = Processing | Done (Maybe input) output
-instance Functor (SinkResult input) where
-    fmap _ Processing = Processing
+-- The @Processing@ constructors provides updated push and close functions to
+-- be used in place of the original @Sink@.
+--
+-- Since 0.2.0
+data SinkResult input m output =
+    Processing (SinkPush input m output) (SinkClose m output)
+  | Done (Maybe input) output
+instance Monad m => Functor (SinkResult input m) where
+    fmap f (Processing push close) = Processing ((fmap . fmap . fmap) f push) (fmap f close)
     fmap f (Done input output) = Done input (f output)
 
+{-
+Note to my future self, and anyone else who reads my code: It's tempting to
+change `Sink` to look like:
+
+    newtype Sink input m output = Sink { runSink :: ResourceT m (SinkResult input m output) }
+
+If you start implementing this, eventually you'll realize that you will have to
+enforce an invariant to make it all work: a `SinkResult` can't return leftovers
+unless data was pushed to it.
+
+The idea is that, with the actual definition of `Sink`, it's impossible to get
+a `SinkResult` without first pushing in some input. Therefore, it's always
+valid at the type level to return leftovers. In this simplified `Sink`, it
+would be possible to have code that looks like:
+
+    sink1 = Sink $ return $ Done (Just "foo") ()
+    fsink2 () = Sink $ return $ Done (Just "bar") ()
+    sink1 >>= fsink2
+
+Now we'd have to coalesce "foo" and "bar" together (e.g., require `Monoid`),
+throw away data, or throw an exception.
+
+So the current three-constructor approach to `Sink` may not be as pretty, but
+it enforce the invariants much better.
+-}
+
 -- | In general, a sink will consume data and eventually produce an output when
 -- it has consumed \"enough\" data. There are two caveats to that statement:
 --
@@ -44,155 +82,67 @@
 -- cannot always produce output, this should be indicated in its return value,
 -- using something like a 'Maybe' or 'Either'.
 --
--- Invariants:
---
--- * After a 'PreparedSink' produces a result (either via 'sinkPush' or
--- 'sinkClose'), neither of those two functions may be called on the @Sink@
--- again.
---
--- * If a @Sink@ needs to clean up any resources (e.g., close a file handle),
--- it must do so whenever it returns a result, either via @sinkPush@ or
--- @sinkClose@. Note that, due to usage of @ResourceT@, this is merely an
--- optimization.
+-- A @Sink@ should clean up any resources it has allocated when it returns a
+-- value, whether that be via @sinkPush@ or @sinkClose@.
 --
--- Since 0.0.0
-data PreparedSink input m output =
+-- Since 0.2.0
+data Sink input m output =
     SinkNoData output
   | SinkData
-        { sinkPush :: input -> ResourceT m (SinkResult input output)
-        , sinkClose :: ResourceT m output
+        { sinkPush :: SinkPush input m output
+        , sinkClose :: SinkClose m output
         }
+  -- | This constructor is provided to allow us to create an efficient
+  -- @MonadTrans@ instance.
+  | SinkLift (ResourceT m (Sink input m output))
 
-instance Monad m => Functor (PreparedSink input m) where
+instance Monad m => Functor (Sink input m) where
     fmap f (SinkNoData x) = SinkNoData (f x)
     fmap f (SinkData p c) = SinkData
         { sinkPush = liftM (fmap f) . p
         , sinkClose = liftM f c
         }
-
--- | Most 'PreparedSink's require some type of state, similar to
--- 'PreparedSource's. Like a @Source@ for a @PreparedSource@, a @Sink@ is a
--- simple monadic wrapper around a @PreparedSink@ which allows initialization
--- of such state. See @Source@ for further caveats.
---
--- Note that this type provides a 'Monad' instance, allowing you to easily
--- compose @Sink@s together.
---
--- Since 0.0.0
-newtype Sink input m output = Sink { prepareSink :: ResourceT m (PreparedSink input m output) }
-
-instance Monad m => Functor (Sink input m) where
-    fmap f (Sink msink) = Sink (liftM (fmap f) msink)
+    fmap f (SinkLift msink) = SinkLift (liftM (fmap f) msink)
 
 instance Resource m => Applicative (Sink input m) where
-    pure x = Sink (return (SinkNoData x))
-    Sink mf <*> Sink ma = Sink $ do
-        f <- mf
-        a <- ma
-        case (f, a) of
-            (SinkNoData f', SinkNoData a') -> return (SinkNoData (f' a'))
-            _ -> do
-                istate <- newRef (toEither f, toEither a)
-                return $ appHelper istate
-
-toEither :: PreparedSink input m output -> SinkEither input m output
-toEither (SinkData x y) = SinkPair x y
-toEither (SinkNoData x) = SinkOutput x
-
-type SinkPush input m output = input -> ResourceT m (SinkResult input output)
-type SinkClose input m output = ResourceT m output
-data SinkEither input m output
-    = SinkPair (SinkPush input m output) (SinkClose input m output)
-    | SinkOutput output
-type SinkState input m a b = Ref (Base m) (SinkEither input m (a -> b), SinkEither input m a)
+    pure = return
+    (<*>) = ap
 
-appHelper :: Resource m => SinkState input m a b -> PreparedSink input m b
-appHelper istate = SinkData (pushHelper istate) (closeHelper istate)
+instance Resource m => Monad (Sink input m) where
+    return = SinkNoData
+    SinkNoData x >>= f = f x
+    SinkLift mx >>= f = SinkLift $ do
+        x <- mx
+        return $ x >>= f
+    SinkData push0 close0 >>= f =
+        SinkData (push push0) (close close0)
+      where
+        push push' input = do
+            res <- push' input
+            case res of
+                Done lo output -> pushHelper lo (f output)
+                Processing push'' close'' ->
+                    return $ Processing (push push'') (close close'')
 
-pushHelper :: Resource m
-           => SinkState input m a b
-           -> input
-           -> ResourceT m (SinkResult input b)
-pushHelper istate stream0 = do
-    state <- readRef istate
-    go state stream0
-  where
-    go (SinkPair f _, eb) stream = do
-        mres <- f stream
-        case mres of
-            Processing -> return Processing
-            Done leftover res -> do
-                let state' = (SinkOutput res, eb)
-                writeRef istate state'
-                maybe (return Processing) (go state') leftover
-    go (f@SinkOutput{}, SinkPair b _) stream = do
-        mres <- b stream
-        case mres of
-            Processing -> return Processing
-            Done leftover res -> do
-                let state' = (f, SinkOutput res)
-                writeRef istate state'
-                maybe (return Processing) (go state') leftover
-    go (SinkOutput f, SinkOutput b) leftover = return $ Done (Just leftover) $ f b
+        pushHelper lo (SinkNoData y) = return $ Done lo y
+        pushHelper (Just l) (SinkData pushF _) = pushF l
+        pushHelper Nothing (SinkData pushF closeF) =
+            return (Processing pushF closeF)
+        pushHelper lo (SinkLift msink) = msink >>= pushHelper lo
 
-closeHelper :: Resource m
-            => SinkState input m a b
-            -> ResourceT m b
-closeHelper istate = do
-    (sf, sa) <- readRef istate
-    case sf of
-        SinkOutput f -> go' f sa
-        SinkPair _ close -> do
-            f <- close
-            go' f sa
-  where
-    go' f (SinkPair _ close) = do
-        a <- close
-        return (f a)
-    go' f (SinkOutput a) = return (f a)
+        close close' = do
+            output <- close'
+            closeHelper (f output)
 
-instance Resource m => Monad (Sink input m) where
-    return = pure
-    mx >>= f = Sink $ do
-        x <- prepareSink mx
-        case x of
-            SinkNoData x' -> prepareSink $ f x'
-            SinkData push' close' -> do
-                istate <- newRef $ Left (push', close')
-                return $ SinkData (push istate) (close istate)
-      where
-        push istate input = do
-            state <- readRef istate
-            case state of
-                Left (push', _) -> do
-                    res <- push' input
-                    case res of
-                        Done leftover output -> do
-                            f' <- prepareSink $ f output
-                            case f' of
-                                SinkNoData y ->
-                                    return $ Done leftover y
-                                SinkData pushF closeF -> do
-                                    writeRef istate $ Right (pushF, closeF)
-                                    maybe (return Processing) (push istate) leftover
-                        Processing -> return Processing
-                Right (push', _) -> push' input
-        close istate = do
-            state <- readRef istate
-            case state of
-                Left (_, close') -> do
-                    output <- close'
-                    f' <- prepareSink $ f output
-                    case f' of
-                        SinkNoData y -> return y
-                        SinkData _ closeF -> closeF
-                Right (_, close') -> close'
+        closeHelper (SinkNoData y) = return y
+        closeHelper (SinkData _ closeF) = closeF
+        closeHelper (SinkLift msink) = msink >>= closeHelper
 
 instance (Resource m, Base m ~ base, Applicative base) => MonadBase base (Sink input m) where
     liftBase = lift . resourceLiftBase
 
 instance MonadTrans (Sink input) where
-    lift f = Sink (lift (liftM SinkNoData f))
+    lift = SinkLift . liftM SinkNoData . lift
 
 instance (Resource m, MonadIO m) => MonadIO (Sink input m) where
     liftIO = lift . liftIO
diff --git a/Data/Conduit/Types/Source.hs b/Data/Conduit/Types/Source.hs
--- a/Data/Conduit/Types/Source.hs
+++ b/Data/Conduit/Types/Source.hs
@@ -1,128 +1,71 @@
 {-# LANGUAGE FlexibleContexts #-}
-{-# LANGUAGE DeriveDataTypeable #-}
 -- | Defines the types for a source, which is a producer of data.
 module Data.Conduit.Types.Source
     ( SourceResult (..)
-    , PreparedSource (..)
     , Source (..)
-    , SourceInvariantException (..)
     ) where
 
 import Control.Monad.Trans.Resource
 import Data.Monoid (Monoid (..))
 import Control.Monad (liftM)
-import Data.Typeable (Typeable)
-import Control.Exception (Exception, throw)
 
 -- | Result of pulling from a source. Either a new piece of data (@Open@), or
 -- indicates that the source is now @Closed@.
 --
--- Since 0.0.0
-data SourceResult a = Open a | Closed
-    deriving (Show, Eq, Ord)
+-- The @Open@ constructor returns both a new value, as well as a new @Source@,
+-- which should be used in place of the previous @Source@.
+--
+-- Since 0.2.0
+data SourceResult m a = Open (Source m a) a | Closed
 
-instance Functor SourceResult where
-    fmap f (Open a) = Open (f a)
+instance Monad m => Functor (SourceResult m) where
+    fmap f (Open p a) = Open (fmap f p) (f a)
     fmap _ Closed = Closed
 
--- | A 'PreparedSource' has two operations on it: pull some data, and close the
--- 'PreparedSource'. Since 'PreparedSource' is built on top of 'ResourceT', all
--- acquired resources should be automatically released anyway. Closing a
--- 'PreparedSource' early
+-- | A @Source@ has two operations on it: pull some data, and close the
+-- @Source@. Since @Source@ is built on top of 'ResourceT', all acquired
+-- resources should be automatically released anyway. Closing a @Source@ early
 -- is merely an optimization to free scarce resources as soon as possible.
 --
--- A 'PreparedSource' has three invariants:
---
--- * It is illegal to call 'sourcePull' after a previous call returns 'Closed', or after a call to 'sourceClose'.
---
--- * It is illegal to call 'sourceClose' multiple times, or after a previous
--- 'sourcePull' returns a 'Closed'.
---
--- * A 'PreparedSource' is responsible to free any resources when either 'sourceClose'
--- is called or a 'Closed' is returned. However, based on the usage of
--- 'ResourceT', this is simply an optimization.
+-- A @Source@ is should free any resources it allocated when either
+-- @sourceClose@ is called or a @Closed@ is returned. However, based on the
+-- usage of @ResourceT@, this is simply an optimization.
 --
--- Since 0.0.0
-data PreparedSource m a = PreparedSource
-    { sourcePull :: ResourceT m (SourceResult a)
+-- Since 0.2.0
+data Source m a = Source
+    { sourcePull :: ResourceT m (SourceResult m a)
     , sourceClose :: ResourceT m ()
     }
 
-instance Monad m => Functor (PreparedSource m) where
+instance Monad m => Functor (Source m) where
     fmap f src = src
         { sourcePull = liftM (fmap f) (sourcePull src)
         }
 
--- | All but the simplest of 'PreparedSource's (e.g., @repeat@) require some
--- type of state to track their current status. This may be in the form of a
--- mutable variable (e.g., @IORef@), or via opening a resource like a @Handle@.
--- While a 'PreparedSource' is given no opportunity to acquire such resources,
--- this type is.
---
--- A 'Source' is simply a monadic action that returns a 'PreparedSource'. One
--- nice consequence of this is the possibility of creating an efficient
--- 'Monoid' instance, which will only acquire one resource at a time, instead
--- of bulk acquiring all resources at the beginning of running the 'Source'.
---
--- Note that each time you \"call\" a @Source@, it is started from scratch. If
--- you want a resumable source (e.g., one which can be passed to multiple
--- @Sink@s), you likely want to use a 'BufferedSource'.
---
--- Since 0.0.0
-newtype Source m a = Source { prepareSource :: ResourceT m (PreparedSource m a) }
-
-instance Monad m => Functor (Source m) where
-    fmap f (Source msrc) = Source (liftM (fmap f) msrc)
-
 instance Resource m => Monoid (Source m a) where
-    mempty = Source (return PreparedSource
+    mempty = Source
         { sourcePull = return Closed
         , sourceClose = return ()
-        })
+        }
     mappend a b = mconcat [a, b]
     mconcat [] = mempty
-    mconcat (Source mnext:rest0) = Source $ do
-        -- open up the first Source...
-        next0 <- mnext
-        -- and place it in a mutable reference along with all of the upcoming
-        -- Sources
-        istate <- newRef (next0, rest0)
-        return PreparedSource
-            { sourcePull = pull istate
-            , sourceClose = close istate
-            }
+    mconcat (next0:rest0) =
+        src next0 rest0
       where
-        pull istate =
-            readRef istate >>= pull'
-          where
-            pull' (current, rest) = do
-                res <- sourcePull current
-                case res of
-                    -- end of the current Source
-                    Closed -> do
-                        case rest of
-                            -- ... and open the next one
-                            Source ma:as -> do
-                                a <- ma
-                                writeRef istate (a, as)
-                                -- continue pulling base on this new state
-                                pull istate
-                            -- no more source, return an EOF
-                            [] -> do
-                                -- give an error message if the first Source
-                                -- invariant is violated (read data after EOF)
-                                writeRef istate $
-                                    throw $ PullAfterEOF "Source:mconcat"
-                                return Closed
-                    Open _ -> return res
-        close istate = do
+        src next rest = Source (pull next rest) (close next rest)
+
+        pull current rest = do
+            res <- sourcePull current
+            case res of
+                -- end of the current Source
+                Closed -> do
+                    case rest of
+                        -- ... and open the next one
+                        a:as -> pull a as
+                        -- no more source, return an EOF
+                        [] -> return Closed
+                Open current' val -> return (Open (src current' rest) val)
+        close current _rest = do
             -- we only need to close the current Source, since they are opened
             -- one at a time
-            (current, _) <- readRef istate
             sourceClose current
-
--- |
--- Since 0.0.0
-data SourceInvariantException = PullAfterEOF String
-    deriving (Show, Typeable)
-instance Exception SourceInvariantException
diff --git a/Data/Conduit/Util/Conduit.hs b/Data/Conduit/Util/Conduit.hs
--- a/Data/Conduit/Util/Conduit.hs
+++ b/Data/Conduit/Util/Conduit.hs
@@ -6,7 +6,9 @@
 -- "Data.Conduit.Types.Conduit" for more information on the base types.
 module Data.Conduit.Util.Conduit
     ( conduitState
+    , ConduitStateResult (..)
     , conduitIO
+    , ConduitIOResult (..)
     , transConduit
       -- *** Sequencing
     , SequencedSink
@@ -18,101 +20,111 @@
 import Control.Monad.Trans.Class
 import Data.Conduit.Types.Conduit
 import Data.Conduit.Types.Sink
-import Control.Monad (liftM)
 
--- | Construct a 'Conduit' with some stateful functions. This function address
--- all mutable state for you.
+-- | A helper type for @conduitState@, indicating the result of being pushed
+-- to.  It can either indicate that processing is done, or to continue with the
+-- updated state.
 --
--- Since 0.0.0
+-- Since 0.2.0
+data ConduitStateResult state input output =
+    StateFinished (Maybe input) [output]
+  | StateProducing state [output]
+
+instance Functor (ConduitStateResult state input) where
+    fmap f (StateFinished a b) = StateFinished a (map f b)
+    fmap f (StateProducing a b) = StateProducing a (map f b)
+
+-- | Construct a 'Conduit' with some stateful functions. This function addresses
+-- threading the state value for you.
+--
+-- Since 0.2.0
 conduitState
     :: Resource m
     => state -- ^ initial state
-    -> (state -> input -> ResourceT m (state, ConduitResult input output)) -- ^ Push function.
+    -> (state -> input -> ResourceT m (ConduitStateResult state input output)) -- ^ Push function.
     -> (state -> ResourceT m [output]) -- ^ Close function. The state need not be returned, since it will not be used again.
     -> Conduit input m output
-conduitState state0 push close = Conduit $ do
-#if DEBUG
-    iclosed <- newRef False
-#endif
-    istate <- newRef state0
-    return PreparedConduit
-        { conduitPush = \input -> do
-#if DEBUG
-            False <- readRef iclosed
-#endif
-            state <- readRef istate
-            (state', res) <- state `seq` push state input
-            writeRef istate state'
-#if DEBUG
-            case res of
-                Finished _ _ -> writeRef iclosed True
-                Producing _ -> return ()
-#endif
-            return res
-        , conduitClose = do
-#if DEBUG
-            False <- readRef iclosed
-            writeRef iclosed True
-#endif
-            readRef istate >>= close
-        }
+conduitState state0 push0 close0 =
+    Conduit (push state0) (close0 state0)
+  where
+    push state input = do
+        res <- state `seq` push0 state input
+        return $ case res of
+            StateFinished a b -> Finished a b
+            StateProducing state' output -> Producing
+                (Conduit (push state') (close0 state'))
+                output
 
+-- | A helper type for @conduitIO@, indicating the result of being pushed to.
+-- It can either indicate that processing is done, or to continue.
+--
+-- Since 0.2.0
+data ConduitIOResult input output =
+    IOFinished (Maybe input) [output]
+  | IOProducing [output]
+
+instance Functor (ConduitIOResult input) where
+    fmap f (IOFinished a b) = IOFinished a (map f b)
+    fmap f (IOProducing b) = IOProducing (map f b)
+
 -- | Construct a 'Conduit'.
 --
--- Since 0.0.0
+-- Since 0.2.0
 conduitIO :: ResourceIO m
            => IO state -- ^ resource and/or state allocation
            -> (state -> IO ()) -- ^ resource and/or state cleanup
-           -> (state -> input -> m (ConduitResult input output)) -- ^ Push function. Note that this need not explicitly perform any cleanup.
+           -> (state -> input -> m (ConduitIOResult input output)) -- ^ Push function. Note that this need not explicitly perform any cleanup.
            -> (state -> m [output]) -- ^ Close function. Note that this need not explicitly perform any cleanup.
            -> Conduit input m output
-conduitIO alloc cleanup push close = Conduit $ do
-#if DEBUG
-    iclosed <- newRef False
-#endif
-    (key, state) <- withIO alloc cleanup
-    return PreparedConduit
-        { conduitPush = \input -> do
-#if DEBUG
-            False <- readRef iclosed
-#endif
-            res <- lift $ push state input
-            case res of
-                Producing{} -> return ()
-                Finished{} -> do
-#if DEBUG
-                    writeRef iclosed True
-#endif
-                    release key
-            return res
-        , conduitClose = do
-#if DEBUG
-            False <- readRef iclosed
-            writeRef iclosed True
-#endif
-            output <- lift $ close state
-            release key
-            return output
-        }
+conduitIO alloc cleanup push0 close0 = Conduit
+    { conduitPush = \input -> do
+        (key, state) <- withIO alloc cleanup
+        push key state input
+    , conduitClose = do
+        (key, state) <- withIO alloc cleanup
+        close key state
+    }
+  where
+    push key state input = do
+        res <- lift $ push0 state input
+        case res of
+            IOProducing output -> return $ Producing
+                (Conduit (push key state) (close key state))
+                output
+            IOFinished a b -> do
+                release key
+                return $ Finished a b
+    close key state = do
+        output <- lift $ close0 state
+        release key
+        return output
 
 -- | Transform the monad a 'Conduit' lives in.
 --
--- Since 0.0.0
+-- See @transSource@ for more information.
+--
+-- Since 0.2.0
 transConduit :: (Monad m, Base m ~ Base n)
+             => (forall a. m a -> n a)
+             -> Conduit input m output
+             -> Conduit input n output
+transConduit f c = c
+    { conduitPush = transResourceT f . fmap (transConduitPush f) . conduitPush c
+    , conduitClose = transResourceT f (conduitClose c)
+    }
+
+transConduitPush :: (Base m ~ Base n, Monad m)
               => (forall a. m a -> n a)
-              -> Conduit input m output
-              -> Conduit input n output
-transConduit f (Conduit mc) =
-    Conduit (transResourceT f (liftM go mc))
-  where
-    go c = c
-        { conduitPush = transResourceT f . conduitPush c
-        , conduitClose = transResourceT f (conduitClose c)
-        }
+              -> ConduitResult input m output
+              -> ConduitResult input n output
+transConduitPush _ (Finished a b) = Finished a b
+transConduitPush f (Producing conduit output) = Producing
+    (transConduit f conduit)
+    output
 
 -- | Return value from a 'SequencedSink'.
 --
--- Since 0.0.0
+-- Since 0.2.0
 data SequencedSinkResponse state input m output =
     Emit state [output] -- ^ Set a new state, and emit some new output.
   | Stop -- ^ End the conduit.
@@ -122,19 +134,19 @@
 -- to write higher-level code that takes advantage of existing conduits and
 -- sinks, and leverages a sink's monadic interface.
 --
--- Since 0.0.0
+-- Since 0.2.0
 type SequencedSink state input m output =
     state -> Sink input m (SequencedSinkResponse state input m output)
 
 data SCState state input m output =
     SCNewState state
-  | SCConduit (PreparedConduit input m output)
-  | SCSink (input -> ResourceT m (SinkResult input (SequencedSinkResponse state input m output)))
+  | SCConduit (Conduit input m output)
+  | SCSink (input -> ResourceT m (SinkResult input m (SequencedSinkResponse state input m output)))
            (ResourceT m (SequencedSinkResponse state input m output))
 
 -- | Convert a 'SequencedSink' into a 'Conduit'.
 --
--- Since 0.0.0
+-- Since 0.2.0
 sequenceSink
     :: Resource m
     => state -- ^ initial state
@@ -150,43 +162,40 @@
       -> Maybe input
       -> ([output] -> [output])
       -> SequencedSink state input m output
-      -> ResourceT m (SCState state input m output, ConduitResult input output)
+      -> ResourceT m (ConduitStateResult (SCState state input m output) input output)
 goRes (Emit state output) (Just input) front fsink =
     scPush (front . (output++)) fsink (SCNewState state) input
 goRes (Emit state output) Nothing front _ =
-    return (SCNewState state, Producing $ front output)
+    return $ StateProducing (SCNewState state) $ front output
 goRes Stop minput front _ =
-    return (error "sequenceSink", Finished minput $ front [])
-goRes (StartConduit c) Nothing front _ = do
-    pc <- prepareConduit c
-    return (SCConduit pc, Producing $ front [])
-goRes (StartConduit c) (Just input) front fsink = do
-    pc <- prepareConduit c
-    scPush front fsink (SCConduit pc) input
+    return $ StateFinished minput $ front []
+goRes (StartConduit c) Nothing front _ =
+    return $ StateProducing (SCConduit c) $ front []
+goRes (StartConduit c) (Just input) front fsink =
+    scPush front fsink (SCConduit c) input
 
 scPush :: Resource m
      => ([output] -> [output])
      -> SequencedSink state input m output
      -> SCState state input m output
      -> input
-     -> ResourceT m (SCState state input m output, ConduitResult input output)
-scPush front fsink (SCNewState state) input = do
-    sink <- prepareSink $ fsink state
-    case sink of
-        SinkData push' close' -> scPush front fsink (SCSink push' close') input
-        SinkNoData res -> goRes res (Just input) front fsink
+     -> ResourceT m (ConduitStateResult (SCState state input m output) input output)
+scPush front fsink (SCNewState state) input =
+    go (fsink state)
+  where
+    go (SinkData push' close') = scPush front fsink (SCSink push' close') input
+    go (SinkNoData res) = goRes res (Just input) front fsink
+    go (SinkLift msink) = msink >>= go
 scPush front _ (SCConduit conduit) input = do
     res <- conduitPush conduit input
-    let res' =
-            case res of
-                Producing x -> Producing $ front x
-                Finished x y -> Finished x $ front y
-    return (SCConduit conduit, res')
-scPush front fsink (SCSink push close) input = do
+    return $ case res of
+        Producing c x -> StateProducing (SCConduit c) $ front x
+        Finished x y -> StateFinished x $ front y
+scPush front fsink (SCSink push _) input = do
     mres <- push input
     case mres of
         Done minput res -> goRes res minput front fsink
-        Processing -> return (SCSink push close, Producing $ front [])
+        Processing push' close' -> return (StateProducing (SCSink push' close') $ front [])
 
 scClose :: Monad m => SCState state inptu m output -> ResourceT m [output]
 scClose (SCNewState _) = return []
@@ -196,6 +205,4 @@
     case res of
         Emit _ os -> return os
         Stop -> return []
-        StartConduit c -> do
-            pc <- prepareConduit c
-            conduitClose pc
+        StartConduit c -> conduitClose c
diff --git a/Data/Conduit/Util/Sink.hs b/Data/Conduit/Util/Sink.hs
--- a/Data/Conduit/Util/Sink.hs
+++ b/Data/Conduit/Util/Sink.hs
@@ -6,7 +6,9 @@
 -- for more information on the base types.
 module Data.Conduit.Util.Sink
     ( sinkState
+    , SinkStateResult (..)
     , sinkIO
+    , SinkIOResult (..)
     , transSink
     ) where
 
@@ -15,93 +17,93 @@
 import Data.Conduit.Types.Sink
 import Control.Monad (liftM)
 
--- | Construct a 'Sink' with some stateful functions. This function address
--- all mutable state for you.
+-- | A helper type for @sinkState@, indicating the result of being pushed to.
+-- It can either indicate that processing is done, or to continue with the
+-- updated state.
 --
--- Since 0.0.0
+-- Since 0.2.0
+data SinkStateResult state input output =
+    StateDone (Maybe input) output
+  | StateProcessing state
+
+-- | Construct a 'Sink' with some stateful functions. This function addresses
+-- threading the state value for you.
+--
+-- Since 0.2.0
 sinkState
     :: Resource m
     => state -- ^ initial state
-    -> (state -> input -> ResourceT m (state, SinkResult input output)) -- ^ push
+    -> (state -> input -> ResourceT m (SinkStateResult state input output)) -- ^ push
     -> (state -> ResourceT m output) -- ^ Close. Note that the state is not returned, as it is not needed.
     -> Sink input m output
-sinkState state0 push close = Sink $ do
-    istate <- newRef state0
-#if DEBUG
-    iclosed <- newRef False
-#endif
-    return SinkData
-        { sinkPush = \input -> do
-#if DEBUG
-            False <- readRef iclosed
-#endif
-            state <- readRef istate
-            (state', res) <- state `seq` push state input
-            writeRef istate state'
-#if DEBUG
-            case res of
-                Done{} -> writeRef iclosed True
-                Processing -> return ()
-#endif
-            return res
-        , sinkClose = do
-#if DEBUG
-            False <- readRef iclosed
-            writeRef iclosed True
-#endif
-            readRef istate >>= close
-        }
+sinkState state0 push0 close0 =
+    SinkData (push state0) (close0 state0)
+  where
+    push state input = do
+        res <- state `seq` push0 state input
+        case res of
+            StateProcessing state' -> return $ Processing (push state') (close0 state')
+            StateDone mleftover output -> return $ Done mleftover output
 
+-- | A helper type for @sinkIO@, indicating the result of being pushed to. It
+-- can either indicate that processing is done, or to continue.
+--
+-- Since 0.2.0
+data SinkIOResult input output = IODone (Maybe input) output | IOProcessing
+
 -- | Construct a 'Sink'. Note that your push and close functions need not
 -- explicitly perform any cleanup.
 --
--- Since 0.0.0
+-- Since 0.2.0
 sinkIO :: ResourceIO m
         => IO state -- ^ resource and/or state allocation
         -> (state -> IO ()) -- ^ resource and/or state cleanup
-        -> (state -> input -> m (SinkResult input output)) -- ^ push
+        -> (state -> input -> m (SinkIOResult input output)) -- ^ push
         -> (state -> m output) -- ^ close
         -> Sink input m output
-sinkIO alloc cleanup push close = Sink $ do
-    (key, state) <- withIO alloc cleanup
-#if DEBUG
-    iclosed <- newRef False
-#endif
-    return SinkData
-        { sinkPush = \input -> do
-#if DEBUG
-            False <- readRef iclosed
-#endif
-            res <- lift $ push state input
-            case res of
-                Done{} -> do
-                    release key
-#if DEBUG
-                    writeRef iclosed True
-#endif
-                Processing -> return ()
-            return res
-        , sinkClose = do
-#if DEBUG
-            False <- readRef iclosed
-            writeRef iclosed True
-#endif
-            res <- lift $ close state
-            release key
-            return res
-        }
+sinkIO alloc cleanup push0 close0 = SinkData
+    { sinkPush = \input -> do
+        (key, state) <- withIO alloc cleanup
+        push key state input
+    , sinkClose = do
+        (key, state) <- withIO alloc cleanup
+        close key state
+    }
+  where
+    push key state input = do
+        res <- lift $ push0 state input
+        case res of
+            IODone a b -> do
+                release key
+                return $ Done a b
+            IOProcessing -> return $ Processing
+                (push key state)
+                (close key state)
+    close key state = do
+        res <- lift $ close0 state
+        release key
+        return res
 
 -- | Transform the monad a 'Sink' lives in.
 --
--- Since 0.0.0
+-- See @transSource@ for more information.
+--
+-- Since 0.2.0
 transSink :: (Base m ~ Base n, Monad m)
-           => (forall a. m a -> n a)
-           -> Sink input m output
-           -> Sink input n output
-transSink f (Sink mc) =
-    Sink (transResourceT f (liftM go mc))
-  where
-    go c = c
-        { sinkPush = transResourceT f . sinkPush c
-        , sinkClose = transResourceT f (sinkClose c)
-        }
+          => (forall a. m a -> n a)
+          -> Sink input m output
+          -> Sink input n output
+transSink _ (SinkNoData x) = SinkNoData x
+transSink f (SinkLift msink) = SinkLift (transResourceT f (liftM (transSink f) msink))
+transSink f (SinkData push close) = SinkData
+    (transResourceT f . fmap (transSinkPush f) . push)
+    (transResourceT f close)
+
+transSinkPush :: (Base m ~ Base n, Monad m)
+              => (forall a. m a -> n a)
+              -> SinkResult input m output
+              -> SinkResult input n output
+transSinkPush _ (Done a b) = Done a b
+transSinkPush f (Processing push close) = Processing
+    (transResourceT f . fmap (transSinkPush f) . push)
+    (transResourceT f close)
diff --git a/Data/Conduit/Util/Source.hs b/Data/Conduit/Util/Source.hs
--- a/Data/Conduit/Util/Source.hs
+++ b/Data/Conduit/Util/Source.hs
@@ -7,100 +7,89 @@
 -- on the base types.
 module Data.Conduit.Util.Source
     ( sourceState
+    , SourceStateResult (..)
     , sourceIO
+    , SourceIOResult (..)
     , transSource
     ) where
 
 import Control.Monad.Trans.Resource
 import Control.Monad.Trans.Class (lift)
 import Data.Conduit.Types.Source
-import Control.Monad (liftM)
 
--- | Construct a 'Source' with some stateful functions. This function address
--- all mutable state for you.
+-- | The return value when pulling in the @sourceState@ function. Either
+-- indicates no more data, or the next value and an updated state.
 --
--- Since 0.0.0
+-- Since 0.2.0
+data SourceStateResult state output = StateOpen state output | StateClosed
+
+-- | Construct a 'Source' with some stateful functions. This function addresses
+-- threading the state value for you.
+--
+-- Since 0.2.0
 sourceState
     :: Resource m
     => state -- ^ Initial state
-    -> (state -> ResourceT m (state, SourceResult output)) -- ^ Pull function
+    -> (state -> ResourceT m (SourceStateResult state output)) -- ^ Pull function
     -> Source m output
-sourceState state0 pull = Source $ do
-    istate <- newRef state0
-#if DEBUG
-    iclosed <- newRef False
-#endif
-    return PreparedSource
-        { sourcePull = do
-#if DEBUG
-            False <- readRef iclosed
-#endif
-            state <- readRef istate
-            (state', res) <- pull state
-#if DEBUG
-            let isClosed =
-                    case res of
-                        Closed -> True
-                        Open _ -> False
-            writeRef iclosed isClosed
-#endif
-            writeRef istate state'
-            return res
-        , sourceClose = do
-#if DEBUG
-            False <- readRef iclosed
-            writeRef iclosed True
-#else
-            return ()
-#endif
-        }
+sourceState state0 pull0 =
+    src state0
+  where
+    src state = Source (pull state) close
 
+    pull state = do
+        res <- pull0 state
+        return $ case res of
+            StateOpen state' val -> Open (src state') val
+            StateClosed -> Closed
+
+    close = return ()
+
+-- | The return value when pulling in the @sourceIO@ function. Either indicates
+-- no more data, or the next value.
+data SourceIOResult output = IOOpen output | IOClosed
+
 -- | Construct a 'Source' based on some IO actions for alloc/release.
 --
--- Since 0.0.0
+-- Since 0.2.0
 sourceIO :: ResourceIO m
           => IO state -- ^ resource and/or state allocation
           -> (state -> IO ()) -- ^ resource and/or state cleanup
-          -> (state -> m (SourceResult output)) -- ^ Pull function. Note that this need not explicitly perform any cleanup.
+          -> (state -> m (SourceIOResult output)) -- ^ Pull function. Note that this need not explicitly perform any cleanup.
           -> Source m output
-sourceIO alloc cleanup pull = Source $ do
-    (key, state) <- withIO alloc cleanup
-#if DEBUG
-    iclosed <- newRef False
-#endif
-    return PreparedSource
+sourceIO alloc cleanup pull0 =
+    Source
         { sourcePull = do
-#if DEBUG
-            False <- readRef iclosed
-#endif
-            res <- lift $ pull state
-            case res of
-                Closed -> do
-#if DEBUG
-                    writeRef iclosed True
-#endif
-                    release key
-                _ -> return ()
-            return res
-        , sourceClose = do
-#if DEBUG
-            False <- readRef iclosed
-            writeRef iclosed True
-#endif
-            release key
+            (key, state) <- withIO alloc cleanup
+            pull key state
+        , sourceClose = return ()
         }
+  where
+    src key state = Source (pull key state) (release key)
 
+    pull key state = do
+        res <- lift $ pull0 state
+        case res of
+            IOClosed -> do
+                release key
+                return Closed
+            IOOpen val -> return $ Open (src key state) val
+
 -- | Transform the monad a 'Source' lives in.
 --
--- Since 0.0.0
+-- Note that this will /not/ thread the individual monads together, meaning
+-- side effects will be lost. This function is most useful for transformers
+-- only providing context and not producing side-effects, such as @ReaderT@.
+--
+-- Since 0.2.0
 transSource :: (Base m ~ Base n, Monad m)
-             => (forall a. m a -> n a)
-             -> Source m output
-             -> Source n output
-transSource f (Source mc) =
-    Source (transResourceT f (liftM go mc))
+            => (forall a. m a -> n a)
+            -> Source m output
+            -> Source n output
+transSource f c = c
+    { sourcePull = transResourceT f (fmap go2 $ sourcePull c)
+    , sourceClose = transResourceT f (sourceClose c)
+    }
   where
-    go c = c
-        { sourcePull = transResourceT f (sourcePull c)
-        , sourceClose = transResourceT f (sourceClose c)
-        }
+    go2 (Open p a) = Open (transSource f p) a
+    go2 Closed = Closed
diff --git a/System/PosixFile.hsc b/System/PosixFile.hsc
--- a/System/PosixFile.hsc
+++ b/System/PosixFile.hsc
@@ -16,7 +16,7 @@
 import qualified Data.ByteString as S
 import qualified Data.ByteString.Unsafe as BU
 import Prelude hiding (read)
-import Data.Conduit.Types.Source (SourceResult (..))
+import Data.Conduit.Util.Source (SourceIOResult (..))
 
 #include <fcntl.h>
 
@@ -45,13 +45,13 @@
         then throwErrno $ "Could not open file: " ++ fp
         else return $ FD h
 
-read :: FD -> IO (SourceResult S.ByteString)
+read :: FD -> IO (SourceIOResult S.ByteString)
 read fd = do
     cstr <- mallocBytes 4096
     len <- c_read fd cstr 4096
     if len == 0
-        then free cstr >> return Closed
-        else fmap Open $ BU.unsafePackCStringFinalizer
+        then free cstr >> return IOClosed
+        else fmap IOOpen $ BU.unsafePackCStringFinalizer
                 cstr
                 (fromIntegral len)
                 (free cstr)
diff --git a/System/Win32File.hsc b/System/Win32File.hsc
--- a/System/Win32File.hsc
+++ b/System/Win32File.hsc
@@ -18,7 +18,7 @@
 import Data.Text.Encoding (encodeUtf16LE)
 import Data.Word (Word8)
 import Prelude hiding (read)
-import Data.Conduit (SourceResult (..))
+import Data.Conduit (SourceIOResult (..))
 
 #include <fcntl.h>
 #include <Share.h>
@@ -74,16 +74,16 @@
         then throwErrno $ "Could not open file: " ++ fp
         else return $ FD h
 
-read :: FD -> IO (SourceResult S.ByteString)
+read :: FD -> IO (SourceIOResult S.ByteString)
 read fd = do
     cstr <- mallocBytes 4096
     len <- c_read fd cstr 4096
     if len == 0
         then do
             free cstr
-            return Closed
+            return IOClosed
         else do
-            fmap Open $ BU.unsafePackCStringFinalizer
+            fmap IOOpen $ BU.unsafePackCStringFinalizer
                 cstr
                 (fromIntegral len)
                 (free cstr)
diff --git a/conduit.cabal b/conduit.cabal
--- a/conduit.cabal
+++ b/conduit.cabal
@@ -1,14 +1,16 @@
 Name:                conduit
-Version:             0.1.1.1
+Version:             0.2.0
 Synopsis:            Streaming data processing library.
 Description:
 	Conduits are an approach to the streaming data problem. It is meant as an alternative to enumerators\/iterators, hoping to address the same issues with different trade-offs based on real-world experience with enumerators. For more information, see <http://www.yesodweb.com/book/conduit>.
 	.
 	Release history:
 	.
-	* 0.1: @BufferedSource@ is now an abstract type, and has a much more efficient internal representation. The result was a 41% speedup on microbenchmarks (note: do not expect speedups anywhere near that in real usage). In general, we are moving towards @BufferedSource@ being a specific tool used internally as needed, but using @Source@ for all external APIs.
+    [0.2] Instead of storing state in mutable variables, we now use CPS. A @Source@ returns the next @Source@, and likewise for @Sink@s and @Conduit@s. Not only does this take better advantage of GHC\'s optimizations (about a 20% speedup), but it allows some operations to have a reduction in algorithmic complexity from exponential to linear. This also allowed us to remove the @Prepared@ set of types. Also, the @State@ functions (e.g., @sinkState@) use better constructors for return types, avoiding the need for a dummy state on completion.
+    .
+	[0.1] @BufferedSource@ is now an abstract type, and has a much more efficient internal representation. The result was a 41% speedup on microbenchmarks (note: do not expect speedups anywhere near that in real usage). In general, we are moving towards @BufferedSource@ being a specific tool used internally as needed, but using @Source@ for all external APIs.
 	.
-	* 0.0: Initial release.
+	[0.0] Initial release.
 License:             BSD3
 License-file:        LICENSE
 Author:              Michael Snoyman
@@ -64,7 +66,7 @@
     cpp-options:   -DTEST
     build-depends:   conduit
                    , base
-                   , hspec
+                   , hspec >= 0.9.1
                    , HUnit
                    , QuickCheck
                    , bytestring
diff --git a/test/main.hs b/test/main.hs
--- a/test/main.hs
+++ b/test/main.hs
@@ -15,10 +15,10 @@
 import Control.Monad.ST (runST)
 import Data.Monoid
 import qualified Data.ByteString as S
+import qualified Data.ByteString.Char8 as S8
 import qualified Data.IORef as I
 import qualified Data.ByteString.Lazy as L
 import Data.ByteString.Lazy.Char8 ()
-import Control.Monad.Trans.Writer (Writer)
 import qualified Data.Text as T
 import qualified Data.Text.Lazy as TL
 import qualified Data.Text.Lazy.Encoding as TLE
@@ -226,13 +226,13 @@
         it' "works inside a ResourceT" $ runResourceT $ do
             counter <- liftIO $ I.newIORef 0
             let incr i = C.sourceIO
-                    (liftIO $ I.newIORef $ C.Open (i :: Int))
+                    (liftIO $ I.newIORef $ C.IOOpen (i :: Int))
                     (const $ return ())
                     (\istate -> do
                         res <- liftIO $ I.atomicModifyIORef istate
-                            (\state -> (C.Closed, state))
+                            (\state -> (C.IOClosed, state))
                         case res of
-                            C.Closed -> return ()
+                            C.IOClosed -> return ()
                             _ -> do
                                 count <- liftIO $ I.atomicModifyIORef counter
                                     (\j -> (j + 1, j + 1))
@@ -400,5 +400,13 @@
                 bsrc C.$= CB.isolate 10 C.$$ CL.head
             x @?= Just "foobarbazb"
 
-it' :: String -> IO () -> Writer [Spec] ()
+    describe "binary" $ do
+        prop "lines" $ \bss' -> runST $ runResourceT $ do
+            let bss = map S.pack bss'
+                bs = S.concat bss
+                src = CL.sourceList bss
+            res <- src C.$$ CB.lines C.=$ CL.consume
+            return $ S8.lines bs == res
+
+it' :: String -> IO () -> Specs
 it' = it