@@ -16,2583 +16,20 @@ module Streamly.Internal.Data.Fold
16 -- * Imports
16 -- * Imports
19− -- * Fold Type
19+ module Streamly.Internal.Data.Fold.Step
20− Step (..)
20+ , module Streamly.Internal.Data.Fold.Type
21− , Fold (..)
21+ , module Streamly.Internal.Data.Fold.Tee
25− -- | Which constructor to use?
27− -- * @foldl*@: If the fold never terminates i.e. does not use the 'Done'
28− -- constructor otherwise use the @foldt*@ variants.
29− -- * @*M@: Use the @M@ suffix variants if any of the step, initial, or
30− -- extract function is monadic, otherwise use the pure variants.
42− -- | Monadic functions useful with mapM/lmapM on folds or streams.
49− -- *** Semigroups and Monoids
64− , rollingHashWithSalt
66− -- , rollingHashLastN
68− -- *** Saturating Reducers
69− -- | 'product' terminates if it becomes 0. Other folds can theoretically
70− -- saturate on bounded types, and therefore terminate, however, they will
71− -- run forever on unbounded types like Integer/Double.
80− -- | Avoid using these folds in scalable or performance critical
81− -- applications, they buffer all the input in GC memory which can be
82− -- detrimental to performance if the input is large.
96− -- | Stateful transformation of the elements. Useful in combination with
97− -- the 'scanMaybe' combinator. For scanners the result of the fold is
98− -- usually a transformation of the current element rather than an
99− -- aggregation of all elements till now.
101− -- , nthLast -- using Ring array
108− -- | Useful in combination with the 'scanMaybe' combinator.
117− -- ** Terminating Folds
119− -- | Folds that return a result without consuming any input.
124− -- *** Singleton folds
125− -- | Folds that terminate after consuming exactly one input element. All
126− -- these can be implemented in terms of the 'maybe' fold.
128− , null -- XXX not very useful and could be problematic, remove it?
133− -- | Terminate after consuming one or more elements.
152− -- | Useful in combination with the 'scanMaybe' combinator.
167− -- * Building Incrementally
180− -- * Combinators
22+ , module Streamly.Internal.Data.Fold.Combinators
181− -- ** Utilities
23+ , module Streamly.Internal.Data.Fold.Container
182− , with
24+ -- , module Streamly.Internal.Data.Fold.Window
184− -- ** Transforming the Monad
188− -- ** Mapping on output
191− -- ** Mapping on Input
197− -- ** Sliding Window
200− -- ** Scanning Input
210− -- ** Filtering Input
226− -- | Insertion adds more elements to the stream.
255− -- , splitIn -- sessionN
257− -- ** Parallel Distribution
268− -- These two can be expressed using lmap/lmapM and unzip
274− -- ** Parallel Alternative
292− , intersperseWithQuotes
315−#include "ArrayMacros.h"
317−import Control.Monad (void)
318−import Control.Monad.IO.Class (MonadIO(..))
319−import Data.Bifunctor (first)
320−import Data.Bits (shiftL, shiftR, (.|.), (.&.))
321−import Data.Either (isLeft, isRight, fromLeft, fromRight)
322−import Data.Int (Int64)
323−import Data.Proxy (Proxy(..))
324−import Data.Word (Word32)
325−import Foreign.Storable (Storable, peek)
326−import Streamly.Internal.Data.MutArray.Type (MutArray(..))
28+import Streamly.Internal.Data.Fold.Combinators
327−import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)
328−import Streamly.Internal.Data.Pipe.Type (Pipe (..), PipeState(..))
329−import Streamly.Internal.Data.Unbox (Unbox, sizeOf)
330−import Streamly.Internal.Data.Unfold.Type (Unfold(..))
331−import Streamly.Internal.Data.Tuple.Strict (Tuple'(..), Tuple3'(..))
332−import Streamly.Internal.Data.Stream.StreamD.Type (Stream)
334−import qualified Prelude
335−import qualified Streamly.Internal.Data.MutArray.Type as MA
336−import qualified Streamly.Internal.Data.Array.Type as Array
337−import qualified Streamly.Internal.Data.Fold.Window as FoldW
29+import Streamly.Internal.Data.Fold.Container
338−import qualified Streamly.Internal.Data.Pipe.Type as Pipe
339−import qualified Streamly.Internal.Data.Ring.Unboxed as Ring
340−import qualified Streamly.Internal.Data.Stream.StreamD.Type as StreamD
342−import Prelude hiding
343− ( Foldable(..), filter, drop, dropWhile, take, takeWhile, zipWith
344− , map, mapM_, sequence, all, any
345− , notElem, head, last, tail
346− , reverse, iterate, init, and, or, lookup, (!!)
347− , scanl, scanl1, replicate, concatMap, mconcat, unzip
348− , span, splitAt, break, mapM, zip, maybe)
349−import Streamly.Internal.Data.Fold.Type
30+import Streamly.Internal.Data.Fold.Step
350import Streamly.Internal.Data.Fold.Tee
31import Streamly.Internal.Data.Fold.Tee
32+import Streamly.Internal.Data.Fold.Type
33+-- import Streamly.Internal.Data.Fold.Window
352#include "DocTestDataFold.hs"
35#include "DocTestDataFold.hs"
354−------------------------------------------------------------------------------
356−------------------------------------------------------------------------------
358−-- | Drive a fold using the supplied 'Stream', reducing the resulting
359−-- expression strictly at each step.
363−-- >>> drive = flip Stream.fold
367−-- >>> Fold.drive (Stream.enumerateFromTo 1 100) Fold.sum
371−drive :: Monad m => Stream m a -> Fold m a b -> m b
372−drive = flip StreamD.fold
375−-- | Like 'drive' but also returns the remaining stream. The resulting stream
376−-- would be 'Stream.nil' if the stream finished before the fold.
380−-- >>> breakStream = flip Stream.foldBreak
384−{-# INLINE breakStreamK #-}
385−breakStreamK :: Monad m => StreamK m a -> Fold m a b -> m (b, StreamK m a)
386−breakStreamK strm fl = fmap f $ K.foldBreak fl (Stream.toStreamK strm)
390− f (b, str) = (b, Stream.fromStreamK str)
393−-- | Append a stream to a fold to build the fold accumulator incrementally. We
394−-- can repeatedly call 'addStream' on the same fold to continue building the
395−-- fold and finally use 'drive' to finish the fold and extract the result. Also
396−-- see the 'Streamly.Data.Fold.addOne' operation which is a singleton version
401−-- >>> addStream stream = Fold.drive stream . Fold.duplicate
403−-- Example, build a list incrementally:
406−-- pure (Fold.toList :: Fold IO Int [Int])
408−-- >>= Fold.addStream (Stream.enumerateFromTo 2 4)
409−-- >>= Fold.drive Stream.nil
414−-- This can be used as an O(n) list append compared to the O(n^2) @++@ when
415−-- used for incrementally building a list.
417−-- Example, build a stream incrementally:
420−-- pure (Fold.toStream :: Fold IO Int (Stream Identity Int))
422−-- >>= Fold.addStream (Stream.enumerateFromTo 2 4)
423−-- >>= Fold.drive Stream.nil
426−-- fromList [1,2,3,4]
428−-- This can be used as an O(n) stream append compared to the O(n^2) @<>@ when
429−-- used for incrementally building a stream.
431−-- Example, build an array incrementally:
434−-- pure (Array.write :: Fold IO Int (Array Int))
436−-- >>= Fold.addStream (Stream.enumerateFromTo 2 4)
437−-- >>= Fold.drive Stream.nil
440−-- fromList [1,2,3,4]
442−-- Example, build an array stream incrementally:
445−-- let f :: Fold IO Int (Stream Identity (Array Int))
446−-- f = Fold.groupsOf 2 (Array.writeN 3) Fold.toStream
449−-- >>= Fold.addStream (Stream.enumerateFromTo 2 4)
450−-- >>= Fold.drive Stream.nil
453−-- fromList [fromList [1,2],fromList [3,4]]
455−addStream :: Monad m => Stream m a -> Fold m a b -> m (Fold m a b)
456−addStream stream = drive stream . duplicate
458−------------------------------------------------------------------------------
459−-- Transformations on fold inputs
460−------------------------------------------------------------------------------
462−-- | Flatten the monadic output of a fold to pure output.
464−{-# DEPRECATED sequence "Use \"rmapM id\" instead" #-}
465−{-# INLINE sequence #-}
466−sequence :: Monad m => Fold m a (m b) -> Fold m a b
469−-- | Map a monadic function on the output of a fold.
471−{-# DEPRECATED mapM "Use rmapM instead" #-}
473−mapM :: Monad m => (b -> m c) -> Fold m a b -> Fold m a c
477−-- >>> mapMaybeM f = Fold.lmapM f . Fold.catMaybes
479−{-# INLINE mapMaybeM #-}
480−mapMaybeM :: Monad m => (a -> m (Maybe b)) -> Fold m b r -> Fold m a r
481−mapMaybeM f = lmapM f . catMaybes
483−-- | @mapMaybe f fold@ maps a 'Maybe' returning function @f@ on the input of
484−-- the fold, filters out 'Nothing' elements, and return the values extracted
487−-- >>> mapMaybe f = Fold.lmap f . Fold.catMaybes
488−-- >>> mapMaybe f = Fold.mapMaybeM (return . f)
490−-- >>> f x = if even x then Just x else Nothing
491−-- >>> fld = Fold.mapMaybe f Fold.toList
492−-- >>> Stream.fold fld (Stream.enumerateFromTo 1 10)
495−{-# INLINE mapMaybe #-}
496−mapMaybe :: Monad m => (a -> Maybe b) -> Fold m b r -> Fold m a r
497−mapMaybe f = lmap f . catMaybes
499−------------------------------------------------------------------------------
500−-- Transformations on fold inputs
501−------------------------------------------------------------------------------
503−-- | Apply a monadic function on the input and return the input.
505−-- >>> Stream.fold (Fold.lmapM (Fold.tracing print) Fold.drain) $ (Stream.enumerateFromTo (1 :: Int) 2)
511−{-# INLINE tracing #-}
512−tracing :: Monad m => (a -> m b) -> (a -> m a)
513−tracing f x = void (f x) >> return x
515−-- | Apply a monadic function to each element flowing through and discard the
518−-- >>> Stream.fold (Fold.trace print Fold.drain) $ (Stream.enumerateFromTo (1 :: Int) 2)
522−-- >>> trace f = Fold.lmapM (Fold.tracing f)
526−trace :: Monad m => (a -> m b) -> Fold m a r -> Fold m a r
527−trace f = lmapM (tracing f)
531−-- | Apply a transformation on a 'Fold' using a 'Pipe'.
534−{-# INLINE transform #-}
535−transform :: Monad m => Pipe m a b -> Fold m b c -> Fold m a c
536−transform (Pipe pstep1 pstep2 pinitial) (Fold fstep finitial fextract) =
537− Fold step initial extract
541− initial = first (Tuple' pinitial) <$> finitial
543− step (Tuple' ps fs) x = do
550− go acc (Pipe.Yield b (Consume ps')) = do
554− Partial s -> Partial $ Tuple' ps' s
556− go acc (Pipe.Yield b (Produce ps')) = do
561− Done b2 -> return $ Done b2
562− go acc (Pipe.Continue (Consume ps')) =
563− return $ Partial $ Tuple' ps' acc
564− go acc (Pipe.Continue (Produce ps')) = do
568− extract (Tuple' _ fs) = fextract fs
570−{-# INLINE scanWith #-}
571−scanWith :: Monad m => Bool -> Fold m a b -> Fold m b c -> Fold m a c
572−scanWith isMany (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
573− Fold step initial extract
577− {-# INLINE runStep #-}
578− runStep actionL sR = do
586− then runStep initialL sR1
587− else Done <$> extractR sR1
588− Done bR -> return $ Done bR
594− Partial sR1 -> Partial (sL, sR1)
600− Partial sR -> runStep initialL sR
601− Done b -> return $ Done b
603− step (sL, sR) x = runStep (stepL sL x) sR
605− extract = extractR . snd
607−-- | Scan the input of a 'Fold' to change it in a stateful manner using another
608−-- 'Fold'. The scan stops as soon as the fold terminates.
612−scan :: Monad m => Fold m a b -> Fold m b c -> Fold m a c
613−scan = scanWith False
615−-- XXX This does not fuse beacuse of the recursive step. Need to investigate.
617−-- | Scan the input of a 'Fold' to change it in a stateful manner using another
618−-- 'Fold'. The scan restarts with a fresh state if the fold terminates.
621−{-# INLINE scanMany #-}
622−scanMany :: Monad m => Fold m a b -> Fold m b c -> Fold m a c
623−scanMany = scanWith True
625−------------------------------------------------------------------------------
627−------------------------------------------------------------------------------
629−-- | Returns the latest element omitting the first occurrence that satisfies
630−-- the given equality predicate.
634−-- >>> input = Stream.fromList [1,3,3,5]
635−-- >>> Stream.fold Fold.toList $ Stream.scanMaybe (Fold.deleteBy (==) 3) input
638−{-# INLINE_NORMAL deleteBy #-}
639−deleteBy :: Monad m => (a -> a -> Bool) -> a -> Fold m a (Maybe a)
640−deleteBy eq x0 = fmap extract $ foldl' step (Tuple' False Nothing)
644− step (Tuple' False _) x =
646− then Tuple' True Nothing
647− else Tuple' False (Just x)
648− step (Tuple' True _) x = Tuple' True (Just x)
650− extract (Tuple' _ x) = x
652−-- | Provide a sliding window of length 2 elements.
654−-- See "Streamly.Internal.Data.Fold.Window".
656−{-# INLINE slide2 #-}
657−slide2 :: Monad m => Fold m (a, Maybe a) b -> Fold m a b
658−slide2 (Fold step1 initial1 extract1) = Fold step initial extract
663− first (Tuple' Nothing) <$> initial1
665− step (Tuple' prev s) cur =
666− first (Tuple' (Just cur)) <$> step1 s (cur, prev)
668− extract (Tuple' _ s) = extract1 s
670−-- | Return the latest unique element using the supplied comparison function.
671−-- Returns 'Nothing' if the current element is same as the last element
672−-- otherwise returns 'Just'.
674−-- Example, strip duplicate path separators:
676−-- >>> input = Stream.fromList "//a//b"
677−-- >>> f x y = x == '/' && y == '/'
678−-- >>> Stream.fold Fold.toList $ Stream.scanMaybe (Fold.uniqBy f) input
685−{-# INLINE uniqBy #-}
686−uniqBy :: Monad m => (a -> a -> Bool) -> Fold m a (Maybe a)
687−uniqBy eq = rollingMap f
693− Nothing -> Just curr
694− Just x -> if x `eq` curr then Nothing else Just curr
700−-- >>> uniq = Fold.uniqBy (==)
703−uniq :: (Monad m, Eq a) => Fold m a (Maybe a)
706−-- | Strip all leading and trailing occurrences of an element passing a
707−-- predicate and make all other consecutive occurrences uniq.
709−-- >> prune p = Stream.dropWhileAround p $ Stream.uniqBy (x y -> p x && p y)
712−-- > Stream.prune isSpace (Stream.fromList " hello world! ")
722− -- (Monad m, Eq a) =>
723− (a -> Bool) -> Fold m a (Maybe a)
724−prune = error "Not implemented yet!"
726−-- | Emit only repeated elements, once.
729−repeated :: -- (Monad m, Eq a) =>
731−repeated = error "Not implemented yet!"
733−------------------------------------------------------------------------------
735−------------------------------------------------------------------------------
737−------------------------------------------------------------------------------
739−------------------------------------------------------------------------------
744−-- >>> drainMapM f = Fold.lmapM f Fold.drain
745−-- >>> drainMapM f = Fold.foldMapM (void . f)
747−-- Drain all input after passing it through a monadic function. This is the
748−-- dual of mapM_ on stream producers.
750−{-# INLINE drainMapM #-}
751−drainMapM :: Monad m => (a -> m b) -> Fold m a ()
752−drainMapM f = lmapM f drain
754−{-# DEPRECATED drainBy "Please use 'drainMapM' instead." #-}
755−{-# INLINE drainBy #-}
756−drainBy :: Monad m => (a -> m b) -> Fold m a ()
759−-- | Returns the latest element of the input stream, if any.
761−-- >>> latest = Fold.foldl1' (\_ x -> x)
762−-- >>> latest = fmap getLast $ Fold.foldMap (Last . Just)
764−{-# INLINE latest #-}
765−latest :: Monad m => Fold m a (Maybe a)
766−latest = foldl1' (\_ x -> x)
768−{-# DEPRECATED last "Please use 'latest' instead." #-}
770−last :: Monad m => Fold m a (Maybe a)
773−-- | Terminates with 'Nothing' as soon as it finds an element different than
774−-- the previous one, returns 'the' element if the entire input consists of the
778−the :: (Monad m, Eq a) => Fold m a (Maybe a)
779−the = foldt' step initial id
783− initial = Partial Nothing
785− step Nothing x = Partial (Just x)
786− step old@(Just x0) x =
791−------------------------------------------------------------------------------
793−------------------------------------------------------------------------------
795−-- | Like 'length', except with a more general 'Num' return value
799−-- >>> lengthGeneric = fmap getSum $ Fold.foldMap (Sum . const 1)
800−-- >>> lengthGeneric = Fold.foldl' (\n _ -> n + 1) 0
803−{-# INLINE lengthGeneric #-}
804−lengthGeneric :: (Monad m, Num b) => Fold m a b
805−lengthGeneric = foldl' (\n _ -> n + 1) 0
807−-- | Determine the length of the input stream.
811−-- >>> length = Fold.lengthGeneric
812−-- >>> length = fmap getSum $ Fold.foldMap (Sum . const 1)
814−{-# INLINE length #-}
815−length :: Monad m => Fold m a Int
816−length = lengthGeneric
819−-- | Determine the sum of all elements of a stream of numbers. Returns additive
820−-- identity (@0@) when the stream is empty. Note that this is not numerically
821−-- stable for floating point numbers.
823−-- >>> sum = FoldW.cumulative FoldW.sum
825−-- Same as following but numerically stable:
827−-- >>> sum = Fold.foldl' (+) 0
828−-- >>> sum = fmap Data.Monoid.getSum $ Fold.foldMap Data.Monoid.Sum
831−sum :: (Monad m, Num a) => Fold m a a
832−sum = FoldW.cumulative FoldW.sum
834−-- | Determine the product of all elements of a stream of numbers. Returns
835−-- multiplicative identity (@1@) when the stream is empty. The fold terminates
836−-- when it encounters (@0@) in its input.
838−-- Same as the following but terminates on multiplication by @0@:
840−-- >>> product = fmap Data.Monoid.getProduct $ Fold.foldMap Data.Monoid.Product
842−{-# INLINE product #-}
843−product :: (Monad m, Num a, Eq a) => Fold m a a
844−product = foldt' step (Partial 1) id
851− else Partial $ x * a
853−------------------------------------------------------------------------------
854−-- To Summary (Maybe)
855−------------------------------------------------------------------------------
857−-- | Determine the maximum element in a stream using the supplied comparison
860−{-# INLINE maximumBy #-}
861−maximumBy :: Monad m => (a -> a -> Ordering) -> Fold m a (Maybe a)
862−maximumBy cmp = foldl1' max'
871−-- | Determine the maximum element in a stream.
875−-- >>> maximum = Fold.maximumBy compare
876−-- >>> maximum = Fold.foldl1' max
878−-- Same as the following but without a default maximum. The 'Max' Monoid uses
879−-- the 'minBound' as the default maximum:
881−-- >>> maximum = fmap Data.Semigroup.getMax $ Fold.foldMap Data.Semigroup.Max
883−{-# INLINE maximum #-}
884−maximum :: (Monad m, Ord a) => Fold m a (Maybe a)
885−maximum = foldl1' max
887−-- | Computes the minimum element with respect to the given comparison function
889−{-# INLINE minimumBy #-}
890−minimumBy :: Monad m => (a -> a -> Ordering) -> Fold m a (Maybe a)
891−minimumBy cmp = foldl1' min'
900−-- | Determine the minimum element in a stream using the supplied comparison
905−-- >>> minimum = Fold.minimumBy compare
906−-- >>> minimum = Fold.foldl1' min
908−-- Same as the following but without a default minimum. The 'Min' Monoid uses the
909−-- 'maxBound' as the default maximum:
911−-- >>> maximum = fmap Data.Semigroup.getMin $ Fold.foldMap Data.Semigroup.Min
913−{-# INLINE minimum #-}
914−minimum :: (Monad m, Ord a) => Fold m a (Maybe a)
915−minimum = foldl1' min
917−------------------------------------------------------------------------------
918−-- To Summary (Statistical)
919−------------------------------------------------------------------------------
921−-- | Compute a numerically stable arithmetic mean of all elements in the input
925−mean :: (Monad m, Fractional a) => Fold m a a
926−mean = fmap done $ foldl' step begin
932− step (Tuple' x n) y =
934− in Tuple' (x + (y - x) / n1) n1
936− done (Tuple' x _) = x
938−-- | Compute a numerically stable (population) variance over all elements in
941−{-# DEPRECATED variance "Use the streamly-statistics package instead" #-}
942−{-# INLINE variance #-}
943−variance :: (Monad m, Fractional a) => Fold m a a
944−variance = fmap done $ foldl' step begin
948− begin = Tuple3' 0 0 0
950− step (Tuple3' n mean_ m2) x = Tuple3' n' mean' m2'
955− mean' = (n * mean_ + x) / (n + 1)
957− m2' = m2 + delta * delta * n / (n + 1)
959− done (Tuple3' n _ m2) = m2 / n
961−-- | Compute a numerically stable (population) standard deviation over all
962−-- elements in the input stream.
964−{-# DEPRECATED stdDev "Use the streamly-statistics package instead" #-}
965−{-# INLINE stdDev #-}
966−stdDev :: (Monad m, Floating a) => Fold m a a
967−stdDev = sqrt <$> variance
969−-- | Compute an 'Int' sized polynomial rolling hash
971−-- > H = salt * k ^ n + c1 * k ^ (n - 1) + c2 * k ^ (n - 2) + ... + cn * k ^ 0
973−-- Where @c1@, @c2@, @cn@ are the elements in the input stream and @k@ is a
976−-- This hash is often used in Rabin-Karp string search algorithm.
978−-- See https://en.wikipedia.org/wiki/Rolling_hash
980−{-# INLINE rollingHashWithSalt #-}
981−rollingHashWithSalt :: (Monad m, Enum a) => Int64 -> Fold m a Int64
982−rollingHashWithSalt = foldl' step
986− k = 2891336453 :: Int64
988− step cksum a = cksum * k + fromIntegral (fromEnum a)
990−-- | A default salt used in the implementation of 'rollingHash'.
991−{-# INLINE defaultSalt #-}
993−defaultSalt = -2578643520546668380
995−-- | Compute an 'Int' sized polynomial rolling hash of a stream.
997−-- >>> rollingHash = Fold.rollingHashWithSalt Fold.defaultSalt
999−{-# INLINE rollingHash #-}
1000−rollingHash :: (Monad m, Enum a) => Fold m a Int64
1001−rollingHash = rollingHashWithSalt defaultSalt
1003−-- | Compute an 'Int' sized polynomial rolling hash of the first n elements of
1006−-- >>> rollingHashFirstN n = Fold.take n Fold.rollingHash
1009−{-# INLINE rollingHashFirstN #-}
1010−rollingHashFirstN :: (Monad m, Enum a) => Int -> Fold m a Int64
1011−rollingHashFirstN n = take n rollingHash
1013−-- XXX Compare this with the implementation in Fold.Window, preferrably use the
1014−-- latter if performance is good.
1016−-- | Apply a function on every two successive elements of a stream. The first
1017−-- argument of the map function is the previous element and the second argument
1018−-- is the current element. When processing the very first element in the
1019−-- stream, the previous element is 'Nothing'.
1023−{-# INLINE rollingMapM #-}
1024−rollingMapM :: Monad m => (Maybe a -> a -> m b) -> Fold m a b
1025−rollingMapM f = Fold step initial extract
1029− -- XXX We need just a postscan. We do not need an initial result here.
1030− -- Or we can supply a default initial result as an argument to rollingMapM.
1031− initial = return $ Partial (Nothing, error "Empty stream")
1033− step (prev, _) cur = do
1035− return $ Partial (Just cur, x)
1037− extract = return . snd
1040−-- >>> rollingMap f = Fold.rollingMapM (\x y -> return $ f x y)
1042−{-# INLINE rollingMap #-}
1043−rollingMap :: Monad m => (Maybe a -> a -> b) -> Fold m a b
1044−rollingMap f = rollingMapM (\x y -> return $ f x y)
1046−------------------------------------------------------------------------------
1047−-- Monoidal left folds
1048−------------------------------------------------------------------------------
1050−-- | Semigroup concat. Append the elements of an input stream to a provided
1055−-- >>> sconcat = Fold.foldl' (<>)
1057−-- >>> semigroups = fmap Data.Monoid.Sum $ Stream.enumerateFromTo 1 10
1058−-- >>> Stream.fold (Fold.sconcat 10) semigroups
1059−-- Sum {getSum = 65}
1061−{-# INLINE sconcat #-}
1062−sconcat :: (Monad m, Semigroup a) => a -> Fold m a a
1063−sconcat = foldl' (<>)
1065−-- | Monoid concat. Fold an input stream consisting of monoidal elements using
1066−-- 'mappend' and 'mempty'.
1070−-- >>> mconcat = Fold.sconcat mempty
1072−-- >>> monoids = fmap Data.Monoid.Sum $ Stream.enumerateFromTo 1 10
1073−-- >>> Stream.fold Fold.mconcat monoids
1074−-- Sum {getSum = 55}
1076−{-# INLINE mconcat #-}
1079− , Monoid a) => Fold m a a
1080−mconcat = sconcat mempty
1085−-- >>> foldMap f = Fold.lmap f Fold.mconcat
1087−-- Make a fold from a pure function that folds the output of the function
1088−-- using 'mappend' and 'mempty'.
1090−-- >>> sum = Fold.foldMap Data.Monoid.Sum
1091−-- >>> Stream.fold sum $ Stream.enumerateFromTo 1 10
1092−-- Sum {getSum = 55}
1094−{-# INLINE foldMap #-}
1095−foldMap :: (Monad m, Monoid b) => (a -> b) -> Fold m a b
1096−foldMap f = lmap f mconcat
1101−-- >>> foldMapM f = Fold.lmapM f Fold.mconcat
1103−-- Make a fold from a monadic function that folds the output of the function
1104−-- using 'mappend' and 'mempty'.
1106−-- >>> sum = Fold.foldMapM (return . Data.Monoid.Sum)
1107−-- >>> Stream.fold sum $ Stream.enumerateFromTo 1 10
1108−-- Sum {getSum = 55}
1110−{-# INLINE foldMapM #-}
1111−foldMapM :: (Monad m, Monoid b) => (a -> m b) -> Fold m a b
1112−foldMapM act = foldlM' step (pure mempty)
1118− return $! mappend m m'
1120−------------------------------------------------------------------------------
1122−------------------------------------------------------------------------------
1125−-- This is more efficient than 'Streamly.Internal.Data.Fold.toList'. toList is
1126−-- exactly the same as reversing the list after 'toListRev'.
1128−-- | Buffers the input stream to a list in the reverse order of the input.
1132−-- >>> toListRev = Fold.foldl' (flip (:)) []
1134−-- /Warning!/ working on large lists accumulated as buffers in memory could be
1135−-- very inefficient, consider using "Streamly.Array" instead.
1138−-- xn : ... : x2 : x1 : []
1139−{-# INLINE toListRev #-}
1140−toListRev :: Monad m => Fold m a [a]
1141−toListRev = foldl' (flip (:)) []
1143−------------------------------------------------------------------------------
1145−------------------------------------------------------------------------------
1147−-- | A fold that drains the first n elements of its input, running the effects
1148−-- and discarding the results.
1152−-- >>> drainN n = Fold.take n Fold.drain
1155−{-# INLINE drainN #-}
1156−drainN :: Monad m => Int -> Fold m a ()
1157−drainN n = take n drain
1159−------------------------------------------------------------------------------
1161−------------------------------------------------------------------------------
1163−-- | Like 'index', except with a more general 'Integral' argument
1166−{-# INLINE indexGeneric #-}
1167−indexGeneric :: (Integral i, Monad m) => i -> Fold m a (Maybe a)
1168−indexGeneric i = foldt' step (Partial 0) (const Nothing)
1175− else Partial (j + 1)
1177−-- | Return the element at the given index.
1181−-- >>> index = Fold.indexGeneric
1183−{-# INLINE index #-}
1184−index :: Monad m => Int -> Fold m a (Maybe a)
1185−index = indexGeneric
1187−-- | Consume a single input and transform it using the supplied 'Maybe'
1188−-- returning function.
1192−{-# INLINE maybe #-}
1193−maybe :: Monad m => (a -> Maybe b) -> Fold m a (Maybe b)
1194−maybe f = foldt' (const (Done . f)) (Partial Nothing) id
1196−-- | Consume a single element and return it if it passes the predicate else
1197−-- return 'Nothing'.
1201−-- >>> satisfy f = Fold.maybe (\a -> if f a then Just a else Nothing)
1204−{-# INLINE satisfy #-}
1205−satisfy :: Monad m => (a -> Bool) -> Fold m a (Maybe a)
1206−satisfy f = maybe (\a -> if f a then Just a else Nothing)
1208−satisfy f = Fold step (return $ Partial ()) (const (return Nothing))
1212− step () a = return $ Done $ if f a then Just a else Nothing
1217−-- "head" and "next" are two alternative names for the same API. head sounds
1218−-- apt in the context of lists but next sounds more apt in the context of
1219−-- streams where we think in terms of generating and consuming the next element
1220−-- rather than taking the head of some static/persistent structure.
1222−-- We also want to keep the nomenclature consistent across folds and parsers,
1223−-- "head" becomes even more unintuitive for parsers because there are two
1224−-- possible variants viz. peek and next.
1226−-- Also, the "head" fold creates confusion in situations like
1227−-- https://github.com/composewell/streamly/issues/1404 where intuitive
1228−-- expectation from head is to consume the entire stream and just give us the
1229−-- head. There we want to convey the notion that we consume one element from
1230−-- the stream and stop. The name "one" already being used in parsers for this
1231−-- purpose sounds more apt from this perspective.
1233−-- The source of confusion is perhaps due to the fact that some folds consume
1234−-- the entire stream and others terminate early. It may have been clearer if we
1235−-- had separate abstractions for the two use cases.
1237−-- XXX We can possibly use "head" for the purposes of reducing the entire
1238−-- stream to the head element i.e. take the head and drain the rest.
1240−-- | Take one element from the stream and stop.
1244−-- >>> one = Fold.maybe Just
1246−-- This is similar to the stream 'Stream.uncons' operation.
1249−one :: Monad m => Fold m a (Maybe a)
1252−-- | Extract the first element of the stream, if any.
1254−-- >>> head = Fold.one
1256−{-# DEPRECATED head "Please use \"one\" instead" #-}
1258−head :: Monad m => Fold m a (Maybe a)
1261−-- | Returns the first element that satisfies the given predicate.
1264−{-# INLINE findM #-}
1265−findM :: Monad m => (a -> m Bool) -> Fold m a (Maybe a)
1266−findM predicate = Fold step (return $ Partial ()) (const $ return Nothing)
1275− in f <$> predicate a
1277−-- | Returns the first element that satisfies the given predicate.
1280−find :: Monad m => (a -> Bool) -> Fold m a (Maybe a)
1281−find p = findM (return . p)
1283−-- | In a stream of (key-value) pairs @(a, b)@, return the value @b@ of the
1284−-- first pair where the key equals the given value @a@.
1288−-- >>> lookup x = fmap snd <$> Fold.find ((== x) . fst)
1290−{-# INLINE lookup #-}
1291−lookup :: (Eq a, Monad m) => a -> Fold m (a,b) (Maybe b)
1292−lookup a0 = foldt' step (Partial ()) (const Nothing)
1301−-- | Returns the first index that satisfies the given predicate.
1303−{-# INLINE findIndex #-}
1304−findIndex :: Monad m => (a -> Bool) -> Fold m a (Maybe Int)
1305−findIndex predicate = foldt' step (Partial 0) (const Nothing)
1312− else Partial (i + 1)
1314−-- | Returns the index of the latest element if the element satisfies the given
1317−{-# INLINE findIndices #-}
1318−findIndices :: Monad m => (a -> Bool) -> Fold m a (Maybe Int)
1319−findIndices predicate =
1320− -- XXX implement by combining indexing and filtering scans
1321− fmap (either (const Nothing) Just) $ foldl' step (Left (-1))
1327− then Right (either id id i + 1)
1328− else Left (either id id i + 1)
1330−-- | Returns the index of the latest element if the element matches the given
1335−-- >>> elemIndices a = Fold.findIndices (== a)
1337−{-# INLINE elemIndices #-}
1338−elemIndices :: (Monad m, Eq a) => a -> Fold m a (Maybe Int)
1339−elemIndices a = findIndices (== a)
1341−-- | Returns the first index where a given value is found in the stream.
1345−-- >>> elemIndex a = Fold.findIndex (== a)
1347−{-# INLINE elemIndex #-}
1348−elemIndex :: (Eq a, Monad m) => a -> Fold m a (Maybe Int)
1349−elemIndex a = findIndex (== a)
1351−------------------------------------------------------------------------------
1353−------------------------------------------------------------------------------
1355−-- Similar to 'eof' parser, but the fold consumes and discards an input element
1356−-- when not at eof. XXX Remove or Rename to "eof"?
1358−-- | Consume one element, return 'True' if successful else return 'False'. In
1359−-- other words, test if the input is empty or not.
1361−-- WARNING! It consumes one element if the stream is not empty. If that is not
1362−-- what you want please use the eof parser instead.
1366−-- >>> null = fmap isJust Fold.one
1369−null :: Monad m => Fold m a Bool
1370−null = foldt' (\() _ -> Done False) (Partial ()) (const True)
1372−-- | Returns 'True' if any element of the input satisfies the predicate.
1376−-- >>> any p = Fold.lmap p Fold.or
1380−-- >>> Stream.fold (Fold.any (== 0)) $ Stream.fromList [1,0,1]
1384−any :: Monad m => (a -> Bool) -> Fold m a Bool
1385−any predicate = foldt' step initial id
1389− initial = Partial False
1396−-- | Return 'True' if the given element is present in the stream.
1400−-- >>> elem a = Fold.any (== a)
1403−elem :: (Eq a, Monad m) => a -> Fold m a Bool
1406−-- | Returns 'True' if all elements of the input satisfy the predicate.
1410−-- >>> all p = Fold.lmap p Fold.and
1414−-- >>> Stream.fold (Fold.all (== 0)) $ Stream.fromList [1,0,1]
1418−all :: Monad m => (a -> Bool) -> Fold m a Bool
1419−all predicate = foldt' step initial id
1423− initial = Partial True
1430−-- | Returns 'True' if the given element is not present in the stream.
1434−-- >>> notElem a = Fold.all (/= a)
1436−{-# INLINE notElem #-}
1437−notElem :: (Eq a, Monad m) => a -> Fold m a Bool
1438−notElem a = all (/= a)
1440−-- | Returns 'True' if all elements are 'True', 'False' otherwise
1444−-- >>> and = Fold.all (== True)
1447−and :: Monad m => Fold m Bool Bool
1450−-- | Returns 'True' if any element is 'True', 'False' otherwise
1454−-- >>> or = Fold.any (== True)
1457−or :: Monad m => Fold m Bool Bool
1460−------------------------------------------------------------------------------
1461−-- Grouping/Splitting
1462−------------------------------------------------------------------------------
1464−------------------------------------------------------------------------------
1465−-- Grouping without looking at elements
1466−------------------------------------------------------------------------------
1468−------------------------------------------------------------------------------
1470−------------------------------------------------------------------------------
1472−-- | @splitAt n f1 f2@ composes folds @f1@ and @f2@ such that first @n@
1473−-- elements of its input are consumed by fold @f1@ and the rest of the stream
1474−-- is consumed by fold @f2@.
1476−-- >>> let splitAt_ n xs = Stream.fold (Fold.splitAt n Fold.toList Fold.toList) $ Stream.fromList xs
1478−-- >>> splitAt_ 6 "Hello World!"
1479−-- ("Hello ","World!")
1481−-- >>> splitAt_ (-1) [1,2,3]
1484−-- >>> splitAt_ 0 [1,2,3]
1487−-- >>> splitAt_ 1 [1,2,3]
1490−-- >>> splitAt_ 3 [1,2,3]
1493−-- >>> splitAt_ 4 [1,2,3]
1496−-- > splitAt n f1 f2 = Fold.splitWith (,) (Fold.take n f1) f2
1500−{-# INLINE splitAt #-}
1507−splitAt n fld = splitWith (,) (take n fld)
1509−------------------------------------------------------------------------------
1510−-- Element Aware APIs
1511−------------------------------------------------------------------------------
1513−------------------------------------------------------------------------------
1515−------------------------------------------------------------------------------
1517−{-# INLINE takingEndByM #-}
1518−takingEndByM :: Monad m => (a -> m Bool) -> Fold m a (Maybe a)
1519−takingEndByM p = Fold step initial (return . toMaybe)
1523− initial = return $ Partial Nothing'
1530− else Partial $ Just' a
1534−-- >>> takingEndBy p = Fold.takingEndByM (return . p)
1536−{-# INLINE takingEndBy #-}
1537−takingEndBy :: Monad m => (a -> Bool) -> Fold m a (Maybe a)
1538−takingEndBy p = takingEndByM (return . p)
1540−{-# INLINE takingEndByM_ #-}
1541−takingEndByM_ :: Monad m => (a -> m Bool) -> Fold m a (Maybe a)
1542−takingEndByM_ p = Fold step initial (return . toMaybe)
1546− initial = return $ Partial Nothing'
1553− else Partial $ Just' a
1557−-- >>> takingEndBy_ p = Fold.takingEndByM_ (return . p)
1559−{-# INLINE takingEndBy_ #-}
1560−takingEndBy_ :: Monad m => (a -> Bool) -> Fold m a (Maybe a)
1561−takingEndBy_ p = takingEndByM_ (return . p)
1563−{-# INLINE droppingWhileM #-}
1564−droppingWhileM :: Monad m => (a -> m Bool) -> Fold m a (Maybe a)
1565−droppingWhileM p = Fold step initial (return . toMaybe)
1569− initial = return $ Partial Nothing'
1571− step Nothing' a = do
1578− step _ a = return $ Partial $ Just' a
1581−-- >>> droppingWhile p = Fold.droppingWhileM (return . p)
1583−{-# INLINE droppingWhile #-}
1584−droppingWhile :: Monad m => (a -> Bool) -> Fold m a (Maybe a)
1585−droppingWhile p = droppingWhileM (return . p)
1587−-- Note: Keep this consistent with S.splitOn. In fact we should eliminate
1588−-- S.splitOn in favor of the fold.
1590−-- XXX Use Fold.many instead once it is fixed.
1591−-- > Stream.splitOnSuffix p f = Stream.foldMany (Fold.takeEndBy_ p f)
1593−-- | Like 'takeEndBy' but drops the element on which the predicate succeeds.
1597−-- >>> input = Stream.fromList "hello\nthere\n"
1598−-- >>> line = Fold.takeEndBy_ (== '\n') Fold.toList
1599−-- >>> Stream.fold line input
1602−-- >>> Stream.fold Fold.toList $ Stream.foldMany line input
1603−-- ["hello","there"]
1605−{-# INLINE takeEndBy_ #-}
1606−takeEndBy_ :: Monad m => (a -> Bool) -> Fold m a b -> Fold m a b
1607−-- takeEndBy_ predicate = scanMaybe (takingEndBy_ predicate)
1608−takeEndBy_ predicate (Fold fstep finitial fextract) =
1609− Fold step finitial fextract
1614− if not (predicate a)
1616− else Done <$> fextract s
1619−-- > Stream.splitWithSuffix p f = Stream.foldMany (Fold.takeEndBy p f)
1621−-- | Take the input, stop when the predicate succeeds taking the succeeding
1626−-- >>> input = Stream.fromList "hello\nthere\n"
1627−-- >>> line = Fold.takeEndBy (== '\n') Fold.toList
1628−-- >>> Stream.fold line input
1631−-- >>> Stream.fold Fold.toList $ Stream.foldMany line input
1632−-- ["hello\n","there\n"]
1634−{-# INLINE takeEndBy #-}
1635−takeEndBy :: Monad m => (a -> Bool) -> Fold m a b -> Fold m a b
1636−-- takeEndBy predicate = scanMaybe (takingEndBy predicate)
1637−takeEndBy predicate (Fold fstep finitial fextract) =
1638− Fold step finitial fextract
1644− if not (predicate a)
1648− Partial s1 -> Done <$> fextract s1
1649− Done b -> return $ Done b
1651−------------------------------------------------------------------------------
1652−-- Binary splitting on a separator
1653−------------------------------------------------------------------------------
1655−data SplitOnSeqState acc a rb rh w ck =
1656− SplitOnSeqEmpty !acc
1657− | SplitOnSeqSingle !acc !a
1658− | SplitOnSeqWord !acc !Int !w
1659− | SplitOnSeqWordLoop !acc !w
1660− | SplitOnSeqKR !acc !Int !rb !rh
1661− | SplitOnSeqKRLoop !acc !ck !rb !rh
1663−-- XXX Need to add tests for takeEndBySeq, we have tests for takeEndBySeq_ .
1665−-- | Continue taking the input until the input sequence matches the supplied
1666−-- sequence, taking the supplied sequence as well. If the pattern is empty this
1667−-- acts as an identity fold.
1669−-- >>> s = Stream.fromList "hello there. How are you?"
1670−-- >>> f = Fold.takeEndBySeq (Array.fromList "re") Fold.toList
1671−-- >>> Stream.fold f s
1674−-- >>> Stream.fold Fold.toList $ Stream.foldMany f s
1675−-- ["hello there",". How are"," you?"]
1678−{-# INLINE takeEndBySeq #-}
1679−takeEndBySeq :: forall m a b. (MonadIO m, Storable a, Unbox a, Enum a, Eq a) =>
1683−takeEndBySeq patArr (Fold fstep finitial fextract) =
1684− Fold step initial extract
1688− patLen = Array.length patArr
1695− -- XXX Should we match nothing or everything on empty
1697− -- Done <$> fextract acc
1698− return $ Partial $ SplitOnSeqEmpty acc
1699− | patLen == 1 -> do
1700− pat <- liftIO $ Array.unsafeIndexIO 0 patArr
1701− return $ Partial $ SplitOnSeqSingle acc pat
1702− | SIZE_OF(a) * patLen <= sizeOf (Proxy :: Proxy Word) ->
1703− return $ Partial $ SplitOnSeqWord acc 0 0
1705− (rb, rhead) <- liftIO $ Ring.new patLen
1706− return $ Partial $ SplitOnSeqKR acc 0 rb rhead
1707− Done b -> return $ Done b
1709− -- Word pattern related
1710− maxIndex = patLen - 1
1712− elemBits = SIZE_OF(a) * 8
1715− wordMask = (1 `shiftL` (elemBits * patLen)) - 1
1718− wordPat = wordMask .&. Array.foldl' addToWord 0 patArr
1720− addToWord wd a = (wd `shiftL` elemBits) .|. fromIntegral (fromEnum a)
1722− -- For Rabin-Karp search
1723− k = 2891336453 :: Word32
1726− addCksum cksum a = cksum * k + fromIntegral (fromEnum a)
1728− deltaCksum cksum old new =
1729− addCksum cksum new - coeff * fromIntegral (fromEnum old)
1731− -- XXX shall we use a random starting hash or 1 instead of 0?
1732− -- XXX Need to keep this cached across fold calls in foldmany
1733− -- XXX We may need refold to inject the cached state instead of
1734− -- initializing the state every time.
1735− -- XXX Allocation of ring buffer should also be done once
1736− patHash = Array.foldl' addCksum 0 patArr
1738− step (SplitOnSeqEmpty s) x = do
1741− Partial s1 -> return $ Partial $ SplitOnSeqEmpty s1
1742− Done b -> return $ Done b
1743− step (SplitOnSeqSingle s pat) x = do
1747− | pat /= x -> return $ Partial $ SplitOnSeqSingle s1 pat
1748− | otherwise -> Done <$> fextract s1
1749− Done b -> return $ Done b
1750− step (SplitOnSeqWord s idx wrd) x = do
1752− let wrd1 = addToWord wrd x
1755− | idx == maxIndex -> do
1756− if wrd1 .&. wordMask == wordPat
1757− then Done <$> fextract s1
1758− else return $ Partial $ SplitOnSeqWordLoop s1 wrd1
1760− return $ Partial $ SplitOnSeqWord s1 (idx + 1) wrd1
1761− Done b -> return $ Done b
1762− step (SplitOnSeqWordLoop s wrd) x = do
1764− let wrd1 = addToWord wrd x
1767− | wrd1 .&. wordMask == wordPat ->
1768− Done <$> fextract s1
1770− return $ Partial $ SplitOnSeqWordLoop s1 wrd1
1771− Done b -> return $ Done b
1772− step (SplitOnSeqKR s idx rb rh) x = do
1776− rh1 <- liftIO $ Ring.unsafeInsert rb rh x
1779− let fld = Ring.unsafeFoldRing (Ring.ringBound rb)
1780− let !ringHash = fld addCksum 0 rb
1781− if ringHash == patHash && Ring.unsafeEqArray rb rh1 patArr
1782− then Done <$> fextract s1
1783− else return $ Partial $ SplitOnSeqKRLoop s1 ringHash rb rh1
1785− return $ Partial $ SplitOnSeqKR s1 (idx + 1) rb rh1
1786− Done b -> return $ Done b
1787− step (SplitOnSeqKRLoop s cksum rb rh) x = do
1791− old <- liftIO $ peek rh
1792− rh1 <- liftIO $ Ring.unsafeInsert rb rh x
1793− let ringHash = deltaCksum cksum old x
1794− if ringHash == patHash && Ring.unsafeEqArray rb rh1 patArr
1795− then Done <$> fextract s1
1796− else return $ Partial $ SplitOnSeqKRLoop s1 ringHash rb rh1
1797− Done b -> return $ Done b
1802− SplitOnSeqEmpty s -> s
1803− SplitOnSeqSingle s _ -> s
1804− SplitOnSeqWord s _ _ -> s
1805− SplitOnSeqWordLoop s _ -> s
1806− SplitOnSeqKR s _ _ _ -> s
1807− SplitOnSeqKRLoop s _ _ _ -> s
1810−-- | Like 'takeEndBySeq' but discards the matched sequence.
1814−{-# INLINE takeEndBySeq_ #-}
1815−takeEndBySeq_ :: forall m a b. (MonadIO m, Storable a, Unbox a, Enum a, Eq a) =>
1819−takeEndBySeq_ patArr (Fold fstep finitial fextract) =
1820− Fold step initial extract
1824− patLen = Array.length patArr
1831− -- XXX Should we match nothing or everything on empty
1833− -- Done <$> fextract acc
1834− return $ Partial $ SplitOnSeqEmpty acc
1835− | patLen == 1 -> do
1836− pat <- liftIO $ Array.unsafeIndexIO 0 patArr
1837− return $ Partial $ SplitOnSeqSingle acc pat
1838− -- XXX Need to add tests for this case
1839− | SIZE_OF(a) * patLen <= sizeOf (Proxy :: Proxy Word) ->
1840− return $ Partial $ SplitOnSeqWord acc 0 0
1842− (rb, rhead) <- liftIO $ Ring.new patLen
1843− return $ Partial $ SplitOnSeqKR acc 0 rb rhead
1844− Done b -> return $ Done b
1846− -- Word pattern related
1847− maxIndex = patLen - 1
1849− elemBits = SIZE_OF(a) * 8
1852− wordMask = (1 `shiftL` (elemBits * patLen)) - 1
1855− elemMask = (1 `shiftL` elemBits) - 1
1858− wordPat = wordMask .&. Array.foldl' addToWord 0 patArr
1860− addToWord wd a = (wd `shiftL` elemBits) .|. fromIntegral (fromEnum a)
1862− -- For Rabin-Karp search
1863− k = 2891336453 :: Word32
1866− addCksum cksum a = cksum * k + fromIntegral (fromEnum a)
1868− deltaCksum cksum old new =
1869− addCksum cksum new - coeff * fromIntegral (fromEnum old)
1871− -- XXX shall we use a random starting hash or 1 instead of 0?
1872− -- XXX Need to keep this cached across fold calls in foldMany
1873− -- XXX We may need refold to inject the cached state instead of
1874− -- initializing the state every time.
1875− -- XXX Allocation of ring buffer should also be done once
1876− patHash = Array.foldl' addCksum 0 patArr
1878− step (SplitOnSeqEmpty s) x = do
1881− Partial s1 -> return $ Partial $ SplitOnSeqEmpty s1
1882− Done b -> return $ Done b
1883− step (SplitOnSeqSingle s pat) x = do
1888− Partial s1 -> return $ Partial $ SplitOnSeqSingle s1 pat
1889− Done b -> return $ Done b
1890− else Done <$> fextract s
1891− step (SplitOnSeqWord s idx wrd) x = do
1892− let wrd1 = addToWord wrd x
1895− if wrd1 .&. wordMask == wordPat
1896− then Done <$> fextract s
1897− else return $ Partial $ SplitOnSeqWordLoop s wrd1
1898− else return $ Partial $ SplitOnSeqWord s (idx + 1) wrd1
1899− step (SplitOnSeqWordLoop s wrd) x = do
1900− let wrd1 = addToWord wrd x
1901− old = (wordMask .&. wrd)
1902− `shiftR` (elemBits * (patLen - 1))
1903− res <- fstep s (toEnum $ fromIntegral old)
1906− | wrd1 .&. wordMask == wordPat ->
1907− Done <$> fextract s1
1909− return $ Partial $ SplitOnSeqWordLoop s1 wrd1
1910− Done b -> return $ Done b
1911− step (SplitOnSeqKR s idx rb rh) x = do
1912− rh1 <- liftIO $ Ring.unsafeInsert rb rh x
1915− let fld = Ring.unsafeFoldRing (Ring.ringBound rb)
1916− let !ringHash = fld addCksum 0 rb
1917− if ringHash == patHash && Ring.unsafeEqArray rb rh1 patArr
1918− then Done <$> fextract s
1919− else return $ Partial $ SplitOnSeqKRLoop s ringHash rb rh1
1920− else return $ Partial $ SplitOnSeqKR s (idx + 1) rb rh1
1921− step (SplitOnSeqKRLoop s cksum rb rh) x = do
1922− old <- liftIO $ peek rh
1926− rh1 <- liftIO $ Ring.unsafeInsert rb rh x
1927− let ringHash = deltaCksum cksum old x
1928− if ringHash == patHash && Ring.unsafeEqArray rb rh1 patArr
1929− then Done <$> fextract s1
1930− else return $ Partial $ SplitOnSeqKRLoop s1 ringHash rb rh1
1931− Done b -> return $ Done b
1933− -- XXX extract should return backtrack count as well. If the fold
1934− -- terminates early inside extract, we may still have buffered data
1935− -- remaining which will be lost if we do not communicate that to the
1938− let consumeWord s n wrd = do
1942− let old = elemMask .&. (wrd `shiftR` (elemBits * (n - 1)))
1943− r <- fstep s (toEnum $ fromIntegral old)
1945− Partial s1 -> consumeWord s1 (n - 1) wrd
1948− let consumeRing s n rb rh =
1952− old <- liftIO $ peek rh
1953− let rh1 = Ring.advance rb rh
1956− Partial s1 -> consumeRing s1 (n - 1) rb rh1
1960− SplitOnSeqEmpty s -> fextract s
1961− SplitOnSeqSingle s _ -> fextract s
1962− SplitOnSeqWord s idx wrd -> consumeWord s idx wrd
1963− SplitOnSeqWordLoop s wrd -> consumeWord s patLen wrd
1964− SplitOnSeqKR s idx rb _ -> consumeRing s idx rb (Ring.startOf rb)
1965− SplitOnSeqKRLoop s _ rb rh -> consumeRing s patLen rb rh
1967−------------------------------------------------------------------------------
1969−------------------------------------------------------------------------------
1971−-- | Distribute one copy of the stream to each fold and zip the results.
1974−-- |-------Fold m a b--------|
1975−-- ---stream m a---| |---m (b,c)
1976−-- |-------Fold m a c--------|
1981−-- >>> tee = Fold.teeWith (,)
1985−-- >>> t = Fold.tee Fold.sum Fold.length
1986−-- >>> Stream.fold t (Stream.enumerateFromTo 1.0 100.0)
1990−tee :: Monad m => Fold m a b -> Fold m a c -> Fold m a (b,c)
1993−-- XXX use "List" instead of "[]"?, use Array for output to scale it to a large
1994−-- number of consumers? For polymorphic case a vector could be helpful. For
1995−-- Storables we can use arrays. Will need separate APIs for those.
1997−-- | Distribute one copy of the stream to each fold and collect the results in
2002−-- |-------Fold m a b--------|
2003−-- ---stream m a---| |---m [b]
2004−-- |-------Fold m a b--------|
2009−-- >>> Stream.fold (Fold.distribute [Fold.sum, Fold.length]) (Stream.enumerateFromTo 1 5)
2012−-- >>> distribute = Prelude.foldr (Fold.teeWith (:)) (Fold.fromPure [])
2014−-- This is the consumer side dual of the producer side 'sequence' operation.
2016−-- Stops when all the folds stop.
2018−{-# INLINE distribute #-}
2019−distribute :: Monad m => [Fold m a b] -> Fold m a [b]
2020−distribute = Prelude.foldr (teeWith (:)) (fromPure [])
2022−------------------------------------------------------------------------------
2024−------------------------------------------------------------------------------
2026−{-# INLINE partitionByMUsing #-}
2027−partitionByMUsing :: Monad m =>
2028− ( (x -> y -> (x, y))
2029− -> Fold m (Either b c) x
2030− -> Fold m (Either b c) y
2031− -> Fold m (Either b c) (x, y)
2033− -> (a -> m (Either b c))
2037−partitionByMUsing t f fld1 fld2 =
2038− let l = lmap (fromLeft undefined) fld1 -- :: Fold m (Either b c) x
2039− r = lmap (fromRight undefined) fld2 -- :: Fold m (Either b c) y
2040− in lmapM f (t (,) (filter isLeft l) (filter isRight r))
2042−-- | Partition the input over two folds using an 'Either' partitioning
2047−-- |-------Fold b x--------|
2048−-- -----stream m a --> (Either b c)----| |----(x,y)
2049−-- |-------Fold c y--------|
2052−-- Example, send input to either fold randomly:
2054−-- >>> :set -package random
2055−-- >>> import System.Random (randomIO)
2056−-- >>> randomly a = randomIO >>= \x -> return $ if x then Left a else Right a
2057−-- >>> f = Fold.partitionByM randomly Fold.length Fold.length
2058−-- >>> Stream.fold f (Stream.enumerateFromTo 1 100)
2061−-- Example, send input to the two folds in a proportion of 2:1:
2064−-- proportionately m n = do
2065−-- ref <- newIORef $ cycle $ concat [replicate m Left, replicate n Right]
2066−-- return $ \a -> do
2067−-- r <- readIORef ref
2068−-- writeIORef ref $ tail r
2069−-- return $ Prelude.head r a
2074−-- g <- proportionately 2 1
2075−-- let f = Fold.partitionByM g Fold.length Fold.length
2076−-- r <- Stream.fold f (Stream.enumerateFromTo (1 :: Int) 100)
2084−-- This is the consumer side dual of the producer side 'mergeBy' operation.
2086−-- When one fold is done, any input meant for it is ignored until the other
2087−-- fold is also done.
2089−-- Stops when both the folds stop.
2091−-- /See also: 'partitionByFstM' and 'partitionByMinM'./
2094−{-# INLINE partitionByM #-}
2095−partitionByM :: Monad m
2096− => (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
2097−partitionByM = partitionByMUsing teeWith
2099−-- | Similar to 'partitionByM' but terminates when the first fold terminates.
2101−{-# INLINE partitionByFstM #-}
2102−partitionByFstM :: Monad m
2103− => (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
2104−partitionByFstM = partitionByMUsing teeWithFst
2106−-- | Similar to 'partitionByM' but terminates when any fold terminates.
2108−{-# INLINE partitionByMinM #-}
2109−partitionByMinM :: Monad m =>
2110− (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
2111−partitionByMinM = partitionByMUsing teeWithMin
2113−-- Note: we could use (a -> Bool) instead of (a -> Either b c), but the latter
2114−-- makes the signature clearer as to which case belongs to which fold.
2115−-- XXX need to check the performance in both cases.
2117−-- | Same as 'partitionByM' but with a pure partition function.
2119−-- Example, count even and odd numbers in a stream:
2122−-- let f = Fold.partitionBy (\n -> if even n then Left n else Right n)
2123−-- (fmap (("Even " ++) . show) Fold.length)
2124−-- (fmap (("Odd " ++) . show) Fold.length)
2125−-- in Stream.fold f (Stream.enumerateFromTo 1 100)
2127−-- ("Even 50","Odd 50")
2130−{-# INLINE partitionBy #-}
2131−partitionBy :: Monad m
2132− => (a -> Either b c) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
2133−partitionBy f = partitionByM (return . f)
2135−-- | Compose two folds such that the combined fold accepts a stream of 'Either'
2136−-- and routes the 'Left' values to the first fold and 'Right' values to the
2141−-- >>> partition = Fold.partitionBy id
2143−{-# INLINE partition #-}
2144−partition :: Monad m
2145− => Fold m b x -> Fold m c y -> Fold m (Either b c) (x, y)
2146−partition = partitionBy id
2149−-- | Send one item to each fold in a round-robin fashion. This is the consumer
2150−-- side dual of producer side 'mergeN' operation.
2152−-- partitionN :: Monad m => [Fold m a b] -> Fold m a [b]
2153−-- partitionN fs = Fold step begin done
2156−------------------------------------------------------------------------------
2158−------------------------------------------------------------------------------
2160−{-# INLINE unzipWithMUsing #-}
2161−unzipWithMUsing :: Monad m =>
2162− ( (x -> y -> (x, y))
2165− -> Fold m (b, c) (x, y)
2171−unzipWithMUsing t f fld1 fld2 =
2172− let f1 = lmap fst fld1 -- :: Fold m (b, c) b
2173− f2 = lmap snd fld2 -- :: Fold m (b, c) c
2174− in lmapM f (t (,) f1 f2)
2176−-- | Like 'unzipWith' but with a monadic splitter function.
2180−-- >>> unzipWithM k f1 f2 = Fold.lmapM k (Fold.unzip f1 f2)
2183−{-# INLINE unzipWithM #-}
2184−unzipWithM :: Monad m
2185− => (a -> m (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
2186−unzipWithM = unzipWithMUsing teeWith
2188−-- | Similar to 'unzipWithM' but terminates when the first fold terminates.
2190−{-# INLINE unzipWithFstM #-}
2191−unzipWithFstM :: Monad m =>
2192− (a -> m (b, c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
2193−unzipWithFstM = unzipWithMUsing teeWithFst
2195−-- | Similar to 'unzipWithM' but terminates when any fold terminates.
2197−{-# INLINE unzipWithMinM #-}
2198−unzipWithMinM :: Monad m =>
2199− (a -> m (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
2200−unzipWithMinM = unzipWithMUsing teeWithMin
2202−-- | Split elements in the input stream into two parts using a pure splitter
2203−-- function, direct each part to a different fold and zip the results.
2207−-- >>> unzipWith f = Fold.unzipWithM (return . f)
2208−-- >>> unzipWith f fld1 fld2 = Fold.lmap f (Fold.unzip fld1 fld2)
2210−-- This fold terminates when both the input folds terminate.
2213−{-# INLINE unzipWith #-}
2214−unzipWith :: Monad m
2215− => (a -> (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
2216−unzipWith f = unzipWithM (return . f)
2218−-- | Send the elements of tuples in a stream of tuples through two different
2223−-- |-------Fold m a x--------|
2224−-- ---------stream of (a,b)--| |----m (x,y)
2225−-- |-------Fold m b y--------|
2231−-- >>> unzip = Fold.unzipWith id
2233−-- This is the consumer side dual of the producer side 'zip' operation.
2235−{-# INLINE unzip #-}
2236−unzip :: Monad m => Fold m a x -> Fold m b y -> Fold m (a,b) (x,y)
2237−unzip = unzipWith id
2239−------------------------------------------------------------------------------
2240−-- Combining streams and folds - Zipping
2241−------------------------------------------------------------------------------
2243−-- XXX These can be implemented using the fold scan, using the stream as a
2245−-- XXX Stream Skip state cannot be efficiently handled in folds but can be
2246−-- handled in parsers using the Continue facility. See zipWithM in the Parser
2249−-- cmpBy, eqBy, isPrefixOf, isSubsequenceOf etc can be implemented using
2252−-- | Zip a stream with the input of a fold using the supplied function.
2256−{-# INLINE zipStreamWithM #-}
2257−zipStreamWithM :: -- Monad m =>
2258− (a -> b -> m c) -> Stream m a -> Fold m c x -> Fold m b x
2259−zipStreamWithM = undefined
2261−-- | Zip a stream with the input of a fold.
2263−-- >>> zip = Fold.zipStreamWithM (curry return)
2267−{-# INLINE zipStream #-}
2268−zipStream :: Monad m => Stream m a -> Fold m (a, b) x -> Fold m b x
2269−zipStream = zipStreamWithM (curry return)
2271−-- | Pair each element of a fold input with its index, starting from index 0.
2273−{-# INLINE indexingWith #-}
2274−indexingWith :: Monad m => Int -> (Int -> Int) -> Fold m a (Maybe (Int, a))
2275−indexingWith i f = fmap toMaybe $ foldl' step initial
2281− step Nothing' a = Just' (i, a)
2282− step (Just' (n, _)) a = Just' (f n, a)
2285−-- >>> indexing = Fold.indexingWith 0 (+ 1)
2287−{-# INLINE indexing #-}
2288−indexing :: Monad m => Fold m a (Maybe (Int, a))
2289−indexing = indexingWith 0 (+ 1)
2292−-- >>> indexingRev n = Fold.indexingWith n (subtract 1)
2294−{-# INLINE indexingRev #-}
2295−indexingRev :: Monad m => Int -> Fold m a (Maybe (Int, a))
2296−indexingRev n = indexingWith n (subtract 1)
2298−-- | Pair each element of a fold input with its index, starting from index 0.
2300−-- >>> indexed = Fold.scanMaybe Fold.indexing
2302−{-# INLINE indexed #-}
2303−indexed :: Monad m => Fold m (Int, a) b -> Fold m a b
2304−indexed = scanMaybe indexing
2306−-- | Change the predicate function of a Fold from @a -> b@ to accept an
2307−-- additional state input @(s, a) -> b@. Convenient to filter with an
2308−-- addiitonal index or time input.
2310−-- >>> filterWithIndex = Fold.with Fold.indexed Fold.filter
2313−-- filterWithAbsTime = with timestamped filter
2314−-- filterWithRelTime = with timeIndexed filter
2320− (Fold m (s, a) b -> Fold m a b)
2321− -> (((s, a) -> c) -> Fold m (s, a) b -> Fold m (s, a) b)
2322− -> (((s, a) -> c) -> Fold m a b -> Fold m a b)
2323−with f comb g = f . comb g . lmap snd
2325−-- XXX Implement as a filter
2326−-- sampleFromthen :: Monad m => Int -> Int -> Fold m a (Maybe a)
2328−-- | @sampleFromthen offset stride@ samples the element at @offset@ index and
2329−-- then every element at strides of @stride@.
2331−{-# INLINE sampleFromthen #-}
2332−sampleFromthen :: Monad m => Int -> Int -> Fold m a b -> Fold m a b
2333−sampleFromthen offset size =
2334− with indexed filter (\(i, _) -> (i + offset) `mod` size == 0)
2336−------------------------------------------------------------------------------
2338−------------------------------------------------------------------------------
2340−-- | @concatSequence f t@ applies folds from stream @t@ sequentially and
2341−-- collects the results using the fold @f@.
2345−{-# INLINE concatSequence #-}
2348− Fold m b c -> t (Fold m a b) -> Fold m a c
2349−concatSequence _f _p = undefined
2351−-- | Group the input stream into groups of elements between @low@ and @high@.
2352−-- Collection starts in chunks of @low@ and then keeps doubling until we reach
2353−-- @high@. Each chunk is folded using the provided fold function.
2355−-- This could be useful, for example, when we are folding a stream of unknown
2356−-- size to a stream of arrays and we want to minimize the number of
2359−-- NOTE: this would be an application of "many" using a terminating fold.
2363−{-# INLINE chunksBetween #-}
2364−chunksBetween :: -- Monad m =>
2365− Int -> Int -> Fold m a b -> Fold m b c -> Fold m a c
2366−chunksBetween _low _high _f1 _f2 = undefined
2368−-- | A fold that buffers its input to a pure stream.
2370−-- /Warning!/ working on large streams accumulated as buffers in memory could
2371−-- be very inefficient, consider using "Streamly.Data.Array" instead.
2373−-- >>> toStream = fmap Stream.fromList Fold.toList
2376−{-# INLINE toStream #-}
2377−toStream :: (Monad m, Monad n) => Fold m a (Stream n a)
2378−toStream = fmap StreamD.fromList toList
2380−-- This is more efficient than 'toStream'. toStream is exactly the same as
2381−-- reversing the stream after toStreamRev.
2383−-- | Buffers the input stream to a pure stream in the reverse order of the
2386−-- >>> toStreamRev = fmap Stream.fromList Fold.toListRev
2388−-- /Warning!/ working on large streams accumulated as buffers in memory could
2389−-- be very inefficient, consider using "Streamly.Data.Array" instead.
2393−-- xn : ... : x2 : x1 : []
2394−{-# INLINE toStreamRev #-}
2395−toStreamRev :: (Monad m, Monad n) => Fold m a (Stream n a)
2396−toStreamRev = fmap StreamD.fromList toListRev
2398−-- XXX This does not fuse. It contains a recursive step function. We will need
2399−-- a Skip input constructor in the fold type to make it fuse.
2401−-- | Unfold and flatten the input stream of a fold.
2404−-- Stream.fold (unfoldMany u f) = Stream.fold f . Stream.unfoldMany u
2408−{-# INLINE unfoldMany #-}
2409−unfoldMany :: Monad m => Unfold m a b -> Fold m b c -> Fold m a c
2410−unfoldMany (Unfold ustep inject) (Fold fstep initial extract) =
2411− Fold consume initial extract
2415− {-# INLINE produce #-}
2419− StreamD.Yield b us1 -> do
2422− Partial fs1 -> produce fs1 us1
2423− -- XXX What to do with the remaining stream?
2424− Done c -> return $ Done c
2425− StreamD.Skip us1 -> produce fs us1
2426− StreamD.Stop -> return $ Partial fs
2428− {-# INLINE_LATE consume #-}
2429− consume s a = inject a >>= produce s
2431−-- | Get the bottom most @n@ elements using the supplied comparison function.
2433−{-# INLINE bottomBy #-}
2434−bottomBy :: (MonadIO m, Unbox a) =>
2435− (a -> a -> Ordering)
2437− -> Fold m a (MutArray a)
2438−bottomBy cmp n = Fold step initial extract
2443− arr <- MA.pinnedNew n
2445− then return $ Done arr
2446− else return $ Partial (arr, 0)
2451− arr' <- MA.snoc arr x
2453− return $ Partial (arr', i + 1)
2455− x1 <- MA.getIndexUnsafe (i - 1) arr
2458− MA.putIndexUnsafe (i - 1) arr x
2460− return $ Partial (arr, i)
2461− _ -> return $ Partial (arr, i)
2463− extract = return . fst
2465−-- | Get the top @n@ elements using the supplied comparison function.
2467−-- To get bottom n elements instead:
2469−-- >>> bottomBy cmp = Fold.topBy (flip cmp)
2473−-- >>> stream = Stream.fromList [2::Int,7,9,3,1,5,6,11,17]
2474−-- >>> Stream.fold (Fold.topBy compare 3) stream >>= MutArray.toList
2479−{-# INLINE topBy #-}
2480−topBy :: (MonadIO m, Unbox a) =>
2481− (a -> a -> Ordering)
2483− -> Fold m a (MutArray a)
2484−topBy cmp = bottomBy (flip cmp)
2486−-- | Fold the input stream to top n elements.
2490−-- >>> top = Fold.topBy compare
2492−-- >>> stream = Stream.fromList [2::Int,7,9,3,1,5,6,11,17]
2493−-- >>> Stream.fold (Fold.top 3) stream >>= MutArray.toList
2498−top :: (MonadIO m, Unbox a, Ord a) => Int -> Fold m a (MutArray a)
2499−top = bottomBy $ flip compare
2501−-- | Fold the input stream to bottom n elements.
2505−-- >>> bottom = Fold.bottomBy compare
2507−-- >>> stream = Stream.fromList [2::Int,7,9,3,1,5,6,11,17]
2508−-- >>> Stream.fold (Fold.bottom 3) stream >>= MutArray.toList
2512−{-# INLINE bottom #-}
2513−bottom :: (MonadIO m, Unbox a, Ord a) => Int -> Fold m a (MutArray a)
2514−bottom = bottomBy compare
2516−------------------------------------------------------------------------------
2517−-- Interspersed parsing
2518−------------------------------------------------------------------------------
2520−data IntersperseQState fs ps =
2521− IntersperseQUnquoted !fs !ps
2522− | IntersperseQQuoted !fs !ps
2523− | IntersperseQQuotedEsc !fs !ps
2525−-- Useful for parsing CSV with quoting and escaping
2526−{-# INLINE intersperseWithQuotes #-}
2527−intersperseWithQuotes :: (Monad m, Eq a) =>
2528− a -> a -> a -> Fold m a b -> Fold m b c -> Fold m a c
2529−intersperseWithQuotes
2533− (Fold stepL initialL extractL)
2534− (Fold stepR initialR extractR) = Fold step initial extract
2539− error $ "intersperseWithQuotes: " ++ p ++ " parsing fold cannot "
2540− ++ status ++ " without input"
2542− {-# INLINE initL #-}
2547− return $ Partial $ mkState sL
2549− errMsg "content" "succeed"
2554− Partial sR -> initL (IntersperseQUnquoted sR)
2555− Done b -> return $ Done b
2557− {-# INLINE collect #-}
2558− collect nextS sR b = do
2563− Done c -> return (Done c)
2565− {-# INLINE process #-}
2566− process a sL sR nextState = do
2569− Partial s -> return $ Partial (nextState sR s)
2570− Done b -> collect nextState sR b
2572− {-# INLINE processQuoted #-}
2573− processQuoted a sL sR nextState = do
2576− Partial s -> return $ Partial (nextState sR s)
2577− Done _ -> error "Collecting fold finished inside quote"
2579− step (IntersperseQUnquoted sR sL) a
2580− | a == separator = do
2582− collect IntersperseQUnquoted sR b
2583− | a == quote = processQuoted a sL sR IntersperseQQuoted
2584− | otherwise = process a sL sR IntersperseQUnquoted
2586− step (IntersperseQQuoted sR sL) a
2587− | a == esc = processQuoted a sL sR IntersperseQQuotedEsc
2588− | a == quote = process a sL sR IntersperseQUnquoted
2589− | otherwise = processQuoted a sL sR IntersperseQQuoted
2591− step (IntersperseQQuotedEsc sR sL) a =
2592− processQuoted a sL sR IntersperseQQuoted
2594− extract (IntersperseQUnquoted sR _) = extractR sR
2595− extract (IntersperseQQuoted _ _) =
2596− error "intersperseWithQuotes: finished inside quote"
2597− extract (IntersperseQQuotedEsc _ _) =
2598− error "intersperseWithQuotes: finished inside quote, at escape char"