1{-# OPTIONS_GHC -Wno-deprecations #-}
4-- Module : Streamly.Internal.Data.Stream.IsStream.Reduce
5-- Copyright : (c) 2017 Composewell Technologies
6-- License : BSD-3-Clause
7-- Maintainer : streamly@composewell.com
8-- Stability : experimental
11-- Reduce streams by streams, folds or parsers.
13module Streamly.Internal.Data.Stream.IsStream.Reduce {-# DEPRECATED "Please use \"Streamly.Data.Stream.*\" instead." #-}
15 -- * Reduce By Streams
22 -- Reduce a stream by folding or parsing chunks of the stream. Functions
23 -- generally ending in these shapes:
26 -- f (Fold m a b) -> t m a -> t m b
27 -- f (Parser a m b) -> t m a -> t m b
31 -- | Apply folds on a stream.
40 -- | Element unaware grouping.
47 -- | Streams can be sliced into segments in space or in time. We use the
48 -- term @chunk@ to refer to a spatial length of the stream (spatial window)
49 -- and the term @session@ to refer to a length in time (time window).
51 -- -- *** Using Element Separators
61 -- -- *** Splitting By Sequences
67 -- Keeping the delimiters
72 -- Splitting using multiple sequence separators
75 -- , splitOnAnyPrefixSeq
77 -- -- *** Splitting By Streams
78 -- -- | Splitting a stream using another stream as separator.
80 -- ** Keyed Window Classification
82 -- | Split the stream into chunks or windows by position or time. Each
83 -- window can be associated with a key, all events associated with a
84 -- particular key in the window can be folded to a single result. The
85 -- window termination can be dynamically controlled by the fold.
87 -- The term "chunk" is used for a window defined by position of elements
88 -- and the term "session" is used for a time window.
90 -- *** Tumbling Windows
91 -- | A new window starts after the previous window is finished.
94 , classifySessionsByGeneric
98 -- *** Keep Alive Windows
99 -- | The window size is extended if an event arrives within the specified
100 -- window size. This can represent sessions with idle or inactive timeout.
102 -- , classifyKeepAliveChunks
103 , classifyKeepAliveSessions
106 -- *** Sliding Windows
107 -- | A new window starts after the specified slide from the previous
108 -- window. Therefore windows can overlap.
109 , classifySlidingChunks
110 , classifySlidingSessions
111 -- *** Sliding Window Buffers
112 -- , slidingChunkBuffer
113 -- , slidingSessionBuffer
116 -- * Reduce By Parsers
117 -- ** Generic Parsing
118 -- | Apply parsers on a stream.
126 -- In imperative terms, grouped folding can be considered as a nested loop
127 -- where we loop over the stream to group elements and then loop over
128 -- individual groups to fold them to a single value that is yielded in the
131 , wordsBy -- stripAndCompactBy
137 -- -- *** Searching Sequences
138 -- , seqIndices -- search a sequence in the stream
140 -- -- *** Searching Multiple Sequences
141 -- , seqIndicesAny -- search any of the given sequence in the stream
143 -- -- -- ** Searching Streams
144 -- -- | Finding a stream within another stream.
146 -- * Nested splitting
154import Control.Exception (assert)
155import Control.Monad.IO.Class (MonadIO(..))
156import Data.Heap (Entry(..))
157import Data.Kind (Type)
159import Data.Maybe (isNothing)
160import Data.Proxy (Proxy(..))
161import Streamly.Internal.Control.Concurrent (MonadAsync)
162import Streamly.Internal.Data.Fold.Type (Fold (..))
163import Streamly.Internal.Data.IsMap (IsMap(..))
164import Streamly.Internal.Data.Refold.Type (Refold (..))
165import Streamly.Internal.Data.Parser (Parser (..), ParseError)
166import Streamly.Internal.Data.Array.Type (Array)
167import Streamly.Internal.Data.Stream.IsStream.Common
176import Streamly.Internal.Data.Stream.IsStream.Type
177 (IsStream(..), fromStreamD, toStreamD, cons)
178import Streamly.Internal.Data.Time.Units
179 ( AbsTime, MilliSecond64(..), addToAbsTime, toRelTime
181import Streamly.Internal.Data.Unboxed (Unbox)
183import qualified Data.Heap as H
184import qualified Streamly.Data.Unfold as Unfold
185import qualified Streamly.Internal.Data.Array.Type as A
187import qualified Streamly.Internal.Data.Fold as FL
188 (Fold, Step(..), takeEndBy_, takeEndBy, catMaybes, take)
189import qualified Streamly.Internal.Data.IsMap as IsMap
190import qualified Streamly.Internal.Data.Parser.ParserD as PRD
191 (Parser(..), fromParserK)
192import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
193import qualified Streamly.Internal.Data.Stream.StreamD as D
208import qualified Streamly.Internal.Data.Stream.IsStream.Expand as Expand
209import qualified Streamly.Internal.Data.Stream.IsStream.Transform as Transform
211import Prelude hiding (concatMap, map)
215-- >>> :set -fno-warn-deprecations
216-- >>> import Prelude hiding (zipWith, concatMap, concat)
217-- >>> import qualified Streamly.Prelude as Stream
218-- >>> import Streamly.Internal.Data.Stream.IsStream as Stream
219-- >>> import qualified Streamly.Data.Fold as Fold
220-- >>> import qualified Streamly.Internal.Data.Fold as Fold
221-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
222-- >>> import qualified Streamly.Internal.Data.Parser as Parser
223-- >>> import qualified Streamly.Data.Array as Array
225------------------------------------------------------------------------------
227------------------------------------------------------------------------------
229-- | Drop prefix from the input stream if present.
234{-# INLINE dropPrefix #-}
236 -- (Eq a, IsStream t, Monad m) =>
237 t m a -> t m a -> t m a
238dropPrefix = error "Not implemented yet!"
240-- | Drop all matching infix from the input stream if present. Infix stream
241-- may be consumed multiple times.
243-- Space: @O(n)@ where n is the length of the infix.
246{-# INLINE dropInfix #-}
248 -- (Eq a, IsStream t, Monad m) =>
249 t m a -> t m a -> t m a
250dropInfix = error "Not implemented yet!"
252-- | Drop suffix from the input stream if present. Suffix stream may be
253-- consumed multiple times.
255-- Space: @O(n)@ where n is the length of the suffix.
258{-# INLINE dropSuffix #-}
260 -- (Eq a, IsStream t, Monad m) =>
261 t m a -> t m a -> t m a
262dropSuffix = error "Not implemented yet!"
264------------------------------------------------------------------------------
266------------------------------------------------------------------------------
268-- Splitting operations that take a predicate and a Fold can be
269-- expressed using parseMany. Operations like chunksOf, intervalsOf, split*,
270-- can be expressed using parseMany when used with an appropriate Parser.
272-- XXX We need takeGE/takeBetween to implement "some" using "many".
274-- | Apply a 'Fold' repeatedly on a stream and emit the fold outputs in the
277-- To sum every two contiguous elements in a stream:
279-- >>> f = Fold.take 2 Fold.sum
280-- >>> Stream.toList $ Stream.foldMany f $ Stream.fromList [1..10]
283-- On an empty stream the output is empty:
285-- >>> Stream.toList $ Stream.foldMany f $ Stream.fromList []
288-- Note @Stream.foldMany (Fold.take 0)@ would result in an infinite loop in a
293{-# INLINE foldMany #-}
295 :: (IsStream t, Monad m)
299foldMany f m = fromStreamD $ D.foldMany f (toStreamD m)
301-- | Like 'foldMany' but using the 'Refold' type instead of 'Fold'.
304{-# INLINE refoldMany #-}
305refoldMany :: (IsStream t, Monad m) =>
306 Refold m c a b -> m c -> t m a -> t m b
307refoldMany f action = fromStreamD . D.refoldMany f action . toStreamD
309-- | Apply a stream of folds to an input stream and emit the results in the
314{-# INLINE foldSequence #-}
316 :: -- (IsStream t, Monad m) =>
320foldSequence _f _m = undefined
322-- | Iterate a fold generator on a stream. The initial value @b@ is used to
323-- generate the first fold, the fold is applied on the stream and the result of
324-- the fold is used to generate the next fold and so on.
327-- >>> import Data.Monoid (Sum(..))
328-- >>> f x = return (Fold.take 2 (Fold.sconcat x))
329-- >>> s = Stream.map Sum $ Stream.fromList [1..10]
330-- >>> Stream.toList $ Stream.map getSum $ Stream.foldIterateM f (pure 0) s
335-- This is the streaming equivalent of monad like sequenced application of
336-- folds where next fold is dependent on the previous fold.
340{-# INLINE foldIterateM #-}
342 (IsStream t, Monad m) => (b -> m (Fold m a b)) -> m b -> t m a -> t m b
343foldIterateM f i m = fromStreamD $ D.foldIterateM f i (toStreamD m)
345-- | Like 'foldIterateM' but using the 'Refold' type instead. This could be
346-- much more efficient due to stream fusion.
349{-# INLINE refoldIterateM #-}
350refoldIterateM :: (IsStream t, Monad m) =>
351 Refold m b a b -> m b -> t m a -> t m b
352refoldIterateM c i m = fromStreamD $ D.refoldIterateM c i (toStreamD m)
354------------------------------------------------------------------------------
356------------------------------------------------------------------------------
358-- | Apply a 'Parser' repeatedly on a stream and emit the parsed values in the
361-- This is the streaming equivalent of the 'Streamly.Internal.Data.Parser.many'
364-- >>> Stream.toList $ Stream.parseMany (Parser.takeBetween 0 2 Fold.sum) $ Stream.fromList [1..10]
365-- [Right 3,Right 7,Right 11,Right 15,Right 19]
368-- > Stream.toList $ Stream.parseMany (Parser.line Fold.toList) $ Stream.fromList "hello\\nworld"
369-- ["hello\\n","world"]
374-- foldMany f = parseMany (fromFold f)
377-- Known Issues: When the parser fails there is no way to get the remaining
382{-# INLINE parseMany #-}
384 :: (IsStream t, Monad m)
387 -> t m (Either ParseError b)
389 fromStreamD $ D.parseMany (PRD.fromParserK p) (toStreamD m)
391-- | Same as parseMany but for StreamD streams.
395{-# INLINE parseManyD #-}
397 :: (IsStream t, Monad m)
400 -> t m (Either ParseError b)
402 fromStreamD $ D.parseMany p (toStreamD m)
404-- | Apply a stream of parsers to an input stream and emit the results in the
409{-# INLINE parseSequence #-}
411 :: -- (IsStream t, Monad m) =>
415parseSequence _f _m = undefined
417-- XXX Change the parser arguments' order
419-- | @parseManyTill collect test stream@ tries the parser @test@ on the input,
420-- if @test@ fails it backtracks and tries @collect@, after @collect@ succeeds
421-- @test@ is tried again and so on. The parser stops when @test@ succeeds. The
422-- output of @test@ is discarded and the output of @collect@ is emitted in the
423-- output stream. The parser fails if @collect@ fails.
427{-# INLINE parseManyTill #-}
429 -- (IsStream t, MonadThrow m) =>
434parseManyTill = undefined
436-- | Iterate a parser generating function on a stream. The initial value @b@ is
437-- used to generate the first parser, the parser is applied on the stream and
438-- the result is used to generate the next parser and so on.
440-- >>> import Data.Monoid (Sum(..))
441-- >>> Stream.toList $ fmap getSum $ Stream.rights $ Stream.parseIterate (\b -> Parser.takeBetween 0 2 (Fold.sconcat b)) (Sum 0) $ fmap Sum $ Stream.fromList [1..10]
444-- This is the streaming equivalent of monad like sequenced application of
445-- parsers where next parser is dependent on the previous parser.
449{-# INLINE parseIterate #-}
451 :: (IsStream t, Monad m)
452 => (b -> Parser a m b)
455 -> t m (Either ParseError b)
456parseIterate f i m = fromStreamD $
457 D.parseIterate (PRD.fromParserK . f) i (toStreamD m)
459------------------------------------------------------------------------------
461------------------------------------------------------------------------------
463-- | @groupsBy cmp f $ S.fromList [a,b,c,...]@ assigns the element @a@ to the
464-- first group, if @b \`cmp` a@ is 'True' then @b@ is also assigned to the same
465-- group. If @c \`cmp` a@ is 'True' then @c@ is also assigned to the same
466-- group and so on. When the comparison fails a new group is started. Each
467-- group is folded using the fold @f@ and the result of the fold is emitted in
470-- >>> Stream.toList $ Stream.groupsBy (>) Fold.toList $ Stream.fromList [1,3,7,0,2,5]
474{-# INLINE groupsBy #-}
476 :: (IsStream t, Monad m)
481groupsBy cmp f m = fromStreamD $ D.groupsBy cmp f (toStreamD m)
483-- | Unlike @groupsBy@ this function performs a rolling comparison of two
484-- successive elements in the input stream. @groupsByRolling cmp f $ S.fromList
485-- [a,b,c,...]@ assigns the element @a@ to the first group, if @a \`cmp` b@ is
486-- 'True' then @b@ is also assigned to the same group. If @b \`cmp` c@ is
487-- 'True' then @c@ is also assigned to the same group and so on. When the
488-- comparison fails a new group is started. Each group is folded using the fold
491-- >>> Stream.toList $ Stream.groupsByRolling (\a b -> a + 1 == b) Fold.toList $ Stream.fromList [1,2,3,7,8,9]
495{-# INLINE groupsByRolling #-}
497 :: (IsStream t, Monad m)
502groupsByRolling cmp f m = fromStreamD $ D.groupsRollingBy cmp f (toStreamD m)
505-- > groups = groupsBy (==)
506-- > groups = groupsByRolling (==)
508-- Groups contiguous spans of equal elements together in individual groups.
510-- >>> Stream.toList $ Stream.groups Fold.toList $ Stream.fromList [1,1,2,2]
515groups :: (IsStream t, Monad m, Eq a) => Fold m a b -> t m a -> t m b
516groups = groupsBy (==)
518------------------------------------------------------------------------------
519-- Splitting - by a predicate
520------------------------------------------------------------------------------
522-- In general we can use deintercalate for splitting. Then we can also use
523-- uniqBy to condense the separators. One way to generalize splitting is to
526-- data Segment a b = Empty | Segment b | Separator a
528-- XXX splitOn and splitOnSuffix have a different behavior on an empty stream,
531-- | Split on an infixed separator element, dropping the separator. The
532-- supplied 'Fold' is applied on the split segments. Splits the stream on
533-- separator elements determined by the supplied predicate, separator is
534-- considered as infixed between two segments:
536-- >>> splitOn' p xs = Stream.toList $ Stream.splitOn p Fold.toList (Stream.fromList xs)
537-- >>> splitOn' (== '.') "a.b"
540-- An empty stream is folded to the default value of the fold:
542-- >>> splitOn' (== '.') ""
545-- If one or both sides of the separator are missing then the empty segment on
546-- that side is folded to the default output of the fold:
548-- >>> splitOn' (== '.') "."
551-- >>> splitOn' (== '.') ".a"
554-- >>> splitOn' (== '.') "a."
557-- >>> splitOn' (== '.') "a..b"
560-- splitOn is an inverse of intercalating single element:
562-- > Stream.intercalate (Stream.fromPure '.') Unfold.fromList . Stream.splitOn (== '.') Fold.toList === id
564-- Assuming the input stream does not contain the separator:
566-- > Stream.splitOn (== '.') Fold.toList . Stream.intercalate (Stream.fromPure '.') Unfold.fromList === id
570{-# INLINE splitOn #-}
572 :: (IsStream t, Monad m)
573 => (a -> Bool) -> Fold m a b -> t m a -> t m b
575 -- We can express the infix splitting in terms of optional suffix split
576 -- fold. After applying a suffix split fold repeatedly if the last segment
577 -- ends with a suffix then we need to return the default output of the fold
578 -- after that to make it an infix split.
580 -- Alternately, we can also express it using an optional prefix split fold.
581 -- If the first segment starts with a prefix then we need to emit the
582 -- default output of the fold before that to make it an infix split, and
583 -- then apply prefix split fold repeatedly.
585 -- Since a suffix split fold can be easily expressed using a
586 -- non-backtracking fold, we use that.
587 foldManyPost (FL.takeEndBy_ predicate f)
589-- | Split on a suffixed separator element, dropping the separator. The
590-- supplied 'Fold' is applied on the split segments.
592-- >>> splitOnSuffix' p xs = Stream.toList $ Stream.splitOnSuffix p Fold.toList (Stream.fromList xs)
593-- >>> splitOnSuffix' (== '.') "a.b."
596-- >>> splitOnSuffix' (== '.') "a."
599-- An empty stream results in an empty output stream:
601-- >>> splitOnSuffix' (== '.') ""
604-- An empty segment consisting of only a suffix is folded to the default output
607-- >>> splitOnSuffix' (== '.') "."
610-- >>> splitOnSuffix' (== '.') "a..b.."
613-- A suffix is optional at the end of the stream:
615-- >>> splitOnSuffix' (== '.') "a"
618-- >>> splitOnSuffix' (== '.') ".a"
621-- >>> splitOnSuffix' (== '.') "a.b"
624-- > lines = splitOnSuffix (== '\n')
626-- 'splitOnSuffix' is an inverse of 'intercalateSuffix' with a single element:
628-- > Stream.intercalateSuffix (Stream.fromPure '.') Unfold.fromList . Stream.splitOnSuffix (== '.') Fold.toList === id
630-- Assuming the input stream does not contain the separator:
632-- > Stream.splitOnSuffix (== '.') Fold.toList . Stream.intercalateSuffix (Stream.fromPure '.') Unfold.fromList === id
636{-# INLINE splitOnSuffix #-}
638 :: (IsStream t, Monad m)
639 => (a -> Bool) -> Fold m a b -> t m a -> t m b
640splitOnSuffix predicate f = foldMany (FL.takeEndBy_ predicate f)
642-- | Split on a prefixed separator element, dropping the separator. The
643-- supplied 'Fold' is applied on the split segments.
646-- > splitOnPrefix' p xs = Stream.toList $ Stream.splitOnPrefix p (Fold.toList) (Stream.fromList xs)
647-- > splitOnPrefix' (== '.') ".a.b"
651-- An empty stream results in an empty output stream:
653-- > splitOnPrefix' (== '.') ""
657-- An empty segment consisting of only a prefix is folded to the default output
661-- > splitOnPrefix' (== '.') "."
664-- > splitOnPrefix' (== '.') ".a.b."
667-- > splitOnPrefix' (== '.') ".a..b"
672-- A prefix is optional at the beginning of the stream:
675-- > splitOnPrefix' (== '.') "a"
678-- > splitOnPrefix' (== '.') "a.b"
682-- 'splitOnPrefix' is an inverse of 'intercalatePrefix' with a single element:
684-- > Stream.intercalatePrefix (Stream.fromPure '.') Unfold.fromList . Stream.splitOnPrefix (== '.') Fold.toList === id
686-- Assuming the input stream does not contain the separator:
688-- > Stream.splitOnPrefix (== '.') Fold.toList . Stream.intercalatePrefix (Stream.fromPure '.') Unfold.fromList === id
692{-# INLINE splitOnPrefix #-}
693splitOnPrefix :: -- (IsStream t, MonadCatch m) =>
694 (a -> Bool) -> Fold m a b -> t m a -> t m b
695splitOnPrefix _predicate _f = undefined
696 -- parseMany (Parser.sliceBeginBy predicate f)
698-- | Like 'splitOn' after stripping leading, trailing, and repeated separators.
699-- Therefore, @".a..b."@ with '.' as the separator would be parsed as
700-- @["a","b"]@. In other words, its like parsing words from whitespace
703-- >>> wordsBy' p xs = Stream.toList $ Stream.wordsBy p Fold.toList (Stream.fromList xs)
705-- >>> wordsBy' (== ',') ""
708-- >>> wordsBy' (== ',') ","
711-- >>> wordsBy' (== ',') ",a,,b,"
714-- > words = wordsBy isSpace
718-- It is equivalent to splitting in any of the infix/prefix/suffix styles
719-- followed by removal of empty segments.
720{-# INLINE wordsBy #-}
722 :: (IsStream t, Monad m)
723 => (a -> Bool) -> Fold m a b -> t m a -> t m b
724wordsBy predicate f m =
725 fromStreamD $ D.wordsBy predicate f (toStreamD m)
727-- | Like 'splitOnSuffix' but keeps the suffix attached to the resulting
730-- >>> splitWithSuffix' p xs = Stream.toList $ splitWithSuffix p Fold.toList (Stream.fromList xs)
732-- >>> splitWithSuffix' (== '.') ""
735-- >>> splitWithSuffix' (== '.') "."
738-- >>> splitWithSuffix' (== '.') "a"
741-- >>> splitWithSuffix' (== '.') ".a"
744-- >>> splitWithSuffix' (== '.') "a."
747-- >>> splitWithSuffix' (== '.') "a.b"
750-- >>> splitWithSuffix' (== '.') "a.b."
753-- >>> splitWithSuffix' (== '.') "a..b.."
754-- ["a.",".","b.","."]
758{-# INLINE splitWithSuffix #-}
760 :: (IsStream t, Monad m)
761 => (a -> Bool) -> Fold m a b -> t m a -> t m b
762splitWithSuffix predicate f = foldMany (FL.takeEndBy predicate f)
764------------------------------------------------------------------------------
765-- Splitting - on a delimiter sequence
766------------------------------------------------------------------------------
768-- Int list examples for splitOn:
770-- >>> splitList [] [1,2,3,3,4]
771-- > [[1],[2],[3],[3],[4]]
773-- >>> splitList [5] [1,2,3,3,4]
776-- >>> splitList [1] [1,2,3,3,4]
779-- >>> splitList [4] [1,2,3,3,4]
782-- >>> splitList [2] [1,2,3,3,4]
785-- >>> splitList [3] [1,2,3,3,4]
788-- >>> splitList [3,3] [1,2,3,3,4]
791-- >>> splitList [1,2,3,3,4] [1,2,3,3,4]
794-- This can be implemented easily using Rabin Karp
795-- | Split on any one of the given patterns.
799{-# INLINE splitOnAny #-}
800splitOnAny :: -- (IsStream t, Monad m, Unbox a, Integral a) =>
801 [Array a] -> Fold m a b -> t m a -> t m b
802splitOnAny _subseq _f _m =
803 undefined -- D.fromStreamD $ D.splitOnAny f subseq (D.toStreamD m)
805-- XXX use a non-monadic intersperse to remove the MonadAsync constraint.
806-- XXX Use two folds, one ring buffer fold for separator sequence and the other
807-- split consumer fold. The input is fed to the ring fold first and the
808-- rejected input is fed to the split fold. If the separator matches, the ring
809-- fold would consume all.
811-- | Like 'splitOnSeq' but splits the separator as well, as an infix token.
813-- >>> splitOn'_ pat xs = Stream.toList $ Stream.splitBySeq (Array.fromList pat) Fold.toList (Stream.fromList xs)
815-- >>> splitOn'_ "" "hello"
816-- ["h","","e","","l","","l","","o"]
818-- >>> splitOn'_ "hello" ""
821-- >>> splitOn'_ "hello" "hello"
824-- >>> splitOn'_ "x" "hello"
827-- >>> splitOn'_ "h" "hello"
830-- >>> splitOn'_ "o" "hello"
833-- >>> splitOn'_ "e" "hello"
836-- >>> splitOn'_ "l" "hello"
837-- ["he","l","","l","o"]
839-- >>> splitOn'_ "ll" "hello"
843{-# INLINE splitBySeq #-}
845 :: (IsStream t, MonadAsync m, Unbox a, Enum a, Eq a)
846 => Array a -> Fold m a b -> t m a -> t m b
848 intersperseM (fold f (A.read patt)) $ splitOnSeq patt f m
850-- | Like 'splitSuffixBy' but the separator is a sequence of elements, instead
851-- of a predicate for a single element.
853-- >>> splitOnSuffixSeq_ pat xs = Stream.toList $ Stream.splitOnSuffixSeq (Array.fromList pat) Fold.toList (Stream.fromList xs)
855-- >>> splitOnSuffixSeq_ "." ""
858-- >>> splitOnSuffixSeq_ "." "."
861-- >>> splitOnSuffixSeq_ "." "a"
864-- >>> splitOnSuffixSeq_ "." ".a"
867-- >>> splitOnSuffixSeq_ "." "a."
870-- >>> splitOnSuffixSeq_ "." "a.b"
873-- >>> splitOnSuffixSeq_ "." "a.b."
876-- >>> splitOnSuffixSeq_ "." "a..b.."
879-- > lines = splitOnSuffixSeq "\n"
881-- 'splitOnSuffixSeq' is an inverse of 'intercalateSuffix'. The following law
884-- > intercalateSuffix . splitOnSuffixSeq == id
886-- The following law holds when the separator is non-empty and contains none of
887-- the elements present in the input lists:
889-- > splitSuffixOn . intercalateSuffix == id
891-- >>> splitOnSuffixSeq pat f = Stream.foldMany (Fold.takeEndBySeq_ pat f)
894{-# INLINE splitOnSuffixSeq #-}
896 :: (IsStream t, MonadIO m, Unbox a, Enum a, Eq a)
897 => Array a -> Fold m a b -> t m a -> t m b
898splitOnSuffixSeq patt f m =
899 fromStreamD $ D.splitOnSuffixSeq False patt f (toStreamD m)
900-- XXX This may have a problem if the stream terminates and we start extracting
901-- the fold and then the fold terminates before consuming all the buffered
902-- input. We have no way to supply the buffered input back to the driver.
903-- splitOnSuffixSeq patt f =
904-- foldMany (FL.takeEndBySeq_ patt f)
906-- | Like 'splitOn' but drops any empty splits.
909{-# INLINE wordsOn #-}
910wordsOn :: -- (IsStream t, Monad m, Unbox a, Eq a) =>
911 Array a -> Fold m a b -> t m a -> t m b
912wordsOn _subseq _f _m =
913 undefined -- D.fromStreamD $ D.wordsOn f subseq (D.toStreamD m)
915-- | Like 'splitOnSuffixSeq' but keeps the suffix intact in the splits.
917-- >>> splitWithSuffixSeq' pat xs = Stream.toList $ Stream.splitWithSuffixSeq (Array.fromList pat) Fold.toList (Stream.fromList xs)
919-- >>> splitWithSuffixSeq' "." ""
922-- >>> splitWithSuffixSeq' "." "."
925-- >>> splitWithSuffixSeq' "." "a"
928-- >>> splitWithSuffixSeq' "." ".a"
931-- >>> splitWithSuffixSeq' "." "a."
934-- >>> splitWithSuffixSeq' "." "a.b"
937-- >>> splitWithSuffixSeq' "." "a.b."
940-- >>> splitWithSuffixSeq' "." "a..b.."
941-- ["a.",".","b.","."]
943-- >>> splitWithSuffixSeq pat f = Stream.foldMany (Fold.takeEndBySeq pat f)
946{-# INLINE splitWithSuffixSeq #-}
948 :: (IsStream t, MonadIO m, Unbox a, Enum a, Eq a)
949 => Array a -> Fold m a b -> t m a -> t m b
950splitWithSuffixSeq patt f m =
951 fromStreamD $ D.splitOnSuffixSeq True patt f (toStreamD m)
952-- splitWithSuffixSeq patt f =
953-- foldMany (FL.takeEndBySeq patt f)
955-- This can be implemented easily using Rabin Karp
956-- | Split post any one of the given patterns.
959{-# INLINE splitOnSuffixSeqAny #-}
960splitOnSuffixSeqAny :: -- (IsStream t, Monad m, Unbox a, Integral a) =>
961 [Array a] -> Fold m a b -> t m a -> t m b
962splitOnSuffixSeqAny _subseq _f _m = undefined
963 -- D.fromStreamD $ D.splitPostAny f subseq (D.toStreamD m)
965------------------------------------------------------------------------------
967------------------------------------------------------------------------------
969-- | Group the input stream into groups of @n@ elements each and then fold each
970-- group using the provided fold function.
972-- >>> Stream.toList $ Stream.chunksOf 2 Fold.sum (Stream.enumerateFromTo 1 10)
975-- This can be considered as an n-fold version of 'take' where we apply
976-- 'take' repeatedly on the leftover stream until the stream exhausts.
978-- @chunksOf n f = foldMany (FL.take n f)@
981{-# INLINE chunksOf #-}
983 :: (IsStream t, Monad m)
984 => Int -> Fold m a b -> t m a -> t m b
985chunksOf n f = fromStreamD . D.chunksOf n f . toStreamD
987-- | @arraysOf n stream@ groups the elements in the input stream into arrays of
990-- Same as the following but may be more efficient:
992-- > arraysOf n = Stream.foldMany (A.writeN n)
995{-# INLINE arraysOf #-}
996arraysOf :: (IsStream t, MonadIO m, Unbox a)
997 => Int -> t m a -> t m (Array a)
998arraysOf n = fromStreamD . A.arraysOf n . toStreamD
1000-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
1001-- XXX add this example after fixing the serial stream rate control
1003-- | Group the input stream into windows of @n@ second each and then fold each
1004-- group using the provided fold function.
1006-- >>> Stream.toList $ Stream.take 5 $ Stream.intervalsOf 1 Fold.sum $ Stream.constRate 2 $ Stream.enumerateFrom 1
1007-- [...,...,...,...,...]
1010{-# INLINE intervalsOf #-}
1012 :: (IsStream t, MonadAsync m)
1013 => Double -> Fold m a b -> t m a -> t m b
1015 splitWithSuffix isNothing (FL.catMaybes f)
1016 (interjectSuffix n (return Nothing) (map Just xs))
1018-- XXX This can be implemented more efficiently by sharing a Clock.Timer across
1019-- parallel threads and resetting it whenever a span is emitted.
1021-- | Like 'chunksOf' but if the chunk is not completed within the specified
1022-- time interval then emit whatever we have collected till now. The chunk
1023-- timeout is reset whenever a chunk is emitted. The granularity of the clock
1026-- >>> s = Stream.delayPost 0.3 $ Stream.fromList [1..1000]
1027-- >>> f = Stream.mapM_ print $ Stream.chunksOfTimeout 5 1 Fold.toList s
1030{-# INLINE chunksOfTimeout #-}
1031chunksOfTimeout :: (IsStream t, MonadAsync m, Functor (t m))
1032 => Int -> Double -> FL.Fold m a b -> t m a -> t m b
1033chunksOfTimeout n timeout f =
1035 . classifySessionsBy
1036 0.1 False (const (return False)) timeout (FL.take n f)
1037 . Transform.timestamped
1040------------------------------------------------------------------------------
1041-- Windowed classification
1042------------------------------------------------------------------------------
1044-- TODO: To mark the position in space or time we can have Indexed or
1045-- TimeStamped types. That can make it easy to deal with the position indices
1048------------------------------------------------------------------------------
1049-- Keyed Sliding Windows
1050------------------------------------------------------------------------------
1053{-# INLINABLE classifySlidingChunks #-}
1054classifySlidingChunks
1055 :: (IsStream t, MonadAsync m, Ord k)
1056 => Int -- ^ window size
1057 -> Int -- ^ window slide
1058 -> Fold m a b -- ^ Fold to be applied to window events
1059 -> t m (k, a, Bool) -- ^ window key, data, close event
1061classifySlidingChunks wsize wslide (Fold step initial extract) str
1064-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
1065-- slide the send window when an ack is received and we slide the receive
1066-- window when a sequence is complete. Sliding is stateful in case of TCP,
1067-- sliding releases the send buffer or makes data available to the user from
1068-- the receive buffer.
1069{-# INLINABLE classifySlidingSessions #-}
1070classifySlidingSessions
1071 :: (IsStream t, MonadAsync m, Ord k)
1072 => Double -- ^ timer tick in seconds
1073 -> Double -- ^ time window size
1074 -> Double -- ^ window slide
1075 -> Fold m a b -- ^ Fold to be applied to window events
1076 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
1078classifySlidingSessions tick interval slide (Fold step initial extract) str
1082------------------------------------------------------------------------------
1083-- Sliding Window Buffers
1084------------------------------------------------------------------------------
1086-- These buffered versions could be faster than concurrent incremental folds of
1087-- all overlapping windows as in many cases we may not need all the values to
1088-- compute the fold, we can just compute the result using the old value and new
1089-- value. However, we may need the buffer once in a while, for example for
1090-- string search we usually compute the hash incrementally but when the hash
1091-- matches the hash of the pattern we need to compare the whole string.
1093-- XXX we should be able to implement sequence based splitting combinators
1094-- using this combinator.
1097-- | Buffer n elements of the input in a ring buffer. When t new elements are
1098-- collected, slide the window to remove the same number of oldest elements,
1099-- insert the new elements, and apply an incremental fold on the sliding
1100-- window, supplying the outgoing elements, the new ring buffer as arguments.
1102 :: (IsStream t, Monad m, Ord a, Unbox a)
1103 => Int -- window size
1104 -> Int -- window slide
1105 -> Fold m (Ring a, Array a) b
1108slidingChunkBuffer = undefined
1110-- Buffer n seconds worth of stream elements of the input in a radix tree.
1111-- Every t seconds, remove the items that are older than n seconds, and apply
1112-- an incremental fold on the sliding window, supplying the outgoing elements,
1113-- and the new radix tree buffer as arguments.
1115 :: (IsStream t, Monad m, Ord a, Unbox a)
1116 => Int -- window size
1118 -> Fold m (RTree a, Array a) b
1121slidingSessionBuffer = undefined
1124------------------------------------------------------------------------------
1125-- Keyed Session Windows
1126------------------------------------------------------------------------------
1129-- | Keyed variable size space windows. Close the window if we do not receive a
1130-- window event in the next "spaceout" elements.
1131{-# INLINABLE classifyChunksBy #-}
1133 :: (IsStream t, MonadAsync m, Ord k)
1134 => Int -- ^ window spaceout (spread)
1135 -> Bool -- ^ reset the spaceout when a chunk window element is received
1136 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1137 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1139classifyChunksBy spanout reset (Fold step initial extract) str = undefined
1141-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
1142-- received within the chunk size window. The chunk gets closed only if no
1143-- element is received within the chunk window.
1145{-# INLINABLE classifyKeepAliveChunks #-}
1146classifyKeepAliveChunks
1147 :: (IsStream t, MonadAsync m, Ord k)
1148 => Int -- ^ window spaceout (spread)
1149 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1150 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1152classifyKeepAliveChunks spanout = classifyChunksBy spanout True
1155data SessionState t m f s b = SessionState
1156 { sessionCurTime :: !AbsTime -- ^ time since last event
1157 , sessionEventTime :: !AbsTime -- ^ time as per last event
1158 -- We can use the Map size instead of maintaining a count, but if we have
1159 -- to switch to HashMap then it can be useful.
1160 , sessionCount :: !Int -- ^ total number of sessions in progress
1161 , sessionTimerHeap :: H.Heap (H.Entry AbsTime (Key f)) -- ^ heap for timeouts
1162 , sessionKeyValueMap :: f s -- ^ Stored sessions for keys
1163 , sessionOutputStream :: t (m :: Type -> Type) (Key f, b) -- ^ Completed sessions
1166data SessionEntry s = LiveSession !AbsTime !s | ZombieSession
1168-- delete from map and output the fold accumulator
1169ejectEntry :: (Monad m, IsStream t, IsMap f) =>
1177 -> m (heap, f entry, t m (Key f, b), Int)
1178ejectEntry extract hp mp out cnt acc key = do
1180 let out1 = (key, sess) `cons` out
1181 let mp1 = IsMap.mapDelete key mp
1182 return (hp, mp1, out1, cnt - 1)
1184{-# NOINLINE flush #-}
1185flush :: (Monad m, IsMap f, IsStream t) =>
1187 -> SessionState t m f (SessionEntry s) b
1188 -> m (SessionState t m f (SessionEntry s) b)
1189flush extract session@SessionState{..} = do
1190 (hp', mp', out, count) <-
1193 , sessionKeyValueMap
1198 { sessionCount = count
1199 , sessionTimerHeap = hp'
1200 , sessionKeyValueMap = mp'
1201 , sessionOutputStream = out
1206 ejectAll (hp, mp, out, !cnt) = do
1207 let hres = H.uncons hp
1209 Just (Entry _ key, hp1) -> do
1210 r <- case IsMap.mapLookup key mp of
1211 Nothing -> return (hp1, mp, out, cnt)
1212 Just ZombieSession ->
1213 return (hp1, IsMap.mapDelete key mp, out, cnt)
1214 Just (LiveSession _ acc) ->
1215 ejectEntry extract hp1 mp out cnt acc key
1218 assert (IsMap.mapNull mp) (return ())
1219 return (hp, mp, out, cnt)
1221{-# NOINLINE ejectOne #-}
1222ejectOne :: (IsMap f, IsStream t, Monad m) =>
1225 -> ( H.Heap (Entry AbsTime (Key f))
1226 , f (SessionEntry acc)
1230 -> m ( H.Heap (Entry AbsTime (Key f))
1231 , f (SessionEntry acc)
1232 , t m (Key f, b), Int
1234ejectOne reset extract = go
1238 go (hp, mp, out, !cnt) = do
1239 let hres = H.uncons hp
1241 Just (Entry expiry key, hp1) ->
1242 case IsMap.mapLookup key mp of
1243 Nothing -> go (hp1, mp, out, cnt)
1244 Just ZombieSession ->
1245 go (hp1, IsMap.mapDelete key mp, out, cnt)
1246 Just (LiveSession expiry1 acc) -> do
1247 if not reset || expiry1 <= expiry
1248 then ejectEntry extract hp1 mp out cnt acc key
1250 -- reset the session timeout and continue
1251 let hp2 = H.insert (Entry expiry1 key) hp1
1252 in go (hp2, mp, out, cnt)
1254 assert (IsMap.mapNull mp) (return ())
1255 return (hp, mp, out, cnt)
1257{-# NOINLINE ejectExpired #-}
1258ejectExpired :: (IsMap f, IsStream t, Monad m) =>
1262 -> SessionState t m f (SessionEntry acc) b
1264 -> m (SessionState t m f (SessionEntry acc) b)
1265ejectExpired reset ejectPred extract session@SessionState{..} curTime = do
1266 (hp', mp', out, count) <-
1268 sessionTimerHeap sessionKeyValueMap IsStream.nil sessionCount
1270 { sessionCurTime = curTime
1271 , sessionCount = count
1272 , sessionTimerHeap = hp'
1273 , sessionKeyValueMap = mp'
1274 , sessionOutputStream = out
1279 ejectLoop hp mp out !cnt = do
1280 let hres = H.uncons hp
1282 Just (Entry expiry key, hp1) -> do
1284 if curTime >= expiry
1285 then return (True, False)
1291 case IsMap.mapLookup key mp of
1292 Nothing -> ejectLoop hp1 mp out cnt
1293 Just ZombieSession ->
1294 ejectLoop hp1 (IsMap.mapDelete key mp) out cnt
1295 Just (LiveSession expiry1 acc) -> do
1296 if expiry1 <= curTime || not reset || force
1298 (hp2,mp1,out1,cnt1) <-
1299 ejectEntry extract hp1 mp out cnt acc key
1300 ejectLoop hp2 mp1 out1 cnt1
1302 -- reset the session timeout and continue
1303 let hp2 = H.insert (Entry expiry1 key) hp1
1304 in ejectLoop hp2 mp out cnt
1305 else return (hp, mp, out, cnt)
1307 assert (IsMap.mapNull mp) (return ())
1308 return (hp, mp, out, cnt)
1310-- XXX Use mutable IORef in accumulator
1311{-# INLINE classifySessionsByGeneric #-}
1312classifySessionsByGeneric
1313 :: forall t m f a b. (IsStream t, MonadAsync m, IsMap f)
1314 => Proxy (f :: (Type -> Type))
1315 -> Double -- ^ timer tick in seconds
1316 -> Bool -- ^ reset the timer when an event is received
1317 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
1318 -> Double -- ^ session timeout in seconds
1319 -> Fold m a b -- ^ Fold to be applied to session data
1320 -> t m (AbsTime, (Key f, a)) -- ^ timestamp, (session key, session data)
1321 -> t m (Key f, b) -- ^ session key, fold result
1322classifySessionsByGeneric _ tick reset ejectPred tmout
1323 (Fold step initial extract) input =
1324 Expand.unfoldMany (Unfold.lmap sessionOutputStream Unfold.fromStream)
1325 $ scanlMAfter' sstep (return szero) (flush extract)
1326 $ interjectSuffix tick (return Nothing)
1331 timeoutMs = toRelTime (round (tmout * 1000) :: MilliSecond64)
1332 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
1333 szero = SessionState
1334 { sessionCurTime = toAbsTime (0 :: MilliSecond64)
1335 , sessionEventTime = toAbsTime (0 :: MilliSecond64)
1337 , sessionTimerHeap = H.empty
1338 , sessionKeyValueMap = IsMap.mapEmpty :: f s
1339 , sessionOutputStream = IsStream.nil
1342 -- We can eject sessions based on the current session count to limit
1343 -- memory consumption. There are two possible strategies:
1345 -- 1) Eject old sessions or sessions beyond a certain/lower timeout
1346 -- threshold even before timeout, effectively reduce the timeout.
1347 -- 2) Drop creation of new sessions but keep accepting new events for the
1350 -- We use the first strategy as of now.
1352 -- Got a new stream input element
1353 sstep session@SessionState{..} (Just (timestamp, (key, value))) = do
1354 -- XXX instead of a heap we could use a timer wheel.
1356 -- XXX if the key is an Int, we can also use an IntMap for slightly
1357 -- better performance.
1361 -- Values for each key are collected in a map using the supplied fold.
1362 -- When we insert a key in the Map we insert an entry into the heap as
1363 -- well with the session expiry as the sort key. The Map entry
1364 -- consists of the fold result, and the expiry time of the session. If
1365 -- "reset" is True the expiry time is readjusted whenever a new event
1366 -- is processed. If the fold terminates and a new session is started
1367 -- for the same key the expiry time is set to the first timestamp of
1370 -- The heap must have at most one entry for any given key. The heap is
1371 -- processed periodically to remove the expired entries. We pick up an
1372 -- expired entry from the top of the heap and if the session has
1373 -- expired based on the expiry time in the Map entry then we remove the
1374 -- session from the Map and yield its fold output. Otherwise, we
1375 -- reinsert the entry into the heap based on the current expiry in the
1378 -- If an entry is removed from the Map and not removed from the heap
1379 -- and in the meantime it is inserted again in the Map (using the same
1380 -- key) then how do we avoid inserting multiple entries in the heap?
1381 -- For this reason we maintain the invariant that the Map entry is
1382 -- removed only when the heap entry is removed. Even if the fold has
1383 -- finished we still keep a dummy Map entry (ZombieSession) until the
1384 -- heap entry is removed. That way if we have a Map entry we do not
1385 -- insert a heap entry because we know it is already there.
1386 -- XXX The ZombieSession mechanism does not work as expected as we
1387 -- ignore ZombieSession when inserting a new entry. Anyway, we can
1388 -- remove this mechanism as at most only two heap entries may be
1389 -- created and they will be ultimately cleaned up.
1391 -- Heap processing needs the map and map processing needs the heap,
1392 -- therefore we cannot separate the two for modularity unless we have a
1393 -- way to achieve mutual recursion.
1395 let curTime = max sessionEventTime timestamp
1396 mOld = IsMap.mapLookup key sessionKeyValueMap
1398 -- deleting a key from the heap is expensive, so we never
1399 -- delete a key from heap, we just purge it from the Map and it
1400 -- gets purged from the heap on timeout. We just need an extra
1401 -- lookup in the Map when the key is purged from the heap, that
1402 -- should not be expensive.
1404 let (mp, cnt) = case mOld of
1405 Just (LiveSession _ _) ->
1407 key ZombieSession sessionKeyValueMap
1410 _ -> (sessionKeyValueMap, sessionCount)
1412 { sessionCurTime = curTime
1413 , sessionEventTime = curTime
1414 , sessionCount = cnt
1415 , sessionKeyValueMap = mp
1416 , sessionOutputStream = fromPure (key, fb)
1419 let expiry = addToAbsTime timestamp timeoutMs
1420 (hp1, mp1, out1, cnt1) <- do
1421 let vars = (sessionTimerHeap, sessionKeyValueMap,
1422 IsStream.nil, sessionCount)
1424 -- inserting new entry
1426 -- Eject a session from heap and map if needed
1427 eject <- ejectPred sessionCount
1428 (hp, mp, out, cnt) <-
1430 then ejectOne reset extract vars
1433 -- Insert the new session in heap
1434 let hp' = H.insert (Entry expiry key) hp
1435 in return (hp', mp, out, cnt + 1)
1436 -- updating old entry
1437 Just _ -> return vars
1439 let acc = LiveSession expiry fs1
1440 mp2 = IsMap.mapInsert key acc mp1
1441 return $ SessionState
1442 { sessionCurTime = curTime
1443 , sessionEventTime = curTime
1444 , sessionCount = cnt1
1445 , sessionTimerHeap = hp1
1446 , sessionKeyValueMap = mp2
1447 , sessionOutputStream = out1
1451 Just (LiveSession _ acc) -> return $ FL.Partial acc
1455 error $ "classifySessionsBy: "
1456 ++ "The supplied fold must consume at least one input"
1458 res <- step fs value
1460 FL.Done fb -> done fb
1461 FL.Partial fs1 -> partial fs1
1463 -- Got a timer tick event
1464 sstep sessionState@SessionState{..} Nothing =
1465 let curTime = addToAbsTime sessionCurTime tickMs
1466 in ejectExpired reset ejectPred extract sessionState curTime
1468-- | @classifySessionsBy tick keepalive predicate timeout fold stream@
1469-- classifies an input event @stream@ consisting of @(timestamp, (key,
1470-- value))@ into sessions based on the @key@, folding all the values
1471-- corresponding to the same key into a session using the supplied @fold@.
1473-- When the fold terminates or a @timeout@ occurs, a tuple consisting of the
1474-- session key and the folded value is emitted in the output stream. The
1475-- timeout is measured from the first event in the session. If the @keepalive@
1476-- option is set to 'True' the timeout is reset to 0 whenever an event is
1479-- The @timestamp@ in the input stream is an absolute time from some epoch,
1480-- characterizing the time when the input event was generated. The notion of
1481-- current time is maintained by a monotonic event time clock using the
1482-- timestamps seen in the input stream. The latest timestamp seen till now is
1483-- used as the base for the current time. When no new events are seen, a timer
1484-- is started with a clock resolution of @tick@ seconds. This timer is used to
1485-- detect session timeouts in the absence of new events.
1487-- To ensure an upper bound on the memory used the number of sessions can be
1488-- limited to an upper bound. If the ejection @predicate@ returns 'True', the
1489-- oldest session is ejected before inserting a new session.
1491-- When the stream ends any buffered sessions are ejected immediately.
1493-- If a session key is received even after a session has finished, another
1494-- session is created for that key.
1497-- Stream.mapM_ print
1498-- $ Stream.classifySessionsBy 1 False (const (return False)) 3 (Fold.take 3 Fold.toList)
1499-- $ Stream.timestamped
1500-- $ Stream.delay 0.1
1501-- $ Stream.fromList ((,) <$> [1,2,3] <*> ['a','b','c'])
1508{-# INLINE classifySessionsBy #-}
1510 :: (IsStream t, MonadAsync m, Ord k)
1511 => Double -- ^ timer tick in seconds
1512 -> Bool -- ^ reset the timer when an event is received
1513 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
1514 -> Double -- ^ session timeout in seconds
1515 -> Fold m a b -- ^ Fold to be applied to session data
1516 -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
1517 -> t m (k, b) -- ^ session key, fold result
1518classifySessionsBy = classifySessionsByGeneric (Proxy :: Proxy (Map k))
1520-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
1521-- option set to 'True'.
1524-- classifyKeepAliveSessions = classifySessionsBy 1 True
1529{-# INLINE classifyKeepAliveSessions #-}
1530classifyKeepAliveSessions ::
1531 (IsStream t, MonadAsync m, Ord k)
1532 => (Int -> m Bool) -- ^ predicate to eject sessions on session count
1533 -> Double -- ^ session inactive timeout
1534 -> Fold m a b -- ^ Fold to be applied to session payload data
1535 -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
1537classifyKeepAliveSessions = classifySessionsBy 1 True
1539------------------------------------------------------------------------------
1540-- Keyed tumbling windows
1541------------------------------------------------------------------------------
1543-- Tumbling windows is a special case of sliding windows where the window slide
1544-- is the same as the window size. Or it can be a special case of session
1545-- windows where the reset flag is set to False.
1547-- XXX instead of using the early termination flag in the stream, we can use an
1548-- early terminating fold instead.
1551-- | Split the stream into fixed size chunks of specified size. Within each
1552-- such chunk fold the elements in buckets identified by the keys. A particular
1553-- bucket fold can be terminated early if a closing flag is encountered in an
1554-- element for that key.
1557{-# INLINABLE classifyChunksOf #-}
1559 :: (IsStream t, MonadAsync m, Ord k)
1560 => Int -- ^ window size
1561 -> Fold m a b -- ^ Fold to be applied to window events
1562 -> t m (k, a, Bool) -- ^ window key, data, close event
1564classifyChunksOf wsize = classifyChunksBy wsize False
1567-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
1568-- option set to 'False'.
1571-- classifySessionsOf = classifySessionsBy 1 False
1576{-# INLINE classifySessionsOf #-}
1577classifySessionsOf ::
1578 (IsStream t, MonadAsync m, Ord k)
1579 => (Int -> m Bool) -- ^ predicate to eject sessions on session count
1580 -> Double -- ^ time window size
1581 -> Fold m a b -- ^ Fold to be applied to session data
1582 -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
1584classifySessionsOf = classifySessionsBy 1 False
1586------------------------------------------------------------------------------
1588------------------------------------------------------------------------------
1590-- | @splitInnerBy splitter joiner stream@ splits the inner containers @f a@ of
1591-- an input stream @t m (f a)@ using the @splitter@ function. Container
1592-- elements @f a@ are collected until a split occurs, then all the elements
1593-- before the split are joined using the @joiner@ function.
1595-- For example, if we have a stream of @Array Word8@, we may want to split the
1596-- stream into arrays representing lines separated by '\n' byte such that the
1597-- resulting stream after a split would be one array for each line.
1599-- CAUTION! This is not a true streaming function as the container size after
1600-- the split and merge may not be bounded.
1603{-# INLINE splitInnerBy #-}
1605 :: (IsStream t, Monad m)
1606 => (f a -> m (f a, Maybe (f a))) -- splitter
1607 -> (f a -> f a -> m (f a)) -- joiner
1610splitInnerBy splitter joiner xs =
1611 fromStreamD $ D.splitInnerBy splitter joiner $ toStreamD xs
1613-- | Like 'splitInnerBy' but splits assuming the separator joins the segment in
1617{-# INLINE splitInnerBySuffix #-}
1619 :: (IsStream t, Monad m, Eq (f a), Monoid (f a))
1620 => (f a -> m (f a, Maybe (f a))) -- splitter
1621 -> (f a -> f a -> m (f a)) -- joiner
1624splitInnerBySuffix splitter joiner xs =
1625 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