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.
135 -- | Apply parsers on a stream.
143 -- | Element unaware grouping.
150 -- In imperative terms, grouped folding can be considered as a nested loop
151 -- where we loop over the stream to group elements and then loop over
152 -- individual groups to fold them to a single value that is yielded in the
159 -- -- *** Searching Sequences
160 -- , seqIndices -- search a sequence in the stream
162 -- -- *** Searching Multiple Sequences
163 -- , seqIndicesAny -- search any of the given sequence in the stream
165 -- -- -- ** Searching Streams
166 -- -- | Finding a stream within another stream.
169 -- | Streams can be sliced into segments in space or in time. We use the
170 -- term @chunk@ to refer to a spatial length of the stream (spatial window)
171 -- and the term @session@ to refer to a length in time (time window).
173 -- -- *** Using Element Separators
181 , wordsBy -- stripAndCompactBy
183 -- -- *** Splitting By Sequences
187 -- , splitOnPrefixSeq
189 -- Keeping the delimiters
191 -- , splitByPrefixSeq
194 -- Splitting using multiple sequence separators
196 -- , splitOnAnySuffixSeq
197 -- , splitOnAnyPrefixSeq
199 -- -- *** Splitting By Streams
200 -- -- | Splitting a stream using another stream as separator.
202 -- ** Windowed Classification
204 -- | Split the stream into windows or chunks in space or time. Each window
205 -- can be associated with a key, all events associated with a particular
206 -- key in the window can be folded to a single result. The stream is split
207 -- into windows of specified size, the window can be terminated early if
208 -- the closing flag is specified in the input stream.
210 -- The term "chunk" is used for a space window and the term "session" is
211 -- 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.Types (Fold (..), Fold2 (..))
256import Streamly.Internal.Data.Parser (Parser (..))
257import Streamly.Internal.Data.Unfold.Types (Unfold)
258import Streamly.Internal.Data.Array.Foreign.Types (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.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.concatUnfold 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.concatUnfold 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.concatUnfold 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{-# INLINE foldMany1 #-}
932 :: (IsStream t, Monad m)
936foldMany1 f m = D.fromStreamD $ D.foldMany1 f (D.toStreamD m)
938-- | Apply a stream of folds to an input stream and emit the results in the
943{-# INLINE foldSequence #-}
945 :: -- (IsStream t, Monad m) =>
949foldSequence _f _m = undefined
951-- | Iterate a fold generator on a stream. The initial value @b@ is used to
952-- generate the first fold, the fold is applied on the stream and the result of
953-- the fold is used to generate the next fold and so on.
955-- >>> f x = Fold.takeLE 2 (Fold.mconcatTo x)
956-- >>> s = Stream.map Sum $ Stream.fromList [1..10]
957-- >>> Stream.toList $ Stream.map getSum $ Stream.foldIterate f 0 s
958-- > [3,10,21,36,55,55]
960-- This is the streaming equivalent of monad like sequenced application of
961-- folds where next fold is dependent on the previous fold.
965{-# INLINE foldIterate #-}
967 :: -- (IsStream t, Monad m) =>
972foldIterate _f _i _m = undefined
973-- D.fromStreamD $ D.foldIterate f i (D.toStreamD m)
975------------------------------------------------------------------------------
977------------------------------------------------------------------------------
979-- | Apply a 'Parser' repeatedly on a stream and emit the parsed values in the
982-- This is the streaming equivalent of the 'Streamly.Internal.Data.Parser.many'
985-- >>> S.toList $ S.parseMany (PR.take 2 $ PR.fromFold FL.sum) $ S.fromList [1..10]
988-- >>> S.toList $ S.parseMany (PR.line FL.toList) $ S.fromList "hello\nworld"
989-- > ["hello\n","world"]
993{-# INLINE parseMany #-}
995 :: (IsStream t, MonadThrow m)
1000 D.fromStreamD $ D.parseMany (PRK.fromParserK p) (D.toStreamD m)
1002{-# INLINE parseManyD #-}
1004 :: (IsStream t, MonadThrow m)
1009 D.fromStreamD $ D.parseMany p (D.toStreamD m)
1011-- | Apply a stream of parsers to an input stream and emit the results in the
1016{-# INLINE parseSequence #-}
1018 :: -- (IsStream t, Monad m) =>
1022parseSequence _f _m = undefined
1024-- | @parseManyTill collect test stream@ tries the parser @test@ on the input,
1025-- if @test@ fails it backtracks and tries @collect@, after @collect@ succeeds
1026-- @test@ is tried again and so on. The parser stops when @test@ succeeds. The
1027-- output of @test@ is discarded and the output of @collect@ is emitted in the
1028-- output stream. The parser fails if @collect@ fails.
1032{-# INLINE parseManyTill #-}
1034 -- (IsStream t, MonadThrow m) =>
1039parseManyTill = undefined
1041-- | Iterate a parser generating function on a stream. The initial value @b@ is
1042-- used to generate the first parser, the parser is applied on the stream and
1043-- the result is used to generate the next parser and so on.
1045-- >>> S.toList $ S.map getSum $ S.parseIterate (\b -> PR.take 2 (FL.mconcatTo b)) 0 $ S.map Sum $ S.fromList [1..10]
1046-- > [3,10,21,36,55,55]
1048-- This is the streaming equivalent of monad like sequenced application of
1049-- parsers where next parser is dependent on the previous parser.
1053{-# INLINE parseIterate #-}
1055 :: (IsStream t, MonadThrow m)
1056 => (b -> Parser m a b)
1060parseIterate f i m = D.fromStreamD $
1061 D.parseIterate (PRK.fromParserK . f) i (D.toStreamD m)
1063------------------------------------------------------------------------------
1065------------------------------------------------------------------------------
1067-- | Drop prefix from the input stream if present.
1071-- /Unimplemented/ - Help wanted.
1072{-# INLINE dropPrefix #-}
1074 -- (Eq a, IsStream t, Monad m) =>
1075 t m a -> t m a -> t m a
1076dropPrefix = error "Not implemented yet!"
1078-- | Drop all matching infix from the input stream if present. Infix stream
1079-- may be consumed multiple times.
1081-- Space: @O(n)@ where n is the length of the infix.
1083-- /Unimplemented/ - Help wanted.
1084{-# INLINE dropInfix #-}
1086 -- (Eq a, IsStream t, Monad m) =>
1087 t m a -> t m a -> t m a
1088dropInfix = error "Not implemented yet!"
1090-- | Drop suffix from the input stream if present. Suffix stream may be
1091-- consumed multiple times.
1093-- Space: @O(n)@ where n is the length of the suffix.
1095-- /Unimplemented/ - Help wanted.
1096{-# INLINE dropSuffix #-}
1098 -- (Eq a, IsStream t, Monad m) =>
1099 t m a -> t m a -> t m a
1100dropSuffix = error "Not implemented yet!"
1102------------------------------------------------------------------------------
1103-- Generalized grouping
1104------------------------------------------------------------------------------
1106-- This combinator is the most general grouping combinator and can be used to
1107-- implement all other grouping combinators.
1109-- XXX check if this can implement the splitOn combinator i.e. we can slide in
1110-- new elements, slide out old elements and incrementally compute the hash.
1111-- Also, can we implement the windowed classification combinators using this?
1113-- In fact this is a parse. Instead of using a special return value in the fold
1114-- we are using a mapping function.
1116-- Note that 'scanl'' (usually followed by a map to extract the desired value
1117-- from the accumulator) can be used to realize many implementations e.g. a
1118-- sliding window implementation. A scan followed by a mapMaybe is also a good
1119-- pattern to express many problems where we want to emit a filtered output and
1120-- not emit an output on every input.
1122-- Passing on of the initial accumulator value to the next fold is equivalent
1123-- to returning the leftover concept.
1126-- | @groupScan splitter fold stream@ folds the input stream using @fold@.
1127-- @splitter@ is applied on the accumulator of the fold every time an item is
1128-- consumed by the fold. The fold continues until @splitter@ returns a 'Just'
1129-- value. A 'Just' result from the @splitter@ specifies a result to be emitted
1130-- in the output stream and the initial value of the accumulator for the next
1131-- group's fold. This allows us to control whether to start fresh for the next
1132-- fold or to continue from the previous fold's output.
1134{-# INLINE groupScan #-}
1136 :: (IsStream t, Monad m)
1137 => (x -> m (Maybe (b, x))) -> Fold m a x -> t m a -> t m b
1138groupScan split fold m = undefined
1141------------------------------------------------------------------------------
1143------------------------------------------------------------------------------
1145-- | @groupsBy cmp f $ S.fromList [a,b,c,...]@ assigns the element @a@ to the
1146-- first group, if @b \`cmp` a@ is 'True' then @b@ is also assigned to the same
1147-- group. If @c \`cmp` a@ is 'True' then @c@ is also assigned to the same
1148-- group and so on. When the comparison fails a new group is started. Each
1149-- group is folded using the fold @f@ and the result of the fold is emitted in
1150-- the output stream.
1152-- >>> S.toList $ S.groupsBy (>) FL.toList $ S.fromList [1,3,7,0,2,5]
1153-- > [[1,3,7],[0,2,5]]
1156{-# INLINE groupsBy #-}
1158 :: (IsStream t, Monad m)
1163groupsBy cmp f m = D.fromStreamD $ D.groupsBy cmp f (D.toStreamD m)
1165-- | Unlike @groupsBy@ this function performs a rolling comparison of two
1166-- successive elements in the input stream. @groupsByRolling cmp f $ S.fromList
1167-- [a,b,c,...]@ assigns the element @a@ to the first group, if @a \`cmp` b@ is
1168-- 'True' then @b@ is also assigned to the same group. If @b \`cmp` c@ is
1169-- 'True' then @c@ is also assigned to the same group and so on. When the
1170-- comparison fails a new group is started. Each group is folded using the fold
1173-- >>> S.toList $ S.groupsByRolling (\a b -> a + 1 == b) FL.toList $ S.fromList [1,2,3,7,8,9]
1174-- > [[1,2,3],[7,8,9]]
1177{-# INLINE groupsByRolling #-}
1179 :: (IsStream t, Monad m)
1184groupsByRolling cmp f m = D.fromStreamD $ D.groupsRollingBy cmp f (D.toStreamD m)
1187-- > groups = groupsBy (==)
1188-- > groups = groupsByRolling (==)
1190-- Groups contiguous spans of equal elements together in individual groups.
1192-- >>> S.toList $ S.groups FL.toList $ S.fromList [1,1,2,2]
1196{-# INLINE groups #-}
1197groups :: (IsStream t, Monad m, Eq a) => Fold m a b -> t m a -> t m b
1198groups = groupsBy (==)
1200------------------------------------------------------------------------------
1201-- Splitting - by a predicate
1202------------------------------------------------------------------------------
1204-- TODO: Use a Splitter configuration similar to the "split" package to make it
1205-- possible to express all splitting combinations. In general, we can have
1206-- infix/suffix/prefix/condensing of separators, dropping both leading/trailing
1207-- separators. We can have a single split operation taking the splitter config
1210-- | Split on an infixed separator element, dropping the separator. Splits the
1211-- stream on separator elements determined by the supplied predicate, separator
1212-- is considered as infixed between two segments, if one side of the separator
1213-- is missing then it is parsed as an empty stream. The supplied 'Fold' is
1214-- applied on the split segments. With '-' representing non-separator elements
1215-- and '.' as separator, 'splitOn' splits as follows:
1218-- "--.--" => "--" "--"
1223-- @splitOn (== x)@ is an inverse of @intercalate (S.yield x)@
1225-- Let's use the following definition for illustration:
1227-- > splitOn' p xs = S.toList $ S.splitOn p (FL.toList) (S.fromList xs)
1229-- >>> splitOn' (== '.') ""
1232-- >>> splitOn' (== '.') "."
1235-- >>> splitOn' (== '.') ".a"
1238-- >>> splitOn' (== '.') "a."
1241-- >>> splitOn' (== '.') "a.b"
1244-- >>> splitOn' (== '.') "a..b"
1249-- This can be considered as an n-fold version of 'breakOn' where we apply
1250-- 'breakOn' successively on the input stream, dropping the first element
1251-- of the second segment after each break.
1253{-# INLINE splitOn #-}
1255 :: (IsStream t, Monad m)
1256 => (a -> Bool) -> Fold m a b -> t m a -> t m b
1257splitOn predicate f = foldMany (FL.sliceSepBy predicate f)
1259-- | Like 'splitOn' but the separator is considered as suffixed to the segments
1260-- in the stream. A missing suffix at the end is allowed. A separator at the
1261-- beginning is parsed as empty segment. With '-' representing elements and
1262-- '.' as separator, 'splitOnSuffix' splits as follows:
1265-- "--.--." => "--" "--"
1266-- "--.--" => "--" "--"
1270-- > splitOnSuffix' p xs = S.toList $ S.splitSuffixBy p (FL.toList) (S.fromList xs)
1272-- >>> splitOnSuffix' (== '.') ""
1275-- >>> splitOnSuffix' (== '.') "."
1278-- >>> splitOnSuffix' (== '.') "a"
1281-- >>> splitOnSuffix' (== '.') ".a"
1284-- >>> splitOnSuffix' (== '.') "a."
1287-- >>> splitOnSuffix' (== '.') "a.b"
1290-- >>> splitOnSuffix' (== '.') "a.b."
1293-- >>> splitOnSuffix' (== '.') "a..b.."
1296-- > lines = splitOnSuffix (== '\n')
1300-- This can be considered as an n-fold version of 'breakPost' where we apply
1301-- 'breakPost' successively on the input stream, dropping the first element
1302-- of the second segment after each break.
1304{-# INLINE splitOnSuffix #-}
1306 :: (IsStream t, Monad m)
1307 => (a -> Bool) -> Fold m a b -> t m a -> t m b
1308splitOnSuffix predicate f m =
1309 D.fromStreamD $ D.foldMany1 (FL.sliceSepBy predicate f) (D.toStreamD m)
1311-- | Like 'splitOn' after stripping leading, trailing, and repeated separators.
1312-- Therefore, @".a..b."@ with '.' as the separator would be parsed as
1313-- @["a","b"]@. In other words, its like parsing words from whitespace
1316-- > wordsBy' p xs = S.toList $ S.wordsBy p (FL.toList) (S.fromList xs)
1318-- >>> wordsBy' (== ',') ""
1321-- >>> wordsBy' (== ',') ","
1324-- >>> wordsBy' (== ',') ",a,,b,"
1327-- > words = wordsBy isSpace
1331-- It is equivalent to splitting in any of the infix/prefix/suffix styles
1332-- followed by removal of empty segments.
1333{-# INLINE wordsBy #-}
1335 :: (IsStream t, Monad m)
1336 => (a -> Bool) -> Fold m a b -> t m a -> t m b
1337wordsBy predicate f m =
1338 D.fromStreamD $ D.wordsBy predicate f (D.toStreamD m)
1340-- | Like 'splitOnSuffix' but keeps the suffix attached to the resulting
1343-- > splitWithSuffix' p xs = S.toList $ S.splitWithSuffix p (FL.toList) (S.fromList xs)
1345-- >>> splitWithSuffix' (== '.') ""
1348-- >>> splitWithSuffix' (== '.') "."
1351-- >>> splitWithSuffix' (== '.') "a"
1354-- >>> splitWithSuffix' (== '.') ".a"
1357-- >>> splitWithSuffix' (== '.') "a."
1360-- >>> splitWithSuffix' (== '.') "a.b"
1363-- >>> splitWithSuffix' (== '.') "a.b."
1366-- >>> splitWithSuffix' (== '.') "a..b.."
1367-- > ["a.",".","b.","."]
1371-- This can be considered as an n-fold version of 'breakPost' where we apply
1372-- 'breakPost' successively on the input stream.
1374{-# INLINE splitWithSuffix #-}
1376 :: (IsStream t, Monad m)
1377 => (a -> Bool) -> Fold m a b -> t m a -> t m b
1378splitWithSuffix predicate f m =
1379 D.fromStreamD $ D.foldMany1 (FL.sliceEndWith predicate f) (D.toStreamD m)
1381------------------------------------------------------------------------------
1382-- Splitting - on a delimiter sequence
1383------------------------------------------------------------------------------
1385-- Int list examples for splitOn:
1387-- >>> splitList [] [1,2,3,3,4]
1388-- > [[1],[2],[3],[3],[4]]
1390-- >>> splitList [5] [1,2,3,3,4]
1393-- >>> splitList [1] [1,2,3,3,4]
1396-- >>> splitList [4] [1,2,3,3,4]
1399-- >>> splitList [2] [1,2,3,3,4]
1402-- >>> splitList [3] [1,2,3,3,4]
1405-- >>> splitList [3,3] [1,2,3,3,4]
1408-- >>> splitList [1,2,3,3,4] [1,2,3,3,4]
1412-- This can be implemented easily using Rabin Karp
1413-- | Split on any one of the given patterns.
1414{-# INLINE splitOnAny #-}
1416 :: (IsStream t, Monad m, Storable a, Integral a)
1417 => [Array a] -> Fold m a b -> t m a -> t m b
1418splitOnAny subseq f m = undefined -- D.fromStreamD $ D.splitOnAny f subseq (D.toStreamD m)
1421-- XXX use a non-monadic intersperse to remove the MonadAsync constraint.
1423-- | Like 'splitOnSeq' but splits the separator as well, as an infix token.
1425-- > splitOn'_ pat xs = S.toList $ S.splitOn' (A.fromList pat) (FL.toList) (S.fromList xs)
1427-- >>> splitOn'_ "" "hello"
1428-- > ["h","","e","","l","","l","","o"]
1430-- >>> splitOn'_ "hello" ""
1433-- >>> splitOn'_ "hello" "hello"
1436-- >>> splitOn'_ "x" "hello"
1439-- >>> splitOn'_ "h" "hello"
1442-- >>> splitOn'_ "o" "hello"
1445-- >>> splitOn'_ "e" "hello"
1448-- >>> splitOn'_ "l" "hello"
1449-- > ["he","l","","l","o"]
1451-- >>> splitOn'_ "ll" "hello"
1455{-# INLINE splitBySeq #-}
1457 :: (IsStream t, MonadAsync m, Storable a, Enum a, Eq a)
1458 => Array a -> Fold m a b -> t m a -> t m b
1459splitBySeq patt f m =
1460 intersperseM (fold f (A.toStream patt)) $ splitOnSeq patt f m
1462-- | Like 'splitSuffixBy' but the separator is a sequence of elements, instead
1463-- of a predicate for a single element.
1465-- > splitOnSuffixSeq_ pat xs = S.toList $ S.splitOnSuffixSeq (A.fromList pat) (FL.toList) (S.fromList xs)
1467-- >>> splitOnSuffixSeq_ "." ""
1470-- >>> splitOnSuffixSeq_ "." "."
1473-- >>> splitOnSuffixSeq_ "." "a"
1476-- >>> splitOnSuffixSeq_ "." ".a"
1479-- >>> splitOnSuffixSeq_ "." "a."
1482-- >>> splitOnSuffixSeq_ "." "a.b"
1485-- >>> splitOnSuffixSeq_ "." "a.b."
1488-- >>> splitOnSuffixSeq_ "." "a..b.."
1491-- > lines = splitOnSuffixSeq "\n"
1493-- 'splitOnSuffixSeq' is an inverse of 'intercalateSuffix'. The following law
1496-- > intercalateSuffix . splitOnSuffixSeq == id
1498-- The following law holds when the separator is non-empty and contains none of
1499-- the elements present in the input lists:
1501-- > splitSuffixOn . intercalateSuffix == id
1504{-# INLINE splitOnSuffixSeq #-}
1506 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
1507 => Array a -> Fold m a b -> t m a -> t m b
1508splitOnSuffixSeq patt f m =
1509 D.fromStreamD $ D.splitOnSuffixSeq False patt f (D.toStreamD m)
1512-- | Like 'splitOn' but drops any empty splits.
1514{-# INLINE wordsOn #-}
1516 :: (IsStream t, Monad m, Storable a, Eq a)
1517 => Array a -> Fold m a b -> t m a -> t m b
1518wordsOn subseq f m = undefined -- D.fromStreamD $ D.wordsOn f subseq (D.toStreamD m)
1521-- | Like 'splitOnSuffixSeq' but keeps the suffix intact in the splits.
1523-- > splitWithSuffixSeq'_ pat xs = S.toList $ S.splitWithSuffixSeq (A.fromList pat) (FL.toList) (S.fromList xs)
1525-- >>> splitWithSuffixSeq' "." ""
1528-- >>> splitWithSuffixSeq' "." "."
1531-- >>> splitWithSuffixSeq' "." "a"
1534-- >>> splitWithSuffixSeq' "." ".a"
1537-- >>> splitWithSuffixSeq' "." "a."
1540-- >>> splitWithSuffixSeq' "." "a.b"
1543-- >>> splitWithSuffixSeq' "." "a.b."
1546-- >>> splitWithSuffixSeq' "." "a..b.."
1547-- > ["a.",".","b.","."]
1550{-# INLINE splitWithSuffixSeq #-}
1552 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
1553 => Array a -> Fold m a b -> t m a -> t m b
1554splitWithSuffixSeq patt f m =
1555 D.fromStreamD $ D.splitOnSuffixSeq True patt f (D.toStreamD m)
1558-- This can be implemented easily using Rabin Karp
1559-- | Split post any one of the given patterns.
1560{-# INLINE splitOnSuffixSeqAny #-}
1562 :: (IsStream t, Monad m, Storable a, Integral a)
1563 => [Array a] -> Fold m a b -> t m a -> t m b
1564splitOnSuffixSeqAny subseq f m = undefined
1565 -- D.fromStreamD $ D.splitPostAny f subseq (D.toStreamD m)
1568------------------------------------------------------------------------------
1570------------------------------------------------------------------------------
1572-- | Group the input stream into groups of @n@ elements each and then fold each
1573-- group using the provided fold function.
1575-- >> S.toList $ S.chunksOf 2 FL.sum (S.enumerateFromTo 1 10)
1578-- This can be considered as an n-fold version of 'takeLE' where we apply
1579-- 'takeLE' repeatedly on the leftover stream until the stream exhausts.
1581-- @chunksOf n f = foldMany (FL.takeLE n f)@
1584{-# INLINE chunksOf #-}
1586 :: (IsStream t, Monad m)
1587 => Int -> Fold m a b -> t m a -> t m b
1588chunksOf n f = D.fromStreamD . D.chunksOf n f . D.toStreamD
1593{-# INLINE chunksOf2 #-}
1595 :: (IsStream t, Monad m)
1596 => Int -> m c -> Fold2 m c a b -> t m a -> t m b
1597chunksOf2 n action f m = D.fromStreamD $ D.groupsOf2 n action f (D.toStreamD m)
1599-- | @arraysOf n stream@ groups the elements in the input stream into arrays of
1600-- @n@ elements each.
1602-- Same as the following but may be more efficient:
1604-- > arraysOf n = Stream.foldMany (A.writeN n)
1607{-# INLINE arraysOf #-}
1608arraysOf :: (IsStream t, MonadIO m, Storable a)
1609 => Int -> t m a -> t m (Array a)
1610arraysOf n = D.fromStreamD . A.arraysOf n . D.toStreamD
1612-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
1613-- XXX add this example after fixing the serial stream rate control
1614-- >>> S.toList $ S.take 5 $ intervalsOf 1 FL.sum $ constRate 2 $ S.enumerateFrom 1
1617-- | Group the input stream into windows of @n@ second each and then fold each
1618-- group using the provided fold function.
1621{-# INLINE intervalsOf #-}
1623 :: (IsStream t, MonadAsync m)
1624 => Double -> Fold m a b -> t m a -> t m b
1626 splitWithSuffix isNothing (FL.lcatMaybes f)
1627 (interjectSuffix n (return Nothing) (Serial.map Just xs))
1629------------------------------------------------------------------------------
1630-- Windowed classification
1631------------------------------------------------------------------------------
1633-- We divide the stream into windows or chunks in space or time and each window
1634-- can be associated with a key, all events associated with a particular key in
1635-- the window can be folded to a single result. The stream can be split into
1636-- windows by size or by using a split predicate on the elements in the stream.
1637-- For example, when we receive a closing flag, we can close the window.
1639-- A "chunk" is a space window and a "session" is a time window. Are there any
1640-- other better short words to describe them. An alternative is to use
1641-- "swindow" and "twindow". Another word for "session" could be "spell".
1643-- TODO: To mark the position in space or time we can have Indexed or
1644-- TimeStamped types. That can make it easy to deal with the position indices
1647------------------------------------------------------------------------------
1648-- Keyed Sliding Windows
1649------------------------------------------------------------------------------
1652{-# INLINABLE classifySlidingChunks #-}
1653classifySlidingChunks
1654 :: (IsStream t, MonadAsync m, Ord k)
1655 => Int -- ^ window size
1656 -> Int -- ^ window slide
1657 -> Fold m a b -- ^ Fold to be applied to window events
1658 -> t m (k, a, Bool) -- ^ window key, data, close event
1660classifySlidingChunks wsize wslide (Fold step initial extract) str
1663-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
1664-- slide the send window when an ack is received and we slide the receive
1665-- window when a sequence is complete. Sliding is stateful in case of TCP,
1666-- sliding releases the send buffer or makes data available to the user from
1667-- the receive buffer.
1668{-# INLINABLE classifySlidingSessions #-}
1669classifySlidingSessions
1670 :: (IsStream t, MonadAsync m, Ord k)
1671 => Double -- ^ timer tick in seconds
1672 -> Double -- ^ time window size
1673 -> Double -- ^ window slide
1674 -> Fold m a b -- ^ Fold to be applied to window events
1675 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
1677classifySlidingSessions tick interval slide (Fold step initial extract) str
1681------------------------------------------------------------------------------
1682-- Sliding Window Buffers
1683------------------------------------------------------------------------------
1685-- These buffered versions could be faster than concurrent incremental folds of
1686-- all overlapping windows as in many cases we may not need all the values to
1687-- compute the fold, we can just compute the result using the old value and new
1688-- value. However, we may need the buffer once in a while, for example for
1689-- string search we usually compute the hash incrementally but when the hash
1690-- matches the hash of the pattern we need to compare the whole string.
1692-- XXX we should be able to implement sequence based splitting combinators
1693-- using this combinator.
1696-- | Buffer n elements of the input in a ring buffer. When t new elements are
1697-- collected, slide the window to remove the same number of oldest elements,
1698-- insert the new elements, and apply an incremental fold on the sliding
1699-- window, supplying the outgoing elements, the new ring buffer as arguments.
1701 :: (IsStream t, Monad m, Ord a, Storable a)
1702 => Int -- window size
1703 -> Int -- window slide
1704 -> Fold m (Ring a, Array a) b
1707slidingChunkBuffer = undefined
1709-- Buffer n seconds worth of stream elements of the input in a radix tree.
1710-- Every t seconds, remove the items that are older than n seconds, and apply
1711-- an incremental fold on the sliding window, supplying the outgoing elements,
1712-- and the new radix tree buffer as arguments.
1714 :: (IsStream t, Monad m, Ord a, Storable a)
1715 => Int -- window size
1717 -> Fold m (RTree a, Array a) b
1720slidingSessionBuffer = undefined
1723------------------------------------------------------------------------------
1724-- Keyed Session Windows
1725------------------------------------------------------------------------------
1728-- | Keyed variable size space windows. Close the window if we do not receive a
1729-- window event in the next "spaceout" elements.
1730{-# INLINABLE classifyChunksBy #-}
1732 :: (IsStream t, MonadAsync m, Ord k)
1733 => Int -- ^ window spaceout (spread)
1734 -> Bool -- ^ reset the spaceout when a chunk window element is received
1735 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1736 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1738classifyChunksBy spanout reset (Fold step initial extract) str = undefined
1740-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
1741-- received within the chunk size window. The chunk gets closed only if no
1742-- element is received within the chunk window.
1744{-# INLINABLE classifyKeepAliveChunks #-}
1745classifyKeepAliveChunks
1746 :: (IsStream t, MonadAsync m, Ord k)
1747 => Int -- ^ window spaceout (spread)
1748 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1749 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1751classifyKeepAliveChunks spanout = classifyChunksBy spanout True
1754data SessionState t m k a b = SessionState
1755 { sessionCurTime :: !AbsTime -- ^ time since last event
1756 , sessionEventTime :: !AbsTime -- ^ time as per last event
1757 , sessionCount :: !Int -- ^ total number sessions in progress
1758 , sessionTimerHeap :: H.Heap (H.Entry AbsTime k) -- ^ heap for timeouts
1759 , sessionKeyValueMap :: Map.Map k a -- ^ Stored sessions for keys
1760 , sessionOutputStream :: t (m :: Type -> Type) (k, b) -- ^ Completed sessions
1763-- XXX Perhaps we should use an "Event a" type to represent timestamped data.
1764-- | @classifySessionsBy tick timeout idle pred f stream@ groups timestamped
1765-- events in an input event stream into sessions based on a session key. Each
1766-- element in the input stream is an event consisting of a triple @(session key,
1767-- sesssion data, timestamp)@. @session key@ is a key that uniquely identifies
1768-- the session. All the events belonging to a session are folded using the fold
1769-- @f@ until the fold terminates or a timeout has occurred. The session key and
1770-- the result of the fold are emitted in the output stream when the session is
1773-- When @idle@ is 'False', @timeout@ is the maximum lifetime of a session in
1774-- seconds, measured from the @timestamp@ of the first event in that session.
1775-- When @idle@ is 'True' then the timeout is an idle timeout, it is reset after
1776-- every event received in the session.
1778-- @timestamp@ in an event characterizes the time when the input event was
1779-- generated, this is an absolute time measured from some @Epoch@. The notion
1780-- of current time is maintained by a monotonic event time clock using the
1781-- timestamps seen in the input stream. The latest timestamp seen till now is
1782-- used as the base for the current time. When no new events are seen, a timer
1783-- is started with a tick duration specified by @tick@. This timer is used to
1784-- detect session timeouts in the absence of new events.
1786-- The predicate @pred@ is invoked with the current session count, if it
1787-- returns 'True' a session is ejected from the session cache before inserting
1788-- a new session. This could be useful to alert or eject sessions when the
1789-- number of sessions becomes too high.
1793{-# INLINABLE classifySessionsBy #-}
1795 :: (IsStream t, MonadAsync m, Ord k)
1796 => Double -- ^ timer tick in seconds
1797 -> Double -- ^ session timeout in seconds
1798 -> Bool -- ^ reset the timeout when an event is received
1799 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
1800 -> Fold m a b -- ^ Fold to be applied to session events
1801 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
1802 -> t m (k, b) -- ^ session key, fold result
1803classifySessionsBy tick tmout reset ejectPred
1804 (Fold step initial extract) str =
1805 concatMap sessionOutputStream $
1806 scanlMAfter' sstep (return szero) flush stream
1810 timeoutMs = toRelTime (round (tmout * 1000) :: MilliSecond64)
1811 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
1812 szero = SessionState
1813 { sessionCurTime = toAbsTime (0 :: MilliSecond64)
1814 , sessionEventTime = toAbsTime (0 :: MilliSecond64)
1816 , sessionTimerHeap = H.empty
1817 , sessionKeyValueMap = Map.empty
1818 , sessionOutputStream = K.nil
1821 -- We can eject sessions based on the current session count to limit
1822 -- memory consumption. There are two possible strategies:
1824 -- 1) Eject old sessions or sessions beyond a certain/lower timeout
1825 -- threshold even before timeout, effectively reduce the timeout.
1826 -- 2) Drop creation of new sessions but keep accepting new events for the
1829 -- We use the first strategy as of now.
1831 -- Got a new stream input element
1832 sstep session@SessionState{..} (Just (key, value, timestamp)) = do
1833 -- XXX we should use a heap in pinned memory to scale it to a large
1836 -- To detect session inactivity we keep a timestamp of the latest event
1837 -- in the Map along with the fold result. When we purge the session
1838 -- from the heap we match the timestamp in the heap with the timestamp
1839 -- in the Map, if the latest timestamp is newer and has not expired we
1840 -- reinsert the key in the heap.
1842 -- XXX if the key is an Int, we can also use an IntMap for slightly
1843 -- better performance.
1845 let curTime = max sessionEventTime timestamp
1846 mOld = Map.lookup key sessionKeyValueMap
1848 -- deleting a key from the heap is expensive, so we never
1849 -- delete a key from heap, we just purge it from the Map and it
1850 -- gets purged from the heap on timeout. We just need an extra
1851 -- lookup in the Map when the key is purged from the heap, that
1852 -- should not be expensive.
1854 let (mp, cnt) = case mOld of
1855 Nothing -> (sessionKeyValueMap, sessionCount)
1856 Just _ -> (Map.delete key sessionKeyValueMap
1859 { sessionCurTime = curTime
1860 , sessionEventTime = curTime
1861 , sessionCount = cnt
1862 , sessionKeyValueMap = mp
1863 , sessionOutputStream = yield (key, fb)
1866 let acc = Tuple' timestamp fs1
1867 (hp1, mp1, out1, cnt1) <- do
1868 let vars = (sessionTimerHeap, sessionKeyValueMap,
1869 K.nil, sessionCount)
1871 -- inserting new entry
1873 -- Eject a session from heap and map is needed
1874 eject <- ejectPred sessionCount
1875 (hp, mp, out, cnt) <-
1880 -- Insert the new session in heap
1881 let expiry = addToAbsTime timestamp timeoutMs
1882 hp' = H.insert (Entry expiry key) hp
1883 in return (hp', mp, out, cnt + 1)
1884 -- updating old entry
1885 Just _ -> return vars
1887 let mp2 = Map.insert key acc mp1
1888 return $ SessionState
1889 { sessionCurTime = curTime
1890 , sessionEventTime = curTime
1891 , sessionCount = cnt1
1892 , sessionTimerHeap = hp1
1893 , sessionKeyValueMap = mp2
1894 , sessionOutputStream = out1
1899 Just (Tuple' _ acc) -> return $ FL.Partial acc
1901 FL.Done fb -> done fb
1903 res <- step fs value
1905 FL.Done fb -> done fb
1906 FL.Partial fs1 -> partial fs1
1908 -- Got a timer tick event
1909 sstep sessionState@SessionState{..} Nothing =
1910 let curTime = addToAbsTime sessionCurTime tickMs
1911 in ejectExpired sessionState curTime
1913 flush session@SessionState{..} = do
1914 (hp', mp', out, count) <-
1917 , sessionKeyValueMap
1922 { sessionCount = count
1923 , sessionTimerHeap = hp'
1924 , sessionKeyValueMap = mp'
1925 , sessionOutputStream = out
1928 -- delete from map and output the fold accumulator
1929 ejectEntry hp mp out cnt acc key = do
1931 let out1 = (key, sess) `K.cons` out
1932 let mp1 = Map.delete key mp
1933 return (hp, mp1, out1, cnt - 1)
1935 ejectAll (hp, mp, out, !cnt) = do
1936 let hres = H.uncons hp
1938 Just (Entry _ key, hp1) -> do
1939 r <- case Map.lookup key mp of
1940 Nothing -> return (hp1, mp, out, cnt)
1941 Just (Tuple' _ acc) -> ejectEntry hp1 mp out cnt acc key
1944 assert (Map.null mp) (return ())
1945 return (hp, mp, out, cnt)
1947 ejectOne (hp, mp, out, !cnt) = do
1948 let hres = H.uncons hp
1950 Just (Entry expiry key, hp1) ->
1951 case Map.lookup key mp of
1952 Nothing -> ejectOne (hp1, mp, out, cnt)
1953 Just (Tuple' latestTS acc) -> do
1954 let expiry1 = addToAbsTime latestTS timeoutMs
1955 if not reset || expiry1 <= expiry
1956 then ejectEntry hp1 mp out cnt acc key
1958 -- reset the session timeout and continue
1959 let hp2 = H.insert (Entry expiry1 key) hp1
1960 in ejectOne (hp2, mp, out, cnt)
1962 assert (Map.null mp) (return ())
1963 return (hp, mp, out, cnt)
1965 ejectExpired session@SessionState{..} curTime = do
1966 (hp', mp', out, count) <-
1967 ejectLoop sessionTimerHeap sessionKeyValueMap K.nil sessionCount
1969 { sessionCurTime = curTime
1970 , sessionCount = count
1971 , sessionTimerHeap = hp'
1972 , sessionKeyValueMap = mp'
1973 , sessionOutputStream = out
1978 ejectLoop hp mp out !cnt = do
1979 let hres = H.uncons hp
1981 Just (Entry expiry key, hp1) -> do
1983 if curTime >= expiry
1984 then return (True, False)
1990 case Map.lookup key mp of
1991 Nothing -> ejectLoop hp1 mp out cnt
1992 Just (Tuple' latestTS acc) -> do
1993 let expiry1 = addToAbsTime latestTS timeoutMs
1994 if expiry1 <= curTime || not reset || force
1996 (hp2,mp1,out1,cnt1) <-
1997 ejectEntry hp1 mp out cnt acc key
1998 ejectLoop hp2 mp1 out1 cnt1
2000 -- reset the session timeout and continue
2001 let hp2 = H.insert (Entry expiry1 key) hp1
2002 in ejectLoop hp2 mp out cnt
2003 else return (hp, mp, out, cnt)
2005 assert (Map.null mp) (return ())
2006 return (hp, mp, out, cnt)
2008 -- merge timer events in the stream
2009 stream = Serial.map Just str `Par.parallelFst` repeatM timer
2011 liftIO $ threadDelay (round $ tick * 1000000)
2014-- | Like 'classifySessionsOf' but the session is kept alive if an event is
2015-- received within the session window. The session times out and gets closed
2016-- only if no event is received within the specified session window size.
2018-- If the ejection predicate returns 'True', the session that was idle for
2019-- the longest time is ejected before inserting a new session.
2022-- classifyKeepAliveSessions timeout pred = classifySessionsBy 1 timeout True pred
2027{-# INLINABLE classifyKeepAliveSessions #-}
2028classifyKeepAliveSessions ::
2029 (IsStream t, MonadAsync m, Ord k)
2030 => Double -- ^ session inactive timeout
2031 -> (Int -> m Bool) -- ^ predicate to eject sessions on session count
2032 -> Fold m a b -- ^ Fold to be applied to session payload data
2033 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
2035classifyKeepAliveSessions tmout =
2036 classifySessionsBy 1 tmout True
2038------------------------------------------------------------------------------
2039-- Keyed tumbling windows
2040------------------------------------------------------------------------------
2042-- Tumbling windows is a special case of sliding windows where the window slide
2043-- is the same as the window size. Or it can be a special case of session
2044-- windows where the reset flag is set to False.
2046-- XXX instead of using the early termination flag in the stream, we can use an
2047-- early terminating fold instead.
2050-- | Split the stream into fixed size chunks of specified size. Within each
2051-- such chunk fold the elements in buckets identified by the keys. A particular
2052-- bucket fold can be terminated early if a closing flag is encountered in an
2053-- element for that key.
2056{-# INLINABLE classifyChunksOf #-}
2058 :: (IsStream t, MonadAsync m, Ord k)
2059 => Int -- ^ window size
2060 -> Fold m a b -- ^ Fold to be applied to window events
2061 -> t m (k, a, Bool) -- ^ window key, data, close event
2063classifyChunksOf wsize = classifyChunksBy wsize False
2066-- | Split the stream into fixed size time windows of specified interval in
2067-- seconds. Within each such window, fold the elements in sessions identified
2068-- by the session keys. The fold result is emitted in the output stream if the
2069-- fold returns a 'Left' result or if the time window ends.
2071-- Session @timestamp@ in the input stream is an absolute time from some epoch,
2072-- characterizing the time when the input element was generated. To detect
2073-- session window end, a monotonic event time clock is maintained synced with
2074-- the timestamps with a clock resolution of 1 second.
2076-- If the ejection predicate returns 'True', the session with the longest
2077-- lifetime is ejected before inserting a new session.
2080-- classifySessionsOf interval pred = classifySessionsBy 1 interval False pred
2085-- $ S.classifySessionsOf 3 (const (return False)) (fmap Right FL.toList)
2086-- $ S.map (\(ts,(k,a)) -> (k, a, ts))
2089-- $ (,) <$> S.fromList [1,2,3] <*> S.fromList [1,2,3]
2094{-# INLINABLE classifySessionsOf #-}
2095classifySessionsOf ::
2096 (IsStream t, MonadAsync m, Ord k)
2097 => Double -- ^ time window size
2098 -> (Int -> m Bool) -- ^ predicate to eject sessions on session count
2099 -> Fold m a b -- ^ Fold to be applied to session events
2100 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
2102classifySessionsOf interval =
2103 classifySessionsBy 1 interval False
2105------------------------------------------------------------------------------
2107------------------------------------------------------------------------------
2109-- | @splitInnerBy splitter joiner stream@ splits the inner containers @f a@ of
2110-- an input stream @t m (f a)@ using the @splitter@ function. Container
2111-- elements @f a@ are collected until a split occurs, then all the elements
2112-- before the split are joined using the @joiner@ function.
2114-- For example, if we have a stream of @Array Word8@, we may want to split the
2115-- stream into arrays representing lines separated by '\n' byte such that the
2116-- resulting stream after a split would be one array for each line.
2118-- CAUTION! This is not a true streaming function as the container size after
2119-- the split and merge may not be bounded.
2122{-# INLINE splitInnerBy #-}
2124 :: (IsStream t, Monad m)
2125 => (f a -> m (f a, Maybe (f a))) -- splitter
2126 -> (f a -> f a -> m (f a)) -- joiner
2129splitInnerBy splitter joiner xs =
2130 D.fromStreamD $ D.splitInnerBy splitter joiner $ D.toStreamD xs
2132-- | Like 'splitInnerBy' but splits assuming the separator joins the segment in
2136{-# INLINE splitInnerBySuffix #-}
2138 :: (IsStream t, Monad m, Eq (f a), Monoid (f a))
2139 => (f a -> m (f a, Maybe (f a))) -- splitter
2140 -> (f a -> f a -> m (f a)) -- joiner
2143splitInnerBySuffix splitter joiner xs =
2144 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