Data.Conduit.Combinators

This module is meant as a replacement for Data.Conduit.List. That module follows a naming scheme which was originally inspired by its enumerator roots. This module is meant to introduce a naming scheme which encourages conduit best practices.

There are two versions of functions in this module. Those with a trailing E work in the individual elements of a chunk of data, e.g., the bytes of a ByteString, the Chars of a Text, or the Ints of a Vector Int. Those without a trailing E work on unchunked streams.

FIXME: discuss overall naming, usage of mono-traversable, etc

Mention take (Conduit) vs drop (Consumer)

Producers

Pure

yieldMany :: (Monad m, MonoFoldable mono) => mono -> Producer m (Element mono)

Yield each of the values contained by the given MonoFoldable.

This will work on many data structures, including lists, ByteStrings, and Vectors.

Since 1.0.0

unfold :: Monad m => (b -> Maybe (a, b)) -> b -> Producer m a

Generate a producer from a seed value.

Since 1.0.0

enumFromTo :: (Monad m, Enum a, Eq a) => a -> a -> Producer m a

Enumerate from a value to a final value, inclusive, via succ.

This is generally more efficient than using Prelude's enumFromTo and combining with sourceList since this avoids any intermediate data structures.

Since 1.0.0

iterate :: Monad m => (a -> a) -> a -> Producer m a

Produces an infinite stream of repeated applications of f to x.

Since 1.0.0

repeat :: Monad m => a -> Producer m a

Produce an infinite stream consisting entirely of the given value.

Since 1.0.0

replicate :: Monad m => Int -> a -> Producer m a

Produce a finite stream consisting of n copies of the given value.

Since 1.0.0

sourceLazy :: (Monad m, LazySequence lazy strict) => lazy -> Producer m strict

Generate a producer by yielding each of the strict chunks in a LazySequence.

For more information, see toChunks.

Since 1.0.0

Monadic

repeatM :: Monad m => m a -> Producer m a

Repeatedly run the given action and yield all values it produces.

Since 1.0.0

repeatWhileM :: Monad m => m a -> (a -> Bool) -> Producer m a

Repeatedly run the given action and yield all values it produces, until the provided predicate returns False.

Since 1.0.0

replicateM :: Monad m => Int -> m a -> Producer m a

Perform the given action n times, yielding each result.

Since 1.0.0

I/O

sourceFile :: (MonadResource m, IOData a) => FilePath -> Producer m a

sourceHandle :: (MonadIO m, IOData a) => Handle -> Producer m a

Read all data from the given Handle.

Does not close the Handle at any point.

Since 1.0.0

sourceIOHandle :: (MonadResource m, IOData a) => IO Handle -> Producer m a

Open a Handle using the given function and stream data from it.

Automatically closes the file at completion.

Since 1.0.0

stdin :: (MonadIO m, IOData a) => Producer m a

sourceHandle applied to stdin.

Since 1.0.0

Consumers

Pure

drop :: Monad m => Int -> Consumer a m ()

Ignore a certain number of values in the stream.

Since 1.0.0

dropE :: (Monad m, IsSequence seq) => Index seq -> Consumer seq m ()

Drop a certain number of elements from a chunked stream.

Since 1.0.0

dropWhile :: Monad m => (a -> Bool) -> Consumer a m ()

Drop all values which match the given predicate.

Since 1.0.0

dropWhileE :: (Monad m, IsSequence seq) => (Element seq -> Bool) -> Consumer seq m ()

Drop all elements in the chunked stream which match the given predicate.

Since 1.0.0

fold :: (Monad m, Monoid a) => Consumer a m a

Monoidally combine all values in the stream.

Since 1.0.0

foldE :: (Monad m, MonoFoldable mono, Monoid (Element mono)) => Consumer mono m (Element mono)

Monoidally combine all elements in the chunked stream.

Since 1.0.0

foldl :: Monad m => (a -> b -> a) -> a -> Consumer b m a

A strict left fold.

Since 1.0.0

foldlE :: (Monad m, MonoFoldable mono) => (a -> Element mono -> a) -> a -> Consumer mono m a

A strict left fold on a chunked stream.

Since 1.0.0

foldMap :: (Monad m, Monoid b) => (a -> b) -> Consumer a m b

Apply the provided mapping function and monoidal combine all values.

Since 1.0.0

foldMapE :: (Monad m, MonoFoldable mono, Monoid w) => (Element mono -> w) -> Consumer mono m w

Apply the provided mapping function and monoidal combine all elements of the chunked stream.

Since 1.0.0

all :: Monad m => (a -> Bool) -> Consumer a m Bool

Check that all values in the stream return True.

Subject to shortcut logic: at the first False, consumption of the stream will stop.

Since 1.0.0

allE :: (Monad m, MonoFoldable mono) => (Element mono -> Bool) -> Consumer mono m Bool

Check that all elements in the chunked stream return True.

Subject to shortcut logic: at the first False, consumption of the stream will stop.

Since 1.0.0

any :: Monad m => (a -> Bool) -> Consumer a m Bool

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.

Since 1.0.0

anyE :: (Monad m, MonoFoldable mono) => (Element mono -> Bool) -> Consumer mono m Bool

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.

Since 1.0.0

and :: Monad m => Consumer Bool m Bool

Are all values in the stream True?

Consumption stops once the first False is encountered.

Since 1.0.0

andE :: (Monad m, MonoFoldable mono, Element mono ~ Bool) => Consumer mono m Bool

Are all elements in the chunked stream True?

Consumption stops once the first False is encountered.

Since 1.0.0

or :: Monad m => Consumer Bool m Bool

Are any values in the stream True?

Consumption stops once the first True is encountered.

Since 1.0.0

orE :: (Monad m, MonoFoldable mono, Element mono ~ Bool) => Consumer mono m Bool

Are any elements in the chunked stream True?

Consumption stops once the first True is encountered.

Since 1.0.0

elem :: (Monad m, Eq a) => a -> Consumer a m Bool

Are any values in the stream equal to the given value?

Stops consuming as soon as a match is found.

Since 1.0.0

elemE :: (Monad m, EqSequence seq) => Element seq -> Consumer seq m Bool

Are any elements in the chunked stream equal to the given element?

Stops consuming as soon as a match is found.

Since 1.0.0

notElem :: (Monad m, Eq a) => a -> Consumer a m Bool

Are no values in the stream equal to the given value?

Stops consuming as soon as a match is found.

Since 1.0.0

notElemE :: (Monad m, EqSequence seq) => Element seq -> Consumer seq m Bool

Are no elements in the chunked stream equal to the given element?

Stops consuming as soon as a match is found.

Since 1.0.0

sinkLazy :: (Monad m, LazySequence lazy strict) => Consumer strict m lazy

Consume all incoming strict chunks into a lazy sequence. Note that the entirety of the sequence will be resident at memory.

This can be used to consume a stream of strict ByteStrings into a lazy ByteString, for example.

Since 1.0.0

sinkList :: Monad m => Consumer a m [a]

Consume all values from the stream and return as a list. Note that this will pull all values into memory.

Since 1.0.0

sinkVector :: (MonadBase base m, Vector v a, PrimMonad base) => Consumer a m (v a)

Sink incoming values into a vector, growing the vector as necessary to fit more elements.

Note that using this function is more memory efficient than sinkList and then converting to a Vector, as it avoids intermediate list constructors.

Since 1.0.0

sinkVectorN

Arguments

:: (MonadBase base m, Vector v a, PrimMonad base)

=> Int

maximum allowed size

-> Consumer a m (v a)

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 present, returns a Vector of smaller size.

Note that using this function is more memory efficient than sinkList and then converting to a Vector, as it avoids intermediate list constructors.

Since 1.0.0

sinkBuilder :: (Monad m, Monoid builder, ToBuilder a builder) => Consumer a m builder

Convert incoming values to a builder and fold together all builder values.

Defined as: foldMap toBuilder.

Since 1.0.0

sinkLazyBuilder :: (Monad m, Monoid builder, ToBuilder a builder, Builder builder lazy) => Consumer a m lazy

Same as sinkBuilder, but afterwards convert the builder to its lazy representation.

Alternatively, this could be considered an alternative to sinkLazy, with the following differences:

• This function will allow multiple input types, not just the strict version of the lazy structure.
• Some buffer copying may occur in this version.

Since 1.0.0

sinkNull :: Monad m => Consumer a m ()

Consume and discard all remaining values in the stream.

Since 1.0.0

awaitNonNull :: (Monad m, MonoFoldable a) => Consumer a m (Maybe (NonNull a))

Same as await, but discards any leading onull values.

Since 1.0.0

headE :: (Monad m, IsSequence seq) => Consumer seq m (Maybe (Element seq))

Get the next element in the chunked stream.

Since 1.0.0

peek :: Monad m => Consumer a m (Maybe a)

View the next value in the stream without consuming it.

Since 1.0.0

peekE :: (Monad m, MonoFoldable mono) => Consumer mono m (Maybe (Element mono))

View the next element in the chunked stream without consuming it.

Since 1.0.0

last :: Monad m => Consumer a m (Maybe a)

Retrieve the last value in the stream, if present.

Since 1.0.0

lastE :: (Monad m, IsSequence seq) => Consumer seq m (Maybe (Element seq))

Retrieve the last element in the chunked stream, if present.

Since 1.0.0

length :: (Monad m, Num len) => Consumer a m len

Count how many values are in the stream.

Since 1.0.0

lengthE :: (Monad m, Num len, MonoFoldable mono) => Consumer mono m len

Count how many elements are in the chunked stream.

Since 1.0.0

maximum :: (Monad m, Ord a) => Consumer a m (Maybe a)

Get the largest value in the stream, if present.

Since 1.0.0

maximumE :: (Monad m, OrdSequence seq) => Consumer seq m (Maybe (Element seq))

Get the largest element in the chunked stream, if present.

Since 1.0.0

minimum :: (Monad m, Ord a) => Consumer a m (Maybe a)

Get the smallest value in the stream, if present.

Since 1.0.0

minimumE :: (Monad m, OrdSequence seq) => Consumer seq m (Maybe (Element seq))

Get the smallest element in the chunked stream, if present.

Since 1.0.0

null :: Monad m => Consumer a m Bool

True if there are no values in the stream.

This function does not modify the stream.

Since 1.0.0

nullE :: (Monad m, MonoFoldable mono) => Consumer mono m Bool

True if there are no elements in the chunked stream.

This function may remove empty leading chunks from the stream, but otherwise will not modify it.

Since 1.0.0

sum :: (Monad m, Num a) => Consumer a m a

Get the sum of all values in the stream.

Since 1.0.0

sumE :: (Monad m, MonoFoldable mono, Num (Element mono)) => Consumer mono m (Element mono)

Get the sum of all elements in the chunked stream.

Since 1.0.0

product :: (Monad m, Num a) => Consumer a m a

Get the product of all values in the stream.

Since 1.0.0

productE :: (Monad m, MonoFoldable mono, Num (Element mono)) => Consumer mono m (Element mono)

Get the product of all elements in the chunked stream.

Since 1.0.0

find :: Monad m => (a -> Bool) -> Consumer a m (Maybe a)

Find the first matching value.

Since 1.0.0

Monadic

mapM_ :: Monad m => (a -> m ()) -> Consumer a m ()

Apply the action to all values in the stream.

Since 1.0.0

mapM_E :: (Monad m, MonoFoldable mono) => (Element mono -> m ()) -> Consumer mono m ()

Apply the action to all elements in the chunked stream.

Since 1.0.0

foldM :: Monad m => (a -> b -> m a) -> a -> Consumer b m a

A monadic strict left fold.

Since 1.0.0

foldME :: (Monad m, MonoFoldable mono) => (a -> Element mono -> m a) -> a -> Consumer mono m a

A monadic strict left fold on a chunked stream.

Since 1.0.0

foldMapM :: (Monad m, Monoid w) => (a -> m w) -> Consumer a m w

Apply the provided monadic mapping function and monoidal combine all values.

Since 1.0.0

foldMapME :: (Monad m, MonoFoldable mono, Monoid w) => (Element mono -> m w) -> Consumer mono m w

Apply the provided monadic mapping function and monoidal combine all elements in the chunked stream.

Since 1.0.0

I/O

sinkFile :: (MonadResource m, IOData a) => FilePath -> Consumer a m ()

sinkHandle :: (MonadIO m, IOData a) => Handle -> Consumer a m ()

Write all data to the given Handle.

Does not close the Handle at any point.

Since 1.0.0

sinkIOHandle :: (MonadResource m, IOData a) => IO Handle -> Consumer a m ()

Open a Handle using the given function and stream data to it.

Automatically closes the file at completion.

Since 1.0.0

print :: (Show a, MonadIO m) => Consumer a m ()

Print all incoming values to stdout.

Since 1.0.0

stdout :: (MonadIO m, IOData a) => Consumer a m ()

sinkHandle applied to stdout.

Since 1.0.0

stderr :: (MonadIO m, IOData a) => Consumer a m ()

sinkHandle applied to stderr.

Since 1.0.0

Transformers

Pure

map :: Monad m => (a -> b) -> Conduit a m b

Apply a transformation to all values in a stream.

Since 1.0.0

mapE :: (Monad m, Functor f) => (a -> b) -> Conduit (f a) m (f b)

Apply a transformation to all elements in a chunked stream.

Since 1.0.0

omapE :: (Monad m, MonoFunctor mono) => (Element mono -> Element mono) -> Conduit mono m mono

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.

Since 1.0.0

concatMap :: (Monad m, MonoFoldable mono) => (a -> mono) -> Conduit a m (Element mono)

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 foldable value separately.

Generalizes concatMap, mapMaybe, and mapFoldable.

Since 1.0.0

concatMapE :: (Monad m, MonoFoldable mono, Monoid w) => (Element mono -> w) -> Conduit mono m w

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 that foldable value separately.

Generalizes concatMap, mapMaybe, and mapFoldable.

Since 1.0.0

take :: Monad m => Int -> Conduit a m a

Stream up to n number of values downstream.

Note that, if downstream terminates early, not all values will be consumed. If you want to force exactly the given number of values to be consumed, see takeExactly.

Since 1.0.0

takeE :: (Monad m, IsSequence seq) => Index seq -> Conduit seq m seq

Stream up to n number of elements downstream in a chunked stream.

Note that, if downstream terminates early, not all values will be consumed. If you want to force exactly the given number of values to be consumed, see takeExactlyE.

Since 1.0.0

takeWhile :: Monad m => (a -> Bool) -> Conduit a m a

Stream all values downstream that match the given predicate.

Same caveats regarding downstream termination apply as with take.

Since 1.0.0

takeWhileE :: (Monad m, IsSequence seq) => (Element seq -> Bool) -> Conduit seq m seq

Stream all elements downstream that match the given predicate in a chunked stream.

Same caveats regarding downstream termination apply as with takeE.

Since 1.0.0

takeExactly :: Monad m => Int -> ConduitM a b m r -> ConduitM a b m r

Consume precisely the given number of values and feed them downstream.

This function is in contrast to take, which will only consume up to the given number of values, and will terminate early if downstream terminates early. This function will discard any additional values in the stream if they are unconsumed.

Note that this function takes a downstream ConduitM as a parameter, as opposed to working with normal fusion. For more information, see http://www.yesodweb.com/blog/2013/10/core-flaw-pipes-conduit, the section titled "pipes and conduit: isolate".

Since 1.0.0

takeExactlyE :: (Monad m, IsSequence a) => Index a -> ConduitM a b m r -> ConduitM a b m r

Same as takeExactly, but for chunked streams.

Since 1.0.0

concat :: (Monad m, MonoFoldable mono) => Conduit mono m (Element mono)

Flatten out a stream by yielding the values contained in an incoming MonoFoldable as individually yielded values.

Since 1.0.0

filter :: Monad m => (a -> Bool) -> Conduit a m a

Keep only values in the stream passing a given predicate.

Since 1.0.0

filterE :: (IsSequence seq, Monad m) => (Element seq -> Bool) -> Conduit seq m seq

Keep only elements in the chunked stream passing a given predicate.

Since 1.0.0

mapWhile :: Monad m => (a -> Maybe b) -> Conduit a m b

Map values as long as the result is Just.

Since 1.0.0

conduitVector

Arguments

:: (MonadBase base m, Vector v a, PrimMonad base)

=> Int

maximum allowed size

-> Conduit a m (v a)

Break up a stream of values into vectors of size n. The final vector may be smaller than n if the total number of values is not a strict multiple of n. No empty vectors will be yielded.

Since 1.0.0

scanl :: Monad m => (a -> b -> a) -> a -> Conduit b m a

Analog of scanl for lists.

Since 1.0.6

concatMapAccum :: Monad m => (a -> accum -> (accum, [b])) -> accum -> Conduit a m b

concatMap with an accumulator.

Since 1.0.0

intersperse :: Monad m => a -> Conduit a m a

Insert the given value between each two values in the stream.

Since 1.0.0

Monadic

mapM :: Monad m => (a -> m b) -> Conduit a m b

Apply a monadic transformation to all values in a stream.

If you do not need the transformed values, and instead just want the monadic side-effects of running the action, see mapM_.

Since 1.0.0

mapME :: (Monad m, Traversable f) => (a -> m b) -> Conduit (f a) m (f b)

Apply a monadic transformation to all elements in a chunked stream.

Since 1.0.0

omapME :: (Monad m, MonoTraversable mono) => (Element mono -> m (Element mono)) -> Conduit mono m mono

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.

Since 1.0.0

concatMapM :: (Monad m, MonoFoldable mono) => (a -> m mono) -> Conduit a m (Element mono)

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 that foldable value separately.

Generalizes concatMapM, mapMaybeM, and mapFoldableM.

Since 1.0.0

filterM :: Monad m => (a -> m Bool) -> Conduit a m a

Keep only values in the stream passing a given monadic predicate.

Since 1.0.0

filterME :: (Monad m, IsSequence seq) => (Element seq -> m Bool) -> Conduit seq m seq

Keep only elements in the chunked stream passing a given monadic predicate.

Since 1.0.0

iterM :: Monad m => (a -> m ()) -> Conduit a m a

Apply a monadic action on all values in a stream.

This Conduit can be used to perform a monadic side-effect for every value, whilst passing the value through the Conduit as-is.

 iterM f = mapM (\a -> f a >>= \() -> return a)

Since 1.0.0

scanlM :: Monad m => (a -> b -> m a) -> a -> Conduit b m a

Analog of scanl for lists, monadic.

Since 1.0.6

concatMapAccumM :: Monad m => (a -> accum -> m (accum, [b])) -> accum -> Conduit a m b

concatMapM with an accumulator.

Since 1.0.0

Textual

encodeUtf8 :: (Monad m, Utf8 text binary) => Conduit text m binary

Encode a stream of text as UTF8.

Since 1.0.0

decodeUtf8 :: MonadThrow m => Conduit ByteString m Text

Decode a stream of binary data as UTF8.

Since 1.0.0

line :: (Monad m, IsSequence seq, Element seq ~ Char) => ConduitM seq o m r -> ConduitM seq o m r

Stream in the entirety of a single line.

Like takeExactly, this will consume the entirety of the line regardless of the behavior of the inner Conduit.

Since 1.0.0

lineAscii :: (Monad m, IsSequence seq, Element seq ~ Word8) => ConduitM seq o m r -> ConduitM seq o m r

Same as line, but operates on ASCII/binary data.

Since 1.0.0

unlines :: (Monad m, IsSequence seq, Element seq ~ Char) => Conduit seq m seq

Insert a newline character after each incoming chunk of data.

Since 1.0.0

unlinesAscii :: (Monad m, IsSequence seq, Element seq ~ Word8) => Conduit seq m seq

Same as unlines, but operates on ASCII/binary data.

Since 1.0.0

linesUnbounded :: (Monad m, IsSequence seq, Element seq ~ Char) => Conduit seq m seq

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 unknownuntrusted input, this function is unsafe/, since it would allow an attacker to form lines of massive length and exhaust memory.

Since 1.0.0

linesUnboundedAscii :: (Monad m, IsSequence seq, Element seq ~ Word8) => Conduit seq m seq

Same as linesUnbounded, but for ASCII/binary data.

Since 1.0.0