2-- Module : Streamly.Internal.Data.Fold
3-- Copyright : (c) 2019 Composewell Technologies
4-- (c) 2013 Gabriel Gonzalez
6-- Maintainer : streamly@composewell.com
7-- Stability : experimental
10-- See "Streamly.Data.Fold" for an overview and
11-- "Streamly.Internal.Data.Fold.Types" for design notes.
13-- IMPORTANT: keep the signatures consistent with the folds in Streamly.Prelude
15module Streamly.Internal.Data.Fold
22 -- | Which constructor to use?
24 -- * @foldl*@: If the fold never terminates i.e. does not use the 'Done'
25 -- constructor otherwise use the @mkFold@ variants.
26 -- * @*M@: Use the @M@ suffix variants if any of the step, initial, or
27 -- extract function is monadic, otherwise use the pure variants.
28 -- * @*_@: Use the @_@ suffix variants if the extract function is 'id' or
47 -- *** Semigroups and Monoids
70 -- *** Saturating Reducers
71 -- | 'product' terminates if it becomes 0. Other folds can theoretically
72 -- saturate on bounded types, and therefore terminate, however, they will
73 -- run forever on unbounded types like Integer/Double.
82 -- | Avoid using these folds in scalable or performance critical
83 -- applications, they buffer all the input in GC memory which can be
84 -- detrimental to performance if the input is large.
92 -- ** Terminating Folds
117 -- ** Transforming the Monad
121 -- ** Mapping on output
124 -- ** Mapping on Input
147 -- ** Mapping Filters
152 -- ** Scanning Filters
158 -- | Insertion adds more elements to the stream.
186 -- , splitIn -- sessionN
188 -- ** Parallel Distribution
199 -- These two can be expressed using lmap/lmapM and unzip
205 -- ** Parallel Alternative
217 -- | Direct values in the input stream to different folds using an n-ary
218 -- fold selector. 'demux' is a generalization of 'classify' (and
219 -- 'partition') where each key of the classifier can use a different fold.
226 -- TODO: These can be implemented using the above operations
227 -- , demuxWithSel -- Stop when the fold for the specified key stops
228 -- , demuxWithMin -- Stop when any of the folds stop
229 -- , demuxWithAll -- Stop when all the folds stop (run once)
232 -- | In an input stream of key value pairs fold values for different keys
233 -- in individual output buckets using the given fold. 'classify' is a
234 -- special case of 'demux' where all the branches of the demultiplexer use
237 -- Different types of maps can be used with these combinators via the IsMap
238 -- type class. Hashmap performs better when there are more collisions, trie
239 -- Map performs better otherwise. Trie has an advantage of sorting the keys
240 -- at the same time. For example if we want to store a dictionary of words
241 -- and their meanings then trie Map would be better if we also want to
242 -- display them in sorted order.
262 -- | Normally you would run a fold to completion by supplying it a stream,
263 -- e.g. using 'Stream.fold'. However, you could also run a fold partially
264 -- by using 'duplicate' on it and then running it with a stream.
265 -- Alternatively, 'initialize', 'snoc' and 'finish' can be used to run a
266 -- fold incrementally, however, that may not be the most efficient way to
283import Control.Monad.IO.Class (MonadIO(..))
284import Data.Bifunctor (first)
285import Data.Either (isLeft, isRight, fromLeft, fromRight)
286import Data.Functor.Identity (Identity(..))
287import Data.Int (Int64)
288import Data.IORef (newIORef, readIORef, writeIORef)
289import Data.Map.Strict (Map)
290import Data.Maybe (isJust, fromJust)
291import Foreign.Storable (Storable)
292import Streamly.Internal.Data.IsMap (IsMap(..))
293import Streamly.Internal.Data.Pipe.Type (Pipe (..), PipeState(..))
294import Streamly.Internal.Data.Unfold.Type (Unfold(..))
295import Streamly.Internal.Data.Tuple.Strict (Tuple'(..), Tuple3'(..))
296import Streamly.Internal.Data.Stream.Serial (SerialT(..))
298import qualified Data.IntSet as IntSet
299import qualified Data.Set as Set
300import qualified Streamly.Internal.Data.IsMap as IsMap
301import qualified Prelude
302import qualified Streamly.Internal.Data.Array.Foreign.Mut.Type as MA
303import qualified Streamly.Internal.Data.Pipe.Type as Pipe
304import qualified Streamly.Internal.Data.Stream.StreamD.Type as StreamD
307 ( filter, foldl1, drop, dropWhile, take, takeWhile, zipWith
308 , foldl, foldr, map, mapM_, sequence, all, any, sum, product, elem
309 , notElem, maximum, minimum, head, last, tail, length, null
310 , reverse, iterate, init, and, or, lookup, (!!)
311 , scanl, scanl1, replicate, concatMap, mconcat, foldMap, unzip
312 , span, splitAt, break, mapM, zip)
313import Streamly.Internal.Data.Fold.Type
317-- >>> :set -package streamly
318-- >>> import Prelude hiding (break, map, span, splitAt)
319-- >>> import qualified Streamly.Prelude as Stream
320-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream (parse, foldMany)
321-- >>> import qualified Streamly.Data.Fold as Fold
322-- >>> import qualified Streamly.Internal.Data.Fold as Fold
323-- >>> import qualified Streamly.Internal.Data.Fold.Type as Fold
324-- >>> import qualified Streamly.Internal.Data.Parser as Parser
325-- >>> import Streamly.Internal.Data.Stream.Serial (SerialT(..))
326-- >>> import Data.IORef (newIORef, readIORef, writeIORef)
327-- >>> import qualified Streamly.Internal.Data.Array.Foreign.Mut.Type as MA
329------------------------------------------------------------------------------
331------------------------------------------------------------------------------
333-- | Change the underlying monad of a fold
336hoist :: (forall x. m x -> n x) -> Fold m a b -> Fold n a b
337hoist f (Fold step initial extract) =
338 Fold (\x a -> f $ step x a) (f initial) (f . extract)
340-- | Adapt a pure fold to any monad
342-- > generally = Fold.hoist (return . runIdentity)
345generally :: Monad m => Fold Identity a b -> Fold m a b
346generally = hoist (return . runIdentity)
348------------------------------------------------------------------------------
349-- Transformations on fold inputs
350------------------------------------------------------------------------------
352-- | Flatten the monadic output of a fold to pure output.
355{-# DEPRECATED sequence "Use \"rmapM id\" instead" #-}
356{-# INLINE sequence #-}
357sequence :: Monad m => Fold m a (m b) -> Fold m a b
360-- | Map a monadic function on the output of a fold.
363{-# DEPRECATED mapM "Use rmapM instead" #-}
365mapM :: Monad m => (b -> m c) -> Fold m a b -> Fold m a c
368-- | @mapMaybe f fold@ maps a 'Maybe' returning function @f@ on the input of
369-- the fold, filters out 'Nothing' elements, and return the values extracted
372-- >>> f x = if even x then Just x else Nothing
373-- >>> fld = Fold.mapMaybe f Fold.toList
374-- >>> Stream.fold fld (Stream.enumerateFromTo 1 10)
378{-# INLINE mapMaybe #-}
379mapMaybe :: (Monad m) => (a -> Maybe b) -> Fold m b r -> Fold m a r
380mapMaybe f = lmap f . filter isJust . lmap fromJust
382------------------------------------------------------------------------------
383-- Transformations on fold inputs
384------------------------------------------------------------------------------
388-- | Apply a transformation on a 'Fold' using a 'Pipe'.
391{-# INLINE transform #-}
392transform :: Monad m => Pipe m a b -> Fold m b c -> Fold m a c
393transform (Pipe pstep1 pstep2 pinitial) (Fold fstep finitial fextract) =
394 Fold step initial extract
398 initial = first (Tuple' pinitial) <$> finitial
400 step (Tuple' ps fs) x = do
407 go acc (Pipe.Yield b (Consume ps')) = do
411 Partial s -> Partial $ Tuple' ps' s
413 go acc (Pipe.Yield b (Produce ps')) = do
418 Done b2 -> return $ Done b2
419 go acc (Pipe.Continue (Consume ps')) =
420 return $ Partial $ Tuple' ps' acc
421 go acc (Pipe.Continue (Produce ps')) = do
425 extract (Tuple' _ fs) = fextract fs
427{-# INLINE scanWith #-}
428scanWith :: Monad m => Bool -> Fold m a b -> Fold m b c -> Fold m a c
429scanWith isMany (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
430 Fold step initial extract
434 {-# INLINE runStep #-}
435 runStep actionL sR = do
443 then runStep initialL sR1
444 else Done <$> extractR sR1
445 Done bR -> return $ Done bR
451 Partial sR1 -> Partial (sL, sR1)
457 Partial sR -> runStep initialL sR
458 Done b -> return $ Done b
460 step (sL, sR) x = runStep (stepL sL x) sR
462 extract = extractR . snd
464-- | Scan the input of a 'Fold' to change it in a stateful manner using another
465-- 'Fold'. The scan stops as soon as the fold terminates.
469scan :: Monad m => Fold m a b -> Fold m b c -> Fold m a c
472-- XXX This does not fuse beacuse of the recursive step. Need to investigate.
474-- | Scan the input of a 'Fold' to change it in a stateful manner using another
475-- 'Fold'. The scan restarts with a fresh state if the fold terminates.
478{-# INLINE scanMany #-}
479scanMany :: Monad m => Fold m a b -> Fold m b c -> Fold m a c
480scanMany = scanWith True
482-- | Postscan the input of a 'Fold' to change it in a stateful manner using
485{-# INLINE postscan #-}
486postscan :: Monad m => Fold m a b -> Fold m b c -> Fold m a c
487postscan (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
488 Fold step initial extract
492 {-# INLINE runStep #-}
493 runStep actionL sR = do
499 Partial sR1 -> Done <$> extractR sR1
500 Done bR -> return $ Done bR
506 Partial sR1 -> Partial (sL, sR1)
515 Done _ -> Done <$> extractR sR
516 Partial sL -> return $ Partial (sL, sR)
517 Done b -> return $ Done b
519 step (sL, sR) x = runStep (stepL sL x) sR
521 extract = extractR . snd
523------------------------------------------------------------------------------
525------------------------------------------------------------------------------
527-- | Convert a predicate into a filtering fold.
529-- >>> f = Fold.foldFilter (Fold.satisfy (> 5)) Fold.sum
530-- >>> Stream.fold f $ Stream.fromList [1..10]
534{-# INLINE satisfy #-}
535satisfy :: Monad m => (a -> Bool) -> Fold m a (Maybe a)
536satisfy f = Fold step (return $ Partial ()) (const (return Nothing))
540 step () a = return $ Done $ if f a then Just a else Nothing
542-- | Use a 'Maybe' returning fold as a filtering scan.
544-- >>> f = Fold.foldFilter (Fold.satisfy (> 5)) Fold.sum
545-- >>> Stream.fold f $ Stream.fromList [1..10]
548-- The above snippet is equivalent to:
550-- >>> f = Fold.filter (> 5) Fold.sum
551-- >>> Stream.fold f $ Stream.fromList [1..10]
555{-# INLINE foldFilter #-}
556foldFilter :: Monad m => Fold m a (Maybe b) -> Fold m b c -> Fold m a c
557foldFilter f1 f2 = many f1 (catMaybes f2)
559------------------------------------------------------------------------------
561------------------------------------------------------------------------------
563------------------------------------------------------------------------------
565------------------------------------------------------------------------------
568-- > drainBy f = lmapM f drain
569-- > drainBy = Fold.foldMapM (void . f)
571-- Drain all input after passing it through a monadic function. This is the
572-- dual of mapM_ on stream producers.
574-- See also: 'Streamly.Prelude.mapM_'
577{-# INLINE drainBy #-}
578drainBy :: Monad m => (a -> m b) -> Fold m a ()
579drainBy f = lmapM f drain
581-- | Extract the last element of the input stream, if any.
583-- > last = fmap getLast $ Fold.foldMap (Last . Just)
587last :: Monad m => Fold m a (Maybe a)
588last = foldl1' (\_ x -> x)
590------------------------------------------------------------------------------
592------------------------------------------------------------------------------
594-- | Like 'length', except with a more general 'Num' return value
596-- > genericLength = fmap getSum $ foldMap (Sum . const 1)
599{-# INLINE genericLength #-}
600genericLength :: (Monad m, Num b) => Fold m a b
601genericLength = foldl' (\n _ -> n + 1) 0
603-- | Determine the length of the input stream.
605-- > length = fmap getSum $ Fold.foldMap (Sum . const 1)
609length :: Monad m => Fold m a Int
610length = genericLength
613-- XXX Add a countDistinct window fold
614-- XXX Add a bloom filter fold
616-- | Count non-duplicate elements in the stream.
618-- Equivalent to using 'nub' followed by 'length' on a stream.
620-- The memory used is proportional to the number of distinct elements in the
621-- stream, to guard against using too much memory use it as a scan and
622-- terminate if the count reaches more than a threshold.
624-- /Space/: \(\mathcal{O}(n)\)
628{-# INLINE countDistinct #-}
629countDistinct :: (Monad m, Ord a) => Fold m a Int
630countDistinct = fmap (\(Tuple' _ n) -> n) $ foldl' step initial
634 initial = Tuple' Set.empty 0
636 step (Tuple' set n) x = do
642 in Tuple' (Set.insert x set) cnt
644-- | Like 'countDistinct' but specialized to a stream of 'Int', for better
648{-# INLINE countDistinctInt #-}
649countDistinctInt :: Monad m => Fold m Int Int
650countDistinctInt = fmap (\(Tuple' _ n) -> n) $ foldl' step initial
654 initial = Tuple' IntSet.empty 0
656 step (Tuple' set n) x = do
657 if IntSet.member x set
662 in Tuple' (IntSet.insert x set) cnt
664-- | Determine the sum of all elements of a stream of numbers. Returns additive
665-- identity (@0@) when the stream is empty. Note that this is not numerically
666-- stable for floating point numbers.
668-- > sum = fmap getSum $ Fold.foldMap Sum
672sum :: (Monad m, Num a) => Fold m a a
675-- | Determine the product of all elements of a stream of numbers. Returns
676-- multiplicative identity (@1@) when the stream is empty. The fold terminates
677-- when it encounters (@0@) in its input.
679-- Compare with @Fold.foldMap Product@.
682-- /Since 0.8.0 (Added 'Eq' constraint)/
683{-# INLINE product #-}
684product :: (Monad m, Num a, Eq a) => Fold m a a
685product = mkFold_ step (Partial 1)
694------------------------------------------------------------------------------
696------------------------------------------------------------------------------
698-- | Determine the maximum element in a stream using the supplied comparison
702{-# INLINE maximumBy #-}
703maximumBy :: Monad m => (a -> a -> Ordering) -> Fold m a (Maybe a)
704maximumBy cmp = foldl1' max'
715-- maximum = Fold.maximumBy compare
718-- Determine the maximum element in a stream.
720-- Compare with @Fold.foldMap Max@.
723{-# INLINE maximum #-}
724maximum :: (Monad m, Ord a) => Fold m a (Maybe a)
727-- | Computes the minimum element with respect to the given comparison function
730{-# INLINE minimumBy #-}
731minimumBy :: Monad m => (a -> a -> Ordering) -> Fold m a (Maybe a)
732minimumBy cmp = foldl1' min'
741-- | Determine the minimum element in a stream using the supplied comparison
745-- minimum = 'minimumBy' compare
748-- Compare with @Fold.foldMap Min@.
751{-# INLINE minimum #-}
752minimum :: (Monad m, Ord a) => Fold m a (Maybe a)
755------------------------------------------------------------------------------
756-- To Summary (Statistical)
757------------------------------------------------------------------------------
759-- | Compute a numerically stable arithmetic mean of all elements in the input
764mean :: (Monad m, Fractional a) => Fold m a a
765mean = fmap done $ foldl' step begin
771 step (Tuple' x n) y =
773 in Tuple' (x + (y - x) / n1) n1
775 done (Tuple' x _) = x
777-- | Compute a numerically stable (population) variance over all elements in
781{-# INLINE variance #-}
782variance :: (Monad m, Fractional a) => Fold m a a
783variance = fmap done $ foldl' step begin
787 begin = Tuple3' 0 0 0
789 step (Tuple3' n mean_ m2) x = Tuple3' n' mean' m2'
794 mean' = (n * mean_ + x) / (n + 1)
796 m2' = m2 + delta * delta * n / (n + 1)
798 done (Tuple3' n _ m2) = m2 / n
800-- | Compute a numerically stable (population) standard deviation over all
801-- elements in the input stream.
805stdDev :: (Monad m, Floating a) => Fold m a a
806stdDev = sqrt <$> variance
808-- | Compute an 'Int' sized polynomial rolling hash
810-- > H = salt * k ^ n + c1 * k ^ (n - 1) + c2 * k ^ (n - 2) + ... + cn * k ^ 0
812-- Where @c1@, @c2@, @cn@ are the elements in the input stream and @k@ is a
815-- This hash is often used in Rabin-Karp string search algorithm.
817-- See https://en.wikipedia.org/wiki/Rolling_hash
820{-# INLINE rollingHashWithSalt #-}
821rollingHashWithSalt :: (Monad m, Enum a) => Int64 -> Fold m a Int64
822rollingHashWithSalt = foldl' step
826 k = 2891336453 :: Int64
828 step cksum a = cksum * k + fromIntegral (fromEnum a)
830-- | A default salt used in the implementation of 'rollingHash'.
831{-# INLINE defaultSalt #-}
833defaultSalt = -2578643520546668380
835-- | Compute an 'Int' sized polynomial rolling hash of a stream.
837-- > rollingHash = Fold.rollingHashWithSalt defaultSalt
840{-# INLINE rollingHash #-}
841rollingHash :: (Monad m, Enum a) => Fold m a Int64
842rollingHash = rollingHashWithSalt defaultSalt
844-- | Compute an 'Int' sized polynomial rolling hash of the first n elements of
847-- > rollingHashFirstN = Fold.take n Fold.rollingHash
850{-# INLINE rollingHashFirstN #-}
851rollingHashFirstN :: (Monad m, Enum a) => Int -> Fold m a Int64
852rollingHashFirstN n = take n rollingHash
854-- | Apply a function on every two successive elements of a stream. The first
855-- argument of the map function is the previous element and the second argument
856-- is the current element. When processing the very first element in the
857-- stream, the previous element is 'Nothing'.
861{-# INLINE rollingMapM #-}
862rollingMapM :: Monad m => (Maybe a -> a -> m b) -> Fold m a b
863rollingMapM f = Fold step initial extract
867 -- XXX We need just a postscan. We do not need an initial result here.
868 -- Or we can supply a default initial result as an argument to rollingMapM.
869 initial = return $ Partial (Nothing, error "Empty stream")
871 step (prev, _) cur = do
873 return $ Partial (Just cur, x)
875 extract = return . snd
877------------------------------------------------------------------------------
878-- Monoidal left folds
879------------------------------------------------------------------------------
881-- | Append the elements of an input stream to a provided starting value.
883-- >>> Stream.fold (Fold.sconcat 10) (Stream.map Data.Monoid.Sum $ Stream.enumerateFromTo 1 10)
887-- sconcat = Fold.foldl' (<>)
891{-# INLINE sconcat #-}
892sconcat :: (Monad m, Semigroup a) => a -> Fold m a a
895-- | Fold an input stream consisting of monoidal elements using 'mappend'
898-- >>> Stream.fold Fold.mconcat (Stream.map Data.Monoid.Sum $ Stream.enumerateFromTo 1 10)
901-- > mconcat = Fold.sconcat mempty
904{-# INLINE mconcat #-}
907 , Monoid a) => Fold m a a
908mconcat = sconcat mempty
911-- > foldMap f = Fold.lmap f Fold.mconcat
913-- Make a fold from a pure function that folds the output of the function
914-- using 'mappend' and 'mempty'.
916-- >>> Stream.fold (Fold.foldMap Data.Monoid.Sum) $ Stream.enumerateFromTo 1 10
920{-# INLINE foldMap #-}
921foldMap :: (Monad m, Monoid b
922 ) => (a -> b) -> Fold m a b
923foldMap f = lmap f mconcat
926-- > foldMapM f = Fold.lmapM f Fold.mconcat
928-- Make a fold from a monadic function that folds the output of the function
929-- using 'mappend' and 'mempty'.
931-- >>> Stream.fold (Fold.foldMapM (return . Data.Monoid.Sum)) $ Stream.enumerateFromTo 1 10
935{-# INLINE foldMapM #-}
936foldMapM :: (Monad m, Monoid b) => (a -> m b) -> Fold m a b
937foldMapM act = foldlM' step (pure mempty)
943 return $! mappend m m'
945------------------------------------------------------------------------------
947------------------------------------------------------------------------------
950-- This is more efficient than 'Streamly.Internal.Data.Fold.toList'. toList is
951-- exactly the same as reversing the list after 'toListRev'.
953-- | Buffers the input stream to a list in the reverse order of the input.
955-- > toListRev = Fold.foldl' (flip (:)) []
957-- /Warning!/ working on large lists accumulated as buffers in memory could be
958-- very inefficient, consider using "Streamly.Array" instead.
962-- xn : ... : x2 : x1 : []
963{-# INLINE toListRev #-}
964toListRev :: Monad m => Fold m a [a]
965toListRev = foldl' (flip (:)) []
967------------------------------------------------------------------------------
969------------------------------------------------------------------------------
971-- | A fold that drains the first n elements of its input, running the effects
972-- and discarding the results.
974-- > drainN n = Fold.take n Fold.drain
978drainN :: Monad m => Int -> Fold m a ()
979drainN n = take n drain
981------------------------------------------------------------------------------
983------------------------------------------------------------------------------
985-- | Like 'index', except with a more general 'Integral' argument
988{-# INLINE genericIndex #-}
989genericIndex :: (Integral i, Monad m) => i -> Fold m a (Maybe a)
990genericIndex i = mkFold step (Partial 0) (const Nothing)
999-- | Lookup the element at the given index.
1001-- See also: 'Streamly.Prelude.!!'
1005index :: Monad m => Int -> Fold m a (Maybe a)
1008-- | Extract the first element of the stream, if any.
1012head :: Monad m => Fold m a (Maybe a)
1013head = mkFold_ (const (Done . Just)) (Partial Nothing)
1015-- | Returns the first element that satisfies the given predicate.
1019find :: Monad m => (a -> Bool) -> Fold m a (Maybe a)
1020find predicate = mkFold step (Partial ()) (const Nothing)
1029-- | In a stream of (key-value) pairs @(a, b)@, return the value @b@ of the
1030-- first pair where the key equals the given value @a@.
1032-- > lookup = snd <$> Fold.find ((==) . fst)
1035{-# INLINE lookup #-}
1036lookup :: (Eq a, Monad m) => a -> Fold m (a,b) (Maybe b)
1037lookup a0 = mkFold step (Partial ()) (const Nothing)
1046-- | Returns the first index that satisfies the given predicate.
1049{-# INLINE findIndex #-}
1050findIndex :: Monad m => (a -> Bool) -> Fold m a (Maybe Int)
1051findIndex predicate = mkFold step (Partial 0) (const Nothing)
1058 else Partial (i + 1)
1060-- XXX If we have a Continue contructor we won't need a Maybe type for the
1061-- scan. The fold will have the ability to skip producing output.
1063-- | Returns all indices that satisfy the given predicate.
1065{-# INLINE findIndices #-}
1066findIndices :: Monad m => (a -> Bool) -> Fold m a (Maybe Int)
1067findIndices predicate =
1068 fmap (either (const Nothing) Just) $ foldl' step (Left (-1))
1074 then Right (either id id i + 1)
1075 else Left (either id id i + 1)
1077-- | Returns the first index where a given value is found in the stream.
1079-- > elemIndex a = Fold.findIndex (== a)
1082{-# INLINE elemIndex #-}
1083elemIndex :: (Eq a, Monad m) => a -> Fold m a (Maybe Int)
1084elemIndex a = findIndex (a ==)
1086------------------------------------------------------------------------------
1088------------------------------------------------------------------------------
1090-- | Return 'True' if the input stream is empty.
1092-- > null = fmap isJust Fold.head
1096null :: Monad m => Fold m a Bool
1097null = mkFold (\() _ -> Done False) (Partial ()) (const True)
1099-- | Returns 'True' if any of the elements of a stream satisfies a predicate.
1101-- >>> Stream.fold (Fold.any (== 0)) $ Stream.fromList [1,0,1]
1104-- > any p = Fold.lmap p Fold.or
1108any :: Monad m => (a -> Bool) -> Fold m a Bool
1109any predicate = mkFold_ step initial
1113 initial = Partial False
1120-- | Return 'True' if the given element is present in the stream.
1122-- > elem a = Fold.any (== a)
1126elem :: (Eq a, Monad m) => a -> Fold m a Bool
1129-- | Returns 'True' if all elements of a stream satisfy a predicate.
1131-- >>> Stream.fold (Fold.all (== 0)) $ Stream.fromList [1,0,1]
1134-- > all p = Fold.lmap p Fold.and
1138all :: Monad m => (a -> Bool) -> Fold m a Bool
1139all predicate = mkFold_ step initial
1143 initial = Partial True
1150-- | Returns 'True' if the given element is not present in the stream.
1152-- > notElem a = Fold.all (/= a)
1155{-# INLINE notElem #-}
1156notElem :: (Eq a, Monad m) => a -> Fold m a Bool
1157notElem a = all (a /=)
1159-- | Returns 'True' if all elements are 'True', 'False' otherwise
1161-- > and = Fold.all (== True)
1165and :: Monad m => Fold m Bool Bool
1168-- | Returns 'True' if any element is 'True', 'False' otherwise
1170-- > or = Fold.any (== True)
1174or :: Monad m => Fold m Bool Bool
1177------------------------------------------------------------------------------
1178-- Grouping/Splitting
1179------------------------------------------------------------------------------
1181------------------------------------------------------------------------------
1182-- Grouping without looking at elements
1183------------------------------------------------------------------------------
1185------------------------------------------------------------------------------
1187------------------------------------------------------------------------------
1189-- | @splitAt n f1 f2@ composes folds @f1@ and @f2@ such that first @n@
1190-- elements of its input are consumed by fold @f1@ and the rest of the stream
1191-- is consumed by fold @f2@.
1193-- >>> let splitAt_ n xs = Stream.fold (Fold.splitAt n Fold.toList Fold.toList) $ Stream.fromList xs
1195-- >>> splitAt_ 6 "Hello World!"
1196-- ("Hello ","World!")
1198-- >>> splitAt_ (-1) [1,2,3]
1201-- >>> splitAt_ 0 [1,2,3]
1204-- >>> splitAt_ 1 [1,2,3]
1207-- >>> splitAt_ 3 [1,2,3]
1210-- >>> splitAt_ 4 [1,2,3]
1213-- > splitAt n f1 f2 = Fold.serialWith (,) (Fold.take n f1) f2
1217{-# INLINE splitAt #-}
1224splitAt n fld = serialWith (,) (take n fld)
1226------------------------------------------------------------------------------
1227-- Element Aware APIs
1228------------------------------------------------------------------------------
1230------------------------------------------------------------------------------
1232------------------------------------------------------------------------------
1234-- Note: Keep this consistent with S.splitOn. In fact we should eliminate
1235-- S.splitOn in favor of the fold.
1237-- XXX Use Fold.many instead once it is fixed.
1239-- | Like 'takeEndBy' but drops the element on which the predicate succeeds.
1241-- >>> Stream.fold (Fold.takeEndBy_ (== '\n') Fold.toList) $ Stream.fromList "hello\nthere\n"
1244-- >>> Stream.toList $ Stream.foldMany (Fold.takeEndBy_ (== '\n') Fold.toList) $ Stream.fromList "hello\nthere\n"
1247-- > Stream.splitOnSuffix p f = Stream.foldMany (Fold.takeEndBy_ p f)
1249-- See 'Streamly.Prelude.splitOnSuffix' for more details on splitting a
1250-- stream using 'takeEndBy_'.
1253{-# INLINE takeEndBy_ #-}
1254takeEndBy_ :: Monad m => (a -> Bool) -> Fold m a b -> Fold m a b
1255takeEndBy_ predicate (Fold fstep finitial fextract) =
1256 Fold step finitial fextract
1261 if not (predicate a)
1263 else Done <$> fextract s
1265-- | Take the input, stop when the predicate succeeds taking the succeeding
1268-- >>> Stream.fold (Fold.takeEndBy (== '\n') Fold.toList) $ Stream.fromList "hello\nthere\n"
1271-- >>> Stream.toList $ Stream.foldMany (Fold.takeEndBy (== '\n') Fold.toList) $ Stream.fromList "hello\nthere\n"
1272-- ["hello\n","there\n"]
1274-- > Stream.splitWithSuffix p f = Stream.foldMany (Fold.takeEndBy p f)
1276-- See 'Streamly.Prelude.splitWithSuffix' for more details on splitting a
1277-- stream using 'takeEndBy'.
1280{-# INLINE takeEndBy #-}
1281takeEndBy :: Monad m => (a -> Bool) -> Fold m a b -> Fold m a b
1282takeEndBy predicate (Fold fstep finitial fextract) =
1283 Fold step finitial fextract
1289 if not (predicate a)
1293 Partial s1 -> Done <$> fextract s1
1294 Done b -> return $ Done b
1296------------------------------------------------------------------------------
1297-- Binary splitting on a separator
1298------------------------------------------------------------------------------
1301-- | Find the first occurrence of the specified sequence in the input stream
1302-- and break the input stream into two parts, the first part consisting of the
1303-- stream before the sequence and the second part consisting of the sequence
1304-- and the rest of the stream.
1306-- > let breakOn_ pat xs = S.fold (S.breakOn pat FL.toList FL.toList) $ S.fromList xs
1308-- >>> breakOn_ "dear" "Hello dear world!"
1309-- > ("Hello ","dear world!")
1311{-# INLINE breakOn #-}
1312breakOn :: Monad m => Array a -> Fold m a b -> Fold m a c -> Fold m a (b,c)
1313breakOn pat f m = undefined
1316------------------------------------------------------------------------------
1318------------------------------------------------------------------------------
1320-- | Distribute one copy of the stream to each fold and zip the results.
1323-- |-------Fold m a b--------|
1324-- ---stream m a---| |---m (b,c)
1325-- |-------Fold m a c--------|
1327-- >>> Stream.fold (Fold.tee Fold.sum Fold.length) (Stream.enumerateFromTo 1.0 100.0)
1330-- > tee = teeWith (,)
1334tee :: Monad m => Fold m a b -> Fold m a c -> Fold m a (b,c)
1337-- XXX use "List" instead of "[]"?, use Array for output to scale it to a large
1338-- number of consumers? For polymorphic case a vector could be helpful. For
1339-- Storables we can use arrays. Will need separate APIs for those.
1341-- | Distribute one copy of the stream to each fold and collect the results in
1346-- |-------Fold m a b--------|
1347-- ---stream m a---| |---m [b]
1348-- |-------Fold m a b--------|
1353-- >>> Stream.fold (Fold.distribute [Fold.sum, Fold.length]) (Stream.enumerateFromTo 1 5)
1356-- >>> distribute = Prelude.foldr (Fold.teeWith (:)) (Fold.fromPure [])
1358-- This is the consumer side dual of the producer side 'sequence' operation.
1360-- Stops when all the folds stop.
1363{-# INLINE distribute #-}
1364distribute :: Monad m => [Fold m a b] -> Fold m a [b]
1365distribute = Prelude.foldr (teeWith (:)) (fromPure [])
1367------------------------------------------------------------------------------
1369------------------------------------------------------------------------------
1371{-# INLINE partitionByMUsing #-}
1372partitionByMUsing :: Monad m =>
1373 ( (x -> y -> (x, y))
1374 -> Fold m (Either b c) x
1375 -> Fold m (Either b c) y
1376 -> Fold m (Either b c) (x, y)
1378 -> (a -> m (Either b c))
1382partitionByMUsing t f fld1 fld2 =
1383 let l = lmap (fromLeft undefined) fld1 -- :: Fold m (Either b c) x
1384 r = lmap (fromRight undefined) fld2 -- :: Fold m (Either b c) y
1385 in lmapM f (t (,) (filter isLeft l) (filter isRight r))
1387-- | Partition the input over two folds using an 'Either' partitioning
1392-- |-------Fold b x--------|
1393-- -----stream m a --> (Either b c)----| |----(x,y)
1394-- |-------Fold c y--------|
1397-- Send input to either fold randomly:
1400-- > import System.Random (randomIO)
1401-- > randomly a = randomIO >>= \\x -> return $ if x then Left a else Right a
1402-- > Stream.fold (Fold.partitionByM randomly Fold.length Fold.length) (Stream.enumerateFromTo 1 100)
1406-- Send input to the two folds in a proportion of 2:1:
1409-- proportionately m n = do
1410-- ref <- newIORef $ cycle $ concat [replicate m Left, replicate n Right]
1412-- r <- readIORef ref
1413-- writeIORef ref $ tail r
1414-- return $ Prelude.head r a
1419-- f <- proportionately 2 1
1420-- r <- Stream.fold (Fold.partitionByM f Fold.length Fold.length) (Stream.enumerateFromTo (1 :: Int) 100)
1428-- This is the consumer side dual of the producer side 'mergeBy' operation.
1430-- When one fold is done, any input meant for it is ignored until the other
1431-- fold is also done.
1433-- Stops when both the folds stop.
1435-- /See also: 'partitionByFstM' and 'partitionByMinM'./
1438{-# INLINE partitionByM #-}
1439partitionByM :: Monad m
1440 => (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1441partitionByM = partitionByMUsing teeWith
1443-- | Similar to 'partitionByM' but terminates when the first fold terminates.
1446{-# INLINE partitionByFstM #-}
1447partitionByFstM :: Monad m
1448 => (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1449partitionByFstM = partitionByMUsing teeWithFst
1451-- | Similar to 'partitionByM' but terminates when any fold terminates.
1454{-# INLINE partitionByMinM #-}
1455partitionByMinM :: Monad m =>
1456 (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1457partitionByMinM = partitionByMUsing teeWithMin
1459-- Note: we could use (a -> Bool) instead of (a -> Either b c), but the latter
1460-- makes the signature clearer as to which case belongs to which fold.
1461-- XXX need to check the performance in both cases.
1462-- | Same as 'partitionByM' but with a pure partition function.
1464-- Count even and odd numbers in a stream:
1467-- let f = Fold.partitionBy (\n -> if even n then Left n else Right n)
1468-- (fmap (("Even " ++) . show) Fold.length)
1469-- (fmap (("Odd " ++) . show) Fold.length)
1470-- in Stream.fold f (Stream.enumerateFromTo 1 100)
1472-- ("Even 50","Odd 50")
1475{-# INLINE partitionBy #-}
1476partitionBy :: Monad m
1477 => (a -> Either b c) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1478partitionBy f = partitionByM (return . f)
1480-- | Compose two folds such that the combined fold accepts a stream of 'Either'
1481-- and routes the 'Left' values to the first fold and 'Right' values to the
1484-- > partition = partitionBy id
1487{-# INLINE partition #-}
1489 => Fold m b x -> Fold m c y -> Fold m (Either b c) (x, y)
1490partition = partitionBy id
1493-- | Send one item to each fold in a round-robin fashion. This is the consumer
1494-- side dual of producer side 'mergeN' operation.
1496-- partitionN :: Monad m => [Fold m a b] -> Fold m a [b]
1497-- partitionN fs = Fold step begin done
1500------------------------------------------------------------------------------
1501-- demux: in a key value stream fold each key sub-stream with a different fold
1502------------------------------------------------------------------------------
1504-- TODO Demultiplex an input element into a number of typed variants. We want
1505-- to statically restrict the target values within a set of predefined types,
1506-- an enumeration of a GADT.
1508-- This is the consumer side dual of the producer side 'mux' operation (XXX to
1511-- XXX If we use Refold in it, it can perhaps fuse/be more efficient. For
1512-- example we can store just the result rather than storing the whole fold in
1515-- Note: There are separate functions to determine Key and Fold from the input
1516-- because key is to be determined on each input whereas fold is to be
1517-- determined only once for a key.
1519-- | In a key value stream, fold values corresponding to each key with a key
1520-- specific fold. The fold returns the fold result as the second component of
1521-- the output tuple whenever a fold terminates. The first component of the
1522-- tuple is a container of in-progress folds. If a fold terminates, another
1523-- instance of the fold is started upon receiving an input with that key.
1525-- This can be used to scan a stream and collect the results from the scan
1530{-# INLINE demuxScanWith #-}
1531demuxScanWith :: (Monad m, IsMap f, Traversable f) =>
1533 -> (a -> m (Fold m a b))
1534 -> Fold m a (m (f b), Maybe (Key f, b))
1535demuxScanWith getKey getFold = fmap extract $ foldlM' step initial
1539 initial = return $ Tuple' IsMap.mapEmpty Nothing
1541 {-# INLINE runFold #-}
1542 runFold kv (Fold step1 initial1 extract1) (k, a) = do
1550 let fld = Fold step1 (return res1) extract1
1551 in Tuple' (IsMap.mapInsert k fld kv) Nothing
1552 Done b -> Tuple' (IsMap.mapDelete k kv) (Just (k, b))
1553 Done b -> return $ Tuple' kv (Just (k, b))
1555 step (Tuple' kv _) a = do
1557 case IsMap.mapLookup k kv of
1560 runFold kv fld (k, a)
1561 Just f -> runFold kv f (k, a)
1563 extract (Tuple' kv x) = (Prelude.mapM f kv, x)
1573-- | This is specialized version of 'demuxScanWith' that uses mutable cells as
1574-- fold accumulators for better performance.
1576-- >>> demuxScanMutWith = Fold.demuxScanWith
1578{-# INLINE demuxScanMutWith #-}
1579demuxScanMutWith :: (MonadIO m, IsMap f, Traversable f) =>
1581 -> (a -> m (Fold m a b))
1582 -> Fold m a (m (f b), Maybe (Key f, b))
1583demuxScanMutWith getKey getFold = fmap extract $ foldlM' step initial
1587 initial = return $ Tuple' IsMap.mapEmpty Nothing
1589 {-# INLINE initFold #-}
1590 initFold kv (Fold step1 initial1 extract1) (k, a) = do
1597 let fld = Fold step1 (return res1) extract1
1598 ref <- liftIO $ newIORef fld
1599 return $ Tuple' (IsMap.mapInsert k ref kv) Nothing
1600 Done b -> return $ Tuple' kv (Just (k, b))
1601 Done b -> return $ Tuple' kv (Just (k, b))
1603 {-# INLINE runFold #-}
1604 runFold kv ref (Fold step1 initial1 extract1) (k, a) = do
1611 let fld = Fold step1 (return res1) extract1
1612 liftIO $ writeIORef ref fld
1613 return $ Tuple' kv Nothing
1615 let kv1 = IsMap.mapDelete k kv
1616 in return $ Tuple' kv1 (Just (k, b))
1617 Done _ -> error "demuxScanMutWith: unreachable"
1619 step (Tuple' kv _) a = do
1621 case IsMap.mapLookup k kv of
1624 initFold kv f (k, a)
1626 f <- liftIO $ readIORef ref
1627 runFold kv ref f (k, a)
1629 extract (Tuple' kv x) = (Prelude.mapM f kv, x)
1634 (Fold _ i e) <- liftIO $ readIORef ref
1640-- | This collects all the results of 'demuxScanWith' in a container.
1642{-# INLINE demuxWith #-}
1643demuxWith :: (Monad m, IsMap f, Traversable f) =>
1644 (a -> Key f) -> (a -> m (Fold m a b)) -> Fold m a (f b)
1645demuxWith getKey getFold =
1647 classifier = demuxScanWith getKey getFold
1648 getMap Nothing = pure IsMap.mapEmpty
1649 getMap (Just action) = action
1651 teeWith IsMap.mapUnion
1652 (rmapM getMap $ lmap fst last)
1653 (lmap snd $ catMaybes toMap)
1654 in postscan classifier aggregator
1656-- | Same as 'demuxWith' but uses 'demuxScanMutWith' for better performance.
1658-- >>> demuxMutWith = Fold.demuxWith
1660{-# INLINE demuxMutWith #-}
1661demuxMutWith :: (MonadIO m, IsMap f, Traversable f) =>
1662 (a -> Key f) -> (a -> m (Fold m a b)) -> Fold m a (f b)
1663demuxMutWith getKey getFold =
1665 classifier = demuxScanMutWith getKey getFold
1666 getMap Nothing = pure IsMap.mapEmpty
1667 getMap (Just action) = action
1669 teeWith IsMap.mapUnion
1670 (rmapM getMap $ lmap fst last)
1671 (lmap snd $ catMaybes toMap)
1672 in postscan classifier aggregator
1674-- | Fold a stream of key value pairs using a function that maps keys to folds.
1676-- >>> import Data.Map (Map)
1678-- let f "SUM" = return Fold.sum
1679-- f _ = return Fold.product
1680-- input = Stream.fromList [("SUM",1),("PRODUCT",2),("SUM",3),("PRODUCT",4)]
1681-- in Stream.fold (Fold.demux f) input :: IO (Map String Int)
1683-- fromList [("PRODUCT",8),("SUM",4)]
1685-- >>> demux f = Fold.demuxWith fst (Fold.lmap snd . f)
1689demux :: (Monad m, IsMap f, Traversable f) =>
1690 (Key f -> m (Fold m a b)) -> Fold m (Key f, a) (f b)
1691demux f = demuxWith fst (\(k, _) -> fmap (lmap snd) (f k))
1693------------------------------------------------------------------------------
1694-- Classify: Like demux but uses the same fold for all keys.
1695------------------------------------------------------------------------------
1697-- XXX Change these to make the behavior similar to demux* variants. We can
1698-- implement this using classifyScanManyWith. Maintain a set of done folds in
1699-- the underlying monad, and when initial is called look it up, if the fold is
1700-- done then initial would set a flag in the state to ignore the input or
1703-- | Folds the values for each key using the supplied fold. When scanning, as
1704-- soon as the fold is complete, its result is available in the second
1705-- component of the tuple. The first component of the tuple is a snapshot of
1706-- the in-progress folds.
1708-- Once the fold for a key is done, any future values of the key are ignored.
1710-- >>> classifyScanWith f fld = Fold.demuxScanWith f (const fld)
1712{-# INLINE classifyScanWith #-}
1713classifyScanWith :: (Monad m, IsMap f, Traversable f, Ord (Key f)) =>
1714 -- Note: we need to return the Map itself to display the in-progress values
1715 -- e.g. to implement top. We could possibly create a separate abstraction
1716 -- for that use case. We return an action because we want it to be lazy so
1717 -- that the downstream consumers can choose to process or discard it.
1718 (a -> Key f) -> Fold m a b -> Fold m a (m (f b), Maybe (Key f, b))
1719classifyScanWith f (Fold step1 initial1 extract1) =
1720 fmap extract $ foldlM' step initial
1724 initial = return $ Tuple3' IsMap.mapEmpty Set.empty Nothing
1726 {-# INLINE initFold #-}
1727 initFold kv set k a = do
1735 Tuple3' (IsMap.mapInsert k s1 kv) set Nothing
1737 Tuple3' kv set (Just (k, b))
1738 Done b -> return (Tuple3' kv (Set.insert k set) (Just (k, b)))
1740 step (Tuple3' kv set _) a = do
1742 case IsMap.mapLookup k kv of
1745 then return (Tuple3' kv set Nothing)
1746 else initFold kv set k a
1752 Tuple3' (IsMap.mapInsert k s1 kv) set Nothing
1754 let kv1 = IsMap.mapDelete k kv
1755 in Tuple3' kv1 (Set.insert k set) (Just (k, b))
1757 extract (Tuple3' kv _ x) = (Prelude.mapM extract1 kv, x)
1759-- XXX we can use a Prim IORef if we can constrain the state "s" to be Prim
1761-- The code is almost the same as classifyScanWith except the IORef operations.
1763-- | Same as classifyScanWith except that it uses mutable IORef cells in the
1764-- Map providing better performance. Be aware that if this is used as a scan,
1765-- the values in the intermediate Maps would be mutable.
1767-- >>> classifyScanMutWith = classifyScanWith
1768-- >>> classifyScanMutWith f fld = demuxScanMutWith f (const fld)
1770{-# INLINE classifyScanMutWith #-}
1771classifyScanMutWith :: (MonadIO m, IsMap f, Traversable f, Ord (Key f)) =>
1772 (a -> Key f) -> Fold m a b -> Fold m a (m (f b), Maybe (Key f, b))
1773classifyScanMutWith f (Fold step1 initial1 extract1) =
1774 fmap extract $ foldlM' step initial
1778 initial = return $ Tuple3' IsMap.mapEmpty Set.empty Nothing
1780 {-# INLINE initFold #-}
1781 initFold kv set k a = do
1788 ref <- liftIO $ newIORef s1
1789 return $ Tuple3' (IsMap.mapInsert k ref kv) set Nothing
1791 return $ Tuple3' kv set (Just (k, b))
1792 Done b -> return (Tuple3' kv (Set.insert k set) (Just (k, b)))
1794 step (Tuple3' kv set _) a = do
1796 case IsMap.mapLookup k kv of
1799 then return (Tuple3' kv set Nothing)
1800 else initFold kv set k a
1802 s <- liftIO $ readIORef ref
1806 liftIO $ writeIORef ref s1
1807 return $ Tuple3' kv set Nothing
1809 let kv1 = IsMap.mapDelete k kv
1811 $ Tuple3' kv1 (Set.insert k set) (Just (k, b))
1813 extract (Tuple3' kv _ x) =
1814 (Prelude.mapM (\ref -> liftIO (readIORef ref) >>= extract1) kv, x)
1816-- | Fold a key value stream to a key-value container. If the same key appears
1817-- multiple times, only the last value is retained.
1819toMap :: (Monad m, IsMap f) => Fold m (Key f, a) (f a)
1820toMap = foldl' (\kv (k, v) -> IsMap.mapInsert k v kv) IsMap.mapEmpty
1822-- | Split the input stream based on a key field and fold each split using the
1823-- given fold. Useful for map/reduce, bucketizing the input in different bins
1824-- or for generating histograms.
1826-- >>> import Data.Map.Strict (Map)
1828-- let input = Stream.fromList [("ONE",1),("ONE",1.1),("TWO",2), ("TWO",2.2)]
1829-- classify = Fold.classifyWith fst (Fold.lmap snd Fold.toList)
1830-- in Stream.fold classify input :: IO (Map String [Double])
1832-- fromList [("ONE",[1.0,1.1]),("TWO",[2.0,2.2])]
1834-- Once the classifier fold terminates for a particular key any further inputs
1835-- in that bucket are ignored.
1837-- Space used is proportional to the number of keys seen till now and
1838-- monotonically increases because it stores whether a key has been seen or
1845{-# INLINE classifyWith #-}
1846classifyWith :: (Monad m, IsMap f, Traversable f, Ord (Key f)) =>
1847 (a -> Key f) -> Fold m a b -> Fold m a (f b)
1850 classifier = classifyScanWith f fld
1851 getMap Nothing = pure IsMap.mapEmpty
1852 getMap (Just action) = action
1854 teeWith IsMap.mapUnion
1855 (rmapM getMap $ lmap fst last)
1856 (lmap snd $ catMaybes toMap)
1857 in postscan classifier aggregator
1859-- | Same as 'classifyWith' but maybe faster because it uses mutable cells as
1860-- fold accumulators in the Map.
1862-- >>> classifyMutWith = Fold.classifyWith
1864{-# INLINE classifyMutWith #-}
1865classifyMutWith :: (MonadIO m, IsMap f, Traversable f, Ord (Key f)) =>
1866 (a -> Key f) -> Fold m a b -> Fold m a (f b)
1867classifyMutWith f fld =
1869 classifier = classifyScanMutWith f fld
1870 getMap Nothing = pure IsMap.mapEmpty
1871 getMap (Just action) = action
1873 teeWith IsMap.mapUnion
1874 (rmapM getMap $ lmap fst last)
1875 (lmap snd $ catMaybes toMap)
1876 in postscan classifier aggregator
1878-- | Given an input stream of key value pairs and a fold for values, fold all
1879-- the values belonging to each key. Useful for map/reduce, bucketizing the
1880-- input in different bins or for generating histograms.
1883-- let input = Stream.fromList [("ONE",1),("ONE",1.1),("TWO",2), ("TWO",2.2)]
1884-- in Stream.fold (Fold.classify Fold.toList) input
1886-- fromList [("ONE",[1.0,1.1]),("TWO",[2.0,2.2])]
1890-- >>> classify = Fold.classifyWith fst . Fold.lmap snd
1893{-# INLINE classify #-}
1894classify :: (Monad m, Ord k) => Fold m a b -> Fold m (k, a) (Map k b)
1895classify = classifyWith fst . lmap snd
1897------------------------------------------------------------------------------
1899------------------------------------------------------------------------------
1901{-# INLINE unzipWithMUsing #-}
1902unzipWithMUsing :: Monad m =>
1903 ( (x -> y -> (x, y))
1906 -> Fold m (b, c) (x, y)
1912unzipWithMUsing t f fld1 fld2 =
1913 let f1 = lmap fst fld1 -- :: Fold m (b, c) b
1914 f2 = lmap snd fld2 -- :: Fold m (b, c) c
1915 in lmapM f (t (,) f1 f2)
1917-- | Like 'unzipWith' but with a monadic splitter function.
1919-- @unzipWithM k f1 f2 = lmapM k (unzip f1 f2)@
1922{-# INLINE unzipWithM #-}
1923unzipWithM :: Monad m
1924 => (a -> m (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
1925unzipWithM = unzipWithMUsing teeWith
1927-- | Similar to 'unzipWithM' but terminates when the first fold terminates.
1930{-# INLINE unzipWithFstM #-}
1931unzipWithFstM :: Monad m =>
1932 (a -> m (b, c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1933unzipWithFstM = unzipWithMUsing teeWithFst
1935-- | Similar to 'unzipWithM' but terminates when any fold terminates.
1938{-# INLINE unzipWithMinM #-}
1939unzipWithMinM :: Monad m =>
1940 (a -> m (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
1941unzipWithMinM = unzipWithMUsing teeWithMin
1943-- | Split elements in the input stream into two parts using a pure splitter
1944-- function, direct each part to a different fold and zip the results.
1946-- @unzipWith f fld1 fld2 = Fold.lmap f (Fold.unzip fld1 fld2)@
1948-- This fold terminates when both the input folds terminate.
1951{-# INLINE unzipWith #-}
1953 => (a -> (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
1954unzipWith f = unzipWithM (return . f)
1956-- | Send the elements of tuples in a stream of tuples through two different
1961-- |-------Fold m a x--------|
1962-- ---------stream of (a,b)--| |----m (x,y)
1963-- |-------Fold m b y--------|
1967-- > unzip = Fold.unzipWith id
1969-- This is the consumer side dual of the producer side 'zip' operation.
1973unzip :: Monad m => Fold m a x -> Fold m b y -> Fold m (a,b) (x,y)
1976------------------------------------------------------------------------------
1977-- Combining streams and folds - Zipping
1978------------------------------------------------------------------------------
1980-- XXX These can be implemented using the fold scan, using the stream as a
1983-- | Zip a stream with the input of a fold using the supplied function.
1987{-# INLINE zipWithM #-}
1988zipWithM :: -- Monad m =>
1989 (a -> b -> m c) -> t m a -> Fold m c x -> Fold m b x
1992-- | Zip a stream with the input of a fold.
1997zip :: Monad m => t m a -> Fold m (a, b) x -> Fold m b x
1998zip = zipWithM (curry return)
2000-- | Pair each element of a fold input with its index, starting from index 0.
2003{-# INLINE indexed #-}
2004indexed :: -- forall m a b. Monad m =>
2005 Fold m (Int, a) b -> Fold m a b
2006indexed = undefined -- zip (Stream.enumerateFrom 0 :: SerialT m Int)
2008-- | Change the predicate function of a Fold from @a -> b@ to accept an
2009-- additional state input @(s, a) -> b@. Convenient to filter with an
2010-- addiitonal index or time input.
2013-- filterWithIndex = with indexed filter
2014-- filterWithAbsTime = with timestamped filter
2015-- filterWithRelTime = with timeIndexed filter
2021 (Fold m (s, a) b -> Fold m a b)
2022 -> (((s, a) -> c) -> Fold m (s, a) b -> Fold m (s, a) b)
2023 -> (((s, a) -> c) -> Fold m a b -> Fold m a b)
2024with f comb g = f . comb g . lmap snd
2026-- | @sampleFromthen offset stride@ samples the element at @offset@ index and
2027-- then every element at strides of @stride@.
2030{-# INLINE sampleFromthen #-}
2031sampleFromthen :: Monad m => Int -> Int -> Fold m a b -> Fold m a b
2032sampleFromthen offset size =
2033 with indexed filter (\(i, _) -> (i + offset) `mod` size == 0)
2035------------------------------------------------------------------------------
2037------------------------------------------------------------------------------
2039-- | @concatSequence f t@ applies folds from stream @t@ sequentially and
2040-- collects the results using the fold @f@.
2044{-# INLINE concatSequence #-}
2047 Fold m b c -> t (Fold m a b) -> Fold m a c
2048concatSequence _f _p = undefined
2050-- | Group the input stream into groups of elements between @low@ and @high@.
2051-- Collection starts in chunks of @low@ and then keeps doubling until we reach
2052-- @high@. Each chunk is folded using the provided fold function.
2054-- This could be useful, for example, when we are folding a stream of unknown
2055-- size to a stream of arrays and we want to minimize the number of
2058-- NOTE: this would be an application of "many" using a terminating fold.
2062{-# INLINE chunksBetween #-}
2063chunksBetween :: -- Monad m =>
2064 Int -> Int -> Fold m a b -> Fold m b c -> Fold m a c
2065chunksBetween _low _high _f1 _f2 = undefined
2067-- | A fold that buffers its input to a pure stream.
2069-- /Warning!/ working on large streams accumulated as buffers in memory could
2070-- be very inefficient, consider using "Streamly.Data.Array" instead.
2072-- >>> toStream = fmap SerialT Fold.toStreamK
2075{-# INLINE toStream #-}
2076toStream :: Monad m => Fold m a (SerialT n a)
2077toStream = fmap SerialT toStreamK
2079-- This is more efficient than 'toStream'. toStream is exactly the same as
2080-- reversing the stream after toStreamRev.
2082-- | Buffers the input stream to a pure stream in the reverse order of the
2085-- >>> toStreamRev = fmap SerialT Fold.toStreamKRev
2087-- /Warning!/ working on large streams accumulated as buffers in memory could
2088-- be very inefficient, consider using "Streamly.Data.Array" instead.
2092-- xn : ... : x2 : x1 : []
2093{-# INLINE toStreamRev #-}
2094toStreamRev :: Monad m => Fold m a (SerialT n a)
2095toStreamRev = fmap SerialT toStreamKRev
2097-- XXX This does not fuse. It contains a recursive step function. We will need
2098-- a Skip input constructor in the fold type to make it fuse.
2100-- | Unfold and flatten the input stream of a fold.
2103-- Stream.fold (unfoldMany u f) = Stream.fold f . Stream.unfoldMany u
2107{-# INLINE unfoldMany #-}
2108unfoldMany :: Monad m => Unfold m a b -> Fold m b c -> Fold m a c
2109unfoldMany (Unfold ustep inject) (Fold fstep initial extract) =
2110 Fold consume initial extract
2114 {-# INLINE produce #-}
2118 StreamD.Yield b us1 -> do
2121 Partial fs1 -> produce fs1 us1
2122 -- XXX What to do with the remaining stream?
2123 Done c -> return $ Done c
2124 StreamD.Skip us1 -> produce fs us1
2125 StreamD.Stop -> return $ Partial fs
2127 {-# INLINE_LATE consume #-}
2128 consume s a = inject a >>= produce s
2131topBy :: (MonadIO m, Storable a) =>
2132 (a -> a -> Ordering)
2134 -> Fold m a (MA.Array a)
2135topBy cmp n = Fold step initial extract
2140 arr <- MA.newArray n
2142 then return $ Done arr
2143 else return $ Partial (arr, 0)
2148 arr' <- MA.snoc arr x
2150 return $ Partial (arr', i + 1)
2152 x1 <- MA.getIndexUnsafe (i - 1) arr
2155 MA.putIndexUnsafe (i - 1) x arr
2157 return $ Partial (arr, i)
2158 _ -> return $ Partial (arr, i)
2160 extract = return . fst
2162-- | Fold the input stream to top n elements.
2164-- >>> stream = Stream.fromList [2::Int,7,9,3,1,5,6,11,17]
2165-- >>> Stream.fold (Fold.top 3) stream >>= MA.toList
2170top :: (MonadIO m, Storable a, Ord a) => Int -> Fold m a (MA.Array a)
2171top = topBy $ flip compare
2173-- | Fold the input stream to bottom n elements.
2175-- >>> stream = Stream.fromList [2::Int,7,9,3,1,5,6,11,17]
2176-- >>> Stream.fold (Fold.bottom 3) stream >>= MA.toList
2180{-# INLINE bottom #-}
2181bottom :: (MonadIO m, Storable a, Ord a) => Int -> Fold m a (MA.Array a)
2182bottom = topBy compare
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list