Streamly.Internal.Data.Fold

See Streamly.Data.Fold for an overview and Streamly.Internal.Data.Fold.Types for design notes.

IMPORTANT: keep the signatures consistent with the folds in Streamly.Prelude

Fold Type

data Step s b

Represents the result of the step of a Fold. Partial returns an intermediate state of the fold, the fold step can be called again with the state or the driver can use extract on the state to get the result out. Done returns the final result and the fold cannot be driven further.

Pre-release

Constructors

Partial !s

Done !b

Instances

Bifunctor Step

first maps over Partial and second maps over Done.

Defined in Streamly.Internal.Data.Fold.Step

Methods

bimap :: (a -> b) -> (c -> d) -> Step a c -> Step b d

first :: (a -> b) -> Step a c -> Step b c

second :: (b -> c) -> Step a b -> Step a c

Functor (Step s)

fmap maps over Done.

fmap = second

Defined in Streamly.Internal.Data.Fold.Step

Methods

fmap :: (a -> b) -> Step s a -> Step s b

(<$) :: a -> Step s b -> Step s a

data Fold m a b

The type Fold m a b having constructor Fold step initial extract represents a fold over an input stream of values of type a to a final value of type b in Monad m.

The fold uses an intermediate state s as accumulator, the type s is internal to the specific fold definition. The initial value of the fold state s is returned by initial. The step function consumes an input and either returns the final result b if the fold is done or the next intermediate state (see Step). At any point the fold driver can extract the result from the intermediate state using the extract function.

NOTE: The constructor is not yet exposed via exposed modules, smart constructors are provided to create folds. If you think you need the constructor of this type please consider using the smart constructors in Streamly.Internal.Data.Fold instead.

since 0.8.0 (type changed)

Since: 0.7.0

Constructors

forall s. Fold (s -> a -> m (Step s b)) (m (Step s b)) (s -> m b)

Fold step initial extract

Instances

Functor m => Functor (Fold m a)

Maps a function on the output of the fold (the type b).

Defined in Streamly.Internal.Data.Fold.Type

Methods

fmap :: (a0 -> b) -> Fold m a a0 -> Fold m a b

(<$) :: a0 -> Fold m a b -> Fold m a a0

Constructors

foldl' :: Monad m => (b -> a -> b) -> b -> Fold m a b

Make a fold from a left fold style pure step function and initial value of the accumulator.

If your Fold returns only Partial (i.e. never returns a Done) then you can use foldl'* constructors.

A fold with an extract function can be expressed using fmap:

mkfoldlx :: Monad m => (s -> a -> s) -> s -> (s -> b) -> Fold m a b
mkfoldlx step initial extract = fmap extract (foldl' step initial)

See also: Streamly.Prelude.foldl'

Since: 0.8.0

foldlM' :: Monad m => (b -> a -> m b) -> m b -> Fold m a b

Make a fold from a left fold style monadic step function and initial value of the accumulator.

A fold with an extract function can be expressed using rmapM:

mkFoldlxM :: Functor m => (s -> a -> m s) -> m s -> (s -> m b) -> Fold m a b
mkFoldlxM step initial extract = rmapM extract (foldlM' step initial)

See also: Streamly.Prelude.foldlM'

Since: 0.8.0

foldl1' :: Monad m => (a -> a -> a) -> Fold m a (Maybe a)

Make a strict left fold, for non-empty streams, using first element as the starting value. Returns Nothing if the stream is empty.

See also: Streamly.Prelude.foldl1'

Pre-release

foldr :: Monad m => (a -> b -> b) -> b -> Fold m a b

Make a fold using a right fold style step function and a terminal value. It performs a strict right fold via a left fold using function composition. Note that this is strict fold, it can only be useful for constructing strict structures in memory. For reductions this will be very inefficient.

For example,

toList = foldr (:) []

See also: foldr

Since: 0.8.0

foldrM :: Monad m => (a -> b -> m b) -> m b -> Fold m a b

Like foldr but with a monadic step function.

For example,

toList = foldrM (\a xs -> return $ a : xs) (return [])

See also: foldrM

Pre-release

mkFold :: Monad m => (s -> a -> Step s b) -> Step s b -> (s -> b) -> Fold m a b

Make a terminating fold using a pure step function, a pure initial state and a pure state extraction function.

Pre-release

mkFold_ :: Monad m => (b -> a -> Step b b) -> Step b b -> Fold m a b

Similar to mkFold but the final state extracted is identical to the intermediate state.

mkFold_ step initial = mkFold step initial id

Pre-release

mkFoldM :: (s -> a -> m (Step s b)) -> m (Step s b) -> (s -> m b) -> Fold m a b

Make a terminating fold with an effectful step function and initial state, and a state extraction function.

mkFoldM = Fold

We can just use Fold but it is provided for completeness.

Pre-release

mkFoldM_ :: Monad m => (b -> a -> m (Step b b)) -> m (Step b b) -> Fold m a b

Similar to mkFoldM but the final state extracted is identical to the intermediate state.

mkFoldM_ step initial = mkFoldM step initial return

Pre-release

Folds

Identity

fromPure :: Applicative m => b -> Fold m a b

A fold that always yields a pure value without consuming any input.

Pre-release

fromEffect :: Applicative m => m b -> Fold m a b

A fold that always yields the result of an effectful action without consuming any input.

Pre-release

Accumulators

Semigroups and Monoids

sconcat :: (Monad m, Semigroup a) => a -> Fold m a a

Append the elements of an input stream to a provided starting value.

>>> Stream.fold (Fold.sconcat 10) (Stream.map Data.Monoid.Sum $ Stream.enumerateFromTo 1 10)
Sum {getSum = 65}

sconcat = Fold.foldl' (<>)

Since: 0.8.0

mconcat :: (Monad m, Monoid a) => Fold m a a

Fold an input stream consisting of monoidal elements using mappend and mempty.

>>> Stream.fold Fold.mconcat (Stream.map Data.Monoid.Sum $ Stream.enumerateFromTo 1 10)
Sum {getSum = 55}

mconcat = Fold.sconcat mempty

Since: 0.7.0

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

foldMap f = Fold.lmap f Fold.mconcat

Make a fold from a pure function that folds the output of the function using mappend and mempty.

>>> Stream.fold (Fold.foldMap Data.Monoid.Sum) $ Stream.enumerateFromTo 1 10
Sum {getSum = 55}

Since: 0.7.0

foldMapM :: (Monad m, Monoid b) => (a -> m b) -> Fold m a b

foldMapM f = Fold.lmapM f Fold.mconcat

Make a fold from a monadic function that folds the output of the function using mappend and mempty.

>>> Stream.fold (Fold.foldMapM (return . Data.Monoid.Sum)) $ Stream.enumerateFromTo 1 10
Sum {getSum = 55}

Since: 0.7.0

Reducers

drain :: Monad m => Fold m a ()

A fold that drains all its input, running the effects and discarding the results.

drain = drainBy (const (return ()))

Since: 0.7.0

drainBy :: Monad m => (a -> m b) -> Fold m a ()

drainBy f = lmapM f drain
drainBy = Fold.foldMapM (void . f)

Drain all input after passing it through a monadic function. This is the dual of mapM_ on stream producers.

See also: mapM_

Since: 0.7.0

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

Extract the last element of the input stream, if any.

last = fmap getLast $ Fold.foldMap (Last . Just)

Since: 0.7.0

length :: Monad m => Fold m a Int

Determine the length of the input stream.

length = fmap getSum $ Fold.foldMap (Sum . const  1)

Since: 0.7.0

mean :: (Monad m, Fractional a) => Fold m a a

Compute a numerically stable arithmetic mean of all elements in the input stream.

Since: 0.7.0

variance :: (Monad m, Fractional a) => Fold m a a

Compute a numerically stable (population) variance over all elements in the input stream.

Since: 0.7.0

stdDev :: (Monad m, Floating a) => Fold m a a

Compute a numerically stable (population) standard deviation over all elements in the input stream.

Since: 0.7.0

rollingHash :: (Monad m, Enum a) => Fold m a Int64

Compute an Int sized polynomial rolling hash of a stream.

rollingHash = Fold.rollingHashWithSalt defaultSalt

Since: 0.8.0

rollingHashWithSalt :: (Monad m, Enum a) => Int64 -> Fold m a Int64

Compute an Int sized polynomial rolling hash

H = salt * k ^ n + c1 * k ^ (n - 1) + c2 * k ^ (n - 2) + ... + cn * k ^ 0

Where c1, c2, cn are the elements in the input stream and k is a constant.

This hash is often used in Rabin-Karp string search algorithm.

See https://en.wikipedia.org/wiki/Rolling_hash

Since: 0.8.0

rollingHashFirstN :: (Monad m, Enum a) => Int -> Fold m a Int64

Compute an Int sized polynomial rolling hash of the first n elements of a stream.

rollingHashFirstN = Fold.take n Fold.rollingHash

Pre-release

Saturating Reducers

product terminates if it becomes 0. Other folds can theoretically saturate on bounded types, and therefore terminate, however, they will run forever on unbounded types like Integer/Double.

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

Determine the sum of all elements of a stream of numbers. Returns additive identity (0) when the stream is empty. Note that this is not numerically stable for floating point numbers.

sum = fmap getSum $ Fold.foldMap Sum

Since: 0.7.0

product :: (Monad m, Num a, Eq a) => Fold m a a

Determine the product of all elements of a stream of numbers. Returns multiplicative identity (1) when the stream is empty. The fold terminates when it encounters (0) in its input.

Compare with Fold.foldMap Product.

Since 0.8.0 (Added Eq constraint)

Since: 0.7.0

maximumBy :: Monad m => (a -> a -> Ordering) -> Fold m a (Maybe a)

Determine the maximum element in a stream using the supplied comparison function.

Since: 0.7.0

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

maximum = Fold.maximumBy compare

Determine the maximum element in a stream.

Compare with Fold.foldMap Max.

Since: 0.7.0

minimumBy :: Monad m => (a -> a -> Ordering) -> Fold m a (Maybe a)

Computes the minimum element with respect to the given comparison function

Since: 0.7.0

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

Determine the minimum element in a stream using the supplied comparison function.

minimum = minimumBy compare

Compare with Fold.foldMap Min.

Since: 0.7.0

Collectors

Avoid using these folds in scalable or performance critical applications, they buffer all the input in GC memory which can be detrimental to performance if the input is large.

toList :: Monad m => Fold m a [a]

Folds the input stream to a list.

Warning! working on large lists accumulated as buffers in memory could be very inefficient, consider using Streamly.Data.Array.Foreign instead.

toList = foldr (:) []

Since: 0.7.0

toListRev :: Monad m => Fold m a [a]

Buffers the input stream to a list in the reverse order of the input.

toListRev = Fold.foldl' (flip (:)) []

Warning! working on large lists accumulated as buffers in memory could be very inefficient, consider using Streamly.Array instead.

Since: 0.8.0

This is more efficient than toList. toList is exactly the same as reversing the list after toListRev.

toStream :: Monad m => Fold m a (SerialT n a)

A fold that buffers its input to a pure stream.

Warning! working on large streams accumulated as buffers in memory could be very inefficient, consider using Streamly.Data.Array instead.

>>> toStream = fmap SerialT Fold.toStreamK

Pre-release

toStreamRev :: Monad m => Fold m a (SerialT n a)

Buffers the input stream to a pure stream in the reverse order of the input.

>>> toStreamRev = fmap SerialT Fold.toStreamKRev

Warning! working on large streams accumulated as buffers in memory could be very inefficient, consider using Streamly.Data.Array instead.

Pre-release

Terminating Folds

drainN :: Monad m => Int -> Fold m a ()

A fold that drains the first n elements of its input, running the effects and discarding the results.

drainN n = Fold.take n Fold.drain

Pre-release

genericIndex :: (Integral i, Monad m) => i -> Fold m a (Maybe a)

Like index, except with a more general Integral argument

Pre-release

index :: Monad m => Int -> Fold m a (Maybe a)

Lookup the element at the given index.

See also: !!

Since: 0.7.0

head :: Monad m => Fold m a (Maybe a)

Extract the first element of the stream, if any.

Since: 0.7.0

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

Returns the first element that satisfies the given predicate.

Since: 0.7.0

lookup :: (Eq a, Monad m) => a -> Fold m (a, b) (Maybe b)

In a stream of (key-value) pairs (a, b), return the value b of the first pair where the key equals the given value a.

lookup = snd <$> Fold.find ((==) . fst)

Since: 0.7.0

findIndex :: Monad m => (a -> Bool) -> Fold m a (Maybe Int)

Returns the first index that satisfies the given predicate.

Since: 0.7.0

elemIndex :: (Eq a, Monad m) => a -> Fold m a (Maybe Int)

Returns the first index where a given value is found in the stream.

elemIndex a = Fold.findIndex (== a)

Since: 0.7.0

null :: Monad m => Fold m a Bool

Return True if the input stream is empty.

null = fmap isJust Fold.head

Since: 0.7.0

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

Return True if the given element is present in the stream.

elem a = Fold.any (== a)

Since: 0.7.0

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

Returns True if the given element is not present in the stream.

notElem a = Fold.all (/= a)

Since: 0.7.0

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

Returns True if all elements of a stream satisfy a predicate.

>>> Stream.fold (Fold.all (== 0)) $ Stream.fromList [1,0,1]
False

all p = Fold.lmap p Fold.and

Since: 0.7.0

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

Returns True if any of the elements of a stream satisfies a predicate.

>>> Stream.fold (Fold.any (== 0)) $ Stream.fromList [1,0,1]
True

any p = Fold.lmap p Fold.or

Since: 0.7.0

and :: Monad m => Fold m Bool Bool

Returns True if all elements are True, False otherwise

and = Fold.all (== True)

Since: 0.7.0

or :: Monad m => Fold m Bool Bool

Returns True if any element is True, False otherwise

or = Fold.any (== True)

Since: 0.7.0

Combinators

Utilities

with :: (Fold m (s, a) b -> Fold m a b) -> (((s, a) -> c) -> Fold m (s, a) b -> Fold m (s, a) b) -> ((s, a) -> c) -> Fold m a b -> Fold m a b

Change the predicate function of a Fold from a -> b to accept an additional state input (s, a) -> b. Convenient to filter with an addiitonal index or time input.

filterWithIndex = with indexed filter
filterWithAbsTime = with timestamped filter
filterWithRelTime = with timeIndexed filter

Pre-release

Transforming the Monad

hoist :: (forall x. m x -> n x) -> Fold m a b -> Fold n a b

Change the underlying monad of a fold

Pre-release

generally :: Monad m => Fold Identity a b -> Fold m a b

Adapt a pure fold to any monad

generally = Fold.hoist (return . runIdentity)

Pre-release

Mapping on output

rmapM :: Monad m => (b -> m c) -> Fold m a b -> Fold m a c

Map a monadic function on the output of a fold.

Since: 0.8.0

Mapping on Input

transform :: Monad m => Pipe m a b -> Fold m b c -> Fold m a c

Apply a transformation on a Fold using a Pipe.

Pre-release

lmap :: (a -> b) -> Fold m b r -> Fold m a r

lmap f fold maps the function f on the input of the fold.

>>> Stream.fold (Fold.lmap (\x -> x * x) Fold.sum) (Stream.enumerateFromTo 1 100)
338350

lmap = Fold.lmapM return

Since: 0.8.0

lmapM :: Monad m => (a -> m b) -> Fold m b r -> Fold m a r

lmapM f fold maps the monadic function f on the input of the fold.

Since: 0.8.0

scan :: Monad m => Fold m a b -> Fold m b c -> Fold m a c

Scan the input of a Fold to change it in a stateful manner using another Fold. Pre-release

postscan :: Monad m => Fold m a b -> Fold m b c -> Fold m a c

Postscan the input of a Fold to change it in a stateful manner using another Fold. Pre-release

indexed :: Fold m (Int, a) b -> Fold m a b

Pair each element of a fold input with its index, starting from index 0.

Unimplemented

Filtering

filter :: Monad m => (a -> Bool) -> Fold m a r -> Fold m a r

Include only those elements that pass a predicate.

>>> Stream.fold (Fold.filter (> 5) Fold.sum) $ Stream.fromList [1..10]
40

filter f = Fold.filterM (return . f)

Since: 0.8.0

filterM :: Monad m => (a -> m Bool) -> Fold m a r -> Fold m a r

Like filter but with a monadic predicate.

Since: 0.8.0

sampleFromthen :: Monad m => Int -> Int -> Fold m a b -> Fold m a b

sampleFromthen offset stride samples the element at offset index and then every element at strides of stride.

Unimplemented

Mapping Filters

catMaybes :: Monad m => Fold m a b -> Fold m (Maybe a) b

Modify a fold to receive a Maybe input, the Just values are unwrapped and sent to the original fold, Nothing values are discarded.

Since: 0.8.0

mapMaybe :: Monad m => (a -> Maybe b) -> Fold m b r -> Fold m a r

mapMaybe f fold maps a Maybe returning function f on the input of the fold, filters out Nothing elements, and return the values extracted from Just.

>>> f x = if even x then Just x else Nothing
>>> fld = Fold.mapMaybe f Fold.toList
>>> Stream.fold fld (Stream.enumerateFromTo 1 10)
[2,4,6,8,10]

Since: 0.8.0

Trimming

take :: Monad m => Int -> Fold m a b -> Fold m a b

Take at most n input elements and fold them using the supplied fold. A negative count is treated as 0.

>>> Stream.fold (Fold.take 2 Fold.toList) $ Stream.fromList [1..10]
[1,2]

Since: 0.8.0

takeEndBy :: Monad m => (a -> Bool) -> Fold m a b -> Fold m a b

Take the input, stop when the predicate succeeds taking the succeeding element as well.

>>> Stream.fold (Fold.takeEndBy (== '\n') Fold.toList) $ Stream.fromList "hello\nthere\n"
"hello\n"

>>> Stream.toList $ Stream.foldMany (Fold.takeEndBy (== '\n') Fold.toList) $ Stream.fromList "hello\nthere\n"
["hello\n","there\n"]

Stream.splitWithSuffix p f = Stream.foldMany (Fold.takeEndBy p f)

See splitWithSuffix for more details on splitting a stream using takeEndBy.

Since: 0.8.0

takeEndBy_ :: Monad m => (a -> Bool) -> Fold m a b -> Fold m a b

Like takeEndBy but drops the element on which the predicate succeeds.

>>> Stream.fold (Fold.takeEndBy_ (== '\n') Fold.toList) $ Stream.fromList "hello\nthere\n"
"hello"

>>> Stream.toList $ Stream.foldMany (Fold.takeEndBy_ (== '\n') Fold.toList) $ Stream.fromList "hello\nthere\n"
["hello","there"]

Stream.splitOnSuffix p f = Stream.foldMany (Fold.takeEndBy_ p f)

See splitOnSuffix for more details on splitting a stream using takeEndBy_.

Since: 0.8.0

Serial Append

serialWith :: Monad m => (a -> b -> c) -> Fold m x a -> Fold m x b -> Fold m x c

Sequential fold application. Apply two folds sequentially to an input stream. The input is provided to the first fold, when it is done - the remaining input is provided to the second fold. When the second fold is done or if the input stream is over, the outputs of the two folds are combined using the supplied function.

>>> f = Fold.serialWith (,) (Fold.take 8 Fold.toList) (Fold.takeEndBy (== '\n') Fold.toList)
>>> Stream.fold f $ Stream.fromList "header: hello\n"
("header: ","hello\n")

Note: This is dual to appending streams using serial.

Note: this implementation allows for stream fusion but has quadratic time complexity, because each composition adds a new branch that each subsequent fold's input element has to traverse, therefore, it cannot scale to a large number of compositions. After around 100 compositions the performance starts dipping rapidly compared to a CPS style implementation.

Time: O(n^2) where n is the number of compositions.

Since: 0.8.0

serial_ :: Monad m => Fold m x a -> Fold m x b -> Fold m x b

Same as applicative *>. Run two folds serially one after the other discarding the result of the first.

This was written in the hope that it might be faster than implementing it using serialWith, but the current benchmarks show that it has the same performance. So do not expose it unless some benchmark shows benefit.

splitAt :: Monad m => Int -> Fold m a b -> Fold m a c -> Fold m a (b, c)

splitAt n f1 f2 composes folds f1 and f2 such that first n elements of its input are consumed by fold f1 and the rest of the stream is consumed by fold f2.

>>> let splitAt_ n xs = Stream.fold (Fold.splitAt n Fold.toList Fold.toList) $ Stream.fromList xs

>>> splitAt_ 6 "Hello World!"
("Hello ","World!")

>>> splitAt_ (-1) [1,2,3]
([],[1,2,3])

>>> splitAt_ 0 [1,2,3]
([],[1,2,3])

>>> splitAt_ 1 [1,2,3]
([1],[2,3])

>>> splitAt_ 3 [1,2,3]
([1,2,3],[])

>>> splitAt_ 4 [1,2,3]
([1,2,3],[])

splitAt n f1 f2 = Fold.serialWith (,) (Fold.take n f1) f2

Internal

Parallel Distribution

teeWith :: Monad m => (a -> b -> c) -> Fold m x a -> Fold m x b -> Fold m x c

teeWith k f1 f2 distributes its input to both f1 and f2 until both of them terminate and combines their output using k.

>>> avg = Fold.teeWith (/) Fold.sum (fmap fromIntegral Fold.length)
>>> Stream.fold avg $ Stream.fromList [1.0..100.0]
50.5

teeWith k f1 f2 = fmap (uncurry k) ((Fold.tee f1 f2)

For applicative composition using this combinator see Streamly.Internal.Data.Fold.Tee.

See also: Streamly.Internal.Data.Fold.Tee

Since: 0.8.0

tee :: Monad m => Fold m a b -> Fold m a c -> Fold m a (b, c)

Distribute one copy of the stream to each fold and zip the results.

                |-------Fold m a b--------|
---stream m a---|                         |---m (b,c)
                |-------Fold m a c--------|

>>> Stream.fold (Fold.tee Fold.sum Fold.length) (Stream.enumerateFromTo 1.0 100.0)
(5050.0,100)

tee = teeWith (,)

Since: 0.7.0

teeWithFst :: Monad m => (b -> c -> d) -> Fold m a b -> Fold m a c -> Fold m a d

Like teeWith but terminates as soon as the first fold terminates.

Pre-release

teeWithMin :: Monad m => (b -> c -> d) -> Fold m a b -> Fold m a c -> Fold m a d

Like teeWith but terminates as soon as any one of the two folds terminates.

Pre-release

distribute :: Monad m => [Fold m a b] -> Fold m a [b]

Distribute one copy of the stream to each fold and collect the results in a container.

                |-------Fold m a b--------|
---stream m a---|                         |---m [b]
                |-------Fold m a b--------|
                |                         |
                           ...

>>> Stream.fold (Fold.distribute [Fold.sum, Fold.length]) (Stream.enumerateFromTo 1 5)
[15,5]

distribute = Prelude.foldr (Fold.teeWith (:)) (Fold.fromPure [])

This is the consumer side dual of the producer side sequence operation.

Stops when all the folds stop.

Since: 0.7.0

Parallel Alternative

shortest :: Monad m => Fold m x a -> Fold m x b -> Fold m x (Either a b)

Shortest alternative. Apply both folds in parallel but choose the result from the one which consumed least input i.e. take the shortest succeeding fold.

If both the folds finish at the same time or if the result is extracted before any of the folds could finish then the left one is taken.

Pre-release

longest :: Monad m => Fold m x a -> Fold m x b -> Fold m x (Either a b)

Longest alternative. Apply both folds in parallel but choose the result from the one which consumed more input i.e. take the longest succeeding fold.

If both the folds finish at the same time or if the result is extracted before any of the folds could finish then the left one is taken.

Pre-release

Partitioning

partitionByM :: Monad m => (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)

Partition the input over two folds using an Either partitioning predicate.

                                    |-------Fold b x--------|
-----stream m a --> (Either b c)----|                       |----(x,y)
                                    |-------Fold c y--------|

Send input to either fold randomly:

> import System.Random (randomIO)
> randomly a = randomIO >>= \x -> return $ if x then Left a else Right a
> Stream.fold (Fold.partitionByM randomly Fold.length Fold.length) (Stream.enumerateFromTo 1 100)
(59,41)

Send input to the two folds in a proportion of 2:1:

>>> :{
proportionately m n = do
 ref <- newIORef $ cycle $ concat [replicate m Left, replicate n Right]
 return $ \a -> do
     r <- readIORef ref
     writeIORef ref $ tail r
     return $ Prelude.head r a
:}

>>> :{
main = do
 f <- proportionately 2 1
 r <- Stream.fold (Fold.partitionByM f Fold.length Fold.length) (Stream.enumerateFromTo (1 :: Int) 100)
 print r
:}

>>> main
(67,33)

This is the consumer side dual of the producer side mergeBy operation.

When one fold is done, any input meant for it is ignored until the other fold is also done.

Stops when both the folds stop.

See also: partitionByFstM and partitionByMinM.

Pre-release

partitionByFstM :: Monad m => (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)

Similar to partitionByM but terminates when the first fold terminates.

partitionByMinM :: Monad m => (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)

Similar to partitionByM but terminates when any fold terminates.

partitionBy :: Monad m => (a -> Either b c) -> Fold m b x -> Fold m c y -> Fold m a (x, y)

Same as partitionByM but with a pure partition function.

Count even and odd numbers in a stream:

>>> :{
 let f = Fold.partitionBy (\n -> if even n then Left n else Right n)
                     (fmap (("Even " ++) . show) Fold.length)
                     (fmap (("Odd "  ++) . show) Fold.length)
  in Stream.fold f (Stream.enumerateFromTo 1 100)
:}
("Even 50","Odd 50")

Pre-release

partition :: Monad m => Fold m b x -> Fold m c y -> Fold m (Either b c) (x, y)

Compose two folds such that the combined fold accepts a stream of Either and routes the Left values to the first fold and Right values to the second fold.

partition = partitionBy id

Since: 0.7.0

Demultiplexing

Direct values in the input stream to different folds using an n-ary fold selector.

demux :: (Monad m, Ord k) => Map k (Fold m a b) -> Fold m (k, a) (Map k b)

Fold a stream of key value pairs using a map of specific folds for each key into a map from keys to the results of fold outputs of the corresponding values.

>>> import qualified Data.Map
>>> :{
 let table = Data.Map.fromList [("SUM", Fold.sum), ("PRODUCT", Fold.product)]
     input = Stream.fromList [("SUM",1),("PRODUCT",2),("SUM",3),("PRODUCT",4)]
  in Stream.fold (Fold.demux table) input
:}
fromList [("PRODUCT",8),("SUM",4)]

demux = demuxWith id

Pre-release

demuxWith :: (Monad m, Ord k) => (a -> (k, a')) -> Map k (Fold m a' b) -> Fold m a (Map k b)

Split the input stream based on a key field and fold each split using a specific fold collecting the results in a map from the keys to the results. Useful for cases like protocol handlers to handle different type of packets using different handlers.

                            |-------Fold m a b
-----stream m a-----Map-----|
                            |-------Fold m a b
                            |
                                      ...

Any input that does not map to a fold in the input Map is silently ignored.

demuxWith f kv = fmap fst $ demuxDefaultWith f kv drain

Pre-release

demuxDefault :: (Monad m, Ord k) => Map k (Fold m a b) -> Fold m (k, a) b -> Fold m (k, a) (Map k b, b)

demuxDefault = demuxDefaultWith id

Pre-release

demuxDefaultWith :: (Monad m, Ord k) => (a -> (k, a')) -> Map k (Fold m a' b) -> Fold m (k, a') c -> Fold m a (Map k b, c)

Like demuxWith but uses a default catchall fold to handle inputs which do not have a specific fold in the map to handle them.

If any fold in the map stops, inputs meant for that fold are sent to the catchall fold. If the catchall fold stops then inputs that do not match any fold are ignored.

Stops when all the folds, including the catchall fold, stop.

Pre-release

Classifying

In an input stream of key value pairs fold values for different keys in individual output buckets using the given fold.

classify :: (Monad m, Ord k) => Fold m a b -> Fold m (k, a) (Map k b)

Given an input stream of key value pairs and a fold for values, fold all the values belonging to each key. Useful for map/reduce, bucketizing the input in different bins or for generating histograms.

>>> :{
 let input = Stream.fromList [("ONE",1),("ONE",1.1),("TWO",2), ("TWO",2.2)]
  in Stream.fold (Fold.classify Fold.toList) input
:}
fromList [("ONE",[1.0,1.1]),("TWO",[2.0,2.2])]

Same as:

classify fld = Fold.classifyWith fst (lmap snd fld)

Pre-release

classifyWith :: (Monad m, Ord k) => (a -> k) -> Fold m a b -> Fold m a (Map k b)

Split the input stream based on a key field and fold each split using the given fold. Useful for map/reduce, bucketizing the input in different bins or for generating histograms.

>>> :{
 let input = Stream.fromList [("ONE",1),("ONE",1.1),("TWO",2), ("TWO",2.2)]
  in Stream.fold (Fold.classifyWith fst (Fold.map snd Fold.toList)) input
:}
fromList [("ONE",[1.0,1.1]),("TWO",[2.0,2.2])]

If the classifier fold stops for a particular key any further inputs in that bucket are ignored.

Stops: never

Pre-release

Unzipping

unzip :: Monad m => Fold m a x -> Fold m b y -> Fold m (a, b) (x, y)

Send the elements of tuples in a stream of tuples through two different folds.

                          |-------Fold m a x--------|
---------stream of (a,b)--|                         |----m (x,y)
                          |-------Fold m b y--------|

unzip = Fold.unzipWith id

This is the consumer side dual of the producer side zip operation.

Since: 0.7.0

unzipWith :: Monad m => (a -> (b, c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)

Split elements in the input stream into two parts using a pure splitter function, direct each part to a different fold and zip the results.

unzipWith f fld1 fld2 = Fold.lmap f (Fold.unzip fld1 fld2)

This fold terminates when both the input folds terminate.

Pre-release

unzipWithM :: Monad m => (a -> m (b, c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)

Like unzipWith but with a monadic splitter function.

unzipWithM k f1 f2 = lmapM k (unzip f1 f2)

Pre-release

unzipWithFstM :: Monad m => (a -> m (b, c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)

Similar to unzipWithM but terminates when the first fold terminates.

unzipWithMinM :: Monad m => (a -> m (b, c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)

Similar to unzipWithM but terminates when any fold terminates.

Zipping

zipWithM :: (a -> b -> m c) -> t m a -> Fold m c x -> Fold m b x

Zip a stream with the input of a fold using the supplied function.

Unimplemented

zip :: Monad m => t m a -> Fold m (a, b) x -> Fold m b x

Zip a stream with the input of a fold.

Unimplemented

Splitting

many :: Monad m => Fold m a b -> Fold m b c -> Fold m a c

Collect zero or more applications of a fold. many split collect applies the split fold repeatedly on the input stream and accumulates zero or more fold results using collect.

>>> two = Fold.take 2 Fold.toList
>>> twos = Fold.many two Fold.toList
>>> Stream.fold twos $ Stream.fromList [1..10]
[[1,2],[3,4],[5,6],[7,8],[9,10]]

Stops when collect stops.

See also: concatMap, foldMany

Since: 0.8.0

chunksOf :: Monad m => Int -> Fold m a b -> Fold m b c -> Fold m a c

chunksOf n split collect repeatedly applies the split fold to chunks of n items in the input stream and supplies the result to the collect fold.

>>> twos = Fold.chunksOf 2 Fold.toList Fold.toList
>>> Stream.fold twos $ Stream.fromList [1..10]
[[1,2],[3,4],[5,6],[7,8],[9,10]]

chunksOf n split = many (take n split)

Stops when collect stops.

Since: 0.8.0

chunksBetween :: Int -> Int -> Fold m a b -> Fold m b c -> Fold m a c

Group the input stream into groups of elements between low and high. Collection starts in chunks of low and then keeps doubling until we reach high. Each chunk is folded using the provided fold function.

This could be useful, for example, when we are folding a stream of unknown size to a stream of arrays and we want to minimize the number of allocations.

NOTE: this would be an application of "many" using a terminating fold.

Unimplemented

Nesting

concatSequence :: Fold m b c -> t (Fold m a b) -> Fold m a c

concatSequence f t applies folds from stream t sequentially and collects the results using the fold f.

Unimplemented

concatMap :: Monad m => (b -> Fold m a c) -> Fold m a b -> Fold m a c

Map a Fold returning function on the result of a Fold and run the returned fold. This operation can be used to express data dependencies between fold operations.

Let's say the first element in the stream is a count of the following elements that we have to add, then:

>>> import Data.Maybe (fromJust)
>>> count = fmap fromJust Fold.head
>>> total n = Fold.take n Fold.sum
>>> Stream.fold (Fold.concatMap total count) $ Stream.fromList [10,9..1]
45

Time: O(n^2) where n is the number of compositions.

See also: foldIterateM

Since: 0.8.0

Running A Fold

Normally you would run a fold to completion by supplying it a stream, e.g. using fold. However, you could also run a fold partially by using duplicate on it and then running it with a stream. Alternatively, initialize, snoc and finish can be used to run a fold incrementally, however, that may not be the most efficient way to run a fold.

initialize :: Monad m => Fold m a b -> m (Fold m a b)

Run the initialization effect of a fold. The returned fold would use the value returned by this effect as its initial value.

Pre-release

snoc :: Monad m => Fold m a b -> a -> m (Fold m a b)

Append a singleton value to the fold.

>>> import qualified Data.Foldable as Foldable
>>> Foldable.foldlM Fold.snoc Fold.toList [1..3] >>= Fold.finish
[1,2,3]

Compare with duplicate which allows appending a stream to the fold.

Pre-release

duplicate :: Monad m => Fold m a b -> Fold m a (Fold m a b)

duplicate provides the ability to run a fold in parts. The duplicated fold consumes the input and returns the same fold as output instead of returning the final result, the returned fold can be run later to consume more input.

We can append a stream to a fold as follows:

>>> :{
foldAppend :: Monad m => Fold m a b -> SerialT m a -> m (Fold m a b)
foldAppend f = Stream.fold (Fold.duplicate f)
:}

>>> :{
do
 sum1 <- foldAppend Fold.sum (Stream.enumerateFromTo 1 10)
 sum2 <- foldAppend sum1 (Stream.enumerateFromTo 11 20)
 Stream.fold sum2 (Stream.enumerateFromTo 21 30)
:}
465

duplicate essentially appends a stream to the fold without finishing the fold. Compare with snoc which appends a singleton value to the fold.

Pre-release

finish :: Monad m => Fold m a b -> m b

Finish the fold to extract the current value of the fold.

>>> Fold.finish Fold.toList
[]

Pre-release

Deprecated

sequence :: Monad m => Fold m a (m b) -> Fold m a b

Deprecated: Use "rmapM id" instead

Flatten the monadic output of a fold to pure output.

Since: 0.7.0

mapM :: Monad m => (b -> m c) -> Fold m a b -> Fold m a c

Deprecated: Use rmapM instead

Map a monadic function on the output of a fold.

Since: 0.7.0