diff --git a/Data/Conduit.hs b/Data/Conduit.hs
--- a/Data/Conduit.hs
+++ b/Data/Conduit.hs
@@ -3,43 +3,43 @@
 {-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE UndecidableInstances #-}
 {-# LANGUAGE DeriveDataTypeable #-}
+{-# LANGUAGE FunctionalDependencies #-}
 -- | The main module, exporting types, utility functions, and fuse and connect
 -- operators.
+--
+-- There are three main types in this package: @Source@ (data producer), @Sink@
+-- (data consumer), and @Conduit@ (data transformer). All three are in fact
+-- type synonyms for the underlying @Pipe@ data type.
+--
+-- The typical approach to use of this package is:
+--
+-- * Compose multiple @Sink@s together using its @Monad@ instance.
+--
+-- * Left-fuse @Source@s and @Conduit@s into new @Conduit@s.
+--
+-- * Right-fuse @Conduit@s and @Sink@s into new @Sink@s.
+--
+-- * Middle-fuse two @Conduit@s into a new @Conduit@.
+--
+-- * Connect a @Source@ to a @Sink@ to obtain a result.
 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
-    , BufferedSource
-    , bufferSource
-    , unbufferSource
-    , bsourceClose
-      -- *** Unifying
-    , IsSource
-      -- ** Sink
-    , module Data.Conduit.Types.Sink
-      -- ** Conduit
-    , module Data.Conduit.Types.Conduit
-    , -- * Connect/fuse operators
-      ($$)
+      Pipe (..)
+    , Source
+    , Conduit
+    , Sink
+      -- * Connect/fuse operators
+    , ($$)
+    , ($$+)
     , ($=)
     , (=$)
     , (=$=)
       -- * Utility functions
+      -- ** General
+    , await
+    , yield
+    , hasInput
+    , transPipe
       -- ** Source
     , module Data.Conduit.Util.Source
       -- ** Sink
@@ -56,387 +56,87 @@
     , runResourceT
     ) where
 
-import Control.Monad (liftM)
 import Control.Monad.Trans.Resource
-import Control.Monad.IO.Class (MonadIO (liftIO))
-import qualified Data.IORef as I
-import Data.Conduit.Types.Source
+import Data.Conduit.Internal
 import Data.Conduit.Util.Source
-import Data.Conduit.Types.Sink
 import Data.Conduit.Util.Sink
-import Data.Conduit.Types.Conduit
 import Data.Conduit.Util.Conduit
 
 -- $typeOverview
 
 infixr 0 $$
+infixr 0 $$+
+infixl 1 $=
+infixr 2 =$
+infixr 2 =$=
 
+
 -- | The connect operator, which pulls data from a source and pushes to a sink.
 -- There are two ways this process can terminate:
 --
 -- 1. If the @Sink@ is a @Done@ constructor, the @Source@ is closed.
 --
--- 2. If the @Source@ is a @Closed@ constructor, the @Sink@ is closed.
+-- 2. If the @Source@ is a @Done@ constructor, the @Sink@ is closed.
 --
--- This function will automatically close any @Source@s, but will not close any
--- @BufferedSource@s, allowing them to be reused. Also, leftover data will be
--- discarded when connecting a @Source@, but will be buffered when using a
--- @BufferedSource@.
+-- In other words, both the @Source@ and @Sink@ will always be closed. If you
+-- would like to keep the @Source@ open to be used for another operations, use
+-- the connect-and-resume operators '$$+'.
 --
--- Since 0.3.0
-($$) :: IsSource src m => src m a -> Sink a m b -> m b
-($$) = connect
+-- Since 0.4.0
+($$) :: Monad m => Source m a -> Sink a m b -> m b
+src $$ sink = runPipe $ pipe src sink
 {-# INLINE ($$) #-}
 
--- | A typeclass allowing us to unify operators for 'Source' and
--- 'BufferedSource'.
+-- | The connect-and-resume operator. Does not close the @Source@, but instead
+-- returns it to be used again. This allows a @Source@ to be used incrementally
+-- in a large program, without forcing the entire program to live in the @Sink@
+-- monad.
 --
--- Since 0.3.0
-class IsSource src m where
-    connect :: src m a -> Sink a m b -> m b
-    fuseLeft :: src m a -> Conduit a m b -> Source m b
-
-instance Monad m => IsSource Source m where
-    connect = normalConnect
-    {-# INLINE connect #-}
-    fuseLeft = normalFuseLeft
-    {-# INLINE fuseLeft #-}
-
-instance MonadIO m => IsSource BufferedSource m where
-    connect = bufferedConnect
-    {-# INLINE connect #-}
-    fuseLeft = bufferedFuseLeft
-    {-# INLINE fuseLeft #-}
-
-normalConnect :: Monad m => Source m a -> Sink a m b -> m b
-
--- @Sink@ cannot handle any more input, close the @Source@ (regardless of its
--- state) and discard leftovers.
-normalConnect src (Done _leftover output) = sourceClose src >> return output
-
--- Run the @Sink@'s monadic action and try again.
-normalConnect src (SinkM msink) = msink >>= normalConnect src
-
--- Run the @Source@'s monadic action and try again.
-normalConnect (SourceM msrc _) sink@Processing{} = msrc >>= flip normalConnect sink
-
--- No more input available from @Source@, close the @Sink@.
-normalConnect Closed (Processing _ close) = close
-
--- More input available, and the @Sink@ wants it: plug it in and keep going.
-normalConnect (Open src _ a) (Processing push _) = normalConnect src $ push a
-
-data FuseLeftState srcState input m output =
-    FLClosed
-  | FLOpen srcState (ConduitPush input m output) (ConduitClose m output)
-  | FLHaveOutput srcState (Conduit input m output) (m ())
-
-infixl 1 $=
+-- Mnemonic: connect + do more.
+--
+-- Since 0.4.0
+($$+) :: Monad m => Source m a -> Sink a m b -> m (Source m a, b)
+src $$+ sink = runPipe $ pipeResume src sink
+{-# INLINE ($$+) #-}
 
 -- | Left fuse, combining a source and a conduit together into a new source.
 --
--- Any @Source@ passed in will be automatically closed, while a
--- @BufferedSource@ will be left open. Leftover input will be discarded for a
--- @Source@, and buffered for a @BufferedSource@.
+-- Both the @Source@ and @Conduit@ will be closed when the newly-created
+-- @Source@ is closed.
 --
--- Since 0.3.0
-($=) :: IsSource src m
-     => src m a
-     -> Conduit a m b
-     -> Source m b
-($=) = fuseLeft
+-- Leftover data from the @Conduit@ will be discarded.
+--
+-- Since 0.4.0
+($=) :: Monad m => Source m a -> Conduit a m b -> Source m b
+($=) = pipe
 {-# INLINE ($=) #-}
 
-normalFuseLeft :: Monad m => Source m a -> Conduit a m b -> Source m b
-
--- No more input, close the @Conduit@.
-normalFuseLeft Closed (NeedInput _ close) = close
-normalFuseLeft Closed (Finished _) = Closed
-
--- @Conduit@ is done, discard leftovers and close the @Source@.
-normalFuseLeft src (Finished _) = SourceM
-    (sourceClose src >> return Closed)
-    (sourceClose src)
-
--- @Conduit@ has some output, return it and keep going.
-normalFuseLeft src (HaveOutput p c x) = Open
-    (normalFuseLeft src p)
-    (sourceClose src >> c)
-    x
-
--- @Source@ provided input, and @Conduit@ wants it. Pipe it through and keep going.
-normalFuseLeft (Open src _ a) (NeedInput push _) = normalFuseLeft src $ push a
-
--- Need to perform a monadic action to get the next @Conduit@.
-normalFuseLeft src (ConduitM mcon conclose) = SourceM
-    (liftM (normalFuseLeft src) mcon)
-    (conclose >> sourceClose src)
-
--- Need to perform a monadic action to get the next @Source@.
-normalFuseLeft (SourceM msrc closeS) conduit@(NeedInput _ closeC) =
-    SourceM (liftM (flip normalFuseLeft conduit) msrc) $ do
-        closeS
-        sourceClose closeC
-
-infixr 0 =$
-
 -- | Right fuse, combining a conduit and a sink together into a new sink.
 --
--- Any leftover data returns from the @Sink@ will be discarded.
---
--- Since 0.3.0
-(=$) :: Monad m => Conduit a m b -> Sink b m c -> Sink a m c
-
--- @Sink@ is complete, discard leftovers, close the @Conduit@ and return the
--- output.
-conduit =$ Done _leftover output = SinkM $ conduitClose conduit >> return (Done Nothing output)
-
--- Need to perform a monadic action to get the next @Sink@.
-conduit =$ SinkM msink = SinkM (liftM (conduit =$) msink)
-
--- Need more input for the @Conduit@, so ask for it.
-NeedInput pushO closeO =$ sink = Processing
-    (\input -> pushO input =$ sink)
-    (closeO $$ sink)
-
--- @Conduit@ can provide no more input to @Sink@. Close the @Sink@ and return
--- the leftovers from the @Conduit@.
-Finished mleftover =$ Processing _ close = SinkM $ liftM (Done mleftover) close
-
--- Perform a monadic action to get the next @Conduit@.
-ConduitM mcon _ =$ sink@Processing{} = SinkM $ liftM (=$ sink) mcon
-
--- @Conduit@ is providing input for the @Sink@, so use it and keep going.
-HaveOutput con _ input =$ Processing pushI _ = con =$ pushI input
-
-infixr 0 =$=
-
--- | Middle fuse, combining two conduits together into a new conduit.
---
--- Any leftovers provided by the inner @Conduit@ will be discarded.
---
--- Since 0.3.0
-(=$=) :: Monad m => Conduit a m b -> Conduit b m c -> Conduit a m c
-
--- No more input from outer conduit, and inner wants more input. Close the
--- inner conduit and convert its source into a conduit.
-Finished mleftover =$= NeedInput _ closeI =
-    go closeI
-  where
-    go Closed = Finished mleftover
-    go (Open src close x) = HaveOutput (go src) close x
-    go (SourceM msrc close) = ConduitM (liftM go msrc) close
-
--- No more input from outer conduit, and inner wants no more input anyway.
--- Discard the leftovers from the inner.
-Finished mleftover =$= Finished _ = Finished mleftover
-
--- Provide more output from the inner conduit.
-conO =$= HaveOutput con close x = HaveOutput (conO =$= con) close x
-
--- Perform a monadic action to get the next outer conduit.
-ConduitM mcon close =$= conI = ConduitM (liftM (=$= conI) mcon) (close >> conduitClose conI)
-
--- Ask for more data for the outer conduit. Note that this clause comes before
--- the inner conduit ConduitM clause, so that we only run that action as
--- necessary.
-NeedInput pushO closeO =$= conI = NeedInput
-    (\input -> pushO input =$= conI)
-    (closeO $= conI)
-
--- Perform a monadic action to get the next inner conduit.
-conO =$= ConduitM mconI close = ConduitM (liftM (conO =$=) mconI) (close >> conduitClose conO)
-
--- Pipe output from outer conduit to inner conduit and keep going.
-HaveOutput conO _ inputI =$= NeedInput pushI _ = conO =$= pushI inputI
-
--- Discard output from outer conduit, and discard leftovers from inner conduit.
--- Close the outer conduit.
-HaveOutput _ close _ =$= Finished _ = ConduitM (close >> return (Finished Nothing)) close
-
--- | 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.
---
--- 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.
---
--- 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.
+-- Both the @Conduit@ and @Sink@ will be closed when the newly-created @Sink@
+-- is closed.
 --
--- 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.
+-- Leftover data returned from the @Sink@ will be discarded.
 --
--- Since 0.3.0
-data BufferedSource m a = BufferedSource (I.IORef (BSState m a))
-
-data BSState m a =
-    ClosedEmpty
-  | OpenEmpty (Source m a)
-  | ClosedFull a
-  | OpenFull (Source m a) a
+-- Since 0.4.0
+(=$) :: Monad m => Conduit a m b -> Sink b m c -> Sink a m c
+(=$) = pipe
+{-# INLINE (=$) #-}
 
--- | 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.
+-- | Fusion operator, combining two @Pipe@s together into a new @Pipe@.
 --
--- Since 0.3.0
-bufferSource :: MonadIO m => Source m a -> m (BufferedSource m a)
-bufferSource src = liftM BufferedSource $ liftIO $ I.newIORef $ OpenEmpty src
-
--- | Turn a 'BufferedSource' into a 'Source'. Note that in general this will
--- mean your original 'BufferedSource' will be closed. Additionally, all
--- leftover data from usage of the returned @Source@ will be discarded. In
--- other words: this is a no-going-back move.
+-- Both @Pipe@s will be closed when the newly-created @Pipe@ is closed.
 --
--- Note: @bufferSource@ . @unbufferSource@ is /not/ the identity function.
+-- Leftover data returned from the right @Pipe@ will be discarded.
 --
--- Since 0.3.0
-unbufferSource :: MonadIO m
-               => BufferedSource m a
-               -> Source m a
-unbufferSource (BufferedSource bs) =
-    SourceM msrc (msrc >>= sourceClose)
-  where
-    msrc = do
-        buf <- liftIO $ I.readIORef bs
-        case buf of
-            OpenEmpty src -> return src
-            OpenFull src a -> return $ Open src (sourceClose src) a
-            ClosedEmpty -> return Closed
-            ClosedFull a -> return $ Open Closed (return ()) a
-
-bufferedConnect :: MonadIO m => BufferedSource m a -> Sink a m b -> m b
-bufferedConnect _ (Done Nothing output) = return output
-bufferedConnect _ (Done Just{} _) = error "Invariant violated: sink returned leftover without input"
-bufferedConnect bsrc (SinkM msink) = msink >>= bufferedConnect bsrc
-bufferedConnect (BufferedSource bs) (Processing push0 close0) = do
-    bsState <- liftIO $ I.readIORef bs
-    case bsState of
-        ClosedEmpty -> close0
-        OpenEmpty src -> connect' src push0 close0
-        ClosedFull a -> onRes Nothing $ push0 a
-        OpenFull src a -> onRes (Just src) $ push0 a
-  where
-    connect' Closed _ close = do
-        liftIO $ I.writeIORef bs ClosedEmpty
-        close
-    connect' (Open src _ x) push _ = onRes (Just src) $ push x
-    connect' (SourceM msrc _) push close = msrc >>= \src -> connect' src push close
-
-    onRes msrc (Done mleftover res) = do
-        let state =
-                case (msrc, mleftover) of
-                    (Nothing, Nothing) -> ClosedEmpty
-                    (Just src, Nothing) -> OpenEmpty src
-                    (Nothing, Just leftover) -> ClosedFull leftover
-                    (Just src, Just leftover) -> OpenFull src leftover
-        liftIO $ I.writeIORef bs state
-        return res
-    onRes Nothing (Processing _ close) = do
-        liftIO $ I.writeIORef bs ClosedEmpty
-        close
-    onRes (Just src) (Processing push close) = connect' src push close
-    onRes msrc (SinkM msink) = msink >>= onRes msrc
-
-bufferedFuseLeft
-    :: MonadIO m
-    => BufferedSource m a
-    -> Conduit a m b
-    -> Source m b
-bufferedFuseLeft bsrc (ConduitM mcon close) = SourceM
-    (liftM (bufferedFuseLeft bsrc) mcon)
-    close
-bufferedFuseLeft _ (Finished _) = Closed
-bufferedFuseLeft bsrc (HaveOutput next close x) = Open
-    (bufferedFuseLeft bsrc next)
-    close
-    x
-bufferedFuseLeft bsrc (NeedInput push0 close0) = SourceM
-    (pullF $ FLOpen () push0 close0)
-    (sourceClose close0)
-  where
-    mkSrc state = SourceM
-        (pullF state)
-        (closeF state)
-
-    pullF state' =
-        case state' of
-            FLClosed -> return Closed
-            FLHaveOutput () pull _ -> goRes pull
-            FLOpen () push close -> do
-                mres <- bsourcePull bsrc
-                case mres of
-                    Nothing -> return close
-                    Just input -> goRes $ push input
-
-    goRes (Finished leftover) = do
-        bsourceUnpull bsrc leftover
-        return Closed
-    goRes (HaveOutput pull close' x) =
-        let state = FLHaveOutput () pull close'
-         in return $ Open (mkSrc state) (closeF state) x
-    goRes (NeedInput pushI closeI) = pullF (FLOpen () pushI closeI)
-    goRes (ConduitM mcon _) = mcon >>= goRes
-
-    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.
-        case state of
-            FLClosed -> return ()
-            FLOpen () _ close -> do
-                () <- sourceClose close
-                return ()
-            FLHaveOutput () _ close -> close
-
-bsourcePull :: MonadIO m => BufferedSource m a -> m (Maybe a)
-bsourcePull (BufferedSource bs) =
-    liftIO (I.readIORef bs) >>= goBuf
-  where
-    goBuf (OpenEmpty Closed) = liftIO $ I.writeIORef bs ClosedEmpty >> return Nothing
-    goBuf (OpenEmpty (Open src _ a)) = do
-        liftIO $ I.writeIORef bs $ OpenEmpty src
-        return $ Just a
-    goBuf (OpenEmpty (SourceM msrc _)) = msrc >>= goBuf . OpenEmpty
-    goBuf ClosedEmpty = return Nothing
-    goBuf (OpenFull src a) = do
-        liftIO $ I.writeIORef bs (OpenEmpty src)
-        return $ Just a
-    goBuf (ClosedFull a) = do
-        liftIO $ I.writeIORef bs ClosedEmpty
-        return $ Just a
-
-bsourceUnpull :: MonadIO m => BufferedSource m a -> Maybe a -> m ()
-bsourceUnpull _ Nothing = return ()
-bsourceUnpull (BufferedSource ref) (Just a) = do
-    buf <- liftIO $ I.readIORef ref
-    case buf of
-        OpenEmpty src -> liftIO $ I.writeIORef ref (OpenFull src a)
-        ClosedEmpty -> liftIO $ I.writeIORef ref (ClosedFull a)
-        _ -> error $ "Invariant violated: bsourceUnpull called on full data"
-
--- | 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 'Source' was previously closed.
+-- Note: in previous versions, this operator would only fuse together two
+-- @Conduit@s (known as middle fusion). This operator is generalized to work on
+-- all @Pipe@s, including @Source@s and @Sink@s.
 --
--- Since 0.3.0
-bsourceClose :: MonadIO m => BufferedSource m a -> m ()
-bsourceClose (BufferedSource ref) = do
-    buf <- liftIO $ I.readIORef ref
-    case buf of
-        OpenEmpty src -> sourceClose src
-        OpenFull src _ -> sourceClose src
-        ClosedEmpty -> return ()
-        ClosedFull _ -> return ()
+-- Since 0.4.0
+(=$=) :: Monad m => Pipe a b m () -> Pipe b c m r -> Pipe a c m r
+(=$=) = pipe
+{-# INLINE (=$=) #-}
 
 -- | Provide for a stream of data that can be flushed.
 --
diff --git a/Data/Conduit/Binary.hs b/Data/Conduit/Binary.hs
--- a/Data/Conduit/Binary.hs
+++ b/Data/Conduit/Binary.hs
@@ -76,13 +76,13 @@
 sourceHandle h =
     src
   where
-    src = SourceM pull close
+    src = PipeM pull close
 
     pull = do
         bs <- liftIO (S.hGetSome h 4096)
         if S.null bs
-            then return Closed
-            else return $ Open src close bs
+            then return $ Done Nothing ()
+            else return $ HaveOutput src close bs
 
     close = return ()
 
@@ -110,9 +110,11 @@
            => IO.Handle
            -> Sink S.ByteString m ()
 sinkHandle h =
-    Processing push close
+    NeedInput push close
   where
-    push input = SinkM $ liftIO (S.hPut h input) >> return (Processing push close)
+    push input = PipeM
+        (liftIO (S.hPut h input) >> return (NeedInput push close))
+        (return ())
     close = return ()
 
 -- | An alternative to 'sinkHandle'.
@@ -137,7 +139,7 @@
                 -> Maybe Integer -- ^ Offset
                 -> Maybe Integer -- ^ Maximum count
                 -> Source m S.ByteString
-sourceFileRange fp offset count = SourceM
+sourceFileRange fp offset count = PipeM
     (do
         (key, handle) <- allocate (IO.openBinaryFile fp IO.ReadMode) IO.hClose
         case offset of
@@ -153,12 +155,12 @@
         if S.null bs
             then do
                 release key
-                return Closed
+                return $ Done Nothing ()
             else do
-                let src = SourceM
+                let src = PipeM
                         (pullUnlimited handle key)
                         (release key)
-                return $ Open src (release key) bs
+                return $ HaveOutput src (release key) bs
 
     pullLimited c0 handle key = do
         let c = fromInteger c0
@@ -168,12 +170,12 @@
             if S.null bs
                 then do
                     release key
-                    return Closed
+                    return $ Done Nothing ()
                 else do
-                    let src = SourceM
+                    let src = PipeM
                             (pullLimited (toInteger c') handle key)
                             (release key)
-                    return $ Open src (release key) bs
+                    return $ HaveOutput src (release key) bs
 
 -- | Stream all incoming data to the given file.
 --
@@ -226,11 +228,11 @@
 -- Since 0.3.0
 head :: Monad m => Sink S.ByteString m (Maybe Word8)
 head =
-    Processing push close
+    NeedInput push close
   where
     push bs =
         case S.uncons bs of
-            Nothing -> Processing push close
+            Nothing -> NeedInput push close
             Just (w, bs') ->
                 let lo = if S.null bs' then Nothing else Just bs'
                  in Done lo (Just w)
@@ -250,7 +252,7 @@
                     then r
                     else HaveOutput r (return ()) x
         | otherwise =
-            let f = Finished $ Just y
+            let f = Done (Just y) ()
              in if S.null x
                     then f
                     else HaveOutput f (return ()) x
@@ -263,10 +265,10 @@
 -- Since 0.3.0
 dropWhile :: Monad m => (Word8 -> Bool) -> Sink S.ByteString m ()
 dropWhile p =
-    Processing push close
+    NeedInput push close
   where
     push bs
-        | S.null bs' = Processing push close
+        | S.null bs' = NeedInput push close
         | otherwise  = Done (Just bs') ()
       where
         bs' = S.dropWhile p bs
diff --git a/Data/Conduit/Internal.hs b/Data/Conduit/Internal.hs
new file mode 100644
--- /dev/null
+++ b/Data/Conduit/Internal.hs
@@ -0,0 +1,281 @@
+{-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE FlexibleInstances #-}
+{-# LANGUAGE MultiParamTypeClasses #-}
+{-# LANGUAGE UndecidableInstances #-}
+{-# LANGUAGE RankNTypes #-}
+module Data.Conduit.Internal
+    ( -- * Types
+      Pipe (..)
+    , Source
+    , Sink
+    , Conduit
+      -- * Functions
+    , pipeClose
+    , pipe
+    , pipeResume
+    , runPipe
+    , sinkToPipe
+    , await
+    , yield
+    , hasInput
+    , transPipe
+    ) where
+
+import Control.Applicative (Applicative (..), (<$>))
+import Control.Monad ((>=>), liftM, ap)
+import Control.Monad.Trans.Class (MonadTrans (lift))
+import Control.Monad.IO.Class (MonadIO (liftIO))
+import Control.Monad.Base (MonadBase (liftBase))
+import Data.Void (Void)
+import Data.Monoid (Monoid (mappend, mempty))
+
+-- | The underlying datatype for all the types in this package.  In has four
+-- type parameters:
+--
+-- * /i/ is the type of values for this @Pipe@'s input stream.
+--
+-- * /o/ is the type of values for this @Pipe@'s output stream.
+--
+-- * /m/ is the underlying monad.
+--
+-- * /r/ is the result type.
+--
+-- Note that /o/ and /r/ are inherently different. /o/ is the type of the
+-- stream of values this @Pipe@ will produce and send downstream. /r/ is the
+-- final output of this @Pipe@.
+--
+-- @Pipe@s can be composed via the 'pipe' function. To do so, the output type
+-- of the left pipe much match the input type of the left pipe, and the result
+-- type of the left pipe must be unit @()@. This is due to the fact that any
+-- result produced by the left pipe must be discarded in favor of the result of
+-- the right pipe.
+--
+-- Since 0.4.0
+data Pipe i o m r =
+    -- | Provide new output to be sent downstream. This constructor has three
+    -- records: the next @Pipe@ to be used, an early-closed function, and the
+    -- output value.
+    HaveOutput (Pipe i o m r) (m r) o
+    -- | Request more input from upstream. The first record takes a new input
+    -- value and provides a new @Pipe@. The second is for early termination. It
+    -- gives a new @Pipe@ which takes no input from upstream. This allows a
+    -- @Pipe@ to provide a final stream of output values after no more input is
+    -- available from upstream.
+  | NeedInput (i -> Pipe i o m r) (Pipe i o m r)
+    -- | Processing with this @Pipe@ is complete. Provides an optional leftover
+    -- input value and and result.
+  | Done (Maybe i) r
+    -- | Require running of a monadic action to get the next @Pipe@. Second
+    -- record is an early cleanup function. Technically, this second record
+    -- could be skipped, but doing so would require extra operations to be
+    -- performed in some cases. For example, for a @Pipe@ pulling data from a
+    -- file, it may be forced to pull an extra, unneeded chunk before closing
+    -- the @Handle@.
+  | PipeM (m (Pipe i o m r)) (m r)
+
+-- | A @Pipe@ which provides a stream of output values, without consuming any
+-- input. The input parameter is set to @()@ instead of @Void@ since there is
+-- no way to statically guarantee that the @NeedInput@ constructor will not be
+-- used. A @Source@ is not used to produce a final result, and thus the result
+-- parameter is set to @()@ as well.
+--
+-- Since 0.4.0
+type Source m a = Pipe Void a m ()
+
+-- | A @Pipe@ which consumes a stream of input values and produces a final
+-- result. It cannot produce any output values, and thus the output parameter
+-- is set to @Void@. In other words, it is impossible to create a @HaveOutput@
+-- constructor for a @Sink@.
+--
+-- Since 0.4.0
+type Sink i m r = Pipe i Void m r
+
+-- | A @Pipe@ which consumes a stream of input values and produces a stream of
+-- output values. It does not produce a result value, and thus the result
+-- parameter is set to @()@.
+--
+-- Since 0.4.0
+type Conduit i m o = Pipe i o m ()
+
+-- | Perform any close actions available for the given @Pipe@.
+--
+-- Since 0.4.0
+pipeClose :: Monad m => Pipe i o m r -> m r
+pipeClose (HaveOutput _ c _) = c
+pipeClose (NeedInput _ p) = pipeClose p
+pipeClose (Done _ r) = return r
+pipeClose (PipeM _ c) = c
+
+pipePush :: Monad m => i -> Pipe i o m r -> Pipe i o m r
+pipePush i (HaveOutput p c o) = HaveOutput (pipePush i p) c o
+pipePush i (NeedInput p _) = p i
+pipePush i (Done _ r) = Done (Just i) r
+pipePush i (PipeM mp c) = PipeM (pipePush i `liftM` mp) c
+
+instance Monad m => Functor (Pipe i o m) where
+    fmap f (HaveOutput p c o) = HaveOutput (f <$> p) (f `liftM` c) o
+    fmap f (NeedInput p c) = NeedInput (fmap f . p) (f <$> c)
+    fmap f (Done l r) = Done l (f r)
+    fmap f (PipeM mp mr) = PipeM ((fmap f) `liftM` mp) (f `liftM` mr)
+
+instance Monad m => Applicative (Pipe i o m) where
+    pure = Done Nothing
+
+    Done Nothing f <*> px = f <$> px
+    Done (Just i) f <*> px = pipePush i $ f <$> px
+    HaveOutput p c o <*> px = HaveOutput (p <*> px) (c `ap` pipeClose px) o
+    NeedInput p c <*> px = NeedInput (\i -> p i <*> px) (c <*> px)
+    PipeM mp c <*> px = PipeM ((<*> px) `liftM` mp) (c `ap` pipeClose px)
+
+instance Monad m => Monad (Pipe i o m) where
+    return = Done Nothing
+
+    Done Nothing x >>= fp = fp x
+    Done (Just i) x >>= fp = pipePush i $ fp x
+    HaveOutput p c o >>= fp = HaveOutput (p >>= fp) (c >>= pipeClose . fp) o
+    NeedInput p c >>= fp = NeedInput (p >=> fp) (c >>= fp)
+    PipeM mp c >>= fp = PipeM ((>>= fp) `liftM` mp) (c >>= pipeClose . fp)
+
+instance MonadBase base m => MonadBase base (Pipe i o m) where
+    liftBase = lift . liftBase
+
+instance MonadTrans (Pipe i o) where
+    lift mr = PipeM (Done Nothing `liftM` mr) mr
+
+instance MonadIO m => MonadIO (Pipe i o m) where
+    liftIO = lift . liftIO
+
+instance Monad m => Monoid (Pipe i o m ()) where
+    mempty = return ()
+    mappend = (>>)
+
+-- | Compose a left and right pipe together into a complete pipe. The left pipe
+-- will be automatically closed when the right pipe finishes, and any leftovers
+-- from the right pipe will be discarded.
+--
+-- This is in fact a wrapper around 'pipeResume'. This function closes the left
+-- @Pipe@ returns by @pipeResume@ and returns only the result.
+--
+-- Since 0.4.0
+pipe :: Monad m => Pipe a b m () -> Pipe b c m r -> Pipe a c m r
+pipe l r = pipeResume l r >>= \(l', res) -> lift (pipeClose l') >> return res
+
+-- | Same as 'pipe', but retain both the new left pipe and the leftovers from
+-- the right pipe. The two components are combined together into a single pipe
+-- and returned, together with the result of the right pipe.
+--
+-- Note: we're biased towards checking the right side first to avoid pulling
+-- extra data which is not needed. Doing so could cause data loss.
+--
+-- Since 0.4.0
+pipeResume :: Monad m => Pipe a b m () -> Pipe b c m r -> Pipe a c m (Pipe a b m (), r)
+
+pipeResume (Done leftoverl ()) (Done leftoverr r) =
+    Done leftoverl (left, r)
+  where
+    left =
+        case leftoverr of
+            Nothing -> mempty
+            Just i -> HaveOutput (Done Nothing ()) (return ()) i
+
+pipeResume left (Done leftoverr r) =
+    Done Nothing (left', r)
+  where
+    left' =
+        case leftoverr of
+            Nothing -> left
+            Just i -> HaveOutput left (pipeClose left) i
+
+-- Left pipe needs more input, ask for it.
+pipeResume (NeedInput p c) right = NeedInput
+    (\a -> pipeResume (p a) right)
+    (do
+        (left, res) <- pipeResume c right
+        lift $ pipeClose left
+        return (mempty, res)
+        )
+
+-- Left pipe has output, right pipe wants it.
+pipeResume (HaveOutput lp _ a) (NeedInput rp _) = pipeResume lp (rp a)
+
+-- Right pipe needs to run a monadic action.
+pipeResume left (PipeM mp c) = PipeM
+    (pipeResume left `liftM` mp)
+    (((,) left) `liftM` c)
+
+-- Right pipe has some output, provide it downstream and continue.
+pipeResume left (HaveOutput p c o) = HaveOutput
+    (pipeResume left p)
+    (((,) left) `liftM` c)
+    o
+
+-- Left pipe is done, right pipe needs input. In such a case, tell the right
+-- pipe there is no more input, and eventually replace its leftovers with the
+-- left pipe's leftover.
+pipeResume (Done l ()) (NeedInput _ c) = ((,) mempty) `liftM` replaceLeftover l c
+
+
+-- Left pipe needs to run a monadic action.
+pipeResume (PipeM mp c) right = PipeM
+    ((`pipeResume` right) `liftM` mp)
+    (c >> liftM ((,) mempty) (pipeClose right))
+
+replaceLeftover :: Monad m => Maybe i -> Pipe i' o m r -> Pipe i o m r
+replaceLeftover l (Done _ r) = Done l r
+replaceLeftover l (HaveOutput p c o) = HaveOutput (replaceLeftover l p) c o
+
+-- This function is only called on pipes when there is no more input available.
+-- Therefore, we can ignore the push record.
+replaceLeftover l (NeedInput _ c) = replaceLeftover l c
+
+replaceLeftover l (PipeM mp c) = PipeM (replaceLeftover l `liftM` mp) c
+
+-- | Run a complete pipeline until processing completes.
+--
+-- Since 0.4.0
+runPipe :: Monad m => Pipe Void Void m r -> m r
+runPipe (HaveOutput _ c _) = c
+runPipe (NeedInput _ c) = runPipe c
+runPipe (Done _ r) = return r
+runPipe (PipeM mp _) = mp >>= runPipe
+
+-- | Send a single output value downstream.
+--
+-- Since 0.4.0
+yield :: Monad m => o -> Pipe i o m ()
+yield = HaveOutput (Done Nothing ()) (return ())
+
+-- | Wait for a single input value from upstream, and remove it from the
+-- stream. Returns @Nothing@ if no more data is available.
+--
+-- Since 0.4.0
+await :: Pipe i o m (Maybe i)
+await = NeedInput (Done Nothing . Just) (Done Nothing Nothing)
+
+-- | Check if input is available from upstream. Will not remove the data from
+-- the stream.
+--
+-- Since 0.4.0
+hasInput :: Pipe i o m Bool
+hasInput = NeedInput (\i -> Done (Just i) True) (Done Nothing False)
+
+-- | A @Sink@ has a @Void@ type parameter for the output, which makes it
+-- difficult to compose with @Source@s and @Conduit@s. This function replaces
+-- that parameter with a free variable. This function is essentially @id@; it
+-- only modifies the types, not the actions performed.
+--
+-- Since 0.4.0
+sinkToPipe :: Monad m => Sink i m r -> Pipe i o m r
+sinkToPipe (HaveOutput _ c _) = lift c
+sinkToPipe (NeedInput p c) = NeedInput (sinkToPipe . p) (sinkToPipe c)
+sinkToPipe (Done i r) = Done i r
+sinkToPipe (PipeM mp c) = PipeM (liftM sinkToPipe mp) c
+
+-- | Transform the monad that a @Pipe@ lives in.
+--
+-- Since 0.4.0
+transPipe :: Monad m => (forall a. m a -> n a) -> Pipe i o m r -> Pipe i o n r
+transPipe f (HaveOutput p c o) = HaveOutput (transPipe f p) (f c) o
+transPipe f (NeedInput p c) = NeedInput (transPipe f . p) (transPipe f c)
+transPipe _ (Done i r) = Done i r
+transPipe f (PipeM mp c) = PipeM (f $ liftM (transPipe f) mp) (f c)
diff --git a/Data/Conduit/Lazy.hs b/Data/Conduit/Lazy.hs
--- a/Data/Conduit/Lazy.hs
+++ b/Data/Conduit/Lazy.hs
@@ -18,12 +18,13 @@
 --
 -- Since 0.3.0
 lazyConsume :: (MonadBaseControl IO m, MonadActive m) => Source m a -> m [a]
-lazyConsume Closed = return []
-lazyConsume (Open src _ x) = do
+lazyConsume (Done _ ()) = return []
+lazyConsume (HaveOutput src _ x) = do
     xs <- lazyConsume src
     return $ x : xs
-lazyConsume (SourceM msrc _) = liftBaseOp_ unsafeInterleaveIO $ do
+lazyConsume (PipeM msrc _) = liftBaseOp_ unsafeInterleaveIO $ do
     a <- monadActive
     if a
         then msrc >>= lazyConsume
         else return []
+lazyConsume (NeedInput _ c) = lazyConsume c
diff --git a/Data/Conduit/List.hs b/Data/Conduit/List.hs
--- a/Data/Conduit/List.hs
+++ b/Data/Conduit/List.hs
@@ -62,7 +62,7 @@
 fold f accum0 =
     go accum0
   where
-    go accum = Processing (push accum) (return accum)
+    go accum = NeedInput (push accum) (return accum)
 
     push accum input =
         let accum' = f accum input
@@ -90,17 +90,17 @@
       => (a -> m ())
       -> Sink a m ()
 mapM_ f =
-    Processing push close
+    NeedInput push close
   where
-    push input = SinkM $ f input >> return (Processing push close)
+    push input = PipeM (f input >> return (NeedInput push close)) (return ())
     close = return ()
 
 -- | Convert a list into a source.
 --
 -- Since 0.3.0
 sourceList :: Monad m => [a] -> Source m a
-sourceList [] = Closed
-sourceList (x:xs) = Open (sourceList xs) (return ()) x
+sourceList [] = Done Nothing ()
+sourceList (x:xs) = HaveOutput (sourceList xs) (return ()) x
 
 -- | Ignore a certain number of values in the stream. This function is
 -- semantically equivalent to:
@@ -114,9 +114,9 @@
 drop :: Monad m
      => Int
      -> Sink a m ()
-drop 0 = Processing (flip Done () . Just) (return ())
+drop 0 = NeedInput (flip Done () . Just) (return ())
 drop count =
-    Processing push (return ())
+    NeedInput push (return ())
   where
     count' = count - 1
     push _
@@ -136,12 +136,12 @@
 take count0 =
     go count0 id
   where
-    go count front = Processing (push count front) (return $ front [])
+    go count front = NeedInput (push count front) (return $ front [])
 
     push 0 front x = Done (Just x) (front [])
     push count front x
         | count' == 0 = Done Nothing (front [x])
-        | otherwise   = Processing (push count' front') (return $ front' [])
+        | otherwise   = NeedInput (push count' front') (return $ front' [])
       where
         count' = count - 1
         front' = front . (x:)
@@ -151,7 +151,7 @@
 -- Since 0.3.0
 head :: Monad m => Sink a m (Maybe a)
 head =
-    Processing push close
+    NeedInput push close
   where
     push x = Done Nothing (Just x)
     close = return Nothing
@@ -162,7 +162,7 @@
 -- Since 0.3.0
 peek :: Monad m => Sink a m (Maybe a)
 peek =
-    Processing push close
+    NeedInput push close
   where
     push x = Done (Just x) (Just x)
     close = return Nothing
@@ -187,7 +187,7 @@
 mapM f =
     NeedInput push close
   where
-    push = flip ConduitM (return ()) . liftM (HaveOutput (NeedInput push close) (return ())) . f
+    push = flip PipeM (return ()) . liftM (HaveOutput (NeedInput push close) (return ())) . f
     close = mempty
 
 -- | Apply a transformation to all values in a stream, concatenating the output
@@ -209,7 +209,7 @@
 concatMapM f =
     NeedInput push close
   where
-    push = flip ConduitM (return ()) . liftM (haveMore (NeedInput push close) (return ())) . f
+    push = flip PipeM (return ()) . liftM (haveMore (NeedInput push close) (return ())) . f
     close = mempty
 
 -- | 'concatMap' with an accumulator.
@@ -241,7 +241,7 @@
 consume =
     go id
   where
-    go front = Processing (push front) (return $ front [])
+    go front = NeedInput (push front) (return $ front [])
     push front x = go (front . (x:))
 
 -- | Grouping input according to an equality function.
@@ -307,7 +307,7 @@
 -- Since 0.3.0
 sinkNull :: Monad m => Sink a m ()
 sinkNull =
-    Processing push close
+    NeedInput push close
   where
     push _ = sinkNull
     close = return ()
@@ -324,12 +324,14 @@
 --
 -- Since 0.3.0
 zip :: Monad m => Source m a -> Source m b -> Source m (a, b)
-zip Closed Closed = Closed
-zip Closed (Open _ close _) = SourceM (close >> return Closed) close
-zip (Open _ close _) Closed = SourceM (close >> return Closed) close
-zip Closed (SourceM _ close) = SourceM (close >> return Closed) close
-zip (SourceM _ close) Closed = SourceM (close >> return Closed) close
-zip (SourceM mx closex) (SourceM my closey) = SourceM (liftM2 zip mx my) (closex >> closey)
-zip (SourceM mx closex) y@(Open _ closey _) = SourceM (liftM (\x -> zip x y) mx) (closex >> closey)
-zip x@(Open _ closex _) (SourceM my closey) = SourceM (liftM (\y -> zip x y) my) (closex >> closey)
-zip (Open srcx closex x) (Open srcy closey y) = Open (zip srcx srcy) (closex >> closey) (x, y)
+zip (Done _ ()) (Done _ ()) = Done Nothing ()
+zip (Done _ ()) (HaveOutput _ close _) = PipeM (close >> return (Done Nothing ())) close
+zip (HaveOutput _ close _) (Done _ ()) = PipeM (close >> return (Done Nothing ())) close
+zip (Done _ ()) (PipeM _ close) = PipeM (close >> return (Done Nothing ())) close
+zip (PipeM _ close) (Done _ ()) = PipeM (close >> return (Done Nothing ())) close
+zip (PipeM mx closex) (PipeM my closey) = PipeM (liftM2 zip mx my) (closex >> closey)
+zip (PipeM mx closex) y@(HaveOutput _ closey _) = PipeM (liftM (\x -> zip x y) mx) (closex >> closey)
+zip x@(HaveOutput _ closex _) (PipeM my closey) = PipeM (liftM (\y -> zip x y) my) (closex >> closey)
+zip (HaveOutput srcx closex x) (HaveOutput srcy closey y) = HaveOutput (zip srcx srcy) (closex >> closey) (x, y)
+zip (NeedInput _ c) right = zip c right
+zip left (NeedInput _ c) = zip left c
diff --git a/Data/Conduit/Types/Conduit.hs b/Data/Conduit/Types/Conduit.hs
deleted file mode 100644
--- a/Data/Conduit/Types/Conduit.hs
+++ /dev/null
@@ -1,63 +0,0 @@
--- | Defines the types for a conduit, which is a transformer of data. A conduit
--- is almost always connected either left (to a source) or right (to a sink).
-module Data.Conduit.Types.Conduit
-    ( Conduit (..)
-    , ConduitPush
-    , ConduitClose
-    ) where
-
-import Control.Monad (liftM)
-import Data.Conduit.Types.Source
-
--- | Pushing new data to a @Conduit@ produces a new @Conduit@.
---
--- Since 0.3.0
-type ConduitPush input m output = input -> Conduit input m output
-
--- | When closing a @Conduit@, it can produce a final stream of values.
---
--- Since 0.3.0
-type ConduitClose m output = Source m output
-
--- | A @Conduit@ allows data to be pushed to it, and for each new input, can
--- produce a stream of output values (possibly an empty stream). It can be
--- considered a hybrid of a @Sink@ and a @Source@.
---
--- A @Conduit@ has four constructors, corresponding to four distinct states of
--- operation.
---
--- Since 0.3.0
-data Conduit input m output =
-    NeedInput (ConduitPush input m output) (ConduitClose m output)
-    -- ^ Indicates that the @Conduit@ needs more input in order to produce
-    -- output. It also provides an action to close the @Conduit@ early, for
-    -- cases when there is no more input available, or when no more output is
-    -- requested. Closing at this point returns a @Source@ to allow for either
-    -- consuming or ignoring the new stream.
-
-  | HaveOutput (Conduit input m output) (m ()) output
-    -- ^ Indicates that the @Conduit@ has more output available. It has three
-    -- records: the next @Conduit@ to continue the stream, a close action for
-    -- early termination, and the output currently available. Note that, unlike
-    -- @NeedInput@, the close action here returns @()@ instead of @Source@. The
-    -- reasoning is that @HaveOutput@ will only be closed early if no more
-    -- output is requested, since no input is required.
-
-  | Finished (Maybe input)
-    -- ^ Indicates that no more output is available, and no more input may be
-    -- sent. It provides an optional leftover input record. Note: It is a
-    -- violation of @Conduit@'s invariants to return leftover output that was
-    -- never consumed, similar to the invariants of a @Sink@.
-
-  | ConduitM (m (Conduit input m output)) (m ())
-    -- ^ Indicates that a monadic action must be taken to determine the next
-    -- @Conduit@. It also provides an early close action. Like @HaveOutput@,
-    -- this action returns @()@, since it should only be used when no more
-    -- output is requested.
-
-instance Monad m => Functor (Conduit input m) where
-    fmap f (NeedInput p c) = NeedInput (fmap f . p) (fmap f c)
-    fmap _ (Finished i) = Finished i
-    fmap f (HaveOutput pull close output) = HaveOutput
-        (fmap f pull) close (f output)
-    fmap f (ConduitM mcon c) = ConduitM (liftM (fmap f) mcon) c
diff --git a/Data/Conduit/Types/Sink.hs b/Data/Conduit/Types/Sink.hs
deleted file mode 100644
--- a/Data/Conduit/Types/Sink.hs
+++ /dev/null
@@ -1,91 +0,0 @@
-{-# LANGUAGE FlexibleContexts #-}
-{-# LANGUAGE MultiParamTypeClasses #-}
-{-# LANGUAGE FlexibleInstances #-}
-{-# LANGUAGE TypeFamilies #-}
-{-# LANGUAGE UndecidableInstances #-}
--- | Defines the types for a sink, which is a consumer of data.
-module Data.Conduit.Types.Sink
-    ( Sink (..)
-    , SinkPush
-    , SinkClose
-    ) where
-
-import Control.Monad.Trans.Class (MonadTrans (lift))
-import Control.Monad.IO.Class (MonadIO (liftIO))
-import Control.Monad (liftM, ap)
-import Control.Applicative (Applicative (..))
-import Control.Monad.Base (MonadBase (liftBase))
-
--- | Push a value into a @Sink@ and get a new @Sink@ as a result.
---
--- Since 0.3.0
-type SinkPush input m output = input -> Sink input m output
-
--- | Closing a @Sink@ returns the final output.
---
--- Since 0.3.0
-type SinkClose m output = m output
-
--- | 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:
---
--- * Some sinks do not actually require any data to produce an output. This is
--- included with a sink in order to allow for a 'Monad' instance.
---
--- * Some sinks will consume all available data and only produce a result at
--- the \"end\" of a data stream (e.g., @sum@).
---
--- Note that you can indicate any leftover data from processing via the @Maybe
--- input@ field of the @Done@ constructor. However, it is a violation of the
--- @Sink@ invariants to return leftover data when no input has been consumed.
--- Concrete, that means that a function like yield is invalid:
---
--- > yield :: input -> Sink input m ()
--- > yield input = Done (Just input) ()
---
--- A @Sink@ should clean up any resources it has allocated when it returns a
--- value.
---
--- Since 0.3.0
-data Sink input m output =
-    Processing (SinkPush input m output) (SinkClose m output) -- ^ Awaiting more input.
-  | Done (Maybe input) output -- ^ Processing complete.
-  | SinkM (m (Sink input m output)) -- ^ Perform some monadic action to continue.
-
-instance Monad m => Functor (Sink input m) where
-    fmap f (Processing push close) = Processing (fmap f . push) (liftM f close)
-    fmap f (Done minput output) = Done minput (f output)
-    fmap f (SinkM msink) = SinkM (liftM (fmap f) msink)
-
-instance Monad m => Applicative (Sink input m) where
-    pure = return
-    (<*>) = ap
-
-instance Monad m => Monad (Sink input m) where
-    return = Done Nothing
-    Done Nothing x >>= f = f x
-    Done (Just leftover) x >>= f =
-        sinkPush (f x)
-      where
-        sinkPush (Processing push _) = push leftover
-        sinkPush (Done Nothing output) = Done (Just leftover) output
-        sinkPush (Done Just{} _) = error $ "Sink invariant violated: leftover input returned without any push"
-        sinkPush (SinkM msink) = SinkM (liftM sinkPush msink)
-    SinkM msink >>= f = SinkM (liftM (>>= f) msink)
-    Processing push close >>= f = Processing
-        (\input -> push input >>= f)
-        (close >>= sinkClose . f)
-
-sinkClose :: Monad m => Sink input m output -> m output
-sinkClose (Done _ output) = return output
-sinkClose (Processing _ close) = close
-sinkClose (SinkM msink) = msink >>= sinkClose
-
-instance MonadBase base m => MonadBase base (Sink input m) where
-    liftBase = lift . liftBase
-
-instance MonadTrans (Sink input) where
-    lift = SinkM . liftM return
-
-instance MonadIO m => MonadIO (Sink input m) where
-    liftIO = lift . liftIO
diff --git a/Data/Conduit/Types/Source.hs b/Data/Conduit/Types/Source.hs
deleted file mode 100644
--- a/Data/Conduit/Types/Source.hs
+++ /dev/null
@@ -1,33 +0,0 @@
-{-# LANGUAGE FlexibleContexts #-}
--- | Defines the types for a source, which is a producer of data.
-module Data.Conduit.Types.Source
-    ( Source (..)
-    ) where
-
-import Data.Monoid (Monoid (..))
-import Control.Monad (liftM)
-
--- | A @Source@ has two operations on it: pull some data, and close the
--- @Source@. A @Source@ should free any resources it allocated when either it
--- returns @Closed@ or when it is explicitly closed (the second record on
--- either the @Open@ or @SourceM@ constructors).
---
--- Since 0.3.0
-data Source m a =
-    Open (Source m a) (m ()) a -- ^ A @Source@ providing more data. Provides records for the next @Source@ in the stream, a close action, and the data provided.
-  | Closed -- ^ A @Source@ which has no more data available.
-  | SourceM (m (Source m a)) (m ()) -- ^ Requires a monadic action to retrieve the next @Source@ in the stream. Second record allows you to close the @Source@.
-
-instance Monad m => Functor (Source m) where
-    fmap f (Open next close a) = Open (fmap f next) close (f a)
-    fmap _ Closed = Closed
-    fmap f (SourceM msrc close) = SourceM (liftM (fmap f) msrc) close
-
-instance Monad m => Monoid (Source m a) where
-    mempty = Closed
-    mappend x Closed = x
-    mappend Closed y = y
-    mappend (Open next close a) y = Open (mappend next y) close a
-    mappend (SourceM msrc close) y = SourceM (do
-        src <- msrc
-        return (mappend src y)) close
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
@@ -10,8 +10,6 @@
     , ConduitStateResult (..)
     , conduitIO
     , ConduitIOResult (..)
-    , transConduit
-    , conduitClose
       -- *** Sequencing
     , SequencedSink
     , sequenceSink
@@ -21,13 +19,8 @@
 
 import Prelude hiding (sequence)
 import Control.Monad.Trans.Resource
-import Data.Conduit.Types.Conduit
-import Data.Conduit.Types.Sink
-import Data.Conduit.Types.Source
-import Data.Conduit.Util.Source
-import Data.Conduit.Util.Sink
+import Data.Conduit.Internal
 import Control.Monad (liftM)
-import Data.Monoid (mempty)
 
 -- | A helper function for returning a list of values from a @Conduit@.
 --
@@ -65,14 +58,14 @@
 conduitState state0 push0 close0 =
     NeedInput (push state0) (close state0)
   where
-    push state input = ConduitM (liftM goRes' $ state `seq` push0 state input) (return ())
+    push state input = PipeM (liftM goRes' $ state `seq` push0 state input) (return ())
 
-    close state = SourceM (do
+    close state = PipeM (do
         os <- close0 state
         return $ fromList os) (return ())
 
     goRes' (StateFinished leftover output) = haveMore
-        (Finished leftover)
+        (Done leftover ())
         (return ())
         output
     goRes' (StateProducing state output) = haveMore
@@ -102,10 +95,10 @@
            -> (state -> m [output]) -- ^ Close function. Note that this need not explicitly perform any cleanup.
            -> Conduit input m output
 conduitIO alloc cleanup push0 close0 = NeedInput
-    (\input -> flip ConduitM (return ()) $ do
+    (\input -> flip PipeM (return ()) $ do
         (key, state) <- allocate alloc cleanup
         push key state input)
-    (SourceM (do
+    (PipeM (do
         (key, state) <- allocate alloc cleanup
         os <- close0 state
         release key
@@ -115,43 +108,24 @@
         res <- push0 state input
         case res of
             IOProducing output -> return $ haveMore
-                (NeedInput (flip ConduitM (release key) . push key state) (close key state))
+                (NeedInput (flip PipeM (release key) . push key state) (close key state))
                 (release key >> return ())
                 output
             IOFinished leftover output -> do
                 release key
                 return $ haveMore
-                    (Finished leftover)
+                    (Done leftover ())
                     (return ())
                     output
 
-    close key state = SourceM (do
+    close key state = PipeM (do
         output <- close0 state
         release key
         return $ fromList output) (release key)
 
-fromList :: Monad m => [a] -> Source m a
-fromList [] = Closed
-fromList (x:xs) = Open (fromList xs) (return ()) x
-
--- | Transform the monad a 'Conduit' lives in.
---
--- See @transSource@ for more information.
---
--- Since 0.3.0
-transConduit :: Monad m
-             => (forall a. m a -> n a)
-             -> Conduit input m output
-             -> Conduit input n output
-transConduit _ (Finished a) = Finished a
-transConduit f (NeedInput push close) = NeedInput
-    (transConduit f . push)
-    (transSource f close)
-transConduit f (HaveOutput pull close output) = HaveOutput
-    (transConduit f pull)
-    (f close)
-    output
-transConduit f (ConduitM mcon close) = ConduitM (f (liftM (transConduit f) mcon)) (f close)
+fromList :: Monad m => [a] -> Pipe i a m ()
+fromList [] = Done Nothing ()
+fromList (x:xs) = HaveOutput (fromList xs) (return ()) x
 
 -- | Return value from a 'SequencedSink'.
 --
@@ -177,50 +151,18 @@
     => state -- ^ initial state
     -> SequencedSink state input m output
     -> Conduit input m output
-sequenceSink state0 fsink = NeedInput (scPush fsink $ fsink state0) mempty -- FIXME investigate if we can bypass getting input
-
-scPush :: Monad m
-       => SequencedSink state input m output
-       -> Sink input m (SequencedSinkResponse state input m output)
-       -> ConduitPush input m output
-scPush fsink (Processing pushI _) input = scGoRes fsink $ pushI input
-scPush fsink (Done Nothing res) input = scGoRes fsink (Done (Just input) res)
-scPush _ (Done Just{} _) _ = error "Invariant violated: Sink returned leftover without input"
-scPush fsink (SinkM msink) input = ConduitM (liftM (\sink -> scPush fsink sink input) msink) (msink >>= sinkClose)
-
-scGoRes :: Monad m
-        => SequencedSink state input m output
-        -> Sink input m (SequencedSinkResponse state input m output)
-        -> Conduit input m output
-scGoRes fsink (Done (Just leftover) (Emit state os)) = haveMore
-    (scPush fsink (fsink state) leftover)
-    (return ())
-    os
-scGoRes fsink (Done Nothing (Emit state os)) = haveMore
-    (NeedInput p c)
-    (return ())
-    os
-  where
-    NeedInput p c = sequenceSink state fsink -- FIXME
-scGoRes fsink (Processing pushI closeI) = NeedInput
-    (scPush fsink (Processing pushI closeI))
-    (SourceM (closeI >>= goRes) (closeI >> return ()))
-  where
-    goRes (Emit _ os) = return $ fromList os
-    goRes Stop = return Closed
-    goRes (StartConduit (NeedInput _ closeC)) = return closeC
-    goRes (StartConduit (Finished _)) = return Closed
-    goRes (StartConduit (ConduitM mcon _)) = mcon >>= goRes . StartConduit
-    goRes (StartConduit HaveOutput{}) = error "scGoRes:goRes: StartConduit HaveOutput not supported yet"
-scGoRes _ (Done mleftover Stop) = Finished mleftover
-scGoRes _ (Done Nothing (StartConduit c)) = c
-scGoRes _ (Done (Just leftover) (StartConduit (Finished Nothing))) = Finished (Just leftover)
-scGoRes _ (Done Just{} (StartConduit (Finished Just{}))) = error "Invariant violated: conduit returns leftover without push"
-scGoRes _ (Done (Just leftover) (StartConduit (NeedInput p _))) = p leftover
-scGoRes _ (Done Just{} (StartConduit HaveOutput{})) = error "scGoRes: StartConduit HaveOutput not supported yet"
-scGoRes fsink (Done mleftover (StartConduit (ConduitM mcon close))) =
-    ConduitM (liftM (scGoRes fsink . Done mleftover . StartConduit) mcon) close
-scGoRes fsink (SinkM msink) = ConduitM (liftM (scGoRes fsink) msink) (msink >>= sinkClose)
+sequenceSink state0 fsink = do
+    x <- hasInput
+    if x
+        then do
+            res <- sinkToPipe $ fsink state0
+            case res of
+                Emit state os -> do
+                    fromList os
+                    sequenceSink state fsink
+                Stop -> return ()
+                StartConduit c -> c
+        else return ()
 
 -- | Specialised version of 'sequenceSink'
 --
@@ -230,43 +172,10 @@
 --
 -- Since 0.3.0
 sequence :: Monad m => Sink input m output -> Conduit input m output
-sequence (Processing spush0 sclose0) =
-    NeedInput (push spush0) (close sclose0)
-  where
-    push spush input = goRes $ spush input
-
-    goRes res =
-        case res of
-            Processing spush'' sclose'' ->
-                NeedInput (push spush'') (close sclose'')
-            Done Nothing output -> HaveOutput
-                (NeedInput (push spush0) (close sclose0))
-                (return ())
-                output
-            Done (Just input') output -> HaveOutput
-                (goRes $ spush0 input')
-                (return ())
-                output
-            SinkM msink -> ConduitM (liftM goRes msink) (msink >>= sinkClose)
-
-    close sclose = SourceM (do
-        output <- sclose
-        return $ Open Closed (return ()) output) (return ())
-
-sequence (Done Nothing output) = NeedInput
-    (\_input ->
-        let x = HaveOutput x (return ()) output
-         in x)
-    (   let src = Open src (return ()) output
-         in src)
-sequence (Done Just{} _) = error "Invariant violated: sink returns leftover without push"
-sequence (SinkM msink) = ConduitM (liftM sequence msink) (msink >>= sinkClose)
-
--- | Close a @Conduit@ early, discarding any output.
---
--- Since 0.3.0
-conduitClose :: Monad m => Conduit input m output -> m ()
-conduitClose (NeedInput _ c) = sourceClose c
-conduitClose Finished{} = return ()
-conduitClose (HaveOutput _ c _) = c
-conduitClose (ConduitM _ c) = c
+sequence sink = do
+    x <- hasInput
+    if x
+        then do
+            sinkToPipe sink >>= yield
+            sequence sink
+        else return ()
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
@@ -9,13 +9,11 @@
     , SinkStateResult (..)
     , sinkIO
     , SinkIOResult (..)
-    , transSink
-    , sinkClose
     ) where
 
 import Control.Monad.Trans.Resource
-import Data.Conduit.Types.Sink
-import Control.Monad (liftM)
+import Control.Monad.Trans.Class (lift)
+import Data.Conduit.Internal
 
 -- | 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
@@ -37,14 +35,18 @@
     -> (state -> m output) -- ^ Close. Note that the state is not returned, as it is not needed.
     -> Sink input m output
 sinkState state0 push0 close0 =
-    Processing (push state0) (close0 state0)
+    NeedInput (push state0) (close state0)
   where
-    push state input = SinkM $ 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
+    push state input = PipeM
+        (do
+            res <- state `seq` push0 state input
+            case res of
+                StateProcessing state' -> return $ NeedInput (push state') (close state')
+                StateDone mleftover output -> return $ Done mleftover output)
+        (close0 state)
 
+    close = lift . close0
+
 -- | A helper type for @sinkIO@, indicating the result of being pushed to. It
 -- can either indicate that processing is done, or to continue.
 --
@@ -61,13 +63,15 @@
        -> (state -> input -> m (SinkIOResult input output)) -- ^ push
        -> (state -> m output) -- ^ close
        -> Sink input m output
-sinkIO alloc cleanup push0 close0 = Processing
-    (\input -> SinkM $ do
+sinkIO alloc cleanup push0 close0 = NeedInput
+    (\input -> PipeM (do
         (key, state) <- allocate alloc cleanup
-        push key state input)
+        push key state input) (do
+            (key, state) <- allocate alloc cleanup
+            close key state))
     (do
-        (key, state) <- allocate alloc cleanup
-        close key state)
+        (key, state) <- lift $ allocate alloc cleanup
+        lift $ close key state)
   where
     push key state input = do
         res <- push0 state input
@@ -75,31 +79,12 @@
             IODone a b -> do
                 release key
                 return $ Done a b
-            IOProcessing -> return $ Processing
-                (SinkM . push key state)
-                (close key state)
+            IOProcessing -> return $ NeedInput
+                (\i ->
+                    let mpipe = push key state i
+                     in PipeM mpipe (mpipe >>= pipeClose))
+                (lift $ close key state)
     close key state = do
         res <- close0 state
         release key
         return res
-
--- | Transform the monad a 'Sink' lives in.
---
--- See @transSource@ for more information.
---
--- Since 0.3.0
-transSink :: Monad m
-          => (forall a. m a -> n a)
-          -> Sink input m output
-          -> Sink input n output
-transSink _ (Done a b) = Done a b
-transSink f (Processing push close) = Processing (transSink f . push) (f close)
-transSink f (SinkM msink) = SinkM (f (liftM (transSink f) msink))
-
--- | Close a @Sink@ if it is still open, discarding any output it produces.
---
--- Since 0.3.0
-sinkClose :: Monad m => Sink input m output -> m ()
-sinkClose (SinkM msink) = msink >>= sinkClose
-sinkClose Done{} = return ()
-sinkClose (Processing _ close) = close >> return ()
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
@@ -11,13 +11,10 @@
     , SourceStateResult (..)
     , sourceIO
     , SourceIOResult (..)
-    , transSource
-    , sourceClose
     ) where
 
 import Control.Monad.Trans.Resource
-import Data.Conduit.Types.Source
-import Control.Monad (liftM)
+import Data.Conduit.Internal
 
 -- | The return value when pulling in the @sourceState@ function. Either
 -- indicates no more data, or the next value and an updated state.
@@ -37,13 +34,13 @@
 sourceState state0 pull0 =
     src state0
   where
-    src state = SourceM (pull state) (return ())
+    src state = PipeM (pull state) (return ())
 
     pull state = do
         res <- pull0 state
         return $ case res of
-            StateOpen state' val -> Open (src state') (return ()) val
-            StateClosed -> Closed
+            StateOpen state' val -> HaveOutput (src state') (return ()) val
+            StateClosed -> Done Nothing ()
 
 -- | The return value when pulling in the @sourceIO@ function. Either indicates
 -- no more data, or the next value.
@@ -60,19 +57,19 @@
           -> (state -> m (SourceIOResult output)) -- ^ Pull function. Note that this should not perform any cleanup.
           -> Source m output
 sourceIO alloc cleanup pull0 =
-    SourceM (do
+    PipeM (do
         (key, state) <- allocate alloc cleanup
         pull key state) (return ())
   where
-    src key state = SourceM (pull key state) (release key)
+    src key state = PipeM (pull key state) (release key)
 
     pull key state = do
         res <- pull0 state
         case res of
             IOClosed -> do
                 release key
-                return Closed
-            IOOpen val -> return $ Open (src key state) (release key) val
+                return $ Done Nothing ()
+            IOOpen val -> return $ HaveOutput (src key state) (release key) val
 
 -- | A combination of 'sourceIO' and 'sourceState'.
 --
@@ -83,39 +80,18 @@
               -> (state -> m (SourceStateResult state output)) -- ^ Pull function. Note that this need not explicitly perform any cleanup.
               -> Source m output
 sourceStateIO alloc cleanup pull0 =
-    SourceM (do
+    PipeM (do
         (key, state) <- allocate alloc cleanup
         pull key state) (return ())
   where
-    src key state = SourceM (pull key state) (release key)
+    src key state = PipeM (pull key state) (release key)
 
     pull key state = do
         res <- pull0 state
         case res of
             StateClosed -> do
                 release key
-                return Closed
-            StateOpen state' val -> return $ Open (src key state') (release key) val
-
--- | Transform the monad a 'Source' lives in.
---
--- 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.3.0
-transSource :: Monad m
-            => (forall a. m a -> n a)
-            -> Source m output
-            -> Source n output
-transSource f (Open next close output) = Open (transSource f next) (f close) output
-transSource _ Closed = Closed
-transSource f (SourceM msrc close) = SourceM (f (liftM (transSource f) msrc)) (f close)
+                return $ Done Nothing ()
+            StateOpen state' val -> return $ HaveOutput (src key state') (release key) val
 
--- | Close a @Source@, regardless of its current state.
---
--- Since 0.3.0
-sourceClose :: Monad m => Source m a -> m ()
-sourceClose Closed = return ()
-sourceClose (Open _ close _) = close
-sourceClose (SourceM _ close) = close
+-- FIXME transPipe
diff --git a/System/PosixFile.hsc b/System/PosixFile.hsc
--- a/System/PosixFile.hsc
+++ b/System/PosixFile.hsc
@@ -8,7 +8,11 @@
 
 import Foreign.C.String (CString, withCString)
 import Foreign.Marshal.Alloc (mallocBytes, free)
+#if __GLASGOW_HASKELL__ >= 704
+import Foreign.C.Types (CInt (..))
+#else
 import Foreign.C.Types (CInt)
+#endif
 import Foreign.C.Error (throwErrno)
 import Foreign.Ptr (Ptr)
 import Data.Bits (Bits)
diff --git a/conduit.cabal b/conduit.cabal
--- a/conduit.cabal
+++ b/conduit.cabal
@@ -1,11 +1,13 @@
 Name:                conduit
-Version:             0.3.0
+Version:             0.4.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.4] Inspired by the design of the pipes package: we now have a single unified type underlying @Source@, @Sink@, and @Conduit@. This type is named @Pipe@. There are type synonyms provided for the other three types. Additionally, @BufferedSource@ is no longer provided. Instead, the connect-and-resume operator, @$$&@, can be used for the same purpose.
+    .
     [0.3] ResourceT has been greatly simplified, specialized for IO, and moved into a separate package. Instead of hard-coding ResourceT into the conduit datatypes, they can now live around any monad. The Conduit datatype has been enhanced to better allow generation of streaming output. The SourceResult, SinkResult, and ConduitResult datatypes have been removed entirely.
 	.
     [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.
@@ -42,10 +44,8 @@
                        Data.Conduit.Text
                        Data.Conduit.List
                        Data.Conduit.Lazy
-  Other-modules:       Data.Conduit.Types.Source
-                       Data.Conduit.Types.Sink
-                       Data.Conduit.Types.Conduit
-                       Data.Conduit.Util.Source
+                       Data.Conduit.Internal
+  Other-modules:       Data.Conduit.Util.Source
                        Data.Conduit.Util.Sink
                        Data.Conduit.Util.Conduit
   Build-depends:       base                     >= 4.3          && < 5
@@ -54,9 +54,10 @@
                      , transformers-base        >= 0.4.1        && < 0.5
                      , monad-control            >= 0.3.1        && < 0.4
                      , containers
-                     , transformers             >= 0.2.2        && < 0.3
+                     , transformers             >= 0.2.2        && < 0.4
                      , bytestring               >= 0.9
                      , text                     >= 0.11
+                     , void                     >= 0.5.5        && < 0.6
   ghc-options:     -Wall
   if flag(debug)
     cpp-options: -DDEBUG
diff --git a/test/main.hs b/test/main.hs
--- a/test/main.hs
+++ b/test/main.hs
@@ -143,11 +143,10 @@
     describe "resumable sources" $ do
         it "simple" $ do
             (x, y, z) <- runResourceT $ do
-                bs <- C.bufferSource $ CL.sourceList [1..10 :: Int]
-                x <- bs C.$$ CL.take 5
-                y <- bs C.$$ CL.fold (+) 0
-                z <- bs C.$$ CL.consume
-                C.bsourceClose bs
+                let src1 = CL.sourceList [1..10 :: Int]
+                (src2, x) <- src1 C.$$+ CL.take 5
+                (src3, y) <- src2 C.$$+ CL.fold (+) 0
+                z <- src3 C.$$ CL.consume
                 return (x, y, z)
             x @?= [1..5] :: IO ()
             y @?= sum [6..10]
@@ -198,12 +197,12 @@
     describe "isolate" $ do
         it "bound to resumable source" $ do
             (x, y) <- runResourceT $ do
-                bsrc <- C.bufferSource $ CL.sourceList [1..10 :: Int]
-                x <- bsrc C.$= CL.isolate 5 C.$$ CL.consume
-                y <- bsrc C.$$ CL.consume
+                let src1 = CL.sourceList [1..10 :: Int]
+                (src2, x) <- src1 C.$= CL.isolate 5 C.$$+ CL.consume
+                y <- src2 C.$$ CL.consume
                 return (x, y)
             x @?= [1..5]
-            y @?= [6..10]
+            y @?= []
 
         it "bound to sink, non-resumable" $ do
             (x, y) <- runResourceT $ do
@@ -216,9 +215,9 @@
 
         it "bound to sink, resumable" $ do
             (x, y) <- runResourceT $ do
-                bsrc <- C.bufferSource $ CL.sourceList [1..10 :: Int]
-                x <- bsrc C.$$ CL.isolate 5 C.=$ CL.consume
-                y <- bsrc C.$$ CL.consume
+                let src1 = CL.sourceList [1..10 :: Int]
+                (src2, x) <- src1 C.$$+ CL.isolate 5 C.=$ CL.consume
+                y <- src2 C.$$ CL.consume
                 return (x, y)
             x @?= [1..5]
             y @?= [6..10]
@@ -354,12 +353,46 @@
     describe "unbuffering" $ do
         it "works" $ do
             x <- runResourceT $ do
-                bsrc <- C.bufferSource $ CL.sourceList [1..10 :: Int]
-                bsrc C.$$ CL.drop 5
-                let src = C.unbufferSource bsrc
-                src C.$$ CL.fold (+) 0
+                let src1 = CL.sourceList [1..10 :: Int]
+                (src2, ()) <- src1 C.$$+ CL.drop 5
+                src2 C.$$ CL.fold (+) 0
             x @?= sum [6..10]
 
+    describe "operators" $ do
+        it "only use =$=" $
+            runIdentity
+            (    CL.sourceList [1..10 :: Int]
+              C.$$ CL.map (+ 1)
+             C.=$= CL.map (subtract 1)
+             C.=$= CL.map (* 2)
+             C.=$= CL.map (`div` 2)
+             C.=$= CL.fold (+) 0
+            ) @?= sum [1..10]
+        it "only use =$" $
+            runIdentity
+            (    CL.sourceList [1..10 :: Int]
+              C.$$ CL.map (+ 1)
+              C.=$ CL.map (subtract 1)
+              C.=$ CL.map (* 2)
+              C.=$ CL.map (`div` 2)
+              C.=$ CL.fold (+) 0
+            ) @?= sum [1..10]
+        it "chain" $ do
+            x <-      CL.sourceList [1..10 :: Int]
+                C.$=  CL.map (+ 1)
+                C.=$= CL.map (+ 1)
+                C.$=  CL.map (+ 1)
+                C.=$= CL.map (subtract 3)
+                C.=$= CL.map (* 2)
+                C.$$  CL.map (`div` 2)
+                C.=$= CL.map (+ 1)
+                C.=$  CL.map (+ 1)
+                C.=$= CL.map (+ 1)
+                C.=$  CL.map (subtract 3)
+                C.=$= CL.fold (+) 0
+            x @?= sum [1..10]
+
+
     describe "properly using binary file reading" $ do
         it "sourceFile" $ do
             x <- runResourceT $ CB.sourceFile "test/random" C.$$ CL.consume
@@ -430,13 +463,6 @@
             x <- runResourceT $ do
                 let src = CL.sourceList ["foobarbazbin"]
                 src C.$= CB.isolate 10 C.$$ CL.head
-            x @?= Just "foobarbazb"
-
-    describe "bufferedFuseLeft" $ do
-        it "does not double close conduit" $ do
-            x <- runResourceT $ do
-                bsrc <- C.bufferSource $ CL.sourceList ["foobarbazbin"]
-                bsrc C.$= CB.isolate 10 C.$$ CL.head
             x @?= Just "foobarbazb"
 
     describe "binary" $ do