2-- Module : Streamly.Internal.Data.Stream.IsStream.Nesting
3-- Copyright : (c) 2017 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
9-- Transformations involving multiple streams, streams and unfolds or streams
12module Streamly.Internal.Data.Stream.IsStream.Nesting
15 -- | Combining streams to generate streams.
17 -- ** Combine Two Streams
18 -- | Functions ending in the shape:
20 -- @t m a -> t m a -> t m a@.
65 -- ** Combine N Streams
66 -- | Functions generally ending in these shapes:
69 -- concat: f (t m a) -> t m a
70 -- concatMap: (a -> t m b) -> t m a -> t m b
71 -- concatUnfold: Unfold m a b -> t m a -> t m b
74 -- *** Flatten Containers of Streams
80 -- | Map and flatten streams.
81 , concatMapFoldableWith
82 , concatForFoldableWith
91 -- | Unfold and flatten streams.
93 , concatUnfoldInterleave
94 , concatUnfoldRoundrobin
97 -- | Insert effects between streams. Like concatUnfold but intersperses an
98 -- effect between the streams. A
99 -- special case of gintercalate.
105 -- | Insert Streams between Streams.
106 -- Like concatUnfold but intersperses streams from another source between
107 -- the streams from the first source.
114 -- | Map and flatten Trees of Streams
117 , iterateMapLeftsWith
120 -- | Folding and Parsing chunks of streams to eliminate nested streams.
121 -- Functions generally ending in these shapes:
124 -- f (Fold m a b) -> t m a -> t m b
125 -- f (Parser m a b) -> t m a -> t m b
129 -- | Apply folds on a stream.
136 -- | Apply parsers on a stream.
144 -- | Element unaware grouping.
151 -- In imperative terms, grouped folding can be considered as a nested loop
152 -- where we loop over the stream to group elements and then loop over
153 -- individual groups to fold them to a single value that is yielded in the
160 -- -- *** Searching Sequences
161 -- , seqIndices -- search a sequence in the stream
163 -- -- *** Searching Multiple Sequences
164 -- , seqIndicesAny -- search any of the given sequence in the stream
166 -- -- -- ** Searching Streams
167 -- -- | Finding a stream within another stream.
170 -- | Streams can be sliced into segments in space or in time. We use the
171 -- term @chunk@ to refer to a spatial length of the stream (spatial window)
172 -- and the term @session@ to refer to a length in time (time window).
174 -- -- *** Using Element Separators
182 , wordsBy -- stripAndCompactBy
184 -- -- *** Splitting By Sequences
188 -- , splitOnPrefixSeq
190 -- Keeping the delimiters
192 -- , splitByPrefixSeq
195 -- Splitting using multiple sequence separators
197 -- , splitOnAnySuffixSeq
198 -- , splitOnAnyPrefixSeq
200 -- -- *** Splitting By Streams
201 -- -- | Splitting a stream using another stream as separator.
203 -- ** Keyed Window Classification
205 -- | Split the stream into chunks or windows by position or time. Each
206 -- window can be associated with a key, all events associated with a
207 -- particular key in the window can be folded to a single result. The
208 -- window termination can be dynamically controlled by the fold.
210 -- The term "chunk" is used for a window defined by position of elements
211 -- and the term "session" is used for a time window.
213 -- *** Tumbling Windows
214 -- | A new window starts after the previous window is finished.
216 -- , classifyChunksOf
220 -- *** Keep Alive Windows
221 -- | The window size is extended if an event arrives within the specified
222 -- window size. This can represent sessions with idle or inactive timeout.
224 -- , classifyKeepAliveChunks
225 , classifyKeepAliveSessions
228 -- *** Sliding Windows
229 -- | A new window starts after the specified slide from the previous
230 -- window. Therefore windows can overlap.
231 , classifySlidingChunks
232 , classifySlidingSessions
233 -- *** Sliding Window Buffers
234 -- , slidingChunkBuffer
235 -- , slidingSessionBuffer
238 -- * Transform (Nested Containers)
247import Control.Concurrent (threadDelay)
248import Control.Exception (assert)
249import Control.Monad.Catch (MonadThrow)
250import Control.Monad.IO.Class (MonadIO(..))
251import Data.Heap (Entry(..))
252import Data.Kind (Type)
253import Data.Maybe (isNothing)
254import Foreign.Storable (Storable)
255import Streamly.Internal.Data.Fold.Type (Fold (..), Fold2 (..))
256import Streamly.Internal.Data.Parser (Parser (..))
257import Streamly.Internal.Data.Unfold.Type (Unfold)
258import Streamly.Internal.Data.Array.Foreign.Type (Array)
259import Streamly.Internal.Data.SVar (MonadAsync)
260import Streamly.Internal.Data.Stream.Ahead (ahead)
261import Streamly.Internal.Data.Stream.Async (async, wAsync)
262import Streamly.Internal.Data.Stream.IsStream.Common
274import Streamly.Internal.Data.Stream.Parallel (parallel)
275import Streamly.Internal.Data.Stream.Prelude
276 ( fromStreamS, toStreamS, concatFoldableWith, concatMapFoldableWith
277 , concatForFoldableWith)
278import Streamly.Internal.Data.Stream.StreamD (fromStreamD, toStreamD)
279import Streamly.Internal.Data.Stream.StreamK (IsStream)
280import Streamly.Internal.Data.Stream.Serial (serial, wSerial)
281import Streamly.Internal.Data.Time.Units
282 ( AbsTime, MilliSecond64(..), addToAbsTime, toRelTime
284import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))
286import qualified Data.Heap as H
287import qualified Data.Map.Strict as Map
288import qualified Streamly.Internal.Data.Array.Foreign.Type as A
289import qualified Streamly.Internal.Data.Fold as FL
290import qualified Streamly.Internal.Data.Parser.ParserK.Type as PRK
291import qualified Streamly.Internal.Data.Parser.ParserD as PRD
292import qualified Streamly.Internal.Data.Stream.Parallel as Par
293import qualified Streamly.Internal.Data.Stream.Serial as Serial
294import qualified Streamly.Internal.Data.Stream.StreamD as D
295import qualified Streamly.Internal.Data.Stream.StreamK as K
296import qualified Streamly.Internal.Data.Stream.Zip as Z
297#ifdef USE_STREAMK_ONLY
298import qualified Streamly.Internal.Data.Stream.StreamK as S
300import qualified Streamly.Internal.Data.Stream.StreamD as S
303import Prelude hiding (zipWith, concatMap, concat)
307-- >>> import Prelude hiding (zipWith, concatMap, concat)
308-- >>> import qualified Streamly.Prelude as Stream
309-- >>> import Streamly.Internal.Data.Stream.IsStream as Stream
310-- >>> import qualified Streamly.Data.Fold as Fold
311-- >>> import qualified Streamly.Internal.Data.Fold as Fold
312-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
313-- >>> import qualified Streamly.Internal.Data.Parser as Parser
314-- >>> import qualified Streamly.Data.Array.Foreign as Array
316-------------------------------------------------------------------------------
318-------------------------------------------------------------------------------
320-- | Append the outputs of two streams, yielding all the elements from the
321-- first stream and then yielding all the elements from the second stream.
323-- IMPORTANT NOTE: This could be 100x faster than @serial/<>@ for appending a
324-- few (say 100) streams because it can fuse via stream fusion. However, it
325-- does not scale for a large number of streams (say 1000s) and becomes
326-- qudartically slow. Therefore use this for custom appending of a few streams
327-- but use 'concatMap' or 'concatMapWith serial' for appending @n@ streams or
328-- infinite containers of streams.
332append ::(IsStream t, Monad m) => t m b -> t m b -> t m b
333append m1 m2 = fromStreamD $ D.append (toStreamD m1) (toStreamD m2)
335-------------------------------------------------------------------------------
337-------------------------------------------------------------------------------
339-- XXX Same as 'wSerial'. We should perhaps rename wSerial to interleave.
340-- XXX Document the interleaving behavior of side effects in all the
341-- interleaving combinators.
342-- XXX Write time-domain equivalents of these. In the time domain we can
343-- interleave two streams such that the value of second stream is always taken
344-- from its last value even if no new value is being yielded, like
345-- zipWithLatest. It would be something like interleaveWithLatest.
347-- | Interleaves the outputs of two streams, yielding elements from each stream
348-- alternately, starting from the first stream. If any of the streams finishes
349-- early the other stream continues alone until it too finishes.
351-- >>> :set -XOverloadedStrings
352-- >>> import Data.Functor.Identity (Identity)
353-- >>> Stream.interleave "ab" ",,,," :: Stream.SerialT Identity Char
355-- >>> Stream.interleave "abcd" ",," :: Stream.SerialT Identity Char
358-- 'interleave' is dual to 'interleaveMin', it can be called @interleaveMax@.
360-- Do not use at scale in concatMapWith.
363{-# INLINE interleave #-}
364interleave ::(IsStream t, Monad m) => t m b -> t m b -> t m b
365interleave m1 m2 = fromStreamD $ D.interleave (toStreamD m1) (toStreamD m2)
367-- | Interleaves the outputs of two streams, yielding elements from each stream
368-- alternately, starting from the first stream. As soon as the first stream
369-- finishes, the output stops, discarding the remaining part of the second
370-- stream. In this case, the last element in the resulting stream would be from
371-- the second stream. If the second stream finishes early then the first stream
372-- still continues to yield elements until it finishes.
374-- >>> :set -XOverloadedStrings
375-- >>> import Data.Functor.Identity (Identity)
376-- >>> Stream.interleaveSuffix "abc" ",,,," :: Stream.SerialT Identity Char
378-- >>> Stream.interleaveSuffix "abc" "," :: Stream.SerialT Identity Char
381-- 'interleaveSuffix' is a dual of 'interleaveInfix'.
383-- Do not use at scale in concatMapWith.
386{-# INLINE interleaveSuffix #-}
387interleaveSuffix ::(IsStream t, Monad m) => t m b -> t m b -> t m b
388interleaveSuffix m1 m2 =
389 fromStreamD $ D.interleaveSuffix (toStreamD m1) (toStreamD m2)
391-- | Interleaves the outputs of two streams, yielding elements from each stream
392-- alternately, starting from the first stream and ending at the first stream.
393-- If the second stream is longer than the first, elements from the second
394-- stream are infixed with elements from the first stream. If the first stream
395-- is longer then it continues yielding elements even after the second stream
398-- >>> :set -XOverloadedStrings
399-- >>> import Data.Functor.Identity (Identity)
400-- >>> Stream.interleaveInfix "abc" ",,,," :: Stream.SerialT Identity Char
402-- >>> Stream.interleaveInfix "abc" "," :: Stream.SerialT Identity Char
405-- 'interleaveInfix' is a dual of 'interleaveSuffix'.
407-- Do not use at scale in concatMapWith.
410{-# INLINE interleaveInfix #-}
411interleaveInfix ::(IsStream t, Monad m) => t m b -> t m b -> t m b
412interleaveInfix m1 m2 =
413 fromStreamD $ D.interleaveInfix (toStreamD m1) (toStreamD m2)
415-- | Interleaves the outputs of two streams, yielding elements from each stream
416-- alternately, starting from the first stream. The output stops as soon as any
417-- of the two streams finishes, discarding the remaining part of the other
418-- stream. The last element of the resulting stream would be from the longer
421-- >>> :set -XOverloadedStrings
422-- >>> import Data.Functor.Identity (Identity)
423-- >>> Stream.interleaveMin "ab" ",,,," :: Stream.SerialT Identity Char
425-- >>> Stream.interleaveMin "abcd" ",," :: Stream.SerialT Identity Char
428-- 'interleaveMin' is dual to 'interleave'.
430-- Do not use at scale in concatMapWith.
433{-# INLINE interleaveMin #-}
434interleaveMin ::(IsStream t, Monad m) => t m b -> t m b -> t m b
435interleaveMin m1 m2 = fromStreamD $ D.interleaveMin (toStreamD m1) (toStreamD m2)
437-------------------------------------------------------------------------------
439-------------------------------------------------------------------------------
441-- | Schedule the execution of two streams in a fair round-robin manner,
442-- executing each stream once, alternately. Execution of a stream may not
443-- necessarily result in an output, a stream may chose to @Skip@ producing an
444-- element until later giving the other stream a chance to run. Therefore, this
445-- combinator fairly interleaves the execution of two streams rather than
446-- fairly interleaving the output of the two streams. This can be useful in
447-- co-operative multitasking without using explicit threads. This can be used
448-- as an alternative to `async`.
450-- Do not use at scale in concatMapWith.
453{-# INLINE roundrobin #-}
454roundrobin ::(IsStream t, Monad m) => t m b -> t m b -> t m b
455roundrobin m1 m2 = fromStreamD $ D.roundRobin (toStreamD m1) (toStreamD m2)
457------------------------------------------------------------------------------
458-- Merging (sorted streams)
459------------------------------------------------------------------------------
461-- | Merge two streams using a comparison function. The head elements of both
462-- the streams are compared and the smaller of the two elements is emitted, if
463-- both elements are equal then the element from the first stream is used
466-- If the streams are sorted in ascending order, the resulting stream would
467-- also remain sorted in ascending order.
470-- >>> Stream.toList $ Stream.mergeBy compare (Stream.fromList [1,3,5]) (Stream.fromList [2,4,6,8])
476{-# INLINABLE mergeBy #-}
478 (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> t m a -> t m a
479mergeBy f m1 m2 = fromStreamS $ S.mergeBy f (toStreamS m1) (toStreamS m2)
481-- | Like 'mergeBy' but with a monadic comparison function.
483-- Merge two streams randomly:
486-- > randomly _ _ = randomIO >>= \x -> return $ if x then LT else GT
487-- > S.toList $ S.mergeByM randomly (S.fromList [1,1,1,1]) (S.fromList [2,2,2,2])
491-- Merge two streams in a proportion of 2:1:
494-- proportionately m n = do
495-- ref <- newIORef $ cycle $ concat [replicate m LT, replicate n GT]
496-- return $ \\_ _ -> do
498-- writeIORef ref $ tail r
502-- f <- proportionately 2 1
503-- xs <- S.toList $ S.mergeByM f (S.fromList [1,1,1,1,1,1]) (S.fromList [2,2,2])
507-- [1,1,2,1,1,2,1,1,2]
511{-# INLINABLE mergeByM #-}
513 :: (IsStream t, Monad m)
514 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
515mergeByM f m1 m2 = fromStreamS $ S.mergeByM f (toStreamS m1) (toStreamS m2)
518-- | Like 'mergeByM' but stops merging as soon as any of the two streams stops.
519{-# INLINABLE mergeEndByAny #-}
521 :: (IsStream t, Monad m)
522 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
523mergeEndByAny f m1 m2 = fromStreamD $
524 D.mergeEndByAny f (toStreamD m1) (toStreamD m2)
526-- Like 'mergeByM' but stops merging as soon as the first stream stops.
527{-# INLINABLE mergeEndByFirst #-}
529 :: (IsStream t, Monad m)
530 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
531mergeEndByFirst f m1 m2 = fromStreamS $
532 D.mergeEndByFirst f (toStreamD m1) (toStreamD m2)
534-- Holding this back for now, we may want to use the name "merge" differently
535-- | Same as @'mergeBy' 'compare'@.
538-- > S.toList $ S.merge (S.fromList [1,3,5]) (S.fromList [2,4,6,8])
543{-# INLINABLE merge #-}
545 (IsStream t, Monad m, Ord a) => t m a -> t m a -> t m a
546merge = mergeBy compare
549-- | Like 'mergeBy' but merges concurrently (i.e. both the elements being
550-- merged are generated concurrently).
553{-# INLINE mergeAsyncBy #-}
554mergeAsyncBy :: (IsStream t, MonadAsync m)
555 => (a -> a -> Ordering) -> t m a -> t m a -> t m a
556mergeAsyncBy f = mergeAsyncByM (\a b -> return $ f a b)
558-- | Like 'mergeByM' but merges concurrently (i.e. both the elements being
559-- merged are generated concurrently).
562{-# INLINE mergeAsyncByM #-}
563mergeAsyncByM :: (IsStream t, MonadAsync m)
564 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
565mergeAsyncByM f m1 m2 =
567 let par = Par.mkParallelD . toStreamD
568 in D.mergeByM f (par m1) (par m2)
570------------------------------------------------------------------------------
571-- Combine N Streams - concatMap
572------------------------------------------------------------------------------
574-- | @concatMapWith merge map stream@ is a two dimensional looping combinator.
575-- The first argument specifies a merge or concat function that is used to
576-- merge the streams generated by applying the second argument i.e. the @map@
577-- function to each element of the input stream. The concat function could be
578-- 'serial', 'parallel', 'async', 'ahead' or any other zip or merge function
579-- and the second argument could be any stream generation function using a
582-- /Compare 'foldMapWith'/
586-- /Since: 0.8.0 (signature change)/
587{-# INLINE concatMapWith #-}
590 => (t m b -> t m b -> t m b)
594concatMapWith = K.concatMapBy
596-- | Like 'concatMapWith' but carries a state which can be used to share
597-- information across multiple steps of concat.
600-- concatSmapMWith combine f initial = concatMapWith combine id . smapM f initial
605{-# INLINE concatSmapMWith #-}
607 :: (IsStream t, Monad m)
608 => (t m b -> t m b -> t m b)
609 -> (s -> a -> m (s, t m b))
613concatSmapMWith combine f initial = concatMapWith combine id . smapM f initial
615-- | Map a stream producing function on each element of the stream and then
616-- flatten the results into a single stream.
619-- concatMap f = 'concat . map f'
620-- concatMap = 'concatMapWith' 'Serial.serial'
621-- concatMap f = 'concatMapM' (return . f)
622-- concatMap f = 'concatUnfold' (UF.lmap f UF.fromStream)
626{-# INLINE concatMap #-}
627concatMap ::(IsStream t, Monad m) => (a -> t m b) -> t m a -> t m b
628concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)
630-- | Flatten a stream of streams to a single stream.
633-- concat = concatMap id
638concat :: (IsStream t, Monad m) => t m (t m a) -> t m a
641-- Keep concating either streams as long as rights are generated, stop as soon
642-- as a left is generated and concat the left stream.
650 :: -- (IsStream t, MonadAsync m) =>
651 (forall x. t m x -> t m x -> t m x)
652 -> (a -> t m (Either (t m b) b))
655concatMapEitherWith = undefined
658-- XXX Implement a StreamD version for fusion.
660-- | Combine streams in pairs using a binary stream combinator, then combine
661-- the resulting streams in pairs recursively until we get to a single combined
664-- For example, you can sort a stream using merge sort like this:
666-- >>> Stream.toList $ Stream.concatPairsWith (Stream.mergeBy compare) Stream.yield $ Stream.fromList [5,1,7,9,2]
669-- /Caution: the stream of streams must be finite/
673{-# INLINE concatPairsWith #-}
674concatPairsWith :: IsStream t =>
675 (t m b -> t m b -> t m b)
679concatPairsWith = K.concatPairsWith
681------------------------------------------------------------------------------
682-- Combine N Streams - concatUnfold
683------------------------------------------------------------------------------
685-- | Like 'concatMap' but uses an 'Unfold' for stream generation. Unlike
686-- 'concatMap' this can fuse the 'Unfold' code with the inner loop and
687-- therefore provide many times better performance.
690{-# INLINE concatUnfold #-}
691concatUnfold ::(IsStream t, Monad m) => Unfold m a b -> t m a -> t m b
692concatUnfold u m = fromStreamD $ D.concatUnfold u (toStreamD m)
694-- | Like 'concatUnfold' but interleaves the streams in the same way as
695-- 'interleave' behaves instead of appending them.
698{-# INLINE concatUnfoldInterleave #-}
699concatUnfoldInterleave ::(IsStream t, Monad m)
700 => Unfold m a b -> t m a -> t m b
701concatUnfoldInterleave u m =
702 fromStreamD $ D.concatUnfoldInterleave u (toStreamD m)
704-- | Like 'concatUnfold' but executes the streams in the same way as
708{-# INLINE concatUnfoldRoundrobin #-}
709concatUnfoldRoundrobin ::(IsStream t, Monad m)
710 => Unfold m a b -> t m a -> t m b
711concatUnfoldRoundrobin u m =
712 fromStreamD $ D.concatUnfoldRoundrobin u (toStreamD m)
714------------------------------------------------------------------------------
715-- Combine N Streams - interpose
716------------------------------------------------------------------------------
718-- > interpose x unf str = gintercalate unf str UF.identity (repeat x)
720-- | Unfold the elements of a stream, intersperse the given element between the
721-- unfolded streams and then concat them into a single stream.
723-- > unwords = S.interpose ' '
726{-# INLINE interpose #-}
727interpose :: (IsStream t, Monad m)
728 => c -> Unfold m b c -> t m b -> t m c
730 D.fromStreamD $ D.interpose (return x) unf (D.toStreamD str)
732-- interposeSuffix x unf str = gintercalateSuffix unf str UF.identity (repeat x)
734-- | Unfold the elements of a stream, append the given element after each
735-- unfolded stream and then concat them into a single stream.
737-- > unlines = S.interposeSuffix '\n'
740{-# INLINE interposeSuffix #-}
741interposeSuffix :: (IsStream t, Monad m)
742 => c -> Unfold m b c -> t m b -> t m c
743interposeSuffix x unf str =
744 D.fromStreamD $ D.interposeSuffix (return x) unf (D.toStreamD str)
746------------------------------------------------------------------------------
747-- Combine N Streams - intercalate
748------------------------------------------------------------------------------
750-- XXX we can swap the order of arguments to gintercalate so that the
751-- definition of concatUnfold becomes simpler? The first stream should be
752-- infixed inside the second one. However, if we change the order in
753-- "interleave" as well similarly, then that will make it a bit unintuitive.
755-- > concatUnfold unf str =
756-- > gintercalate unf str (UF.nilM (\_ -> return ())) (repeat ())
758-- | 'interleaveInfix' followed by unfold and concat.
761{-# INLINE gintercalate #-}
763 :: (IsStream t, Monad m)
764 => Unfold m a c -> t m a -> Unfold m b c -> t m b -> t m c
765gintercalate unf1 str1 unf2 str2 =
766 D.fromStreamD $ D.gintercalate
767 unf1 (D.toStreamD str1)
768 unf2 (D.toStreamD str2)
770-- XXX The order of arguments in "intercalate" is consistent with the list
771-- intercalate but inconsistent with gintercalate and other stream interleaving
772-- combinators. We can change the order of the arguments in other combinators
773-- but then 'interleave' combinator may become a bit unintuitive because we
774-- will be starting with the second stream.
776-- > intercalate seed unf str = gintercalate unf str unf (repeatM seed)
777-- > intercalate a unf str = concatUnfold unf $ intersperse a str
779-- | 'intersperse' followed by unfold and concat.
781-- > unwords = intercalate " " Unfold.fromList
783-- >>> Stream.toList $ Stream.intercalate " " Unfold.fromList $ Stream.fromList ["abc", "def", "ghi"]
787{-# INLINE intercalate #-}
788intercalate :: (IsStream t, Monad m)
789 => b -> Unfold m b c -> t m b -> t m c
790intercalate seed unf str = D.fromStreamD $
791 D.concatUnfold unf $ D.intersperse seed (toStreamD str)
793-- | 'interleaveSuffix' followed by unfold and concat.
796{-# INLINE gintercalateSuffix #-}
798 :: (IsStream t, Monad m)
799 => Unfold m a c -> t m a -> Unfold m b c -> t m b -> t m c
800gintercalateSuffix unf1 str1 unf2 str2 =
801 D.fromStreamD $ D.gintercalateSuffix
802 unf1 (D.toStreamD str1)
803 unf2 (D.toStreamD str2)
805-- > intercalateSuffix seed unf str = gintercalateSuffix unf str unf (repeatM seed)
806-- > intercalateSuffix a unf str = concatUnfold unf $ intersperseSuffix a str
808-- | 'intersperseSuffix' followed by unfold and concat.
810-- > unlines = intercalateSuffix "\n" Unfold.fromList
812-- >>> Stream.toList $ Stream.intercalateSuffix "\n" Unfold.fromList $ Stream.fromList ["abc", "def", "ghi"]
816{-# INLINE intercalateSuffix #-}
817intercalateSuffix :: (IsStream t, Monad m)
818 => b -> Unfold m b c -> t m b -> t m c
819intercalateSuffix seed unf str = fromStreamD $ D.concatUnfold unf
820 $ D.intersperseSuffix (return seed) (D.toStreamD str)
822------------------------------------------------------------------------------
823-- IterateMap - Map and flatten Trees of Streams
824------------------------------------------------------------------------------
826-- | Like 'iterateM' but iterates after mapping a stream generator on the
829-- Yield an input element in the output stream, map a stream generator on it
830-- and then do the same on the resulting stream. This can be used for a depth
831-- first traversal of a tree like structure.
833-- Note that 'iterateM' is a special case of 'iterateMapWith':
836-- iterateM f = iterateMapWith serial (yieldM . f) . yieldM
839-- It can be used to traverse a tree structure. For example, to list a
843-- Stream.iterateMapWith Stream.serial
844-- (either Dir.toEither (const nil))
845-- (yield (Left "tmp"))
850{-# INLINE iterateMapWith #-}
853 => (t m a -> t m a -> t m a)
857iterateMapWith combine f = concatMapWith combine go
859 go x = yield x `combine` concatMapWith combine go (f x)
862{-# INLINE iterateUnfold #-}
863iterateUnfold :: (IsStream t, MonadAsync m)
864 => Unfold m a a -> t m a -> t m a
865iterateUnfold unf xs = undefined
868------------------------------------------------------------------------------
870------------------------------------------------------------------------------
872-- To traverse graphs we need a state to be carried around in the traversal.
873-- For example, we can use a hashmap to store the visited status of nodes.
875-- | Like 'iterateMap' but carries a state in the stream generation function.
876-- This can be used to traverse graph like structures, we can remember the
877-- visited nodes in the state to avoid cycles.
879-- Note that a combination of 'iterateMap' and 'usingState' can also be used to
880-- traverse graphs. However, this function provides a more localized state
881-- instead of using a global state.
887{-# INLINE iterateSmapMWith #-}
889 :: (IsStream t, Monad m)
890 => (t m a -> t m a -> t m a)
891 -> (b -> a -> m (b, t m a))
895iterateSmapMWith combine f initial stream =
896 concatMap (\b -> concatMapWith combine (go b) stream) (yieldM initial)
900 go b a = yield a `combine` feedback b a
904 (\(b1, s) -> concatMapWith combine (go b1) s)
907------------------------------------------------------------------------------
908-- iterateMap - Either streams
909------------------------------------------------------------------------------
911-- | In an 'Either' stream iterate on 'Left's. This is a special case of
915-- iterateMapLeftsWith combine f = iterateMapWith combine (either f (const nil))
918-- To traverse a directory tree:
921-- iterateMapLeftsWith serial Dir.toEither (yield (Left "tmp"))
926{-# INLINE iterateMapLeftsWith #-}
929 => (t m (Either a b) -> t m (Either a b) -> t m (Either a b))
930 -> (a -> t m (Either a b))
933iterateMapLeftsWith combine f = iterateMapWith combine (either f (const K.nil))
935------------------------------------------------------------------------------
937------------------------------------------------------------------------------
939-- Splitting operations that take a predicate and a Fold can be
940-- expressed using parseMany. Operations like chunksOf, intervalsOf, split*,
941-- can be expressed using parseMany when used with an appropriate Parser.
943-- XXX We need takeGE/takeBetween to implement "some" using "many".
945-- | Like 'foldMany' but appends empty fold output if the fold and stream
946-- termination aligns:
948-- >>> f = Fold.take 2 Fold.sum
949-- >>> Stream.toList $ Stream.foldManyPost f $ Stream.fromList []
951-- >>> Stream.toList $ Stream.foldManyPost f $ Stream.fromList [1..9]
953-- >>> Stream.toList $ Stream.foldManyPost f $ Stream.fromList [1..10]
958{-# INLINE foldManyPost #-}
960 :: (IsStream t, Monad m)
964foldManyPost f m = D.fromStreamD $ D.foldManyPost f (D.toStreamD m)
966-- | Apply a 'Fold' repeatedly on a stream and emit the fold outputs in the
969-- To sum every two contiguous elements in a stream:
971-- >>> f = Fold.take 2 Fold.sum
972-- >>> Stream.toList $ Stream.foldMany f $ Stream.fromList [1..10]
975-- On an empty stream the output is empty:
977-- >>> Stream.toList $ Stream.foldMany f $ Stream.fromList []
980-- Note @foldMany (take 0)@ would result in an infinite loop in a non-empty
985{-# INLINE foldMany #-}
987 :: (IsStream t, Monad m)
991foldMany f m = D.fromStreamD $ D.foldMany f (D.toStreamD m)
993-- | Apply a stream of folds to an input stream and emit the results in the
998{-# INLINE foldSequence #-}
1000 :: -- (IsStream t, Monad m) =>
1004foldSequence _f _m = undefined
1006-- | Iterate a fold generator on a stream. The initial value @b@ is used to
1007-- generate the first fold, the fold is applied on the stream and the result of
1008-- the fold is used to generate the next fold and so on.
1011-- > import Data.Monoid (Sum(..))
1012-- > f x = return (Fold.take 2 (Fold.sconcat x))
1013-- > s = Stream.map Sum $ Stream.fromList [1..10]
1014-- > Stream.toList $ Stream.map getSum $ Stream.foldIterateM f 0 s
1015-- [3,10,21,36,55,55]
1019-- This is the streaming equivalent of monad like sequenced application of
1020-- folds where next fold is dependent on the previous fold.
1024{-# INLINE foldIterateM #-}
1026 (IsStream t, Monad m) => (b -> m (Fold m a b)) -> b -> t m a -> t m b
1027foldIterateM f i m = D.fromStreamD $ D.foldIterateM f i (D.toStreamD m)
1029------------------------------------------------------------------------------
1031------------------------------------------------------------------------------
1033-- | Apply a 'Parser' repeatedly on a stream and emit the parsed values in the
1036-- This is the streaming equivalent of the 'Streamly.Internal.Data.Parser.many'
1039-- >>> Stream.toList $ Stream.parseMany (Parser.takeBetween 0 2 Fold.sum) $ Stream.fromList [1..10]
1043-- > Stream.toList $ Stream.parseMany (Parser.line Fold.toList) $ Stream.fromList "hello\\nworld"
1044-- ["hello\\n","world"]
1049-- foldMany f = parseMany (fromFold f)
1052-- Known Issues: When the parser fails there is no way to get the remaining
1057{-# INLINE parseMany #-}
1059 :: (IsStream t, MonadThrow m)
1064 D.fromStreamD $ D.parseMany (PRK.fromParserK p) (D.toStreamD m)
1066{-# INLINE parseManyD #-}
1068 :: (IsStream t, MonadThrow m)
1073 D.fromStreamD $ D.parseMany p (D.toStreamD m)
1075-- | Apply a stream of parsers to an input stream and emit the results in the
1080{-# INLINE parseSequence #-}
1082 :: -- (IsStream t, Monad m) =>
1086parseSequence _f _m = undefined
1088-- | @parseManyTill collect test stream@ tries the parser @test@ on the input,
1089-- if @test@ fails it backtracks and tries @collect@, after @collect@ succeeds
1090-- @test@ is tried again and so on. The parser stops when @test@ succeeds. The
1091-- output of @test@ is discarded and the output of @collect@ is emitted in the
1092-- output stream. The parser fails if @collect@ fails.
1096{-# INLINE parseManyTill #-}
1098 -- (IsStream t, MonadThrow m) =>
1103parseManyTill = undefined
1105-- | Iterate a parser generating function on a stream. The initial value @b@ is
1106-- used to generate the first parser, the parser is applied on the stream and
1107-- the result is used to generate the next parser and so on.
1109-- >>> import Data.Monoid (Sum(..))
1110-- >>> Stream.toList $ Stream.map getSum $ Stream.parseIterate (\b -> Parser.takeBetween 0 2 (Fold.sconcat b)) 0 $ Stream.map Sum $ Stream.fromList [1..10]
1111-- [3,10,21,36,55,55]
1113-- This is the streaming equivalent of monad like sequenced application of
1114-- parsers where next parser is dependent on the previous parser.
1118{-# INLINE parseIterate #-}
1120 :: (IsStream t, MonadThrow m)
1121 => (b -> Parser m a b)
1125parseIterate f i m = D.fromStreamD $
1126 D.parseIterate (PRK.fromParserK . f) i (D.toStreamD m)
1128------------------------------------------------------------------------------
1130------------------------------------------------------------------------------
1132-- | Drop prefix from the input stream if present.
1136-- /Unimplemented/ - Help wanted.
1137{-# INLINE dropPrefix #-}
1139 -- (Eq a, IsStream t, Monad m) =>
1140 t m a -> t m a -> t m a
1141dropPrefix = error "Not implemented yet!"
1143-- | Drop all matching infix from the input stream if present. Infix stream
1144-- may be consumed multiple times.
1146-- Space: @O(n)@ where n is the length of the infix.
1148-- /Unimplemented/ - Help wanted.
1149{-# INLINE dropInfix #-}
1151 -- (Eq a, IsStream t, Monad m) =>
1152 t m a -> t m a -> t m a
1153dropInfix = error "Not implemented yet!"
1155-- | Drop suffix from the input stream if present. Suffix stream may be
1156-- consumed multiple times.
1158-- Space: @O(n)@ where n is the length of the suffix.
1160-- /Unimplemented/ - Help wanted.
1161{-# INLINE dropSuffix #-}
1163 -- (Eq a, IsStream t, Monad m) =>
1164 t m a -> t m a -> t m a
1165dropSuffix = error "Not implemented yet!"
1167------------------------------------------------------------------------------
1168-- Generalized grouping
1169------------------------------------------------------------------------------
1171-- This combinator is the most general grouping combinator and can be used to
1172-- implement all other grouping combinators.
1174-- XXX check if this can implement the splitOn combinator i.e. we can slide in
1175-- new elements, slide out old elements and incrementally compute the hash.
1176-- Also, can we implement the windowed classification combinators using this?
1178-- In fact this is a parse. Instead of using a special return value in the fold
1179-- we are using a mapping function.
1181-- Note that 'scanl'' (usually followed by a map to extract the desired value
1182-- from the accumulator) can be used to realize many implementations e.g. a
1183-- sliding window implementation. A scan followed by a mapMaybe is also a good
1184-- pattern to express many problems where we want to emit a filtered output and
1185-- not emit an output on every input.
1187-- Passing on of the initial accumulator value to the next fold is equivalent
1188-- to returning the leftover concept.
1191-- | @groupScan splitter fold stream@ folds the input stream using @fold@.
1192-- @splitter@ is applied on the accumulator of the fold every time an item is
1193-- consumed by the fold. The fold continues until @splitter@ returns a 'Just'
1194-- value. A 'Just' result from the @splitter@ specifies a result to be emitted
1195-- in the output stream and the initial value of the accumulator for the next
1196-- group's fold. This allows us to control whether to start fresh for the next
1197-- fold or to continue from the previous fold's output.
1199{-# INLINE groupScan #-}
1201 :: (IsStream t, Monad m)
1202 => (x -> m (Maybe (b, x))) -> Fold m a x -> t m a -> t m b
1203groupScan split fold m = undefined
1206------------------------------------------------------------------------------
1208------------------------------------------------------------------------------
1210-- | @groupsBy cmp f $ S.fromList [a,b,c,...]@ assigns the element @a@ to the
1211-- first group, if @b \`cmp` a@ is 'True' then @b@ is also assigned to the same
1212-- group. If @c \`cmp` a@ is 'True' then @c@ is also assigned to the same
1213-- group and so on. When the comparison fails a new group is started. Each
1214-- group is folded using the fold @f@ and the result of the fold is emitted in
1215-- the output stream.
1217-- >>> Stream.toList $ Stream.groupsBy (>) Fold.toList $ Stream.fromList [1,3,7,0,2,5]
1221{-# INLINE groupsBy #-}
1223 :: (IsStream t, Monad m)
1228groupsBy cmp f m = D.fromStreamD $ D.groupsBy cmp f (D.toStreamD m)
1230-- | Unlike @groupsBy@ this function performs a rolling comparison of two
1231-- successive elements in the input stream. @groupsByRolling cmp f $ S.fromList
1232-- [a,b,c,...]@ assigns the element @a@ to the first group, if @a \`cmp` b@ is
1233-- 'True' then @b@ is also assigned to the same group. If @b \`cmp` c@ is
1234-- 'True' then @c@ is also assigned to the same group and so on. When the
1235-- comparison fails a new group is started. Each group is folded using the fold
1238-- >>> Stream.toList $ Stream.groupsByRolling (\a b -> a + 1 == b) Fold.toList $ Stream.fromList [1,2,3,7,8,9]
1242{-# INLINE groupsByRolling #-}
1244 :: (IsStream t, Monad m)
1249groupsByRolling cmp f m = D.fromStreamD $ D.groupsRollingBy cmp f (D.toStreamD m)
1252-- > groups = groupsBy (==)
1253-- > groups = groupsByRolling (==)
1255-- Groups contiguous spans of equal elements together in individual groups.
1257-- >>> Stream.toList $ Stream.groups Fold.toList $ Stream.fromList [1,1,2,2]
1261{-# INLINE groups #-}
1262groups :: (IsStream t, Monad m, Eq a) => Fold m a b -> t m a -> t m b
1263groups = groupsBy (==)
1265------------------------------------------------------------------------------
1266-- Splitting - by a predicate
1267------------------------------------------------------------------------------
1269-- In general we can use deintercalate for splitting. Then we can also use
1270-- uniqBy to condense the separators. One way to generalize splitting is to
1273-- data Segment a b = Empty | Segment b | Separator a
1275-- XXX splitOn and splitOnSuffix have a different behavior on an empty stream,
1276-- is that desirable?
1278-- | Split on an infixed separator element, dropping the separator. The
1279-- supplied 'Fold' is applied on the split segments. Splits the stream on
1280-- separator elements determined by the supplied predicate, separator is
1281-- considered as infixed between two segments:
1283-- >>> splitOn' p xs = Stream.toList $ Stream.splitOn p Fold.toList (Stream.fromList xs)
1284-- >>> splitOn' (== '.') "a.b"
1287-- An empty stream is folded to the default value of the fold:
1289-- >>> splitOn' (== '.') ""
1292-- If one or both sides of the separator are missing then the empty segment on
1293-- that side is folded to the default output of the fold:
1295-- >>> splitOn' (== '.') "."
1298-- >>> splitOn' (== '.') ".a"
1301-- >>> splitOn' (== '.') "a."
1304-- >>> splitOn' (== '.') "a..b"
1307-- splitOn is an inverse of intercalating single element:
1309-- @Stream.intercalate (Stream.yield '.') Unfold.fromList . Stream.splitOn (== '.') Fold.toList === id@
1311-- Assuming the input stream does not contain the separator:
1313-- @Stream.splitOn (== '.') Fold.toList . Stream.intercalate (Stream.yield '.') Unfold.fromList === id@
1317{-# INLINE splitOn #-}
1319 :: (IsStream t, Monad m)
1320 => (a -> Bool) -> Fold m a b -> t m a -> t m b
1321splitOn predicate f =
1322 -- We can express the infix splitting in terms of optional suffix split
1323 -- fold. After applying a suffix split fold repeatedly if the last segment
1324 -- ends with a suffix then we need to return the default output of the fold
1325 -- after that to make it an infix split.
1327 -- Alternately, we can also express it using an optional prefix split fold.
1328 -- If the first segment starts with a prefix then we need to emit the
1329 -- default output of the fold before that to make it an infix split, and
1330 -- then apply prefix split fold repeatedly.
1332 -- Since a suffix split fold can be easily expressed using a
1333 -- non-backtracking fold, we use that.
1334 foldManyPost (FL.takeEndBy_ predicate f)
1336-- | Split on a suffixed separator element, dropping the separator. The
1337-- supplied 'Fold' is applied on the split segments.
1339-- >>> splitOnSuffix' p xs = Stream.toList $ Stream.splitOnSuffix p Fold.toList (Stream.fromList xs)
1340-- >>> splitOnSuffix' (== '.') "a.b."
1343-- >>> splitOnSuffix' (== '.') "a."
1346-- An empty stream results in an empty output stream:
1348-- >>> splitOnSuffix' (== '.') ""
1351-- An empty segment consisting of only a suffix is folded to the default output
1354-- >>> splitOnSuffix' (== '.') "."
1357-- >>> splitOnSuffix' (== '.') "a..b.."
1360-- A suffix is optional at the end of the stream:
1362-- >>> splitOnSuffix' (== '.') "a"
1365-- >>> splitOnSuffix' (== '.') ".a"
1368-- >>> splitOnSuffix' (== '.') "a.b"
1371-- > lines = splitOnSuffix (== '\n')
1373-- 'splitOnSuffix' is an inverse of 'intercalateSuffix' with a single element:
1375-- @Stream.intercalateSuffix (Stream.yield '.') Unfold.fromList . Stream.splitOnSuffix (== '.') Fold.toList === id@
1377-- Assuming the input stream does not contain the separator:
1379-- @Stream.splitOnSuffix (== '.') Fold.toList . Stream.intercalateSuffix (Stream.yield '.') Unfold.fromList === id@
1383{-# INLINE splitOnSuffix #-}
1385 :: (IsStream t, Monad m)
1386 => (a -> Bool) -> Fold m a b -> t m a -> t m b
1387splitOnSuffix predicate f = foldMany (FL.takeEndBy_ predicate f)
1389-- | Split on a prefixed separator element, dropping the separator. The
1390-- supplied 'Fold' is applied on the split segments.
1393-- > splitOnPrefix' p xs = Stream.toList $ Stream.splitOnPrefix p (Fold.toList) (Stream.fromList xs)
1394-- > splitOnPrefix' (== '.') ".a.b"
1398-- An empty stream results in an empty output stream:
1400-- > splitOnPrefix' (== '.') ""
1404-- An empty segment consisting of only a prefix is folded to the default output
1408-- > splitOnPrefix' (== '.') "."
1411-- > splitOnPrefix' (== '.') ".a.b."
1414-- > splitOnPrefix' (== '.') ".a..b"
1419-- A prefix is optional at the beginning of the stream:
1422-- > splitOnPrefix' (== '.') "a"
1425-- > splitOnPrefix' (== '.') "a.b"
1429-- 'splitOnPrefix' is an inverse of 'intercalatePrefix' with a single element:
1431-- @Stream.intercalatePrefix (Stream.yield '.') Unfold.fromList . Stream.splitOnPrefix (== '.') Fold.toList === id@
1433-- Assuming the input stream does not contain the separator:
1435-- @Stream.splitOnPrefix (== '.') Fold.toList . Stream.intercalatePrefix (Stream.yield '.') Unfold.fromList === id@
1439{-# INLINE splitOnPrefix #-}
1440splitOnPrefix :: -- (IsStream t, MonadCatch m) =>
1441 (a -> Bool) -> Fold m a b -> t m a -> t m b
1442splitOnPrefix _predicate _f = undefined
1443 -- parseMany (Parser.sliceBeginBy predicate f)
1445-- | Like 'splitOn' after stripping leading, trailing, and repeated separators.
1446-- Therefore, @".a..b."@ with '.' as the separator would be parsed as
1447-- @["a","b"]@. In other words, its like parsing words from whitespace
1450-- >>> wordsBy' p xs = Stream.toList $ Stream.wordsBy p Fold.toList (Stream.fromList xs)
1452-- >>> wordsBy' (== ',') ""
1455-- >>> wordsBy' (== ',') ","
1458-- >>> wordsBy' (== ',') ",a,,b,"
1461-- > words = wordsBy isSpace
1465-- It is equivalent to splitting in any of the infix/prefix/suffix styles
1466-- followed by removal of empty segments.
1467{-# INLINE wordsBy #-}
1469 :: (IsStream t, Monad m)
1470 => (a -> Bool) -> Fold m a b -> t m a -> t m b
1471wordsBy predicate f m =
1472 D.fromStreamD $ D.wordsBy predicate f (D.toStreamD m)
1474-- | Like 'splitOnSuffix' but keeps the suffix attached to the resulting
1477-- >>> splitWithSuffix' p xs = Stream.toList $ splitWithSuffix p Fold.toList (Stream.fromList xs)
1479-- >>> splitWithSuffix' (== '.') ""
1482-- >>> splitWithSuffix' (== '.') "."
1485-- >>> splitWithSuffix' (== '.') "a"
1488-- >>> splitWithSuffix' (== '.') ".a"
1491-- >>> splitWithSuffix' (== '.') "a."
1494-- >>> splitWithSuffix' (== '.') "a.b"
1497-- >>> splitWithSuffix' (== '.') "a.b."
1500-- >>> splitWithSuffix' (== '.') "a..b.."
1501-- ["a.",".","b.","."]
1505{-# INLINE splitWithSuffix #-}
1507 :: (IsStream t, Monad m)
1508 => (a -> Bool) -> Fold m a b -> t m a -> t m b
1509splitWithSuffix predicate f = foldMany (FL.takeEndBy predicate f)
1511------------------------------------------------------------------------------
1512-- Splitting - on a delimiter sequence
1513------------------------------------------------------------------------------
1515-- Int list examples for splitOn:
1517-- >>> splitList [] [1,2,3,3,4]
1518-- > [[1],[2],[3],[3],[4]]
1520-- >>> splitList [5] [1,2,3,3,4]
1523-- >>> splitList [1] [1,2,3,3,4]
1526-- >>> splitList [4] [1,2,3,3,4]
1529-- >>> splitList [2] [1,2,3,3,4]
1532-- >>> splitList [3] [1,2,3,3,4]
1535-- >>> splitList [3,3] [1,2,3,3,4]
1538-- >>> splitList [1,2,3,3,4] [1,2,3,3,4]
1542-- This can be implemented easily using Rabin Karp
1543-- | Split on any one of the given patterns.
1544{-# INLINE splitOnAny #-}
1546 :: (IsStream t, Monad m, Storable a, Integral a)
1547 => [Array a] -> Fold m a b -> t m a -> t m b
1548splitOnAny subseq f m = undefined -- D.fromStreamD $ D.splitOnAny f subseq (D.toStreamD m)
1551-- XXX use a non-monadic intersperse to remove the MonadAsync constraint.
1552-- XXX Use two folds, one ring buffer fold for separator sequence and the other
1553-- split consumer fold. The input is fed to the ring fold first and the
1554-- rejected input is fed to the split fold. If the separator matches, the ring
1555-- fold would consume all.
1557-- | Like 'splitOnSeq' but splits the separator as well, as an infix token.
1559-- >>> splitOn'_ pat xs = Stream.toList $ Stream.splitBySeq (Array.fromList pat) Fold.toList (Stream.fromList xs)
1561-- >>> splitOn'_ "" "hello"
1562-- ["h","","e","","l","","l","","o"]
1564-- >>> splitOn'_ "hello" ""
1567-- >>> splitOn'_ "hello" "hello"
1570-- >>> splitOn'_ "x" "hello"
1573-- >>> splitOn'_ "h" "hello"
1576-- >>> splitOn'_ "o" "hello"
1579-- >>> splitOn'_ "e" "hello"
1582-- >>> splitOn'_ "l" "hello"
1583-- ["he","l","","l","o"]
1585-- >>> splitOn'_ "ll" "hello"
1589{-# INLINE splitBySeq #-}
1591 :: (IsStream t, MonadAsync m, Storable a, Enum a, Eq a)
1592 => Array a -> Fold m a b -> t m a -> t m b
1593splitBySeq patt f m =
1594 intersperseM (fold f (A.toStream patt)) $ splitOnSeq patt f m
1596-- | Like 'splitSuffixBy' but the separator is a sequence of elements, instead
1597-- of a predicate for a single element.
1599-- >>> splitOnSuffixSeq_ pat xs = Stream.toList $ Stream.splitOnSuffixSeq (Array.fromList pat) Fold.toList (Stream.fromList xs)
1601-- >>> splitOnSuffixSeq_ "." ""
1604-- >>> splitOnSuffixSeq_ "." "."
1607-- >>> splitOnSuffixSeq_ "." "a"
1610-- >>> splitOnSuffixSeq_ "." ".a"
1613-- >>> splitOnSuffixSeq_ "." "a."
1616-- >>> splitOnSuffixSeq_ "." "a.b"
1619-- >>> splitOnSuffixSeq_ "." "a.b."
1622-- >>> splitOnSuffixSeq_ "." "a..b.."
1625-- > lines = splitOnSuffixSeq "\n"
1627-- 'splitOnSuffixSeq' is an inverse of 'intercalateSuffix'. The following law
1630-- > intercalateSuffix . splitOnSuffixSeq == id
1632-- The following law holds when the separator is non-empty and contains none of
1633-- the elements present in the input lists:
1635-- > splitSuffixOn . intercalateSuffix == id
1638{-# INLINE splitOnSuffixSeq #-}
1640 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
1641 => Array a -> Fold m a b -> t m a -> t m b
1642splitOnSuffixSeq patt f m =
1643 D.fromStreamD $ D.splitOnSuffixSeq False patt f (D.toStreamD m)
1646-- | Like 'splitOn' but drops any empty splits.
1648{-# INLINE wordsOn #-}
1650 :: (IsStream t, Monad m, Storable a, Eq a)
1651 => Array a -> Fold m a b -> t m a -> t m b
1652wordsOn subseq f m = undefined -- D.fromStreamD $ D.wordsOn f subseq (D.toStreamD m)
1655-- | Like 'splitOnSuffixSeq' but keeps the suffix intact in the splits.
1657-- >>> splitWithSuffixSeq' pat xs = Stream.toList $ Stream.splitWithSuffixSeq (Array.fromList pat) Fold.toList (Stream.fromList xs)
1659-- >>> splitWithSuffixSeq' "." ""
1662-- >>> splitWithSuffixSeq' "." "."
1665-- >>> splitWithSuffixSeq' "." "a"
1668-- >>> splitWithSuffixSeq' "." ".a"
1671-- >>> splitWithSuffixSeq' "." "a."
1674-- >>> splitWithSuffixSeq' "." "a.b"
1677-- >>> splitWithSuffixSeq' "." "a.b."
1680-- >>> splitWithSuffixSeq' "." "a..b.."
1681-- ["a.",".","b.","."]
1684{-# INLINE splitWithSuffixSeq #-}
1686 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
1687 => Array a -> Fold m a b -> t m a -> t m b
1688splitWithSuffixSeq patt f m =
1689 D.fromStreamD $ D.splitOnSuffixSeq True patt f (D.toStreamD m)
1692-- This can be implemented easily using Rabin Karp
1693-- | Split post any one of the given patterns.
1694{-# INLINE splitOnSuffixSeqAny #-}
1696 :: (IsStream t, Monad m, Storable a, Integral a)
1697 => [Array a] -> Fold m a b -> t m a -> t m b
1698splitOnSuffixSeqAny subseq f m = undefined
1699 -- D.fromStreamD $ D.splitPostAny f subseq (D.toStreamD m)
1702------------------------------------------------------------------------------
1704------------------------------------------------------------------------------
1706-- | Group the input stream into groups of @n@ elements each and then fold each
1707-- group using the provided fold function.
1709-- >>> Stream.toList $ Stream.chunksOf 2 Fold.sum (Stream.enumerateFromTo 1 10)
1712-- This can be considered as an n-fold version of 'take' where we apply
1713-- 'take' repeatedly on the leftover stream until the stream exhausts.
1715-- @chunksOf n f = foldMany (FL.take n f)@
1718{-# INLINE chunksOf #-}
1720 :: (IsStream t, Monad m)
1721 => Int -> Fold m a b -> t m a -> t m b
1722chunksOf n f = D.fromStreamD . D.chunksOf n f . D.toStreamD
1727{-# INLINE chunksOf2 #-}
1729 :: (IsStream t, Monad m)
1730 => Int -> m c -> Fold2 m c a b -> t m a -> t m b
1731chunksOf2 n action f m = D.fromStreamD $ D.groupsOf2 n action f (D.toStreamD m)
1733-- | @arraysOf n stream@ groups the elements in the input stream into arrays of
1734-- @n@ elements each.
1736-- Same as the following but may be more efficient:
1738-- > arraysOf n = Stream.foldMany (A.writeN n)
1741{-# INLINE arraysOf #-}
1742arraysOf :: (IsStream t, MonadIO m, Storable a)
1743 => Int -> t m a -> t m (Array a)
1744arraysOf n = D.fromStreamD . A.arraysOf n . D.toStreamD
1746-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
1747-- XXX add this example after fixing the serial stream rate control
1749-- | Group the input stream into windows of @n@ second each and then fold each
1750-- group using the provided fold function.
1752-- >>> Stream.toList $ Stream.take 5 $ Stream.intervalsOf 1 Fold.sum $ Stream.constRate 2 $ Stream.enumerateFrom 1
1753-- [...,...,...,...,...]
1756{-# INLINE intervalsOf #-}
1758 :: (IsStream t, MonadAsync m)
1759 => Double -> Fold m a b -> t m a -> t m b
1761 splitWithSuffix isNothing (FL.catMaybes f)
1762 (interjectSuffix n (return Nothing) (Serial.map Just xs))
1764------------------------------------------------------------------------------
1765-- Windowed classification
1766------------------------------------------------------------------------------
1768-- We divide the stream into windows or chunks in space or time and each window
1769-- can be associated with a key, all events associated with a particular key in
1770-- the window can be folded to a single result. The stream can be split into
1771-- windows by size or by using a split predicate on the elements in the stream.
1772-- For example, when we receive a closing flag, we can close the window.
1774-- A "chunk" is a space window and a "session" is a time window. Are there any
1775-- other better short words to describe them. An alternative is to use
1776-- "swindow" and "twindow". Another word for "session" could be "spell".
1778-- TODO: To mark the position in space or time we can have Indexed or
1779-- TimeStamped types. That can make it easy to deal with the position indices
1782------------------------------------------------------------------------------
1783-- Keyed Sliding Windows
1784------------------------------------------------------------------------------
1787{-# INLINABLE classifySlidingChunks #-}
1788classifySlidingChunks
1789 :: (IsStream t, MonadAsync m, Ord k)
1790 => Int -- ^ window size
1791 -> Int -- ^ window slide
1792 -> Fold m a b -- ^ Fold to be applied to window events
1793 -> t m (k, a, Bool) -- ^ window key, data, close event
1795classifySlidingChunks wsize wslide (Fold step initial extract) str
1798-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
1799-- slide the send window when an ack is received and we slide the receive
1800-- window when a sequence is complete. Sliding is stateful in case of TCP,
1801-- sliding releases the send buffer or makes data available to the user from
1802-- the receive buffer.
1803{-# INLINABLE classifySlidingSessions #-}
1804classifySlidingSessions
1805 :: (IsStream t, MonadAsync m, Ord k)
1806 => Double -- ^ timer tick in seconds
1807 -> Double -- ^ time window size
1808 -> Double -- ^ window slide
1809 -> Fold m a b -- ^ Fold to be applied to window events
1810 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
1812classifySlidingSessions tick interval slide (Fold step initial extract) str
1816------------------------------------------------------------------------------
1817-- Sliding Window Buffers
1818------------------------------------------------------------------------------
1820-- These buffered versions could be faster than concurrent incremental folds of
1821-- all overlapping windows as in many cases we may not need all the values to
1822-- compute the fold, we can just compute the result using the old value and new
1823-- value. However, we may need the buffer once in a while, for example for
1824-- string search we usually compute the hash incrementally but when the hash
1825-- matches the hash of the pattern we need to compare the whole string.
1827-- XXX we should be able to implement sequence based splitting combinators
1828-- using this combinator.
1831-- | Buffer n elements of the input in a ring buffer. When t new elements are
1832-- collected, slide the window to remove the same number of oldest elements,
1833-- insert the new elements, and apply an incremental fold on the sliding
1834-- window, supplying the outgoing elements, the new ring buffer as arguments.
1836 :: (IsStream t, Monad m, Ord a, Storable a)
1837 => Int -- window size
1838 -> Int -- window slide
1839 -> Fold m (Ring a, Array a) b
1842slidingChunkBuffer = undefined
1844-- Buffer n seconds worth of stream elements of the input in a radix tree.
1845-- Every t seconds, remove the items that are older than n seconds, and apply
1846-- an incremental fold on the sliding window, supplying the outgoing elements,
1847-- and the new radix tree buffer as arguments.
1849 :: (IsStream t, Monad m, Ord a, Storable a)
1850 => Int -- window size
1852 -> Fold m (RTree a, Array a) b
1855slidingSessionBuffer = undefined
1858------------------------------------------------------------------------------
1859-- Keyed Session Windows
1860------------------------------------------------------------------------------
1863-- | Keyed variable size space windows. Close the window if we do not receive a
1864-- window event in the next "spaceout" elements.
1865{-# INLINABLE classifyChunksBy #-}
1867 :: (IsStream t, MonadAsync m, Ord k)
1868 => Int -- ^ window spaceout (spread)
1869 -> Bool -- ^ reset the spaceout when a chunk window element is received
1870 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1871 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1873classifyChunksBy spanout reset (Fold step initial extract) str = undefined
1875-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
1876-- received within the chunk size window. The chunk gets closed only if no
1877-- element is received within the chunk window.
1879{-# INLINABLE classifyKeepAliveChunks #-}
1880classifyKeepAliveChunks
1881 :: (IsStream t, MonadAsync m, Ord k)
1882 => Int -- ^ window spaceout (spread)
1883 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1884 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1886classifyKeepAliveChunks spanout = classifyChunksBy spanout True
1889data SessionState t m k a b = SessionState
1890 { sessionCurTime :: !AbsTime -- ^ time since last event
1891 , sessionEventTime :: !AbsTime -- ^ time as per last event
1892 , sessionCount :: !Int -- ^ total number sessions in progress
1893 , sessionTimerHeap :: H.Heap (H.Entry AbsTime k) -- ^ heap for timeouts
1894 , sessionKeyValueMap :: Map.Map k a -- ^ Stored sessions for keys
1895 , sessionOutputStream :: t (m :: Type -> Type) (k, b) -- ^ Completed sessions
1898-- | @classifySessionsBy tick keepalive predicate timeout fold stream@
1899-- classifies an input event @stream@ consisting of @(timestamp, (key,
1900-- value))@ into sessions based on the @key@, folding all the values
1901-- corresponding to the same key into a session using the supplied @fold@.
1903-- When the fold terminates or a @timeout@ occurs, a tuple consisting of the
1904-- session key and the folded value is emitted in the output stream. The
1905-- timeout is measured from the first event in the session. If the @keepalive@
1906-- option is set to 'True' the timeout is reset to 0 whenever an event is
1909-- The @timestamp@ in the input stream is an absolute time from some epoch,
1910-- characterizing the time when the input event was generated. The notion of
1911-- current time is maintained by a monotonic event time clock using the
1912-- timestamps seen in the input stream. The latest timestamp seen till now is
1913-- used as the base for the current time. When no new events are seen, a timer
1914-- is started with a clock resolution of @tick@ seconds. This timer is used to
1915-- detect session timeouts in the absence of new events.
1917-- To ensure an upper bound on the memory used the number of sessions can be
1918-- limited to an upper bound. If the ejection @predicate@ returns 'True', the
1919-- oldest session is ejected before inserting a new session.
1922-- Stream.mapM_ print
1923-- $ Stream.classifySessionsBy 1 False (const (return False)) 3 (Fold.take 3 Fold.toList)
1924-- $ Stream.timestamped
1925-- $ Stream.delay 0.1
1926-- $ (,) <$> Stream.fromList [1,2,3] <*> Stream.fromList ['a','b','c']
1934{-# INLINABLE classifySessionsBy #-}
1936 :: (IsStream t, MonadAsync m, Ord k)
1937 => Double -- ^ timer tick in seconds
1938 -> Bool -- ^ reset the timer when an event is received
1939 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
1940 -> Double -- ^ session timeout in seconds
1941 -> Fold m a b -- ^ Fold to be applied to session data
1942 -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
1943 -> t m (k, b) -- ^ session key, fold result
1944classifySessionsBy tick reset ejectPred tmout
1945 (Fold step initial extract) str =
1946 concatMap sessionOutputStream $
1947 scanlMAfter' sstep (return szero) flush stream
1951 timeoutMs = toRelTime (round (tmout * 1000) :: MilliSecond64)
1952 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
1953 szero = SessionState
1954 { sessionCurTime = toAbsTime (0 :: MilliSecond64)
1955 , sessionEventTime = toAbsTime (0 :: MilliSecond64)
1957 , sessionTimerHeap = H.empty
1958 , sessionKeyValueMap = Map.empty
1959 , sessionOutputStream = K.nil
1962 -- We can eject sessions based on the current session count to limit
1963 -- memory consumption. There are two possible strategies:
1965 -- 1) Eject old sessions or sessions beyond a certain/lower timeout
1966 -- threshold even before timeout, effectively reduce the timeout.
1967 -- 2) Drop creation of new sessions but keep accepting new events for the
1970 -- We use the first strategy as of now.
1972 -- Got a new stream input element
1973 sstep session@SessionState{..} (Just (timestamp, (key, value))) = do
1974 -- XXX we should use a heap in pinned memory to scale it to a large
1977 -- To detect session inactivity we keep a timestamp of the latest event
1978 -- in the Map along with the fold result. When we purge the session
1979 -- from the heap we match the timestamp in the heap with the timestamp
1980 -- in the Map, if the latest timestamp is newer and has not expired we
1981 -- reinsert the key in the heap.
1983 -- XXX if the key is an Int, we can also use an IntMap for slightly
1984 -- better performance.
1986 let curTime = max sessionEventTime timestamp
1987 mOld = Map.lookup key sessionKeyValueMap
1989 -- deleting a key from the heap is expensive, so we never
1990 -- delete a key from heap, we just purge it from the Map and it
1991 -- gets purged from the heap on timeout. We just need an extra
1992 -- lookup in the Map when the key is purged from the heap, that
1993 -- should not be expensive.
1995 let (mp, cnt) = case mOld of
1996 Nothing -> (sessionKeyValueMap, sessionCount)
1997 Just _ -> (Map.delete key sessionKeyValueMap
2000 { sessionCurTime = curTime
2001 , sessionEventTime = curTime
2002 , sessionCount = cnt
2003 , sessionKeyValueMap = mp
2004 , sessionOutputStream = yield (key, fb)
2007 let acc = Tuple' timestamp fs1
2008 (hp1, mp1, out1, cnt1) <- do
2009 let vars = (sessionTimerHeap, sessionKeyValueMap,
2010 K.nil, sessionCount)
2012 -- inserting new entry
2014 -- Eject a session from heap and map is needed
2015 eject <- ejectPred sessionCount
2016 (hp, mp, out, cnt) <-
2021 -- Insert the new session in heap
2022 let expiry = addToAbsTime timestamp timeoutMs
2023 hp' = H.insert (Entry expiry key) hp
2024 in return (hp', mp, out, cnt + 1)
2025 -- updating old entry
2026 Just _ -> return vars
2028 let mp2 = Map.insert key acc mp1
2029 return $ SessionState
2030 { sessionCurTime = curTime
2031 , sessionEventTime = curTime
2032 , sessionCount = cnt1
2033 , sessionTimerHeap = hp1
2034 , sessionKeyValueMap = mp2
2035 , sessionOutputStream = out1
2040 Just (Tuple' _ acc) -> return $ FL.Partial acc
2042 FL.Done fb -> done fb
2044 res <- step fs value
2046 FL.Done fb -> done fb
2047 FL.Partial fs1 -> partial fs1
2049 -- Got a timer tick event
2050 sstep sessionState@SessionState{..} Nothing =
2051 let curTime = addToAbsTime sessionCurTime tickMs
2052 in ejectExpired sessionState curTime
2054 flush session@SessionState{..} = do
2055 (hp', mp', out, count) <-
2058 , sessionKeyValueMap
2063 { sessionCount = count
2064 , sessionTimerHeap = hp'
2065 , sessionKeyValueMap = mp'
2066 , sessionOutputStream = out
2069 -- delete from map and output the fold accumulator
2070 ejectEntry hp mp out cnt acc key = do
2072 let out1 = (key, sess) `K.cons` out
2073 let mp1 = Map.delete key mp
2074 return (hp, mp1, out1, cnt - 1)
2076 ejectAll (hp, mp, out, !cnt) = do
2077 let hres = H.uncons hp
2079 Just (Entry _ key, hp1) -> do
2080 r <- case Map.lookup key mp of
2081 Nothing -> return (hp1, mp, out, cnt)
2082 Just (Tuple' _ acc) -> ejectEntry hp1 mp out cnt acc key
2085 assert (Map.null mp) (return ())
2086 return (hp, mp, out, cnt)
2088 ejectOne (hp, mp, out, !cnt) = do
2089 let hres = H.uncons hp
2091 Just (Entry expiry key, hp1) ->
2092 case Map.lookup key mp of
2093 Nothing -> ejectOne (hp1, mp, out, cnt)
2094 Just (Tuple' latestTS acc) -> do
2095 let expiry1 = addToAbsTime latestTS timeoutMs
2096 if not reset || expiry1 <= expiry
2097 then ejectEntry hp1 mp out cnt acc key
2099 -- reset the session timeout and continue
2100 let hp2 = H.insert (Entry expiry1 key) hp1
2101 in ejectOne (hp2, mp, out, cnt)
2103 assert (Map.null mp) (return ())
2104 return (hp, mp, out, cnt)
2106 ejectExpired session@SessionState{..} curTime = do
2107 (hp', mp', out, count) <-
2108 ejectLoop sessionTimerHeap sessionKeyValueMap K.nil sessionCount
2110 { sessionCurTime = curTime
2111 , sessionCount = count
2112 , sessionTimerHeap = hp'
2113 , sessionKeyValueMap = mp'
2114 , sessionOutputStream = out
2119 ejectLoop hp mp out !cnt = do
2120 let hres = H.uncons hp
2122 Just (Entry expiry key, hp1) -> do
2124 if curTime >= expiry
2125 then return (True, False)
2131 case Map.lookup key mp of
2132 Nothing -> ejectLoop hp1 mp out cnt
2133 Just (Tuple' latestTS acc) -> do
2134 let expiry1 = addToAbsTime latestTS timeoutMs
2135 if expiry1 <= curTime || not reset || force
2137 (hp2,mp1,out1,cnt1) <-
2138 ejectEntry hp1 mp out cnt acc key
2139 ejectLoop hp2 mp1 out1 cnt1
2141 -- reset the session timeout and continue
2142 let hp2 = H.insert (Entry expiry1 key) hp1
2143 in ejectLoop hp2 mp out cnt
2144 else return (hp, mp, out, cnt)
2146 assert (Map.null mp) (return ())
2147 return (hp, mp, out, cnt)
2149 -- merge timer events in the stream
2150 stream = Serial.map Just str `Par.parallelFst` repeatM timer
2152 liftIO $ threadDelay (round $ tick * 1000000)
2155-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
2156-- option set to 'True'.
2159-- classifyKeepAliveSessions = classifySessionsBy 1 True
2164{-# INLINABLE classifyKeepAliveSessions #-}
2165classifyKeepAliveSessions ::
2166 (IsStream t, MonadAsync m, Ord k)
2167 => (Int -> m Bool) -- ^ predicate to eject sessions on session count
2168 -> Double -- ^ session inactive timeout
2169 -> Fold m a b -- ^ Fold to be applied to session payload data
2170 -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
2172classifyKeepAliveSessions = classifySessionsBy 1 True
2174------------------------------------------------------------------------------
2175-- Keyed tumbling windows
2176------------------------------------------------------------------------------
2178-- Tumbling windows is a special case of sliding windows where the window slide
2179-- is the same as the window size. Or it can be a special case of session
2180-- windows where the reset flag is set to False.
2182-- XXX instead of using the early termination flag in the stream, we can use an
2183-- early terminating fold instead.
2186-- | Split the stream into fixed size chunks of specified size. Within each
2187-- such chunk fold the elements in buckets identified by the keys. A particular
2188-- bucket fold can be terminated early if a closing flag is encountered in an
2189-- element for that key.
2192{-# INLINABLE classifyChunksOf #-}
2194 :: (IsStream t, MonadAsync m, Ord k)
2195 => Int -- ^ window size
2196 -> Fold m a b -- ^ Fold to be applied to window events
2197 -> t m (k, a, Bool) -- ^ window key, data, close event
2199classifyChunksOf wsize = classifyChunksBy wsize False
2202-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
2203-- option set to 'False'.
2206-- classifySessionsOf = classifySessionsBy 1 False
2211{-# INLINABLE classifySessionsOf #-}
2212classifySessionsOf ::
2213 (IsStream t, MonadAsync m, Ord k)
2214 => (Int -> m Bool) -- ^ predicate to eject sessions on session count
2215 -> Double -- ^ time window size
2216 -> Fold m a b -- ^ Fold to be applied to session data
2217 -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
2219classifySessionsOf = classifySessionsBy 1 False
2221------------------------------------------------------------------------------
2223------------------------------------------------------------------------------
2225-- | @splitInnerBy splitter joiner stream@ splits the inner containers @f a@ of
2226-- an input stream @t m (f a)@ using the @splitter@ function. Container
2227-- elements @f a@ are collected until a split occurs, then all the elements
2228-- before the split are joined using the @joiner@ function.
2230-- For example, if we have a stream of @Array Word8@, we may want to split the
2231-- stream into arrays representing lines separated by '\n' byte such that the
2232-- resulting stream after a split would be one array for each line.
2234-- CAUTION! This is not a true streaming function as the container size after
2235-- the split and merge may not be bounded.
2238{-# INLINE splitInnerBy #-}
2240 :: (IsStream t, Monad m)
2241 => (f a -> m (f a, Maybe (f a))) -- splitter
2242 -> (f a -> f a -> m (f a)) -- joiner
2245splitInnerBy splitter joiner xs =
2246 D.fromStreamD $ D.splitInnerBy splitter joiner $ D.toStreamD xs
2248-- | Like 'splitInnerBy' but splits assuming the separator joins the segment in
2252{-# INLINE splitInnerBySuffix #-}
2254 :: (IsStream t, Monad m, Eq (f a), Monoid (f a))
2255 => (f a -> m (f a, Maybe (f a))) -- splitter
2256 -> (f a -> f a -> m (f a)) -- joiner
2259splitInnerBySuffix splitter joiner xs =
2260 D.fromStreamD $ D.splitInnerBySuffix splitter joiner $ D.toStreamD xs
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