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.Foreign.Types (Array)
258import Streamly.Internal.Data.SVar (MonadAsync)
259import Streamly.Internal.Data.Stream.Ahead (ahead)
260import Streamly.Internal.Data.Stream.Async (async, wAsync)
261import Streamly.Internal.Data.Stream.IsStream.Common
273import Streamly.Internal.Data.Stream.Parallel (parallel)
274import Streamly.Internal.Data.Stream.Prelude
275 ( fromStreamS, toStreamS, concatFoldableWith, concatMapFoldableWith
276 , concatForFoldableWith)
277import Streamly.Internal.Data.Stream.StreamD (fromStreamD, toStreamD)
278import Streamly.Internal.Data.Stream.StreamK (IsStream)
279import Streamly.Internal.Data.Stream.Serial (serial, wSerial)
280import Streamly.Internal.Data.Time.Units
281 ( AbsTime, MilliSecond64(..), addToAbsTime, toRelTime
283import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))
285import qualified Data.Heap as H
286import qualified Data.Map.Strict as Map
287import qualified Streamly.Internal.Data.Array.Foreign.Types as A
288import qualified Streamly.Internal.Data.Fold as FL
289import qualified Streamly.Internal.Data.Parser.ParserK.Types as PRK
290import qualified Streamly.Internal.Data.Parser.ParserD as PRD
291import qualified Streamly.Internal.Data.Stream.Parallel as Par
292import qualified Streamly.Internal.Data.Stream.Serial as Serial
293import qualified Streamly.Internal.Data.Stream.StreamD as D
294import qualified Streamly.Internal.Data.Stream.StreamK as K
295import qualified Streamly.Internal.Data.Stream.Zip as Z
296#ifdef USE_STREAMK_ONLY
297import qualified Streamly.Internal.Data.Stream.StreamK as S
299import qualified Streamly.Internal.Data.Stream.StreamD as S
302import Prelude hiding (zipWith, concatMap, concat)
304-------------------------------------------------------------------------------
306-------------------------------------------------------------------------------
308-- | Append the outputs of two streams, yielding all the elements from the
309-- first stream and then yielding all the elements from the second stream.
311-- IMPORTANT NOTE: This could be 100x faster than @serial/<>@ for appending a
312-- few (say 100) streams because it can fuse via stream fusion. However, it
313-- does not scale for a large number of streams (say 1000s) and becomes
314-- qudartically slow. Therefore use this for custom appending of a few streams
315-- but use 'concatMap' or 'concatMapWith serial' for appending @n@ streams or
316-- infinite containers of streams.
320append ::(IsStream t, Monad m) => t m b -> t m b -> t m b
321append m1 m2 = fromStreamD $ D.append (toStreamD m1) (toStreamD m2)
323-------------------------------------------------------------------------------
325-------------------------------------------------------------------------------
327-- XXX Same as 'wSerial'. We should perhaps rename wSerial to interleave.
328-- XXX Document the interleaving behavior of side effects in all the
329-- interleaving combinators.
330-- XXX Write time-domain equivalents of these. In the time domain we can
331-- interleave two streams such that the value of second stream is always taken
332-- from its last value even if no new value is being yielded, like
333-- zipWithLatest. It would be something like interleaveWithLatest.
335-- | Interleaves the outputs of two streams, yielding elements from each stream
336-- alternately, starting from the first stream. If any of the streams finishes
337-- early the other stream continues alone until it too finishes.
339-- >>> :set -XOverloadedStrings
340-- >>> interleave "ab" ",,,," :: SerialT Identity Char
342-- >>> interleave "abcd" ",," :: SerialT Identity Char
345-- 'interleave' is dual to 'interleaveMin', it can be called @interleaveMax@.
347-- Do not use at scale in concatMapWith.
350{-# INLINE interleave #-}
351interleave ::(IsStream t, Monad m) => t m b -> t m b -> t m b
352interleave m1 m2 = fromStreamD $ D.interleave (toStreamD m1) (toStreamD m2)
354-- | Interleaves the outputs of two streams, yielding elements from each stream
355-- alternately, starting from the first stream. As soon as the first stream
356-- finishes, the output stops, discarding the remaining part of the second
357-- stream. In this case, the last element in the resulting stream would be from
358-- the second stream. If the second stream finishes early then the first stream
359-- still continues to yield elements until it finishes.
361-- >>> :set -XOverloadedStrings
362-- >>> interleaveSuffix "abc" ",,,," :: SerialT Identity Char
364-- >>> interleaveSuffix "abc" "," :: SerialT Identity Char
367-- 'interleaveSuffix' is a dual of 'interleaveInfix'.
369-- Do not use at scale in concatMapWith.
372{-# INLINE interleaveSuffix #-}
373interleaveSuffix ::(IsStream t, Monad m) => t m b -> t m b -> t m b
374interleaveSuffix m1 m2 =
375 fromStreamD $ D.interleaveSuffix (toStreamD m1) (toStreamD m2)
377-- | Interleaves the outputs of two streams, yielding elements from each stream
378-- alternately, starting from the first stream and ending at the first stream.
379-- If the second stream is longer than the first, elements from the second
380-- stream are infixed with elements from the first stream. If the first stream
381-- is longer then it continues yielding elements even after the second stream
384-- >>> :set -XOverloadedStrings
385-- >>> interleaveInfix "abc" ",,,," :: SerialT Identity Char
387-- >>> interleaveInfix "abc" "," :: SerialT Identity Char
390-- 'interleaveInfix' is a dual of 'interleaveSuffix'.
392-- Do not use at scale in concatMapWith.
395{-# INLINE interleaveInfix #-}
396interleaveInfix ::(IsStream t, Monad m) => t m b -> t m b -> t m b
397interleaveInfix m1 m2 =
398 fromStreamD $ D.interleaveInfix (toStreamD m1) (toStreamD m2)
400-- | Interleaves the outputs of two streams, yielding elements from each stream
401-- alternately, starting from the first stream. The output stops as soon as any
402-- of the two streams finishes, discarding the remaining part of the other
403-- stream. The last element of the resulting stream would be from the longer
406-- >>> :set -XOverloadedStrings
407-- >>> interleaveMin "ab" ",,,," :: SerialT Identity Char
409-- >>> interleaveMin "abcd" ",," :: SerialT Identity Char
412-- 'interleaveMin' is dual to 'interleave'.
414-- Do not use at scale in concatMapWith.
417{-# INLINE interleaveMin #-}
418interleaveMin ::(IsStream t, Monad m) => t m b -> t m b -> t m b
419interleaveMin m1 m2 = fromStreamD $ D.interleaveMin (toStreamD m1) (toStreamD m2)
421-------------------------------------------------------------------------------
423-------------------------------------------------------------------------------
425-- | Schedule the execution of two streams in a fair round-robin manner,
426-- executing each stream once, alternately. Execution of a stream may not
427-- necessarily result in an output, a stream may chose to @Skip@ producing an
428-- element until later giving the other stream a chance to run. Therefore, this
429-- combinator fairly interleaves the execution of two streams rather than
430-- fairly interleaving the output of the two streams. This can be useful in
431-- co-operative multitasking without using explicit threads. This can be used
432-- as an alternative to `async`.
434-- Do not use at scale in concatMapWith.
437{-# INLINE roundrobin #-}
438roundrobin ::(IsStream t, Monad m) => t m b -> t m b -> t m b
439roundrobin m1 m2 = fromStreamD $ D.roundRobin (toStreamD m1) (toStreamD m2)
441------------------------------------------------------------------------------
443------------------------------------------------------------------------------
445-- | Merge two streams using a comparison function. The head elements of both
446-- the streams are compared and the smaller of the two elements is emitted, if
447-- both elements are equal then the element from the first stream is used
450-- If the streams are sorted in ascending order, the resulting stream would
451-- also remain sorted in ascending order.
454-- > S.toList $ S.mergeBy compare (S.fromList [1,3,5]) (S.fromList [2,4,6,8])
459{-# INLINABLE mergeBy #-}
461 (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> t m a -> t m a
462mergeBy f m1 m2 = fromStreamS $ S.mergeBy f (toStreamS m1) (toStreamS m2)
464-- | Like 'mergeBy' but with a monadic comparison function.
466-- Merge two streams randomly:
469-- > randomly _ _ = randomIO >>= \x -> return $ if x then LT else GT
470-- > S.toList $ S.mergeByM randomly (S.fromList [1,1,1,1]) (S.fromList [2,2,2,2])
474-- Merge two streams in a proportion of 2:1:
477-- proportionately m n = do
478-- ref <- newIORef $ cycle $ concat [replicate m LT, replicate n GT]
479-- return $ \\_ _ -> do
481-- writeIORef ref $ tail r
485-- f <- proportionately 2 1
486-- xs <- S.toList $ S.mergeByM f (S.fromList [1,1,1,1,1,1]) (S.fromList [2,2,2])
490-- [1,1,2,1,1,2,1,1,2]
494{-# INLINABLE mergeByM #-}
496 :: (IsStream t, Monad m)
497 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
498mergeByM f m1 m2 = fromStreamS $ S.mergeByM f (toStreamS m1) (toStreamS m2)
501-- | Like 'mergeByM' but stops merging as soon as any of the two streams stops.
502{-# INLINABLE mergeEndByAny #-}
504 :: (IsStream t, Monad m)
505 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
506mergeEndByAny f m1 m2 = fromStreamD $
507 D.mergeEndByAny f (toStreamD m1) (toStreamD m2)
509-- Like 'mergeByM' but stops merging as soon as the first stream stops.
510{-# INLINABLE mergeEndByFirst #-}
512 :: (IsStream t, Monad m)
513 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
514mergeEndByFirst f m1 m2 = fromStreamS $
515 D.mergeEndByFirst f (toStreamD m1) (toStreamD m2)
517-- Holding this back for now, we may want to use the name "merge" differently
518-- | Same as @'mergeBy' 'compare'@.
521-- > S.toList $ S.merge (S.fromList [1,3,5]) (S.fromList [2,4,6,8])
526{-# INLINABLE merge #-}
528 (IsStream t, Monad m, Ord a) => t m a -> t m a -> t m a
529merge = mergeBy compare
532-- | Like 'mergeBy' but merges concurrently (i.e. both the elements being
533-- merged are generated concurrently).
536{-# INLINE mergeAsyncBy #-}
537mergeAsyncBy :: (IsStream t, MonadAsync m)
538 => (a -> a -> Ordering) -> t m a -> t m a -> t m a
539mergeAsyncBy f = mergeAsyncByM (\a b -> return $ f a b)
541-- | Like 'mergeByM' but merges concurrently (i.e. both the elements being
542-- merged are generated concurrently).
545{-# INLINE mergeAsyncByM #-}
546mergeAsyncByM :: (IsStream t, MonadAsync m)
547 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
548mergeAsyncByM f m1 m2 =
550 let par = Par.mkParallelD . toStreamD
551 in D.mergeByM f (par m1) (par m2)
553------------------------------------------------------------------------------
554-- Combine N Streams - concatMap
555------------------------------------------------------------------------------
557-- | @concatMapWith merge map stream@ is a two dimensional looping combinator.
558-- The first argument specifies a merge or concat function that is used to
559-- merge the streams generated by applying the second argument i.e. the @map@
560-- function to each element of the input stream. The concat function could be
561-- 'serial', 'parallel', 'async', 'ahead' or any other zip or merge function
562-- and the second argument could be any stream generation function using a
565-- /Compare 'foldMapWith'/
569-- /Since: 0.8.0 (signature change)/
570{-# INLINE concatMapWith #-}
573 => (t m b -> t m b -> t m b)
577concatMapWith = K.concatMapBy
579-- | Like 'concatMapWith' but carries a state which can be used to share
580-- information across multiple steps of concat.
583-- concatSmapMWith combine f initial = concatMapWith combine id . smapM f initial
588{-# INLINE concatSmapMWith #-}
590 :: (IsStream t, Monad m)
591 => (t m b -> t m b -> t m b)
592 -> (s -> a -> m (s, t m b))
596concatSmapMWith combine f initial = concatMapWith combine id . smapM f initial
598-- | Map a stream producing function on each element of the stream and then
599-- flatten the results into a single stream.
602-- concatMap f = 'concat . map f'
603-- concatMap = 'concatMapWith' 'Serial.serial'
604-- concatMap f = 'concatMapM' (return . f)
605-- concatMap f = 'concatUnfold' (UF.lmap f UF.fromStream)
609{-# INLINE concatMap #-}
610concatMap ::(IsStream t, Monad m) => (a -> t m b) -> t m a -> t m b
611concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)
613-- | Flatten a stream of streams to a single stream.
616-- concat = concatMap id
621concat :: (IsStream t, Monad m) => t m (t m a) -> t m a
624-- Keep concating either streams as long as rights are generated, stop as soon
625-- as a left is generated and concat the left stream.
633 :: -- (IsStream t, MonadAsync m) =>
634 (forall x. t m x -> t m x -> t m x)
635 -> (a -> t m (Either (t m b) b))
638concatMapEitherWith = undefined
641------------------------------------------------------------------------------
642-- Combine N Streams - concatUnfold
643------------------------------------------------------------------------------
645-- | Like 'concatMap' but uses an 'Unfold' for stream generation. Unlike
646-- 'concatMap' this can fuse the 'Unfold' code with the inner loop and
647-- therefore provide many times better performance.
650{-# INLINE concatUnfold #-}
651concatUnfold ::(IsStream t, Monad m) => Unfold m a b -> t m a -> t m b
652concatUnfold u m = fromStreamD $ D.concatUnfold u (toStreamD m)
654-- | Like 'concatUnfold' but interleaves the streams in the same way as
655-- 'interleave' behaves instead of appending them.
658{-# INLINE concatUnfoldInterleave #-}
659concatUnfoldInterleave ::(IsStream t, Monad m)
660 => Unfold m a b -> t m a -> t m b
661concatUnfoldInterleave u m =
662 fromStreamD $ D.concatUnfoldInterleave u (toStreamD m)
664-- | Like 'concatUnfold' but executes the streams in the same way as
668{-# INLINE concatUnfoldRoundrobin #-}
669concatUnfoldRoundrobin ::(IsStream t, Monad m)
670 => Unfold m a b -> t m a -> t m b
671concatUnfoldRoundrobin u m =
672 fromStreamD $ D.concatUnfoldRoundrobin u (toStreamD m)
674------------------------------------------------------------------------------
675-- Combine N Streams - interpose
676------------------------------------------------------------------------------
678-- > interpose x unf str = gintercalate unf str UF.identity (repeat x)
680-- | Unfold the elements of a stream, intersperse the given element between the
681-- unfolded streams and then concat them into a single stream.
683-- > unwords = S.interpose ' '
686{-# INLINE interpose #-}
687interpose :: (IsStream t, Monad m)
688 => c -> Unfold m b c -> t m b -> t m c
690 D.fromStreamD $ D.interpose (return x) unf (D.toStreamD str)
692-- interposeSuffix x unf str = gintercalateSuffix unf str UF.identity (repeat x)
694-- | Unfold the elements of a stream, append the given element after each
695-- unfolded stream and then concat them into a single stream.
697-- > unlines = S.interposeSuffix '\n'
700{-# INLINE interposeSuffix #-}
701interposeSuffix :: (IsStream t, Monad m)
702 => c -> Unfold m b c -> t m b -> t m c
703interposeSuffix x unf str =
704 D.fromStreamD $ D.interposeSuffix (return x) unf (D.toStreamD str)
706------------------------------------------------------------------------------
707-- Combine N Streams - intercalate
708------------------------------------------------------------------------------
710-- XXX we can swap the order of arguments to gintercalate so that the
711-- definition of concatUnfold becomes simpler? The first stream should be
712-- infixed inside the second one. However, if we change the order in
713-- "interleave" as well similarly, then that will make it a bit unintuitive.
715-- > concatUnfold unf str =
716-- > gintercalate unf str (UF.nilM (\_ -> return ())) (repeat ())
718-- | 'interleaveInfix' followed by unfold and concat.
721{-# INLINE gintercalate #-}
723 :: (IsStream t, Monad m)
724 => Unfold m a c -> t m a -> Unfold m b c -> t m b -> t m c
725gintercalate unf1 str1 unf2 str2 =
726 D.fromStreamD $ D.gintercalate
727 unf1 (D.toStreamD str1)
728 unf2 (D.toStreamD str2)
730-- XXX The order of arguments in "intercalate" is consistent with the list
731-- intercalate but inconsistent with gintercalate and other stream interleaving
732-- combinators. We can change the order of the arguments in other combinators
733-- but then 'interleave' combinator may become a bit unintuitive because we
734-- will be starting with the second stream.
736-- > intercalate seed unf str = gintercalate unf str unf (repeatM seed)
737-- > intercalate a unf str = concatUnfold unf $ intersperse a str
739-- | 'intersperse' followed by unfold and concat.
741-- > unwords = intercalate " " UF.fromList
743-- >>> intercalate " " UF.fromList ["abc", "def", "ghi"]
747{-# INLINE intercalate #-}
748intercalate :: (IsStream t, Monad m)
749 => b -> Unfold m b c -> t m b -> t m c
750intercalate seed unf str = D.fromStreamD $
751 D.concatUnfold unf $ D.intersperse seed (toStreamD str)
753-- | 'interleaveSuffix' followed by unfold and concat.
756{-# INLINE gintercalateSuffix #-}
758 :: (IsStream t, Monad m)
759 => Unfold m a c -> t m a -> Unfold m b c -> t m b -> t m c
760gintercalateSuffix unf1 str1 unf2 str2 =
761 D.fromStreamD $ D.gintercalateSuffix
762 unf1 (D.toStreamD str1)
763 unf2 (D.toStreamD str2)
765-- > intercalateSuffix seed unf str = gintercalateSuffix unf str unf (repeatM seed)
766-- > intercalateSuffix a unf str = concatUnfold unf $ intersperseSuffix a str
768-- | 'intersperseSuffix' followed by unfold and concat.
770-- > unlines = intercalateSuffix "\n" UF.fromList
772-- >>> intercalate "\n" UF.fromList ["abc", "def", "ghi"]
773-- > "abc\ndef\nghi\n"
776{-# INLINE intercalateSuffix #-}
777intercalateSuffix :: (IsStream t, Monad m)
778 => b -> Unfold m b c -> t m b -> t m c
779intercalateSuffix seed unf str = fromStreamD $ D.concatUnfold unf
780 $ D.intersperseSuffix (return seed) (D.toStreamD str)
782------------------------------------------------------------------------------
783-- IterateMap - Map and flatten Trees of Streams
784------------------------------------------------------------------------------
786-- | Like 'iterateM' but iterates after mapping a stream generator on the
789-- Yield an input element in the output stream, map a stream generator on it
790-- and then do the same on the resulting stream. This can be used for a depth
791-- first traversal of a tree like structure.
793-- Note that 'iterateM' is a special case of 'iterateMapWith':
796-- iterateM f = iterateMapWith serial (yieldM . f) . yieldM
799-- It can be used to traverse a tree structure. For example, to list a
803-- Stream.iterateMapWith Stream.serial
804-- (either Dir.toEither (const nil))
805-- (yield (Left "tmp"))
810{-# INLINE iterateMapWith #-}
813 => (t m a -> t m a -> t m a)
817iterateMapWith combine f = concatMapWith combine go
819 go x = yield x `combine` concatMapWith combine go (f x)
822{-# INLINE iterateUnfold #-}
823iterateUnfold :: (IsStream t, MonadAsync m)
824 => Unfold m a a -> t m a -> t m a
825iterateUnfold unf xs = undefined
828------------------------------------------------------------------------------
830------------------------------------------------------------------------------
832-- To traverse graphs we need a state to be carried around in the traversal.
833-- For example, we can use a hashmap to store the visited status of nodes.
835-- | Like 'iterateMap' but carries a state in the stream generation function.
836-- This can be used to traverse graph like structures, we can remember the
837-- visited nodes in the state to avoid cycles.
839-- Note that a combination of 'iterateMap' and 'usingState' can also be used to
840-- traverse graphs. However, this function provides a more localized state
841-- instead of using a global state.
847{-# INLINE iterateSmapMWith #-}
849 :: (IsStream t, Monad m)
850 => (t m a -> t m a -> t m a)
851 -> (b -> a -> m (b, t m a))
855iterateSmapMWith combine f initial stream =
856 concatMap (\b -> concatMapWith combine (go b) stream) (yieldM initial)
860 go b a = yield a `combine` feedback b a
864 (\(b1, s) -> concatMapWith combine (go b1) s)
867------------------------------------------------------------------------------
868-- iterateMap - Either streams
869------------------------------------------------------------------------------
871-- | In an 'Either' stream iterate on 'Left's. This is a special case of
875-- iterateMapLeftsWith combine f = iterateMapWith combine (either f (const nil))
878-- To traverse a directory tree:
881-- iterateMapLeftsWith serial Dir.toEither (yield (Left "tmp"))
886{-# INLINE iterateMapLeftsWith #-}
889 => (t m (Either a b) -> t m (Either a b) -> t m (Either a b))
890 -> (a -> t m (Either a b))
893iterateMapLeftsWith combine f = iterateMapWith combine (either f (const K.nil))
895------------------------------------------------------------------------------
897------------------------------------------------------------------------------
899-- Splitting operations that take a predicate and a Fold can be
900-- expressed using parseMany. Operations like chunksOf, intervalsOf, split*,
901-- can be expressed using parseMany when used with an appropriate Parser.
903-- XXX We need takeGE/takeBetween to implement "some" using "many".
905-- | Apply a 'Fold' repeatedly on a stream and emit the parsed values in the
908-- This is the streaming dual of the 'Streamly.Internal.Data.Fold.many'
911-- >>> f = Fold.takeLE 2 Fold.sum
912-- >>> Stream.toList $ Stream.foldMany f $ Stream.fromList [1..10]
915-- >>> f = Fold.sliceEndWith Fold.toList
916-- >>> Stream.toList $ Stream.foldMany f $ Stream.fromList "hello\nworld"
917-- > ["hello\n","world"]
921{-# INLINE foldMany #-}
923 :: (IsStream t, Monad m)
927foldMany f m = D.fromStreamD $ D.foldMany f (D.toStreamD m)
929-- | Apply a stream of folds to an input stream and emit the results in the
934{-# INLINE foldSequence #-}
936 :: -- (IsStream t, Monad m) =>
940foldSequence _f _m = undefined
942-- | Iterate a fold generator on a stream. The initial value @b@ is used to
943-- generate the first fold, the fold is applied on the stream and the result of
944-- the fold is used to generate the next fold and so on.
946-- >>> f x = Fold.takeLE 2 (Fold.mconcatTo x)
947-- >>> s = Stream.map Sum $ Stream.fromList [1..10]
948-- >>> Stream.toList $ Stream.map getSum $ Stream.foldIterate f 0 s
949-- > [3,10,21,36,55,55]
951-- This is the streaming equivalent of monad like sequenced application of
952-- folds where next fold is dependent on the previous fold.
956{-# INLINE foldIterate #-}
958 :: -- (IsStream t, Monad m) =>
963foldIterate _f _i _m = undefined
964-- D.fromStreamD $ D.foldIterate f i (D.toStreamD m)
966------------------------------------------------------------------------------
968------------------------------------------------------------------------------
970-- | Apply a 'Parser' repeatedly on a stream and emit the parsed values in the
973-- This is the streaming equivalent of the 'Streamly.Internal.Data.Parser.many'
976-- >>> S.toList $ S.parseMany (PR.take 2 $ PR.fromFold FL.sum) $ S.fromList [1..10]
979-- >>> S.toList $ S.parseMany (PR.line FL.toList) $ S.fromList "hello\nworld"
980-- > ["hello\n","world"]
984{-# INLINE parseMany #-}
986 :: (IsStream t, MonadThrow m)
991 D.fromStreamD $ D.parseMany (PRK.fromParserK p) (D.toStreamD m)
993{-# INLINE parseManyD #-}
995 :: (IsStream t, MonadThrow m)
1000 D.fromStreamD $ D.parseMany p (D.toStreamD m)
1002-- | Apply a stream of parsers to an input stream and emit the results in the
1007{-# INLINE parseSequence #-}
1009 :: -- (IsStream t, Monad m) =>
1013parseSequence _f _m = undefined
1015-- | @parseManyTill collect test stream@ tries the parser @test@ on the input,
1016-- if @test@ fails it backtracks and tries @collect@, after @collect@ succeeds
1017-- @test@ is tried again and so on. The parser stops when @test@ succeeds. The
1018-- output of @test@ is discarded and the output of @collect@ is emitted in the
1019-- output stream. The parser fails if @collect@ fails.
1023{-# INLINE parseManyTill #-}
1025 -- (IsStream t, MonadThrow m) =>
1030parseManyTill = undefined
1032-- | Iterate a parser generating function on a stream. The initial value @b@ is
1033-- used to generate the first parser, the parser is applied on the stream and
1034-- the result is used to generate the next parser and so on.
1036-- >>> S.toList $ S.map getSum $ S.parseIterate (\b -> PR.take 2 (FL.mconcatTo b)) 0 $ S.map Sum $ S.fromList [1..10]
1037-- > [3,10,21,36,55,55]
1039-- This is the streaming equivalent of monad like sequenced application of
1040-- parsers where next parser is dependent on the previous parser.
1044{-# INLINE parseIterate #-}
1046 :: (IsStream t, MonadThrow m)
1047 => (b -> Parser m a b)
1051parseIterate f i m = D.fromStreamD $
1052 D.parseIterate (PRK.fromParserK . f) i (D.toStreamD m)
1054------------------------------------------------------------------------------
1056------------------------------------------------------------------------------
1058-- | Drop prefix from the input stream if present.
1062-- /Unimplemented/ - Help wanted.
1063{-# INLINE dropPrefix #-}
1065 -- (Eq a, IsStream t, Monad m) =>
1066 t m a -> t m a -> t m a
1067dropPrefix = error "Not implemented yet!"
1069-- | Drop all matching infix from the input stream if present. Infix stream
1070-- may be consumed multiple times.
1072-- Space: @O(n)@ where n is the length of the infix.
1074-- /Unimplemented/ - Help wanted.
1075{-# INLINE dropInfix #-}
1077 -- (Eq a, IsStream t, Monad m) =>
1078 t m a -> t m a -> t m a
1079dropInfix = error "Not implemented yet!"
1081-- | Drop suffix from the input stream if present. Suffix stream may be
1082-- consumed multiple times.
1084-- Space: @O(n)@ where n is the length of the suffix.
1086-- /Unimplemented/ - Help wanted.
1087{-# INLINE dropSuffix #-}
1089 -- (Eq a, IsStream t, Monad m) =>
1090 t m a -> t m a -> t m a
1091dropSuffix = error "Not implemented yet!"
1093------------------------------------------------------------------------------
1094-- Generalized grouping
1095------------------------------------------------------------------------------
1097-- This combinator is the most general grouping combinator and can be used to
1098-- implement all other grouping combinators.
1100-- XXX check if this can implement the splitOn combinator i.e. we can slide in
1101-- new elements, slide out old elements and incrementally compute the hash.
1102-- Also, can we implement the windowed classification combinators using this?
1104-- In fact this is a parse. Instead of using a special return value in the fold
1105-- we are using a mapping function.
1107-- Note that 'scanl'' (usually followed by a map to extract the desired value
1108-- from the accumulator) can be used to realize many implementations e.g. a
1109-- sliding window implementation. A scan followed by a mapMaybe is also a good
1110-- pattern to express many problems where we want to emit a filtered output and
1111-- not emit an output on every input.
1113-- Passing on of the initial accumulator value to the next fold is equivalent
1114-- to returning the leftover concept.
1117-- | @groupScan splitter fold stream@ folds the input stream using @fold@.
1118-- @splitter@ is applied on the accumulator of the fold every time an item is
1119-- consumed by the fold. The fold continues until @splitter@ returns a 'Just'
1120-- value. A 'Just' result from the @splitter@ specifies a result to be emitted
1121-- in the output stream and the initial value of the accumulator for the next
1122-- group's fold. This allows us to control whether to start fresh for the next
1123-- fold or to continue from the previous fold's output.
1125{-# INLINE groupScan #-}
1127 :: (IsStream t, Monad m)
1128 => (x -> m (Maybe (b, x))) -> Fold m a x -> t m a -> t m b
1129groupScan split fold m = undefined
1132------------------------------------------------------------------------------
1134------------------------------------------------------------------------------
1136-- | @groupsBy cmp f $ S.fromList [a,b,c,...]@ assigns the element @a@ to the
1137-- first group, if @b \`cmp` a@ is 'True' then @b@ is also assigned to the same
1138-- group. If @c \`cmp` a@ is 'True' then @c@ is also assigned to the same
1139-- group and so on. When the comparison fails a new group is started. Each
1140-- group is folded using the fold @f@ and the result of the fold is emitted in
1141-- the output stream.
1143-- >>> S.toList $ S.groupsBy (>) FL.toList $ S.fromList [1,3,7,0,2,5]
1144-- > [[1,3,7],[0,2,5]]
1147{-# INLINE groupsBy #-}
1149 :: (IsStream t, Monad m)
1154groupsBy cmp f m = D.fromStreamD $ D.groupsBy cmp f (D.toStreamD m)
1156-- | Unlike @groupsBy@ this function performs a rolling comparison of two
1157-- successive elements in the input stream. @groupsByRolling cmp f $ S.fromList
1158-- [a,b,c,...]@ assigns the element @a@ to the first group, if @a \`cmp` b@ is
1159-- 'True' then @b@ is also assigned to the same group. If @b \`cmp` c@ is
1160-- 'True' then @c@ is also assigned to the same group and so on. When the
1161-- comparison fails a new group is started. Each group is folded using the fold
1164-- >>> S.toList $ S.groupsByRolling (\a b -> a + 1 == b) FL.toList $ S.fromList [1,2,3,7,8,9]
1165-- > [[1,2,3],[7,8,9]]
1168{-# INLINE groupsByRolling #-}
1170 :: (IsStream t, Monad m)
1175groupsByRolling cmp f m = D.fromStreamD $ D.groupsRollingBy cmp f (D.toStreamD m)
1178-- > groups = groupsBy (==)
1179-- > groups = groupsByRolling (==)
1181-- Groups contiguous spans of equal elements together in individual groups.
1183-- >>> S.toList $ S.groups FL.toList $ S.fromList [1,1,2,2]
1187{-# INLINE groups #-}
1188groups :: (IsStream t, Monad m, Eq a) => Fold m a b -> t m a -> t m b
1189groups = groupsBy (==)
1191------------------------------------------------------------------------------
1192-- Splitting - by a predicate
1193------------------------------------------------------------------------------
1195-- TODO: Use a Splitter configuration similar to the "split" package to make it
1196-- possible to express all splitting combinations. In general, we can have
1197-- infix/suffix/prefix/condensing of separators, dropping both leading/trailing
1198-- separators. We can have a single split operation taking the splitter config
1201-- | Split on an infixed separator element, dropping the separator. Splits the
1202-- stream on separator elements determined by the supplied predicate, separator
1203-- is considered as infixed between two segments, if one side of the separator
1204-- is missing then it is parsed as an empty stream. The supplied 'Fold' is
1205-- applied on the split segments. With '-' representing non-separator elements
1206-- and '.' as separator, 'splitOn' splits as follows:
1209-- "--.--" => "--" "--"
1214-- @splitOn (== x)@ is an inverse of @intercalate (S.yield x)@
1216-- Let's use the following definition for illustration:
1218-- > splitOn' p xs = S.toList $ S.splitOn p (FL.toList) (S.fromList xs)
1220-- >>> splitOn' (== '.') ""
1223-- >>> splitOn' (== '.') "."
1226-- >>> splitOn' (== '.') ".a"
1229-- >>> splitOn' (== '.') "a."
1232-- >>> splitOn' (== '.') "a.b"
1235-- >>> splitOn' (== '.') "a..b"
1240-- This can be considered as an n-fold version of 'breakOn' where we apply
1241-- 'breakOn' successively on the input stream, dropping the first element
1242-- of the second segment after each break.
1244{-# INLINE splitOn #-}
1246 :: (IsStream t, Monad m)
1247 => (a -> Bool) -> Fold m a b -> t m a -> t m b
1248splitOn predicate f = foldMany (FL.sliceSepBy predicate f)
1250-- | Like 'splitOn' but the separator is considered as suffixed to the segments
1251-- in the stream. A missing suffix at the end is allowed. A separator at the
1252-- beginning is parsed as empty segment. With '-' representing elements and
1253-- '.' as separator, 'splitOnSuffix' splits as follows:
1256-- "--.--." => "--" "--"
1257-- "--.--" => "--" "--"
1261-- > splitOnSuffix' p xs = S.toList $ S.splitSuffixBy p (FL.toList) (S.fromList xs)
1263-- >>> splitOnSuffix' (== '.') ""
1266-- >>> splitOnSuffix' (== '.') "."
1269-- >>> splitOnSuffix' (== '.') "a"
1272-- >>> splitOnSuffix' (== '.') ".a"
1275-- >>> splitOnSuffix' (== '.') "a."
1278-- >>> splitOnSuffix' (== '.') "a.b"
1281-- >>> splitOnSuffix' (== '.') "a.b."
1284-- >>> splitOnSuffix' (== '.') "a..b.."
1287-- > lines = splitOnSuffix (== '\n')
1291-- This can be considered as an n-fold version of 'breakPost' where we apply
1292-- 'breakPost' successively on the input stream, dropping the first element
1293-- of the second segment after each break.
1295{-# INLINE splitOnSuffix #-}
1297 :: (IsStream t, Monad m)
1298 => (a -> Bool) -> Fold m a b -> t m a -> t m b
1299splitOnSuffix predicate f m =
1300 D.fromStreamD $ D.foldMany1 (FL.sliceSepBy predicate f) (D.toStreamD m)
1302-- | Like 'splitOn' after stripping leading, trailing, and repeated separators.
1303-- Therefore, @".a..b."@ with '.' as the separator would be parsed as
1304-- @["a","b"]@. In other words, its like parsing words from whitespace
1307-- > wordsBy' p xs = S.toList $ S.wordsBy p (FL.toList) (S.fromList xs)
1309-- >>> wordsBy' (== ',') ""
1312-- >>> wordsBy' (== ',') ","
1315-- >>> wordsBy' (== ',') ",a,,b,"
1318-- > words = wordsBy isSpace
1322-- It is equivalent to splitting in any of the infix/prefix/suffix styles
1323-- followed by removal of empty segments.
1324{-# INLINE wordsBy #-}
1326 :: (IsStream t, Monad m)
1327 => (a -> Bool) -> Fold m a b -> t m a -> t m b
1328wordsBy predicate f m =
1329 D.fromStreamD $ D.wordsBy predicate f (D.toStreamD m)
1331-- | Like 'splitOnSuffix' but keeps the suffix attached to the resulting
1334-- > splitWithSuffix' p xs = S.toList $ S.splitWithSuffix p (FL.toList) (S.fromList xs)
1336-- >>> splitWithSuffix' (== '.') ""
1339-- >>> splitWithSuffix' (== '.') "."
1342-- >>> splitWithSuffix' (== '.') "a"
1345-- >>> splitWithSuffix' (== '.') ".a"
1348-- >>> splitWithSuffix' (== '.') "a."
1351-- >>> splitWithSuffix' (== '.') "a.b"
1354-- >>> splitWithSuffix' (== '.') "a.b."
1357-- >>> splitWithSuffix' (== '.') "a..b.."
1358-- > ["a.",".","b.","."]
1362-- This can be considered as an n-fold version of 'breakPost' where we apply
1363-- 'breakPost' successively on the input stream.
1365{-# INLINE splitWithSuffix #-}
1367 :: (IsStream t, Monad m)
1368 => (a -> Bool) -> Fold m a b -> t m a -> t m b
1369splitWithSuffix predicate f m =
1370 D.fromStreamD $ D.foldMany1 (FL.sliceEndWith predicate f) (D.toStreamD m)
1372------------------------------------------------------------------------------
1373-- Splitting - on a delimiter sequence
1374------------------------------------------------------------------------------
1376-- Int list examples for splitOn:
1378-- >>> splitList [] [1,2,3,3,4]
1379-- > [[1],[2],[3],[3],[4]]
1381-- >>> splitList [5] [1,2,3,3,4]
1384-- >>> splitList [1] [1,2,3,3,4]
1387-- >>> splitList [4] [1,2,3,3,4]
1390-- >>> splitList [2] [1,2,3,3,4]
1393-- >>> splitList [3] [1,2,3,3,4]
1396-- >>> splitList [3,3] [1,2,3,3,4]
1399-- >>> splitList [1,2,3,3,4] [1,2,3,3,4]
1403-- This can be implemented easily using Rabin Karp
1404-- | Split on any one of the given patterns.
1405{-# INLINE splitOnAny #-}
1407 :: (IsStream t, Monad m, Storable a, Integral a)
1408 => [Array a] -> Fold m a b -> t m a -> t m b
1409splitOnAny subseq f m = undefined -- D.fromStreamD $ D.splitOnAny f subseq (D.toStreamD m)
1412-- XXX use a non-monadic intersperse to remove the MonadAsync constraint.
1414-- | Like 'splitOnSeq' but splits the separator as well, as an infix token.
1416-- > splitOn'_ pat xs = S.toList $ S.splitOn' (A.fromList pat) (FL.toList) (S.fromList xs)
1418-- >>> splitOn'_ "" "hello"
1419-- > ["h","","e","","l","","l","","o"]
1421-- >>> splitOn'_ "hello" ""
1424-- >>> splitOn'_ "hello" "hello"
1427-- >>> splitOn'_ "x" "hello"
1430-- >>> splitOn'_ "h" "hello"
1433-- >>> splitOn'_ "o" "hello"
1436-- >>> splitOn'_ "e" "hello"
1439-- >>> splitOn'_ "l" "hello"
1440-- > ["he","l","","l","o"]
1442-- >>> splitOn'_ "ll" "hello"
1446{-# INLINE splitBySeq #-}
1448 :: (IsStream t, MonadAsync m, Storable a, Enum a, Eq a)
1449 => Array a -> Fold m a b -> t m a -> t m b
1450splitBySeq patt f m =
1451 intersperseM (fold f (A.toStream patt)) $ splitOnSeq patt f m
1453-- | Like 'splitSuffixBy' but the separator is a sequence of elements, instead
1454-- of a predicate for a single element.
1456-- > splitOnSuffixSeq_ pat xs = S.toList $ S.splitOnSuffixSeq (A.fromList pat) (FL.toList) (S.fromList xs)
1458-- >>> splitOnSuffixSeq_ "." ""
1461-- >>> splitOnSuffixSeq_ "." "."
1464-- >>> splitOnSuffixSeq_ "." "a"
1467-- >>> splitOnSuffixSeq_ "." ".a"
1470-- >>> splitOnSuffixSeq_ "." "a."
1473-- >>> splitOnSuffixSeq_ "." "a.b"
1476-- >>> splitOnSuffixSeq_ "." "a.b."
1479-- >>> splitOnSuffixSeq_ "." "a..b.."
1482-- > lines = splitOnSuffixSeq "\n"
1484-- 'splitOnSuffixSeq' is an inverse of 'intercalateSuffix'. The following law
1487-- > intercalateSuffix . splitOnSuffixSeq == id
1489-- The following law holds when the separator is non-empty and contains none of
1490-- the elements present in the input lists:
1492-- > splitSuffixOn . intercalateSuffix == id
1495{-# INLINE splitOnSuffixSeq #-}
1497 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
1498 => Array a -> Fold m a b -> t m a -> t m b
1499splitOnSuffixSeq patt f m =
1500 D.fromStreamD $ D.splitOnSuffixSeq False patt f (D.toStreamD m)
1503-- | Like 'splitOn' but drops any empty splits.
1505{-# INLINE wordsOn #-}
1507 :: (IsStream t, Monad m, Storable a, Eq a)
1508 => Array a -> Fold m a b -> t m a -> t m b
1509wordsOn subseq f m = undefined -- D.fromStreamD $ D.wordsOn f subseq (D.toStreamD m)
1512-- | Like 'splitOnSuffixSeq' but keeps the suffix intact in the splits.
1514-- > splitWithSuffixSeq'_ pat xs = S.toList $ S.splitWithSuffixSeq (A.fromList pat) (FL.toList) (S.fromList xs)
1516-- >>> splitWithSuffixSeq' "." ""
1519-- >>> splitWithSuffixSeq' "." "."
1522-- >>> splitWithSuffixSeq' "." "a"
1525-- >>> splitWithSuffixSeq' "." ".a"
1528-- >>> splitWithSuffixSeq' "." "a."
1531-- >>> splitWithSuffixSeq' "." "a.b"
1534-- >>> splitWithSuffixSeq' "." "a.b."
1537-- >>> splitWithSuffixSeq' "." "a..b.."
1538-- > ["a.",".","b.","."]
1541{-# INLINE splitWithSuffixSeq #-}
1543 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
1544 => Array a -> Fold m a b -> t m a -> t m b
1545splitWithSuffixSeq patt f m =
1546 D.fromStreamD $ D.splitOnSuffixSeq True patt f (D.toStreamD m)
1549-- This can be implemented easily using Rabin Karp
1550-- | Split post any one of the given patterns.
1551{-# INLINE splitOnSuffixSeqAny #-}
1553 :: (IsStream t, Monad m, Storable a, Integral a)
1554 => [Array a] -> Fold m a b -> t m a -> t m b
1555splitOnSuffixSeqAny subseq f m = undefined
1556 -- D.fromStreamD $ D.splitPostAny f subseq (D.toStreamD m)
1559------------------------------------------------------------------------------
1561------------------------------------------------------------------------------
1563-- | Group the input stream into groups of @n@ elements each and then fold each
1564-- group using the provided fold function.
1566-- >> S.toList $ S.chunksOf 2 FL.sum (S.enumerateFromTo 1 10)
1569-- This can be considered as an n-fold version of 'takeLE' where we apply
1570-- 'takeLE' repeatedly on the leftover stream until the stream exhausts.
1572-- @chunksOf n f = foldMany (FL.takeLE n f)@
1575{-# INLINE chunksOf #-}
1577 :: (IsStream t, Monad m)
1578 => Int -> Fold m a b -> t m a -> t m b
1579chunksOf n f = D.fromStreamD . D.chunksOf n f . D.toStreamD
1584{-# INLINE chunksOf2 #-}
1586 :: (IsStream t, Monad m)
1587 => Int -> m c -> Fold2 m c a b -> t m a -> t m b
1588chunksOf2 n action f m = D.fromStreamD $ D.groupsOf2 n action f (D.toStreamD m)
1590-- | @arraysOf n stream@ groups the elements in the input stream into arrays of
1591-- @n@ elements each.
1593-- Same as the following but may be more efficient:
1595-- > arraysOf n = Stream.foldMany (A.writeN n)
1598{-# INLINE arraysOf #-}
1599arraysOf :: (IsStream t, MonadIO m, Storable a)
1600 => Int -> t m a -> t m (Array a)
1601arraysOf n = D.fromStreamD . A.arraysOf n . D.toStreamD
1603-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
1604-- XXX add this example after fixing the serial stream rate control
1605-- >>> S.toList $ S.take 5 $ intervalsOf 1 FL.sum $ constRate 2 $ S.enumerateFrom 1
1608-- | Group the input stream into windows of @n@ second each and then fold each
1609-- group using the provided fold function.
1612{-# INLINE intervalsOf #-}
1614 :: (IsStream t, MonadAsync m)
1615 => Double -> Fold m a b -> t m a -> t m b
1617 splitWithSuffix isNothing (FL.lcatMaybes f)
1618 (interjectSuffix n (return Nothing) (Serial.map Just xs))
1620------------------------------------------------------------------------------
1621-- Windowed classification
1622------------------------------------------------------------------------------
1624-- We divide the stream into windows or chunks in space or time and each window
1625-- can be associated with a key, all events associated with a particular key in
1626-- the window can be folded to a single result. The stream can be split into
1627-- windows by size or by using a split predicate on the elements in the stream.
1628-- For example, when we receive a closing flag, we can close the window.
1630-- A "chunk" is a space window and a "session" is a time window. Are there any
1631-- other better short words to describe them. An alternative is to use
1632-- "swindow" and "twindow". Another word for "session" could be "spell".
1634-- TODO: To mark the position in space or time we can have Indexed or
1635-- TimeStamped types. That can make it easy to deal with the position indices
1638------------------------------------------------------------------------------
1639-- Keyed Sliding Windows
1640------------------------------------------------------------------------------
1643{-# INLINABLE classifySlidingChunks #-}
1644classifySlidingChunks
1645 :: (IsStream t, MonadAsync m, Ord k)
1646 => Int -- ^ window size
1647 -> Int -- ^ window slide
1648 -> Fold m a b -- ^ Fold to be applied to window events
1649 -> t m (k, a, Bool) -- ^ window key, data, close event
1651classifySlidingChunks wsize wslide (Fold step initial extract) str
1654-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
1655-- slide the send window when an ack is received and we slide the receive
1656-- window when a sequence is complete. Sliding is stateful in case of TCP,
1657-- sliding releases the send buffer or makes data available to the user from
1658-- the receive buffer.
1659{-# INLINABLE classifySlidingSessions #-}
1660classifySlidingSessions
1661 :: (IsStream t, MonadAsync m, Ord k)
1662 => Double -- ^ timer tick in seconds
1663 -> Double -- ^ time window size
1664 -> Double -- ^ window slide
1665 -> Fold m a b -- ^ Fold to be applied to window events
1666 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
1668classifySlidingSessions tick interval slide (Fold step initial extract) str
1672------------------------------------------------------------------------------
1673-- Sliding Window Buffers
1674------------------------------------------------------------------------------
1676-- These buffered versions could be faster than concurrent incremental folds of
1677-- all overlapping windows as in many cases we may not need all the values to
1678-- compute the fold, we can just compute the result using the old value and new
1679-- value. However, we may need the buffer once in a while, for example for
1680-- string search we usually compute the hash incrementally but when the hash
1681-- matches the hash of the pattern we need to compare the whole string.
1683-- XXX we should be able to implement sequence based splitting combinators
1684-- using this combinator.
1687-- | Buffer n elements of the input in a ring buffer. When t new elements are
1688-- collected, slide the window to remove the same number of oldest elements,
1689-- insert the new elements, and apply an incremental fold on the sliding
1690-- window, supplying the outgoing elements, the new ring buffer as arguments.
1692 :: (IsStream t, Monad m, Ord a, Storable a)
1693 => Int -- window size
1694 -> Int -- window slide
1695 -> Fold m (Ring a, Array a) b
1698slidingChunkBuffer = undefined
1700-- Buffer n seconds worth of stream elements of the input in a radix tree.
1701-- Every t seconds, remove the items that are older than n seconds, and apply
1702-- an incremental fold on the sliding window, supplying the outgoing elements,
1703-- and the new radix tree buffer as arguments.
1705 :: (IsStream t, Monad m, Ord a, Storable a)
1706 => Int -- window size
1708 -> Fold m (RTree a, Array a) b
1711slidingSessionBuffer = undefined
1714------------------------------------------------------------------------------
1715-- Keyed Session Windows
1716------------------------------------------------------------------------------
1719-- | Keyed variable size space windows. Close the window if we do not receive a
1720-- window event in the next "spaceout" elements.
1721{-# INLINABLE classifyChunksBy #-}
1723 :: (IsStream t, MonadAsync m, Ord k)
1724 => Int -- ^ window spaceout (spread)
1725 -> Bool -- ^ reset the spaceout when a chunk window element is received
1726 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1727 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1729classifyChunksBy spanout reset (Fold step initial extract) str = undefined
1731-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
1732-- received within the chunk size window. The chunk gets closed only if no
1733-- element is received within the chunk window.
1735{-# INLINABLE classifyKeepAliveChunks #-}
1736classifyKeepAliveChunks
1737 :: (IsStream t, MonadAsync m, Ord k)
1738 => Int -- ^ window spaceout (spread)
1739 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1740 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1742classifyKeepAliveChunks spanout = classifyChunksBy spanout True
1745data SessionState t m k a b = SessionState
1746 { sessionCurTime :: !AbsTime -- ^ time since last event
1747 , sessionEventTime :: !AbsTime -- ^ time as per last event
1748 , sessionCount :: !Int -- ^ total number sessions in progress
1749 , sessionTimerHeap :: H.Heap (H.Entry AbsTime k) -- ^ heap for timeouts
1750 , sessionKeyValueMap :: Map.Map k a -- ^ Stored sessions for keys
1751 , sessionOutputStream :: t (m :: Type -> Type) (k, b) -- ^ Completed sessions
1754-- XXX Perhaps we should use an "Event a" type to represent timestamped data.
1755-- | @classifySessionsBy tick timeout idle pred f stream@ groups timestamped
1756-- events in an input event stream into sessions based on a session key. Each
1757-- element in the input stream is an event consisting of a triple @(session key,
1758-- sesssion data, timestamp)@. @session key@ is a key that uniquely identifies
1759-- the session. All the events belonging to a session are folded using the fold
1760-- @f@ until the fold terminates or a timeout has occurred. The session key and
1761-- the result of the fold are emitted in the output stream when the session is
1764-- When @idle@ is 'False', @timeout@ is the maximum lifetime of a session in
1765-- seconds, measured from the @timestamp@ of the first event in that session.
1766-- When @idle@ is 'True' then the timeout is an idle timeout, it is reset after
1767-- every event received in the session.
1769-- @timestamp@ in an event characterizes the time when the input event was
1770-- generated, this is an absolute time measured from some @Epoch@. The notion
1771-- of current time is maintained by a monotonic event time clock using the
1772-- timestamps seen in the input stream. The latest timestamp seen till now is
1773-- used as the base for the current time. When no new events are seen, a timer
1774-- is started with a tick duration specified by @tick@. This timer is used to
1775-- detect session timeouts in the absence of new events.
1777-- The predicate @pred@ is invoked with the current session count, if it
1778-- returns 'True' a session is ejected from the session cache before inserting
1779-- a new session. This could be useful to alert or eject sessions when the
1780-- number of sessions becomes too high.
1784{-# INLINABLE classifySessionsBy #-}
1786 :: (IsStream t, MonadAsync m, Ord k)
1787 => Double -- ^ timer tick in seconds
1788 -> Double -- ^ session timeout in seconds
1789 -> Bool -- ^ reset the timeout when an event is received
1790 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
1791 -> Fold m a b -- ^ Fold to be applied to session events
1792 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
1793 -> t m (k, b) -- ^ session key, fold result
1794classifySessionsBy tick tmout reset ejectPred
1795 (Fold step initial extract) str =
1796 concatMap sessionOutputStream $
1797 scanlMAfter' sstep (return szero) flush stream
1801 timeoutMs = toRelTime (round (tmout * 1000) :: MilliSecond64)
1802 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
1803 szero = SessionState
1804 { sessionCurTime = toAbsTime (0 :: MilliSecond64)
1805 , sessionEventTime = toAbsTime (0 :: MilliSecond64)
1807 , sessionTimerHeap = H.empty
1808 , sessionKeyValueMap = Map.empty
1809 , sessionOutputStream = K.nil
1812 -- We can eject sessions based on the current session count to limit
1813 -- memory consumption. There are two possible strategies:
1815 -- 1) Eject old sessions or sessions beyond a certain/lower timeout
1816 -- threshold even before timeout, effectively reduce the timeout.
1817 -- 2) Drop creation of new sessions but keep accepting new events for the
1820 -- We use the first strategy as of now.
1822 -- Got a new stream input element
1823 sstep session@SessionState{..} (Just (key, value, timestamp)) = do
1824 -- XXX we should use a heap in pinned memory to scale it to a large
1827 -- To detect session inactivity we keep a timestamp of the latest event
1828 -- in the Map along with the fold result. When we purge the session
1829 -- from the heap we match the timestamp in the heap with the timestamp
1830 -- in the Map, if the latest timestamp is newer and has not expired we
1831 -- reinsert the key in the heap.
1833 -- XXX if the key is an Int, we can also use an IntMap for slightly
1834 -- better performance.
1836 let curTime = max sessionEventTime timestamp
1837 mOld = Map.lookup key sessionKeyValueMap
1839 -- deleting a key from the heap is expensive, so we never
1840 -- delete a key from heap, we just purge it from the Map and it
1841 -- gets purged from the heap on timeout. We just need an extra
1842 -- lookup in the Map when the key is purged from the heap, that
1843 -- should not be expensive.
1845 let (mp, cnt) = case mOld of
1846 Nothing -> (sessionKeyValueMap, sessionCount)
1847 Just _ -> (Map.delete key sessionKeyValueMap
1850 { sessionCurTime = curTime
1851 , sessionEventTime = curTime
1852 , sessionCount = cnt
1853 , sessionKeyValueMap = mp
1854 , sessionOutputStream = yield (key, fb)
1857 let acc = Tuple' timestamp fs1
1858 (hp1, mp1, out1, cnt1) <- do
1859 let vars = (sessionTimerHeap, sessionKeyValueMap,
1860 K.nil, sessionCount)
1862 -- inserting new entry
1864 -- Eject a session from heap and map is needed
1865 eject <- ejectPred sessionCount
1866 (hp, mp, out, cnt) <-
1871 -- Insert the new session in heap
1872 let expiry = addToAbsTime timestamp timeoutMs
1873 hp' = H.insert (Entry expiry key) hp
1874 in return (hp', mp, out, cnt + 1)
1875 -- updating old entry
1876 Just _ -> return vars
1878 let mp2 = Map.insert key acc mp1
1879 return $ SessionState
1880 { sessionCurTime = curTime
1881 , sessionEventTime = curTime
1882 , sessionCount = cnt1
1883 , sessionTimerHeap = hp1
1884 , sessionKeyValueMap = mp2
1885 , sessionOutputStream = out1
1890 Just (Tuple' _ acc) -> return $ FL.Partial acc
1892 FL.Done fb -> done fb
1894 res <- step fs value
1896 FL.Done fb -> done fb
1897 FL.Partial fs1 -> partial fs1
1899 -- Got a timer tick event
1900 sstep sessionState@SessionState{..} Nothing =
1901 let curTime = addToAbsTime sessionCurTime tickMs
1902 in ejectExpired sessionState curTime
1904 flush session@SessionState{..} = do
1905 (hp', mp', out, count) <-
1908 , sessionKeyValueMap
1913 { sessionCount = count
1914 , sessionTimerHeap = hp'
1915 , sessionKeyValueMap = mp'
1916 , sessionOutputStream = out
1919 -- delete from map and output the fold accumulator
1920 ejectEntry hp mp out cnt acc key = do
1922 let out1 = (key, sess) `K.cons` out
1923 let mp1 = Map.delete key mp
1924 return (hp, mp1, out1, cnt - 1)
1926 ejectAll (hp, mp, out, !cnt) = do
1927 let hres = H.uncons hp
1929 Just (Entry _ key, hp1) -> do
1930 r <- case Map.lookup key mp of
1931 Nothing -> return (hp1, mp, out, cnt)
1932 Just (Tuple' _ acc) -> ejectEntry hp1 mp out cnt acc key
1935 assert (Map.null mp) (return ())
1936 return (hp, mp, out, cnt)
1938 ejectOne (hp, mp, out, !cnt) = do
1939 let hres = H.uncons hp
1941 Just (Entry expiry key, hp1) ->
1942 case Map.lookup key mp of
1943 Nothing -> ejectOne (hp1, mp, out, cnt)
1944 Just (Tuple' latestTS acc) -> do
1945 let expiry1 = addToAbsTime latestTS timeoutMs
1946 if not reset || expiry1 <= expiry
1947 then ejectEntry hp1 mp out cnt acc key
1949 -- reset the session timeout and continue
1950 let hp2 = H.insert (Entry expiry1 key) hp1
1951 in ejectOne (hp2, mp, out, cnt)
1953 assert (Map.null mp) (return ())
1954 return (hp, mp, out, cnt)
1956 ejectExpired session@SessionState{..} curTime = do
1957 (hp', mp', out, count) <-
1958 ejectLoop sessionTimerHeap sessionKeyValueMap K.nil sessionCount
1960 { sessionCurTime = curTime
1961 , sessionCount = count
1962 , sessionTimerHeap = hp'
1963 , sessionKeyValueMap = mp'
1964 , sessionOutputStream = out
1969 ejectLoop hp mp out !cnt = do
1970 let hres = H.uncons hp
1972 Just (Entry expiry key, hp1) -> do
1974 if curTime >= expiry
1975 then return (True, False)
1981 case Map.lookup key mp of
1982 Nothing -> ejectLoop hp1 mp out cnt
1983 Just (Tuple' latestTS acc) -> do
1984 let expiry1 = addToAbsTime latestTS timeoutMs
1985 if expiry1 <= curTime || not reset || force
1987 (hp2,mp1,out1,cnt1) <-
1988 ejectEntry hp1 mp out cnt acc key
1989 ejectLoop hp2 mp1 out1 cnt1
1991 -- reset the session timeout and continue
1992 let hp2 = H.insert (Entry expiry1 key) hp1
1993 in ejectLoop hp2 mp out cnt
1994 else return (hp, mp, out, cnt)
1996 assert (Map.null mp) (return ())
1997 return (hp, mp, out, cnt)
1999 -- merge timer events in the stream
2000 stream = Serial.map Just str `Par.parallelFst` repeatM timer
2002 liftIO $ threadDelay (round $ tick * 1000000)
2005-- | Like 'classifySessionsOf' but the session is kept alive if an event is
2006-- received within the session window. The session times out and gets closed
2007-- only if no event is received within the specified session window size.
2009-- If the ejection predicate returns 'True', the session that was idle for
2010-- the longest time is ejected before inserting a new session.
2013-- classifyKeepAliveSessions timeout pred = classifySessionsBy 1 timeout True pred
2018{-# INLINABLE classifyKeepAliveSessions #-}
2019classifyKeepAliveSessions ::
2020 (IsStream t, MonadAsync m, Ord k)
2021 => Double -- ^ session inactive timeout
2022 -> (Int -> m Bool) -- ^ predicate to eject sessions on session count
2023 -> Fold m a b -- ^ Fold to be applied to session payload data
2024 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
2026classifyKeepAliveSessions tmout =
2027 classifySessionsBy 1 tmout True
2029------------------------------------------------------------------------------
2030-- Keyed tumbling windows
2031------------------------------------------------------------------------------
2033-- Tumbling windows is a special case of sliding windows where the window slide
2034-- is the same as the window size. Or it can be a special case of session
2035-- windows where the reset flag is set to False.
2037-- XXX instead of using the early termination flag in the stream, we can use an
2038-- early terminating fold instead.
2041-- | Split the stream into fixed size chunks of specified size. Within each
2042-- such chunk fold the elements in buckets identified by the keys. A particular
2043-- bucket fold can be terminated early if a closing flag is encountered in an
2044-- element for that key.
2047{-# INLINABLE classifyChunksOf #-}
2049 :: (IsStream t, MonadAsync m, Ord k)
2050 => Int -- ^ window size
2051 -> Fold m a b -- ^ Fold to be applied to window events
2052 -> t m (k, a, Bool) -- ^ window key, data, close event
2054classifyChunksOf wsize = classifyChunksBy wsize False
2057-- | Split the stream into fixed size time windows of specified interval in
2058-- seconds. Within each such window, fold the elements in sessions identified
2059-- by the session keys. The fold result is emitted in the output stream if the
2060-- fold returns a 'Left' result or if the time window ends.
2062-- Session @timestamp@ in the input stream is an absolute time from some epoch,
2063-- characterizing the time when the input element was generated. To detect
2064-- session window end, a monotonic event time clock is maintained synced with
2065-- the timestamps with a clock resolution of 1 second.
2067-- If the ejection predicate returns 'True', the session with the longest
2068-- lifetime is ejected before inserting a new session.
2071-- classifySessionsOf interval pred = classifySessionsBy 1 interval False pred
2076-- $ S.classifySessionsOf 3 (const (return False)) (fmap Right FL.toList)
2077-- $ S.map (\(ts,(k,a)) -> (k, a, ts))
2080-- $ (,) <$> S.fromList [1,2,3] <*> S.fromList [1,2,3]
2085{-# INLINABLE classifySessionsOf #-}
2086classifySessionsOf ::
2087 (IsStream t, MonadAsync m, Ord k)
2088 => Double -- ^ time window size
2089 -> (Int -> m Bool) -- ^ predicate to eject sessions on session count
2090 -> Fold m a b -- ^ Fold to be applied to session events
2091 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
2093classifySessionsOf interval =
2094 classifySessionsBy 1 interval False
2096------------------------------------------------------------------------------
2098------------------------------------------------------------------------------
2100-- | @splitInnerBy splitter joiner stream@ splits the inner containers @f a@ of
2101-- an input stream @t m (f a)@ using the @splitter@ function. Container
2102-- elements @f a@ are collected until a split occurs, then all the elements
2103-- before the split are joined using the @joiner@ function.
2105-- For example, if we have a stream of @Array Word8@, we may want to split the
2106-- stream into arrays representing lines separated by '\n' byte such that the
2107-- resulting stream after a split would be one array for each line.
2109-- CAUTION! This is not a true streaming function as the container size after
2110-- the split and merge may not be bounded.
2113{-# INLINE splitInnerBy #-}
2115 :: (IsStream t, Monad m)
2116 => (f a -> m (f a, Maybe (f a))) -- splitter
2117 -> (f a -> f a -> m (f a)) -- joiner
2120splitInnerBy splitter joiner xs =
2121 D.fromStreamD $ D.splitInnerBy splitter joiner $ D.toStreamD xs
2123-- | Like 'splitInnerBy' but splits assuming the separator joins the segment in
2127{-# INLINE splitInnerBySuffix #-}
2129 :: (IsStream t, Monad m, Eq (f a), Monoid (f a))
2130 => (f a -> m (f a, Maybe (f a))) -- splitter
2131 -> (f a -> f a -> m (f a)) -- joiner
2134splitInnerBySuffix splitter joiner xs =
2135 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