conduit 0.4.0.1 → 0.4.1

raw patch · 9 files changed

+302/−87 lines, 9 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• Data/Conduit.hs +1/−0
• Data/Conduit/Binary.hs +14/−1
• Data/Conduit/Internal.hs +181/−66
• Data/Conduit/List.hs +40/−4
• Data/Conduit/Text.hs +28/−1
• Data/Conduit/Util/Conduit.hs +8/−10
• Data/Conduit/Util/Sink.hs +3/−3
• conduit.cabal +1/−1
• test/main.hs +26/−1

@@ -40,6 +40,7 @@     , yield     , hasInput     , transPipe+    , mapOutput       -- ** Source     , module Data.Conduit.Util.Source       -- ** Sink

@@ -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.

@@ -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)

@@ -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+

@@ -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

@@ -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

@@ -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

@@ -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>.

@@ -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"