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
38 -- *** Semigroups and Monoids
57 -- *** Saturating Reducers
58 -- | 'product' terminates if it becomes 0. Other folds can theoretically
59 -- saturate on bounded types, and therefore terminate, however, they will
60 -- run forever on unbounded types like Integer/Double.
69 -- | Avoid using these folds in scalable or performance critical
70 -- applications, they buffer all the input in GC memory which can be
71 -- detrimental to performance if the input is large.
78 -- ** Terminating Folds
103 -- ** Transforming the Monad
107 -- ** Mapping on output
110 -- ** Mapping on Input
125 -- ** Mapping Filters
131 -- ** Scanning Filters
136 -- | Insertion adds more elements to the stream.
164 -- , splitIn -- sessionN
166 -- ** Parallel Distribution
175 -- ** Parallel Alternative
187 -- | Direct values in the input stream to different folds using an n-ary
189 , demux -- XXX rename this to demux_
191 , demuxDefault -- XXX rename this to demux
197 -- | In an input stream of key value pairs fold values for different keys
198 -- in individual output buckets using the given fold.
206 -- These two can be expressed using lmap/lmapM and unzip
226 -- * Running Partially
240import Control.Monad (void)
241import Data.Bifunctor (first)
242import Data.Functor.Identity (Identity(..))
243import Data.Int (Int64)
244import Data.Map.Strict (Map)
245import Data.Maybe (isJust, fromJust)
246#if __GLASGOW_HASKELL__ < 804
247import Data.Semigroup (Semigroup((<>)))
249import Streamly.Internal.Data.Either.Strict
250 (Either'(..), fromLeft', fromRight', isLeft', isRight')
251import Streamly.Internal.Data.Pipe.Type (Pipe (..), PipeState(..))
252import Streamly.Internal.Data.Tuple.Strict (Tuple'(..), Tuple3'(..))
253import Streamly.Internal.Data.Stream.Serial (SerialT)
255import qualified Data.Map.Strict as Map
256import qualified Streamly.Internal.Data.Pipe.Type as Pipe
257import qualified Streamly.Internal.Data.Stream.IsStream.Enumeration as Stream
258import qualified Streamly.Internal.Data.Stream.StreamK as K
259import qualified Prelude
262 ( filter, foldl1, drop, dropWhile, take, takeWhile, zipWith
263 , foldl, foldr, map, mapM_, sequence, all, any, sum, product, elem
264 , notElem, maximum, minimum, head, last, tail, length, null
265 , reverse, iterate, init, and, or, lookup, (!!)
266 , scanl, scanl1, replicate, concatMap, mconcat, foldMap, unzip
267 , span, splitAt, break, mapM, zip)
268import Streamly.Internal.Data.Fold.Type
272-- >>> import Prelude hiding (break, map, span, splitAt)
273-- >>> import qualified Streamly.Prelude as Stream
274-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream (parse, foldMany)
275-- >>> import qualified Streamly.Data.Fold as Fold
276-- >>> import qualified Streamly.Internal.Data.Fold as Fold
277-- >>> import qualified Streamly.Internal.Data.Parser as Parser
279------------------------------------------------------------------------------
281------------------------------------------------------------------------------
283-- | Change the underlying monad of a fold
286hoist :: (forall x. m x -> n x) -> Fold m a b -> Fold n a b
287hoist f (Fold step initial extract) =
288 Fold (\x a -> f $ step x a) (f initial) (f . extract)
290-- | Adapt a pure fold to any monad
292-- > generally = Fold.hoist (return . runIdentity)
295generally :: Monad m => Fold Identity a b -> Fold m a b
296generally = hoist (return . runIdentity)
298------------------------------------------------------------------------------
299-- Transformations on fold inputs
300------------------------------------------------------------------------------
302-- | Flatten the monadic output of a fold to pure output.
305{-# DEPRECATED sequence "Use \"rmapM id\" instead" #-}
306{-# INLINE sequence #-}
307sequence :: Monad m => Fold m a (m b) -> Fold m a b
310-- | Map a monadic function on the output of a fold.
313{-# DEPRECATED mapM "Use rmapM instead" #-}
315mapM :: Monad m => (b -> m c) -> Fold m a b -> Fold m a c
318-- | @mapMaybe f fold@ maps a 'Maybe' returning function @f@ on the input of
319-- the fold, filters out 'Nothing' elements, and return the values extracted
322-- >>> f x = if even x then Just x else Nothing
323-- >>> fld = Fold.mapMaybe f Fold.toList
324-- >>> Stream.fold fld (Stream.enumerateFromTo 1 10)
328{-# INLINE mapMaybe #-}
329mapMaybe :: (Monad m) => (a -> Maybe b) -> Fold m b r -> Fold m a r
330mapMaybe f = map f . filter isJust . map fromJust
332------------------------------------------------------------------------------
333-- Transformations on fold inputs
334------------------------------------------------------------------------------
338-- | Apply a transformation on a 'Fold' using a 'Pipe'.
341{-# INLINE transform #-}
342transform :: Monad m => Pipe m a b -> Fold m b c -> Fold m a c
343transform (Pipe pstep1 pstep2 pinitial) (Fold fstep finitial fextract) =
344 Fold step initial extract
348 initial = first (Tuple' pinitial) <$> finitial
350 step (Tuple' ps fs) x = do
357 go acc (Pipe.Yield b (Consume ps')) = do
361 Partial s -> Partial $ Tuple' ps' s
363 go acc (Pipe.Yield b (Produce ps')) = do
368 Done b2 -> return $ Done b2
369 go acc (Pipe.Continue (Consume ps')) =
370 return $ Partial $ Tuple' ps' acc
371 go acc (Pipe.Continue (Produce ps')) = do
375 extract (Tuple' _ fs) = fextract fs
377------------------------------------------------------------------------------
379------------------------------------------------------------------------------
381------------------------------------------------------------------------------
383------------------------------------------------------------------------------
386-- > drainBy f = lmapM f drain
387-- > drainBy = Fold.foldMapM (void . f)
389-- Drain all input after passing it through a monadic function. This is the
390-- dual of mapM_ on stream producers.
392-- See also: 'Streamly.Prelude.mapM_'
395{-# INLINABLE drainBy #-}
396drainBy :: Monad m => (a -> m b) -> Fold m a ()
397drainBy f = lmapM f drain
402{-# INLINABLE drainBy2 #-}
403drainBy2 :: Monad m => (a -> m b) -> Fold2 m c a ()
404drainBy2 f = Fold2 (const (void . f)) (\_ -> return ()) return
406-- | Extract the last element of the input stream, if any.
408-- > last = fmap getLast $ Fold.foldMap (Last . Just)
411{-# INLINABLE last #-}
412last :: Monad m => Fold m a (Maybe a)
413last = foldl1' (\_ x -> x)
415------------------------------------------------------------------------------
417------------------------------------------------------------------------------
419-- | Like 'length', except with a more general 'Num' return value
421-- > genericLength = fmap getSum $ foldMap (Sum . const 1)
424{-# INLINE genericLength #-}
425genericLength :: (Monad m, Num b) => Fold m a b
426genericLength = foldl' (\n _ -> n + 1) 0
428-- | Determine the length of the input stream.
430-- > length = fmap getSum $ Fold.foldMap (Sum . const 1)
434length :: Monad m => Fold m a Int
435length = genericLength
437-- | Determine the sum of all elements of a stream of numbers. Returns additive
438-- identity (@0@) when the stream is empty. Note that this is not numerically
439-- stable for floating point numbers.
441-- > sum = fmap getSum $ Fold.foldMap Sum
445sum :: (Monad m, Num a) => Fold m a a
448-- | Determine the product of all elements of a stream of numbers. Returns
449-- multiplicative identity (@1@) when the stream is empty. The fold terminates
450-- when it encounters (@0@) in its input.
452-- Compare with @Fold.foldMap Product@.
455-- /Since 0.8.0 (Added 'Eq' constraint)/
456{-# INLINE product #-}
457product :: (Monad m, Num a, Eq a) => Fold m a a
458product = mkFold_ step (Partial 1)
467------------------------------------------------------------------------------
469------------------------------------------------------------------------------
471-- | Determine the maximum element in a stream using the supplied comparison
475{-# INLINE maximumBy #-}
476maximumBy :: Monad m => (a -> a -> Ordering) -> Fold m a (Maybe a)
477maximumBy cmp = foldl1' max'
488-- maximum = Fold.maximumBy compare
491-- Determine the maximum element in a stream.
493-- Compare with @Fold.foldMap Max@.
496{-# INLINE maximum #-}
497maximum :: (Monad m, Ord a) => Fold m a (Maybe a)
500-- | Computes the minimum element with respect to the given comparison function
503{-# INLINE minimumBy #-}
504minimumBy :: Monad m => (a -> a -> Ordering) -> Fold m a (Maybe a)
505minimumBy cmp = foldl1' min'
514-- | Determine the minimum element in a stream using the supplied comparison
518-- minimum = 'minimumBy' compare
521-- Compare with @Fold.foldMap Min@.
524{-# INLINE minimum #-}
525minimum :: (Monad m, Ord a) => Fold m a (Maybe a)
528------------------------------------------------------------------------------
529-- To Summary (Statistical)
530------------------------------------------------------------------------------
532-- | Compute a numerically stable arithmetic mean of all elements in the input
536{-# INLINABLE mean #-}
537mean :: (Monad m, Fractional a) => Fold m a a
538mean = fmap done $ foldl' step begin
544 step (Tuple' x n) y =
546 in Tuple' (x + (y - x) / n1) n1
548 done (Tuple' x _) = x
550-- | Compute a numerically stable (population) variance over all elements in
554{-# INLINABLE variance #-}
555variance :: (Monad m, Fractional a) => Fold m a a
556variance = fmap done $ foldl' step begin
560 begin = Tuple3' 0 0 0
562 step (Tuple3' n mean_ m2) x = Tuple3' n' mean' m2'
567 mean' = (n * mean_ + x) / (n + 1)
569 m2' = m2 + delta * delta * n / (n + 1)
571 done (Tuple3' n _ m2) = m2 / n
573-- | Compute a numerically stable (population) standard deviation over all
574-- elements in the input stream.
577{-# INLINABLE stdDev #-}
578stdDev :: (Monad m, Floating a) => Fold m a a
579stdDev = sqrt <$> variance
581-- | Compute an 'Int' sized polynomial rolling hash
583-- > H = salt * k ^ n + c1 * k ^ (n - 1) + c2 * k ^ (n - 2) + ... + cn * k ^ 0
585-- Where @c1@, @c2@, @cn@ are the elements in the input stream and @k@ is a
588-- This hash is often used in Rabin-Karp string search algorithm.
590-- See https://en.wikipedia.org/wiki/Rolling_hash
593{-# INLINABLE rollingHashWithSalt #-}
594rollingHashWithSalt :: (Monad m, Enum a) => Int64 -> Fold m a Int64
595rollingHashWithSalt = foldl' step
599 k = 2891336453 :: Int64
601 step cksum a = cksum * k + fromIntegral (fromEnum a)
603-- | A default salt used in the implementation of 'rollingHash'.
604{-# INLINE defaultSalt #-}
606defaultSalt = -2578643520546668380
608-- | Compute an 'Int' sized polynomial rolling hash of a stream.
610-- > rollingHash = Fold.rollingHashWithSalt defaultSalt
613{-# INLINABLE rollingHash #-}
614rollingHash :: (Monad m, Enum a) => Fold m a Int64
615rollingHash = rollingHashWithSalt defaultSalt
617-- | Compute an 'Int' sized polynomial rolling hash of the first n elements of
620-- > rollingHashFirstN = Fold.take n Fold.rollingHash
623{-# INLINABLE rollingHashFirstN #-}
624rollingHashFirstN :: (Monad m, Enum a) => Int -> Fold m a Int64
625rollingHashFirstN n = take n rollingHash
627------------------------------------------------------------------------------
628-- Monoidal left folds
629------------------------------------------------------------------------------
631-- | Append the elements of an input stream to a provided starting value.
633-- >>> Stream.fold (Fold.sconcat 10) (Stream.map Data.Monoid.Sum $ Stream.enumerateFromTo 1 10)
637-- sconcat = Fold.foldl' (<>)
641{-# INLINE sconcat #-}
642sconcat :: (Monad m, Semigroup a) => a -> Fold m a a
645-- | Fold an input stream consisting of monoidal elements using 'mappend'
648-- >>> Stream.fold Fold.mconcat (Stream.map Data.Monoid.Sum $ Stream.enumerateFromTo 1 10)
651-- > mconcat = Fold.sconcat mempty
654{-# INLINE mconcat #-}
657#if !MIN_VERSION_base(4,11,0)
660 , Monoid a) => Fold m a a
661mconcat = sconcat mempty
664-- > foldMap f = Fold.lmap f Fold.mconcat
666-- Make a fold from a pure function that folds the output of the function
667-- using 'mappend' and 'mempty'.
669-- >>> Stream.fold (Fold.foldMap Data.Monoid.Sum) $ Stream.enumerateFromTo 1 10
673{-# INLINABLE foldMap #-}
674foldMap :: (Monad m, Monoid b
675#if !MIN_VERSION_base(4,11,0)
678 ) => (a -> b) -> Fold m a b
679foldMap f = lmap f mconcat
682-- > foldMapM f = Fold.lmapM f Fold.mconcat
684-- Make a fold from a monadic function that folds the output of the function
685-- using 'mappend' and 'mempty'.
687-- >>> Stream.fold (Fold.foldMapM (return . Data.Monoid.Sum)) $ Stream.enumerateFromTo 1 10
691{-# INLINABLE foldMapM #-}
692foldMapM :: (Monad m, Monoid b) => (a -> m b) -> Fold m a b
693foldMapM act = foldlM' step (pure mempty)
699 return $! mappend m m'
701------------------------------------------------------------------------------
703------------------------------------------------------------------------------
706-- This is more efficient than 'Streamly.Internal.Data.Fold.toList'. toList is
707-- exactly the same as reversing the list after 'toListRev'.
709-- | Buffers the input stream to a list in the reverse order of the input.
711-- > toListRev = Fold.foldl' (flip (:)) []
713-- /Warning!/ working on large lists accumulated as buffers in memory could be
714-- very inefficient, consider using "Streamly.Array" instead.
718-- xn : ... : x2 : x1 : []
719{-# INLINABLE toListRev #-}
720toListRev :: Monad m => Fold m a [a]
721toListRev = foldl' (flip (:)) []
723------------------------------------------------------------------------------
725------------------------------------------------------------------------------
727-- | A fold that drains the first n elements of its input, running the effects
728-- and discarding the results.
730-- > drainN n = Fold.take n Fold.drain
733{-# INLINABLE drainN #-}
734drainN :: Monad m => Int -> Fold m a ()
735drainN n = take n drain
737------------------------------------------------------------------------------
739------------------------------------------------------------------------------
741-- | Like 'index', except with a more general 'Integral' argument
744{-# INLINABLE genericIndex #-}
745genericIndex :: (Integral i, Monad m) => i -> Fold m a (Maybe a)
746genericIndex i = mkFold step (Partial 0) (const Nothing)
755-- | Lookup the element at the given index.
757-- See also: 'Streamly.Prelude.!!'
760{-# INLINABLE index #-}
761index :: Monad m => Int -> Fold m a (Maybe a)
764-- | Extract the first element of the stream, if any.
767{-# INLINABLE head #-}
768head :: Monad m => Fold m a (Maybe a)
769head = mkFold_ (const (Done . Just)) (Partial Nothing)
771-- | Returns the first element that satisfies the given predicate.
774{-# INLINABLE find #-}
775find :: Monad m => (a -> Bool) -> Fold m a (Maybe a)
776find predicate = mkFold step (Partial ()) (const Nothing)
785-- | In a stream of (key-value) pairs @(a, b)@, return the value @b@ of the
786-- first pair where the key equals the given value @a@.
788-- > lookup = snd <$> Fold.find ((==) . fst)
791{-# INLINABLE lookup #-}
792lookup :: (Eq a, Monad m) => a -> Fold m (a,b) (Maybe b)
793lookup a0 = mkFold step (Partial ()) (const Nothing)
802-- | Returns the first index that satisfies the given predicate.
805{-# INLINABLE findIndex #-}
806findIndex :: Monad m => (a -> Bool) -> Fold m a (Maybe Int)
807findIndex predicate = mkFold step (Partial 0) (const Nothing)
816-- | Returns the first index where a given value is found in the stream.
818-- > elemIndex a = Fold.findIndex (== a)
821{-# INLINABLE elemIndex #-}
822elemIndex :: (Eq a, Monad m) => a -> Fold m a (Maybe Int)
823elemIndex a = findIndex (a ==)
825------------------------------------------------------------------------------
827------------------------------------------------------------------------------
829-- | Return 'True' if the input stream is empty.
831-- > null = fmap isJust Fold.head
834{-# INLINABLE null #-}
835null :: Monad m => Fold m a Bool
836null = mkFold (\() _ -> Done False) (Partial ()) (const True)
838-- | Returns 'True' if any of the elements of a stream satisfies a predicate.
840-- >>> Stream.fold (Fold.any (== 0)) $ Stream.fromList [1,0,1]
843-- > any p = Fold.lmap p Fold.or
847any :: Monad m => (a -> Bool) -> Fold m a Bool
848any predicate = mkFold_ step initial
852 initial = Partial False
859-- | Return 'True' if the given element is present in the stream.
861-- > elem a = Fold.any (== a)
864{-# INLINABLE elem #-}
865elem :: (Eq a, Monad m) => a -> Fold m a Bool
868-- | Returns 'True' if all elements of a stream satisfy a predicate.
870-- >>> Stream.fold (Fold.all (== 0)) $ Stream.fromList [1,0,1]
873-- > all p = Fold.lmap p Fold.and
877all :: Monad m => (a -> Bool) -> Fold m a Bool
878all predicate = mkFold_ step initial
882 initial = Partial True
889-- | Returns 'True' if the given element is not present in the stream.
891-- > notElem a = Fold.all (/= a)
894{-# INLINABLE notElem #-}
895notElem :: (Eq a, Monad m) => a -> Fold m a Bool
896notElem a = all (a /=)
898-- | Returns 'True' if all elements are 'True', 'False' otherwise
900-- > and = Fold.all (== True)
904and :: Monad m => Fold m Bool Bool
907-- | Returns 'True' if any element is 'True', 'False' otherwise
909-- > or = Fold.any (== True)
913or :: Monad m => Fold m Bool Bool
916------------------------------------------------------------------------------
918------------------------------------------------------------------------------
920------------------------------------------------------------------------------
921-- Grouping without looking at elements
922------------------------------------------------------------------------------
924------------------------------------------------------------------------------
926------------------------------------------------------------------------------
928-- | @splitAt n f1 f2@ composes folds @f1@ and @f2@ such that first @n@
929-- elements of its input are consumed by fold @f1@ and the rest of the stream
930-- is consumed by fold @f2@.
932-- >>> let splitAt_ n xs = Stream.fold (Fold.splitAt n Fold.toList Fold.toList) $ Stream.fromList xs
934-- >>> splitAt_ 6 "Hello World!"
935-- ("Hello ","World!")
937-- >>> splitAt_ (-1) [1,2,3]
940-- >>> splitAt_ 0 [1,2,3]
943-- >>> splitAt_ 1 [1,2,3]
946-- >>> splitAt_ 3 [1,2,3]
949-- >>> splitAt_ 4 [1,2,3]
952-- > splitAt n f1 f2 = Fold.serialWith (,) (Fold.take n f1) f2
956{-# INLINE splitAt #-}
963splitAt n fld = serialWith (,) (take n fld)
965------------------------------------------------------------------------------
967------------------------------------------------------------------------------
969------------------------------------------------------------------------------
971------------------------------------------------------------------------------
973-- Note: Keep this consistent with S.splitOn. In fact we should eliminate
974-- S.splitOn in favor of the fold.
976-- XXX Use Fold.many instead once it is fixed.
978-- | Like 'takeEndBy' but drops the element on which the predicate succeeds.
980-- >>> Stream.fold (Fold.takeEndBy_ (== '\n') Fold.toList) $ Stream.fromList "hello\nthere\n"
983-- >>> Stream.toList $ Stream.foldMany (Fold.takeEndBy_ (== '\n') Fold.toList) $ Stream.fromList "hello\nthere\n"
986-- > Stream.splitOnSuffix p f = Stream.foldMany (Fold.takeEndBy_ p f)
988-- See 'Streamly.Prelude.splitOnSuffix' for more details on splitting a
989-- stream using 'takeEndBy_'.
992{-# INLINE takeEndBy_ #-}
993takeEndBy_ :: Monad m => (a -> Bool) -> Fold m a b -> Fold m a b
994takeEndBy_ predicate (Fold fstep finitial fextract) =
995 Fold step finitial fextract
1000 if not (predicate a)
1002 else Done <$> fextract s
1004-- | Take the input, stop when the predicate succeeds taking the succeeding
1007-- >>> Stream.fold (Fold.takeEndBy (== '\n') Fold.toList) $ Stream.fromList "hello\nthere\n"
1010-- >>> Stream.toList $ Stream.foldMany (Fold.takeEndBy (== '\n') Fold.toList) $ Stream.fromList "hello\nthere\n"
1011-- ["hello\n","there\n"]
1013-- > Stream.splitWithSuffix p f = Stream.foldMany (Fold.takeEndBy p f)
1015-- See 'Streamly.Prelude.splitWithSuffix' for more details on splitting a
1016-- stream using 'takeEndBy'.
1019{-# INLINE takeEndBy #-}
1020takeEndBy :: Monad m => (a -> Bool) -> Fold m a b -> Fold m a b
1021takeEndBy predicate (Fold fstep finitial fextract) =
1022 Fold step finitial fextract
1028 if not (predicate a)
1032 Partial s1 -> Done <$> fextract s1
1033 Done b -> return $ Done b
1035------------------------------------------------------------------------------
1036-- Binary splitting on a separator
1037------------------------------------------------------------------------------
1040-- | Find the first occurrence of the specified sequence in the input stream
1041-- and break the input stream into two parts, the first part consisting of the
1042-- stream before the sequence and the second part consisting of the sequence
1043-- and the rest of the stream.
1045-- > let breakOn_ pat xs = S.fold (S.breakOn pat FL.toList FL.toList) $ S.fromList xs
1047-- >>> breakOn_ "dear" "Hello dear world!"
1048-- > ("Hello ","dear world!")
1050{-# INLINE breakOn #-}
1051breakOn :: Monad m => Array a -> Fold m a b -> Fold m a c -> Fold m a (b,c)
1052breakOn pat f m = undefined
1055------------------------------------------------------------------------------
1057------------------------------------------------------------------------------
1059-- | Distribute one copy of the stream to each fold and zip the results.
1062-- |-------Fold m a b--------|
1063-- ---stream m a---| |---m (b,c)
1064-- |-------Fold m a c--------|
1066-- >>> Stream.fold (Fold.tee Fold.sum Fold.length) (Stream.enumerateFromTo 1.0 100.0)
1069-- > tee = teeWith (,)
1073tee :: Monad m => Fold m a b -> Fold m a c -> Fold m a (b,c)
1076-- XXX use "List" instead of "[]"?, use Array for output to scale it to a large
1077-- number of consumers? For polymorphic case a vector could be helpful. For
1078-- Storables we can use arrays. Will need separate APIs for those.
1080-- | Distribute one copy of the stream to each fold and collect the results in
1085-- |-------Fold m a b--------|
1086-- ---stream m a---| |---m [b]
1087-- |-------Fold m a b--------|
1092-- >>> Stream.fold (Fold.distribute [Fold.sum, Fold.length]) (Stream.enumerateFromTo 1 5)
1095-- > distribute = Prelude.foldr (Fold.teeWith (:)) (Fold.fromPure [])
1097-- This is the consumer side dual of the producer side 'sequence' operation.
1099-- Stops when all the folds stop.
1102{-# INLINE distribute #-}
1103distribute :: Monad m => [Fold m a b] -> Fold m a [b]
1104distribute = Prelude.foldr (teeWith (:)) (fromPure [])
1106------------------------------------------------------------------------------
1108------------------------------------------------------------------------------
1110-- | Partition the input over two folds using an 'Either' partitioning
1115-- |-------Fold b x--------|
1116-- -----stream m a --> (Either b c)----| |----(x,y)
1117-- |-------Fold c y--------|
1120-- Send input to either fold randomly:
1123-- > import System.Random (randomIO)
1124-- > randomly a = randomIO >>= \\x -> return $ if x then Left a else Right a
1125-- > Stream.fold (Fold.partitionByM randomly Fold.length Fold.length) (Stream.enumerateFromTo 1 100)
1129-- Send input to the two folds in a proportion of 2:1:
1132-- import Data.IORef (newIORef, readIORef, writeIORef)
1133-- proportionately m n = do
1134-- ref <- newIORef $ cycle $ concat [replicate m Left, replicate n Right]
1135-- return $ \\a -> do
1136-- r <- readIORef ref
1137-- writeIORef ref $ tail r
1141-- f <- proportionately 2 1
1142-- r <- S.fold (FL.partitionByM f FL.length FL.length) (S.enumerateFromTo (1 :: Int) 100)
1149-- This is the consumer side dual of the producer side 'mergeBy' operation.
1151-- When one fold is done, any input meant for it is ignored until the other
1152-- fold is also done.
1154-- Stops when both the folds stop.
1156-- /See also: 'partitionByFstM' and 'partitionByMinM'./
1159{-# INLINE partitionByM #-}
1160partitionByM :: Monad m
1161 => (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1162partitionByM f (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1163 Fold step begin done
1175 Partial sR -> RunBoth sL sR
1176 Done bR -> RunLeft sL bR
1179 Partial sR -> Partial $ RunRight bL sR
1180 Done bR -> Done (bL, bR)
1182 step (RunBoth sL sR) a = do
1190 Partial sres -> RunBoth sres sR
1191 Done bres -> RunRight bres sR
1197 Partial sres -> RunBoth sL sres
1198 Done bres -> RunLeft sL bres
1199 step (RunLeft sL bR) a = do
1206 Partial sres -> Partial $ RunLeft sres bR
1207 Done bres -> Done (bres, bR)
1208 Right _ -> return $ Partial $ RunLeft sL bR
1209 step (RunRight bL sR) a = do
1212 Left _ -> return $ Partial $ RunRight bL sR
1217 Partial sres -> Partial $ RunRight bL sres
1218 Done bres -> Done (bL, bres)
1220 done (RunBoth sL sR) = (,) <$> doneL sL <*> doneR sR
1221 done (RunLeft sL bR) = (,bR) <$> doneL sL
1222 done (RunRight bL sR) = (bL,) <$> doneR sR
1224-- | Similar to 'partitionByM' but terminates when the first fold terminates.
1228{-# INLINE partitionByFstM #-}
1229partitionByFstM :: -- Monad m =>
1230 (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1231partitionByFstM = undefined
1233-- | Similar to 'partitionByM' but terminates when any fold terminates.
1237{-# INLINE partitionByMinM #-}
1238partitionByMinM :: -- Monad m =>
1239 (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1240partitionByMinM = undefined
1242-- Note: we could use (a -> Bool) instead of (a -> Either b c), but the latter
1243-- makes the signature clearer as to which case belongs to which fold.
1244-- XXX need to check the performance in both cases.
1245-- | Same as 'partitionByM' but with a pure partition function.
1247-- Count even and odd numbers in a stream:
1250-- let f = Fold.partitionBy (\n -> if even n then Left n else Right n)
1251-- (fmap (("Even " ++) . show) Fold.length)
1252-- (fmap (("Odd " ++) . show) Fold.length)
1253-- in Stream.fold f (Stream.enumerateFromTo 1 100)
1255-- ("Even 50","Odd 50")
1258{-# INLINE partitionBy #-}
1259partitionBy :: Monad m
1260 => (a -> Either b c) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1261partitionBy f = partitionByM (return . f)
1263-- | Compose two folds such that the combined fold accepts a stream of 'Either'
1264-- and routes the 'Left' values to the first fold and 'Right' values to the
1267-- > partition = partitionBy id
1270{-# INLINE partition #-}
1272 => Fold m b x -> Fold m c y -> Fold m (Either b c) (x, y)
1273partition = partitionBy id
1276-- | Send one item to each fold in a round-robin fashion. This is the consumer
1277-- side dual of producer side 'mergeN' operation.
1279-- partitionN :: Monad m => [Fold m a b] -> Fold m a [b]
1280-- partitionN fs = Fold step begin done
1283-- TODO Demultiplex an input element into a number of typed variants. We want
1284-- to statically restrict the target values within a set of predefined types,
1285-- an enumeration of a GADT. We also want to make sure that the Map contains
1286-- only those types and the full set of those types.
1288-- TODO Instead of the input Map it should probably be a lookup-table using an
1289-- array and not in GC memory. The same applies to the output Map as well.
1290-- However, that would only be helpful if we have a very large data structure,
1291-- need to measure and see how it scales.
1293-- This is the consumer side dual of the producer side 'mux' operation (XXX to
1296-- | Split the input stream based on a key field and fold each split using a
1297-- specific fold collecting the results in a map from the keys to the results.
1298-- Useful for cases like protocol handlers to handle different type of packets
1299-- using different handlers.
1303-- |-------Fold m a b
1304-- -----stream m a-----Map-----|
1305-- |-------Fold m a b
1310-- Any input that does not map to a fold in the input Map is silently ignored.
1312-- > demuxWith f kv = fmap fst $ demuxDefaultWith f kv drain
1316{-# INLINE demuxWith #-}
1317demuxWith :: (Monad m, Ord k)
1318 => (a -> (k, a')) -> Map k (Fold m a' b) -> Fold m a (Map k b)
1319demuxWith f kv = fmap fst $ demuxDefaultWith f kv drain
1321-- | Fold a stream of key value pairs using a map of specific folds for each
1322-- key into a map from keys to the results of fold outputs of the corresponding
1325-- >>> import qualified Data.Map
1327-- let table = Data.Map.fromList [("SUM", Fold.sum), ("PRODUCT", Fold.product)]
1328-- input = Stream.fromList [("SUM",1),("PRODUCT",2),("SUM",3),("PRODUCT",4)]
1329-- in Stream.fold (Fold.demux table) input
1331-- fromList [("PRODUCT",8),("SUM",4)]
1333-- > demux = demuxWith id
1337demux :: (Monad m, Ord k)
1338 => Map k (Fold m a b) -> Fold m (k, a) (Map k b)
1341data DemuxState s b doneMap runMap =
1342 DemuxMapAndDefault !s !doneMap !runMap
1343 | DemuxOnlyMap b !doneMap !runMap
1344 | DemuxOnlyDefault s !doneMap
1346-- | Like 'demuxWith' but uses a default catchall fold to handle inputs which
1347-- do not have a specific fold in the map to handle them.
1349-- If any fold in the map stops, inputs meant for that fold are sent to the
1350-- catchall fold. If the catchall fold stops then inputs that do not match any
1353-- Stops when all the folds, including the catchall fold, stop.
1357{-# INLINE demuxDefaultWith #-}
1358demuxDefaultWith :: (Monad m, Ord k)
1360 -> Map k (Fold m a' b)
1362 -> Fold m a (Map k b, c)
1363demuxDefaultWith f kv (Fold dstep dinitial dextract) =
1364 Fold step initial extract
1369 let runInit (Fold step1 initial1 done1) = do
1373 Partial _ -> Right' (Fold step1 (return r) done1)
1376 -- initialize folds in the kv map and separate the ones that are done
1377 -- from running ones
1378 kv1 <- Prelude.mapM runInit kv
1379 let runMap = Map.map fromRight' $ Map.filter isRight' kv1
1380 doneMap = Map.map fromLeft' $ Map.filter isLeft' kv1
1382 -- Run the default fold, and decide the next state based on its result
1388 $ if Map.size runMap > 0
1389 then DemuxMapAndDefault s doneMap runMap
1390 else DemuxOnlyDefault s doneMap
1392 if Map.size runMap > 0
1393 then Partial $ DemuxOnlyMap b doneMap runMap
1394 else Done (doneMap, b)
1396 {-# INLINE runFold #-}
1397 runFold fPartial fDone doneMap runMap (Fold step1 initial1 done1) k a1 = do
1402 return $ case res of
1404 let fld = Fold step1 (return $ Partial s) done1
1405 runMap1 = Map.insert k fld runMap
1406 in Partial $ fPartial doneMap runMap1
1408 let runMap1 = Map.delete k runMap
1409 doneMap1 = Map.insert k b doneMap
1410 if Map.size runMap1 == 0
1412 else Partial $ fPartial doneMap1 runMap1
1413 Done _ -> error "Bug: demuxDefaultWith: Done fold"
1415 step (DemuxMapAndDefault dacc doneMap runMap) a = do
1417 case Map.lookup k runMap of
1419 res <- dstep dacc (k, a1)
1423 Partial s -> DemuxMapAndDefault s doneMap runMap
1424 Done b -> DemuxOnlyMap b doneMap runMap
1427 (DemuxMapAndDefault dacc)
1428 (Partial . DemuxOnlyDefault dacc)
1429 doneMap runMap fld k a1
1431 step (DemuxOnlyMap dval doneMap runMap) a = do
1433 case Map.lookup k runMap of
1434 Nothing -> return $ Partial $ DemuxOnlyMap dval doneMap runMap
1439 doneMap runMap fld k a1
1440 step (DemuxOnlyDefault dacc doneMap) a = do
1442 res <- dstep dacc (k, a1)
1445 Partial s -> Partial $ DemuxOnlyDefault s doneMap
1446 Done b -> Done (doneMap, b)
1448 runExtract (Fold _ initial1 done1) = do
1451 Partial s -> done1 s
1454 extract (DemuxMapAndDefault dacc doneMap runMap) = do
1456 runMap1 <- Prelude.mapM runExtract runMap
1457 return (doneMap `Map.union` runMap1, b)
1458 extract (DemuxOnlyMap dval doneMap runMap) = do
1459 runMap1 <- Prelude.mapM runExtract runMap
1460 return (doneMap `Map.union` runMap1, dval)
1461 extract (DemuxOnlyDefault dacc doneMap) = do
1466-- > demuxDefault = demuxDefaultWith id
1469{-# INLINE demuxDefault #-}
1470demuxDefault :: (Monad m, Ord k)
1471 => Map k (Fold m a b) -> Fold m (k, a) b -> Fold m (k, a) (Map k b, b)
1472demuxDefault = demuxDefaultWith id
1474-- TODO If the data is large we may need a map/hashmap in pinned memory instead
1475-- of a regular Map. That may require a serializable constraint though. We can
1476-- have another API for that.
1478-- | Split the input stream based on a key field and fold each split using the
1479-- given fold. Useful for map/reduce, bucketizing the input in different bins
1480-- or for generating histograms.
1483-- let input = Stream.fromList [("ONE",1),("ONE",1.1),("TWO",2), ("TWO",2.2)]
1484-- in Stream.fold (Fold.classifyWith fst (Fold.map snd Fold.toList)) input
1486-- fromList [("ONE",[1.0,1.1]),("TWO",[2.0,2.2])]
1488-- If the classifier fold stops for a particular key any further inputs in that
1489-- bucket are ignored.
1495{-# INLINE classifyWith #-}
1496classifyWith :: (Monad m, Ord k) => (a -> k) -> Fold m a b -> Fold m a (Map k b)
1497classifyWith f (Fold step1 initial1 extract1) =
1498 rmapM extract $ foldlM' step initial
1502 initial = return Map.empty
1505 case Map.lookup k kv of
1512 $ flip (Map.insert k) kv
1514 Partial s1 -> Left' s1
1516 Done b -> return $ Map.insert k (Right' b) kv
1522 $ flip (Map.insert k) kv
1524 Partial s1 -> Left' s1
1526 Right' _ -> return kv
1535 Left' s -> extract1 s
1536 Right' b -> return b)
1538-- | Given an input stream of key value pairs and a fold for values, fold all
1539-- the values belonging to each key. Useful for map/reduce, bucketizing the
1540-- input in different bins or for generating histograms.
1543-- let input = Stream.fromList [("ONE",1),("ONE",1.1),("TWO",2), ("TWO",2.2)]
1544-- in Stream.fold (Fold.classify Fold.toList) input
1546-- fromList [("ONE",[1.0,1.1]),("TWO",[2.0,2.2])]
1550-- > classify fld = Fold.classifyWith fst (map snd fld)
1553{-# INLINE classify #-}
1554classify :: (Monad m, Ord k) => Fold m a b -> Fold m (k, a) (Map k b)
1555classify fld = classifyWith fst (map snd fld)
1557------------------------------------------------------------------------------
1559------------------------------------------------------------------------------
1561-- | Like 'unzipWith' but with a monadic splitter function.
1563-- @unzipWithM k f1 f2 = lmapM k (unzip f1 f2)@
1566{-# INLINE unzipWithM #-}
1567unzipWithM :: Monad m
1568 => (a -> m (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
1569unzipWithM f (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1570 Fold step begin done
1582 Partial sR -> RunBoth sL sR
1583 Done bR -> RunLeft sL bR
1586 Partial sR -> Partial $ RunRight bL sR
1587 Done bR -> Done (bL, bR)
1589 step (RunBoth sL sR) a = do
1598 Partial sresR -> RunBoth sresL sresR
1599 Done bresR -> RunLeft sresL bresR
1603 Partial sresR -> Partial $ RunRight bresL sresR
1604 Done bresR -> Done (bresL, bresR)
1605 step (RunLeft sL bR) a = do
1610 Partial sresL -> Partial $ RunLeft sresL bR
1611 Done bresL -> Done (bresL, bR)
1612 step (RunRight bL sR) a = do
1617 Partial sresR -> Partial $ RunRight bL sresR
1618 Done bresR -> Done (bL, bresR)
1620 done (RunBoth sL sR) = (,) <$> doneL sL <*> doneR sR
1621 done (RunLeft sL bR) = (,bR) <$> doneL sL
1622 done (RunRight bL sR) = (bL,) <$> doneR sR
1624-- | Similar to 'unzipWithM' but terminates when the first fold terminates.
1628{-# INLINE unzipWithFstM #-}
1629unzipWithFstM :: -- Monad m =>
1630 (a -> m (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
1631unzipWithFstM = undefined
1633-- | Similar to 'unzipWithM' but terminates when any fold terminates.
1637{-# INLINE unzipWithMinM #-}
1638unzipWithMinM :: -- Monad m =>
1639 (a -> m (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
1640unzipWithMinM = undefined
1642-- | Split elements in the input stream into two parts using a pure splitter
1643-- function, direct each part to a different fold and zip the results.
1645-- @unzipWith f fld1 fld2 = Fold.lmap f (Fold.unzip fld1 fld2)@
1647-- This fold terminates when both the input folds terminate.
1650{-# INLINE unzipWith #-}
1652 => (a -> (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
1653unzipWith f = unzipWithM (return . f)
1655-- | Send the elements of tuples in a stream of tuples through two different
1660-- |-------Fold m a x--------|
1661-- ---------stream of (a,b)--| |----m (x,y)
1662-- |-------Fold m b y--------|
1666-- > unzip = Fold.unzipWith id
1668-- This is the consumer side dual of the producer side 'zip' operation.
1672unzip :: Monad m => Fold m a x -> Fold m b y -> Fold m (a,b) (x,y)
1675------------------------------------------------------------------------------
1676-- Combining streams and folds - Zipping
1677------------------------------------------------------------------------------
1679-- | Zip a stream with the input of a fold using the supplied function.
1683{-# INLINE zipWithM #-}
1684zipWithM :: -- Monad m =>
1685 (a -> b -> m c) -> t m a -> Fold m c x -> Fold m b x
1688-- | Zip a stream with the input of a fold.
1693zip :: Monad m => t m a -> Fold m (a, b) x -> Fold m b x
1694zip = zipWithM (curry return)
1696-- | Pair each element of a fold input with its index, starting from index 0.
1699{-# INLINE indexed #-}
1700indexed :: forall m a b. Monad m => Fold m (Int, a) b -> Fold m a b
1701indexed = zip (Stream.enumerateFrom 0 :: SerialT m Int)
1703-- | Change the predicate function of a Fold from @a -> b@ to accept an
1704-- additional state input @(s, a) -> b@. Convenient to filter with an
1705-- addiitonal index or time input.
1708-- filterWithIndex = with indexed filter
1709-- filterWithAbsTime = with timestamped filter
1710-- filterWithRelTime = with timeIndexed filter
1716 (Fold m (s, a) b -> Fold m a b)
1717 -> (((s, a) -> c) -> Fold m (s, a) b -> Fold m (s, a) b)
1718 -> (((s, a) -> c) -> Fold m a b -> Fold m a b)
1719with f comb g = f . comb g . map snd
1721-- | @sampleFromthen offset stride@ samples the element at @offset@ index and
1722-- then every element at strides of @stride@.
1725{-# INLINE sampleFromthen #-}
1726sampleFromthen :: Monad m => Int -> Int -> Fold m a b -> Fold m a b
1727sampleFromthen offset size =
1728 with indexed filter (\(i, _) -> (i + offset) `mod` size == 0)
1730------------------------------------------------------------------------------
1732------------------------------------------------------------------------------
1734-- | @concatSequence f t@ applies folds from stream @t@ sequentially and
1735-- collects the results using the fold @f@.
1739{-# INLINE concatSequence #-}
1742 Fold m b c -> t (Fold m a b) -> Fold m a c
1743concatSequence _f _p = undefined
1745-- | Group the input stream into groups of elements between @low@ and @high@.
1746-- Collection starts in chunks of @low@ and then keeps doubling until we reach
1747-- @high@. Each chunk is folded using the provided fold function.
1749-- This could be useful, for example, when we are folding a stream of unknown
1750-- size to a stream of arrays and we want to minimize the number of
1753-- NOTE: this would be an application of "many" using a terminating fold.
1757{-# INLINE chunksBetween #-}
1758chunksBetween :: -- Monad m =>
1759 Int -> Int -> Fold m a b -> Fold m b c -> Fold m a c
1760chunksBetween _low _high _f1 _f2 = undefined
1762-- | A fold that buffers its input to a pure stream.
1764-- /Warning!/ working on large streams accumulated as buffers in memory could
1765-- be very inefficient, consider using "Streamly.Data.Array" instead.
1767-- > toStream = foldr K.cons K.nil
1770{-# INLINE toStream #-}
1771toStream :: Monad m => Fold m a (SerialT Identity a)
1772toStream = foldr K.cons K.nil
1774-- This is more efficient than 'toStream'. toStream is exactly the same as
1775-- reversing the stream after toStreamRev.
1777-- | Buffers the input stream to a pure stream in the reverse order of the
1780-- > toStreamRev = foldl' (flip K.cons) K.nil
1782-- /Warning!/ working on large streams accumulated as buffers in memory could
1783-- be very inefficient, consider using "Streamly.Data.Array" instead.
1787-- xn : ... : x2 : x1 : []
1788{-# INLINABLE toStreamRev #-}
1789toStreamRev :: Monad m => Fold m a (SerialT Identity a)
1790toStreamRev = foldl' (flip K.cons) K.nil
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