conduit-combinators 0.2.8.3 → 0.3.0

raw patch · 9 files changed

+1473/−397 lines, 9 filesdep +QuickCheckdep +containersdep +directorydep ~basedep ~chunked-datadep ~conduitPVP ok

version bump matches the API change (PVP)

Dependencies added: QuickCheck, containers, directory, mtl, safe

Dependency ranges changed: base, chunked-data, conduit, conduit-extra, mono-traversable

API changes (from Hackage documentation)

Files

• ChangeLog.md +1/−0
• Data/Conduit/Combinators.hs +484/−372
• Data/Conduit/Combinators/Internal.hs +17/−15
• Data/Conduit/Combinators/Stream.hs +451/−0
• Data/Conduit/Combinators/Unqualified.hs +0/−5
• conduit-combinators.cabal +13/−3
• fusion-macros.h +23/−0
• test/Spec.hs +4/−2
• test/StreamSpec.hs +480/−0

@@ -0,0 +1,1 @@+__0.3.0__ Stream fusion enabled, drop compatibility with older conduit

@@ -1,4 +1,5 @@ {-# LANGUAGE CPP #-}+{-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE GADTs #-}@@ -60,6 +61,7 @@     , fold     , foldE     , foldl+    , foldl1     , foldlE     , foldMap     , foldMapE@@ -170,8 +172,10 @@     , lineAscii     , unlines     , unlinesAscii+    , takeExactlyUntilE     , linesUnbounded     , linesUnboundedAscii+    , splitOnUnboundedE        -- * Special     , vectorBuilder@@ -196,9 +200,11 @@ import           Control.Monad.Trans.Class   (lift) import           Control.Monad.Trans.Resource (MonadResource, MonadThrow) import           Data.Conduit+import qualified Data.Conduit.Filesystem as CF import           Data.Conduit.Internal       (ConduitM (..), Pipe (..)) import qualified Data.Conduit.List           as CL import           Data.IOData+import           Data.Maybe                  (isNothing, isJust) import           Data.Monoid                 (Monoid (..)) import           Data.MonoTraversable import qualified Data.Sequences              as Seq@@ -225,6 +231,8 @@ import Data.Text (Text) import qualified System.Random.MWC as MWC import Data.Conduit.Combinators.Internal+import Data.Conduit.Combinators.Stream+import Data.Conduit.Internal.Fusion import qualified System.PosixCompat.Files as PosixC import           Data.Primitive.MutVar       (MutVar, newMutVar, readMutVar,                                               writeMutVar)@@ -233,32 +241,74 @@ import qualified System.Posix.Directory as Dir #endif -#if MIN_VERSION_conduit(1,1,0)-import qualified Data.Conduit.Filesystem as CF-#endif+-- Defines INLINE_RULE0, INLINE_RULE, STREAMING0, and STREAMING.+#include "fusion-macros.h"  -- END IMPORTS +-- TODO:+--+--   * The functions sourceRandom* are based on, initReplicate and+--   initRepeat have specialized versions for when they're used with+--   ($$).  How does this interact with stream fusion?+--+--   * Is it possible to implement fusion for vectorBuilder?  Since it+--   takes a Sink yielding function as an input, the rewrite rule+--   would need to trigger when that parameter looks something like+--   (\x -> unstream (...)).  I don't see anything preventing doing+--   this, but it would be quite a bit of code.++-- NOTE: Fusion isn't possible for the following operations:+--+--   * Due to a lack of leftovers:+--     - dropE, dropWhile, dropWhileE+--     - headE+--     - peek, peekE+--     - null, nullE+--     - takeE, takeWhile, takeWhileE+--     - mapWhile+--     - codeWith+--     - line+--     - lineAscii+--+--   * Due to a use of leftover in a dependency:+--     - Due to "codeWith": encodeBase64, decodeBase64, encodeBase64URL, decodeBase64URL, decodeBase16+--     - due to "CT.decode": decodeUtf8, decodeUtf8Lenient+--+--   * Due to lack of resource cleanup (e.g. bracketP):+--     - sourceDirectory+--     - sourceDirectoryDeep+--     - sourceFile+--+--   * takeExactly / takeExactlyE - no monadic bind.  Another way to+--   look at this is that subsequent streams drive stream evaluation,+--   so there's no way for the conduit to guarantee a certain amount+--   of demand from the upstream.+ -- | Yield each of the values contained by the given @MonoFoldable@. -- -- This will work on many data structures, including lists, @ByteString@s, and @Vector@s. --+-- Subject to fusion+-- -- Since 1.0.0-yieldMany :: (Monad m, MonoFoldable mono)-          => mono-          -> Producer m (Element mono)-yieldMany = ofoldMap yield-{-# INLINE yieldMany #-}+yieldMany, yieldManyC :: (Monad m, MonoFoldable mono)+                      => mono+                      -> Producer m (Element mono)+yieldManyC = ofoldMap yield+{-# INLINE yieldManyC #-}+STREAMING(yieldMany, x)  -- | Generate a producer from a seed value. --+-- Subject to fusion+-- -- Since 1.0.0 unfold :: Monad m        => (b -> Maybe (a, b))        -> b        -> Producer m a-unfold = CL.unfold-{-# INLINE unfold #-}+INLINE_RULE(unfold, f x, CL.unfold f x)  -- | Enumerate from a value to a final value, inclusive, via 'succ'. --@@ -266,88 +316,91 @@ -- combining with @sourceList@ since this avoids any intermediate data -- structures. --+-- Subject to fusion+-- -- Since 1.0.0 enumFromTo :: (Monad m, Enum a, Ord a) => a -> a -> Producer m a-enumFromTo = CL.enumFromTo+INLINE_RULE(enumFromTo, f t, CL.enumFromTo f t)  -- | Produces an infinite stream of repeated applications of f to x. --+-- Subject to fusion+-- -- Since 1.0.0 iterate :: Monad m => (a -> a) -> a -> Producer m a-iterate = CL.iterate-{-# INLINE iterate #-}+INLINE_RULE(iterate, f t, CL.iterate f t)  -- | Produce an infinite stream consisting entirely of the given value. --+-- Subject to fusion+-- -- Since 1.0.0 repeat :: Monad m => a -> Producer m a-repeat = iterate id-{-# INLINE repeat #-}+INLINE_RULE(repeat, x, iterate id x)  -- | Produce a finite stream consisting of n copies of the given value. --+-- Subject to fusion+-- -- Since 1.0.0 replicate :: Monad m           => Int           -> a           -> Producer m a-replicate count0 a =-    loop count0-  where-    loop count = if count <= 0-        then return ()-        else yield a >> loop (count - 1)-{-# INLINE replicate #-}+INLINE_RULE(replicate, n x, CL.replicate n x)  -- | Generate a producer by yielding each of the strict chunks in a @LazySequence@. -- -- For more information, see 'toChunks'. --+-- Subject to fusion+-- -- Since 1.0.0 sourceLazy :: (Monad m, LazySequence lazy strict)            => lazy            -> Producer m strict-sourceLazy = yieldMany . toChunks-{-# INLINE sourceLazy #-}+INLINE_RULE(sourceLazy, x, yieldMany (toChunks x))  -- | Repeatedly run the given action and yield all values it produces. --+-- Subject to fusion+-- -- Since 1.0.0-repeatM :: Monad m-        => m a-        -> Producer m a-repeatM m = forever $ lift m >>= yield-{-# INLINE repeatM #-}+repeatM, repeatMC :: Monad m+                  => m a+                  -> Producer m a+repeatMC m = forever $ lift m >>= yield+{-# INLINE repeatMC #-}+STREAMING(repeatM, m)  -- | Repeatedly run the given action and yield all values it produces, until -- the provided predicate returns @False@. --+-- Subject to fusion+-- -- Since 1.0.0-repeatWhileM :: Monad m-             => m a-             -> (a -> Bool)-             -> Producer m a-repeatWhileM m f =+repeatWhileM, repeatWhileMC :: Monad m+                            => m a+                            -> (a -> Bool)+                            -> Producer m a+repeatWhileMC m f =     loop   where     loop = do         x <- lift m         when (f x) $ yield x >> loop+STREAMING(repeatWhileM, m f)  -- | Perform the given action n times, yielding each result. --+-- Subject to fusion+-- -- Since 1.0.0 replicateM :: Monad m            => Int            -> m a            -> Producer m a-replicateM count0 m =-    loop count0-  where-    loop count = if count <= 0-        then return ()-        else lift m >>= yield >> loop (count - 1)-{-# INLINE replicateM #-}+INLINE_RULE(replicateM, n m, CL.replicateM n m)  -- | Read all data from the given file. --@@ -364,9 +417,11 @@ -- -- Does not close the @Handle@ at any point. --+-- Subject to fusion+-- -- Since 1.0.0-sourceHandle :: (MonadIO m, IOData a) => Handle -> Producer m a-sourceHandle h =+sourceHandle, sourceHandleC :: (MonadIO m, IOData a) => Handle -> Producer m a+sourceHandleC h =     loop   where     loop = do@@ -374,7 +429,8 @@         if onull x             then return ()             else yield x >> loop-{-# INLINEABLE sourceHandle #-}+{-# INLINEABLE sourceHandleC #-}+STREAMING(sourceHandle, h)  -- | Open a @Handle@ using the given function and stream data from it. --@@ -387,48 +443,54 @@  -- | @sourceHandle@ applied to @stdin@. --+-- Subject to fusion+-- -- Since 1.0.0 stdin :: (MonadIO m, IOData a) => Producer m a-stdin = sourceHandle SIO.stdin+INLINE_RULE0(stdin, sourceHandle SIO.stdin)  -- | Create an infinite stream of random values, seeding from the system random -- number. --+-- Subject to fusion+-- -- Since 1.0.0 sourceRandom :: (MWC.Variate a, MonadIO m) => Producer m a-sourceRandom = initRepeat (liftIO MWC.createSystemRandom) (liftIO . MWC.uniform)-{-# INLINE sourceRandom #-}+INLINE_RULE0(sourceRandom, initRepeat (liftIO MWC.createSystemRandom) (liftIO . MWC.uniform))  -- | Create a stream of random values of length n, seeding from the system -- random number. --+-- Subject to fusion+-- -- Since 1.0.0 sourceRandomN :: (MWC.Variate a, MonadIO m)               => Int -- ^ count               -> Producer m a-sourceRandomN = initReplicate (liftIO MWC.createSystemRandom) (liftIO . MWC.uniform)-{-# INLINE [0] sourceRandomN #-}+INLINE_RULE(sourceRandomN, cnt, initReplicate (liftIO MWC.createSystemRandom) (liftIO . MWC.uniform) cnt)  -- | Create an infinite stream of random values, using the given random number -- generator. --+-- Subject to fusion+-- -- Since 1.0.0 sourceRandomGen :: (MWC.Variate a, MonadBase base m, PrimMonad base)                 => MWC.Gen (PrimState base)                 -> Producer m a-sourceRandomGen gen = initRepeat (return gen) (liftBase . MWC.uniform)-{-# INLINE sourceRandomGen #-}+INLINE_RULE(sourceRandomGen, gen, initRepeat (return gen) (liftBase . MWC.uniform))  -- | Create a stream of random values of length n, seeding from the system -- random number. --+-- Subject to fusion+-- -- Since 1.0.0 sourceRandomNGen :: (MWC.Variate a, MonadBase base m, PrimMonad base)                  => MWC.Gen (PrimState base)                  -> Int -- ^ count                  -> Producer m a-sourceRandomNGen gen = initReplicate (return gen) (liftBase . MWC.uniform)-{-# INLINE sourceRandomNGen #-}+INLINE_RULE(sourceRandomNGen, gen cnt, initReplicate (return gen) (liftBase . MWC.uniform) cnt)  -- | Stream the contents of the given directory, without traversing deeply. --@@ -442,26 +504,8 @@ -- -- Since 1.0.0 sourceDirectory :: MonadResource m => FilePath -> Producer m FilePath-#if MIN_VERSION_conduit(1,1,0) sourceDirectory = mapOutput decodeString . CF.sourceDirectory . encodeString-#else -#ifdef WINDOWS-sourceDirectory = (liftIO . F.listDirectory) >=> yieldMany-#else-sourceDirectory dir =-    bracketP (Dir.openDirStream $ encodeString dir) Dir.closeDirStream loop-  where-    loop ds = do-        fp <- liftIO $ Dir.readDirStream ds-        unless (Prelude.null fp) $ do-            unless (fp == "." || fp == "..")-                $ yield $ dir </> decodeString fp-            loop ds-#endif--#endif- -- | Deeply stream the contents of the given directory. -- -- This works the same as @sourceDirectory@, but will not return directories at@@ -473,46 +517,15 @@                     => Bool -- ^ Follow directory symlinks                     -> FilePath -- ^ Root directory                     -> Producer m FilePath-#if MIN_VERSION_conduit(1,1,0) sourceDirectoryDeep follow = mapOutput decodeString . CF.sourceDirectoryDeep follow . encodeString-#else -sourceDirectoryDeep followSymlinks =-    start-  where-    start :: MonadResource m => FilePath -> Producer m FilePath-    start dir = sourceDirectory dir =$= awaitForever go--    go :: MonadResource m => FilePath -> Producer m FilePath-    go fp = do-        isFile' <- liftIO $ F.isFile fp-        if isFile'-            then yield fp-            else do-                follow' <- liftIO $ follow fp-                when follow' (start fp)--    follow :: FilePath -> Prelude.IO Bool-    follow p = do-        let path = encodeString p-        stat <- if followSymlinks-            then PosixC.getFileStatus path-            else PosixC.getSymbolicLinkStatus path-        return (PosixC.isDirectory stat)-#endif- -- | Ignore a certain number of values in the stream. -- -- Since 1.0.0 drop :: Monad m      => Int      -> Consumer a m ()-drop =-    loop-  where-    loop i | i <= 0 = return ()-    loop count = await >>= maybe (return ()) (\_ -> loop (count - 1))-{-# INLINE drop #-}+INLINE_RULE(drop, n, CL.drop n)  -- | Drop a certain number of elements from a chunked stream. --@@ -567,184 +580,241 @@  -- | Monoidally combine all values in the stream. --+-- Subject to fusion+-- -- Since 1.0.0 fold :: (Monad m, Monoid a)      => Consumer a m a-fold = CL.foldMap id-{-# INLINE fold #-}+INLINE_RULE0(fold, CL.foldMap id)  -- | Monoidally combine all elements in the chunked stream. --+-- Subject to fusion+-- -- Since 1.0.0 foldE :: (Monad m, MonoFoldable mono, Monoid (Element mono))       => Consumer mono m (Element mono)-foldE = CL.fold (\accum mono -> accum `mappend` ofoldMap id mono) mempty-{-# INLINE foldE #-}+INLINE_RULE0(foldE, CL.fold (\accum mono -> accum `mappend` ofoldMap id mono) mempty)  -- | A strict left fold. --+-- Subject to fusion+-- -- Since 1.0.0 foldl :: Monad m => (a -> b -> a) -> a -> Consumer b m a-foldl = CL.fold-{-# INLINE foldl #-}+INLINE_RULE(foldl, f x, CL.fold f x)  -- | A strict left fold on a chunked stream. --+-- Subject to fusion+-- -- Since 1.0.0 foldlE :: (Monad m, MonoFoldable mono)        => (a -> Element mono -> a)        -> a        -> Consumer mono m a-foldlE f = CL.fold (ofoldl' f)-{-# INLINE foldlE #-}+INLINE_RULE(foldlE, f x, CL.fold (ofoldlPrime f) x) +-- Work around CPP not supporting identifiers with primes...+ofoldlPrime :: MonoFoldable mono => (a -> Element mono -> a) -> a -> mono -> a+ofoldlPrime = ofoldl'+ -- | Apply the provided mapping function and monoidal combine all values. --+-- Subject to fusion+-- -- Since 1.0.0 foldMap :: (Monad m, Monoid b)         => (a -> b)         -> Consumer a m b-foldMap = CL.foldMap-{-# INLINE foldMap #-}+INLINE_RULE(foldMap, f, CL.foldMap f)  -- | Apply the provided mapping function and monoidal combine all elements of the chunked stream. --+-- Subject to fusion+-- -- Since 1.0.0 foldMapE :: (Monad m, MonoFoldable mono, Monoid w)          => (Element mono -> w)          -> Consumer mono m w-foldMapE = CL.foldMap . ofoldMap-{-# INLINE foldMapE #-}+INLINE_RULE(foldMapE, f, CL.foldMap (ofoldMap f)) +-- | A strict left fold with no starting value.  Returns 'Nothing'+-- when the stream is empty.+--+-- Subject to fusion+foldl1, foldl1C :: Monad m => (a -> a -> a) -> Consumer a m (Maybe a)+foldl1C f =+    await >>= maybe (return Nothing) loop+  where+    loop prev = await >>= maybe (return $ Just prev) (loop . f prev)+STREAMING(foldl1, f)++-- | A strict left fold on a chunked stream, with no starting value.+-- Returns 'Nothing' when the stream is empty.+--+-- Subject to fusion+--+-- Since 1.0.0+foldl1E :: (Monad m, MonoFoldable mono, a ~ Element mono)+        => (a -> a -> a)+        -> Consumer mono m (Maybe a)+INLINE_RULE(foldl1E, f, foldl (foldMaybeNull f) Nothing)++-- Helper for foldl1E+foldMaybeNull :: (MonoFoldable mono, e ~ Element mono)+              => (e -> e -> e)+              -> Maybe e+              -> mono+              -> Maybe e+foldMaybeNull f macc mono =+    case (macc, NonNull.fromNullable mono) of+        (Just acc, Just nn) -> Just $ ofoldl' f acc nn+        (Nothing, Just nn) -> Just $ NonNull.ofoldl1' f nn+        _ -> macc+{-# INLINE foldMaybeNull #-}+ -- | Check that all values in the stream return True. -- -- Subject to shortcut logic: at the first False, consumption of the stream -- will stop. --+-- Subject to fusion+-- -- Since 1.0.0-all :: Monad m-    => (a -> Bool)-    -> Consumer a m Bool-all f =-    loop-  where-    loop = await >>= maybe (return True) go-    go x = if f x then loop else return False-{-# INLINE all #-}+all, allC :: Monad m+          => (a -> Bool)+          -> Consumer a m Bool+allC f = fmap isNothing $ find (Prelude.not . f)+{-# INLINE allC #-}+STREAMING(all, f)  -- | Check that all elements in the chunked stream return True. -- -- Subject to shortcut logic: at the first False, consumption of the stream -- will stop. --+-- Subject to fusion+-- -- Since 1.0.0 allE :: (Monad m, MonoFoldable mono)      => (Element mono -> Bool)      -> Consumer mono m Bool-allE = all . oall+INLINE_RULE(allE, f, all (oall f))  -- | Check that at least one value in the stream returns True. -- -- Subject to shortcut logic: at the first True, consumption of the stream -- will stop. --+-- Subject to fusion+-- -- Since 1.0.0-any :: Monad m-    => (a -> Bool)-    -> Consumer a m Bool-any f =-    loop-  where-    loop = await >>= maybe (return False) go-    go x = if f x then return True else loop-{-# INLINE any #-}+any, anyC :: Monad m+          => (a -> Bool)+          -> Consumer a m Bool+anyC = fmap isJust . find+{-# INLINE anyC #-}+STREAMING(any, f)  -- | Check that at least one element in the chunked stream returns True. -- -- Subject to shortcut logic: at the first True, consumption of the stream -- will stop. --+-- Subject to fusion+-- -- Since 1.0.0 anyE :: (Monad m, MonoFoldable mono)      => (Element mono -> Bool)      -> Consumer mono m Bool-anyE = any . oany+INLINE_RULE(anyE, f, any (oany f))  -- | Are all values in the stream True? -- -- Consumption stops once the first False is encountered. --+-- Subject to fusion+-- -- Since 1.0.0 and :: Monad m => Consumer Bool m Bool-and = all id-{-# INLINE and #-}+INLINE_RULE0(and, all id)  -- | Are all elements in the chunked stream True? -- -- Consumption stops once the first False is encountered. --+-- Subject to fusion+-- -- Since 1.0.0 andE :: (Monad m, MonoFoldable mono, Element mono ~ Bool)      => Consumer mono m Bool-andE = allE id-{-# INLINE andE #-}+INLINE_RULE0(andE, allE id)  -- | Are any values in the stream True? -- -- Consumption stops once the first True is encountered. --+-- Subject to fusion+-- -- Since 1.0.0 or :: Monad m => Consumer Bool m Bool-or = any id-{-# INLINE or #-}+INLINE_RULE0(or, any id)  -- | Are any elements in the chunked stream True? -- -- Consumption stops once the first True is encountered. --+-- Subject to fusion+-- -- Since 1.0.0 orE :: (Monad m, MonoFoldable mono, Element mono ~ Bool)     => Consumer mono m Bool-orE  = anyE id-{-# INLINE orE #-}+INLINE_RULE0(orE, anyE id)  -- | Are any values in the stream equal to the given value? -- -- Stops consuming as soon as a match is found. --+-- Subject to fusion+-- -- Since 1.0.0 elem :: (Monad m, Eq a) => a -> Consumer a m Bool-elem x = any (== x)-{-# INLINE elem #-}+INLINE_RULE(elem, x, any (== x))  -- | Are any elements in the chunked stream equal to the given element? -- -- Stops consuming as soon as a match is found. --+-- Subject to fusion+-- -- Since 1.0.0 elemE :: (Monad m, Seq.EqSequence seq)       => Element seq       -> Consumer seq m Bool-elemE = any . Seq.elem+INLINE_RULE(elemE, f, any (Seq.elem f))  -- | Are no values in the stream equal to the given value? -- -- Stops consuming as soon as a match is found. --+-- Subject to fusion+-- -- Since 1.0.0 notElem :: (Monad m, Eq a) => a -> Consumer a m Bool-notElem x = all (/= x)-{-# INLINE notElem #-}+INLINE_RULE(notElem, x, all (/= x))  -- | Are no elements in the chunked stream equal to the given element? -- -- Stops consuming as soon as a match is found. --+-- Subject to fusion+-- -- Since 1.0.0 notElemE :: (Monad m, Seq.EqSequence seq)          => Element seq          -> Consumer seq m Bool-notElemE = all . Seq.notElem+INLINE_RULE(notElemE, x, all (Seq.notElem x))  -- | Consume all incoming strict chunks into a lazy sequence. -- Note that the entirety of the sequence will be resident at memory.@@ -752,19 +822,23 @@ -- This can be used to consume a stream of strict ByteStrings into a lazy -- ByteString, for example. --+-- Subject to fusion+-- -- Since 1.0.0-sinkLazy :: (Monad m, LazySequence lazy strict)-         => Consumer strict m lazy-sinkLazy = (fromChunks . ($ [])) <$> CL.fold (\front next -> front . (next:)) id-{-# INLINE sinkLazy #-}+sinkLazy, sinkLazyC :: (Monad m, LazySequence lazy strict)+                    => Consumer strict m lazy+sinkLazyC = (fromChunks . ($ [])) <$> CL.fold (\front next -> front . (next:)) id+{-# INLINE sinkLazyC #-}+STREAMING0(sinkLazy)  -- | Consume all values from the stream and return as a list. Note that this -- will pull all values into memory. --+-- Subject to fusion+-- -- Since 1.0.0 sinkList :: Monad m => Consumer a m [a]-sinkList = CL.consume-{-# INLINE sinkList #-}+INLINE_RULE0(sinkList, CL.consume)  -- | Sink incoming values into a vector, growing the vector as necessary to fit -- more elements.@@ -772,10 +846,12 @@ -- Note that using this function is more memory efficient than @sinkList@ and -- then converting to a @Vector@, as it avoids intermediate list constructors. --+-- Subject to fusion+-- -- Since 1.0.0-sinkVector :: (MonadBase base m, V.Vector v a, PrimMonad base)-           => Consumer a m (v a)-sinkVector = do+sinkVector, sinkVectorC :: (MonadBase base m, V.Vector v a, PrimMonad base)+                        => Consumer a m (v a)+sinkVectorC = do     let initSize = 10     mv0 <- liftBase $ VM.new initSize     let go maxSize i mv | i >= maxSize = do@@ -790,7 +866,8 @@                     liftBase $ VM.write mv i x                     go maxSize (i + 1) mv     go initSize 0 mv0-{-# INLINEABLE sinkVector #-}+{-# INLINEABLE sinkVectorC #-}+STREAMING0(sinkVector)  -- | Sink incoming values into a vector, up until size @maxSize@.  Subsequent -- values will be left in the stream. If there are less than @maxSize@ values@@ -799,11 +876,13 @@ -- Note that using this function is more memory efficient than @sinkList@ and -- then converting to a @Vector@, as it avoids intermediate list constructors. --+-- Subject to fusion+-- -- Since 1.0.0-sinkVectorN :: (MonadBase base m, V.Vector v a, PrimMonad base)-            => Int -- ^ maximum allowed size-            -> Consumer a m (v a)-sinkVectorN maxSize = do+sinkVectorN, sinkVectorNC :: (MonadBase base m, V.Vector v a, PrimMonad base)+                          => Int -- ^ maximum allowed size+                          -> Consumer a m (v a)+sinkVectorNC maxSize = do     mv <- liftBase $ VM.new maxSize     let go i | i >= maxSize = liftBase $ V.unsafeFreeze mv         go i = do@@ -814,17 +893,19 @@                     liftBase $ VM.write mv i x                     go (i + 1)     go 0-{-# INLINEABLE sinkVectorN #-}+{-# INLINEABLE sinkVectorNC #-}+STREAMING(sinkVectorN, maxSize)  -- | Convert incoming values to a builder and fold together all builder values. -- -- Defined as: @foldMap toBuilder@. --+-- Subject to fusion+-- -- Since 1.0.0 sinkBuilder :: (Monad m, Monoid builder, ToBuilder a builder)             => Consumer a m builder-sinkBuilder = foldMap toBuilder-{-# INLINE sinkBuilder #-}+INLINE_RULE0(sinkBuilder, foldMap toBuilder)  -- | Same as @sinkBuilder@, but afterwards convert the builder to its lazy -- representation.@@ -837,18 +918,22 @@ -- -- * Some buffer copying may occur in this version. --+-- Subject to fusion+-- -- Since 1.0.0-sinkLazyBuilder :: (Monad m, Monoid builder, ToBuilder a builder, Builder builder lazy)-                => Consumer a m lazy-sinkLazyBuilder = fmap builderToLazy sinkBuilder-{-# INLINE sinkLazyBuilder #-}+sinkLazyBuilder, sinkLazyBuilderC :: (Monad m, Monoid builder, ToBuilder a builder, Builder builder lazy)+                                  => Consumer a m lazy+sinkLazyBuilderC = fmap builderToLazy sinkBuilder+{-# INLINE sinkLazyBuilderC #-}+STREAMING0(sinkLazyBuilder)  -- | Consume and discard all remaining values in the stream. --+-- Subject to fusion+-- -- Since 1.0.0 sinkNull :: Monad m => Consumer a m ()-sinkNull = CL.sinkNull-{-# INLINE sinkNull #-}+INLINE_RULE0(sinkNull, CL.sinkNull)  -- | Same as @await@, but discards any leading 'onull' values. --@@ -903,103 +988,92 @@  -- | Retrieve the last value in the stream, if present. --+-- Subject to fusion+-- -- Since 1.0.0-last :: Monad m => Consumer a m (Maybe a)-last =+last, lastC :: Monad m => Consumer a m (Maybe a)+lastC =     await >>= maybe (return Nothing) loop   where     loop prev = await >>= maybe (return $ Just prev) loop-{-# INLINE last #-}+STREAMING0(last)  -- | Retrieve the last element in the chunked stream, if present. --+-- Subject to fusion+-- -- Since 1.0.0-lastE :: (Monad m, Seq.IsSequence seq) => Consumer seq m (Maybe (Element seq))-lastE =+lastE, lastEC :: (Monad m, Seq.IsSequence seq) => Consumer seq m (Maybe (Element seq))+lastEC =     awaitNonNull >>= maybe (return Nothing) (loop . NonNull.last)   where-     loop prev = awaitNonNull >>= maybe (return $ Just prev) (loop . NonNull.last)-{-# INLINE lastE #-}+STREAMING0(lastE)  -- | Count how many values are in the stream. --+-- Subject to fusion+-- -- Since 1.0.0 length :: (Monad m, Num len) => Consumer a m len-length = foldl (\x _ -> x + 1) 0-{-# INLINE length #-}+INLINE_RULE0(length, foldl (\x _ -> x + 1) 0)  -- | Count how many elements are in the chunked stream. --+-- Subject to fusion+-- -- Since 1.0.0 lengthE :: (Monad m, Num len, MonoFoldable mono) => Consumer mono m len-lengthE = foldl (\x y -> x + fromIntegral (olength y)) 0-{-# INLINE lengthE #-}+INLINE_RULE0(lengthE, foldl (\x y -> x + fromIntegral (olength y)) 0)  -- | Count how many values in the stream pass the given predicate. --+-- Subject to fusion+-- -- Since 1.0.0 lengthIf :: (Monad m, Num len) => (a -> Bool) -> Consumer a m len-lengthIf f = foldl (\cnt a -> if f a then (cnt + 1) else cnt) 0-{-# INLINE lengthIf #-}+INLINE_RULE(lengthIf, f, foldl (\cnt a -> if f a then (cnt + 1) else cnt) 0)  -- | Count how many elements in the chunked stream pass the given predicate. --+-- Subject to fusion+-- -- Since 1.0.0 lengthIfE :: (Monad m, Num len, MonoFoldable mono)           => (Element mono -> Bool) -> Consumer mono m len-lengthIfE f = foldlE (\cnt a -> if f a then (cnt + 1) else cnt) 0-{-# INLINE lengthIfE #-}+INLINE_RULE(lengthIfE, f, foldlE (\cnt a -> if f a then (cnt + 1) else cnt) 0)  -- | Get the largest value in the stream, if present. --+-- Subject to fusion+-- -- Since 1.0.0 maximum :: (Monad m, Ord a) => Consumer a m (Maybe a)-maximum =-    await >>= maybe (return Nothing) loop-  where-    loop prev = await >>= maybe (return $ Just prev) (loop . max prev)-{-# INLINE maximum #-}+INLINE_RULE0(maximum, foldl1 max)  -- | Get the largest element in the chunked stream, if present. --+-- Subject to fusion+-- -- Since 1.0.0 maximumE :: (Monad m, Seq.OrdSequence seq) => Consumer seq m (Maybe (Element seq))-maximumE =-    start-  where-    start = await >>= maybe (return Nothing) start'-    start' x =-        case NonNull.fromNullable x of-            Nothing -> start-            Just y -> loop $ NonNull.maximum y-    loop prev = await >>= maybe (return $ Just prev) (loop . ofoldl' max prev)-{-# INLINE maximumE #-}+INLINE_RULE0(maximumE, foldl1E max)  -- | Get the smallest value in the stream, if present. --+-- Subject to fusion+-- -- Since 1.0.0 minimum :: (Monad m, Ord a) => Consumer a m (Maybe a)-minimum =-    await >>= maybe (return Nothing) loop-  where-    loop prev = await >>= maybe (return $ Just prev) (loop . min prev)-{-# INLINE minimum #-}+INLINE_RULE0(minimum, foldl1 min)  -- | Get the smallest element in the chunked stream, if present. --+-- Subject to fusion+-- -- Since 1.0.0 minimumE :: (Monad m, Seq.OrdSequence seq) => Consumer seq m (Maybe (Element seq))-minimumE =-    start-  where-    start = await >>= maybe (return Nothing) start'-    start' x =-        case NonNull.fromNullable x of-            Nothing -> start-            Just y -> loop $ NonNull.minimum y-    loop prev = await >>= maybe (return $ Just prev) (loop . ofoldl' min prev)-{-# INLINE minimumE #-}+INLINE_RULE0(minimumE, foldl1E min)  -- | True if there are no values in the stream. --@@ -1027,92 +1101,104 @@  -- | Get the sum of all values in the stream. --+-- Subject to fusion+-- -- Since 1.0.0 sum :: (Monad m, Num a) => Consumer a m a-sum = foldl (+) 0-{-# INLINE sum #-}+INLINE_RULE0(sum, foldl (+) 0)  -- | Get the sum of all elements in the chunked stream. --+-- Subject to fusion+-- -- Since 1.0.0 sumE :: (Monad m, MonoFoldable mono, Num (Element mono)) => Consumer mono m (Element mono)-sumE = foldlE (+) 0-{-# INLINE sumE #-}+INLINE_RULE0(sumE, foldlE (+) 0)  -- | Get the product of all values in the stream. --+-- Subject to fusion+-- -- Since 1.0.0 product :: (Monad m, Num a) => Consumer a m a-product = foldl (*) 1-{-# INLINE product #-}+INLINE_RULE0(product, foldl (*) 1)  -- | Get the product of all elements in the chunked stream. --+-- Subject to fusion+-- -- Since 1.0.0 productE :: (Monad m, MonoFoldable mono, Num (Element mono)) => Consumer mono m (Element mono)-productE = foldlE (*) 1-{-# INLINE productE #-}+INLINE_RULE0(productE, foldlE (*) 1)  -- | Find the first matching value. --+-- Subject to fusion+-- -- Since 1.0.0-find :: Monad m => (a -> Bool) -> Consumer a m (Maybe a)-find f =+find, findC :: Monad m => (a -> Bool) -> Consumer a m (Maybe a)+findC f =     loop   where     loop = await >>= maybe (return Nothing) go     go x = if f x then return (Just x) else loop+{-# INLINE findC #-}+STREAMING(find, f)  -- | Apply the action to all values in the stream. --+-- Subject to fusion+-- -- Since 1.0.0 mapM_ :: Monad m => (a -> m ()) -> Consumer a m ()-mapM_ = CL.mapM_-{-# INLINE mapM_ #-}+INLINE_RULE(mapM_, f, CL.mapM_ f)  -- | Apply the action to all elements in the chunked stream. --+-- Subject to fusion+-- -- Since 1.0.0 mapM_E :: (Monad m, MonoFoldable mono) => (Element mono -> m ()) -> Consumer mono m ()-mapM_E = CL.mapM_ . omapM_-{-# INLINE mapM_E #-}+INLINE_RULE(mapM_E, f, CL.mapM_ (omapM_ f))  -- | A monadic strict left fold. --+-- Subject to fusion+-- -- Since 1.0.0 foldM :: Monad m => (a -> b -> m a) -> a -> Consumer b m a-foldM = CL.foldM-{-# INLINE foldM #-}+INLINE_RULE(foldM, f x, CL.foldM f x)  -- | A monadic strict left fold on a chunked stream. --+-- Subject to fusion+-- -- Since 1.0.0 foldME :: (Monad m, MonoFoldable mono)        => (a -> Element mono -> m a)        -> a        -> Consumer mono m a-foldME f = foldM (ofoldlM f)-{-# INLINE foldME #-}+INLINE_RULE(foldME, f x, foldM (ofoldlM f) x)  -- | Apply the provided monadic mapping function and monoidal combine all values. --+-- Subject to fusion+-- -- Since 1.0.0 foldMapM :: (Monad m, Monoid w) => (a -> m w) -> Consumer a m w-foldMapM = CL.foldMapM-{-# INLINE foldMapM #-}+INLINE_RULE(foldMapM, f, CL.foldMapM f)  -- | Apply the provided monadic mapping function and monoidal combine all -- elements in the chunked stream. --+-- Subject to fusion+-- -- Since 1.0.0 foldMapME :: (Monad m, MonoFoldable mono, Monoid w)           => (Element mono -> m w)           -> Consumer mono m w-foldMapME f =-    CL.foldM go mempty-  where-    go = ofoldlM (\accum e -> mappend accum `liftM` f e)-{-# INLINE foldMapME #-}+INLINE_RULE(foldMapME, f,+    CL.foldM (ofoldlM (\accum e -> mappend accum `liftM` f e)) mempty)  -- | Write all data to the given file. --@@ -1127,30 +1213,37 @@  -- | Print all incoming values to stdout. --+-- Subject to fusion+-- -- Since 1.0.0 print :: (Show a, MonadIO m) => Consumer a m ()-print = mapM_ (liftIO . Prelude.print)+INLINE_RULE0(print, mapM_ (liftIO . Prelude.print))  -- | @sinkHandle@ applied to @stdout@. --+-- Subject to fusion+-- -- Since 1.0.0 stdout :: (MonadIO m, IOData a) => Consumer a m ()-stdout = sinkHandle SIO.stdout+INLINE_RULE0(stdout, sinkHandle SIO.stdout)  -- | @sinkHandle@ applied to @stderr@. --+-- Subject to fusion+-- -- Since 1.0.0 stderr :: (MonadIO m, IOData a) => Consumer a m ()-stderr = sinkHandle SIO.stderr+INLINE_RULE0(stderr, sinkHandle SIO.stderr)  -- | Write all data to the given @Handle@. -- -- Does not close the @Handle@ at any point. --+-- Subject to fusion+-- -- Since 1.0.0 sinkHandle :: (MonadIO m, IOData a) => Handle -> Consumer a m ()-sinkHandle = CL.mapM_ . hPut-{-# INLINE sinkHandle #-}+INLINE_RULE(sinkHandle, h, CL.mapM_ (hPut h))  -- | Open a @Handle@ using the given function and stream data to it. --@@ -1163,27 +1256,30 @@  -- | Apply a transformation to all values in a stream. --+-- Subject to fusion+-- -- Since 1.0.0 map :: Monad m => (a -> b) -> Conduit a m b-map = CL.map-{-# INLINE map #-}+INLINE_RULE(map, f, CL.map f)  -- | Apply a transformation to all elements in a chunked stream. --+-- Subject to fusion+-- -- Since 1.0.0 mapE :: (Monad m, Functor f) => (a -> b) -> Conduit (f a) m (f b)-mapE = CL.map . fmap-{-# INLINE mapE #-}+INLINE_RULE(mapE, f, CL.map (fmap f))  -- | Apply a monomorphic transformation to all elements in a chunked stream. -- -- Unlike @mapE@, this will work on types like @ByteString@ and @Text@ which -- are @MonoFunctor@ but not @Functor@. --+-- Subject to fusion+-- -- Since 1.0.0 omapE :: (Monad m, MonoFunctor mono) => (Element mono -> Element mono) -> Conduit mono m mono-omapE = CL.map . omap-{-# INLINE omapE #-}+INLINE_RULE(omapE, f, CL.map (omap f))  -- | Apply the function to each value in the stream, resulting in a foldable -- value (e.g., a list). Then yield each of the individual values in that@@ -1191,12 +1287,15 @@ -- -- Generalizes concatMap, mapMaybe, and mapFoldable. --+-- Subject to fusion+-- -- Since 1.0.0-concatMap :: (Monad m, MonoFoldable mono)-          => (a -> mono)-          -> Conduit a m (Element mono)-concatMap f = awaitForever (yieldMany . f)-{-# INLINE concatMap #-}+concatMap, concatMapC :: (Monad m, MonoFoldable mono)+                      => (a -> mono)+                      -> Conduit a m (Element mono)+concatMapC f = awaitForever (yieldMany . f)+{-# INLINE concatMapC #-}+STREAMING(concatMap, f)  -- | Apply the function to each element in the chunked stream, resulting in a -- foldable value (e.g., a list). Then yield each of the individual values in@@ -1204,12 +1303,13 @@ -- -- Generalizes concatMap, mapMaybe, and mapFoldable. --+-- Subject to fusion+-- -- Since 1.0.0 concatMapE :: (Monad m, MonoFoldable mono, Monoid w)            => (Element mono -> w)            -> Conduit mono m w-concatMapE = CL.map . ofoldMap-{-# INLINE concatMapE #-}+INLINE_RULE(concatMapE, f, CL.map (ofoldMap f))  -- | Stream up to n number of values downstream. --@@ -1217,15 +1317,11 @@ -- If you want to force /exactly/ the given number of values to be consumed, -- see 'takeExactly'. --+-- Subject to fusion+-- -- Since 1.0.0 take :: Monad m => Int -> Conduit a m a-take =-    loop-  where-    loop count = if count <= 0-        then return ()-        else await >>= maybe (return ()) (\i -> yield i >> loop (count - 1))-{-# INLINE take #-}+INLINE_RULE(take, n, CL.isolate n)  -- | Stream up to n number of elements downstream in a chunked stream. --@@ -1313,7 +1409,6 @@     r <- inner     CL.sinkNull     return r-{-# INLINE takeExactly #-}  -- | Same as 'takeExactly', but for chunked streams. --@@ -1331,25 +1426,29 @@ -- | Flatten out a stream by yielding the values contained in an incoming -- @MonoFoldable@ as individually yielded values. --+-- Subject to fusion+-- -- Since 1.0.0-concat :: (Monad m, MonoFoldable mono)-       => Conduit mono m (Element mono)-concat = awaitForever yieldMany-{-# INLINE concat #-}+concat, concatC :: (Monad m, MonoFoldable mono)+                => Conduit mono m (Element mono)+concatC = awaitForever yieldMany+STREAMING0(concat)  -- | Keep only values in the stream passing a given predicate. --+-- Subject to fusion+-- -- Since 1.0.0 filter :: Monad m => (a -> Bool) -> Conduit a m a-filter = CL.filter-{-# INLINE filter #-}+INLINE_RULE(filter, f, CL.filter f)  -- | Keep only elements in the chunked stream passing a given predicate. --+-- Subject to fusion+-- -- Since 1.0.0 filterE :: (Seq.IsSequence seq, Monad m) => (Element seq -> Bool) -> Conduit seq m seq-filterE = CL.map . Seq.filter-{-# INLINE filterE #-}+INLINE_RULE(filterE, f, CL.map (Seq.filter f))  -- | Map values as long as the result is @Just@. --@@ -1385,9 +1484,11 @@  -- | Analog of 'Prelude.scanl' for lists. --+-- Subject to fusion+-- -- Since 1.0.6-scanl :: Monad m => (a -> b -> a) -> a -> Conduit b m a-scanl f =+scanl, scanlC :: Monad m => (a -> b -> a) -> a -> Conduit b m a+scanlC f =     loop   where     loop seed =@@ -1397,24 +1498,27 @@             let seed' = f seed b             seed' `seq` yield seed             loop seed'-{-# INLINE scanl #-}+STREAMING(scanl, f x)  -- | 'concatMap' with an accumulator. --+-- Subject to fusion+-- -- Since 1.0.0 concatMapAccum :: Monad m => (a -> accum -> (accum, [b])) -> accum -> Conduit a m b-concatMapAccum = CL.concatMapAccum-{-# INLINE concatMapAccum #-}+INLINE_RULE0(concatMapAccum, CL.concatMapAccum)  -- | Insert the given value between each two values in the stream. --+-- Subject to fusion+-- -- Since 1.0.0-intersperse :: Monad m => a -> Conduit a m a-intersperse x =+intersperse, intersperseC :: Monad m => a -> Conduit a m a+intersperseC x =     await >>= omapM_ go   where     go y = yield y >> concatMap (\z -> [x, z])-{-# INLINE intersperse #-}+STREAMING(intersperse, x)  -- | Sliding window of values -- 1,2,3,4,5 with window size 2 gives@@ -1422,9 +1526,11 @@ -- -- Best used with structures that support O(1) snoc. --+-- Subject to fusion+-- -- Since 1.0.0-slidingWindow :: (Monad m, Seq.IsSequence seq, Element seq ~ a) => Int -> Conduit a m seq-slidingWindow sz = go (if sz <= 0 then 1 else sz) mempty+slidingWindow, slidingWindowC :: (Monad m, Seq.IsSequence seq, Element seq ~ a) => Int -> Conduit a m seq+slidingWindowC sz = go (max 1 sz) mempty     where goContinue st = await >>=                           maybe (return ())                                 (\x -> do@@ -1436,6 +1542,7 @@                      case m of                        Nothing -> yield st                        Just x -> go (n-1) (Seq.snoc st x)+STREAMING(slidingWindow, sz)  codeWith :: Monad m          => Int@@ -1512,10 +1619,11 @@  -- | Apply base16-encoding to the stream. --+-- Subject to fusion+-- -- Since 1.0.0 encodeBase16 :: Monad m => Conduit ByteString m ByteString-encodeBase16 = map B16.encode-{-# INLINE encodeBase16 #-}+INLINE_RULE0(encodeBase16, map B16.encode)  -- | Apply base16-decoding to the stream. Will stop decoding on the first -- invalid chunk.@@ -1537,29 +1645,32 @@ -- If you do not need the transformed values, and instead just want the monadic -- side-effects of running the action, see 'mapM_'. --+-- Subject to fusion+-- -- Since 1.0.0 mapM :: Monad m => (a -> m b) -> Conduit a m b-mapM = CL.mapM-{-# INLINE mapM #-}+INLINE_RULE(mapM, f, CL.mapM f)  -- | Apply a monadic transformation to all elements in a chunked stream. --+-- Subject to fusion+-- -- Since 1.0.0 mapME :: (Monad m, Data.Traversable.Traversable f) => (a -> m b) -> Conduit (f a) m (f b)-mapME = CL.mapM . Data.Traversable.mapM-{-# INLINE mapME #-}+INLINE_RULE(mapME, f, CL.mapM (Data.Traversable.mapM f))  -- | Apply a monadic monomorphic transformation to all elements in a chunked stream. -- -- Unlike @mapME@, this will work on types like @ByteString@ and @Text@ which -- are @MonoFunctor@ but not @Functor@. --+-- Subject to fusion+-- -- Since 1.0.0 omapME :: (Monad m, MonoTraversable mono)        => (Element mono -> m (Element mono))        -> Conduit mono m mono-omapME = CL.mapM . omapM-{-# INLINE omapME #-}+INLINE_RULE(omapME, f, CL.mapM (omapM f))  -- | Apply the monadic function to each value in the stream, resulting in a -- foldable value (e.g., a list). Then yield each of the individual values in@@ -1567,33 +1678,38 @@ -- -- Generalizes concatMapM, mapMaybeM, and mapFoldableM. --+-- Subject to fusion+-- -- Since 1.0.0-concatMapM :: (Monad m, MonoFoldable mono)-           => (a -> m mono)-           -> Conduit a m (Element mono)-concatMapM f = awaitForever (lift . f >=> yieldMany)-{-# INLINE concatMapM #-}+concatMapM, concatMapMC :: (Monad m, MonoFoldable mono)+                        => (a -> m mono)+                        -> Conduit a m (Element mono)+concatMapMC f = awaitForever (lift . f >=> yieldMany)+STREAMING(concatMapM, f)  -- | Keep only values in the stream passing a given monadic predicate. --+-- Subject to fusion+-- -- Since 1.0.0-filterM :: Monad m-        => (a -> m Bool)-        -> Conduit a m a-filterM f =+filterM, filterMC :: Monad m+                  => (a -> m Bool)+                  -> Conduit a m a+filterMC f =     awaitForever go   where     go x = do         b <- lift $ f x         when b $ yield x-{-# INLINE filterM #-}+STREAMING(filterM, f)  -- | Keep only elements in the chunked stream passing a given monadic predicate. --+-- Subject to fusion+-- -- Since 1.0.0 filterME :: (Monad m, Seq.IsSequence seq) => (Element seq -> m Bool) -> Conduit seq m seq-filterME = CL.mapM . Seq.filterM-{-# INLINE filterME #-}+INLINE_RULE(filterME, f, CL.mapM (Seq.filterM f))  -- | Apply a monadic action on all values in a stream. --@@ -1602,15 +1718,19 @@ -- -- > iterM f = mapM (\a -> f a >>= \() -> return a) --+-- Subject to fusion+-- -- Since 1.0.0 iterM :: Monad m => (a -> m ()) -> Conduit a m a-iterM = CL.iterM+INLINE_RULE(iterM, f, CL.iterM f)  -- | Analog of 'Prelude.scanl' for lists, monadic. --+-- Subject to fusion+-- -- Since 1.0.6-scanlM :: Monad m => (a -> b -> m a) -> a -> Conduit b m a-scanlM f =+scanlM, scanlMC :: Monad m => (a -> b -> m a) -> a -> Conduit b m a+scanlMC f =     loop   where     loop seed =@@ -1620,20 +1740,23 @@             seed' <- lift $ f seed b             seed' `seq` yield seed             loop seed'-{-# INLINE scanlM #-}+STREAMING(scanlM, f x)  -- | 'concatMapM' with an accumulator. --+-- Subject to fusion+-- -- Since 1.0.0 concatMapAccumM :: Monad m => (a -> accum -> m (accum, [b])) -> accum -> Conduit a m b-concatMapAccumM = CL.concatMapAccumM-{-# INLINE concatMapAccumM #-}+INLINE_RULE(concatMapAccumM, f x, CL.concatMapAccumM f x)  -- | Encode a stream of text as UTF8. --+-- Subject to fusion+-- -- Since 1.0.0 encodeUtf8 :: (Monad m, DTE.Utf8 text binary) => Conduit text m binary-encodeUtf8 = map DTE.encodeUtf8+INLINE_RULE0(encodeUtf8, map DTE.encodeUtf8)  -- | Decode a stream of binary data as UTF8. --@@ -1657,22 +1780,7 @@ line :: (Monad m, Seq.IsSequence seq, Element seq ~ Char)      => ConduitM seq o m r      -> ConduitM seq o m r-line inner = do-    loop =$= do-        x <- inner-        sinkNull-        return x-  where-    loop = await >>= omapM_ go-    go t =-        if onull y-            then yield x >> loop-            else do-                unless (onull x) $ yield x-                let y' = Seq.drop 1 y-                unless (onull y') $ leftover y'-      where-        (x, y) = Seq.break (== '\n') t+line = takeExactlyUntilE (== '\n') {-# INLINE line #-}  -- | Same as 'line', but operates on ASCII/binary data.@@ -1681,7 +1789,18 @@ lineAscii :: (Monad m, Seq.IsSequence seq, Element seq ~ Word8)           => ConduitM seq o m r           -> ConduitM seq o m r-lineAscii inner =+lineAscii = takeExactlyUntilE (== 10)+{-# INLINE lineAscii #-}++-- | Stream in the chunked input until an element matches a predicate.+--+-- Like @takeExactly@, this will consume the entirety of the prefix+-- regardless of the behavior of the inner Conduit.+takeExactlyUntilE :: (Monad m, Seq.IsSequence seq)+                  => (Element seq -> Bool)+                  -> ConduitM seq o m r+                  -> ConduitM seq o m r+takeExactlyUntilE f inner =     loop =$= do         x <- inner         sinkNull@@ -1696,32 +1815,35 @@                 let y' = Seq.drop 1 y                 unless (onull y') $ leftover y'       where-        (x, y) = Seq.break (== 10) t-{-# INLINE lineAscii #-}+        (x, y) = Seq.break f t+{-# INLINE takeExactlyUntilE #-}  -- | Insert a newline character after each incoming chunk of data. --+-- Subject to fusion+-- -- Since 1.0.0 unlines :: (Monad m, Seq.IsSequence seq, Element seq ~ Char) => Conduit seq m seq-unlines = concatMap (:[Seq.singleton '\n'])-{-# INLINE unlines #-}+INLINE_RULE0(unlines, concatMap (:[Seq.singleton '\n']))  -- | Same as 'unlines', but operates on ASCII/binary data. --+-- Subject to fusion+-- -- Since 1.0.0 unlinesAscii :: (Monad m, Seq.IsSequence seq, Element seq ~ Word8) => Conduit seq m seq-unlinesAscii = concatMap (:[Seq.singleton 10])-{-# INLINE unlinesAscii #-}+INLINE_RULE0(unlinesAscii, concatMap (:[Seq.singleton 10])) --- | Convert a stream of arbitrarily-chunked textual data into a stream of data--- where each chunk represents a single line. Note that, if you have--- unknown/untrusted input, this function is /unsafe/, since it would allow an--- attacker to form lines of massive length and exhaust memory.------ Since 1.0.0-linesUnbounded :: (Monad m, Seq.IsSequence seq, Element seq ~ Char)-               => Conduit seq m seq-linesUnbounded =+-- | Split a stream of arbitrarily-chunked data, based on a predicate+-- on elements.  Elements that satisfy the predicate will cause chunks+-- to be split, and aren't included in these output chunks.  Note+-- that, if you have unknown/untrusted input, this function is+-- /unsafe/, since it would allow an attacker to form chunks of+-- massive length and exhaust memory.+splitOnUnboundedE, splitOnUnboundedEC+    :: (Monad m, Seq.IsSequence seq)+    => (Element seq -> Bool) -> Conduit seq m seq+splitOnUnboundedEC f =     start   where     start = await >>= maybe (return ()) loop@@ -1735,28 +1857,29 @@                     Just t' -> loop (t `mappend` t')             else yield x >> loop (Seq.drop 1 y)       where-        (x, y) = Seq.break (== '\n') t+        (x, y) = Seq.break f t+STREAMING(splitOnUnboundedE, f) +-- | Convert a stream of arbitrarily-chunked textual data into a stream of data+-- where each chunk represents a single line. Note that, if you have+-- unknown/untrusted input, this function is /unsafe/, since it would allow an+-- attacker to form lines of massive length and exhaust memory.+--+-- Subject to fusion+--+-- Since 1.0.0+linesUnbounded :: (Monad m, Seq.IsSequence seq, Element seq ~ Char)+               => Conduit seq m seq+INLINE_RULE0(linesUnbounded, splitOnUnboundedE (== '\n'))+ -- | Same as 'linesUnbounded', but for ASCII/binary data. --+-- Subject to fusion+-- -- Since 1.0.0 linesUnboundedAscii :: (Monad m, Seq.IsSequence seq, Element seq ~ Word8)                     => Conduit seq m seq-linesUnboundedAscii =-    start-  where-    start = await >>= maybe (return ()) loop--    loop t =-        if onull y-            then do-                mt <- await-                case mt of-                    Nothing -> unless (onull t) $ yield t-                    Just t' -> loop (t `mappend` t')-            else yield x >> loop (Seq.drop 1 y)-      where-        (x, y) = Seq.break (== 10) t+INLINE_RULE0(linesUnboundedAscii, splitOnUnboundedE (== 10))  -- | Generally speaking, yielding values from inside a Conduit requires -- some allocation for constructors. This can introduce an overhead,@@ -1807,7 +1930,6 @@         => ConduitM i o m ()         -> Sink i m r         -> ConduitM i o m r-#if MIN_VERSION_conduit(1, 2, 0) onAwait (ConduitM callback) (ConduitM sink0) = ConduitM $ \rest -> let     go (Done r) = rest r     go (HaveOutput _ _ o) = absurd o@@ -1815,16 +1937,6 @@     go (PipeM mp) = PipeM (liftM go mp)     go (Leftover f i) = Leftover (go f) i     in go (sink0 Done)-#else-onAwait (ConduitM callback) =-    ConduitM . go . unConduitM-  where-    go (Done r) = Done r-    go (HaveOutput _ _ o) = absurd o-    go (NeedInput f g) = callback >> NeedInput (go . f) (go . g)-    go (PipeM mp) = PipeM (liftM go mp)-    go (Leftover f i) = Leftover (go f) i-#endif {-# INLINE onAwait #-}  yieldS :: (PrimMonad base, MonadBase base m)

@@ -13,16 +13,24 @@ import Data.Void (absurd) import Control.Monad.Trans.Class (lift) import Control.Monad (replicateM_, forever)+import Data.Conduit.Combinators.Stream+import Data.Conduit.Internal.Fusion +-- Defines INLINE_RULE0, INLINE_RULE, STREAMING0, and STREAMING.+#include "fusion-macros.h"+ -- | Acquire the seed value and perform the given action with it n times, -- yielding each result. --+-- Subject to fusion+-- -- Since 0.2.1-initReplicate :: Monad m => m seed -> (seed -> m a) -> Int -> Producer m a-initReplicate mseed f cnt = do+initReplicate, initReplicateC :: Monad m => m seed -> (seed -> m a) -> Int -> Producer m a+initReplicateC mseed f cnt = do     seed <- lift mseed     replicateM_ cnt (lift (f seed) >>= yield)-{-# INLINE [1] initReplicate #-}+{-# INLINE [1] initReplicateC #-}+STREAMING(initReplicate, mseed f cnt)  -- | Optimized version of initReplicate for the special case of connecting with -- a @Sink@.@@ -42,12 +50,7 @@         loop _ (HaveOutput _ _ o) = absurd o         loop cnt (PipeM mp) = mp >>= loop cnt         loop _ (Leftover _ i) = absurd i--#if MIN_VERSION_conduit(1, 2, 0)     loop cnt0 (injectLeftovers $ sink0 Done)-#else-    loop cnt0 (injectLeftovers sink0)-#endif   where     finish (Done r) = return r     finish (HaveOutput _ _ o) = absurd o@@ -62,11 +65,15 @@ -- | Acquire the seed value and perform the given action with it forever, -- yielding each result. --+-- Subject to fusion+-- -- Since 0.2.1-initRepeat :: Monad m => m seed -> (seed -> m a) -> Producer m a-initRepeat mseed f = do+initRepeat, initRepeatC :: Monad m => m seed -> (seed -> m a) -> Producer m a+initRepeatC mseed f = do     seed <- lift mseed     forever $ lift (f seed) >>= yield+{-# INLINE [1] initRepeatC #-}+STREAMING(initRepeat, mseed f)  -- | Optimized version of initRepeat for the special case of connecting with -- a @Sink@.@@ -84,12 +91,7 @@         loop (HaveOutput _ _ o) = absurd o         loop (PipeM mp) = mp >>= loop         loop (Leftover _ i) = absurd i--#if MIN_VERSION_conduit(1, 2, 0)     loop (injectLeftovers (sink0 Done))-#else-    loop (injectLeftovers sink0)-#endif {-# RULES "initRepeatConnect" forall mseed f sink.     initRepeat mseed f $$ sink     = initRepeatConnect mseed f sink

@@ -0,0 +1,451 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE ViewPatterns #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE TypeFamilies #-}+-- | These are stream fusion versions of some of the functions in+-- "Data.Conduit.Combinators".  Many functions don't have stream+-- versions here because instead they have @RULES@ which inline a+-- definition that fuses.+module Data.Conduit.Combinators.Stream+  ( yieldManyS+  , repeatMS+  , repeatWhileMS+  , sourceHandleS+  , foldl1S+  , allS+  , anyS+  , sinkLazyS+  , sinkVectorS+  , sinkVectorNS+  , sinkLazyBuilderS+  , lastS+  , lastES+  , findS+  , concatMapS+  , concatMapMS+  , concatS+  , scanlS+  , scanlMS+  , intersperseS+  , slidingWindowS+  , filterMS+  , splitOnUnboundedES+  , initReplicateS+  , initRepeatS+  )+  where++-- BEGIN IMPORTS++import           Control.Monad (liftM)+import           Control.Monad.Base (MonadBase (liftBase))+import           Control.Monad.IO.Class (MonadIO (..))+import           Control.Monad.Primitive (PrimMonad)+import           Data.Builder+import           Data.Conduit.Internal.Fusion+import           Data.Conduit.Internal.List.Stream (foldS)+import           Data.IOData+import           Data.Maybe (isNothing, isJust)+import           Data.MonoTraversable+import           Data.Monoid (Monoid (..))+import qualified Data.NonNull as NonNull+import qualified Data.Sequences as Seq+import           Data.Sequences.Lazy+import qualified Data.Vector.Generic as V+import qualified Data.Vector.Generic.Mutable as VM+import           Prelude+import           System.IO (Handle)++-- END IMPORTS++yieldManyS :: (Monad m, MonoFoldable mono)+            => mono+            -> StreamProducer m (Element mono)+yieldManyS mono _ =+    Stream (return . step) (return (otoList mono))+  where+    step [] = Stop ()+    step (x:xs) = Emit xs x+{-# INLINE yieldManyS #-}++repeatMS :: Monad m+         => m a+         -> StreamProducer m a+repeatMS m _ =+    Stream step (return ())+  where+    step _ = liftM (Emit ()) m+{-# INLINE repeatMS #-}++repeatWhileMS :: Monad m+              => m a+              -> (a -> Bool)+              -> StreamProducer m a+repeatWhileMS m f _ =+    Stream step (return ())+  where+    step _ = do+        x <- m+        return $ if f x+            then Emit () x+            else Stop ()+{-# INLINE repeatWhileMS #-}++sourceHandleS :: (MonadIO m, IOData a) => Handle -> StreamProducer m a+sourceHandleS h _ =+    Stream step (return ())+  where+    step () = do+        x <- liftIO (hGetChunk h)+        return $ if onull x+            then Stop ()+            else Emit () x+{-# INLINE sourceHandleS #-}++foldl1S :: Monad m+        => (a -> a -> a)+        -> StreamConsumer a m (Maybe a)+foldl1S f (Stream step ms0) =+    Stream step' (liftM (Nothing, ) ms0)+  where+    step' (mprev, s) = do+        res <- step s+        return $ case res of+            Stop () -> Stop mprev+            Skip s' -> Skip (mprev, s')+            Emit s' a -> Skip (Just $ maybe a (`f` a) mprev, s')+{-# INLINE foldl1S #-}++allS :: Monad m+     => (a -> Bool)+     -> StreamConsumer a m Bool+allS f = fmapS isNothing (findS (Prelude.not . f))+{-# INLINE allS #-}++anyS :: Monad m+     => (a -> Bool)+     -> StreamConsumer a m Bool+anyS f = fmapS isJust (findS f)+{-# INLINE anyS #-}++--TODO: use a definition like+-- fmapS (fromChunks . ($ [])) <$> CL.fold (\front next -> front . (next:)) id++sinkLazyS :: (Monad m, LazySequence lazy strict)+          => StreamConsumer strict m lazy+sinkLazyS = fmapS (fromChunks . ($ [])) $ foldS (\front next -> front . (next:)) id+{-# INLINE sinkLazyS #-}++sinkVectorS :: (MonadBase base m, V.Vector v a, PrimMonad base)+            => StreamConsumer a m (v a)+sinkVectorS (Stream step ms0) = do+    Stream step' $ do+        s0 <- ms0+        mv0 <- liftBase $ VM.new initSize+        return (initSize, 0, mv0, s0)+  where+    initSize = 10+    step' (maxSize, i, mv, s) = do+        res <- step s+        case res of+            Stop () -> liftM (Stop . V.slice 0 i) $ liftBase (V.unsafeFreeze mv)+            Skip s' -> return $ Skip (maxSize, i, mv, s')+            Emit s' x -> do+                liftBase $ VM.write mv i x+                let i' = i + 1+                if i' >= maxSize+                    then do+                        let newMax = maxSize * 2+                        mv' <- liftBase $ VM.grow mv maxSize+                        return $ Skip (newMax, i', mv', s')+                    else return $ Skip (maxSize, i', mv, s')+{-# INLINE sinkVectorS #-}++sinkVectorNS :: (MonadBase base m, V.Vector v a, PrimMonad base)+             => Int -- ^ maximum allowed size+             -> StreamConsumer a m (v a)+sinkVectorNS maxSize (Stream step ms0) = do+    Stream step' $ do+        s0 <- ms0+        mv0 <- liftBase $ VM.new maxSize+        return (0, mv0, s0)+  where+    step' (i, mv, _) | i >= maxSize = liftM Stop $ liftBase $ V.unsafeFreeze mv+    step' (i, mv, s) = do+        res <- step s+        case res of+            Stop () -> liftM (Stop . V.slice 0 i) $ liftBase (V.unsafeFreeze mv)+            Skip s' -> return $ Skip (i, mv, s')+            Emit s' x -> do+                liftBase $ VM.write mv i x+                let i' = i + 1+                return $ Skip (i', mv, s')+{-# INLINE sinkVectorNS #-}++sinkLazyBuilderS :: (Monad m, Monoid builder, ToBuilder a builder, Builder builder lazy)+                 => StreamConsumer a m lazy+sinkLazyBuilderS = fmapS builderToLazy (foldS combiner mempty)+  where+    combiner accum = mappend accum . toBuilder+{-# INLINE sinkLazyBuilderS #-}++lastS :: Monad m+      => StreamConsumer a m (Maybe a)+lastS (Stream step ms0) =+    Stream step' (liftM (Nothing,) ms0)+  where+    step' (mlast, s) = do+        res <- step s+        return $ case res of+            Stop () -> Stop mlast+            Skip s' -> Skip (mlast, s')+            Emit s' x -> Skip (Just x, s')+{-# INLINE lastS #-}++lastES :: (Monad m, Seq.IsSequence seq)+       => StreamConsumer seq m (Maybe (Element seq))+lastES (Stream step ms0) =+    Stream step' (liftM (Nothing, ) ms0)+  where+    step' (mlast, s) = do+        res <- step s+        return $ case res of+            Stop () -> Stop (fmap NonNull.last mlast)+            Skip s' -> Skip (mlast, s')+            Emit s' (NonNull.fromNullable -> mlast'@(Just _)) -> Skip (mlast', s')+            Emit s' _ -> Skip (mlast, s')+{-# INLINE lastES #-}++findS :: Monad m+      => (a -> Bool) -> StreamConsumer a m (Maybe a)+findS f (Stream step ms0) =+    Stream step' ms0+  where+    step' s = do+      res <- step s+      return $ case res of+          Stop () -> Stop Nothing+          Skip s' -> Skip s'+          Emit s' x ->+              if f x+                  then Stop (Just x)+                  else Skip s'+{-# INLINE findS #-}++concatMapS :: (Monad m, MonoFoldable mono)+           => (a -> mono)+           -> StreamConduit a m (Element mono)+concatMapS f (Stream step ms0) =+    Stream step' (liftM ([], ) ms0)+  where+    step' ([], s) = do+        res <- step s+        return $ case res of+            Stop () -> Stop ()+            Skip s' -> Skip ([], s')+            Emit s' x -> Skip (otoList (f x), s')+    step' ((x:xs), s) = return (Emit (xs, s) x)+{-# INLINE concatMapS #-}++concatMapMS :: (Monad m, MonoFoldable mono)+             => (a -> m mono)+             -> StreamConduit a m (Element mono)+concatMapMS f (Stream step ms0) =+    Stream step' (liftM ([], ) ms0)+  where+    step' ([], s) = do+        res <- step s+        case res of+            Stop () -> return $ Stop ()+            Skip s' -> return $ Skip ([], s')+            Emit s' x -> do+                o <- f x+                return $ Skip (otoList o, s')+    step' ((x:xs), s) = return (Emit (xs, s) x)+{-# INLINE concatMapMS #-}++concatS :: (Monad m, MonoFoldable mono)+         => StreamConduit mono m (Element mono)+concatS = concatMapS id+{-# INLINE concatS #-}++data ScanState a s+    = ScanEnded+    | ScanContinues a s++scanlS :: Monad m => (a -> b -> a) -> a -> StreamConduit b m a+scanlS f seed0 (Stream step ms0) =+    Stream step' (liftM (ScanContinues seed0) ms0)+  where+    step' ScanEnded = return $ Stop ()+    step' (ScanContinues seed s) = do+        res <- step s+        return $ case res of+            Stop () -> Emit ScanEnded seed+            Skip s' -> Skip (ScanContinues seed s')+            Emit s' x -> Emit (ScanContinues seed' s') seed+              where+                !seed' = f seed x+{-# INLINE scanlS #-}++scanlMS :: Monad m => (a -> b -> m a) -> a -> StreamConduit b m a+scanlMS f seed0 (Stream step ms0) =+    Stream step' (liftM (ScanContinues seed0) ms0)+  where+    step' ScanEnded = return $ Stop ()+    step' (ScanContinues seed s) = do+        res <- step s+        case res of+            Stop () -> return $ Emit ScanEnded seed+            Skip s' -> return $ Skip (ScanContinues seed s')+            Emit s' x -> do+                !seed' <- f seed x+                return $ Emit (ScanContinues seed' s') seed+{-# INLINE scanlMS #-}++data IntersperseState a s+    = IFirstValue s+    | IGotValue s a+    | IEmitValue s a++intersperseS :: Monad m => a -> StreamConduit a m a+intersperseS sep (Stream step ms0) =+    Stream step' (liftM IFirstValue ms0)+  where+    step' (IFirstValue s) = do+        res <- step s+        return $ case res of+            Stop () -> Stop ()+            Skip s' -> Skip (IFirstValue s')+            Emit s' x -> Emit (IGotValue s' x) x+    -- Emit the separator once we know it's not the end of the list.+    step' (IGotValue s x) = do+        res <- step s+        return $ case res of+            Stop () -> Stop ()+            Skip s' -> Skip (IGotValue s' x)+            Emit s' x' -> Emit (IEmitValue s' x') sep+    -- We emitted a separator, now emit the value that comes after.+    step' (IEmitValue s x) = return $ Emit (IGotValue s x) x+{-# INLINE intersperseS #-}++data SlidingWindowState seq s+    = SWInitial Int seq s+    | SWSliding seq s+    | SWEarlyExit++slidingWindowS :: (Monad m, Seq.IsSequence seq, Element seq ~ a) => Int -> StreamConduit a m seq+slidingWindowS sz (Stream step ms0) =+    Stream step' (liftM (SWInitial (max 1 sz) mempty) ms0)+  where+    step' (SWInitial n st s) = do+        res <- step s+        return $ case res of+            Stop () -> Emit SWEarlyExit st+            Skip s' -> Skip (SWInitial n st s')+            Emit s' x ->+                if n == 1+                    then Emit (SWSliding (Seq.unsafeTail st') s') st'+                    else Skip (SWInitial (n - 1) st' s')+              where+                st' = Seq.snoc st x+    -- After collecting the initial window, each upstream element+    -- causes an additional window to be yielded.+    step' (SWSliding st s) = do+        res <- step s+        return $ case res of+            Stop () -> Stop ()+            Skip s' -> Skip (SWSliding st s')+            Emit s' x -> Emit (SWSliding (Seq.unsafeTail st') s') st'+              where+                st' = Seq.snoc st x+    step' SWEarlyExit = return $ Stop ()++{-# INLINE slidingWindowS #-}++filterMS :: Monad m+         => (a -> m Bool)+         -> StreamConduit a m a+filterMS f (Stream step ms0) = do+    Stream step' ms0+  where+    step' s = do+        res <- step s+        case res of+            Stop () -> return $ Stop ()+            Skip s' -> return $ Skip s'+            Emit s' x -> do+                r <- f x+                return $+                    if r+                        then Emit s' x+                        else Skip s'+{-# INLINE filterMS #-}++data SplitState seq s+    = SplitDone+    -- When no element of seq passes the predicate.  This allows+    -- 'splitOnUnboundedES' to not run 'Seq.break' multiple times due+    -- to 'Skip's being sent by the upstream.+    | SplitNoSep seq s+    | SplitState seq s++splitOnUnboundedES :: (Monad m, Seq.IsSequence seq)+                   => (Element seq -> Bool) -> StreamConduit seq m seq+splitOnUnboundedES f (Stream step ms0) =+    Stream step' (liftM (SplitState mempty) ms0)+  where+    step' SplitDone = return $ Stop ()+    step' (SplitNoSep t s) = do+        res <- step s+        return $ case res of+            Stop () | not (onull t) -> Emit SplitDone t+                    | otherwise -> Stop ()+            Skip s' -> Skip (SplitNoSep t s')+            Emit s' t' -> Skip (SplitState (t `mappend` t') s')+    step' (SplitState t s) = do+        if onull y+            then do+                res <- step s+                return $ case res of+                    Stop () | not (onull t) -> Emit SplitDone t+                            | otherwise -> Stop ()+                    Skip s' -> Skip (SplitNoSep t s')+                    Emit s' t' -> Skip (SplitState (t `mappend` t') s')+            else return $ Emit (SplitState (Seq.drop 1 y) s) x+      where+        (x, y) = Seq.break f t+{-# INLINE splitOnUnboundedES #-}++-- | Streaming versions of @Data.Conduit.Combinators.Internal.initReplicate@+initReplicateS :: Monad m => m seed -> (seed -> m a) -> Int -> StreamProducer m a+initReplicateS mseed f cnt _ =+    Stream step (liftM (cnt, ) mseed)+  where+    step (ix, _) | ix <= 0 = return $ Stop ()+    step (ix, seed) = do+        x <- f seed+        return $ Emit (ix - 1, seed) x+{-# INLINE initReplicateS #-}++-- | Streaming versions of @Data.Conduit.Combinators.Internal.initRepeat@+initRepeatS :: Monad m => m seed -> (seed -> m a) -> StreamProducer m a+initRepeatS mseed f _ =+    Stream step mseed+  where+    step seed = do+        x <- f seed+        return $ Emit seed x+{-# INLINE initRepeatS #-}++-- | Utility function+fmapS :: Monad m+      => (a -> b)+      -> StreamConduitM i o m a+      -> StreamConduitM i o m b+fmapS f s inp =+    case s inp of+        Stream step ms0 -> Stream (fmap (liftM (fmap f)) step) ms0+{-# INLINE fmapS #-}

@@ -225,11 +225,6 @@ import qualified System.Posix.Directory as Dir #endif -#if MIN_VERSION_conduit(1,1,0)-import qualified Data.Conduit.Filesystem as CF-#endif-- -- END IMPORTS  -- | Yield each of the values contained by the given @MonoFoldable@.

@@ -1,5 +1,5 @@ name:                conduit-combinators-version:             0.2.8.3+version:             0.3.0 synopsis:            Commonly used conduit functions, for both chunked and unchunked data description:         Provides a replacement for Data.Conduit.List, as well as a convenient Conduit module. homepage:            https://github.com/fpco/conduit-combinators@@ -10,16 +10,17 @@ category:            Data, Conduit build-type:          Simple cabal-version:       >=1.8-extra-source-files:  test/subdir/dummyfile.txt+extra-source-files:  test/subdir/dummyfile.txt fusion-macros.h ChangeLog.md  library   exposed-modules:     Conduit                        Data.Conduit.Combinators                        Data.Conduit.Combinators.Internal+                       Data.Conduit.Combinators.Stream   other-modules:       Data.Conduit.Combinators.Unqualified   build-depends:       base >= 4 && < 5                      , chunked-data-                     , conduit >= 1.0.12+                     , conduit >= 1.2.2                      , conduit-extra >= 1.1.1                      , transformers                      , transformers-base@@ -41,10 +42,13 @@       cpp-options:     -DWINDOWS   else       build-depends:   unix+  include-dirs:        .+  ghc-options:         -Wall -O2 -rtsopts  test-suite test   hs-source-dirs: test   main-is:        Spec.hs+  other-modules:  StreamSpec   type:           exitcode-stdio-1.0   cpp-options:    -DTEST   build-depends:  conduit-combinators@@ -61,6 +65,12 @@                 , base16-bytestring                 , base64-bytestring                 , system-filepath+                , mtl+                , conduit+                , containers+                , safe+                , QuickCheck+                , directory   ghc-options:    -Wall  source-repository head

@@ -0,0 +1,23 @@+#define INLINE_RULE0(new,old)            ;\+    new = old                            ;\+    {-# INLINE [0] new #-}               ;\+    {-# RULES "inline new" new = old #-}++#define INLINE_RULE(new,vars,body)                          ;\+    new vars = body                                         ;\+    {-# INLINE [0] new #-}                                  ;\+    {-# RULES "inline new" forall vars. new vars = body #-}++#define STREAMING0(name)                                 ;\+    name = name/**/C                                     ;\+    {-# INLINE [0] name #-}                              ;\+    {-# RULES "unstream name"                             \+      name = unstream (streamConduit name/**/C name/**/S) \+      #-}++#define STREAMING(name,vars)                                                ;\+    name = name/**/C                                                        ;\+    {-# INLINE [0] name #-}                                                 ;\+    {-# RULES "unstream name" forall vars.                                   \+      name vars = unstream (streamConduit (name/**/C vars) (name/**/S vars)) \+      #-}

@@ -39,6 +39,7 @@ import qualified Data.ByteString.Base64.URL.Lazy as B64LU import qualified Data.ByteString.Base64.URL as B64U import Control.Monad.ST (runST)+import qualified StreamSpec  main :: IO () main = hspec $ do@@ -140,13 +141,13 @@     it "sourceDirectory" $ do         res <- runResourceT              $ sourceDirectory "test" $$ filterC (not . flip hasExtension "swp") =$ sinkList-        sort res `shouldBe` ["test/Spec.hs", "test/subdir"]+        sort res `shouldBe` ["test/Spec.hs", "test/StreamSpec.hs", "test/subdir"]     it "sourceDirectoryDeep" $ do         res1 <- runResourceT               $ sourceDirectoryDeep False "test" $$ filterC (not . flip hasExtension "swp") =$ sinkList         res2 <- runResourceT               $ sourceDirectoryDeep True "test" $$ filterC (not . flip hasExtension "swp") =$ sinkList-        sort res1 `shouldBe` ["test/Spec.hs", "test/subdir/dummyfile.txt"]+        sort res1 `shouldBe` ["test/Spec.hs", "test/StreamSpec.hs", "test/subdir/dummyfile.txt"]         sort res1 `shouldBe` sort res2     prop "drop" $ \(T.pack -> input) count ->         runIdentity (yieldMany input $$ (dropC count >>= \() -> sinkList))@@ -625,6 +626,7 @@                   where                     (y, z) = splitAt size x         res `shouldBe` expected+    StreamSpec.spec  evenInt :: Int -> Bool evenInt = even

@@ -0,0 +1,480 @@+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ViewPatterns #-}+{-# LANGUAGE TypeFamilies #-}+module StreamSpec where++import           Control.Applicative+import qualified Control.Monad+import           Control.Monad (liftM)+import           Control.Monad.Identity (Identity, runIdentity)+import           Control.Monad.State (StateT(..), get, put)+import           Data.Conduit+import           Data.Conduit.Combinators+import           Data.Conduit.Combinators.Internal+import           Data.Conduit.Combinators.Stream+import           Data.Conduit.Internal.Fusion+import           Data.Conduit.Internal.List.Stream (takeS, sourceListS, mapS)+import           Data.Conduit.List (consume, isolate, sourceList)+import qualified Data.List+import           Data.MonoTraversable+import           Data.Monoid (Monoid(..))+import qualified Data.NonNull as NonNull+import           Data.Sequence (Seq)+import qualified Data.Sequences as Seq+import qualified Data.Text.Lazy as TL+import           Data.Vector (Vector)+import qualified Prelude+import           Prelude+    ((.), ($), (=<<), return, id, Maybe(..), Monad, Bool(..), Int,+     Eq, Show, String, Functor, fst, snd)+import qualified Safe+import           System.Directory (removeFile)+import qualified System.IO as IO+import           System.IO.Unsafe+import           Test.Hspec+import           Test.QuickCheck++spec :: Spec+spec = do+    it "sourceHandleS works" $ do+        let contents = Prelude.concat $ Prelude.replicate 10000 $ "this is some content\n"+            fp = "tmp"+        IO.writeFile fp contents+        (res, ()) <- IO.withBinaryFile "tmp" IO.ReadMode $ \h ->+            evalStream $ sourceHandleS h emptyStream+        (TL.concat res) `shouldBe` TL.pack contents+        removeFile "tmp"+    describe "Comparing list function to" $ do+        qit "yieldMany" $+            \(mono :: Seq Int) ->+                yieldMany mono `checkProducer`+                otoList mono+        qit "yieldManyS" $+            \(mono :: Seq Int) ->+                yieldManyS mono `checkStreamProducer`+                otoList mono+        qit "repeatM" $+            \(getBlind -> (f :: M Int)) ->+                repeatM f `checkInfiniteProducerM`+                repeatML f+        qit "repeatMS" $+            \(getBlind -> (f :: M Int)) ->+                repeatMS f `checkInfiniteStreamProducerM`+                repeatML f+        qit "repeatWhileM" $+            \(getBlind -> (f :: M Int), getBlind -> g) ->+                repeatWhileM f g `checkInfiniteProducerM`+                repeatWhileML f g+        qit "repeatWhileMS" $+            \(getBlind -> (f :: M Int), getBlind -> g) ->+                repeatWhileMS f g `checkInfiniteStreamProducerM`+                repeatWhileML f g+        qit "foldl1" $+            \(getBlind -> f) ->+                foldl1 f `checkConsumer`+                foldl1L f+        qit "foldl1S" $+            \(getBlind -> f) ->+                foldl1S f `checkStreamConsumer`+                foldl1L f+        qit "all" $+            \(getBlind -> f) ->+                all f `checkConsumer`+                Prelude.all f+        qit "allS" $+            \(getBlind -> f) ->+                allS f `checkStreamConsumer`+                Prelude.all f+        qit "any" $+            \(getBlind -> f) ->+                any f `checkConsumer`+                Prelude.any f+        qit "anyS" $+            \(getBlind -> f) ->+                anyS f `checkStreamConsumer`+                Prelude.any f+        qit "last" $+            \() ->+                last `checkConsumer`+                Safe.lastMay+        qit "lastS" $+            \() ->+                lastS `checkStreamConsumer`+                Safe.lastMay+        qit "lastE" $+            \(getBlind -> f) ->+                let g x = Seq.replicate (Prelude.abs (f x)) x :: Seq Int+                 in (map g =$= lastE) `checkConsumer`+                    (lastEL . Prelude.map g :: [Int] -> Maybe Int)+        qit "lastES" $+            \(getBlind -> f) ->+                let g x = Seq.replicate (Prelude.abs (f x)) x :: Seq Int+                 in (lastES . mapS g) `checkStreamConsumer`+                    (lastEL . Prelude.map g :: [Int] -> Maybe Int)+        qit "find" $+            \(getBlind -> f) ->+                find f `checkConsumer`+                Data.List.find f+        qit "findS" $+            \(getBlind -> f) ->+                findS f `checkStreamConsumer`+                Data.List.find f+        qit "concatMap" $+            \(getBlind -> (f :: Int -> Seq Int)) ->+                concatMap f `checkConduit`+                concatMapL f+        qit "concatMapS" $+            \(getBlind -> (f :: Int -> Seq Int)) ->+                concatMapS f `checkStreamConduit`+                concatMapL f+        qit "concatMapM" $+            \(getBlind -> (f :: Int -> M (Seq Int))) ->+                concatMapM f `checkConduitM`+                concatMapML f+        qit "concatMapMS" $+            \(getBlind -> (f :: Int -> M (Seq Int))) ->+                concatMapMS f `checkStreamConduitM`+                concatMapML f+        qit "concat" $+            \() ->+                concat `checkConduit`+                (concatL :: [Seq Int] -> [Int])+        qit "concatS" $+            \() ->+                concatS `checkStreamConduit`+                (concatL :: [Seq Int] -> [Int])+        qit "scanl" $+            \(getBlind -> (f :: Int -> Int -> Int), initial) ->+                scanl f initial `checkConduit`+                Prelude.scanl f initial+        qit "scanlS" $+            \(getBlind -> (f :: Int -> Int -> Int), initial) ->+                scanlS f initial `checkStreamConduit`+                Prelude.scanl f initial+        qit "scanlM" $+            \(getBlind -> (f :: Int -> Int -> M Int), initial) ->+                scanlM f initial `checkConduitM`+                scanlML f initial+        qit "scanlMS" $+            \(getBlind -> (f :: Int -> Int -> M Int), initial) ->+                scanlMS f initial `checkStreamConduitM`+                scanlML f initial+        qit "intersperse" $+            \(sep :: Int) ->+                intersperse sep `checkConduit`+                Data.List.intersperse sep+        qit "intersperseS" $+            \(sep :: Int) ->+                intersperseS sep `checkStreamConduit`+                Data.List.intersperse sep+        qit "filterM" $+            \(getBlind -> (f :: Int -> M Bool)) ->+                filterM f `checkConduitM`+                Control.Monad.filterM f+        qit "filterMS" $+            \(getBlind -> (f :: Int -> M Bool)) ->+                filterMS f `checkStreamConduitM`+                Control.Monad.filterM f+    describe "comparing normal conduit function to" $ do+        qit "slidingWindowS" $+            \(getSmall -> n) ->+                slidingWindowS n `checkStreamConduit`+                (\xs -> runIdentity $+                    sourceList xs $= preventFusion (slidingWindow n) $$ consume+                    :: [Seq Int])+        qit "splitOnUnboundedES" $+            \(getBlind -> (f :: Int -> Bool)) ->+                splitOnUnboundedES f `checkStreamConduit`+                (\xs -> runIdentity $+                    sourceList xs $= preventFusion (splitOnUnboundedE f) $$ consume+                    :: [Seq Int])+        qit "initReplicateS" $+            \(getBlind -> (mseed :: M Int), getBlind -> (f :: Int -> M Int), getSmall -> cnt) ->+                initReplicateS mseed f cnt `checkStreamProducerM`+                (preventFusion (initReplicate mseed f cnt) $$ consume)+        qit "initRepeatS" $+            \(getBlind -> (mseed :: M Int), getBlind -> (f :: Int -> M Int)) ->+                initRepeatS mseed f `checkInfiniteStreamProducerM`+                (preventFusion (initRepeat mseed f) $= take 10 $$ consume)+        qit "sinkVectorS" $+            \() -> checkStreamConsumerM'+                unsafePerformIO+                (sinkVectorS :: forall o. StreamConduitM Int o IO.IO (Vector Int))+                (\xs -> sourceList xs $$ preventFusion sinkVector)+        qit "sinkVectorNS" $+            \(getSmall . getNonNegative -> n) -> checkStreamConsumerM'+                unsafePerformIO+                (sinkVectorNS n :: forall o. StreamConduitM Int o IO.IO (Vector Int))+                (\xs -> sourceList xs $$ preventFusion (sinkVectorN n))++instance Arbitrary a => Arbitrary (Seq a) where+    arbitrary = Seq.fromList <$> arbitrary++repeatML :: Monad m => m a -> m [a]+repeatML = Prelude.sequence . Prelude.repeat++repeatWhileML :: Monad m => m a -> (a -> Bool) -> m [a]+repeatWhileML m f = go+  where+    go = do+        x <- m+        if f x+           then liftM (x:) go+           else return []++foldl1L :: (a -> a -> a) -> [a] -> Maybe a+foldl1L _ [] = Nothing+foldl1L f xs = Just $ Prelude.foldl1 f xs++lastEL :: Seq.IsSequence seq+       => [seq] -> Maybe (Element seq)+lastEL = Prelude.foldl go Nothing+  where+    go _ (NonNull.fromNullable -> Just l) = Just (NonNull.last l)+    go mlast _ = mlast++concatMapL :: MonoFoldable mono+           => (a -> mono) -> [a] -> [Element mono]+concatMapL f = Prelude.concatMap (otoList . f)++concatMapML :: (Monad m, MonoFoldable mono)+             => (a -> m mono) -> [a] -> m [Element mono]+concatMapML f = liftM (Prelude.concatMap otoList) . Prelude.mapM f++concatL :: MonoFoldable mono+        => [mono] -> [Element mono]+concatL = Prelude.concatMap otoList++scanlML :: Monad m => (a -> b -> m a) -> a -> [b] -> m [a]+scanlML f = go+  where+    go l [] = return [l]+    go l (r:rs) = do+        l' <- f l r+        liftM (l:) (go l' rs)++--FIXME: the following code is directly copied from the conduit test+--suite.  How to share this code??++qit :: (Arbitrary a, Testable prop, Show a)+     => String -> (a -> prop) -> Spec+qit n f = it n $ property $ forAll arbitrary f++--------------------------------------------------------------------------------+-- Quickcheck utilities for pure conduits / streams++checkProducer :: (Show a, Eq a) => Source Identity a -> [a] -> Property+checkProducer c l  = checkProducerM' runIdentity c (return l)++checkStreamProducer :: (Show a, Eq a) => StreamSource Identity a -> [a] -> Property+checkStreamProducer s l = checkStreamProducerM' runIdentity s (return l)++checkInfiniteProducer :: (Show a, Eq a) => Source Identity a -> [a] -> Property+checkInfiniteProducer c l = checkInfiniteProducerM' runIdentity c (return l)++checkInfiniteStreamProducer :: (Show a, Eq a) => StreamSource Identity a -> [a] -> Property+checkInfiniteStreamProducer s l = checkInfiniteStreamProducerM' runIdentity s (return l)++checkConsumer :: (Show b, Eq b) => Consumer Int Identity b -> ([Int] -> b) -> Property+checkConsumer c l = checkConsumerM' runIdentity c (return . l)++checkStreamConsumer :: (Show b, Eq b) => StreamConsumer Int Identity b -> ([Int] -> b) -> Property+checkStreamConsumer c l = checkStreamConsumerM' runIdentity c (return . l)++checkConduit :: (Show a, Arbitrary a, Show b, Eq b) => Conduit a Identity b -> ([a] -> [b]) -> Property+checkConduit c l = checkConduitM' runIdentity c (return . l)++checkStreamConduit :: (Show a, Arbitrary a, Show b, Eq b) => StreamConduit a Identity b -> ([a] -> [b]) -> Property+checkStreamConduit c l = checkStreamConduitM' runIdentity c (return . l)++-- checkConduitResult :: (Show a, Arbitrary a, Show b, Eq b, Show r, Eq r) => ConduitM a b Identity r -> ([a] -> ([b], r)) -> Property+-- checkConduitResult c l = checkConduitResultM' runIdentity c (return . l)++checkStreamConduitResult :: (Show a, Arbitrary a, Show b, Eq b, Show r, Eq r) => StreamConduitM a b Identity r -> ([a] -> ([b], r)) -> Property+checkStreamConduitResult c l = checkStreamConduitResultM' runIdentity c (return . l)++--------------------------------------------------------------------------------+-- Quickcheck utilities for conduits / streams in the M monad.++checkProducerM :: (Show a, Eq a) => Source M a -> M [a] -> Property+checkProducerM = checkProducerM' runM++checkStreamProducerM :: (Show a, Eq a) => StreamSource M a -> M [a] -> Property+checkStreamProducerM = checkStreamProducerM' runM++checkInfiniteProducerM :: (Show a, Eq a) => Source M a -> M [a] -> Property+checkInfiniteProducerM = checkInfiniteProducerM' (fst . runM)++checkInfiniteStreamProducerM :: (Show a, Eq a) => StreamSource M a -> M [a] -> Property+checkInfiniteStreamProducerM = checkInfiniteStreamProducerM' (fst . runM)++checkConsumerM :: (Show b, Eq b) => Consumer Int M b -> ([Int] -> M b) -> Property+checkConsumerM  = checkConsumerM' runM++checkStreamConsumerM :: (Show b, Eq b) => StreamConsumer Int M b -> ([Int] -> M b) -> Property+checkStreamConsumerM  = checkStreamConsumerM' runM++checkConduitM :: (Show a, Arbitrary a, Show b, Eq b) => Conduit a M b -> ([a] -> M [b]) -> Property+checkConduitM = checkConduitM' runM++checkStreamConduitM :: (Show a, Arbitrary a, Show b, Eq b) => StreamConduit a M b -> ([a] -> M [b]) -> Property+checkStreamConduitM = checkStreamConduitM' runM++-- checkConduitResultM :: (Show a, Arbitrary a, Show b, Eq b, Show r, Eq r) => ConduitM a b M r -> ([a] -> M ([b], r)) -> Property+-- checkConduitResultM = checkConduitResultM' runM++checkStreamConduitResultM :: (Show a, Arbitrary a, Show b, Eq b, Show r, Eq r) => StreamConduitM a b M r -> ([a] -> M ([b], r)) -> Property+checkStreamConduitResultM = checkStreamConduitResultM' runM++--------------------------------------------------------------------------------+-- Quickcheck utilities for monadic streams / conduits+-- These are polymorphic in which Monad is used.++checkProducerM' :: (Show a, Monad m, Show b, Eq b)+                => (m [a] -> b)+                -> Source m a+                -> m [a]+                -> Property+checkProducerM' f c l =+    f (preventFusion c $$ consume)+    ===+    f l++checkStreamProducerM' :: (Show a, Monad m, Show b, Eq b)+                      => (m [a] -> b)+                      -> StreamSource m a+                      -> m [a]+                      -> Property+checkStreamProducerM' f s l =+    f (liftM fst $ evalStream $ s emptyStream)+    ===+    f l++checkInfiniteProducerM' :: (Show a, Monad m, Show b, Eq b)+                        => (m [a] -> b)+                        -> Source m a+                        -> m [a]+                        -> Property+checkInfiniteProducerM' f s l =+    checkProducerM' f+        (preventFusion s $= isolate 10)+        (liftM (Prelude.take 10) l)++checkInfiniteStreamProducerM' :: (Show a, Monad m, Show b, Eq b)+                              => (m [a] -> b)+                              -> StreamSource m a+                              -> m [a]+                              -> Property+checkInfiniteStreamProducerM' f s l =+    f (liftM snd $ evalStream $ takeS 10 $ s emptyStream)+    ===+    f (liftM (Prelude.take 10) l)++checkConsumerM' :: (Show a, Monad m, Show b, Eq b)+                => (m a -> b)+                -> Consumer Int m a+                -> ([Int] -> m a)+                -> Property+checkConsumerM' f c l = forAll arbitrary $ \xs ->+    f (sourceList xs $$ preventFusion c)+    ===+    f (l xs)++checkStreamConsumerM' :: (Show a, Monad m, Show b, Eq b)+                      => (m a -> b)+                      -> StreamConsumer Int m a+                      -> ([Int] -> m a)+                      -> Property+checkStreamConsumerM' f s l = forAll arbitrary $ \xs ->+    f (liftM snd $ evalStream $ s $ sourceListS xs emptyStream)+    ===+    f (l xs)++checkConduitM' :: (Show a, Arbitrary a, Monad m, Show c, Eq c)+               => (m [b] -> c)+               -> Conduit a m b+               -> ([a] -> m [b])+               -> Property+checkConduitM' f c l = forAll arbitrary $ \xs ->+    f (sourceList xs $= preventFusion c $$ consume)+    ===+    f (l xs)++checkStreamConduitM' :: (Show a, Arbitrary a, Monad m, Show c, Eq c)+                     =>  (m [b] -> c)+                     -> StreamConduit a m b+                     -> ([a] -> m [b])+                     -> Property+checkStreamConduitM' f s l = forAll arbitrary $ \xs ->+    f (liftM fst $ evalStream $ s $ sourceListS xs emptyStream)+    ===+    f (l xs)++-- TODO: Fixing this would allow comparing conduit consumers against+-- their list versions.+--+-- checkConduitResultM' :: (Show a, Arbitrary a, Monad m, Show c, Eq c)+--                      => (m ([b], r) -> c)+--                      -> ConduitM a b m r+--                      -> ([a] -> m ([b], r))+--                      -> Property+-- checkConduitResultM' f c l = FIXME forAll arbitrary $ \xs ->+--     f (sourceList xs $= preventFusion c $$ consume)+--     ===+--     f (l xs)++checkStreamConduitResultM' :: (Show a, Arbitrary a, Monad m, Show c, Eq c)+                           =>  (m ([b], r) -> c)+                           -> StreamConduitM a b m r+                           -> ([a] -> m ([b], r))+                           -> Property+checkStreamConduitResultM' f s l = forAll arbitrary $ \xs ->+    f (evalStream $ s $ sourceListS xs emptyStream)+    ===+    f (l xs)++emptyStream :: Monad m => Stream m () ()+emptyStream = Stream (\_ -> return $ Stop ()) (return ())++evalStream :: Monad m => Stream m o r -> m ([o], r)+evalStream (Stream step s0) = go =<< s0+  where+    go s = do+        res <- step s+        case res of+            Stop r -> return ([], r)+            Skip s' -> go s'+            Emit s' x -> liftM (\(l, r) -> (x:l, r)) (go s')++--------------------------------------------------------------------------------+-- Misc utilities++-- Prefer this to creating an orphan instance for Data.Monoid.Sum:++newtype Sum a = Sum a+  deriving (Eq, Show, Arbitrary)++instance Prelude.Num a => Monoid (Sum a) where+  mempty = Sum 0+  mappend (Sum x) (Sum y) = Sum $ x Prelude.+ y++preventFusion :: a -> a+preventFusion = id+{-# INLINE [0] preventFusion #-}++newtype M a = M (StateT Int Identity a)+  deriving (Functor, Applicative, Monad)++instance Arbitrary a => Arbitrary (M a) where+    arbitrary = do+        f <- arbitrary+        return $ do+            s <- M get+            let (x, s') = f s+            M (put s')+            return x++runM :: M a -> (a, Int)+runM (M m) = runIdentity $ runStateT m 0