diff --git a/Data/Conduit.hs b/Data/Conduit.hs
--- a/Data/Conduit.hs
+++ b/Data/Conduit.hs
@@ -40,6 +40,7 @@
     , yield
     , hasInput
     , transPipe
+    , mapOutput
       -- ** Source
     , module Data.Conduit.Util.Source
       -- ** Sink
diff --git a/Data/Conduit/Binary.hs b/Data/Conduit/Binary.hs
--- a/Data/Conduit/Binary.hs
+++ b/Data/Conduit/Binary.hs
@@ -278,7 +278,20 @@
 --
 -- Since 0.3.0
 take :: Monad m => Int -> Sink S.ByteString m L.ByteString
-take n = L.fromChunks `liftM` (isolate n =$ CL.consume)
+take n0 =
+    go n0 id
+  where
+    go n front =
+        NeedInput push close
+      where
+        push bs =
+            case S.length bs `compare` n of
+                LT -> go (n - S.length bs) (front . (bs:))
+                EQ -> Done Nothing $ L.fromChunks $ front [bs]
+                GT ->
+                    let (x, y) = S.splitAt n bs
+                     in Done (Just y) $ L.fromChunks $ front [x]
+        close = return $ L.fromChunks $ front []
 
 -- | Split the input bytes into lines. In other words, split on the LF byte
 -- (10), and strip it from the output.
diff --git a/Data/Conduit/Internal.hs b/Data/Conduit/Internal.hs
--- a/Data/Conduit/Internal.hs
+++ b/Data/Conduit/Internal.hs
@@ -1,3 +1,4 @@
+{-# OPTIONS_HADDOCK not-home #-}
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE FlexibleInstances #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
@@ -9,6 +10,7 @@
     , Source
     , Sink
     , Conduit
+    , Finalize (..)
       -- * Functions
     , pipeClose
     , pipe
@@ -19,6 +21,9 @@
     , yield
     , hasInput
     , transPipe
+    , mapOutput
+    , runFinalize
+    , addCleanup
     ) where
 
 import Control.Applicative (Applicative (..), (<$>))
@@ -26,9 +31,53 @@
 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.Void (Void, absurd)
 import Data.Monoid (Monoid (mappend, mempty))
+import Control.Monad.Trans.Resource
 
+-- | A cleanup action to be performed.
+--
+-- Previously, we just had a plain action. However, most @Pipe@s simply have
+-- empty cleanup actions, and storing a large set of them wastes memory. But
+-- having strict fields and distinguishing between pure and impure actions, we
+-- can keep memory usage constant, and only allocate memory for the actual
+-- actions we have to track.
+--
+-- Since 0.4.1
+data Finalize m r = FinalizePure !r
+                  | FinalizeM !(m r)
+
+instance Monad m => Functor (Finalize m) where
+    fmap f (FinalizePure r) = FinalizePure (f r)
+    fmap f (FinalizeM mr) = FinalizeM (liftM f mr)
+
+instance Monad m => Applicative (Finalize m) where
+    pure = FinalizePure
+    (<*>) = ap
+
+instance Monad m => Monad (Finalize m) where
+    return = FinalizePure
+    FinalizePure x >>= f = f x
+    FinalizeM mx >>= f = FinalizeM $ mx >>= \x ->
+        case f x of
+            FinalizePure y -> return y
+            FinalizeM my -> my
+
+instance MonadTrans Finalize where
+    lift = FinalizeM
+
+instance MonadThrow m => MonadThrow (Finalize m) where
+    monadThrow = lift . monadThrow
+
+instance MonadIO m => MonadIO (Finalize m) where
+    liftIO = lift . liftIO
+
+instance MonadResource m => MonadResource (Finalize m) where
+    allocate a = lift . allocate a
+    register = lift . register
+    release = lift . release
+    resourceMask = lift . resourceMask
+
 -- | The underlying datatype for all the types in this package.  In has four
 -- type parameters:
 --
@@ -53,31 +102,30 @@
 -- 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
+    -- fields: 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
+    HaveOutput (Pipe i o m r) !(Finalize m r) !o
+    -- | Request more input from upstream. The first field 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)
+  | 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
+  | 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
+    -- field is an early cleanup function. Technically, this second field
     -- 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)
+  | PipeM !(m (Pipe i o m r)) !(Finalize 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.
+-- input. The input parameter is set to @Void@ to indicate that this @Pipe@
+-- takes no input.  A @Source@ is not used to produce a final result, and thus
+-- the result parameter is set to @()@.
 --
 -- Since 0.4.0
 type Source m a = Pipe Void a m ()
@@ -100,10 +148,10 @@
 -- | Perform any close actions available for the given @Pipe@.
 --
 -- Since 0.4.0
-pipeClose :: Monad m => Pipe i o m r -> m r
+pipeClose :: Monad m => Pipe i o m r -> Finalize m r
 pipeClose (HaveOutput _ c _) = c
 pipeClose (NeedInput _ p) = pipeClose p
-pipeClose (Done _ r) = return r
+pipeClose (Done _ r) = FinalizePure r
 pipeClose (PipeM _ c) = c
 
 pipePush :: Monad m => i -> Pipe i o m r -> Pipe i o m r
@@ -140,7 +188,7 @@
     liftBase = lift . liftBase
 
 instance MonadTrans (Pipe i o) where
-    lift mr = PipeM (Done Nothing `liftM` mr) mr
+    lift mr = PipeM (Done Nothing `liftM` mr) (FinalizeM mr)
 
 instance MonadIO m => MonadIO (Pipe i o m) where
     liftIO = lift . liftIO
@@ -158,7 +206,7 @@
 --
 -- 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
+pipe l r = pipeResume l r >>= \(l', res) -> lift (runFinalize $ 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
@@ -169,59 +217,82 @@
 --
 -- 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
+pipeResume left right =
+    -- We're using a case statement instead of pattern matching in the function
+    -- itself to make the logic explicit. We first check the right pipe, and
+    -- only if the right pipe is asking for more input do we process the left
+    -- pipe.
+    case right of
+        -- Right pipe is done, grab leftovers and the left pipe
+        Done leftoverr r ->
+            -- Get any leftovers from the left pipe, the current state of the
+            -- left pipe (sans leftovers), and a close action for the left
+            -- pipe.
+            let (leftover, left', leftClose) =
+                    case left of
+                        Done leftoverl () -> (leftoverl, Done Nothing (), FinalizePure ())
+                        _ -> (Nothing, left, pipeClose left)
+            -- Combine the current state of the left pipe with any leftovers
+            -- from the right pipe.
+                left'' =
+                    case leftoverr of
+                        Just a -> HaveOutput left' leftClose a
+                        Nothing -> left'
+            -- Return the leftovers, the final left pipe state, and the result.
+             in Done leftover (left'', r)
 
--- 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.
+        PipeM mp c -> PipeM
+            (pipeResume left `liftM` mp)
+            (((,) left) `fmap` c)
 
--- 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.
+        HaveOutput p c o -> HaveOutput
+            (pipeResume left p)
+            (((,) left) `fmap` c)
+            o
 
--- Right pipe has some output, provide it downstream and continue.
-pipeResume left (HaveOutput p c o) = HaveOutput
-    (pipeResume left p)
-    (((,) left) `liftM` c)
-    o
+        -- Right pipe needs input, so let's get it
+        NeedInput rp rc ->
+            case left of
+                -- Left pipe has output, right pipe wants it.
+                HaveOutput lp _ a -> pipeResume lp $ rp a
 
--- Now we've dealt with all right constructor except for NeedInput. Since the
--- right pipe needs more input, we can process the left pipe.
+                -- Left pipe needs more input, ask for it.
+                NeedInput p c -> NeedInput
+                    (\a -> pipeResume (p a) right)
+                    (do
+                        -- There is no more input available, so connect the
+                        -- no-more-input record with the right.
+                        (left', res) <- pipeResume c right
 
--- Left pipe needs more input, ask for it.
-pipeResume (NeedInput p c) right@NeedInput{} = NeedInput
-    (\a -> pipeResume (p a) right)
-    (do
-        (left, res) <- pipeResume c right
-        lift $ pipeClose left
-        return (mempty, res)
-        )
+                        -- Theoretically, we could return the left' value as
+                        -- the first element in the tuple. However, it is not
+                        -- recommended to give input to a pipe after it has
+                        -- been told there is no more input. Instead, we close
+                        -- the pipe and return mempty in its place.
+                        lift $ runFinalize $ pipeClose left'
+                        return (mempty, res)
+                        )
 
--- 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 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.
+                Done l () -> ((,) mempty) `liftM` replaceLeftover l rc
 
+                -- Left pipe needs to run a monadic action.
+                PipeM mp c -> PipeM
+                    ((`pipeResume` right) `liftM` mp)
+                    (fmap ((,) mempty) $ combineFinalize c $ pipeClose right)
 
--- Left pipe needs to run a monadic action.
-pipeResume (PipeM mp c) right@NeedInput{} = PipeM
-    ((`pipeResume` right) `liftM` mp)
-    (c >> liftM ((,) mempty) (pipeClose right))
+-- | A minor optimization on @>>@ which does not cause any allocations for the
+-- common case of missing left actions.
+--
+-- Since 0.4.1
+combineFinalize :: Monad m => Finalize m () -> Finalize m r -> Finalize m r
+combineFinalize (FinalizePure ()) f = f
+combineFinalize (FinalizeM x) (FinalizeM y) = FinalizeM $ x >> y
+combineFinalize (FinalizeM x) (FinalizePure y) = FinalizeM $ x >> return y
 
 replaceLeftover :: Monad m => Maybe i -> Pipe i' o m r -> Pipe i o m r
 replaceLeftover l (Done _ r) = Done l r
@@ -237,16 +308,23 @@
 --
 -- Since 0.4.0
 runPipe :: Monad m => Pipe Void Void m r -> m r
-runPipe (HaveOutput _ c _) = c
+runPipe (HaveOutput _ c _) = runFinalize c
 runPipe (NeedInput _ c) = runPipe c
 runPipe (Done _ r) = return r
 runPipe (PipeM mp _) = mp >>= runPipe
 
+-- | Perform any necessary finalization actions.
+--
+-- Since 0.4.1
+runFinalize :: Monad m => Finalize m r -> m r
+runFinalize (FinalizePure r) = return r
+runFinalize (FinalizeM mr) = mr
+
 -- | Send a single output value downstream.
 --
 -- Since 0.4.0
 yield :: Monad m => o -> Pipe i o m ()
-yield = HaveOutput (Done Nothing ()) (return ())
+yield = HaveOutput (Done Nothing ()) (FinalizePure ())
 
 -- | Wait for a single input value from upstream, and remove it from the
 -- stream. Returns @Nothing@ if no more data is available.
@@ -269,7 +347,7 @@
 --
 -- Since 0.4.0
 sinkToPipe :: Monad m => Sink i m r -> Pipe i o m r
-sinkToPipe (HaveOutput _ c _) = lift c
+sinkToPipe (HaveOutput _ _ o) = absurd o
 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
@@ -278,7 +356,44 @@
 --
 -- 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 (HaveOutput p c o) = HaveOutput (transPipe f p) (transFinalize 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)
+transPipe f (PipeM mp c) = PipeM (f $ liftM (transPipe f) mp) (transFinalize f c)
+
+transFinalize :: (forall a. m a -> n a) -> Finalize m r -> Finalize n r
+transFinalize _ (FinalizePure r) = FinalizePure r
+transFinalize f (FinalizeM mr) = FinalizeM $ f mr
+
+-- | Apply a function to all the output values of a `Pipe`.
+--
+-- This mimics the behavior of `fmap` for a `Source` and `Conduit` in pre-0.4
+-- days.
+--
+-- Since 0.4.1
+mapOutput :: Monad m => (o1 -> o2) -> Pipe i o1 m r -> Pipe i o2 m r
+mapOutput f (HaveOutput p c o) = HaveOutput (mapOutput f p) c (f o)
+mapOutput f (NeedInput p c) = NeedInput (mapOutput f . p) (mapOutput f c)
+mapOutput _ (Done i r) = Done i r
+mapOutput f (PipeM mp c) = PipeM (liftM (mapOutput f) mp) c
+
+-- | Add some code to be run when the given @Pipe@ cleans up.
+--
+-- Since 0.4.1
+addCleanup :: Monad m
+           => (Bool -> m ()) -- ^ @True@ if @Pipe@ ran to completion, @False@ for early termination.
+           -> Pipe i o m r
+           -> Pipe i o m r
+addCleanup cleanup (Done leftover r) = PipeM
+    (cleanup True >> return (Done leftover r))
+    (lift (cleanup True) >> return r)
+addCleanup cleanup (HaveOutput src close x) = HaveOutput
+    (addCleanup cleanup src)
+    (lift (cleanup False) >> close)
+    x
+addCleanup cleanup (PipeM msrc close) = PipeM
+    (liftM (addCleanup cleanup) msrc)
+    (lift (cleanup False) >> close)
+addCleanup cleanup (NeedInput p c) = NeedInput
+    (addCleanup cleanup . p)
+    (addCleanup cleanup c)
diff --git a/Data/Conduit/List.hs b/Data/Conduit/List.hs
--- a/Data/Conduit/List.hs
+++ b/Data/Conduit/List.hs
@@ -19,6 +19,7 @@
     , drop
     , head
     , zip
+    , zipSinks
     , peek
     , consume
     , sinkNull
@@ -43,13 +44,16 @@
     ( ($), return, (==), (-), Int
     , (.), id, Maybe (..), Monad
     , Bool (..)
+    , Ordering (..)
     , (>>)
     , flip
     , seq
     , otherwise
     )
 import Data.Conduit
+import Data.Conduit.Internal (pipeClose, runFinalize)
 import Data.Monoid (mempty)
+import Data.Void (absurd)
 import Control.Monad (liftM, liftM2)
 
 -- | A strict left fold.
@@ -325,13 +329,45 @@
 -- Since 0.3.0
 zip :: Monad m => Source m a -> Source m b -> Source m (a, b)
 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 (Done _ ()) (HaveOutput _ close _) = PipeM (runFinalize close >> return (Done Nothing ())) close
+zip (HaveOutput _ close _) (Done _ ()) = PipeM (runFinalize close >> return (Done Nothing ())) close
+zip (Done _ ()) (PipeM _ close) = PipeM (runFinalize close >> return (Done Nothing ())) close
+zip (PipeM _ close) (Done _ ()) = PipeM (runFinalize 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
+
+
+-- | Combines two sinks. The new sink will complete when both input sinks have
+--   completed.
+--
+-- If both sinks finish on the same chunk, and both report leftover input,
+-- arbitrarily yield the left sink's leftover input.
+--
+-- Since 0.4.1
+zipSinks :: Monad m => Sink i m r -> Sink i m r' -> Sink i m (r, r')
+zipSinks = zipSinks' EQ
+
+zipSinks' :: Monad m => Ordering -> Sink i m r -> Sink i m r' -> Sink i m (r, r')
+zipSinks' byInputUsed = (><)
+  where
+    PipeM mpx mx     >< py               = PipeM (liftM (>< py) mpx) (liftM2 (,) mx (pipeClose py))
+    px               >< PipeM mpy my     = PipeM (liftM (px ><) mpy) (liftM2 (,) (pipeClose px) my)
+
+    Done ix x        >< Done iy y        = Done i (x, y)
+      where
+        i = case byInputUsed of
+                 EQ -> iy >> ix
+                 GT -> ix
+                 LT -> iy
+
+    NeedInput fpx px >< NeedInput fpy py = NeedInput (\i -> zipSinks' EQ (fpx i) (fpy i)) (px >< py)
+    NeedInput fpx px >< py               = NeedInput (\i -> zipSinks' GT (fpx i) py)      (px >< py)
+    px               >< NeedInput fpy py = NeedInput (\i -> zipSinks' LT px (fpy i))      (px >< py)
+
+    HaveOutput _ _ o >< _                = absurd o
+    _                >< HaveOutput _ _ o = absurd o
+
diff --git a/Data/Conduit/Text.hs b/Data/Conduit/Text.hs
--- a/Data/Conduit/Text.hs
+++ b/Data/Conduit/Text.hs
@@ -21,6 +21,7 @@
     , utf32_be
     , ascii
     , iso8859_1
+    , lines
 
     ) where
 
@@ -43,6 +44,7 @@
 import qualified Data.Conduit as C
 import qualified Data.Conduit.List as CL
 import Control.Monad.Trans.Resource (MonadThrow (..))
+import Control.Monad.Trans.Class (lift)
 
 -- | A specific character encoding.
 --
@@ -63,6 +65,31 @@
     showsPrec d c = showParen (d > 10) $
         showString "Codec " . shows (codecName c)
 
+-- | Emit each line separately
+--
+-- Since 0.4.1
+lines :: Monad m => C.Conduit T.Text m T.Text
+lines =
+    C.conduitState id push close
+  where
+    push front bs' = return $ C.StateProducing leftover ls
+      where
+        bs = front bs'
+        (leftover, ls) = getLines id bs
+
+    getLines front bs
+        | T.null bs = (id, front [])
+        | T.null y = (T.append x, front [])
+        | otherwise = getLines (front . (x:)) (T.drop 1 y)
+      where
+        (x, y) = T.break (== '\n') bs
+
+    close front
+        | T.null bs = return []
+        | otherwise = return [bs]
+      where
+        bs = front T.empty
+
 -- | Convert text into bytes, using the provided codec. If the codec is
 -- not capable of representing an input character, an exception will be thrown.
 --
@@ -96,7 +123,7 @@
             Nothing -> C.Done Nothing ()
             Just (w, _) ->
                 let exc = monadThrow $ DecodeException codec w
-                 in C.PipeM exc exc
+                 in C.PipeM exc (lift exc)
 
     close2 bs =
         case B.uncons bs of
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
@@ -20,7 +20,7 @@
 import Prelude hiding (sequence)
 import Control.Monad.Trans.Resource
 import Data.Conduit.Internal
-import Control.Monad (liftM)
+import Control.Monad (liftM, when)
 
 -- | A helper function for returning a list of values from a @Conduit@.
 --
@@ -30,7 +30,7 @@
          -> [b] -- ^ The values to send down the stream.
          -> Conduit a m b
 haveMore res _ [] = res
-haveMore res close (x:xs) = HaveOutput (haveMore res close xs) close x
+haveMore res close (x:xs) = HaveOutput (haveMore res close xs) (FinalizeM close) x
 
 -- | 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
@@ -108,7 +108,7 @@
         res <- push0 state input
         case res of
             IOProducing output -> return $ haveMore
-                (NeedInput (flip PipeM (release key) . push key state) (close key state))
+                (NeedInput (flip PipeM (FinalizeM $ release key) . push key state) (close key state))
                 (release key >> return ())
                 output
             IOFinished leftover output -> do
@@ -121,7 +121,7 @@
     close key state = PipeM (do
         output <- close0 state
         release key
-        return $ fromList output) (release key)
+        return $ fromList output) (FinalizeM $ release key)
 
 fromList :: Monad m => [a] -> Pipe i a m ()
 fromList [] = Done Nothing ()
@@ -167,15 +167,13 @@
 -- | Specialised version of 'sequenceSink'
 --
 -- Note that this function will return an infinite stream if provided a
--- @SinkNoData@ constructor. In other words, you probably don\'t want to do
+-- @Sink@ which does not consume data. In other words, you probably don\'t want to do
 -- @sequence . return@.
 --
 -- Since 0.3.0
 sequence :: Monad m => Sink input m output -> Conduit input m output
 sequence sink = do
     x <- hasInput
-    if x
-        then do
-            sinkToPipe sink >>= yield
-            sequence sink
-        else return ()
+    when x $ do
+      sinkToPipe sink >>= yield
+      sequence sink
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
@@ -43,7 +43,7 @@
             case res of
                 StateProcessing state' -> return $ NeedInput (push state') (close state')
                 StateDone mleftover output -> return $ Done mleftover output)
-        (close0 state)
+        (FinalizeM $ close0 state)
 
     close = lift . close0
 
@@ -66,7 +66,7 @@
 sinkIO alloc cleanup push0 close0 = NeedInput
     (\input -> PipeM (do
         (key, state) <- allocate alloc cleanup
-        push key state input) (do
+        push key state input) (FinalizeM $ do
             (key, state) <- allocate alloc cleanup
             close key state))
     (do
@@ -82,7 +82,7 @@
             IOProcessing -> return $ NeedInput
                 (\i ->
                     let mpipe = push key state i
-                     in PipeM mpipe (mpipe >>= pipeClose))
+                     in PipeM mpipe (lift mpipe >>= pipeClose))
                 (lift $ close key state)
     close key state = do
         res <- close0 state
diff --git a/conduit.cabal b/conduit.cabal
--- a/conduit.cabal
+++ b/conduit.cabal
@@ -1,5 +1,5 @@
 Name:                conduit
-Version:             0.4.0.1
+Version:             0.4.1
 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>.
diff --git a/test/main.hs b/test/main.hs
--- a/test/main.hs
+++ b/test/main.hs
@@ -127,6 +127,17 @@
             res <- runResourceT $ CL.zip (CL.sourceList [1..10]) (CL.sourceList [11..12]) C.$$ CL.consume
             res @=? zip [1..10 :: Int] [11..12 :: Int]
 
+    describe "zipping sinks" $ do
+        it "take all" $ do
+            res <- runResourceT $ CL.sourceList [1..10] C.$$ CL.zipSinks CL.consume CL.consume
+            res @=? ([1..10 :: Int], [1..10 :: Int])
+        it "take fewer on left" $ do
+            res <- runResourceT $ CL.sourceList [1..10] C.$$ CL.zipSinks (CL.take 4) CL.consume
+            res @=? ([1..4 :: Int], [1..10 :: Int])
+        it "take fewer on right" $ do
+            res <- runResourceT $ CL.sourceList [1..10] C.$$ CL.zipSinks CL.consume (CL.take 4)
+            res @=? ([1..10 :: Int], [1..4 :: Int])
+
     describe "Monad instance for Sink" $ do
         it "binding" $ do
             x <- runResourceT $ CL.sourceList [1..10] C.$$ do
@@ -278,8 +289,8 @@
                              C.$= C.sequence sumSink
                              C.$$ CL.consume
             res @?= [3, 5, 7, 9, 11, 13, 15, 17, 19, 21, 11]
-           
 
+
     describe "sequenceSink" $ do
         it "simple sink" $ do
             let sink () = do
@@ -343,6 +354,20 @@
         go "utf16_be" TLE.encodeUtf16BE CT.utf16_be
         go "utf32_le" TLE.encodeUtf32LE CT.utf32_le
         go "utf32_be" TLE.encodeUtf32BE CT.utf32_be
+
+    describe "text lines" $ do
+        it "works across split lines" $
+            (CL.sourceList [T.pack "abc", T.pack "d\nef"] C.$= CT.lines C.$$ CL.consume) ==
+                [[T.pack "abcd", T.pack "ef"]]
+        it "works with multiple lines in an item" $
+            (CL.sourceList [T.pack "ab\ncd\ne"] C.$= CT.lines C.$$ CL.consume) ==
+                [[T.pack "ab", T.pack "cd", T.pack "e"]]
+        it "works with ending on a newline" $
+            (CL.sourceList [T.pack "ab\n"] C.$= CT.lines C.$$ CL.consume) ==
+                [[T.pack "ab"]]
+        it "works with ending a middle item on a newline" $
+            (CL.sourceList [T.pack "ab\n", T.pack "cd\ne"] C.$= CT.lines C.$$ CL.consume) ==
+                [[T.pack "ab", T.pack "cd", T.pack "e"]]
         it "is not too eager" $ do
             x <- CL.sourceList ["foobarbaz", error "ignore me"] C.$$ CT.decode CT.utf8 C.=$ CL.head
             x @?= Just "foobarbaz"