diff --git a/ChangeLog.md b/ChangeLog.md new file mode 100644 --- /dev/null +++ b/ChangeLog.md @@ -0,0 +1,1 @@ +__0.3.0__ Stream fusion enabled, drop compatibility with older conduit diff --git a/Data/Conduit/Combinators.hs b/Data/Conduit/Combinators.hs --- a/Data/Conduit/Combinators.hs +++ b/Data/Conduit/Combinators.hs @@ -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) diff --git a/Data/Conduit/Combinators/Internal.hs b/Data/Conduit/Combinators/Internal.hs --- a/Data/Conduit/Combinators/Internal.hs +++ b/Data/Conduit/Combinators/Internal.hs @@ -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 diff --git a/Data/Conduit/Combinators/Stream.hs b/Data/Conduit/Combinators/Stream.hs new file mode 100644 --- /dev/null +++ b/Data/Conduit/Combinators/Stream.hs @@ -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 #-} diff --git a/Data/Conduit/Combinators/Unqualified.hs b/Data/Conduit/Combinators/Unqualified.hs --- a/Data/Conduit/Combinators/Unqualified.hs +++ b/Data/Conduit/Combinators/Unqualified.hs @@ -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@. diff --git a/conduit-combinators.cabal b/conduit-combinators.cabal --- a/conduit-combinators.cabal +++ b/conduit-combinators.cabal @@ -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 diff --git a/fusion-macros.h b/fusion-macros.h new file mode 100644 --- /dev/null +++ b/fusion-macros.h @@ -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)) \ + #-} diff --git a/test/Spec.hs b/test/Spec.hs --- a/test/Spec.hs +++ b/test/Spec.hs @@ -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 diff --git a/test/StreamSpec.hs b/test/StreamSpec.hs new file mode 100644 --- /dev/null +++ b/test/StreamSpec.hs @@ -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