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
90 -- | Unfold and flatten streams.
92 , concatUnfoldInterleave
93 , concatUnfoldRoundrobin
96 -- | Insert effects between streams. Like concatUnfold but intersperses an
97 -- effect between the streams. A
98 -- special case of gintercalate.
104 -- | Insert Streams between Streams.
105 -- Like concatUnfold but intersperses streams from another source between
106 -- the streams from the first source.
113 -- | Map and flatten Trees of Streams
116 , iterateMapLeftsWith
119 -- | Folding and Parsing chunks of streams to eliminate nested streams.
120 -- Functions generally ending in these shapes:
123 -- f (Fold m a b) -> t m a -> t m b
124 -- f (Parser m a b) -> t m a -> t m b
128 -- | Apply folds on a stream.
134 -- | Apply parsers on a stream.
142 -- | Element unaware grouping.
149 -- In imperative terms, grouped folding can be considered as a nested loop
150 -- where we loop over the stream to group elements and then loop over
151 -- individual groups to fold them to a single value that is yielded in the
158 -- -- *** Searching Sequences
159 -- , seqIndices -- search a sequence in the stream
161 -- -- *** Searching Multiple Sequences
162 -- , seqIndicesAny -- search any of the given sequence in the stream
164 -- -- -- ** Searching Streams
165 -- -- | Finding a stream within another stream.
168 -- | Streams can be sliced into segments in space or in time. We use the
169 -- term @chunk@ to refer to a spatial length of the stream (spatial window)
170 -- and the term @session@ to refer to a length in time (time window).
172 -- -- *** Using Element Separators
180 , wordsBy -- stripAndCompactBy
182 -- -- *** Splitting By Sequences
186 -- , splitOnPrefixSeq
188 -- Keeping the delimiters
190 -- , splitByPrefixSeq
193 -- Splitting using multiple sequence separators
195 -- , splitOnAnySuffixSeq
196 -- , splitOnAnyPrefixSeq
198 -- -- *** Splitting By Streams
199 -- -- | Splitting a stream using another stream as separator.
201 -- ** Windowed Classification
203 -- | Split the stream into windows or chunks in space or time. Each window
204 -- can be associated with a key, all events associated with a particular
205 -- key in the window can be folded to a single result. The stream is split
206 -- into windows of specified size, the window can be terminated early if
207 -- the closing flag is specified in the input stream.
209 -- The term "chunk" is used for a space window and the term "session" is
210 -- used for a time window.
212 -- *** Tumbling Windows
213 -- | A new window starts after the previous window is finished.
215 -- , classifyChunksOf
219 -- *** Keep Alive Windows
220 -- | The window size is extended if an event arrives within the specified
221 -- window size. This can represent sessions with idle or inactive timeout.
223 -- , classifyKeepAliveChunks
224 , classifyKeepAliveSessions
227 -- *** Sliding Windows
228 -- | A new window starts after the specified slide from the previous
229 -- window. Therefore windows can overlap.
230 , classifySlidingChunks
231 , classifySlidingSessions
232 -- *** Sliding Window Buffers
233 -- , slidingChunkBuffer
234 -- , slidingSessionBuffer
237 -- * Transform (Nested Containers)
246import Control.Concurrent (threadDelay)
247import Control.Exception (assert)
248import Control.Monad.Catch (MonadThrow)
249import Control.Monad.IO.Class (MonadIO(..))
250import Data.Heap (Entry(..))
251import Data.Kind (Type)
252import Data.Maybe (isNothing)
253import Foreign.Storable (Storable)
254import Streamly.Internal.Data.Fold.Types (Fold (..), Fold2 (..))
255import Streamly.Internal.Data.Parser (Parser (..))
256import Streamly.Internal.Data.Unfold.Types (Unfold)
257import Streamly.Internal.Data.Array.Storable.Foreign.Types
258 (Array, writeNUnsafe)
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.Storable.Foreign.Types as A
289import qualified Streamly.Internal.Data.Fold as FL
290import qualified Streamly.Internal.Data.Parser.ParserK.Types 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)
305-------------------------------------------------------------------------------
307-------------------------------------------------------------------------------
309-- | Append the outputs of two streams, yielding all the elements from the
310-- first stream and then yielding all the elements from the second stream.
312-- IMPORTANT NOTE: This could be 100x faster than @serial/<>@ for appending a
313-- few (say 100) streams because it can fuse via stream fusion. However, it
314-- does not scale for a large number of streams (say 1000s) and becomes
315-- qudartically slow. Therefore use this for custom appending of a few streams
316-- but use 'concatMap' or 'concatMapWith serial' for appending @n@ streams or
317-- infinite containers of streams.
321append ::(IsStream t, Monad m) => t m b -> t m b -> t m b
322append m1 m2 = fromStreamD $ D.append (toStreamD m1) (toStreamD m2)
324-------------------------------------------------------------------------------
326-------------------------------------------------------------------------------
328-- XXX Same as 'wSerial'. We should perhaps rename wSerial to interleave.
329-- XXX Document the interleaving behavior of side effects in all the
330-- interleaving combinators.
331-- XXX Write time-domain equivalents of these. In the time domain we can
332-- interleave two streams such that the value of second stream is always taken
333-- from its last value even if no new value is being yielded, like
334-- zipWithLatest. It would be something like interleaveWithLatest.
336-- | Interleaves the outputs of two streams, yielding elements from each stream
337-- alternately, starting from the first stream. If any of the streams finishes
338-- early the other stream continues alone until it too finishes.
340-- >>> :set -XOverloadedStrings
341-- >>> interleave "ab" ",,,," :: SerialT Identity Char
343-- >>> interleave "abcd" ",," :: SerialT Identity Char
346-- 'interleave' is dual to 'interleaveMin', it can be called @interleaveMax@.
348-- Do not use at scale in concatMapWith.
351{-# INLINE interleave #-}
352interleave ::(IsStream t, Monad m) => t m b -> t m b -> t m b
353interleave m1 m2 = fromStreamD $ D.interleave (toStreamD m1) (toStreamD m2)
355-- | Interleaves the outputs of two streams, yielding elements from each stream
356-- alternately, starting from the first stream. As soon as the first stream
357-- finishes, the output stops, discarding the remaining part of the second
358-- stream. In this case, the last element in the resulting stream would be from
359-- the second stream. If the second stream finishes early then the first stream
360-- still continues to yield elements until it finishes.
362-- >>> :set -XOverloadedStrings
363-- >>> interleaveSuffix "abc" ",,,," :: SerialT Identity Char
365-- >>> interleaveSuffix "abc" "," :: SerialT Identity Char
368-- 'interleaveSuffix' is a dual of 'interleaveInfix'.
370-- Do not use at scale in concatMapWith.
373{-# INLINE interleaveSuffix #-}
374interleaveSuffix ::(IsStream t, Monad m) => t m b -> t m b -> t m b
375interleaveSuffix m1 m2 =
376 fromStreamD $ D.interleaveSuffix (toStreamD m1) (toStreamD m2)
378-- | Interleaves the outputs of two streams, yielding elements from each stream
379-- alternately, starting from the first stream and ending at the first stream.
380-- If the second stream is longer than the first, elements from the second
381-- stream are infixed with elements from the first stream. If the first stream
382-- is longer then it continues yielding elements even after the second stream
385-- >>> :set -XOverloadedStrings
386-- >>> interleaveInfix "abc" ",,,," :: SerialT Identity Char
388-- >>> interleaveInfix "abc" "," :: SerialT Identity Char
391-- 'interleaveInfix' is a dual of 'interleaveSuffix'.
393-- Do not use at scale in concatMapWith.
396{-# INLINE interleaveInfix #-}
397interleaveInfix ::(IsStream t, Monad m) => t m b -> t m b -> t m b
398interleaveInfix m1 m2 =
399 fromStreamD $ D.interleaveInfix (toStreamD m1) (toStreamD m2)
401-- | Interleaves the outputs of two streams, yielding elements from each stream
402-- alternately, starting from the first stream. The output stops as soon as any
403-- of the two streams finishes, discarding the remaining part of the other
404-- stream. The last element of the resulting stream would be from the longer
407-- >>> :set -XOverloadedStrings
408-- >>> interleaveMin "ab" ",,,," :: SerialT Identity Char
410-- >>> interleaveMin "abcd" ",," :: SerialT Identity Char
413-- 'interleaveMin' is dual to 'interleave'.
415-- Do not use at scale in concatMapWith.
418{-# INLINE interleaveMin #-}
419interleaveMin ::(IsStream t, Monad m) => t m b -> t m b -> t m b
420interleaveMin m1 m2 = fromStreamD $ D.interleaveMin (toStreamD m1) (toStreamD m2)
422-------------------------------------------------------------------------------
424-------------------------------------------------------------------------------
426-- | Schedule the execution of two streams in a fair round-robin manner,
427-- executing each stream once, alternately. Execution of a stream may not
428-- necessarily result in an output, a stream may chose to @Skip@ producing an
429-- element until later giving the other stream a chance to run. Therefore, this
430-- combinator fairly interleaves the execution of two streams rather than
431-- fairly interleaving the output of the two streams. This can be useful in
432-- co-operative multitasking without using explicit threads. This can be used
433-- as an alternative to `async`.
435-- Do not use at scale in concatMapWith.
438{-# INLINE roundrobin #-}
439roundrobin ::(IsStream t, Monad m) => t m b -> t m b -> t m b
440roundrobin m1 m2 = fromStreamD $ D.roundRobin (toStreamD m1) (toStreamD m2)
442------------------------------------------------------------------------------
444------------------------------------------------------------------------------
446-- | Merge two streams using a comparison function. The head elements of both
447-- the streams are compared and the smaller of the two elements is emitted, if
448-- both elements are equal then the element from the first stream is used
451-- If the streams are sorted in ascending order, the resulting stream would
452-- also remain sorted in ascending order.
455-- > S.toList $ S.mergeBy compare (S.fromList [1,3,5]) (S.fromList [2,4,6,8])
460{-# INLINABLE mergeBy #-}
462 (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> t m a -> t m a
463mergeBy f m1 m2 = fromStreamS $ S.mergeBy f (toStreamS m1) (toStreamS m2)
465-- | Like 'mergeBy' but with a monadic comparison function.
467-- Merge two streams randomly:
470-- > randomly _ _ = randomIO >>= \x -> return $ if x then LT else GT
471-- > S.toList $ S.mergeByM randomly (S.fromList [1,1,1,1]) (S.fromList [2,2,2,2])
475-- Merge two streams in a proportion of 2:1:
478-- proportionately m n = do
479-- ref <- newIORef $ cycle $ concat [replicate m LT, replicate n GT]
480-- return $ \\_ _ -> do
482-- writeIORef ref $ tail r
486-- f <- proportionately 2 1
487-- xs <- S.toList $ S.mergeByM f (S.fromList [1,1,1,1,1,1]) (S.fromList [2,2,2])
491-- [1,1,2,1,1,2,1,1,2]
495{-# INLINABLE mergeByM #-}
497 :: (IsStream t, Monad m)
498 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
499mergeByM f m1 m2 = fromStreamS $ S.mergeByM f (toStreamS m1) (toStreamS m2)
502-- | Like 'mergeByM' but stops merging as soon as any of the two streams stops.
503{-# INLINABLE mergeEndByAny #-}
505 :: (IsStream t, Monad m)
506 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
507mergeEndByAny f m1 m2 = fromStreamD $
508 D.mergeEndByAny f (toStreamD m1) (toStreamD m2)
510-- Like 'mergeByM' but stops merging as soon as the first stream stops.
511{-# INLINABLE mergeEndByFirst #-}
513 :: (IsStream t, Monad m)
514 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
515mergeEndByFirst f m1 m2 = fromStreamS $
516 D.mergeEndByFirst f (toStreamD m1) (toStreamD m2)
518-- Holding this back for now, we may want to use the name "merge" differently
519-- | Same as @'mergeBy' 'compare'@.
522-- > S.toList $ S.merge (S.fromList [1,3,5]) (S.fromList [2,4,6,8])
527{-# INLINABLE merge #-}
529 (IsStream t, Monad m, Ord a) => t m a -> t m a -> t m a
530merge = mergeBy compare
533-- | Like 'mergeBy' but merges concurrently (i.e. both the elements being
534-- merged are generated concurrently).
537{-# INLINE mergeAsyncBy #-}
538mergeAsyncBy :: (IsStream t, MonadAsync m)
539 => (a -> a -> Ordering) -> t m a -> t m a -> t m a
540mergeAsyncBy f = mergeAsyncByM (\a b -> return $ f a b)
542-- | Like 'mergeByM' but merges concurrently (i.e. both the elements being
543-- merged are generated concurrently).
546{-# INLINE mergeAsyncByM #-}
547mergeAsyncByM :: (IsStream t, MonadAsync m)
548 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
549mergeAsyncByM f m1 m2 =
551 let par = Par.mkParallelD . toStreamD
552 in D.mergeByM f (par m1) (par m2)
554------------------------------------------------------------------------------
555-- Combine N Streams - concatMap
556------------------------------------------------------------------------------
558-- | @concatMapWith merge map stream@ is a two dimensional looping combinator.
559-- The first argument specifies a merge or concat function that is used to
560-- merge the streams generated by applying the second argument i.e. the @map@
561-- function to each element of the input stream. The concat function could be
562-- 'serial', 'parallel', 'async', 'ahead' or any other zip or merge function
563-- and the second argument could be any stream generation function using a
566-- /Compare 'foldMapWith'/
570-- /Since: 0.8.0 (signature change)/
571{-# INLINE concatMapWith #-}
574 => (t m b -> t m b -> t m b)
578concatMapWith = K.concatMapBy
580-- | Like 'concatMapWith' but carries a state which can be used to share
581-- information across multiple steps of concat.
584-- concatSmapMWith combine f initial = concatMapWith combine id . smapM f initial
589{-# INLINE concatSmapMWith #-}
591 :: (IsStream t, Monad m)
592 => (t m b -> t m b -> t m b)
593 -> (s -> a -> m (s, t m b))
597concatSmapMWith combine f initial = concatMapWith combine id . smapM f initial
599-- | Map a stream producing function on each element of the stream and then
600-- flatten the results into a single stream.
603-- concatMap f = 'concat . map f'
604-- concatMap = 'concatMapWith' 'Serial.serial'
605-- concatMap f = 'concatMapM' (return . f)
606-- concatMap f = 'concatUnfold' (UF.lmap f UF.fromStream)
610{-# INLINE concatMap #-}
611concatMap ::(IsStream t, Monad m) => (a -> t m b) -> t m a -> t m b
612concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)
614-- | Flatten a stream of streams to a single stream.
617-- concat = concatMap id
622concat :: (IsStream t, Monad m) => t m (t m a) -> t m a
625-- Keep concating either streams as long as rights are generated, stop as soon
626-- as a left is generated and concat the left stream.
634 :: -- (IsStream t, MonadAsync m) =>
635 (forall x. t m x -> t m x -> t m x)
636 -> (a -> t m (Either (t m b) b))
639concatMapEitherWith = undefined
642------------------------------------------------------------------------------
643-- Combine N Streams - concatUnfold
644------------------------------------------------------------------------------
646-- | Like 'concatMap' but uses an 'Unfold' for stream generation. Unlike
647-- 'concatMap' this can fuse the 'Unfold' code with the inner loop and
648-- therefore provide many times better performance.
651{-# INLINE concatUnfold #-}
652concatUnfold ::(IsStream t, Monad m) => Unfold m a b -> t m a -> t m b
653concatUnfold u m = fromStreamD $ D.concatMapU u (toStreamD m)
655-- | Like 'concatUnfold' but interleaves the streams in the same way as
656-- 'interleave' behaves instead of appending them.
659{-# INLINE concatUnfoldInterleave #-}
660concatUnfoldInterleave ::(IsStream t, Monad m)
661 => Unfold m a b -> t m a -> t m b
662concatUnfoldInterleave u m =
663 fromStreamD $ D.concatUnfoldInterleave u (toStreamD m)
665-- | Like 'concatUnfold' but executes the streams in the same way as
669{-# INLINE concatUnfoldRoundrobin #-}
670concatUnfoldRoundrobin ::(IsStream t, Monad m)
671 => Unfold m a b -> t m a -> t m b
672concatUnfoldRoundrobin u m =
673 fromStreamD $ D.concatUnfoldRoundrobin u (toStreamD m)
675------------------------------------------------------------------------------
676-- Combine N Streams - interpose
677------------------------------------------------------------------------------
679-- > interpose x unf str = gintercalate unf str UF.identity (repeat x)
681-- | Unfold the elements of a stream, intersperse the given element between the
682-- unfolded streams and then concat them into a single stream.
684-- > unwords = S.interpose ' '
687{-# INLINE interpose #-}
688interpose :: (IsStream t, Monad m)
689 => c -> Unfold m b c -> t m b -> t m c
691 D.fromStreamD $ D.interpose (return x) unf (D.toStreamD str)
693-- interposeSuffix x unf str = gintercalateSuffix unf str UF.identity (repeat x)
695-- | Unfold the elements of a stream, append the given element after each
696-- unfolded stream and then concat them into a single stream.
698-- > unlines = S.interposeSuffix '\n'
701{-# INLINE interposeSuffix #-}
702interposeSuffix :: (IsStream t, Monad m)
703 => c -> Unfold m b c -> t m b -> t m c
704interposeSuffix x unf str =
705 D.fromStreamD $ D.interposeSuffix (return x) unf (D.toStreamD str)
707------------------------------------------------------------------------------
708-- Combine N Streams - intercalate
709------------------------------------------------------------------------------
711-- XXX we can swap the order of arguments to gintercalate so that the
712-- definition of concatUnfold becomes simpler? The first stream should be
713-- infixed inside the second one. However, if we change the order in
714-- "interleave" as well similarly, then that will make it a bit unintuitive.
716-- > concatUnfold unf str =
717-- > gintercalate unf str (UF.nilM (\_ -> return ())) (repeat ())
719-- | 'interleaveInfix' followed by unfold and concat.
722{-# INLINE gintercalate #-}
724 :: (IsStream t, Monad m)
725 => Unfold m a c -> t m a -> Unfold m b c -> t m b -> t m c
726gintercalate unf1 str1 unf2 str2 =
727 D.fromStreamD $ D.gintercalate
728 unf1 (D.toStreamD str1)
729 unf2 (D.toStreamD str2)
731-- XXX The order of arguments in "intercalate" is consistent with the list
732-- intercalate but inconsistent with gintercalate and other stream interleaving
733-- combinators. We can change the order of the arguments in other combinators
734-- but then 'interleave' combinator may become a bit unintuitive because we
735-- will be starting with the second stream.
737-- > intercalate seed unf str = gintercalate unf str unf (repeatM seed)
738-- > intercalate a unf str = concatUnfold unf $ intersperse a str
740-- | 'intersperse' followed by unfold and concat.
742-- > unwords = intercalate " " UF.fromList
744-- >>> intercalate " " UF.fromList ["abc", "def", "ghi"]
748{-# INLINE intercalate #-}
749intercalate :: (IsStream t, Monad m)
750 => b -> Unfold m b c -> t m b -> t m c
751intercalate seed unf str = D.fromStreamD $
752 D.concatMapU unf $ D.intersperse seed (toStreamD str)
754-- | 'interleaveSuffix' followed by unfold and concat.
757{-# INLINE gintercalateSuffix #-}
759 :: (IsStream t, Monad m)
760 => Unfold m a c -> t m a -> Unfold m b c -> t m b -> t m c
761gintercalateSuffix unf1 str1 unf2 str2 =
762 D.fromStreamD $ D.gintercalateSuffix
763 unf1 (D.toStreamD str1)
764 unf2 (D.toStreamD str2)
766-- > intercalateSuffix seed unf str = gintercalateSuffix unf str unf (repeatM seed)
767-- > intercalateSuffix a unf str = concatUnfold unf $ intersperseSuffix a str
769-- | 'intersperseSuffix' followed by unfold and concat.
771-- > unlines = intercalateSuffix "\n" UF.fromList
773-- >>> intercalate "\n" UF.fromList ["abc", "def", "ghi"]
774-- > "abc\ndef\nghi\n"
777{-# INLINE intercalateSuffix #-}
778intercalateSuffix :: (IsStream t, Monad m)
779 => b -> Unfold m b c -> t m b -> t m c
780intercalateSuffix seed unf str = fromStreamD $ D.concatMapU unf
781 $ D.intersperseSuffix (return seed) (D.toStreamD str)
783------------------------------------------------------------------------------
784-- IterateMap - Map and flatten Trees of Streams
785------------------------------------------------------------------------------
787-- | Like 'iterateM' but iterates after mapping a stream generator on the
790-- Yield an input element in the output stream, map a stream generator on it
791-- and then do the same on the resulting stream. This can be used for a depth
792-- first traversal of a tree like structure.
794-- Note that 'iterateM' is a special case of 'iterateMapWith':
797-- iterateM f = iterateMapWith serial (yieldM . f) . yieldM
800-- It can be used to traverse a tree structure. For example, to list a
804-- Stream.iterateMapWith Stream.serial
805-- (either Dir.toEither (const nil))
806-- (yield (Left "tmp"))
811{-# INLINE iterateMapWith #-}
814 => (t m a -> t m a -> t m a)
818iterateMapWith combine f = concatMapWith combine go
820 go x = yield x `combine` concatMapWith combine go (f x)
823{-# INLINE iterateUnfold #-}
824iterateUnfold :: (IsStream t, MonadAsync m)
825 => Unfold m a a -> t m a -> t m a
826iterateUnfold unf xs = undefined
829------------------------------------------------------------------------------
831------------------------------------------------------------------------------
833-- To traverse graphs we need a state to be carried around in the traversal.
834-- For example, we can use a hashmap to store the visited status of nodes.
836-- | Like 'iterateMap' but carries a state in the stream generation function.
837-- This can be used to traverse graph like structures, we can remember the
838-- visited nodes in the state to avoid cycles.
840-- Note that a combination of 'iterateMap' and 'usingState' can also be used to
841-- traverse graphs. However, this function provides a more localized state
842-- instead of using a global state.
848{-# INLINE iterateSmapMWith #-}
850 :: (IsStream t, Monad m)
851 => (t m a -> t m a -> t m a)
852 -> (b -> a -> m (b, t m a))
856iterateSmapMWith combine f initial stream =
857 concatMap (\b -> concatMapWith combine (go b) stream) (yieldM initial)
861 go b a = yield a `combine` feedback b a
865 (\(b1, s) -> concatMapWith combine (go b1) s)
868------------------------------------------------------------------------------
869-- iterateMap - Either streams
870------------------------------------------------------------------------------
872-- | In an 'Either' stream iterate on 'Left's. This is a special case of
876-- iterateMapLeftsWith combine f = iterateMapWith combine (either f (const nil))
879-- To traverse a directory tree:
882-- iterateMapLeftsWith serial Dir.toEither (yield (Left "tmp"))
887{-# INLINE iterateMapLeftsWith #-}
890 => (t m (Either a b) -> t m (Either a b) -> t m (Either a b))
891 -> (a -> t m (Either a b))
894iterateMapLeftsWith combine f = iterateMapWith combine (either f (const K.nil))
896------------------------------------------------------------------------------
898------------------------------------------------------------------------------
900-- Splitting operations that take a predicate and a Fold can be
901-- expressed using parseMany. Operations like chunksOf, intervalsOf, split*,
902-- can be expressed using parseMany when used with an appropriate Parser.
904-- XXX We need takeGE/takeBetween to implement "some" using "many".
906-- | Apply a 'Fold' repeatedly on a stream and emit the parsed values in the
909-- This is the streaming dual of the 'Streamly.Internal.Data.Fold.many'
912-- >>> f = Fold.takeLE 2 Fold.sum
913-- >>> Stream.toList $ Stream.foldMany f $ Stream.fromList [1..10]
916-- >>> f = Fold.sliceEndWith Fold.toList
917-- >>> Stream.toList $ Stream.foldMany f $ Stream.fromList "hello\nworld"
918-- > ["hello\n","world"]
922{-# INLINE foldMany #-}
924 :: (IsStream t, Monad m)
928foldMany f m = D.fromStreamD $ D.foldMany f (D.toStreamD m)
930-- | Apply a stream of folds to an input stream and emit the results in the
935{-# INLINE foldSequence #-}
937 :: -- (IsStream t, Monad m) =>
941foldSequence _f _m = undefined
943-- | Iterate a fold generator on a stream. The initial value @b@ is used to
944-- generate the first fold, the fold is applied on the stream and the result of
945-- the fold is used to generate the next fold and so on.
947-- >>> f x = Fold.takeLE 2 (Fold.mconcatTo x)
948-- >>> s = Stream.map Sum $ Stream.fromList [1..10]
949-- >>> Stream.toList $ Stream.map getSum $ Stream.foldIterate f 0 s
950-- > [3,10,21,36,55,55]
952-- This is the streaming equivalent of monad like sequenced application of
953-- folds where next fold is dependent on the previous fold.
957{-# INLINE foldIterate #-}
959 :: -- (IsStream t, Monad m) =>
964foldIterate _f _i _m = undefined
965-- D.fromStreamD $ D.foldIterate f i (D.toStreamD m)
967------------------------------------------------------------------------------
969------------------------------------------------------------------------------
971-- | Apply a 'Parser' repeatedly on a stream and emit the parsed values in the
974-- This is the streaming equivalent of the 'Streamly.Internal.Data.Parser.many'
977-- >>> S.toList $ S.parseMany (PR.take 2 $ PR.fromFold FL.sum) $ S.fromList [1..10]
980-- >>> S.toList $ S.parseMany (PR.line FL.toList) $ S.fromList "hello\nworld"
981-- > ["hello\n","world"]
985{-# INLINE parseMany #-}
987 :: (IsStream t, MonadThrow m)
992 D.fromStreamD $ D.parseMany (PRK.fromParserK p) (D.toStreamD m)
994{-# INLINE parseManyD #-}
996 :: (IsStream t, MonadThrow m)
1001 D.fromStreamD $ D.parseMany p (D.toStreamD m)
1003-- | Apply a stream of parsers to an input stream and emit the results in the
1008{-# INLINE parseSequence #-}
1010 :: -- (IsStream t, Monad m) =>
1014parseSequence _f _m = undefined
1016-- | @parseManyTill collect test stream@ tries the parser @test@ on the input,
1017-- if @test@ fails it backtracks and tries @collect@, after @collect@ succeeds
1018-- @test@ is tried again and so on. The parser stops when @test@ succeeds. The
1019-- output of @test@ is discarded and the output of @collect@ is emitted in the
1020-- output stream. The parser fails if @collect@ fails.
1024{-# INLINE parseManyTill #-}
1026 -- (IsStream t, MonadThrow m) =>
1031parseManyTill = undefined
1033-- | Iterate a parser generating function on a stream. The initial value @b@ is
1034-- used to generate the first parser, the parser is applied on the stream and
1035-- the result is used to generate the next parser and so on.
1037-- >>> S.toList $ S.map getSum $ S.parseIterate (\b -> PR.take 2 (FL.mconcatTo b)) 0 $ S.map Sum $ S.fromList [1..10]
1038-- > [3,10,21,36,55,55]
1040-- This is the streaming equivalent of monad like sequenced application of
1041-- parsers where next parser is dependent on the previous parser.
1045{-# INLINE parseIterate #-}
1047 :: (IsStream t, MonadThrow m)
1048 => (b -> Parser m a b)
1052parseIterate f i m = D.fromStreamD $
1053 D.parseIterate (PRK.fromParserK . f) i (D.toStreamD m)
1055------------------------------------------------------------------------------
1057------------------------------------------------------------------------------
1059-- | Drop prefix from the input stream if present.
1063-- /Unimplemented/ - Help wanted.
1064{-# INLINE dropPrefix #-}
1066 -- (Eq a, IsStream t, Monad m) =>
1067 t m a -> t m a -> t m a
1068dropPrefix = error "Not implemented yet!"
1070-- | Drop all matching infix from the input stream if present. Infix stream
1071-- may be consumed multiple times.
1073-- Space: @O(n)@ where n is the length of the infix.
1075-- /Unimplemented/ - Help wanted.
1076{-# INLINE dropInfix #-}
1078 -- (Eq a, IsStream t, Monad m) =>
1079 t m a -> t m a -> t m a
1080dropInfix = error "Not implemented yet!"
1082-- | Drop suffix from the input stream if present. Suffix stream may be
1083-- consumed multiple times.
1085-- Space: @O(n)@ where n is the length of the suffix.
1087-- /Unimplemented/ - Help wanted.
1088{-# INLINE dropSuffix #-}
1090 -- (Eq a, IsStream t, Monad m) =>
1091 t m a -> t m a -> t m a
1092dropSuffix = error "Not implemented yet!"
1094------------------------------------------------------------------------------
1095-- Generalized grouping
1096------------------------------------------------------------------------------
1098-- This combinator is the most general grouping combinator and can be used to
1099-- implement all other grouping combinators.
1101-- XXX check if this can implement the splitOn combinator i.e. we can slide in
1102-- new elements, slide out old elements and incrementally compute the hash.
1103-- Also, can we implement the windowed classification combinators using this?
1105-- In fact this is a parse. Instead of using a special return value in the fold
1106-- we are using a mapping function.
1108-- Note that 'scanl'' (usually followed by a map to extract the desired value
1109-- from the accumulator) can be used to realize many implementations e.g. a
1110-- sliding window implementation. A scan followed by a mapMaybe is also a good
1111-- pattern to express many problems where we want to emit a filtered output and
1112-- not emit an output on every input.
1114-- Passing on of the initial accumulator value to the next fold is equivalent
1115-- to returning the leftover concept.
1118-- | @groupScan splitter fold stream@ folds the input stream using @fold@.
1119-- @splitter@ is applied on the accumulator of the fold every time an item is
1120-- consumed by the fold. The fold continues until @splitter@ returns a 'Just'
1121-- value. A 'Just' result from the @splitter@ specifies a result to be emitted
1122-- in the output stream and the initial value of the accumulator for the next
1123-- group's fold. This allows us to control whether to start fresh for the next
1124-- fold or to continue from the previous fold's output.
1126{-# INLINE groupScan #-}
1128 :: (IsStream t, Monad m)
1129 => (x -> m (Maybe (b, x))) -> Fold m a x -> t m a -> t m b
1130groupScan split fold m = undefined
1133------------------------------------------------------------------------------
1135------------------------------------------------------------------------------
1137-- | @groupsBy cmp f $ S.fromList [a,b,c,...]@ assigns the element @a@ to the
1138-- first group, if @b \`cmp` a@ is 'True' then @b@ is also assigned to the same
1139-- group. If @c \`cmp` a@ is 'True' then @c@ is also assigned to the same
1140-- group and so on. When the comparison fails a new group is started. Each
1141-- group is folded using the fold @f@ and the result of the fold is emitted in
1142-- the output stream.
1144-- >>> S.toList $ S.groupsBy (>) FL.toList $ S.fromList [1,3,7,0,2,5]
1145-- > [[1,3,7],[0,2,5]]
1148{-# INLINE groupsBy #-}
1150 :: (IsStream t, Monad m)
1155groupsBy cmp f m = D.fromStreamD $ D.groupsBy cmp f (D.toStreamD m)
1157-- | Unlike @groupsBy@ this function performs a rolling comparison of two
1158-- successive elements in the input stream. @groupsByRolling cmp f $ S.fromList
1159-- [a,b,c,...]@ assigns the element @a@ to the first group, if @a \`cmp` b@ is
1160-- 'True' then @b@ is also assigned to the same group. If @b \`cmp` c@ is
1161-- 'True' then @c@ is also assigned to the same group and so on. When the
1162-- comparison fails a new group is started. Each group is folded using the fold
1165-- >>> S.toList $ S.groupsByRolling (\a b -> a + 1 == b) FL.toList $ S.fromList [1,2,3,7,8,9]
1166-- > [[1,2,3],[7,8,9]]
1169{-# INLINE groupsByRolling #-}
1171 :: (IsStream t, Monad m)
1176groupsByRolling cmp f m = D.fromStreamD $ D.groupsRollingBy cmp f (D.toStreamD m)
1179-- > groups = groupsBy (==)
1180-- > groups = groupsByRolling (==)
1182-- Groups contiguous spans of equal elements together in individual groups.
1184-- >>> S.toList $ S.groups FL.toList $ S.fromList [1,1,2,2]
1188{-# INLINE groups #-}
1189groups :: (IsStream t, Monad m, Eq a) => Fold m a b -> t m a -> t m b
1190groups = groupsBy (==)
1192------------------------------------------------------------------------------
1193-- Splitting - by a predicate
1194------------------------------------------------------------------------------
1196-- TODO: Use a Splitter configuration similar to the "split" package to make it
1197-- possible to express all splitting combinations. In general, we can have
1198-- infix/suffix/prefix/condensing of separators, dropping both leading/trailing
1199-- separators. We can have a single split operation taking the splitter config
1202-- | Split on an infixed separator element, dropping the separator. Splits the
1203-- stream on separator elements determined by the supplied predicate, separator
1204-- is considered as infixed between two segments, if one side of the separator
1205-- is missing then it is parsed as an empty stream. The supplied 'Fold' is
1206-- applied on the split segments. With '-' representing non-separator elements
1207-- and '.' as separator, 'splitOn' splits as follows:
1210-- "--.--" => "--" "--"
1215-- @splitOn (== x)@ is an inverse of @intercalate (S.yield x)@
1217-- Let's use the following definition for illustration:
1219-- > splitOn' p xs = S.toList $ S.splitOn p (FL.toList) (S.fromList xs)
1221-- >>> splitOn' (== '.') ""
1224-- >>> splitOn' (== '.') "."
1227-- >>> splitOn' (== '.') ".a"
1230-- >>> splitOn' (== '.') "a."
1233-- >>> splitOn' (== '.') "a.b"
1236-- >>> splitOn' (== '.') "a..b"
1241-- This can be considered as an n-fold version of 'breakOn' where we apply
1242-- 'breakOn' successively on the input stream, dropping the first element
1243-- of the second segment after each break.
1245{-# INLINE splitOn #-}
1247 :: (IsStream t, Monad m)
1248 => (a -> Bool) -> Fold m a b -> t m a -> t m b
1249splitOn predicate f = foldMany (FL.sliceSepBy predicate f)
1251-- | Like 'splitOn' but the separator is considered as suffixed to the segments
1252-- in the stream. A missing suffix at the end is allowed. A separator at the
1253-- beginning is parsed as empty segment. With '-' representing elements and
1254-- '.' as separator, 'splitOnSuffix' splits as follows:
1257-- "--.--." => "--" "--"
1258-- "--.--" => "--" "--"
1262-- > splitOnSuffix' p xs = S.toList $ S.splitSuffixBy p (FL.toList) (S.fromList xs)
1264-- >>> splitOnSuffix' (== '.') ""
1267-- >>> splitOnSuffix' (== '.') "."
1270-- >>> splitOnSuffix' (== '.') "a"
1273-- >>> splitOnSuffix' (== '.') ".a"
1276-- >>> splitOnSuffix' (== '.') "a."
1279-- >>> splitOnSuffix' (== '.') "a.b"
1282-- >>> splitOnSuffix' (== '.') "a.b."
1285-- >>> splitOnSuffix' (== '.') "a..b.."
1288-- > lines = splitOnSuffix (== '\n')
1292-- This can be considered as an n-fold version of 'breakPost' where we apply
1293-- 'breakPost' successively on the input stream, dropping the first element
1294-- of the second segment after each break.
1296{-# INLINE splitOnSuffix #-}
1298 :: (IsStream t, Monad m)
1299 => (a -> Bool) -> Fold m a b -> t m a -> t m b
1300splitOnSuffix predicate f m =
1301 D.fromStreamD $ D.foldMany1 (FL.sliceSepBy predicate f) (D.toStreamD m)
1303-- | Like 'splitOn' after stripping leading, trailing, and repeated separators.
1304-- Therefore, @".a..b."@ with '.' as the separator would be parsed as
1305-- @["a","b"]@. In other words, its like parsing words from whitespace
1308-- > wordsBy' p xs = S.toList $ S.wordsBy p (FL.toList) (S.fromList xs)
1310-- >>> wordsBy' (== ',') ""
1313-- >>> wordsBy' (== ',') ","
1316-- >>> wordsBy' (== ',') ",a,,b,"
1319-- > words = wordsBy isSpace
1323-- It is equivalent to splitting in any of the infix/prefix/suffix styles
1324-- followed by removal of empty segments.
1325{-# INLINE wordsBy #-}
1327 :: (IsStream t, Monad m)
1328 => (a -> Bool) -> Fold m a b -> t m a -> t m b
1329wordsBy predicate f m =
1330 D.fromStreamD $ D.wordsBy predicate f (D.toStreamD m)
1332-- | Like 'splitOnSuffix' but keeps the suffix attached to the resulting
1335-- > splitWithSuffix' p xs = S.toList $ S.splitWithSuffix p (FL.toList) (S.fromList xs)
1337-- >>> splitWithSuffix' (== '.') ""
1340-- >>> splitWithSuffix' (== '.') "."
1343-- >>> splitWithSuffix' (== '.') "a"
1346-- >>> splitWithSuffix' (== '.') ".a"
1349-- >>> splitWithSuffix' (== '.') "a."
1352-- >>> splitWithSuffix' (== '.') "a.b"
1355-- >>> splitWithSuffix' (== '.') "a.b."
1358-- >>> splitWithSuffix' (== '.') "a..b.."
1359-- > ["a.",".","b.","."]
1363-- This can be considered as an n-fold version of 'breakPost' where we apply
1364-- 'breakPost' successively on the input stream.
1366{-# INLINE splitWithSuffix #-}
1368 :: (IsStream t, Monad m)
1369 => (a -> Bool) -> Fold m a b -> t m a -> t m b
1370splitWithSuffix predicate f m =
1371 D.fromStreamD $ D.foldMany1 (FL.sliceEndWith predicate f) (D.toStreamD m)
1373------------------------------------------------------------------------------
1374-- Splitting - on a delimiter sequence
1375------------------------------------------------------------------------------
1377-- Int list examples for splitOn:
1379-- >>> splitList [] [1,2,3,3,4]
1380-- > [[1],[2],[3],[3],[4]]
1382-- >>> splitList [5] [1,2,3,3,4]
1385-- >>> splitList [1] [1,2,3,3,4]
1388-- >>> splitList [4] [1,2,3,3,4]
1391-- >>> splitList [2] [1,2,3,3,4]
1394-- >>> splitList [3] [1,2,3,3,4]
1397-- >>> splitList [3,3] [1,2,3,3,4]
1400-- >>> splitList [1,2,3,3,4] [1,2,3,3,4]
1404-- This can be implemented easily using Rabin Karp
1405-- | Split on any one of the given patterns.
1406{-# INLINE splitOnAny #-}
1408 :: (IsStream t, Monad m, Storable a, Integral a)
1409 => [Array a] -> Fold m a b -> t m a -> t m b
1410splitOnAny subseq f m = undefined -- D.fromStreamD $ D.splitOnAny f subseq (D.toStreamD m)
1413-- XXX use a non-monadic intersperse to remove the MonadAsync constraint.
1415-- | Like 'splitOnSeq' but splits the separator as well, as an infix token.
1417-- > splitOn'_ pat xs = S.toList $ S.splitOn' (A.fromList pat) (FL.toList) (S.fromList xs)
1419-- >>> splitOn'_ "" "hello"
1420-- > ["h","","e","","l","","l","","o"]
1422-- >>> splitOn'_ "hello" ""
1425-- >>> splitOn'_ "hello" "hello"
1428-- >>> splitOn'_ "x" "hello"
1431-- >>> splitOn'_ "h" "hello"
1434-- >>> splitOn'_ "o" "hello"
1437-- >>> splitOn'_ "e" "hello"
1440-- >>> splitOn'_ "l" "hello"
1441-- > ["he","l","","l","o"]
1443-- >>> splitOn'_ "ll" "hello"
1447{-# INLINE splitBySeq #-}
1449 :: (IsStream t, MonadAsync m, Storable a, Enum a, Eq a)
1450 => Array a -> Fold m a b -> t m a -> t m b
1451splitBySeq patt f m =
1452 intersperseM (fold f (A.toStream patt)) $ splitOnSeq patt f m
1454-- | Like 'splitSuffixBy' but the separator is a sequence of elements, instead
1455-- of a predicate for a single element.
1457-- > splitOnSuffixSeq_ pat xs = S.toList $ S.splitOnSuffixSeq (A.fromList pat) (FL.toList) (S.fromList xs)
1459-- >>> splitOnSuffixSeq_ "." ""
1462-- >>> splitOnSuffixSeq_ "." "."
1465-- >>> splitOnSuffixSeq_ "." "a"
1468-- >>> splitOnSuffixSeq_ "." ".a"
1471-- >>> splitOnSuffixSeq_ "." "a."
1474-- >>> splitOnSuffixSeq_ "." "a.b"
1477-- >>> splitOnSuffixSeq_ "." "a.b."
1480-- >>> splitOnSuffixSeq_ "." "a..b.."
1483-- > lines = splitOnSuffixSeq "\n"
1485-- 'splitOnSuffixSeq' is an inverse of 'intercalateSuffix'. The following law
1488-- > intercalateSuffix . splitOnSuffixSeq == id
1490-- The following law holds when the separator is non-empty and contains none of
1491-- the elements present in the input lists:
1493-- > splitSuffixOn . intercalateSuffix == id
1496{-# INLINE splitOnSuffixSeq #-}
1498 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
1499 => Array a -> Fold m a b -> t m a -> t m b
1500splitOnSuffixSeq patt f m =
1501 D.fromStreamD $ D.splitOnSuffixSeq False patt f (D.toStreamD m)
1504-- | Like 'splitOn' but drops any empty splits.
1506{-# INLINE wordsOn #-}
1508 :: (IsStream t, Monad m, Storable a, Eq a)
1509 => Array a -> Fold m a b -> t m a -> t m b
1510wordsOn subseq f m = undefined -- D.fromStreamD $ D.wordsOn f subseq (D.toStreamD m)
1513-- | Like 'splitOnSuffixSeq' but keeps the suffix intact in the splits.
1515-- > splitWithSuffixSeq'_ pat xs = S.toList $ S.splitWithSuffixSeq (A.fromList pat) (FL.toList) (S.fromList xs)
1517-- >>> splitWithSuffixSeq' "." ""
1520-- >>> splitWithSuffixSeq' "." "."
1523-- >>> splitWithSuffixSeq' "." "a"
1526-- >>> splitWithSuffixSeq' "." ".a"
1529-- >>> splitWithSuffixSeq' "." "a."
1532-- >>> splitWithSuffixSeq' "." "a.b"
1535-- >>> splitWithSuffixSeq' "." "a.b."
1538-- >>> splitWithSuffixSeq' "." "a..b.."
1539-- > ["a.",".","b.","."]
1542{-# INLINE splitWithSuffixSeq #-}
1544 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
1545 => Array a -> Fold m a b -> t m a -> t m b
1546splitWithSuffixSeq patt f m =
1547 D.fromStreamD $ D.splitOnSuffixSeq True patt f (D.toStreamD m)
1550-- This can be implemented easily using Rabin Karp
1551-- | Split post any one of the given patterns.
1552{-# INLINE splitOnSuffixSeqAny #-}
1554 :: (IsStream t, Monad m, Storable a, Integral a)
1555 => [Array a] -> Fold m a b -> t m a -> t m b
1556splitOnSuffixSeqAny subseq f m = undefined
1557 -- D.fromStreamD $ D.splitPostAny f subseq (D.toStreamD m)
1560------------------------------------------------------------------------------
1562------------------------------------------------------------------------------
1564-- | Group the input stream into groups of @n@ elements each and then fold each
1565-- group using the provided fold function.
1567-- >> S.toList $ S.chunksOf 2 FL.sum (S.enumerateFromTo 1 10)
1570-- This can be considered as an n-fold version of 'takeLE' where we apply
1571-- 'takeLE' repeatedly on the leftover stream until the stream exhausts.
1574{-# INLINE chunksOf #-}
1576 :: (IsStream t, Monad m)
1577 => Int -> Fold m a b -> t m a -> t m b
1578chunksOf n f = foldMany (FL.takeLE n f)
1583{-# INLINE chunksOf2 #-}
1585 :: (IsStream t, Monad m)
1586 => Int -> m c -> Fold2 m c a b -> t m a -> t m b
1587chunksOf2 n action f m = D.fromStreamD $ D.groupsOf2 n action f (D.toStreamD m)
1589-- | @arraysOf n stream@ groups the elements in the input stream into arrays of
1590-- @n@ elements each.
1592-- Same as the following but may be more efficient:
1594-- > arraysOf n = S.chunksOf n (A.writeN n)
1597{-# INLINE arraysOf #-}
1598arraysOf :: (IsStream t, MonadIO m, Storable a)
1599 => Int -> t m a -> t m (Array a)
1600arraysOf n = chunksOf n (writeNUnsafe n)
1602-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
1603-- XXX add this example after fixing the serial stream rate control
1604-- >>> S.toList $ S.take 5 $ intervalsOf 1 FL.sum $ constRate 2 $ S.enumerateFrom 1
1607-- | Group the input stream into windows of @n@ second each and then fold each
1608-- group using the provided fold function.
1611{-# INLINE intervalsOf #-}
1613 :: (IsStream t, MonadAsync m)
1614 => Double -> Fold m a b -> t m a -> t m b
1616 splitWithSuffix isNothing (FL.lcatMaybes f)
1617 (interjectSuffix n (return Nothing) (Serial.map Just xs))
1619------------------------------------------------------------------------------
1620-- Windowed classification
1621------------------------------------------------------------------------------
1623-- We divide the stream into windows or chunks in space or time and each window
1624-- can be associated with a key, all events associated with a particular key in
1625-- the window can be folded to a single result. The stream can be split into
1626-- windows by size or by using a split predicate on the elements in the stream.
1627-- For example, when we receive a closing flag, we can close the window.
1629-- A "chunk" is a space window and a "session" is a time window. Are there any
1630-- other better short words to describe them. An alternative is to use
1631-- "swindow" and "twindow". Another word for "session" could be "spell".
1633-- TODO: To mark the position in space or time we can have Indexed or
1634-- TimeStamped types. That can make it easy to deal with the position indices
1637------------------------------------------------------------------------------
1638-- Keyed Sliding Windows
1639------------------------------------------------------------------------------
1642{-# INLINABLE classifySlidingChunks #-}
1643classifySlidingChunks
1644 :: (IsStream t, MonadAsync m, Ord k)
1645 => Int -- ^ window size
1646 -> Int -- ^ window slide
1647 -> Fold m a b -- ^ Fold to be applied to window events
1648 -> t m (k, a, Bool) -- ^ window key, data, close event
1650classifySlidingChunks wsize wslide (Fold step initial extract) str
1653-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
1654-- slide the send window when an ack is received and we slide the receive
1655-- window when a sequence is complete. Sliding is stateful in case of TCP,
1656-- sliding releases the send buffer or makes data available to the user from
1657-- the receive buffer.
1658{-# INLINABLE classifySlidingSessions #-}
1659classifySlidingSessions
1660 :: (IsStream t, MonadAsync m, Ord k)
1661 => Double -- ^ timer tick in seconds
1662 -> Double -- ^ time window size
1663 -> Double -- ^ window slide
1664 -> Fold m a b -- ^ Fold to be applied to window events
1665 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
1667classifySlidingSessions tick interval slide (Fold step initial extract) str
1671------------------------------------------------------------------------------
1672-- Sliding Window Buffers
1673------------------------------------------------------------------------------
1675-- These buffered versions could be faster than concurrent incremental folds of
1676-- all overlapping windows as in many cases we may not need all the values to
1677-- compute the fold, we can just compute the result using the old value and new
1678-- value. However, we may need the buffer once in a while, for example for
1679-- string search we usually compute the hash incrementally but when the hash
1680-- matches the hash of the pattern we need to compare the whole string.
1682-- XXX we should be able to implement sequence based splitting combinators
1683-- using this combinator.
1686-- | Buffer n elements of the input in a ring buffer. When t new elements are
1687-- collected, slide the window to remove the same number of oldest elements,
1688-- insert the new elements, and apply an incremental fold on the sliding
1689-- window, supplying the outgoing elements, the new ring buffer as arguments.
1691 :: (IsStream t, Monad m, Ord a, Storable a)
1692 => Int -- window size
1693 -> Int -- window slide
1694 -> Fold m (Ring a, Array a) b
1697slidingChunkBuffer = undefined
1699-- Buffer n seconds worth of stream elements of the input in a radix tree.
1700-- Every t seconds, remove the items that are older than n seconds, and apply
1701-- an incremental fold on the sliding window, supplying the outgoing elements,
1702-- and the new radix tree buffer as arguments.
1704 :: (IsStream t, Monad m, Ord a, Storable a)
1705 => Int -- window size
1707 -> Fold m (RTree a, Array a) b
1710slidingSessionBuffer = undefined
1713------------------------------------------------------------------------------
1714-- Keyed Session Windows
1715------------------------------------------------------------------------------
1718-- | Keyed variable size space windows. Close the window if we do not receive a
1719-- window event in the next "spaceout" elements.
1720{-# INLINABLE classifyChunksBy #-}
1722 :: (IsStream t, MonadAsync m, Ord k)
1723 => Int -- ^ window spaceout (spread)
1724 -> Bool -- ^ reset the spaceout when a chunk window element is received
1725 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1726 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1728classifyChunksBy spanout reset (Fold step initial extract) str = undefined
1730-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
1731-- received within the chunk size window. The chunk gets closed only if no
1732-- element is received within the chunk window.
1734{-# INLINABLE classifyKeepAliveChunks #-}
1735classifyKeepAliveChunks
1736 :: (IsStream t, MonadAsync m, Ord k)
1737 => Int -- ^ window spaceout (spread)
1738 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1739 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1741classifyKeepAliveChunks spanout = classifyChunksBy spanout True
1744data SessionState t m k a b = SessionState
1745 { sessionCurTime :: !AbsTime -- ^ time since last event
1746 , sessionEventTime :: !AbsTime -- ^ time as per last event
1747 , sessionCount :: !Int -- ^ total number sessions in progress
1748 , sessionTimerHeap :: H.Heap (H.Entry AbsTime k) -- ^ heap for timeouts
1749 , sessionKeyValueMap :: Map.Map k a -- ^ Stored sessions for keys
1750 , sessionOutputStream :: t (m :: Type -> Type) (k, b) -- ^ Completed sessions
1753-- XXX Perhaps we should use an "Event a" type to represent timestamped data.
1754-- | @classifySessionsBy tick timeout idle pred f stream@ groups timestamped
1755-- events in an input event stream into sessions based on a session key. Each
1756-- element in the input stream is an event consisting of a triple @(session key,
1757-- sesssion data, timestamp)@. @session key@ is a key that uniquely identifies
1758-- the session. All the events belonging to a session are folded using the fold
1759-- @f@ until the fold terminates or a timeout has occurred. The session key and
1760-- the result of the fold are emitted in the output stream when the session is
1763-- When @idle@ is 'False', @timeout@ is the maximum lifetime of a session in
1764-- seconds, measured from the @timestamp@ of the first event in that session.
1765-- When @idle@ is 'True' then the timeout is an idle timeout, it is reset after
1766-- every event received in the session.
1768-- @timestamp@ in an event characterizes the time when the input event was
1769-- generated, this is an absolute time measured from some @Epoch@. The notion
1770-- of current time is maintained by a monotonic event time clock using the
1771-- timestamps seen in the input stream. The latest timestamp seen till now is
1772-- used as the base for the current time. When no new events are seen, a timer
1773-- is started with a tick duration specified by @tick@. This timer is used to
1774-- detect session timeouts in the absence of new events.
1776-- The predicate @pred@ is invoked with the current session count, if it
1777-- returns 'True' a session is ejected from the session cache before inserting
1778-- a new session. This could be useful to alert or eject sessions when the
1779-- number of sessions becomes too high.
1783{-# INLINABLE classifySessionsBy #-}
1785 :: (IsStream t, MonadAsync m, Ord k)
1786 => Double -- ^ timer tick in seconds
1787 -> Double -- ^ session timeout in seconds
1788 -> Bool -- ^ reset the timeout when an event is received
1789 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
1790 -> Fold m a b -- ^ Fold to be applied to session events
1791 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
1792 -> t m (k, b) -- ^ session key, fold result
1793classifySessionsBy tick tmout reset ejectPred
1794 (Fold step initial extract) str =
1795 concatMap sessionOutputStream $
1796 scanlMAfter' sstep (return szero) flush stream
1800 timeoutMs = toRelTime (round (tmout * 1000) :: MilliSecond64)
1801 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
1802 szero = SessionState
1803 { sessionCurTime = toAbsTime (0 :: MilliSecond64)
1804 , sessionEventTime = toAbsTime (0 :: MilliSecond64)
1806 , sessionTimerHeap = H.empty
1807 , sessionKeyValueMap = Map.empty
1808 , sessionOutputStream = K.nil
1811 -- We can eject sessions based on the current session count to limit
1812 -- memory consumption. There are two possible strategies:
1814 -- 1) Eject old sessions or sessions beyond a certain/lower timeout
1815 -- threshold even before timeout, effectively reduce the timeout.
1816 -- 2) Drop creation of new sessions but keep accepting new events for the
1819 -- We use the first strategy as of now.
1821 -- Got a new stream input element
1822 sstep session@SessionState{..} (Just (key, value, timestamp)) = do
1823 -- XXX we should use a heap in pinned memory to scale it to a large
1826 -- To detect session inactivity we keep a timestamp of the latest event
1827 -- in the Map along with the fold result. When we purge the session
1828 -- from the heap we match the timestamp in the heap with the timestamp
1829 -- in the Map, if the latest timestamp is newer and has not expired we
1830 -- reinsert the key in the heap.
1832 -- XXX if the key is an Int, we can also use an IntMap for slightly
1833 -- better performance.
1835 let curTime = max sessionEventTime timestamp
1836 mOld = Map.lookup key sessionKeyValueMap
1838 -- deleting a key from the heap is expensive, so we never
1839 -- delete a key from heap, we just purge it from the Map and it
1840 -- gets purged from the heap on timeout. We just need an extra
1841 -- lookup in the Map when the key is purged from the heap, that
1842 -- should not be expensive.
1844 let (mp, cnt) = case mOld of
1845 Nothing -> (sessionKeyValueMap, sessionCount)
1846 Just _ -> (Map.delete key sessionKeyValueMap
1849 { sessionCurTime = curTime
1850 , sessionEventTime = curTime
1851 , sessionCount = cnt
1852 , sessionKeyValueMap = mp
1853 , sessionOutputStream = yield (key, fb)
1856 let acc = Tuple' timestamp fs1
1857 (hp1, mp1, out1, cnt1) <- do
1858 let vars = (sessionTimerHeap, sessionKeyValueMap,
1859 K.nil, sessionCount)
1861 -- inserting new entry
1863 -- Eject a session from heap and map is needed
1864 eject <- ejectPred sessionCount
1865 (hp, mp, out, cnt) <-
1870 -- Insert the new session in heap
1871 let expiry = addToAbsTime timestamp timeoutMs
1872 hp' = H.insert (Entry expiry key) hp
1873 in return (hp', mp, out, cnt + 1)
1874 -- updating old entry
1875 Just _ -> return vars
1877 let mp2 = Map.insert key acc mp1
1878 return $ SessionState
1879 { sessionCurTime = curTime
1880 , sessionEventTime = curTime
1881 , sessionCount = cnt1
1882 , sessionTimerHeap = hp1
1883 , sessionKeyValueMap = mp2
1884 , sessionOutputStream = out1
1889 Just (Tuple' _ acc) -> return $ FL.Partial acc
1891 FL.Done fb -> done fb
1893 res <- step fs value
1895 FL.Done fb -> done fb
1896 FL.Partial fs1 -> partial fs1
1898 -- Got a timer tick event
1899 sstep sessionState@SessionState{..} Nothing =
1900 let curTime = addToAbsTime sessionCurTime tickMs
1901 in ejectExpired sessionState curTime
1903 flush session@SessionState{..} = do
1904 (hp', mp', out, count) <-
1907 , sessionKeyValueMap
1912 { sessionCount = count
1913 , sessionTimerHeap = hp'
1914 , sessionKeyValueMap = mp'
1915 , sessionOutputStream = out
1918 -- delete from map and output the fold accumulator
1919 ejectEntry hp mp out cnt acc key = do
1921 let out1 = (key, sess) `K.cons` out
1922 let mp1 = Map.delete key mp
1923 return (hp, mp1, out1, cnt - 1)
1925 ejectAll (hp, mp, out, !cnt) = do
1926 let hres = H.uncons hp
1928 Just (Entry _ key, hp1) -> do
1929 r <- case Map.lookup key mp of
1930 Nothing -> return (hp1, mp, out, cnt)
1931 Just (Tuple' _ acc) -> ejectEntry hp1 mp out cnt acc key
1934 assert (Map.null mp) (return ())
1935 return (hp, mp, out, cnt)
1937 ejectOne (hp, mp, out, !cnt) = do
1938 let hres = H.uncons hp
1940 Just (Entry expiry key, hp1) ->
1941 case Map.lookup key mp of
1942 Nothing -> ejectOne (hp1, mp, out, cnt)
1943 Just (Tuple' latestTS acc) -> do
1944 let expiry1 = addToAbsTime latestTS timeoutMs
1945 if not reset || expiry1 <= expiry
1946 then ejectEntry hp1 mp out cnt acc key
1948 -- reset the session timeout and continue
1949 let hp2 = H.insert (Entry expiry1 key) hp1
1950 in ejectOne (hp2, mp, out, cnt)
1952 assert (Map.null mp) (return ())
1953 return (hp, mp, out, cnt)
1955 ejectExpired session@SessionState{..} curTime = do
1956 (hp', mp', out, count) <-
1957 ejectLoop sessionTimerHeap sessionKeyValueMap K.nil sessionCount
1959 { sessionCurTime = curTime
1960 , sessionCount = count
1961 , sessionTimerHeap = hp'
1962 , sessionKeyValueMap = mp'
1963 , sessionOutputStream = out
1968 ejectLoop hp mp out !cnt = do
1969 let hres = H.uncons hp
1971 Just (Entry expiry key, hp1) -> do
1973 if curTime >= expiry
1974 then return (True, False)
1980 case Map.lookup key mp of
1981 Nothing -> ejectLoop hp1 mp out cnt
1982 Just (Tuple' latestTS acc) -> do
1983 let expiry1 = addToAbsTime latestTS timeoutMs
1984 if expiry1 <= curTime || not reset || force
1986 (hp2,mp1,out1,cnt1) <-
1987 ejectEntry hp1 mp out cnt acc key
1988 ejectLoop hp2 mp1 out1 cnt1
1990 -- reset the session timeout and continue
1991 let hp2 = H.insert (Entry expiry1 key) hp1
1992 in ejectLoop hp2 mp out cnt
1993 else return (hp, mp, out, cnt)
1995 assert (Map.null mp) (return ())
1996 return (hp, mp, out, cnt)
1998 -- merge timer events in the stream
1999 stream = Serial.map Just str `Par.parallelFst` repeatM timer
2001 liftIO $ threadDelay (round $ tick * 1000000)
2004-- | Like 'classifySessionsOf' but the session is kept alive if an event is
2005-- received within the session window. The session times out and gets closed
2006-- only if no event is received within the specified session window size.
2008-- If the ejection predicate returns 'True', the session that was idle for
2009-- the longest time is ejected before inserting a new session.
2012-- classifyKeepAliveSessions timeout pred = classifySessionsBy 1 timeout True pred
2017{-# INLINABLE classifyKeepAliveSessions #-}
2018classifyKeepAliveSessions ::
2019 (IsStream t, MonadAsync m, Ord k)
2020 => Double -- ^ session inactive timeout
2021 -> (Int -> m Bool) -- ^ predicate to eject sessions on session count
2022 -> Fold m a b -- ^ Fold to be applied to session payload data
2023 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
2025classifyKeepAliveSessions tmout =
2026 classifySessionsBy 1 tmout True
2028------------------------------------------------------------------------------
2029-- Keyed tumbling windows
2030------------------------------------------------------------------------------
2032-- Tumbling windows is a special case of sliding windows where the window slide
2033-- is the same as the window size. Or it can be a special case of session
2034-- windows where the reset flag is set to False.
2036-- XXX instead of using the early termination flag in the stream, we can use an
2037-- early terminating fold instead.
2040-- | Split the stream into fixed size chunks of specified size. Within each
2041-- such chunk fold the elements in buckets identified by the keys. A particular
2042-- bucket fold can be terminated early if a closing flag is encountered in an
2043-- element for that key.
2046{-# INLINABLE classifyChunksOf #-}
2048 :: (IsStream t, MonadAsync m, Ord k)
2049 => Int -- ^ window size
2050 -> Fold m a b -- ^ Fold to be applied to window events
2051 -> t m (k, a, Bool) -- ^ window key, data, close event
2053classifyChunksOf wsize = classifyChunksBy wsize False
2056-- | Split the stream into fixed size time windows of specified interval in
2057-- seconds. Within each such window, fold the elements in sessions identified
2058-- by the session keys. The fold result is emitted in the output stream if the
2059-- fold returns a 'Left' result or if the time window ends.
2061-- Session @timestamp@ in the input stream is an absolute time from some epoch,
2062-- characterizing the time when the input element was generated. To detect
2063-- session window end, a monotonic event time clock is maintained synced with
2064-- the timestamps with a clock resolution of 1 second.
2066-- If the ejection predicate returns 'True', the session with the longest
2067-- lifetime is ejected before inserting a new session.
2070-- classifySessionsOf interval pred = classifySessionsBy 1 interval False pred
2075-- $ S.classifySessionsOf 3 (const (return False)) (fmap Right FL.toList)
2076-- $ S.map (\(ts,(k,a)) -> (k, a, ts))
2079-- $ (,) <$> S.fromList [1,2,3] <*> S.fromList [1,2,3]
2084{-# INLINABLE classifySessionsOf #-}
2085classifySessionsOf ::
2086 (IsStream t, MonadAsync m, Ord k)
2087 => Double -- ^ time window size
2088 -> (Int -> m Bool) -- ^ predicate to eject sessions on session count
2089 -> Fold m a b -- ^ Fold to be applied to session events
2090 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
2092classifySessionsOf interval =
2093 classifySessionsBy 1 interval False
2095------------------------------------------------------------------------------
2097------------------------------------------------------------------------------
2099-- | @splitInnerBy splitter joiner stream@ splits the inner containers @f a@ of
2100-- an input stream @t m (f a)@ using the @splitter@ function. Container
2101-- elements @f a@ are collected until a split occurs, then all the elements
2102-- before the split are joined using the @joiner@ function.
2104-- For example, if we have a stream of @Array Word8@, we may want to split the
2105-- stream into arrays representing lines separated by '\n' byte such that the
2106-- resulting stream after a split would be one array for each line.
2108-- CAUTION! This is not a true streaming function as the container size after
2109-- the split and merge may not be bounded.
2112{-# INLINE splitInnerBy #-}
2114 :: (IsStream t, Monad m)
2115 => (f a -> m (f a, Maybe (f a))) -- splitter
2116 -> (f a -> f a -> m (f a)) -- joiner
2119splitInnerBy splitter joiner xs =
2120 D.fromStreamD $ D.splitInnerBy splitter joiner $ D.toStreamD xs
2122-- | Like 'splitInnerBy' but splits assuming the separator joins the segment in
2126{-# INLINE splitInnerBySuffix #-}
2128 :: (IsStream t, Monad m, Eq (f a), Monoid (f a))
2129 => (f a -> m (f a, Maybe (f a))) -- splitter
2130 -> (f a -> f a -> m (f a)) -- joiner
2133splitInnerBySuffix splitter joiner xs =
2134 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