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 Foreign.Storable (Storable)
162import Streamly.Internal.Control.Concurrent (MonadAsync)
163import Streamly.Internal.Data.Fold.Type (Fold (..))
164import Streamly.Internal.Data.IsMap (IsMap(..))
165import Streamly.Internal.Data.Refold.Type (Refold (..))
166import Streamly.Internal.Data.Parser (Parser (..), ParseError)
167import Streamly.Internal.Data.Array.Type (Array)
168import Streamly.Internal.Data.Stream.IsStream.Common
177import Streamly.Internal.Data.Stream.IsStream.Type
178 (IsStream(..), fromStreamD, toStreamD, cons)
179import Streamly.Internal.Data.Stream.Serial(toStreamK)
180import Streamly.Internal.Data.Time.Units
181 ( AbsTime, MilliSecond64(..), addToAbsTime, toRelTime
183import Streamly.Internal.Data.Unboxed (Unbox)
185import qualified Data.Heap as H
186import qualified Streamly.Internal.Data.Unfold as Unfold
187import qualified Streamly.Internal.Data.Array.Type as A
189import qualified Streamly.Internal.Data.Fold as FL
190 (Fold, Step(..), takeEndBy_, takeEndBy, catMaybes, take)
191import qualified Streamly.Internal.Data.IsMap as IsMap
192import qualified Streamly.Internal.Data.Parser.ParserD as PRD (Parser(..))
193import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
194import qualified Streamly.Internal.Data.Stream.StreamD as D
209import qualified Streamly.Internal.Data.Stream.IsStream.Expand as Expand
210import qualified Streamly.Internal.Data.Stream.IsStream.Transform as Transform
212import Prelude hiding (concatMap, map)
216-- >>> :set -fno-warn-deprecations
217-- >>> import Prelude hiding (zipWith, concatMap, concat)
218-- >>> import qualified Streamly.Prelude as Stream
219-- >>> import Streamly.Internal.Data.Stream.IsStream as Stream
220-- >>> import qualified Streamly.Data.Fold as Fold
221-- >>> import qualified Streamly.Internal.Data.Fold as Fold
222-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
223-- >>> import qualified Streamly.Internal.Data.Parser as Parser
224-- >>> import qualified Streamly.Data.Array as Array
226------------------------------------------------------------------------------
228------------------------------------------------------------------------------
230-- | Drop prefix from the input stream if present.
235{-# INLINE dropPrefix #-}
237 -- (Eq a, IsStream t, Monad m) =>
238 t m a -> t m a -> t m a
239dropPrefix = error "Not implemented yet!"
241-- | Drop all matching infix from the input stream if present. Infix stream
242-- may be consumed multiple times.
244-- Space: @O(n)@ where n is the length of the infix.
247{-# INLINE dropInfix #-}
249 -- (Eq a, IsStream t, Monad m) =>
250 t m a -> t m a -> t m a
251dropInfix = error "Not implemented yet!"
253-- | Drop suffix from the input stream if present. Suffix stream may be
254-- consumed multiple times.
256-- Space: @O(n)@ where n is the length of the suffix.
259{-# INLINE dropSuffix #-}
261 -- (Eq a, IsStream t, Monad m) =>
262 t m a -> t m a -> t m a
263dropSuffix = error "Not implemented yet!"
265------------------------------------------------------------------------------
267------------------------------------------------------------------------------
269-- Splitting operations that take a predicate and a Fold can be
270-- expressed using parseMany. Operations like chunksOf, intervalsOf, split*,
271-- can be expressed using parseMany when used with an appropriate Parser.
273-- XXX We need takeGE/takeBetween to implement "some" using "many".
275-- | Apply a 'Fold' repeatedly on a stream and emit the fold outputs in the
278-- To sum every two contiguous elements in a stream:
280-- >>> f = Fold.take 2 Fold.sum
281-- >>> Stream.toList $ Stream.foldMany f $ Stream.fromList [1..10]
284-- On an empty stream the output is empty:
286-- >>> Stream.toList $ Stream.foldMany f $ Stream.fromList []
289-- Note @Stream.foldMany (Fold.take 0)@ would result in an infinite loop in a
294{-# INLINE foldMany #-}
296 :: (IsStream t, Monad m)
300foldMany f m = fromStreamD $ D.foldMany f (toStreamD m)
302-- | Like 'foldMany' but using the 'Refold' type instead of 'Fold'.
305{-# INLINE refoldMany #-}
306refoldMany :: (IsStream t, Monad m) =>
307 Refold m c a b -> m c -> t m a -> t m b
308refoldMany f action = fromStreamD . D.refoldMany f action . toStreamD
310-- | Apply a stream of folds to an input stream and emit the results in the
315{-# INLINE foldSequence #-}
317 :: -- (IsStream t, Monad m) =>
321foldSequence _f _m = undefined
323-- | Iterate a fold generator on a stream. The initial value @b@ is used to
324-- generate the first fold, the fold is applied on the stream and the result of
325-- the fold is used to generate the next fold and so on.
328-- >>> import Data.Monoid (Sum(..))
329-- >>> f x = return (Fold.take 2 (Fold.sconcat x))
330-- >>> s = Stream.map Sum $ Stream.fromList [1..10]
331-- >>> Stream.toList $ Stream.map getSum $ Stream.foldIterateM f (pure 0) s
336-- This is the streaming equivalent of monad like sequenced application of
337-- folds where next fold is dependent on the previous fold.
341{-# INLINE foldIterateM #-}
343 (IsStream t, Monad m) => (b -> m (Fold m a b)) -> m b -> t m a -> t m b
344foldIterateM f i m = fromStreamD $ D.foldIterateM f i (toStreamD m)
346-- | Like 'foldIterateM' but using the 'Refold' type instead. This could be
347-- much more efficient due to stream fusion.
350{-# INLINE refoldIterateM #-}
351refoldIterateM :: (IsStream t, Monad m) =>
352 Refold m b a b -> m b -> t m a -> t m b
353refoldIterateM c i m = fromStreamD $ D.refoldIterateM c i (toStreamD m)
355------------------------------------------------------------------------------
357------------------------------------------------------------------------------
359-- | Apply a 'Parser' repeatedly on a stream and emit the parsed values in the
362-- This is the streaming equivalent of the 'Streamly.Internal.Data.Parser.many'
365-- >>> Stream.toList $ Stream.parseMany (Parser.takeBetween 0 2 Fold.sum) $ Stream.fromList [1..10]
366-- [Right 3,Right 7,Right 11,Right 15,Right 19]
369-- > Stream.toList $ Stream.parseMany (Parser.line Fold.toList) $ Stream.fromList "hello\\nworld"
370-- ["hello\\n","world"]
375-- foldMany f = parseMany (fromFold f)
378-- Known Issues: When the parser fails there is no way to get the remaining
383{-# INLINE parseMany #-}
385 :: (IsStream t, Monad m)
388 -> t m (Either ParseError b)
390 fromStreamD $ D.parseManyD p (toStreamD m)
392-- | Same as parseMany but for StreamD streams.
396{-# INLINE parseManyD #-}
398 :: (IsStream t, Monad m)
401 -> t m (Either ParseError b)
403 fromStreamD $ D.parseManyD p (toStreamD m)
405-- | Apply a stream of parsers to an input stream and emit the results in the
410{-# INLINE parseSequence #-}
412 :: -- (IsStream t, Monad m) =>
416parseSequence _f _m = undefined
418-- XXX Change the parser arguments' order
420-- | @parseManyTill collect test stream@ tries the parser @test@ on the input,
421-- if @test@ fails it backtracks and tries @collect@, after @collect@ succeeds
422-- @test@ is tried again and so on. The parser stops when @test@ succeeds. The
423-- output of @test@ is discarded and the output of @collect@ is emitted in the
424-- output stream. The parser fails if @collect@ fails.
428{-# INLINE parseManyTill #-}
430 -- (IsStream t, MonadThrow m) =>
435parseManyTill = undefined
437-- | Iterate a parser generating function on a stream. The initial value @b@ is
438-- used to generate the first parser, the parser is applied on the stream and
439-- the result is used to generate the next parser and so on.
441-- >>> import Data.Monoid (Sum(..))
442-- >>> Stream.toList $ fmap getSum $ Stream.rights $ Stream.parseIterate (\b -> Parser.takeBetween 0 2 (Fold.sconcat b)) (Sum 0) $ fmap Sum $ Stream.fromList [1..10]
445-- This is the streaming equivalent of monad like sequenced application of
446-- parsers where next parser is dependent on the previous parser.
450{-# INLINE parseIterate #-}
452 :: (IsStream t, Monad m)
453 => (b -> Parser a m b)
456 -> t m (Either ParseError b)
457parseIterate f i m = fromStreamD $
458 D.parseIterateD f i (toStreamD m)
460------------------------------------------------------------------------------
462------------------------------------------------------------------------------
464-- | @groupsBy cmp f $ S.fromList [a,b,c,...]@ assigns the element @a@ to the
465-- first group, if @b \`cmp` a@ is 'True' then @b@ is also assigned to the same
466-- group. If @c \`cmp` a@ is 'True' then @c@ is also assigned to the same
467-- group and so on. When the comparison fails a new group is started. Each
468-- group is folded using the fold @f@ and the result of the fold is emitted in
471-- >>> Stream.toList $ Stream.groupsBy (>) Fold.toList $ Stream.fromList [1,3,7,0,2,5]
475{-# INLINE groupsBy #-}
477 :: (IsStream t, Monad m)
482groupsBy cmp f m = fromStreamD $ D.groupsBy cmp f (toStreamD m)
484-- | Unlike @groupsBy@ this function performs a rolling comparison of two
485-- successive elements in the input stream. @groupsByRolling cmp f $ S.fromList
486-- [a,b,c,...]@ assigns the element @a@ to the first group, if @a \`cmp` b@ is
487-- 'True' then @b@ is also assigned to the same group. If @b \`cmp` c@ is
488-- 'True' then @c@ is also assigned to the same group and so on. When the
489-- comparison fails a new group is started. Each group is folded using the fold
492-- >>> Stream.toList $ Stream.groupsByRolling (\a b -> a + 1 == b) Fold.toList $ Stream.fromList [1,2,3,7,8,9]
496{-# INLINE groupsByRolling #-}
498 :: (IsStream t, Monad m)
503groupsByRolling cmp f m = fromStreamD $ D.groupsRollingBy cmp f (toStreamD m)
506-- > groups = groupsBy (==)
507-- > groups = groupsByRolling (==)
509-- Groups contiguous spans of equal elements together in individual groups.
511-- >>> Stream.toList $ Stream.groups Fold.toList $ Stream.fromList [1,1,2,2]
516groups :: (IsStream t, Monad m, Eq a) => Fold m a b -> t m a -> t m b
517groups = groupsBy (==)
519------------------------------------------------------------------------------
520-- Splitting - by a predicate
521------------------------------------------------------------------------------
523-- In general we can use deintercalate for splitting. Then we can also use
524-- uniqBy to condense the separators. One way to generalize splitting is to
527-- data Segment a b = Empty | Segment b | Separator a
529-- XXX splitOn and splitOnSuffix have a different behavior on an empty stream,
532-- | Split on an infixed separator element, dropping the separator. The
533-- supplied 'Fold' is applied on the split segments. Splits the stream on
534-- separator elements determined by the supplied predicate, separator is
535-- considered as infixed between two segments:
537-- >>> splitOn' p xs = Stream.toList $ Stream.splitOn p Fold.toList (Stream.fromList xs)
538-- >>> splitOn' (== '.') "a.b"
541-- An empty stream is folded to the default value of the fold:
543-- >>> splitOn' (== '.') ""
546-- If one or both sides of the separator are missing then the empty segment on
547-- that side is folded to the default output of the fold:
549-- >>> splitOn' (== '.') "."
552-- >>> splitOn' (== '.') ".a"
555-- >>> splitOn' (== '.') "a."
558-- >>> splitOn' (== '.') "a..b"
561-- splitOn is an inverse of intercalating single element:
563-- > Stream.intercalate (Stream.fromPure '.') Unfold.fromList . Stream.splitOn (== '.') Fold.toList === id
565-- Assuming the input stream does not contain the separator:
567-- > Stream.splitOn (== '.') Fold.toList . Stream.intercalate (Stream.fromPure '.') Unfold.fromList === id
571{-# INLINE splitOn #-}
573 :: (IsStream t, Monad m)
574 => (a -> Bool) -> Fold m a b -> t m a -> t m b
576 -- We can express the infix splitting in terms of optional suffix split
577 -- fold. After applying a suffix split fold repeatedly if the last segment
578 -- ends with a suffix then we need to return the default output of the fold
579 -- after that to make it an infix split.
581 -- Alternately, we can also express it using an optional prefix split fold.
582 -- If the first segment starts with a prefix then we need to emit the
583 -- default output of the fold before that to make it an infix split, and
584 -- then apply prefix split fold repeatedly.
586 -- Since a suffix split fold can be easily expressed using a
587 -- non-backtracking fold, we use that.
588 foldManyPost (FL.takeEndBy_ predicate f)
590-- | Split on a suffixed separator element, dropping the separator. The
591-- supplied 'Fold' is applied on the split segments.
593-- >>> splitOnSuffix' p xs = Stream.toList $ Stream.splitOnSuffix p Fold.toList (Stream.fromList xs)
594-- >>> splitOnSuffix' (== '.') "a.b."
597-- >>> splitOnSuffix' (== '.') "a."
600-- An empty stream results in an empty output stream:
602-- >>> splitOnSuffix' (== '.') ""
605-- An empty segment consisting of only a suffix is folded to the default output
608-- >>> splitOnSuffix' (== '.') "."
611-- >>> splitOnSuffix' (== '.') "a..b.."
614-- A suffix is optional at the end of the stream:
616-- >>> splitOnSuffix' (== '.') "a"
619-- >>> splitOnSuffix' (== '.') ".a"
622-- >>> splitOnSuffix' (== '.') "a.b"
625-- > lines = splitOnSuffix (== '\n')
627-- 'splitOnSuffix' is an inverse of 'intercalateSuffix' with a single element:
629-- > Stream.intercalateSuffix (Stream.fromPure '.') Unfold.fromList . Stream.splitOnSuffix (== '.') Fold.toList === id
631-- Assuming the input stream does not contain the separator:
633-- > Stream.splitOnSuffix (== '.') Fold.toList . Stream.intercalateSuffix (Stream.fromPure '.') Unfold.fromList === id
637{-# INLINE splitOnSuffix #-}
639 :: (IsStream t, Monad m)
640 => (a -> Bool) -> Fold m a b -> t m a -> t m b
641splitOnSuffix predicate f = foldMany (FL.takeEndBy_ predicate f)
643-- | Split on a prefixed separator element, dropping the separator. The
644-- supplied 'Fold' is applied on the split segments.
647-- > splitOnPrefix' p xs = Stream.toList $ Stream.splitOnPrefix p (Fold.toList) (Stream.fromList xs)
648-- > splitOnPrefix' (== '.') ".a.b"
652-- An empty stream results in an empty output stream:
654-- > splitOnPrefix' (== '.') ""
658-- An empty segment consisting of only a prefix is folded to the default output
662-- > splitOnPrefix' (== '.') "."
665-- > splitOnPrefix' (== '.') ".a.b."
668-- > splitOnPrefix' (== '.') ".a..b"
673-- A prefix is optional at the beginning of the stream:
676-- > splitOnPrefix' (== '.') "a"
679-- > splitOnPrefix' (== '.') "a.b"
683-- 'splitOnPrefix' is an inverse of 'intercalatePrefix' with a single element:
685-- > Stream.intercalatePrefix (Stream.fromPure '.') Unfold.fromList . Stream.splitOnPrefix (== '.') Fold.toList === id
687-- Assuming the input stream does not contain the separator:
689-- > Stream.splitOnPrefix (== '.') Fold.toList . Stream.intercalatePrefix (Stream.fromPure '.') Unfold.fromList === id
693{-# INLINE splitOnPrefix #-}
694splitOnPrefix :: -- (IsStream t, MonadCatch m) =>
695 (a -> Bool) -> Fold m a b -> t m a -> t m b
696splitOnPrefix _predicate _f = undefined
697 -- parseMany (Parser.sliceBeginBy predicate f)
699-- | Like 'splitOn' after stripping leading, trailing, and repeated separators.
700-- Therefore, @".a..b."@ with '.' as the separator would be parsed as
701-- @["a","b"]@. In other words, its like parsing words from whitespace
704-- >>> wordsBy' p xs = Stream.toList $ Stream.wordsBy p Fold.toList (Stream.fromList xs)
706-- >>> wordsBy' (== ',') ""
709-- >>> wordsBy' (== ',') ","
712-- >>> wordsBy' (== ',') ",a,,b,"
715-- > words = wordsBy isSpace
719-- It is equivalent to splitting in any of the infix/prefix/suffix styles
720-- followed by removal of empty segments.
721{-# INLINE wordsBy #-}
723 :: (IsStream t, Monad m)
724 => (a -> Bool) -> Fold m a b -> t m a -> t m b
725wordsBy predicate f m =
726 fromStreamD $ D.wordsBy predicate f (toStreamD m)
728-- | Like 'splitOnSuffix' but keeps the suffix attached to the resulting
731-- >>> splitWithSuffix' p xs = Stream.toList $ splitWithSuffix p Fold.toList (Stream.fromList xs)
733-- >>> splitWithSuffix' (== '.') ""
736-- >>> splitWithSuffix' (== '.') "."
739-- >>> splitWithSuffix' (== '.') "a"
742-- >>> splitWithSuffix' (== '.') ".a"
745-- >>> splitWithSuffix' (== '.') "a."
748-- >>> splitWithSuffix' (== '.') "a.b"
751-- >>> splitWithSuffix' (== '.') "a.b."
754-- >>> splitWithSuffix' (== '.') "a..b.."
755-- ["a.",".","b.","."]
759{-# INLINE splitWithSuffix #-}
761 :: (IsStream t, Monad m)
762 => (a -> Bool) -> Fold m a b -> t m a -> t m b
763splitWithSuffix predicate f = foldMany (FL.takeEndBy predicate f)
765------------------------------------------------------------------------------
766-- Splitting - on a delimiter sequence
767------------------------------------------------------------------------------
769-- Int list examples for splitOn:
771-- >>> splitList [] [1,2,3,3,4]
772-- > [[1],[2],[3],[3],[4]]
774-- >>> splitList [5] [1,2,3,3,4]
777-- >>> splitList [1] [1,2,3,3,4]
780-- >>> splitList [4] [1,2,3,3,4]
783-- >>> splitList [2] [1,2,3,3,4]
786-- >>> splitList [3] [1,2,3,3,4]
789-- >>> splitList [3,3] [1,2,3,3,4]
792-- >>> splitList [1,2,3,3,4] [1,2,3,3,4]
795-- This can be implemented easily using Rabin Karp
796-- | Split on any one of the given patterns.
800{-# INLINE splitOnAny #-}
801splitOnAny :: -- (IsStream t, Monad m, Unbox a, Integral a) =>
802 [Array a] -> Fold m a b -> t m a -> t m b
803splitOnAny _subseq _f _m =
804 undefined -- D.fromStreamD $ D.splitOnAny f subseq (D.toStreamD m)
806-- XXX use a non-monadic intersperse to remove the MonadAsync constraint.
807-- XXX Use two folds, one ring buffer fold for separator sequence and the other
808-- split consumer fold. The input is fed to the ring fold first and the
809-- rejected input is fed to the split fold. If the separator matches, the ring
810-- fold would consume all.
812-- | Like 'splitOnSeq' but splits the separator as well, as an infix token.
814-- >>> splitOn'_ pat xs = Stream.toList $ Stream.splitBySeq (Array.fromList pat) Fold.toList (Stream.fromList xs)
816-- >>> splitOn'_ "" "hello"
817-- ["h","","e","","l","","l","","o"]
819-- >>> splitOn'_ "hello" ""
822-- >>> splitOn'_ "hello" "hello"
825-- >>> splitOn'_ "x" "hello"
828-- >>> splitOn'_ "h" "hello"
831-- >>> splitOn'_ "o" "hello"
834-- >>> splitOn'_ "e" "hello"
837-- >>> splitOn'_ "l" "hello"
838-- ["he","l","","l","o"]
840-- >>> splitOn'_ "ll" "hello"
844{-# INLINE splitBySeq #-}
846 :: (IsStream t, MonadAsync m, Storable a, Unbox a, Enum a, Eq a)
847 => Array a -> Fold m a b -> t m a -> t m b
849 intersperseM (fold f (fromStreamD $ A.read patt)) $ splitOnSeq patt f m
851-- | Like 'splitSuffixBy' but the separator is a sequence of elements, instead
852-- of a predicate for a single element.
854-- >>> splitOnSuffixSeq_ pat xs = Stream.toList $ Stream.splitOnSuffixSeq (Array.fromList pat) Fold.toList (Stream.fromList xs)
856-- >>> splitOnSuffixSeq_ "." ""
859-- >>> splitOnSuffixSeq_ "." "."
862-- >>> splitOnSuffixSeq_ "." "a"
865-- >>> splitOnSuffixSeq_ "." ".a"
868-- >>> splitOnSuffixSeq_ "." "a."
871-- >>> splitOnSuffixSeq_ "." "a.b"
874-- >>> splitOnSuffixSeq_ "." "a.b."
877-- >>> splitOnSuffixSeq_ "." "a..b.."
880-- > lines = splitOnSuffixSeq "\n"
882-- 'splitOnSuffixSeq' is an inverse of 'intercalateSuffix'. The following law
885-- > intercalateSuffix . splitOnSuffixSeq == id
887-- The following law holds when the separator is non-empty and contains none of
888-- the elements present in the input lists:
890-- > splitSuffixOn . intercalateSuffix == id
892-- >>> splitOnSuffixSeq pat f = Stream.foldMany (Fold.takeEndBySeq_ pat f)
895{-# INLINE splitOnSuffixSeq #-}
897 :: (IsStream t, MonadIO m, Storable a, Unbox a, Enum a, Eq a)
898 => Array a -> Fold m a b -> t m a -> t m b
899splitOnSuffixSeq patt f m =
900 fromStreamD $ D.splitOnSuffixSeq False patt f (toStreamD m)
901-- XXX This may have a problem if the stream terminates and we start extracting
902-- the fold and then the fold terminates before consuming all the buffered
903-- input. We have no way to supply the buffered input back to the driver.
904-- splitOnSuffixSeq patt f =
905-- foldMany (FL.takeEndBySeq_ patt f)
907-- | Like 'splitOn' but drops any empty splits.
910{-# INLINE wordsOn #-}
911wordsOn :: -- (IsStream t, Monad m, Unbox a, Eq a) =>
912 Array a -> Fold m a b -> t m a -> t m b
913wordsOn _subseq _f _m =
914 undefined -- D.fromStreamD $ D.wordsOn f subseq (D.toStreamD m)
916-- | Like 'splitOnSuffixSeq' but keeps the suffix intact in the splits.
918-- >>> splitWithSuffixSeq' pat xs = Stream.toList $ Stream.splitWithSuffixSeq (Array.fromList pat) Fold.toList (Stream.fromList xs)
920-- >>> splitWithSuffixSeq' "." ""
923-- >>> splitWithSuffixSeq' "." "."
926-- >>> splitWithSuffixSeq' "." "a"
929-- >>> splitWithSuffixSeq' "." ".a"
932-- >>> splitWithSuffixSeq' "." "a."
935-- >>> splitWithSuffixSeq' "." "a.b"
938-- >>> splitWithSuffixSeq' "." "a.b."
941-- >>> splitWithSuffixSeq' "." "a..b.."
942-- ["a.",".","b.","."]
944-- >>> splitWithSuffixSeq pat f = Stream.foldMany (Fold.takeEndBySeq pat f)
947{-# INLINE splitWithSuffixSeq #-}
949 :: (IsStream t, MonadIO m, Storable a, Unbox a, Enum a, Eq a)
950 => Array a -> Fold m a b -> t m a -> t m b
951splitWithSuffixSeq patt f m =
952 fromStreamD $ D.splitOnSuffixSeq True patt f (toStreamD m)
953-- splitWithSuffixSeq patt f =
954-- foldMany (FL.takeEndBySeq patt f)
956-- This can be implemented easily using Rabin Karp
957-- | Split post any one of the given patterns.
960{-# INLINE splitOnSuffixSeqAny #-}
961splitOnSuffixSeqAny :: -- (IsStream t, Monad m, Unbox a, Integral a) =>
962 [Array a] -> Fold m a b -> t m a -> t m b
963splitOnSuffixSeqAny _subseq _f _m = undefined
964 -- D.fromStreamD $ D.splitPostAny f subseq (D.toStreamD m)
966------------------------------------------------------------------------------
968------------------------------------------------------------------------------
970-- | Group the input stream into groups of @n@ elements each and then fold each
971-- group using the provided fold function.
973-- >>> Stream.toList $ Stream.chunksOf 2 Fold.sum (Stream.enumerateFromTo 1 10)
976-- This can be considered as an n-fold version of 'take' where we apply
977-- 'take' repeatedly on the leftover stream until the stream exhausts.
979-- @chunksOf n f = foldMany (FL.take n f)@
982{-# INLINE chunksOf #-}
984 :: (IsStream t, Monad m)
985 => Int -> Fold m a b -> t m a -> t m b
986chunksOf n f = fromStreamD . D.groupsOf n f . toStreamD
988-- | @arraysOf n stream@ groups the elements in the input stream into arrays of
991-- Same as the following but may be more efficient:
993-- > arraysOf n = Stream.foldMany (A.writeN n)
996{-# INLINE arraysOf #-}
997arraysOf :: (IsStream t, MonadIO m, Unbox a)
998 => Int -> t m a -> t m (Array a)
999arraysOf n = fromStreamD . A.chunksOf n . toStreamD
1001-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
1002-- XXX add this example after fixing the serial stream rate control
1004-- | Group the input stream into windows of @n@ second each and then fold each
1005-- group using the provided fold function.
1007-- >>> Stream.toList $ Stream.take 5 $ Stream.intervalsOf 1 Fold.sum $ Stream.constRate 2 $ Stream.enumerateFrom 1
1008-- [...,...,...,...,...]
1011{-# INLINE intervalsOf #-}
1013 :: (IsStream t, MonadAsync m)
1014 => Double -> Fold m a b -> t m a -> t m b
1016 splitWithSuffix isNothing (FL.catMaybes f)
1017 (interjectSuffix n (return Nothing) (map Just xs))
1019-- XXX This can be implemented more efficiently by sharing a Clock.Timer across
1020-- parallel threads and resetting it whenever a span is emitted.
1022-- | Like 'chunksOf' but if the chunk is not completed within the specified
1023-- time interval then emit whatever we have collected till now. The chunk
1024-- timeout is reset whenever a chunk is emitted. The granularity of the clock
1027-- >>> s = Stream.delayPost 0.3 $ Stream.fromList [1..1000]
1028-- >>> f = Stream.mapM_ print $ Stream.chunksOfTimeout 5 1 Fold.toList s
1031{-# INLINE chunksOfTimeout #-}
1032chunksOfTimeout :: (IsStream t, MonadAsync m, Functor (t m))
1033 => Int -> Double -> FL.Fold m a b -> t m a -> t m b
1034chunksOfTimeout n timeout f =
1036 . classifySessionsBy
1037 0.1 False (const (return False)) timeout (FL.take n f)
1038 . Transform.timestamped
1041------------------------------------------------------------------------------
1042-- Windowed classification
1043------------------------------------------------------------------------------
1045-- TODO: To mark the position in space or time we can have Indexed or
1046-- TimeStamped types. That can make it easy to deal with the position indices
1049------------------------------------------------------------------------------
1050-- Keyed Sliding Windows
1051------------------------------------------------------------------------------
1054{-# INLINABLE classifySlidingChunks #-}
1055classifySlidingChunks
1056 :: (IsStream t, MonadAsync m, Ord k)
1057 => Int -- ^ window size
1058 -> Int -- ^ window slide
1059 -> Fold m a b -- ^ Fold to be applied to window events
1060 -> t m (k, a, Bool) -- ^ window key, data, close event
1062classifySlidingChunks wsize wslide (Fold step initial extract) str
1065-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
1066-- slide the send window when an ack is received and we slide the receive
1067-- window when a sequence is complete. Sliding is stateful in case of TCP,
1068-- sliding releases the send buffer or makes data available to the user from
1069-- the receive buffer.
1070{-# INLINABLE classifySlidingSessions #-}
1071classifySlidingSessions
1072 :: (IsStream t, MonadAsync m, Ord k)
1073 => Double -- ^ timer tick in seconds
1074 -> Double -- ^ time window size
1075 -> Double -- ^ window slide
1076 -> Fold m a b -- ^ Fold to be applied to window events
1077 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
1079classifySlidingSessions tick interval slide (Fold step initial extract) str
1083------------------------------------------------------------------------------
1084-- Sliding Window Buffers
1085------------------------------------------------------------------------------
1087-- These buffered versions could be faster than concurrent incremental folds of
1088-- all overlapping windows as in many cases we may not need all the values to
1089-- compute the fold, we can just compute the result using the old value and new
1090-- value. However, we may need the buffer once in a while, for example for
1091-- string search we usually compute the hash incrementally but when the hash
1092-- matches the hash of the pattern we need to compare the whole string.
1094-- XXX we should be able to implement sequence based splitting combinators
1095-- using this combinator.
1098-- | Buffer n elements of the input in a ring buffer. When t new elements are
1099-- collected, slide the window to remove the same number of oldest elements,
1100-- insert the new elements, and apply an incremental fold on the sliding
1101-- window, supplying the outgoing elements, the new ring buffer as arguments.
1103 :: (IsStream t, Monad m, Ord a, Unbox a)
1104 => Int -- window size
1105 -> Int -- window slide
1106 -> Fold m (Ring a, Array a) b
1109slidingChunkBuffer = undefined
1111-- Buffer n seconds worth of stream elements of the input in a radix tree.
1112-- Every t seconds, remove the items that are older than n seconds, and apply
1113-- an incremental fold on the sliding window, supplying the outgoing elements,
1114-- and the new radix tree buffer as arguments.
1116 :: (IsStream t, Monad m, Ord a, Unbox a)
1117 => Int -- window size
1119 -> Fold m (RTree a, Array a) b
1122slidingSessionBuffer = undefined
1125------------------------------------------------------------------------------
1126-- Keyed Session Windows
1127------------------------------------------------------------------------------
1130-- | Keyed variable size space windows. Close the window if we do not receive a
1131-- window event in the next "spaceout" elements.
1132{-# INLINABLE classifyChunksBy #-}
1134 :: (IsStream t, MonadAsync m, Ord k)
1135 => Int -- ^ window spaceout (spread)
1136 -> Bool -- ^ reset the spaceout when a chunk window element is received
1137 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1138 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1140classifyChunksBy spanout reset (Fold step initial extract) str = undefined
1142-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
1143-- received within the chunk size window. The chunk gets closed only if no
1144-- element is received within the chunk window.
1146{-# INLINABLE classifyKeepAliveChunks #-}
1147classifyKeepAliveChunks
1148 :: (IsStream t, MonadAsync m, Ord k)
1149 => Int -- ^ window spaceout (spread)
1150 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1151 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1153classifyKeepAliveChunks spanout = classifyChunksBy spanout True
1156data SessionState t m f s b = SessionState
1157 { sessionCurTime :: !AbsTime -- ^ time since last event
1158 , sessionEventTime :: !AbsTime -- ^ time as per last event
1159 -- We can use the Map size instead of maintaining a count, but if we have
1160 -- to switch to HashMap then it can be useful.
1161 , sessionCount :: !Int -- ^ total number of sessions in progress
1162 , sessionTimerHeap :: H.Heap (H.Entry AbsTime (Key f)) -- ^ heap for timeouts
1163 , sessionKeyValueMap :: f s -- ^ Stored sessions for keys
1164 , sessionOutputStream :: t (m :: Type -> Type) (Key f, b) -- ^ Completed sessions
1167data SessionEntry s = LiveSession !AbsTime !s | ZombieSession
1169-- delete from map and output the fold accumulator
1170ejectEntry :: (Monad m, IsStream t, IsMap f) =>
1178 -> m (heap, f entry, t m (Key f, b), Int)
1179ejectEntry extract hp mp out cnt acc key = do
1181 let out1 = (key, sess) `cons` out
1182 let mp1 = IsMap.mapDelete key mp
1183 return (hp, mp1, out1, cnt - 1)
1185{-# NOINLINE flush #-}
1186flush :: (Monad m, IsMap f, IsStream t) =>
1188 -> SessionState t m f (SessionEntry s) b
1189 -> m (SessionState t m f (SessionEntry s) b)
1190flush extract session@SessionState{..} = do
1191 (hp', mp', out, count) <-
1194 , sessionKeyValueMap
1199 { sessionCount = count
1200 , sessionTimerHeap = hp'
1201 , sessionKeyValueMap = mp'
1202 , sessionOutputStream = out
1207 ejectAll (hp, mp, out, !cnt) = do
1208 let hres = H.uncons hp
1210 Just (Entry _ key, hp1) -> do
1211 r <- case IsMap.mapLookup key mp of
1212 Nothing -> return (hp1, mp, out, cnt)
1213 Just ZombieSession ->
1214 return (hp1, IsMap.mapDelete key mp, out, cnt)
1215 Just (LiveSession _ acc) ->
1216 ejectEntry extract hp1 mp out cnt acc key
1219 assert (IsMap.mapNull mp) (return ())
1220 return (hp, mp, out, cnt)
1222{-# NOINLINE ejectOne #-}
1223ejectOne :: (IsMap f, IsStream t, Monad m) =>
1226 -> ( H.Heap (Entry AbsTime (Key f))
1227 , f (SessionEntry acc)
1231 -> m ( H.Heap (Entry AbsTime (Key f))
1232 , f (SessionEntry acc)
1233 , t m (Key f, b), Int
1235ejectOne reset extract = go
1239 go (hp, mp, out, !cnt) = do
1240 let hres = H.uncons hp
1242 Just (Entry expiry key, hp1) ->
1243 case IsMap.mapLookup key mp of
1244 Nothing -> go (hp1, mp, out, cnt)
1245 Just ZombieSession ->
1246 go (hp1, IsMap.mapDelete key mp, out, cnt)
1247 Just (LiveSession expiry1 acc) -> do
1248 if not reset || expiry1 <= expiry
1249 then ejectEntry extract hp1 mp out cnt acc key
1251 -- reset the session timeout and continue
1252 let hp2 = H.insert (Entry expiry1 key) hp1
1253 in go (hp2, mp, out, cnt)
1255 assert (IsMap.mapNull mp) (return ())
1256 return (hp, mp, out, cnt)
1258{-# NOINLINE ejectExpired #-}
1259ejectExpired :: (IsMap f, IsStream t, Monad m) =>
1263 -> SessionState t m f (SessionEntry acc) b
1265 -> m (SessionState t m f (SessionEntry acc) b)
1266ejectExpired reset ejectPred extract session@SessionState{..} curTime = do
1267 (hp', mp', out, count) <-
1269 sessionTimerHeap sessionKeyValueMap IsStream.nil sessionCount
1271 { sessionCurTime = curTime
1272 , sessionCount = count
1273 , sessionTimerHeap = hp'
1274 , sessionKeyValueMap = mp'
1275 , sessionOutputStream = out
1280 ejectLoop hp mp out !cnt = do
1281 let hres = H.uncons hp
1283 Just (Entry expiry key, hp1) -> do
1285 if curTime >= expiry
1286 then return (True, False)
1292 case IsMap.mapLookup key mp of
1293 Nothing -> ejectLoop hp1 mp out cnt
1294 Just ZombieSession ->
1295 ejectLoop hp1 (IsMap.mapDelete key mp) out cnt
1296 Just (LiveSession expiry1 acc) -> do
1297 if expiry1 <= curTime || not reset || force
1299 (hp2,mp1,out1,cnt1) <-
1300 ejectEntry extract hp1 mp out cnt acc key
1301 ejectLoop hp2 mp1 out1 cnt1
1303 -- reset the session timeout and continue
1304 let hp2 = H.insert (Entry expiry1 key) hp1
1305 in ejectLoop hp2 mp out cnt
1306 else return (hp, mp, out, cnt)
1308 assert (IsMap.mapNull mp) (return ())
1309 return (hp, mp, out, cnt)
1311-- XXX Use mutable IORef in accumulator
1312{-# INLINE classifySessionsByGeneric #-}
1313classifySessionsByGeneric
1314 :: forall t m f a b. (IsStream t, MonadAsync m, IsMap f)
1315 => Proxy (f :: (Type -> Type))
1316 -> Double -- ^ timer tick in seconds
1317 -> Bool -- ^ reset the timer when an event is received
1318 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
1319 -> Double -- ^ session timeout in seconds
1320 -> Fold m a b -- ^ Fold to be applied to session data
1321 -> t m (AbsTime, (Key f, a)) -- ^ timestamp, (session key, session data)
1322 -> t m (Key f, b) -- ^ session key, fold result
1323classifySessionsByGeneric _ tick reset ejectPred tmout
1324 (Fold step initial extract) input =
1326 (Unfold.lmap (toStreamK . sessionOutputStream) Unfold.fromStreamK)
1327 $ scanlMAfter' sstep (return szero) (flush extract)
1328 $ interjectSuffix tick (return Nothing)
1333 timeoutMs = toRelTime (round (tmout * 1000) :: MilliSecond64)
1334 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
1335 szero = SessionState
1336 { sessionCurTime = toAbsTime (0 :: MilliSecond64)
1337 , sessionEventTime = toAbsTime (0 :: MilliSecond64)
1339 , sessionTimerHeap = H.empty
1340 , sessionKeyValueMap = IsMap.mapEmpty :: f s
1341 , sessionOutputStream = IsStream.nil
1344 -- We can eject sessions based on the current session count to limit
1345 -- memory consumption. There are two possible strategies:
1347 -- 1) Eject old sessions or sessions beyond a certain/lower timeout
1348 -- threshold even before timeout, effectively reduce the timeout.
1349 -- 2) Drop creation of new sessions but keep accepting new events for the
1352 -- We use the first strategy as of now.
1354 -- Got a new stream input element
1355 sstep session@SessionState{..} (Just (timestamp, (key, value))) = do
1356 -- XXX instead of a heap we could use a timer wheel.
1358 -- XXX if the key is an Int, we can also use an IntMap for slightly
1359 -- better performance.
1363 -- Values for each key are collected in a map using the supplied fold.
1364 -- When we insert a key in the Map we insert an entry into the heap as
1365 -- well with the session expiry as the sort key. The Map entry
1366 -- consists of the fold result, and the expiry time of the session. If
1367 -- "reset" is True the expiry time is readjusted whenever a new event
1368 -- is processed. If the fold terminates and a new session is started
1369 -- for the same key the expiry time is set to the first timestamp of
1372 -- The heap must have at most one entry for any given key. The heap is
1373 -- processed periodically to remove the expired entries. We pick up an
1374 -- expired entry from the top of the heap and if the session has
1375 -- expired based on the expiry time in the Map entry then we remove the
1376 -- session from the Map and yield its fold output. Otherwise, we
1377 -- reinsert the entry into the heap based on the current expiry in the
1380 -- If an entry is removed from the Map and not removed from the heap
1381 -- and in the meantime it is inserted again in the Map (using the same
1382 -- key) then how do we avoid inserting multiple entries in the heap?
1383 -- For this reason we maintain the invariant that the Map entry is
1384 -- removed only when the heap entry is removed. Even if the fold has
1385 -- finished we still keep a dummy Map entry (ZombieSession) until the
1386 -- heap entry is removed. That way if we have a Map entry we do not
1387 -- insert a heap entry because we know it is already there.
1388 -- XXX The ZombieSession mechanism does not work as expected as we
1389 -- ignore ZombieSession when inserting a new entry. Anyway, we can
1390 -- remove this mechanism as at most only two heap entries may be
1391 -- created and they will be ultimately cleaned up.
1393 -- Heap processing needs the map and map processing needs the heap,
1394 -- therefore we cannot separate the two for modularity unless we have a
1395 -- way to achieve mutual recursion.
1397 let curTime = max sessionEventTime timestamp
1398 mOld = IsMap.mapLookup key sessionKeyValueMap
1400 -- deleting a key from the heap is expensive, so we never
1401 -- delete a key from heap, we just purge it from the Map and it
1402 -- gets purged from the heap on timeout. We just need an extra
1403 -- lookup in the Map when the key is purged from the heap, that
1404 -- should not be expensive.
1406 let (mp, cnt) = case mOld of
1407 Just (LiveSession _ _) ->
1409 key ZombieSession sessionKeyValueMap
1412 _ -> (sessionKeyValueMap, sessionCount)
1414 { sessionCurTime = curTime
1415 , sessionEventTime = curTime
1416 , sessionCount = cnt
1417 , sessionKeyValueMap = mp
1418 , sessionOutputStream = fromPure (key, fb)
1421 let expiry = addToAbsTime timestamp timeoutMs
1422 (hp1, mp1, out1, cnt1) <- do
1423 let vars = (sessionTimerHeap, sessionKeyValueMap,
1424 IsStream.nil, sessionCount)
1426 -- inserting new entry
1428 -- Eject a session from heap and map if needed
1429 eject <- ejectPred sessionCount
1430 (hp, mp, out, cnt) <-
1432 then ejectOne reset extract vars
1435 -- Insert the new session in heap
1436 let hp' = H.insert (Entry expiry key) hp
1437 in return (hp', mp, out, cnt + 1)
1438 -- updating old entry
1439 Just _ -> return vars
1441 let acc = LiveSession expiry fs1
1442 mp2 = IsMap.mapInsert key acc mp1
1443 return $ SessionState
1444 { sessionCurTime = curTime
1445 , sessionEventTime = curTime
1446 , sessionCount = cnt1
1447 , sessionTimerHeap = hp1
1448 , sessionKeyValueMap = mp2
1449 , sessionOutputStream = out1
1453 Just (LiveSession _ acc) -> return $ FL.Partial acc
1457 error $ "classifySessionsBy: "
1458 ++ "The supplied fold must consume at least one input"
1460 res <- step fs value
1462 FL.Done fb -> done fb
1463 FL.Partial fs1 -> partial fs1
1465 -- Got a timer tick event
1466 sstep sessionState@SessionState{..} Nothing =
1467 let curTime = addToAbsTime sessionCurTime tickMs
1468 in ejectExpired reset ejectPred extract sessionState curTime
1470-- | @classifySessionsBy tick keepalive predicate timeout fold stream@
1471-- classifies an input event @stream@ consisting of @(timestamp, (key,
1472-- value))@ into sessions based on the @key@, folding all the values
1473-- corresponding to the same key into a session using the supplied @fold@.
1475-- When the fold terminates or a @timeout@ occurs, a tuple consisting of the
1476-- session key and the folded value is emitted in the output stream. The
1477-- timeout is measured from the first event in the session. If the @keepalive@
1478-- option is set to 'True' the timeout is reset to 0 whenever an event is
1481-- The @timestamp@ in the input stream is an absolute time from some epoch,
1482-- characterizing the time when the input event was generated. The notion of
1483-- current time is maintained by a monotonic event time clock using the
1484-- timestamps seen in the input stream. The latest timestamp seen till now is
1485-- used as the base for the current time. When no new events are seen, a timer
1486-- is started with a clock resolution of @tick@ seconds. This timer is used to
1487-- detect session timeouts in the absence of new events.
1489-- To ensure an upper bound on the memory used the number of sessions can be
1490-- limited to an upper bound. If the ejection @predicate@ returns 'True', the
1491-- oldest session is ejected before inserting a new session.
1493-- When the stream ends any buffered sessions are ejected immediately.
1495-- If a session key is received even after a session has finished, another
1496-- session is created for that key.
1499-- Stream.mapM_ print
1500-- $ Stream.classifySessionsBy 1 False (const (return False)) 3 (Fold.take 3 Fold.toList)
1501-- $ Stream.timestamped
1502-- $ Stream.delay 0.1
1503-- $ Stream.fromList ((,) <$> [1,2,3] <*> ['a','b','c'])
1510{-# INLINE classifySessionsBy #-}
1512 :: (IsStream t, MonadAsync m, Ord k)
1513 => Double -- ^ timer tick in seconds
1514 -> Bool -- ^ reset the timer when an event is received
1515 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
1516 -> Double -- ^ session timeout in seconds
1517 -> Fold m a b -- ^ Fold to be applied to session data
1518 -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
1519 -> t m (k, b) -- ^ session key, fold result
1520classifySessionsBy = classifySessionsByGeneric (Proxy :: Proxy (Map k))
1522-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
1523-- option set to 'True'.
1526-- classifyKeepAliveSessions = classifySessionsBy 1 True
1531{-# INLINE classifyKeepAliveSessions #-}
1532classifyKeepAliveSessions ::
1533 (IsStream t, MonadAsync m, Ord k)
1534 => (Int -> m Bool) -- ^ predicate to eject sessions on session count
1535 -> Double -- ^ session inactive timeout
1536 -> Fold m a b -- ^ Fold to be applied to session payload data
1537 -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
1539classifyKeepAliveSessions = classifySessionsBy 1 True
1541------------------------------------------------------------------------------
1542-- Keyed tumbling windows
1543------------------------------------------------------------------------------
1545-- Tumbling windows is a special case of sliding windows where the window slide
1546-- is the same as the window size. Or it can be a special case of session
1547-- windows where the reset flag is set to False.
1549-- XXX instead of using the early termination flag in the stream, we can use an
1550-- early terminating fold instead.
1553-- | Split the stream into fixed size chunks of specified size. Within each
1554-- such chunk fold the elements in buckets identified by the keys. A particular
1555-- bucket fold can be terminated early if a closing flag is encountered in an
1556-- element for that key.
1559{-# INLINABLE classifyChunksOf #-}
1561 :: (IsStream t, MonadAsync m, Ord k)
1562 => Int -- ^ window size
1563 -> Fold m a b -- ^ Fold to be applied to window events
1564 -> t m (k, a, Bool) -- ^ window key, data, close event
1566classifyChunksOf wsize = classifyChunksBy wsize False
1569-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
1570-- option set to 'False'.
1573-- classifySessionsOf = classifySessionsBy 1 False
1578{-# INLINE classifySessionsOf #-}
1579classifySessionsOf ::
1580 (IsStream t, MonadAsync m, Ord k)
1581 => (Int -> m Bool) -- ^ predicate to eject sessions on session count
1582 -> Double -- ^ time window size
1583 -> Fold m a b -- ^ Fold to be applied to session data
1584 -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
1586classifySessionsOf = classifySessionsBy 1 False
1588------------------------------------------------------------------------------
1590------------------------------------------------------------------------------
1592-- | @splitInnerBy splitter joiner stream@ splits the inner containers @f a@ of
1593-- an input stream @t m (f a)@ using the @splitter@ function. Container
1594-- elements @f a@ are collected until a split occurs, then all the elements
1595-- before the split are joined using the @joiner@ function.
1597-- For example, if we have a stream of @Array Word8@, we may want to split the
1598-- stream into arrays representing lines separated by '\n' byte such that the
1599-- resulting stream after a split would be one array for each line.
1601-- CAUTION! This is not a true streaming function as the container size after
1602-- the split and merge may not be bounded.
1605{-# INLINE splitInnerBy #-}
1607 :: (IsStream t, Monad m)
1608 => (f a -> m (f a, Maybe (f a))) -- splitter
1609 -> (f a -> f a -> m (f a)) -- joiner
1612splitInnerBy splitter joiner xs =
1613 fromStreamD $ D.splitInnerBy splitter joiner $ toStreamD xs
1615-- | Like 'splitInnerBy' but splits assuming the separator joins the segment in
1619{-# INLINE splitInnerBySuffix #-}
1621 :: (IsStream t, Monad m, Eq (f a), Monoid (f a))
1622 => (f a -> m (f a, Maybe (f a))) -- splitter
1623 -> (f a -> f a -> m (f a)) -- joiner
1626splitInnerBySuffix splitter joiner xs =
1627 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