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-- | Like 'foldMany' but appends empty fold output if the fold and stream
907-- termination aligns:
909-- >>> f = Fold.takeLE 2 Fold.sum
910-- >>> Stream.toList $ Stream.foldManyPost f $ Stream.fromList []
912-- >>> Stream.toList $ Stream.foldManyPost f $ Stream.fromList [1..9]
914-- >>> Stream.toList $ Stream.foldManyPost f $ Stream.fromList [1..10]
919{-# INLINE foldManyPost #-}
921 :: (IsStream t, Monad m)
925foldManyPost f m = D.fromStreamD $ D.foldManyPost f (D.toStreamD m)
927-- | Apply a 'Fold' repeatedly on a stream and emit the fold outputs in the
930-- To sum every two contiguous elements in a stream:
932-- >>> f = Fold.takeLE 2 Fold.sum
933-- >>> Stream.toList $ Stream.foldMany f $ Stream.fromList [1..10]
936-- On an empty stream the output is empty:
938-- >>> Stream.toList $ Stream.foldMany f $ Stream.fromList []
941-- Note @foldMany (takeLE 0)@ would result in an infinite loop in a non-empty
946{-# INLINE foldMany #-}
948 :: (IsStream t, Monad m)
952foldMany f m = D.fromStreamD $ D.foldMany f (D.toStreamD m)
954-- | Apply a stream of folds to an input stream and emit the results in the
959{-# INLINE foldSequence #-}
961 :: -- (IsStream t, Monad m) =>
965foldSequence _f _m = undefined
967-- | Iterate a fold generator on a stream. The initial value @b@ is used to
968-- generate the first fold, the fold is applied on the stream and the result of
969-- the fold is used to generate the next fold and so on.
971-- >>> f x = Fold.takeLE 2 (Fold.mconcatTo x)
972-- >>> s = Stream.map Sum $ Stream.fromList [1..10]
973-- >>> Stream.toList $ Stream.map getSum $ Stream.foldIterate f 0 s
974-- > [3,10,21,36,55,55]
976-- This is the streaming equivalent of monad like sequenced application of
977-- folds where next fold is dependent on the previous fold.
981{-# INLINE foldIterate #-}
983 :: -- (IsStream t, Monad m) =>
988foldIterate _f _i _m = undefined
989-- D.fromStreamD $ D.foldIterate f i (D.toStreamD m)
991------------------------------------------------------------------------------
993------------------------------------------------------------------------------
995-- | Apply a 'Parser' repeatedly on a stream and emit the parsed values in the
998-- This is the streaming equivalent of the 'Streamly.Internal.Data.Parser.many'
1001-- >>> S.toList $ S.parseMany (PR.take 2 $ PR.fromFold FL.sum) $ S.fromList [1..10]
1004-- >>> S.toList $ S.parseMany (PR.line FL.toList) $ S.fromList "hello\nworld"
1005-- > ["hello\n","world"]
1008-- foldMany f = parseMany (fromFold f)
1011-- Known Issues: When the parser fails there is no way to get the remaining
1016{-# INLINE parseMany #-}
1018 :: (IsStream t, MonadThrow m)
1023 D.fromStreamD $ D.parseMany (PRK.fromParserK p) (D.toStreamD m)
1025{-# INLINE parseManyD #-}
1027 :: (IsStream t, MonadThrow m)
1032 D.fromStreamD $ D.parseMany p (D.toStreamD m)
1034-- | Apply a stream of parsers to an input stream and emit the results in the
1039{-# INLINE parseSequence #-}
1041 :: -- (IsStream t, Monad m) =>
1045parseSequence _f _m = undefined
1047-- | @parseManyTill collect test stream@ tries the parser @test@ on the input,
1048-- if @test@ fails it backtracks and tries @collect@, after @collect@ succeeds
1049-- @test@ is tried again and so on. The parser stops when @test@ succeeds. The
1050-- output of @test@ is discarded and the output of @collect@ is emitted in the
1051-- output stream. The parser fails if @collect@ fails.
1055{-# INLINE parseManyTill #-}
1057 -- (IsStream t, MonadThrow m) =>
1062parseManyTill = undefined
1064-- | Iterate a parser generating function on a stream. The initial value @b@ is
1065-- used to generate the first parser, the parser is applied on the stream and
1066-- the result is used to generate the next parser and so on.
1068-- >>> S.toList $ S.map getSum $ S.parseIterate (\b -> PR.take 2 (FL.mconcatTo b)) 0 $ S.map Sum $ S.fromList [1..10]
1069-- > [3,10,21,36,55,55]
1071-- This is the streaming equivalent of monad like sequenced application of
1072-- parsers where next parser is dependent on the previous parser.
1076{-# INLINE parseIterate #-}
1078 :: (IsStream t, MonadThrow m)
1079 => (b -> Parser m a b)
1083parseIterate f i m = D.fromStreamD $
1084 D.parseIterate (PRK.fromParserK . f) i (D.toStreamD m)
1086------------------------------------------------------------------------------
1088------------------------------------------------------------------------------
1090-- | Drop prefix from the input stream if present.
1094-- /Unimplemented/ - Help wanted.
1095{-# INLINE dropPrefix #-}
1097 -- (Eq a, IsStream t, Monad m) =>
1098 t m a -> t m a -> t m a
1099dropPrefix = error "Not implemented yet!"
1101-- | Drop all matching infix from the input stream if present. Infix stream
1102-- may be consumed multiple times.
1104-- Space: @O(n)@ where n is the length of the infix.
1106-- /Unimplemented/ - Help wanted.
1107{-# INLINE dropInfix #-}
1109 -- (Eq a, IsStream t, Monad m) =>
1110 t m a -> t m a -> t m a
1111dropInfix = error "Not implemented yet!"
1113-- | Drop suffix from the input stream if present. Suffix stream may be
1114-- consumed multiple times.
1116-- Space: @O(n)@ where n is the length of the suffix.
1118-- /Unimplemented/ - Help wanted.
1119{-# INLINE dropSuffix #-}
1121 -- (Eq a, IsStream t, Monad m) =>
1122 t m a -> t m a -> t m a
1123dropSuffix = error "Not implemented yet!"
1125------------------------------------------------------------------------------
1126-- Generalized grouping
1127------------------------------------------------------------------------------
1129-- This combinator is the most general grouping combinator and can be used to
1130-- implement all other grouping combinators.
1132-- XXX check if this can implement the splitOn combinator i.e. we can slide in
1133-- new elements, slide out old elements and incrementally compute the hash.
1134-- Also, can we implement the windowed classification combinators using this?
1136-- In fact this is a parse. Instead of using a special return value in the fold
1137-- we are using a mapping function.
1139-- Note that 'scanl'' (usually followed by a map to extract the desired value
1140-- from the accumulator) can be used to realize many implementations e.g. a
1141-- sliding window implementation. A scan followed by a mapMaybe is also a good
1142-- pattern to express many problems where we want to emit a filtered output and
1143-- not emit an output on every input.
1145-- Passing on of the initial accumulator value to the next fold is equivalent
1146-- to returning the leftover concept.
1149-- | @groupScan splitter fold stream@ folds the input stream using @fold@.
1150-- @splitter@ is applied on the accumulator of the fold every time an item is
1151-- consumed by the fold. The fold continues until @splitter@ returns a 'Just'
1152-- value. A 'Just' result from the @splitter@ specifies a result to be emitted
1153-- in the output stream and the initial value of the accumulator for the next
1154-- group's fold. This allows us to control whether to start fresh for the next
1155-- fold or to continue from the previous fold's output.
1157{-# INLINE groupScan #-}
1159 :: (IsStream t, Monad m)
1160 => (x -> m (Maybe (b, x))) -> Fold m a x -> t m a -> t m b
1161groupScan split fold m = undefined
1164------------------------------------------------------------------------------
1166------------------------------------------------------------------------------
1168-- | @groupsBy cmp f $ S.fromList [a,b,c,...]@ assigns the element @a@ to the
1169-- first group, if @b \`cmp` a@ is 'True' then @b@ is also assigned to the same
1170-- group. If @c \`cmp` a@ is 'True' then @c@ is also assigned to the same
1171-- group and so on. When the comparison fails a new group is started. Each
1172-- group is folded using the fold @f@ and the result of the fold is emitted in
1173-- the output stream.
1175-- >>> S.toList $ S.groupsBy (>) FL.toList $ S.fromList [1,3,7,0,2,5]
1176-- > [[1,3,7],[0,2,5]]
1179{-# INLINE groupsBy #-}
1181 :: (IsStream t, Monad m)
1186groupsBy cmp f m = D.fromStreamD $ D.groupsBy cmp f (D.toStreamD m)
1188-- | Unlike @groupsBy@ this function performs a rolling comparison of two
1189-- successive elements in the input stream. @groupsByRolling cmp f $ S.fromList
1190-- [a,b,c,...]@ assigns the element @a@ to the first group, if @a \`cmp` b@ is
1191-- 'True' then @b@ is also assigned to the same group. If @b \`cmp` c@ is
1192-- 'True' then @c@ is also assigned to the same group and so on. When the
1193-- comparison fails a new group is started. Each group is folded using the fold
1196-- >>> S.toList $ S.groupsByRolling (\a b -> a + 1 == b) FL.toList $ S.fromList [1,2,3,7,8,9]
1197-- > [[1,2,3],[7,8,9]]
1200{-# INLINE groupsByRolling #-}
1202 :: (IsStream t, Monad m)
1207groupsByRolling cmp f m = D.fromStreamD $ D.groupsRollingBy cmp f (D.toStreamD m)
1210-- > groups = groupsBy (==)
1211-- > groups = groupsByRolling (==)
1213-- Groups contiguous spans of equal elements together in individual groups.
1215-- >>> S.toList $ S.groups FL.toList $ S.fromList [1,1,2,2]
1219{-# INLINE groups #-}
1220groups :: (IsStream t, Monad m, Eq a) => Fold m a b -> t m a -> t m b
1221groups = groupsBy (==)
1223------------------------------------------------------------------------------
1224-- Splitting - by a predicate
1225------------------------------------------------------------------------------
1227-- In general we can use deintercalate for splitting. Then we can also use
1228-- uniqBy to condense the separators. One way to generalize splitting is to
1231-- data Segment a b = Empty | Segment b | Separator a
1233-- XXX splitOn and splitOnSuffix have a different behavior on an empty stream,
1234-- is that desirable?
1236-- | Split on an infixed separator element, dropping the separator. The
1237-- supplied 'Fold' is applied on the split segments. Splits the stream on
1238-- separator elements determined by the supplied predicate, separator is
1239-- considered as infixed between two segments:
1241-- >>> splitOn' p xs = S.toList $ S.splitOn p FL.toList (S.fromList xs)
1242-- >>> splitOn' (== '.') "a.b"
1245-- An empty stream is folded to the default value of the fold:
1247-- >>> splitOn' (== '.') ""
1250-- If one or both sides of the separator are missing then the empty segment on
1251-- that side is folded to the default output of the fold:
1253-- >>> splitOn' (== '.') "."
1256-- >>> splitOn' (== '.') ".a"
1259-- >>> splitOn' (== '.') "a."
1262-- >>> splitOn' (== '.') "a..b"
1265-- splitOn is an inverse of intercalating single element:
1267-- @Stream.intercalate (Stream.yield '.') Unfold.fromList . Stream.splitOn (== '.') Fold.toList === id@
1269-- Assuming the input stream does not contain the separator:
1271-- @Stream.splitOn (== '.') Fold.toList . Stream.intercalate (Stream.yield '.') Unfold.fromList === id@
1275{-# INLINE splitOn #-}
1277 :: (IsStream t, Monad m)
1278 => (a -> Bool) -> Fold m a b -> t m a -> t m b
1279splitOn predicate f =
1280 -- We can express the infix splitting in terms of optional suffix split
1281 -- fold. After applying a suffix split fold repeatedly if the last segment
1282 -- ends with a suffix then we need to return the default output of the fold
1283 -- after that to make it an infix split.
1285 -- Alternately, we can also express it using an optional prefix split fold.
1286 -- If the first segment starts with a prefix then we need to emit the
1287 -- default output of the fold before that to make it an infix split, and
1288 -- then apply prefix split fold repeatedly.
1290 -- Since a suffix split fold can be easily expressed using a
1291 -- non-backtracking fold, we use that.
1292 foldManyPost (FL.sliceSepBy predicate f)
1294-- | Split on a suffixed separator element, dropping the separator. The
1295-- supplied 'Fold' is applied on the split segments.
1297-- > splitOnSuffix' p xs = S.toList $ S.splitOnSuffix p (FL.toList) (S.fromList xs)
1298-- >>> splitOnSuffix' (== '.') "a.b."
1301-- >>> splitOnSuffix' (== '.') "a."
1304-- An empty stream results in an empty output stream:
1306-- >>> splitOnSuffix' (== '.') ""
1309-- An empty segment consisting of only a suffix is folded to the default output
1312-- >>> splitOnSuffix' (== '.') "."
1315-- >>> splitOnSuffix' (== '.') "a..b.."
1318-- A suffix is optional at the end of the stream:
1320-- >>> splitOnSuffix' (== '.') "a"
1323-- >>> splitOnSuffix' (== '.') ".a"
1326-- >>> splitOnSuffix' (== '.') "a.b"
1329-- > lines = splitOnSuffix (== '\n')
1331-- 'splitOnSuffix' is an inverse of 'intercalateSuffix' with a single element:
1333-- @Stream.intercalateSuffix (Stream.yield '.') Unfold.fromList . Stream.splitOnSuffix (== '.') Fold.toList === id@
1335-- Assuming the input stream does not contain the separator:
1337-- @Stream.splitOnSuffix (== '.') Fold.toList . Stream.intercalateSuffix (Stream.yield '.') Unfold.fromList === id@
1341{-# INLINE splitOnSuffix #-}
1343 :: (IsStream t, Monad m)
1344 => (a -> Bool) -> Fold m a b -> t m a -> t m b
1345splitOnSuffix predicate f = foldMany (FL.sliceSepBy predicate f)
1347-- | Split on a prefixed separator element, dropping the separator. The
1348-- supplied 'Fold' is applied on the split segments.
1350-- >>> splitOnPrefix' p xs = S.toList $ S.splitOnPrefix p (FL.toList) (S.fromList xs)
1351-- >>> splitOnPrefix' (== '.') ".a.b"
1354-- An empty stream results in an empty output stream:
1355-- >>> splitOnPrefix' (== '.') ""
1358-- An empty segment consisting of only a prefix is folded to the default output
1361-- >>> splitOnPrefix' (== '.') "."
1364-- >>> splitOnPrefix' (== '.') ".a.b."
1367-- >>> splitOnPrefix' (== '.') ".a..b"
1370-- A prefix is optional at the beginning of the stream:
1372-- >>> splitOnPrefix' (== '.') "a"
1375-- >>> splitOnPrefix' (== '.') "a.b"
1378-- 'splitOnPrefix' is an inverse of 'intercalatePrefix' with a single element:
1380-- @Stream.intercalatePrefix (Stream.yield '.') Unfold.fromList . Stream.splitOnPrefix (== '.') Fold.toList === id@
1382-- Assuming the input stream does not contain the separator:
1384-- @Stream.splitOnPrefix (== '.') Fold.toList . Stream.intercalatePrefix (Stream.yield '.') Unfold.fromList === id@
1388{-# INLINE splitOnPrefix #-}
1389splitOnPrefix :: -- (IsStream t, MonadCatch m) =>
1390 (a -> Bool) -> Fold m a b -> t m a -> t m b
1391splitOnPrefix _predicate _f = undefined
1392 -- parseMany (Parser.sliceBeginBy predicate f)
1394-- | Like 'splitOn' after stripping leading, trailing, and repeated separators.
1395-- Therefore, @".a..b."@ with '.' as the separator would be parsed as
1396-- @["a","b"]@. In other words, its like parsing words from whitespace
1399-- > wordsBy' p xs = S.toList $ S.wordsBy p (FL.toList) (S.fromList xs)
1401-- >>> wordsBy' (== ',') ""
1404-- >>> wordsBy' (== ',') ","
1407-- >>> wordsBy' (== ',') ",a,,b,"
1410-- > words = wordsBy isSpace
1414-- It is equivalent to splitting in any of the infix/prefix/suffix styles
1415-- followed by removal of empty segments.
1416{-# INLINE wordsBy #-}
1418 :: (IsStream t, Monad m)
1419 => (a -> Bool) -> Fold m a b -> t m a -> t m b
1420wordsBy predicate f m =
1421 D.fromStreamD $ D.wordsBy predicate f (D.toStreamD m)
1423-- | Like 'splitOnSuffix' but keeps the suffix attached to the resulting
1426-- > splitWithSuffix' p xs = S.toList $ S.splitWithSuffix p (FL.toList) (S.fromList xs)
1428-- >>> splitWithSuffix' (== '.') ""
1431-- >>> splitWithSuffix' (== '.') "."
1434-- >>> splitWithSuffix' (== '.') "a"
1437-- >>> splitWithSuffix' (== '.') ".a"
1440-- >>> splitWithSuffix' (== '.') "a."
1443-- >>> splitWithSuffix' (== '.') "a.b"
1446-- >>> splitWithSuffix' (== '.') "a.b."
1449-- >>> splitWithSuffix' (== '.') "a..b.."
1450-- > ["a.",".","b.","."]
1454{-# INLINE splitWithSuffix #-}
1456 :: (IsStream t, Monad m)
1457 => (a -> Bool) -> Fold m a b -> t m a -> t m b
1458splitWithSuffix predicate f = foldMany (FL.sliceEndWith predicate f)
1460------------------------------------------------------------------------------
1461-- Splitting - on a delimiter sequence
1462------------------------------------------------------------------------------
1464-- Int list examples for splitOn:
1466-- >>> splitList [] [1,2,3,3,4]
1467-- > [[1],[2],[3],[3],[4]]
1469-- >>> splitList [5] [1,2,3,3,4]
1472-- >>> splitList [1] [1,2,3,3,4]
1475-- >>> splitList [4] [1,2,3,3,4]
1478-- >>> splitList [2] [1,2,3,3,4]
1481-- >>> splitList [3] [1,2,3,3,4]
1484-- >>> splitList [3,3] [1,2,3,3,4]
1487-- >>> splitList [1,2,3,3,4] [1,2,3,3,4]
1491-- This can be implemented easily using Rabin Karp
1492-- | Split on any one of the given patterns.
1493{-# INLINE splitOnAny #-}
1495 :: (IsStream t, Monad m, Storable a, Integral a)
1496 => [Array a] -> Fold m a b -> t m a -> t m b
1497splitOnAny subseq f m = undefined -- D.fromStreamD $ D.splitOnAny f subseq (D.toStreamD m)
1500-- XXX use a non-monadic intersperse to remove the MonadAsync constraint.
1501-- XXX Use two folds, one ring buffer fold for separator sequence and the other
1502-- split consumer fold. The input is fed to the ring fold first and the
1503-- rejected input is fed to the split fold. If the separator matches, the ring
1504-- fold would consume all.
1506-- | Like 'splitOnSeq' but splits the separator as well, as an infix token.
1508-- > splitOn'_ pat xs = S.toList $ S.splitOn' (A.fromList pat) (FL.toList) (S.fromList xs)
1510-- >>> splitOn'_ "" "hello"
1511-- > ["h","","e","","l","","l","","o"]
1513-- >>> splitOn'_ "hello" ""
1516-- >>> splitOn'_ "hello" "hello"
1519-- >>> splitOn'_ "x" "hello"
1522-- >>> splitOn'_ "h" "hello"
1525-- >>> splitOn'_ "o" "hello"
1528-- >>> splitOn'_ "e" "hello"
1531-- >>> splitOn'_ "l" "hello"
1532-- > ["he","l","","l","o"]
1534-- >>> splitOn'_ "ll" "hello"
1538{-# INLINE splitBySeq #-}
1540 :: (IsStream t, MonadAsync m, Storable a, Enum a, Eq a)
1541 => Array a -> Fold m a b -> t m a -> t m b
1542splitBySeq patt f m =
1543 intersperseM (fold f (A.toStream patt)) $ splitOnSeq patt f m
1545-- | Like 'splitSuffixBy' but the separator is a sequence of elements, instead
1546-- of a predicate for a single element.
1548-- > splitOnSuffixSeq_ pat xs = S.toList $ S.splitOnSuffixSeq (A.fromList pat) (FL.toList) (S.fromList xs)
1550-- >>> splitOnSuffixSeq_ "." ""
1553-- >>> splitOnSuffixSeq_ "." "."
1556-- >>> splitOnSuffixSeq_ "." "a"
1559-- >>> splitOnSuffixSeq_ "." ".a"
1562-- >>> splitOnSuffixSeq_ "." "a."
1565-- >>> splitOnSuffixSeq_ "." "a.b"
1568-- >>> splitOnSuffixSeq_ "." "a.b."
1571-- >>> splitOnSuffixSeq_ "." "a..b.."
1574-- > lines = splitOnSuffixSeq "\n"
1576-- 'splitOnSuffixSeq' is an inverse of 'intercalateSuffix'. The following law
1579-- > intercalateSuffix . splitOnSuffixSeq == id
1581-- The following law holds when the separator is non-empty and contains none of
1582-- the elements present in the input lists:
1584-- > splitSuffixOn . intercalateSuffix == id
1587{-# INLINE splitOnSuffixSeq #-}
1589 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
1590 => Array a -> Fold m a b -> t m a -> t m b
1591splitOnSuffixSeq patt f m =
1592 D.fromStreamD $ D.splitOnSuffixSeq False patt f (D.toStreamD m)
1595-- | Like 'splitOn' but drops any empty splits.
1597{-# INLINE wordsOn #-}
1599 :: (IsStream t, Monad m, Storable a, Eq a)
1600 => Array a -> Fold m a b -> t m a -> t m b
1601wordsOn subseq f m = undefined -- D.fromStreamD $ D.wordsOn f subseq (D.toStreamD m)
1604-- | Like 'splitOnSuffixSeq' but keeps the suffix intact in the splits.
1606-- > splitWithSuffixSeq'_ pat xs = S.toList $ S.splitWithSuffixSeq (A.fromList pat) (FL.toList) (S.fromList xs)
1608-- >>> splitWithSuffixSeq' "." ""
1611-- >>> splitWithSuffixSeq' "." "."
1614-- >>> splitWithSuffixSeq' "." "a"
1617-- >>> splitWithSuffixSeq' "." ".a"
1620-- >>> splitWithSuffixSeq' "." "a."
1623-- >>> splitWithSuffixSeq' "." "a.b"
1626-- >>> splitWithSuffixSeq' "." "a.b."
1629-- >>> splitWithSuffixSeq' "." "a..b.."
1630-- > ["a.",".","b.","."]
1633{-# INLINE splitWithSuffixSeq #-}
1635 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
1636 => Array a -> Fold m a b -> t m a -> t m b
1637splitWithSuffixSeq patt f m =
1638 D.fromStreamD $ D.splitOnSuffixSeq True patt f (D.toStreamD m)
1641-- This can be implemented easily using Rabin Karp
1642-- | Split post any one of the given patterns.
1643{-# INLINE splitOnSuffixSeqAny #-}
1645 :: (IsStream t, Monad m, Storable a, Integral a)
1646 => [Array a] -> Fold m a b -> t m a -> t m b
1647splitOnSuffixSeqAny subseq f m = undefined
1648 -- D.fromStreamD $ D.splitPostAny f subseq (D.toStreamD m)
1651------------------------------------------------------------------------------
1653------------------------------------------------------------------------------
1655-- | Group the input stream into groups of @n@ elements each and then fold each
1656-- group using the provided fold function.
1658-- >> S.toList $ S.chunksOf 2 FL.sum (S.enumerateFromTo 1 10)
1661-- This can be considered as an n-fold version of 'takeLE' where we apply
1662-- 'takeLE' repeatedly on the leftover stream until the stream exhausts.
1664-- @chunksOf n f = foldMany (FL.takeLE n f)@
1667{-# INLINE chunksOf #-}
1669 :: (IsStream t, Monad m)
1670 => Int -> Fold m a b -> t m a -> t m b
1671chunksOf n f = D.fromStreamD . D.chunksOf n f . D.toStreamD
1676{-# INLINE chunksOf2 #-}
1678 :: (IsStream t, Monad m)
1679 => Int -> m c -> Fold2 m c a b -> t m a -> t m b
1680chunksOf2 n action f m = D.fromStreamD $ D.groupsOf2 n action f (D.toStreamD m)
1682-- | @arraysOf n stream@ groups the elements in the input stream into arrays of
1683-- @n@ elements each.
1685-- Same as the following but may be more efficient:
1687-- > arraysOf n = Stream.foldMany (A.writeN n)
1690{-# INLINE arraysOf #-}
1691arraysOf :: (IsStream t, MonadIO m, Storable a)
1692 => Int -> t m a -> t m (Array a)
1693arraysOf n = D.fromStreamD . A.arraysOf n . D.toStreamD
1695-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
1696-- XXX add this example after fixing the serial stream rate control
1697-- >>> S.toList $ S.take 5 $ intervalsOf 1 FL.sum $ constRate 2 $ S.enumerateFrom 1
1700-- | Group the input stream into windows of @n@ second each and then fold each
1701-- group using the provided fold function.
1704{-# INLINE intervalsOf #-}
1706 :: (IsStream t, MonadAsync m)
1707 => Double -> Fold m a b -> t m a -> t m b
1709 splitWithSuffix isNothing (FL.lcatMaybes f)
1710 (interjectSuffix n (return Nothing) (Serial.map Just xs))
1712------------------------------------------------------------------------------
1713-- Windowed classification
1714------------------------------------------------------------------------------
1716-- We divide the stream into windows or chunks in space or time and each window
1717-- can be associated with a key, all events associated with a particular key in
1718-- the window can be folded to a single result. The stream can be split into
1719-- windows by size or by using a split predicate on the elements in the stream.
1720-- For example, when we receive a closing flag, we can close the window.
1722-- A "chunk" is a space window and a "session" is a time window. Are there any
1723-- other better short words to describe them. An alternative is to use
1724-- "swindow" and "twindow". Another word for "session" could be "spell".
1726-- TODO: To mark the position in space or time we can have Indexed or
1727-- TimeStamped types. That can make it easy to deal with the position indices
1730------------------------------------------------------------------------------
1731-- Keyed Sliding Windows
1732------------------------------------------------------------------------------
1735{-# INLINABLE classifySlidingChunks #-}
1736classifySlidingChunks
1737 :: (IsStream t, MonadAsync m, Ord k)
1738 => Int -- ^ window size
1739 -> Int -- ^ window slide
1740 -> Fold m a b -- ^ Fold to be applied to window events
1741 -> t m (k, a, Bool) -- ^ window key, data, close event
1743classifySlidingChunks wsize wslide (Fold step initial extract) str
1746-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
1747-- slide the send window when an ack is received and we slide the receive
1748-- window when a sequence is complete. Sliding is stateful in case of TCP,
1749-- sliding releases the send buffer or makes data available to the user from
1750-- the receive buffer.
1751{-# INLINABLE classifySlidingSessions #-}
1752classifySlidingSessions
1753 :: (IsStream t, MonadAsync m, Ord k)
1754 => Double -- ^ timer tick in seconds
1755 -> Double -- ^ time window size
1756 -> Double -- ^ window slide
1757 -> Fold m a b -- ^ Fold to be applied to window events
1758 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
1760classifySlidingSessions tick interval slide (Fold step initial extract) str
1764------------------------------------------------------------------------------
1765-- Sliding Window Buffers
1766------------------------------------------------------------------------------
1768-- These buffered versions could be faster than concurrent incremental folds of
1769-- all overlapping windows as in many cases we may not need all the values to
1770-- compute the fold, we can just compute the result using the old value and new
1771-- value. However, we may need the buffer once in a while, for example for
1772-- string search we usually compute the hash incrementally but when the hash
1773-- matches the hash of the pattern we need to compare the whole string.
1775-- XXX we should be able to implement sequence based splitting combinators
1776-- using this combinator.
1779-- | Buffer n elements of the input in a ring buffer. When t new elements are
1780-- collected, slide the window to remove the same number of oldest elements,
1781-- insert the new elements, and apply an incremental fold on the sliding
1782-- window, supplying the outgoing elements, the new ring buffer as arguments.
1784 :: (IsStream t, Monad m, Ord a, Storable a)
1785 => Int -- window size
1786 -> Int -- window slide
1787 -> Fold m (Ring a, Array a) b
1790slidingChunkBuffer = undefined
1792-- Buffer n seconds worth of stream elements of the input in a radix tree.
1793-- Every t seconds, remove the items that are older than n seconds, and apply
1794-- an incremental fold on the sliding window, supplying the outgoing elements,
1795-- and the new radix tree buffer as arguments.
1797 :: (IsStream t, Monad m, Ord a, Storable a)
1798 => Int -- window size
1800 -> Fold m (RTree a, Array a) b
1803slidingSessionBuffer = undefined
1806------------------------------------------------------------------------------
1807-- Keyed Session Windows
1808------------------------------------------------------------------------------
1811-- | Keyed variable size space windows. Close the window if we do not receive a
1812-- window event in the next "spaceout" elements.
1813{-# INLINABLE classifyChunksBy #-}
1815 :: (IsStream t, MonadAsync m, Ord k)
1816 => Int -- ^ window spaceout (spread)
1817 -> Bool -- ^ reset the spaceout when a chunk window element is received
1818 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1819 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1821classifyChunksBy spanout reset (Fold step initial extract) str = undefined
1823-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
1824-- received within the chunk size window. The chunk gets closed only if no
1825-- element is received within the chunk window.
1827{-# INLINABLE classifyKeepAliveChunks #-}
1828classifyKeepAliveChunks
1829 :: (IsStream t, MonadAsync m, Ord k)
1830 => Int -- ^ window spaceout (spread)
1831 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1832 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1834classifyKeepAliveChunks spanout = classifyChunksBy spanout True
1837data SessionState t m k a b = SessionState
1838 { sessionCurTime :: !AbsTime -- ^ time since last event
1839 , sessionEventTime :: !AbsTime -- ^ time as per last event
1840 , sessionCount :: !Int -- ^ total number sessions in progress
1841 , sessionTimerHeap :: H.Heap (H.Entry AbsTime k) -- ^ heap for timeouts
1842 , sessionKeyValueMap :: Map.Map k a -- ^ Stored sessions for keys
1843 , sessionOutputStream :: t (m :: Type -> Type) (k, b) -- ^ Completed sessions
1846-- XXX Perhaps we should use an "Event a" type to represent timestamped data.
1847-- | @classifySessionsBy tick timeout idle pred f stream@ groups timestamped
1848-- events in an input event stream into sessions based on a session key. Each
1849-- element in the input stream is an event consisting of a triple @(session key,
1850-- sesssion data, timestamp)@. @session key@ is a key that uniquely identifies
1851-- the session. All the events belonging to a session are folded using the fold
1852-- @f@ until the fold terminates or a timeout has occurred. The session key and
1853-- the result of the fold are emitted in the output stream when the session is
1856-- When @idle@ is 'False', @timeout@ is the maximum lifetime of a session in
1857-- seconds, measured from the @timestamp@ of the first event in that session.
1858-- When @idle@ is 'True' then the timeout is an idle timeout, it is reset after
1859-- every event received in the session.
1861-- @timestamp@ in an event characterizes the time when the input event was
1862-- generated, this is an absolute time measured from some @Epoch@. The notion
1863-- of current time is maintained by a monotonic event time clock using the
1864-- timestamps seen in the input stream. The latest timestamp seen till now is
1865-- used as the base for the current time. When no new events are seen, a timer
1866-- is started with a tick duration specified by @tick@. This timer is used to
1867-- detect session timeouts in the absence of new events.
1869-- The predicate @pred@ is invoked with the current session count, if it
1870-- returns 'True' a session is ejected from the session cache before inserting
1871-- a new session. This could be useful to alert or eject sessions when the
1872-- number of sessions becomes too high.
1876{-# INLINABLE classifySessionsBy #-}
1878 :: (IsStream t, MonadAsync m, Ord k)
1879 => Double -- ^ timer tick in seconds
1880 -> Double -- ^ session timeout in seconds
1881 -> Bool -- ^ reset the timeout when an event is received
1882 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
1883 -> Fold m a b -- ^ Fold to be applied to session events
1884 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
1885 -> t m (k, b) -- ^ session key, fold result
1886classifySessionsBy tick tmout reset ejectPred
1887 (Fold step initial extract) str =
1888 concatMap sessionOutputStream $
1889 scanlMAfter' sstep (return szero) flush stream
1893 timeoutMs = toRelTime (round (tmout * 1000) :: MilliSecond64)
1894 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
1895 szero = SessionState
1896 { sessionCurTime = toAbsTime (0 :: MilliSecond64)
1897 , sessionEventTime = toAbsTime (0 :: MilliSecond64)
1899 , sessionTimerHeap = H.empty
1900 , sessionKeyValueMap = Map.empty
1901 , sessionOutputStream = K.nil
1904 -- We can eject sessions based on the current session count to limit
1905 -- memory consumption. There are two possible strategies:
1907 -- 1) Eject old sessions or sessions beyond a certain/lower timeout
1908 -- threshold even before timeout, effectively reduce the timeout.
1909 -- 2) Drop creation of new sessions but keep accepting new events for the
1912 -- We use the first strategy as of now.
1914 -- Got a new stream input element
1915 sstep session@SessionState{..} (Just (key, value, timestamp)) = do
1916 -- XXX we should use a heap in pinned memory to scale it to a large
1919 -- To detect session inactivity we keep a timestamp of the latest event
1920 -- in the Map along with the fold result. When we purge the session
1921 -- from the heap we match the timestamp in the heap with the timestamp
1922 -- in the Map, if the latest timestamp is newer and has not expired we
1923 -- reinsert the key in the heap.
1925 -- XXX if the key is an Int, we can also use an IntMap for slightly
1926 -- better performance.
1928 let curTime = max sessionEventTime timestamp
1929 mOld = Map.lookup key sessionKeyValueMap
1931 -- deleting a key from the heap is expensive, so we never
1932 -- delete a key from heap, we just purge it from the Map and it
1933 -- gets purged from the heap on timeout. We just need an extra
1934 -- lookup in the Map when the key is purged from the heap, that
1935 -- should not be expensive.
1937 let (mp, cnt) = case mOld of
1938 Nothing -> (sessionKeyValueMap, sessionCount)
1939 Just _ -> (Map.delete key sessionKeyValueMap
1942 { sessionCurTime = curTime
1943 , sessionEventTime = curTime
1944 , sessionCount = cnt
1945 , sessionKeyValueMap = mp
1946 , sessionOutputStream = yield (key, fb)
1949 let acc = Tuple' timestamp fs1
1950 (hp1, mp1, out1, cnt1) <- do
1951 let vars = (sessionTimerHeap, sessionKeyValueMap,
1952 K.nil, sessionCount)
1954 -- inserting new entry
1956 -- Eject a session from heap and map is needed
1957 eject <- ejectPred sessionCount
1958 (hp, mp, out, cnt) <-
1963 -- Insert the new session in heap
1964 let expiry = addToAbsTime timestamp timeoutMs
1965 hp' = H.insert (Entry expiry key) hp
1966 in return (hp', mp, out, cnt + 1)
1967 -- updating old entry
1968 Just _ -> return vars
1970 let mp2 = Map.insert key acc mp1
1971 return $ SessionState
1972 { sessionCurTime = curTime
1973 , sessionEventTime = curTime
1974 , sessionCount = cnt1
1975 , sessionTimerHeap = hp1
1976 , sessionKeyValueMap = mp2
1977 , sessionOutputStream = out1
1982 Just (Tuple' _ acc) -> return $ FL.Partial acc
1984 FL.Done fb -> done fb
1986 res <- step fs value
1988 FL.Done fb -> done fb
1989 FL.Partial fs1 -> partial fs1
1991 -- Got a timer tick event
1992 sstep sessionState@SessionState{..} Nothing =
1993 let curTime = addToAbsTime sessionCurTime tickMs
1994 in ejectExpired sessionState curTime
1996 flush session@SessionState{..} = do
1997 (hp', mp', out, count) <-
2000 , sessionKeyValueMap
2005 { sessionCount = count
2006 , sessionTimerHeap = hp'
2007 , sessionKeyValueMap = mp'
2008 , sessionOutputStream = out
2011 -- delete from map and output the fold accumulator
2012 ejectEntry hp mp out cnt acc key = do
2014 let out1 = (key, sess) `K.cons` out
2015 let mp1 = Map.delete key mp
2016 return (hp, mp1, out1, cnt - 1)
2018 ejectAll (hp, mp, out, !cnt) = do
2019 let hres = H.uncons hp
2021 Just (Entry _ key, hp1) -> do
2022 r <- case Map.lookup key mp of
2023 Nothing -> return (hp1, mp, out, cnt)
2024 Just (Tuple' _ acc) -> ejectEntry hp1 mp out cnt acc key
2027 assert (Map.null mp) (return ())
2028 return (hp, mp, out, cnt)
2030 ejectOne (hp, mp, out, !cnt) = do
2031 let hres = H.uncons hp
2033 Just (Entry expiry key, hp1) ->
2034 case Map.lookup key mp of
2035 Nothing -> ejectOne (hp1, mp, out, cnt)
2036 Just (Tuple' latestTS acc) -> do
2037 let expiry1 = addToAbsTime latestTS timeoutMs
2038 if not reset || expiry1 <= expiry
2039 then ejectEntry hp1 mp out cnt acc key
2041 -- reset the session timeout and continue
2042 let hp2 = H.insert (Entry expiry1 key) hp1
2043 in ejectOne (hp2, mp, out, cnt)
2045 assert (Map.null mp) (return ())
2046 return (hp, mp, out, cnt)
2048 ejectExpired session@SessionState{..} curTime = do
2049 (hp', mp', out, count) <-
2050 ejectLoop sessionTimerHeap sessionKeyValueMap K.nil sessionCount
2052 { sessionCurTime = curTime
2053 , sessionCount = count
2054 , sessionTimerHeap = hp'
2055 , sessionKeyValueMap = mp'
2056 , sessionOutputStream = out
2061 ejectLoop hp mp out !cnt = do
2062 let hres = H.uncons hp
2064 Just (Entry expiry key, hp1) -> do
2066 if curTime >= expiry
2067 then return (True, False)
2073 case Map.lookup key mp of
2074 Nothing -> ejectLoop hp1 mp out cnt
2075 Just (Tuple' latestTS acc) -> do
2076 let expiry1 = addToAbsTime latestTS timeoutMs
2077 if expiry1 <= curTime || not reset || force
2079 (hp2,mp1,out1,cnt1) <-
2080 ejectEntry hp1 mp out cnt acc key
2081 ejectLoop hp2 mp1 out1 cnt1
2083 -- reset the session timeout and continue
2084 let hp2 = H.insert (Entry expiry1 key) hp1
2085 in ejectLoop hp2 mp out cnt
2086 else return (hp, mp, out, cnt)
2088 assert (Map.null mp) (return ())
2089 return (hp, mp, out, cnt)
2091 -- merge timer events in the stream
2092 stream = Serial.map Just str `Par.parallelFst` repeatM timer
2094 liftIO $ threadDelay (round $ tick * 1000000)
2097-- | Like 'classifySessionsOf' but the session is kept alive if an event is
2098-- received within the session window. The session times out and gets closed
2099-- only if no event is received within the specified session window size.
2101-- If the ejection predicate returns 'True', the session that was idle for
2102-- the longest time is ejected before inserting a new session.
2105-- classifyKeepAliveSessions timeout pred = classifySessionsBy 1 timeout True pred
2110{-# INLINABLE classifyKeepAliveSessions #-}
2111classifyKeepAliveSessions ::
2112 (IsStream t, MonadAsync m, Ord k)
2113 => Double -- ^ session inactive timeout
2114 -> (Int -> m Bool) -- ^ predicate to eject sessions on session count
2115 -> Fold m a b -- ^ Fold to be applied to session payload data
2116 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
2118classifyKeepAliveSessions tmout =
2119 classifySessionsBy 1 tmout True
2121------------------------------------------------------------------------------
2122-- Keyed tumbling windows
2123------------------------------------------------------------------------------
2125-- Tumbling windows is a special case of sliding windows where the window slide
2126-- is the same as the window size. Or it can be a special case of session
2127-- windows where the reset flag is set to False.
2129-- XXX instead of using the early termination flag in the stream, we can use an
2130-- early terminating fold instead.
2133-- | Split the stream into fixed size chunks of specified size. Within each
2134-- such chunk fold the elements in buckets identified by the keys. A particular
2135-- bucket fold can be terminated early if a closing flag is encountered in an
2136-- element for that key.
2139{-# INLINABLE classifyChunksOf #-}
2141 :: (IsStream t, MonadAsync m, Ord k)
2142 => Int -- ^ window size
2143 -> Fold m a b -- ^ Fold to be applied to window events
2144 -> t m (k, a, Bool) -- ^ window key, data, close event
2146classifyChunksOf wsize = classifyChunksBy wsize False
2149-- | Split the stream into fixed size time windows of specified interval in
2150-- seconds. Within each such window, fold the elements in sessions identified
2151-- by the session keys. The fold result is emitted in the output stream if the
2152-- fold returns a 'Left' result or if the time window ends.
2154-- Session @timestamp@ in the input stream is an absolute time from some epoch,
2155-- characterizing the time when the input element was generated. To detect
2156-- session window end, a monotonic event time clock is maintained synced with
2157-- the timestamps with a clock resolution of 1 second.
2159-- If the ejection predicate returns 'True', the session with the longest
2160-- lifetime is ejected before inserting a new session.
2163-- classifySessionsOf interval pred = classifySessionsBy 1 interval False pred
2168-- $ S.classifySessionsOf 3 (const (return False)) (fmap Right FL.toList)
2169-- $ S.map (\(ts,(k,a)) -> (k, a, ts))
2172-- $ (,) <$> S.fromList [1,2,3] <*> S.fromList [1,2,3]
2177{-# INLINABLE classifySessionsOf #-}
2178classifySessionsOf ::
2179 (IsStream t, MonadAsync m, Ord k)
2180 => Double -- ^ time window size
2181 -> (Int -> m Bool) -- ^ predicate to eject sessions on session count
2182 -> Fold m a b -- ^ Fold to be applied to session events
2183 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
2185classifySessionsOf interval =
2186 classifySessionsBy 1 interval False
2188------------------------------------------------------------------------------
2190------------------------------------------------------------------------------
2192-- | @splitInnerBy splitter joiner stream@ splits the inner containers @f a@ of
2193-- an input stream @t m (f a)@ using the @splitter@ function. Container
2194-- elements @f a@ are collected until a split occurs, then all the elements
2195-- before the split are joined using the @joiner@ function.
2197-- For example, if we have a stream of @Array Word8@, we may want to split the
2198-- stream into arrays representing lines separated by '\n' byte such that the
2199-- resulting stream after a split would be one array for each line.
2201-- CAUTION! This is not a true streaming function as the container size after
2202-- the split and merge may not be bounded.
2205{-# INLINE splitInnerBy #-}
2207 :: (IsStream t, Monad m)
2208 => (f a -> m (f a, Maybe (f a))) -- splitter
2209 -> (f a -> f a -> m (f a)) -- joiner
2212splitInnerBy splitter joiner xs =
2213 D.fromStreamD $ D.splitInnerBy splitter joiner $ D.toStreamD xs
2215-- | Like 'splitInnerBy' but splits assuming the separator joins the segment in
2219{-# INLINE splitInnerBySuffix #-}
2221 :: (IsStream t, Monad m, Eq (f a), Monoid (f a))
2222 => (f a -> m (f a, Maybe (f a))) -- splitter
2223 -> (f a -> f a -> m (f a)) -- joiner
2226splitInnerBySuffix splitter joiner xs =
2227 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