2-- Module : Streamly.Internal.Data.Stream.IsStream.Reduce
3-- Copyright : (c) 2017 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
9-- Reduce streams by streams, folds or parsers.
11module Streamly.Internal.Data.Stream.IsStream.Reduce
13 -- * Reduce By Streams
20 -- Reduce a stream by folding or parsing chunks of the stream. Functions
21 -- generally ending in these shapes:
24 -- f (Fold m a b) -> t m a -> t m b
25 -- f (Parser m a b) -> t m a -> t m b
29 -- | Apply folds on a stream.
38 -- | Element unaware grouping.
45 -- | Streams can be sliced into segments in space or in time. We use the
46 -- term @chunk@ to refer to a spatial length of the stream (spatial window)
47 -- and the term @session@ to refer to a length in time (time window).
49 -- -- *** Using Element Separators
59 -- -- *** Splitting By Sequences
65 -- Keeping the delimiters
70 -- Splitting using multiple sequence separators
73 -- , splitOnAnyPrefixSeq
75 -- -- *** Splitting By Streams
76 -- -- | Splitting a stream using another stream as separator.
78 -- ** Keyed Window Classification
80 -- | Split the stream into chunks or windows by position or time. Each
81 -- window can be associated with a key, all events associated with a
82 -- particular key in the window can be folded to a single result. The
83 -- window termination can be dynamically controlled by the fold.
85 -- The term "chunk" is used for a window defined by position of elements
86 -- and the term "session" is used for a time window.
88 -- *** Tumbling Windows
89 -- | A new window starts after the previous window is finished.
92 , classifySessionsByGeneric
96 -- *** Keep Alive Windows
97 -- | The window size is extended if an event arrives within the specified
98 -- window size. This can represent sessions with idle or inactive timeout.
100 -- , classifyKeepAliveChunks
101 , classifyKeepAliveSessions
104 -- *** Sliding Windows
105 -- | A new window starts after the specified slide from the previous
106 -- window. Therefore windows can overlap.
107 , classifySlidingChunks
108 , classifySlidingSessions
109 -- *** Sliding Window Buffers
110 -- , slidingChunkBuffer
111 -- , slidingSessionBuffer
114 -- * Reduce By Parsers
115 -- ** Generic Parsing
116 -- | Apply parsers on a stream.
124 -- In imperative terms, grouped folding can be considered as a nested loop
125 -- where we loop over the stream to group elements and then loop over
126 -- individual groups to fold them to a single value that is yielded in the
129 , wordsBy -- stripAndCompactBy
135 -- -- *** Searching Sequences
136 -- , seqIndices -- search a sequence in the stream
138 -- -- *** Searching Multiple Sequences
139 -- , seqIndicesAny -- search any of the given sequence in the stream
141 -- -- -- ** Searching Streams
142 -- -- | Finding a stream within another stream.
144 -- * Nested splitting
152import Control.Exception (assert)
153import Control.Monad.Catch (MonadThrow)
154import Control.Monad.IO.Class (MonadIO(..))
155import Data.Heap (Entry(..))
156import Data.Kind (Type)
158import Data.Maybe (isNothing)
159import Data.Proxy (Proxy(..))
160import Streamly.Internal.Control.Concurrent (MonadAsync)
161import Streamly.Internal.Data.Fold.Type (Fold (..))
162import Streamly.Internal.Data.IsMap (IsMap(..))
163import Streamly.Internal.Data.Refold.Type (Refold (..))
164import Streamly.Internal.Data.Parser (Parser (..))
165import Streamly.Internal.Data.Array.Unboxed.Type (Array)
166import Streamly.Internal.Data.Stream.IsStream.Common
175import Streamly.Internal.Data.Stream.IsStream.Type
176 (IsStream(..), fromStreamD, toStreamD, cons)
177import Streamly.Internal.Data.Time.Units
178 ( AbsTime, MilliSecond64(..), addToAbsTime, toRelTime
180import Streamly.Internal.Data.Unboxed (Unboxed)
182import qualified Data.Heap as H
183import qualified Streamly.Data.Unfold as Unfold
184import qualified Streamly.Internal.Data.Array.Unboxed.Type as A
186import qualified Streamly.Internal.Data.Fold as FL
187 (Fold, Step(..), takeEndBy_, takeEndBy, catMaybes, take)
188import qualified Streamly.Internal.Data.IsMap as IsMap
189import qualified Streamly.Internal.Data.Parser.ParserD as PRD
190 (Parser(..), fromParserK)
191import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
192import qualified Streamly.Internal.Data.Stream.StreamD as D
207import qualified Streamly.Internal.Data.Stream.IsStream.Expand as Expand
208import qualified Streamly.Internal.Data.Stream.IsStream.Transform as Transform
210import Prelude hiding (concatMap, map)
214-- >>> import Prelude hiding (zipWith, concatMap, concat)
215-- >>> import qualified Streamly.Prelude as Stream
216-- >>> import Streamly.Internal.Data.Stream.IsStream as Stream
217-- >>> import qualified Streamly.Data.Fold as Fold
218-- >>> import qualified Streamly.Internal.Data.Fold as Fold
219-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
220-- >>> import qualified Streamly.Internal.Data.Parser as Parser
221-- >>> import qualified Streamly.Data.Array.Unboxed as Array
223------------------------------------------------------------------------------
225------------------------------------------------------------------------------
227-- | Drop prefix from the input stream if present.
232{-# INLINE dropPrefix #-}
234 -- (Eq a, IsStream t, Monad m) =>
235 t m a -> t m a -> t m a
236dropPrefix = error "Not implemented yet!"
238-- | Drop all matching infix from the input stream if present. Infix stream
239-- may be consumed multiple times.
241-- Space: @O(n)@ where n is the length of the infix.
244{-# INLINE dropInfix #-}
246 -- (Eq a, IsStream t, Monad m) =>
247 t m a -> t m a -> t m a
248dropInfix = error "Not implemented yet!"
250-- | Drop suffix from the input stream if present. Suffix stream may be
251-- consumed multiple times.
253-- Space: @O(n)@ where n is the length of the suffix.
256{-# INLINE dropSuffix #-}
258 -- (Eq a, IsStream t, Monad m) =>
259 t m a -> t m a -> t m a
260dropSuffix = error "Not implemented yet!"
262------------------------------------------------------------------------------
264------------------------------------------------------------------------------
266-- Splitting operations that take a predicate and a Fold can be
267-- expressed using parseMany. Operations like chunksOf, intervalsOf, split*,
268-- can be expressed using parseMany when used with an appropriate Parser.
270-- XXX We need takeGE/takeBetween to implement "some" using "many".
272-- | Apply a 'Fold' repeatedly on a stream and emit the fold outputs in the
275-- To sum every two contiguous elements in a stream:
277-- >>> f = Fold.take 2 Fold.sum
278-- >>> Stream.toList $ Stream.foldMany f $ Stream.fromList [1..10]
281-- On an empty stream the output is empty:
283-- >>> Stream.toList $ Stream.foldMany f $ Stream.fromList []
286-- Note @Stream.foldMany (Fold.take 0)@ would result in an infinite loop in a
291{-# INLINE foldMany #-}
293 :: (IsStream t, Monad m)
297foldMany f m = fromStreamD $ D.foldMany f (toStreamD m)
299-- | Like 'foldMany' but using the 'Refold' type instead of 'Fold'.
302{-# INLINE refoldMany #-}
303refoldMany :: (IsStream t, Monad m) =>
304 Refold m c a b -> m c -> t m a -> t m b
305refoldMany f action = fromStreamD . D.refoldMany f action . toStreamD
307-- | Apply a stream of folds to an input stream and emit the results in the
312{-# INLINE foldSequence #-}
314 :: -- (IsStream t, Monad m) =>
318foldSequence _f _m = undefined
320-- | Iterate a fold generator on a stream. The initial value @b@ is used to
321-- generate the first fold, the fold is applied on the stream and the result of
322-- the fold is used to generate the next fold and so on.
325-- >>> import Data.Monoid (Sum(..))
326-- >>> f x = return (Fold.take 2 (Fold.sconcat x))
327-- >>> s = Stream.map Sum $ Stream.fromList [1..10]
328-- >>> Stream.toList $ Stream.map getSum $ Stream.foldIterateM f (pure 0) s
333-- This is the streaming equivalent of monad like sequenced application of
334-- folds where next fold is dependent on the previous fold.
338{-# INLINE foldIterateM #-}
340 (IsStream t, Monad m) => (b -> m (Fold m a b)) -> m b -> t m a -> t m b
341foldIterateM f i m = fromStreamD $ D.foldIterateM f i (toStreamD m)
343-- | Like 'foldIterateM' but using the 'Refold' type instead. This could be
344-- much more efficient due to stream fusion.
347{-# INLINE refoldIterateM #-}
348refoldIterateM :: (IsStream t, Monad m) =>
349 Refold m b a b -> m b -> t m a -> t m b
350refoldIterateM c i m = fromStreamD $ D.refoldIterateM c i (toStreamD m)
352------------------------------------------------------------------------------
354------------------------------------------------------------------------------
356-- | Apply a 'Parser' repeatedly on a stream and emit the parsed values in the
359-- This is the streaming equivalent of the 'Streamly.Internal.Data.Parser.many'
362-- >>> Stream.toList $ Stream.parseMany (Parser.takeBetween 0 2 Fold.sum) $ Stream.fromList [1..10]
366-- > Stream.toList $ Stream.parseMany (Parser.line Fold.toList) $ Stream.fromList "hello\\nworld"
367-- ["hello\\n","world"]
372-- foldMany f = parseMany (fromFold f)
375-- Known Issues: When the parser fails there is no way to get the remaining
380{-# INLINE parseMany #-}
382 :: (IsStream t, MonadThrow m)
387 fromStreamD $ D.parseMany (PRD.fromParserK p) (toStreamD m)
389-- | Same as parseMany but for StreamD streams.
393{-# INLINE parseManyD #-}
395 :: (IsStream t, MonadThrow m)
400 fromStreamD $ D.parseMany p (toStreamD m)
402-- | Apply a stream of parsers to an input stream and emit the results in the
407{-# INLINE parseSequence #-}
409 :: -- (IsStream t, Monad m) =>
413parseSequence _f _m = undefined
415-- XXX Change the parser arguments' order
417-- | @parseManyTill collect test stream@ tries the parser @test@ on the input,
418-- if @test@ fails it backtracks and tries @collect@, after @collect@ succeeds
419-- @test@ is tried again and so on. The parser stops when @test@ succeeds. The
420-- output of @test@ is discarded and the output of @collect@ is emitted in the
421-- output stream. The parser fails if @collect@ fails.
425{-# INLINE parseManyTill #-}
427 -- (IsStream t, MonadThrow m) =>
432parseManyTill = undefined
434-- | Iterate a parser generating function on a stream. The initial value @b@ is
435-- used to generate the first parser, the parser is applied on the stream and
436-- the result is used to generate the next parser and so on.
438-- >>> import Data.Monoid (Sum(..))
439-- >>> Stream.toList $ Stream.map getSum $ Stream.parseIterate (\b -> Parser.takeBetween 0 2 (Fold.sconcat b)) 0 $ Stream.map Sum $ Stream.fromList [1..10]
442-- This is the streaming equivalent of monad like sequenced application of
443-- parsers where next parser is dependent on the previous parser.
447{-# INLINE parseIterate #-}
449 :: (IsStream t, MonadThrow m)
450 => (b -> Parser m a b)
454parseIterate f i m = fromStreamD $
455 D.parseIterate (PRD.fromParserK . f) i (toStreamD m)
457------------------------------------------------------------------------------
459------------------------------------------------------------------------------
461-- | @groupsBy cmp f $ S.fromList [a,b,c,...]@ assigns the element @a@ to the
462-- first group, if @b \`cmp` a@ is 'True' then @b@ is also assigned to the same
463-- group. If @c \`cmp` a@ is 'True' then @c@ is also assigned to the same
464-- group and so on. When the comparison fails a new group is started. Each
465-- group is folded using the fold @f@ and the result of the fold is emitted in
468-- >>> Stream.toList $ Stream.groupsBy (>) Fold.toList $ Stream.fromList [1,3,7,0,2,5]
472{-# INLINE groupsBy #-}
474 :: (IsStream t, Monad m)
479groupsBy cmp f m = fromStreamD $ D.groupsBy cmp f (toStreamD m)
481-- | Unlike @groupsBy@ this function performs a rolling comparison of two
482-- successive elements in the input stream. @groupsByRolling cmp f $ S.fromList
483-- [a,b,c,...]@ assigns the element @a@ to the first group, if @a \`cmp` b@ is
484-- 'True' then @b@ is also assigned to the same group. If @b \`cmp` c@ is
485-- 'True' then @c@ is also assigned to the same group and so on. When the
486-- comparison fails a new group is started. Each group is folded using the fold
489-- >>> Stream.toList $ Stream.groupsByRolling (\a b -> a + 1 == b) Fold.toList $ Stream.fromList [1,2,3,7,8,9]
493{-# INLINE groupsByRolling #-}
495 :: (IsStream t, Monad m)
500groupsByRolling cmp f m = fromStreamD $ D.groupsRollingBy cmp f (toStreamD m)
503-- > groups = groupsBy (==)
504-- > groups = groupsByRolling (==)
506-- Groups contiguous spans of equal elements together in individual groups.
508-- >>> Stream.toList $ Stream.groups Fold.toList $ Stream.fromList [1,1,2,2]
513groups :: (IsStream t, Monad m, Eq a) => Fold m a b -> t m a -> t m b
514groups = groupsBy (==)
516------------------------------------------------------------------------------
517-- Splitting - by a predicate
518------------------------------------------------------------------------------
520-- In general we can use deintercalate for splitting. Then we can also use
521-- uniqBy to condense the separators. One way to generalize splitting is to
524-- data Segment a b = Empty | Segment b | Separator a
526-- XXX splitOn and splitOnSuffix have a different behavior on an empty stream,
529-- | Split on an infixed separator element, dropping the separator. The
530-- supplied 'Fold' is applied on the split segments. Splits the stream on
531-- separator elements determined by the supplied predicate, separator is
532-- considered as infixed between two segments:
534-- >>> splitOn' p xs = Stream.toList $ Stream.splitOn p Fold.toList (Stream.fromList xs)
535-- >>> splitOn' (== '.') "a.b"
538-- An empty stream is folded to the default value of the fold:
540-- >>> splitOn' (== '.') ""
543-- If one or both sides of the separator are missing then the empty segment on
544-- that side is folded to the default output of the fold:
546-- >>> splitOn' (== '.') "."
549-- >>> splitOn' (== '.') ".a"
552-- >>> splitOn' (== '.') "a."
555-- >>> splitOn' (== '.') "a..b"
558-- splitOn is an inverse of intercalating single element:
560-- > Stream.intercalate (Stream.fromPure '.') Unfold.fromList . Stream.splitOn (== '.') Fold.toList === id
562-- Assuming the input stream does not contain the separator:
564-- > Stream.splitOn (== '.') Fold.toList . Stream.intercalate (Stream.fromPure '.') Unfold.fromList === id
568{-# INLINE splitOn #-}
570 :: (IsStream t, Monad m)
571 => (a -> Bool) -> Fold m a b -> t m a -> t m b
573 -- We can express the infix splitting in terms of optional suffix split
574 -- fold. After applying a suffix split fold repeatedly if the last segment
575 -- ends with a suffix then we need to return the default output of the fold
576 -- after that to make it an infix split.
578 -- Alternately, we can also express it using an optional prefix split fold.
579 -- If the first segment starts with a prefix then we need to emit the
580 -- default output of the fold before that to make it an infix split, and
581 -- then apply prefix split fold repeatedly.
583 -- Since a suffix split fold can be easily expressed using a
584 -- non-backtracking fold, we use that.
585 foldManyPost (FL.takeEndBy_ predicate f)
587-- | Split on a suffixed separator element, dropping the separator. The
588-- supplied 'Fold' is applied on the split segments.
590-- >>> splitOnSuffix' p xs = Stream.toList $ Stream.splitOnSuffix p Fold.toList (Stream.fromList xs)
591-- >>> splitOnSuffix' (== '.') "a.b."
594-- >>> splitOnSuffix' (== '.') "a."
597-- An empty stream results in an empty output stream:
599-- >>> splitOnSuffix' (== '.') ""
602-- An empty segment consisting of only a suffix is folded to the default output
605-- >>> splitOnSuffix' (== '.') "."
608-- >>> splitOnSuffix' (== '.') "a..b.."
611-- A suffix is optional at the end of the stream:
613-- >>> splitOnSuffix' (== '.') "a"
616-- >>> splitOnSuffix' (== '.') ".a"
619-- >>> splitOnSuffix' (== '.') "a.b"
622-- > lines = splitOnSuffix (== '\n')
624-- 'splitOnSuffix' is an inverse of 'intercalateSuffix' with a single element:
626-- > Stream.intercalateSuffix (Stream.fromPure '.') Unfold.fromList . Stream.splitOnSuffix (== '.') Fold.toList === id
628-- Assuming the input stream does not contain the separator:
630-- > Stream.splitOnSuffix (== '.') Fold.toList . Stream.intercalateSuffix (Stream.fromPure '.') Unfold.fromList === id
634{-# INLINE splitOnSuffix #-}
636 :: (IsStream t, Monad m)
637 => (a -> Bool) -> Fold m a b -> t m a -> t m b
638splitOnSuffix predicate f = foldMany (FL.takeEndBy_ predicate f)
640-- | Split on a prefixed separator element, dropping the separator. The
641-- supplied 'Fold' is applied on the split segments.
644-- > splitOnPrefix' p xs = Stream.toList $ Stream.splitOnPrefix p (Fold.toList) (Stream.fromList xs)
645-- > splitOnPrefix' (== '.') ".a.b"
649-- An empty stream results in an empty output stream:
651-- > splitOnPrefix' (== '.') ""
655-- An empty segment consisting of only a prefix is folded to the default output
659-- > splitOnPrefix' (== '.') "."
662-- > splitOnPrefix' (== '.') ".a.b."
665-- > splitOnPrefix' (== '.') ".a..b"
670-- A prefix is optional at the beginning of the stream:
673-- > splitOnPrefix' (== '.') "a"
676-- > splitOnPrefix' (== '.') "a.b"
680-- 'splitOnPrefix' is an inverse of 'intercalatePrefix' with a single element:
682-- > Stream.intercalatePrefix (Stream.fromPure '.') Unfold.fromList . Stream.splitOnPrefix (== '.') Fold.toList === id
684-- Assuming the input stream does not contain the separator:
686-- > Stream.splitOnPrefix (== '.') Fold.toList . Stream.intercalatePrefix (Stream.fromPure '.') Unfold.fromList === id
690{-# INLINE splitOnPrefix #-}
691splitOnPrefix :: -- (IsStream t, MonadCatch m) =>
692 (a -> Bool) -> Fold m a b -> t m a -> t m b
693splitOnPrefix _predicate _f = undefined
694 -- parseMany (Parser.sliceBeginBy predicate f)
696-- | Like 'splitOn' after stripping leading, trailing, and repeated separators.
697-- Therefore, @".a..b."@ with '.' as the separator would be parsed as
698-- @["a","b"]@. In other words, its like parsing words from whitespace
701-- >>> wordsBy' p xs = Stream.toList $ Stream.wordsBy p Fold.toList (Stream.fromList xs)
703-- >>> wordsBy' (== ',') ""
706-- >>> wordsBy' (== ',') ","
709-- >>> wordsBy' (== ',') ",a,,b,"
712-- > words = wordsBy isSpace
716-- It is equivalent to splitting in any of the infix/prefix/suffix styles
717-- followed by removal of empty segments.
718{-# INLINE wordsBy #-}
720 :: (IsStream t, Monad m)
721 => (a -> Bool) -> Fold m a b -> t m a -> t m b
722wordsBy predicate f m =
723 fromStreamD $ D.wordsBy predicate f (toStreamD m)
725-- | Like 'splitOnSuffix' but keeps the suffix attached to the resulting
728-- >>> splitWithSuffix' p xs = Stream.toList $ splitWithSuffix p Fold.toList (Stream.fromList xs)
730-- >>> splitWithSuffix' (== '.') ""
733-- >>> splitWithSuffix' (== '.') "."
736-- >>> splitWithSuffix' (== '.') "a"
739-- >>> splitWithSuffix' (== '.') ".a"
742-- >>> splitWithSuffix' (== '.') "a."
745-- >>> splitWithSuffix' (== '.') "a.b"
748-- >>> splitWithSuffix' (== '.') "a.b."
751-- >>> splitWithSuffix' (== '.') "a..b.."
752-- ["a.",".","b.","."]
756{-# INLINE splitWithSuffix #-}
758 :: (IsStream t, Monad m)
759 => (a -> Bool) -> Fold m a b -> t m a -> t m b
760splitWithSuffix predicate f = foldMany (FL.takeEndBy predicate f)
762------------------------------------------------------------------------------
763-- Splitting - on a delimiter sequence
764------------------------------------------------------------------------------
766-- Int list examples for splitOn:
768-- >>> splitList [] [1,2,3,3,4]
769-- > [[1],[2],[3],[3],[4]]
771-- >>> splitList [5] [1,2,3,3,4]
774-- >>> splitList [1] [1,2,3,3,4]
777-- >>> splitList [4] [1,2,3,3,4]
780-- >>> splitList [2] [1,2,3,3,4]
783-- >>> splitList [3] [1,2,3,3,4]
786-- >>> splitList [3,3] [1,2,3,3,4]
789-- >>> splitList [1,2,3,3,4] [1,2,3,3,4]
792-- This can be implemented easily using Rabin Karp
793-- | Split on any one of the given patterns.
797{-# INLINE splitOnAny #-}
798splitOnAny :: -- (IsStream t, Monad m, Unboxed a, Integral a) =>
799 [Array a] -> Fold m a b -> t m a -> t m b
800splitOnAny _subseq _f _m =
801 undefined -- D.fromStreamD $ D.splitOnAny f subseq (D.toStreamD m)
803-- XXX use a non-monadic intersperse to remove the MonadAsync constraint.
804-- XXX Use two folds, one ring buffer fold for separator sequence and the other
805-- split consumer fold. The input is fed to the ring fold first and the
806-- rejected input is fed to the split fold. If the separator matches, the ring
807-- fold would consume all.
809-- | Like 'splitOnSeq' but splits the separator as well, as an infix token.
811-- >>> splitOn'_ pat xs = Stream.toList $ Stream.splitBySeq (Array.fromList pat) Fold.toList (Stream.fromList xs)
813-- >>> splitOn'_ "" "hello"
814-- ["h","","e","","l","","l","","o"]
816-- >>> splitOn'_ "hello" ""
819-- >>> splitOn'_ "hello" "hello"
822-- >>> splitOn'_ "x" "hello"
825-- >>> splitOn'_ "h" "hello"
828-- >>> splitOn'_ "o" "hello"
831-- >>> splitOn'_ "e" "hello"
834-- >>> splitOn'_ "l" "hello"
835-- ["he","l","","l","o"]
837-- >>> splitOn'_ "ll" "hello"
841{-# INLINE splitBySeq #-}
843 :: (IsStream t, MonadAsync m, Unboxed a, Enum a, Eq a)
844 => Array a -> Fold m a b -> t m a -> t m b
846 intersperseM (fold f (A.toStream patt)) $ splitOnSeq patt f m
848-- | Like 'splitSuffixBy' but the separator is a sequence of elements, instead
849-- of a predicate for a single element.
851-- >>> splitOnSuffixSeq_ pat xs = Stream.toList $ Stream.splitOnSuffixSeq (Array.fromList pat) Fold.toList (Stream.fromList xs)
853-- >>> splitOnSuffixSeq_ "." ""
856-- >>> splitOnSuffixSeq_ "." "."
859-- >>> splitOnSuffixSeq_ "." "a"
862-- >>> splitOnSuffixSeq_ "." ".a"
865-- >>> splitOnSuffixSeq_ "." "a."
868-- >>> splitOnSuffixSeq_ "." "a.b"
871-- >>> splitOnSuffixSeq_ "." "a.b."
874-- >>> splitOnSuffixSeq_ "." "a..b.."
877-- > lines = splitOnSuffixSeq "\n"
879-- 'splitOnSuffixSeq' is an inverse of 'intercalateSuffix'. The following law
882-- > intercalateSuffix . splitOnSuffixSeq == id
884-- The following law holds when the separator is non-empty and contains none of
885-- the elements present in the input lists:
887-- > splitSuffixOn . intercalateSuffix == id
889-- >>> splitOnSuffixSeq pat f = Stream.foldMany (Fold.takeEndBySeq_ pat f)
892{-# INLINE splitOnSuffixSeq #-}
894 :: (IsStream t, MonadIO m, Unboxed a, Enum a, Eq a)
895 => Array a -> Fold m a b -> t m a -> t m b
896splitOnSuffixSeq patt f m =
897 fromStreamD $ D.splitOnSuffixSeq False patt f (toStreamD m)
898-- XXX This may have a problem if the stream terminates and we start extracting
899-- the fold and then the fold terminates before consuming all the buffered
900-- input. We have no way to supply the buffered input back to the driver.
901-- splitOnSuffixSeq patt f =
902-- foldMany (FL.takeEndBySeq_ patt f)
904-- | Like 'splitOn' but drops any empty splits.
907{-# INLINE wordsOn #-}
908wordsOn :: -- (IsStream t, Monad m, Unboxed a, Eq a) =>
909 Array a -> Fold m a b -> t m a -> t m b
910wordsOn _subseq _f _m =
911 undefined -- D.fromStreamD $ D.wordsOn f subseq (D.toStreamD m)
913-- | Like 'splitOnSuffixSeq' but keeps the suffix intact in the splits.
915-- >>> splitWithSuffixSeq' pat xs = Stream.toList $ Stream.splitWithSuffixSeq (Array.fromList pat) Fold.toList (Stream.fromList xs)
917-- >>> splitWithSuffixSeq' "." ""
920-- >>> splitWithSuffixSeq' "." "."
923-- >>> splitWithSuffixSeq' "." "a"
926-- >>> splitWithSuffixSeq' "." ".a"
929-- >>> splitWithSuffixSeq' "." "a."
932-- >>> splitWithSuffixSeq' "." "a.b"
935-- >>> splitWithSuffixSeq' "." "a.b."
938-- >>> splitWithSuffixSeq' "." "a..b.."
939-- ["a.",".","b.","."]
941-- >>> splitWithSuffixSeq pat f = Stream.foldMany (Fold.takeEndBySeq pat f)
944{-# INLINE splitWithSuffixSeq #-}
946 :: (IsStream t, MonadIO m, Unboxed a, Enum a, Eq a)
947 => Array a -> Fold m a b -> t m a -> t m b
948splitWithSuffixSeq patt f m =
949 fromStreamD $ D.splitOnSuffixSeq True patt f (toStreamD m)
950-- splitWithSuffixSeq patt f =
951-- foldMany (FL.takeEndBySeq patt f)
953-- This can be implemented easily using Rabin Karp
954-- | Split post any one of the given patterns.
957{-# INLINE splitOnSuffixSeqAny #-}
958splitOnSuffixSeqAny :: -- (IsStream t, Monad m, Unboxed a, Integral a) =>
959 [Array a] -> Fold m a b -> t m a -> t m b
960splitOnSuffixSeqAny _subseq _f _m = undefined
961 -- D.fromStreamD $ D.splitPostAny f subseq (D.toStreamD m)
963------------------------------------------------------------------------------
965------------------------------------------------------------------------------
967-- | Group the input stream into groups of @n@ elements each and then fold each
968-- group using the provided fold function.
970-- >>> Stream.toList $ Stream.chunksOf 2 Fold.sum (Stream.enumerateFromTo 1 10)
973-- This can be considered as an n-fold version of 'take' where we apply
974-- 'take' repeatedly on the leftover stream until the stream exhausts.
976-- @chunksOf n f = foldMany (FL.take n f)@
979{-# INLINE chunksOf #-}
981 :: (IsStream t, Monad m)
982 => Int -> Fold m a b -> t m a -> t m b
983chunksOf n f = fromStreamD . D.chunksOf n f . toStreamD
985-- | @arraysOf n stream@ groups the elements in the input stream into arrays of
988-- Same as the following but may be more efficient:
990-- > arraysOf n = Stream.foldMany (A.writeN n)
993{-# INLINE arraysOf #-}
994arraysOf :: (IsStream t, MonadIO m, Unboxed a)
995 => Int -> t m a -> t m (Array a)
996arraysOf n = fromStreamD . A.arraysOf n . toStreamD
998-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
999-- XXX add this example after fixing the serial stream rate control
1001-- | Group the input stream into windows of @n@ second each and then fold each
1002-- group using the provided fold function.
1004-- >>> Stream.toList $ Stream.take 5 $ Stream.intervalsOf 1 Fold.sum $ Stream.constRate 2 $ Stream.enumerateFrom 1
1005-- [...,...,...,...,...]
1008{-# INLINE intervalsOf #-}
1010 :: (IsStream t, MonadAsync m)
1011 => Double -> Fold m a b -> t m a -> t m b
1013 splitWithSuffix isNothing (FL.catMaybes f)
1014 (interjectSuffix n (return Nothing) (map Just xs))
1016-- XXX This can be implemented more efficiently by sharing a Clock.Timer across
1017-- parallel threads and resetting it whenever a span is emitted.
1019-- | Like 'chunksOf' but if the chunk is not completed within the specified
1020-- time interval then emit whatever we have collected till now. The chunk
1021-- timeout is reset whenever a chunk is emitted. The granularity of the clock
1024-- >>> s = Stream.delayPost 0.3 $ Stream.fromList [1..1000]
1025-- >>> f = Stream.mapM_ print $ Stream.chunksOfTimeout 5 1 Fold.toList s
1028{-# INLINE chunksOfTimeout #-}
1029chunksOfTimeout :: (IsStream t, MonadAsync m, Functor (t m))
1030 => Int -> Double -> FL.Fold m a b -> t m a -> t m b
1031chunksOfTimeout n timeout f =
1033 . classifySessionsBy
1034 0.1 False (const (return False)) timeout (FL.take n f)
1035 . Transform.timestamped
1038------------------------------------------------------------------------------
1039-- Windowed classification
1040------------------------------------------------------------------------------
1042-- TODO: To mark the position in space or time we can have Indexed or
1043-- TimeStamped types. That can make it easy to deal with the position indices
1046------------------------------------------------------------------------------
1047-- Keyed Sliding Windows
1048------------------------------------------------------------------------------
1051{-# INLINABLE classifySlidingChunks #-}
1052classifySlidingChunks
1053 :: (IsStream t, MonadAsync m, Ord k)
1054 => Int -- ^ window size
1055 -> Int -- ^ window slide
1056 -> Fold m a b -- ^ Fold to be applied to window events
1057 -> t m (k, a, Bool) -- ^ window key, data, close event
1059classifySlidingChunks wsize wslide (Fold step initial extract) str
1062-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
1063-- slide the send window when an ack is received and we slide the receive
1064-- window when a sequence is complete. Sliding is stateful in case of TCP,
1065-- sliding releases the send buffer or makes data available to the user from
1066-- the receive buffer.
1067{-# INLINABLE classifySlidingSessions #-}
1068classifySlidingSessions
1069 :: (IsStream t, MonadAsync m, Ord k)
1070 => Double -- ^ timer tick in seconds
1071 -> Double -- ^ time window size
1072 -> Double -- ^ window slide
1073 -> Fold m a b -- ^ Fold to be applied to window events
1074 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
1076classifySlidingSessions tick interval slide (Fold step initial extract) str
1080------------------------------------------------------------------------------
1081-- Sliding Window Buffers
1082------------------------------------------------------------------------------
1084-- These buffered versions could be faster than concurrent incremental folds of
1085-- all overlapping windows as in many cases we may not need all the values to
1086-- compute the fold, we can just compute the result using the old value and new
1087-- value. However, we may need the buffer once in a while, for example for
1088-- string search we usually compute the hash incrementally but when the hash
1089-- matches the hash of the pattern we need to compare the whole string.
1091-- XXX we should be able to implement sequence based splitting combinators
1092-- using this combinator.
1095-- | Buffer n elements of the input in a ring buffer. When t new elements are
1096-- collected, slide the window to remove the same number of oldest elements,
1097-- insert the new elements, and apply an incremental fold on the sliding
1098-- window, supplying the outgoing elements, the new ring buffer as arguments.
1100 :: (IsStream t, Monad m, Ord a, Unboxed a)
1101 => Int -- window size
1102 -> Int -- window slide
1103 -> Fold m (Ring a, Array a) b
1106slidingChunkBuffer = undefined
1108-- Buffer n seconds worth of stream elements of the input in a radix tree.
1109-- Every t seconds, remove the items that are older than n seconds, and apply
1110-- an incremental fold on the sliding window, supplying the outgoing elements,
1111-- and the new radix tree buffer as arguments.
1113 :: (IsStream t, Monad m, Ord a, Unboxed a)
1114 => Int -- window size
1116 -> Fold m (RTree a, Array a) b
1119slidingSessionBuffer = undefined
1122------------------------------------------------------------------------------
1123-- Keyed Session Windows
1124------------------------------------------------------------------------------
1127-- | Keyed variable size space windows. Close the window if we do not receive a
1128-- window event in the next "spaceout" elements.
1129{-# INLINABLE classifyChunksBy #-}
1131 :: (IsStream t, MonadAsync m, Ord k)
1132 => Int -- ^ window spaceout (spread)
1133 -> Bool -- ^ reset the spaceout when a chunk window element is received
1134 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1135 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1137classifyChunksBy spanout reset (Fold step initial extract) str = undefined
1139-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
1140-- received within the chunk size window. The chunk gets closed only if no
1141-- element is received within the chunk window.
1143{-# INLINABLE classifyKeepAliveChunks #-}
1144classifyKeepAliveChunks
1145 :: (IsStream t, MonadAsync m, Ord k)
1146 => Int -- ^ window spaceout (spread)
1147 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1148 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1150classifyKeepAliveChunks spanout = classifyChunksBy spanout True
1153data SessionState t m f s b = SessionState
1154 { sessionCurTime :: !AbsTime -- ^ time since last event
1155 , sessionEventTime :: !AbsTime -- ^ time as per last event
1156 -- We can use the Map size instead of maintaining a count, but if we have
1157 -- to switch to HashMap then it can be useful.
1158 , sessionCount :: !Int -- ^ total number of sessions in progress
1159 , sessionTimerHeap :: H.Heap (H.Entry AbsTime (Key f)) -- ^ heap for timeouts
1160 , sessionKeyValueMap :: f s -- ^ Stored sessions for keys
1161 , sessionOutputStream :: t (m :: Type -> Type) (Key f, b) -- ^ Completed sessions
1164data SessionEntry s = LiveSession !AbsTime !s | ZombieSession
1166-- delete from map and output the fold accumulator
1167ejectEntry :: (Monad m, IsStream t, IsMap f) =>
1175 -> m (heap, f entry, t m (Key f, b), Int)
1176ejectEntry extract hp mp out cnt acc key = do
1178 let out1 = (key, sess) `cons` out
1179 let mp1 = IsMap.mapDelete key mp
1180 return (hp, mp1, out1, cnt - 1)
1182{-# NOINLINE flush #-}
1183flush :: (Monad m, IsMap f, IsStream t) =>
1185 -> SessionState t m f (SessionEntry s) b
1186 -> m (SessionState t m f (SessionEntry s) b)
1187flush extract session@SessionState{..} = do
1188 (hp', mp', out, count) <-
1191 , sessionKeyValueMap
1196 { sessionCount = count
1197 , sessionTimerHeap = hp'
1198 , sessionKeyValueMap = mp'
1199 , sessionOutputStream = out
1204 ejectAll (hp, mp, out, !cnt) = do
1205 let hres = H.uncons hp
1207 Just (Entry _ key, hp1) -> do
1208 r <- case IsMap.mapLookup key mp of
1209 Nothing -> return (hp1, mp, out, cnt)
1210 Just ZombieSession ->
1211 return (hp1, IsMap.mapDelete key mp, out, cnt)
1212 Just (LiveSession _ acc) ->
1213 ejectEntry extract hp1 mp out cnt acc key
1216 assert (IsMap.mapNull mp) (return ())
1217 return (hp, mp, out, cnt)
1219{-# NOINLINE ejectOne #-}
1220ejectOne :: (IsMap f, IsStream t, Monad m) =>
1223 -> ( H.Heap (Entry AbsTime (Key f))
1224 , f (SessionEntry acc)
1228 -> m ( H.Heap (Entry AbsTime (Key f))
1229 , f (SessionEntry acc)
1230 , t m (Key f, b), Int
1232ejectOne reset extract = go
1236 go (hp, mp, out, !cnt) = do
1237 let hres = H.uncons hp
1239 Just (Entry expiry key, hp1) ->
1240 case IsMap.mapLookup key mp of
1241 Nothing -> go (hp1, mp, out, cnt)
1242 Just ZombieSession ->
1243 go (hp1, IsMap.mapDelete key mp, out, cnt)
1244 Just (LiveSession expiry1 acc) -> do
1245 if not reset || expiry1 <= expiry
1246 then ejectEntry extract hp1 mp out cnt acc key
1248 -- reset the session timeout and continue
1249 let hp2 = H.insert (Entry expiry1 key) hp1
1250 in go (hp2, mp, out, cnt)
1252 assert (IsMap.mapNull mp) (return ())
1253 return (hp, mp, out, cnt)
1255{-# NOINLINE ejectExpired #-}
1256ejectExpired :: (IsMap f, IsStream t, Monad m) =>
1260 -> SessionState t m f (SessionEntry acc) b
1262 -> m (SessionState t m f (SessionEntry acc) b)
1263ejectExpired reset ejectPred extract session@SessionState{..} curTime = do
1264 (hp', mp', out, count) <-
1266 sessionTimerHeap sessionKeyValueMap IsStream.nil sessionCount
1268 { sessionCurTime = curTime
1269 , sessionCount = count
1270 , sessionTimerHeap = hp'
1271 , sessionKeyValueMap = mp'
1272 , sessionOutputStream = out
1277 ejectLoop hp mp out !cnt = do
1278 let hres = H.uncons hp
1280 Just (Entry expiry key, hp1) -> do
1282 if curTime >= expiry
1283 then return (True, False)
1289 case IsMap.mapLookup key mp of
1290 Nothing -> ejectLoop hp1 mp out cnt
1291 Just ZombieSession ->
1292 ejectLoop hp1 (IsMap.mapDelete key mp) out cnt
1293 Just (LiveSession expiry1 acc) -> do
1294 if expiry1 <= curTime || not reset || force
1296 (hp2,mp1,out1,cnt1) <-
1297 ejectEntry extract hp1 mp out cnt acc key
1298 ejectLoop hp2 mp1 out1 cnt1
1300 -- reset the session timeout and continue
1301 let hp2 = H.insert (Entry expiry1 key) hp1
1302 in ejectLoop hp2 mp out cnt
1303 else return (hp, mp, out, cnt)
1305 assert (IsMap.mapNull mp) (return ())
1306 return (hp, mp, out, cnt)
1308-- XXX Use mutable IORef in accumulator
1309{-# INLINE classifySessionsByGeneric #-}
1310classifySessionsByGeneric
1311 :: forall t m f a b. (IsStream t, MonadAsync m, IsMap f)
1312 => Proxy (f :: (Type -> Type))
1313 -> Double -- ^ timer tick in seconds
1314 -> Bool -- ^ reset the timer when an event is received
1315 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
1316 -> Double -- ^ session timeout in seconds
1317 -> Fold m a b -- ^ Fold to be applied to session data
1318 -> t m (AbsTime, (Key f, a)) -- ^ timestamp, (session key, session data)
1319 -> t m (Key f, b) -- ^ session key, fold result
1320classifySessionsByGeneric _ tick reset ejectPred tmout
1321 (Fold step initial extract) input =
1322 Expand.unfoldMany (Unfold.lmap sessionOutputStream Unfold.fromStream)
1323 $ scanlMAfter' sstep (return szero) (flush extract)
1324 $ interjectSuffix tick (return Nothing)
1329 timeoutMs = toRelTime (round (tmout * 1000) :: MilliSecond64)
1330 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
1331 szero = SessionState
1332 { sessionCurTime = toAbsTime (0 :: MilliSecond64)
1333 , sessionEventTime = toAbsTime (0 :: MilliSecond64)
1335 , sessionTimerHeap = H.empty
1336 , sessionKeyValueMap = IsMap.mapEmpty :: f s
1337 , sessionOutputStream = IsStream.nil
1340 -- We can eject sessions based on the current session count to limit
1341 -- memory consumption. There are two possible strategies:
1343 -- 1) Eject old sessions or sessions beyond a certain/lower timeout
1344 -- threshold even before timeout, effectively reduce the timeout.
1345 -- 2) Drop creation of new sessions but keep accepting new events for the
1348 -- We use the first strategy as of now.
1350 -- Got a new stream input element
1351 sstep session@SessionState{..} (Just (timestamp, (key, value))) = do
1352 -- XXX instead of a heap we could use a timer wheel.
1354 -- XXX if the key is an Int, we can also use an IntMap for slightly
1355 -- better performance.
1359 -- Values for each key are collected in a map using the supplied fold.
1360 -- When we insert a key in the Map we insert an entry into the heap as
1361 -- well with the session expiry as the sort key. The Map entry
1362 -- consists of the fold result, and the expiry time of the session. If
1363 -- "reset" is True the expiry time is readjusted whenever a new event
1364 -- is processed. If the fold terminates and a new session is started
1365 -- for the same key the expiry time is set to the first timestamp of
1368 -- The heap must have at most one entry for any given key. The heap is
1369 -- processed periodically to remove the expired entries. We pick up an
1370 -- expired entry from the top of the heap and if the session has
1371 -- expired based on the expiry time in the Map entry then we remove the
1372 -- session from the Map and yield its fold output. Otherwise, we
1373 -- reinsert the entry into the heap based on the current expiry in the
1376 -- If an entry is removed from the Map and not removed from the heap
1377 -- and in the meantime it is inserted again in the Map (using the same
1378 -- key) then how do we avoid inserting multiple entries in the heap?
1379 -- For this reason we maintain the invariant that the Map entry is
1380 -- removed only when the heap entry is removed. Even if the fold has
1381 -- finished we still keep a dummy Map entry (ZombieSession) until the
1382 -- heap entry is removed. That way if we have a Map entry we do not
1383 -- insert a heap entry because we know it is already there.
1384 -- XXX The ZombieSession mechanism does not work as expected as we
1385 -- ignore ZombieSession when inserting a new entry. Anyway, we can
1386 -- remove this mechanism as at most only two heap entries may be
1387 -- created and they will be ultimately cleaned up.
1389 -- Heap processing needs the map and map processing needs the heap,
1390 -- therefore we cannot separate the two for modularity unless we have a
1391 -- way to achieve mutual recursion.
1393 let curTime = max sessionEventTime timestamp
1394 mOld = IsMap.mapLookup key sessionKeyValueMap
1396 -- deleting a key from the heap is expensive, so we never
1397 -- delete a key from heap, we just purge it from the Map and it
1398 -- gets purged from the heap on timeout. We just need an extra
1399 -- lookup in the Map when the key is purged from the heap, that
1400 -- should not be expensive.
1402 let (mp, cnt) = case mOld of
1403 Just (LiveSession _ _) ->
1405 key ZombieSession sessionKeyValueMap
1408 _ -> (sessionKeyValueMap, sessionCount)
1410 { sessionCurTime = curTime
1411 , sessionEventTime = curTime
1412 , sessionCount = cnt
1413 , sessionKeyValueMap = mp
1414 , sessionOutputStream = fromPure (key, fb)
1417 let expiry = addToAbsTime timestamp timeoutMs
1418 (hp1, mp1, out1, cnt1) <- do
1419 let vars = (sessionTimerHeap, sessionKeyValueMap,
1420 IsStream.nil, sessionCount)
1422 -- inserting new entry
1424 -- Eject a session from heap and map if needed
1425 eject <- ejectPred sessionCount
1426 (hp, mp, out, cnt) <-
1428 then ejectOne reset extract vars
1431 -- Insert the new session in heap
1432 let hp' = H.insert (Entry expiry key) hp
1433 in return (hp', mp, out, cnt + 1)
1434 -- updating old entry
1435 Just _ -> return vars
1437 let acc = LiveSession expiry fs1
1438 mp2 = IsMap.mapInsert key acc mp1
1439 return $ SessionState
1440 { sessionCurTime = curTime
1441 , sessionEventTime = curTime
1442 , sessionCount = cnt1
1443 , sessionTimerHeap = hp1
1444 , sessionKeyValueMap = mp2
1445 , sessionOutputStream = out1
1449 Just (LiveSession _ acc) -> return $ FL.Partial acc
1453 error $ "classifySessionsBy: "
1454 ++ "The supplied fold must consume at least one input"
1456 res <- step fs value
1458 FL.Done fb -> done fb
1459 FL.Partial fs1 -> partial fs1
1461 -- Got a timer tick event
1462 sstep sessionState@SessionState{..} Nothing =
1463 let curTime = addToAbsTime sessionCurTime tickMs
1464 in ejectExpired reset ejectPred extract sessionState curTime
1466-- | @classifySessionsBy tick keepalive predicate timeout fold stream@
1467-- classifies an input event @stream@ consisting of @(timestamp, (key,
1468-- value))@ into sessions based on the @key@, folding all the values
1469-- corresponding to the same key into a session using the supplied @fold@.
1471-- When the fold terminates or a @timeout@ occurs, a tuple consisting of the
1472-- session key and the folded value is emitted in the output stream. The
1473-- timeout is measured from the first event in the session. If the @keepalive@
1474-- option is set to 'True' the timeout is reset to 0 whenever an event is
1477-- The @timestamp@ in the input stream is an absolute time from some epoch,
1478-- characterizing the time when the input event was generated. The notion of
1479-- current time is maintained by a monotonic event time clock using the
1480-- timestamps seen in the input stream. The latest timestamp seen till now is
1481-- used as the base for the current time. When no new events are seen, a timer
1482-- is started with a clock resolution of @tick@ seconds. This timer is used to
1483-- detect session timeouts in the absence of new events.
1485-- To ensure an upper bound on the memory used the number of sessions can be
1486-- limited to an upper bound. If the ejection @predicate@ returns 'True', the
1487-- oldest session is ejected before inserting a new session.
1489-- When the stream ends any buffered sessions are ejected immediately.
1491-- If a session key is received even after a session has finished, another
1492-- session is created for that key.
1495-- Stream.mapM_ print
1496-- $ Stream.classifySessionsBy 1 False (const (return False)) 3 (Fold.take 3 Fold.toList)
1497-- $ Stream.timestamped
1498-- $ Stream.delay 0.1
1499-- $ (,) <$> Stream.fromList [1,2,3] <*> Stream.fromList ['a','b','c']
1506{-# INLINE classifySessionsBy #-}
1508 :: (IsStream t, MonadAsync m, Ord k)
1509 => Double -- ^ timer tick in seconds
1510 -> Bool -- ^ reset the timer when an event is received
1511 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
1512 -> Double -- ^ session timeout in seconds
1513 -> Fold m a b -- ^ Fold to be applied to session data
1514 -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
1515 -> t m (k, b) -- ^ session key, fold result
1516classifySessionsBy = classifySessionsByGeneric (Proxy :: Proxy (Map k))
1518-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
1519-- option set to 'True'.
1522-- classifyKeepAliveSessions = classifySessionsBy 1 True
1527{-# INLINE classifyKeepAliveSessions #-}
1528classifyKeepAliveSessions ::
1529 (IsStream t, MonadAsync m, Ord k)
1530 => (Int -> m Bool) -- ^ predicate to eject sessions on session count
1531 -> Double -- ^ session inactive timeout
1532 -> Fold m a b -- ^ Fold to be applied to session payload data
1533 -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
1535classifyKeepAliveSessions = classifySessionsBy 1 True
1537------------------------------------------------------------------------------
1538-- Keyed tumbling windows
1539------------------------------------------------------------------------------
1541-- Tumbling windows is a special case of sliding windows where the window slide
1542-- is the same as the window size. Or it can be a special case of session
1543-- windows where the reset flag is set to False.
1545-- XXX instead of using the early termination flag in the stream, we can use an
1546-- early terminating fold instead.
1549-- | Split the stream into fixed size chunks of specified size. Within each
1550-- such chunk fold the elements in buckets identified by the keys. A particular
1551-- bucket fold can be terminated early if a closing flag is encountered in an
1552-- element for that key.
1555{-# INLINABLE classifyChunksOf #-}
1557 :: (IsStream t, MonadAsync m, Ord k)
1558 => Int -- ^ window size
1559 -> Fold m a b -- ^ Fold to be applied to window events
1560 -> t m (k, a, Bool) -- ^ window key, data, close event
1562classifyChunksOf wsize = classifyChunksBy wsize False
1565-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
1566-- option set to 'False'.
1569-- classifySessionsOf = classifySessionsBy 1 False
1574{-# INLINE classifySessionsOf #-}
1575classifySessionsOf ::
1576 (IsStream t, MonadAsync m, Ord k)
1577 => (Int -> m Bool) -- ^ predicate to eject sessions on session count
1578 -> Double -- ^ time window size
1579 -> Fold m a b -- ^ Fold to be applied to session data
1580 -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
1582classifySessionsOf = classifySessionsBy 1 False
1584------------------------------------------------------------------------------
1586------------------------------------------------------------------------------
1588-- | @splitInnerBy splitter joiner stream@ splits the inner containers @f a@ of
1589-- an input stream @t m (f a)@ using the @splitter@ function. Container
1590-- elements @f a@ are collected until a split occurs, then all the elements
1591-- before the split are joined using the @joiner@ function.
1593-- For example, if we have a stream of @Array Word8@, we may want to split the
1594-- stream into arrays representing lines separated by '\n' byte such that the
1595-- resulting stream after a split would be one array for each line.
1597-- CAUTION! This is not a true streaming function as the container size after
1598-- the split and merge may not be bounded.
1601{-# INLINE splitInnerBy #-}
1603 :: (IsStream t, Monad m)
1604 => (f a -> m (f a, Maybe (f a))) -- splitter
1605 -> (f a -> f a -> m (f a)) -- joiner
1608splitInnerBy splitter joiner xs =
1609 fromStreamD $ D.splitInnerBy splitter joiner $ toStreamD xs
1611-- | Like 'splitInnerBy' but splits assuming the separator joins the segment in
1615{-# INLINE splitInnerBySuffix #-}
1617 :: (IsStream t, Monad m, Eq (f a), Monoid (f a))
1618 => (f a -> m (f a, Maybe (f a))) -- splitter
1619 -> (f a -> f a -> m (f a)) -- joiner
1622splitInnerBySuffix splitter joiner xs =
1623 fromStreamD $ D.splitInnerBySuffix splitter joiner $ 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