2-- Module : Streamly.Internal.Data.Stream.IsStream.Reduce
3-- Copyright : (c) 2017 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
9-- Reduce streams by streams, folds or parsers.
11module Streamly.Internal.Data.Stream.IsStream.Reduce
13 -- * Reduce By Streams
20 -- Reduce a stream by folding or parsing chunks of the stream. Functions
21 -- generally ending in these shapes:
24 -- f (Fold m a b) -> t m a -> t m b
25 -- f (Parser m a b) -> t m a -> t m b
29 -- | Apply folds on a stream.
38 -- | Element unaware grouping.
45 -- | Streams can be sliced into segments in space or in time. We use the
46 -- term @chunk@ to refer to a spatial length of the stream (spatial window)
47 -- and the term @session@ to refer to a length in time (time window).
49 -- -- *** Using Element Separators
59 -- -- *** Splitting By Sequences
65 -- Keeping the delimiters
70 -- Splitting using multiple sequence separators
73 -- , splitOnAnyPrefixSeq
75 -- -- *** Splitting By Streams
76 -- -- | Splitting a stream using another stream as separator.
78 -- ** Keyed Window Classification
80 -- | Split the stream into chunks or windows by position or time. Each
81 -- window can be associated with a key, all events associated with a
82 -- particular key in the window can be folded to a single result. The
83 -- window termination can be dynamically controlled by the fold.
85 -- The term "chunk" is used for a window defined by position of elements
86 -- and the term "session" is used for a time window.
88 -- *** Tumbling Windows
89 -- | A new window starts after the previous window is finished.
92 , classifySessionsByGeneric
96 -- *** Keep Alive Windows
97 -- | The window size is extended if an event arrives within the specified
98 -- window size. This can represent sessions with idle or inactive timeout.
100 -- , classifyKeepAliveChunks
101 , classifyKeepAliveSessions
104 -- *** Sliding Windows
105 -- | A new window starts after the specified slide from the previous
106 -- window. Therefore windows can overlap.
107 , classifySlidingChunks
108 , classifySlidingSessions
109 -- *** Sliding Window Buffers
110 -- , slidingChunkBuffer
111 -- , slidingSessionBuffer
114 -- * Reduce By Parsers
115 -- ** Generic Parsing
116 -- | Apply parsers on a stream.
124 -- In imperative terms, grouped folding can be considered as a nested loop
125 -- where we loop over the stream to group elements and then loop over
126 -- individual groups to fold them to a single value that is yielded in the
129 , wordsBy -- stripAndCompactBy
135 -- -- *** Searching Sequences
136 -- , seqIndices -- search a sequence in the stream
138 -- -- *** Searching Multiple Sequences
139 -- , seqIndicesAny -- search any of the given sequence in the stream
141 -- -- -- ** Searching Streams
142 -- -- | Finding a stream within another stream.
144 -- * Nested splitting
152import Control.Exception (assert)
153import Control.Monad.Catch (MonadThrow)
154import Control.Monad.IO.Class (MonadIO(..))
155import Data.Heap (Entry(..))
156import Data.Kind (Type)
158import Data.Maybe (isNothing)
159import Data.Proxy (Proxy(..))
160import Foreign.Storable (Storable)
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 (..))
166import Streamly.Internal.Data.Array.Foreign.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
182import qualified Data.Heap as H
183import qualified Streamly.Internal.Data.Array.Foreign.Type as A
184import qualified Streamly.Internal.Data.Fold as FL
185import qualified Streamly.Internal.Data.IsMap as IsMap
186import qualified Streamly.Internal.Data.Parser.ParserD as PRD
187import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
188import qualified Streamly.Internal.Data.Stream.StreamD as D
189import qualified Streamly.Internal.Data.Stream.IsStream.Expand as Expand
190import qualified Streamly.Internal.Data.Stream.IsStream.Transform as Transform
191import qualified Streamly.Internal.Data.Unfold as Unfold
193import Prelude hiding (concatMap, map)
197-- >>> import Prelude hiding (zipWith, concatMap, concat)
198-- >>> import qualified Streamly.Prelude as Stream
199-- >>> import Streamly.Internal.Data.Stream.IsStream as Stream
200-- >>> import qualified Streamly.Data.Fold as Fold
201-- >>> import qualified Streamly.Internal.Data.Fold as Fold
202-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
203-- >>> import qualified Streamly.Internal.Data.Parser as Parser
204-- >>> import qualified Streamly.Data.Array.Foreign as Array
206------------------------------------------------------------------------------
208------------------------------------------------------------------------------
210-- | Drop prefix from the input stream if present.
215{-# INLINE dropPrefix #-}
217 -- (Eq a, IsStream t, Monad m) =>
218 t m a -> t m a -> t m a
219dropPrefix = error "Not implemented yet!"
221-- | Drop all matching infix from the input stream if present. Infix stream
222-- may be consumed multiple times.
224-- Space: @O(n)@ where n is the length of the infix.
227{-# INLINE dropInfix #-}
229 -- (Eq a, IsStream t, Monad m) =>
230 t m a -> t m a -> t m a
231dropInfix = error "Not implemented yet!"
233-- | Drop suffix from the input stream if present. Suffix stream may be
234-- consumed multiple times.
236-- Space: @O(n)@ where n is the length of the suffix.
239{-# INLINE dropSuffix #-}
241 -- (Eq a, IsStream t, Monad m) =>
242 t m a -> t m a -> t m a
243dropSuffix = error "Not implemented yet!"
245------------------------------------------------------------------------------
247------------------------------------------------------------------------------
249-- Splitting operations that take a predicate and a Fold can be
250-- expressed using parseMany. Operations like chunksOf, intervalsOf, split*,
251-- can be expressed using parseMany when used with an appropriate Parser.
253-- XXX We need takeGE/takeBetween to implement "some" using "many".
255-- | Apply a 'Fold' repeatedly on a stream and emit the fold outputs in the
258-- To sum every two contiguous elements in a stream:
260-- >>> f = Fold.take 2 Fold.sum
261-- >>> Stream.toList $ Stream.foldMany f $ Stream.fromList [1..10]
264-- On an empty stream the output is empty:
266-- >>> Stream.toList $ Stream.foldMany f $ Stream.fromList []
269-- Note @Stream.foldMany (Fold.take 0)@ would result in an infinite loop in a
274{-# INLINE foldMany #-}
276 :: (IsStream t, Monad m)
280foldMany f m = fromStreamD $ D.foldMany f (toStreamD m)
282-- | Like 'foldMany' but using the 'Refold' type instead of 'Fold'.
285{-# INLINE refoldMany #-}
286refoldMany :: (IsStream t, Monad m) =>
287 Refold m c a b -> m c -> t m a -> t m b
288refoldMany f action = fromStreamD . D.refoldMany f action . toStreamD
290-- | Apply a stream of folds to an input stream and emit the results in the
295{-# INLINE foldSequence #-}
297 :: -- (IsStream t, Monad m) =>
301foldSequence _f _m = undefined
303-- | Iterate a fold generator on a stream. The initial value @b@ is used to
304-- generate the first fold, the fold is applied on the stream and the result of
305-- the fold is used to generate the next fold and so on.
308-- >>> import Data.Monoid (Sum(..))
309-- >>> f x = return (Fold.take 2 (Fold.sconcat x))
310-- >>> s = Stream.map Sum $ Stream.fromList [1..10]
311-- >>> Stream.toList $ Stream.map getSum $ Stream.foldIterateM f (pure 0) s
316-- This is the streaming equivalent of monad like sequenced application of
317-- folds where next fold is dependent on the previous fold.
321{-# INLINE foldIterateM #-}
323 (IsStream t, Monad m) => (b -> m (Fold m a b)) -> m b -> t m a -> t m b
324foldIterateM f i m = fromStreamD $ D.foldIterateM f i (toStreamD m)
326-- | Like 'foldIterateM' but using the 'Refold' type instead. This could be
327-- much more efficient due to stream fusion.
330{-# INLINE refoldIterateM #-}
331refoldIterateM :: (IsStream t, Monad m) =>
332 Refold m b a b -> m b -> t m a -> t m b
333refoldIterateM c i m = fromStreamD $ D.refoldIterateM c i (toStreamD m)
335------------------------------------------------------------------------------
337------------------------------------------------------------------------------
339-- | Apply a 'Parser' repeatedly on a stream and emit the parsed values in the
342-- This is the streaming equivalent of the 'Streamly.Internal.Data.Parser.many'
345-- >>> Stream.toList $ Stream.parseMany (Parser.takeBetween 0 2 Fold.sum) $ Stream.fromList [1..10]
349-- > Stream.toList $ Stream.parseMany (Parser.line Fold.toList) $ Stream.fromList "hello\\nworld"
350-- ["hello\\n","world"]
355-- foldMany f = parseMany (fromFold f)
358-- Known Issues: When the parser fails there is no way to get the remaining
363{-# INLINE parseMany #-}
365 :: (IsStream t, MonadThrow m)
370 fromStreamD $ D.parseMany (PRD.fromParserK p) (toStreamD m)
372-- | Same as parseMany but for StreamD streams.
376{-# INLINE parseManyD #-}
378 :: (IsStream t, MonadThrow m)
383 fromStreamD $ D.parseMany p (toStreamD m)
385-- | Apply a stream of parsers to an input stream and emit the results in the
390{-# INLINE parseSequence #-}
392 :: -- (IsStream t, Monad m) =>
396parseSequence _f _m = undefined
398-- XXX Change the parser arguments' order
400-- | @parseManyTill collect test stream@ tries the parser @test@ on the input,
401-- if @test@ fails it backtracks and tries @collect@, after @collect@ succeeds
402-- @test@ is tried again and so on. The parser stops when @test@ succeeds. The
403-- output of @test@ is discarded and the output of @collect@ is emitted in the
404-- output stream. The parser fails if @collect@ fails.
408{-# INLINE parseManyTill #-}
410 -- (IsStream t, MonadThrow m) =>
415parseManyTill = undefined
417-- | Iterate a parser generating function on a stream. The initial value @b@ is
418-- used to generate the first parser, the parser is applied on the stream and
419-- the result is used to generate the next parser and so on.
421-- >>> import Data.Monoid (Sum(..))
422-- >>> Stream.toList $ Stream.map getSum $ Stream.parseIterate (\b -> Parser.takeBetween 0 2 (Fold.sconcat b)) 0 $ Stream.map Sum $ Stream.fromList [1..10]
425-- This is the streaming equivalent of monad like sequenced application of
426-- parsers where next parser is dependent on the previous parser.
430{-# INLINE parseIterate #-}
432 :: (IsStream t, MonadThrow m)
433 => (b -> Parser m a b)
437parseIterate f i m = fromStreamD $
438 D.parseIterate (PRD.fromParserK . f) i (toStreamD m)
440------------------------------------------------------------------------------
442------------------------------------------------------------------------------
444-- | @groupsBy cmp f $ S.fromList [a,b,c,...]@ assigns the element @a@ to the
445-- first group, if @b \`cmp` a@ is 'True' then @b@ is also assigned to the same
446-- group. If @c \`cmp` a@ is 'True' then @c@ is also assigned to the same
447-- group and so on. When the comparison fails a new group is started. Each
448-- group is folded using the fold @f@ and the result of the fold is emitted in
451-- >>> Stream.toList $ Stream.groupsBy (>) Fold.toList $ Stream.fromList [1,3,7,0,2,5]
455{-# INLINE groupsBy #-}
457 :: (IsStream t, Monad m)
462groupsBy cmp f m = fromStreamD $ D.groupsBy cmp f (toStreamD m)
464-- | Unlike @groupsBy@ this function performs a rolling comparison of two
465-- successive elements in the input stream. @groupsByRolling cmp f $ S.fromList
466-- [a,b,c,...]@ assigns the element @a@ to the first group, if @a \`cmp` b@ is
467-- 'True' then @b@ is also assigned to the same group. If @b \`cmp` c@ is
468-- 'True' then @c@ is also assigned to the same group and so on. When the
469-- comparison fails a new group is started. Each group is folded using the fold
472-- >>> Stream.toList $ Stream.groupsByRolling (\a b -> a + 1 == b) Fold.toList $ Stream.fromList [1,2,3,7,8,9]
476{-# INLINE groupsByRolling #-}
478 :: (IsStream t, Monad m)
483groupsByRolling cmp f m = fromStreamD $ D.groupsRollingBy cmp f (toStreamD m)
486-- > groups = groupsBy (==)
487-- > groups = groupsByRolling (==)
489-- Groups contiguous spans of equal elements together in individual groups.
491-- >>> Stream.toList $ Stream.groups Fold.toList $ Stream.fromList [1,1,2,2]
496groups :: (IsStream t, Monad m, Eq a) => Fold m a b -> t m a -> t m b
497groups = groupsBy (==)
499------------------------------------------------------------------------------
500-- Splitting - by a predicate
501------------------------------------------------------------------------------
503-- In general we can use deintercalate for splitting. Then we can also use
504-- uniqBy to condense the separators. One way to generalize splitting is to
507-- data Segment a b = Empty | Segment b | Separator a
509-- XXX splitOn and splitOnSuffix have a different behavior on an empty stream,
512-- | Split on an infixed separator element, dropping the separator. The
513-- supplied 'Fold' is applied on the split segments. Splits the stream on
514-- separator elements determined by the supplied predicate, separator is
515-- considered as infixed between two segments:
517-- >>> splitOn' p xs = Stream.toList $ Stream.splitOn p Fold.toList (Stream.fromList xs)
518-- >>> splitOn' (== '.') "a.b"
521-- An empty stream is folded to the default value of the fold:
523-- >>> splitOn' (== '.') ""
526-- If one or both sides of the separator are missing then the empty segment on
527-- that side is folded to the default output of the fold:
529-- >>> splitOn' (== '.') "."
532-- >>> splitOn' (== '.') ".a"
535-- >>> splitOn' (== '.') "a."
538-- >>> splitOn' (== '.') "a..b"
541-- splitOn is an inverse of intercalating single element:
543-- > Stream.intercalate (Stream.fromPure '.') Unfold.fromList . Stream.splitOn (== '.') Fold.toList === id
545-- Assuming the input stream does not contain the separator:
547-- > Stream.splitOn (== '.') Fold.toList . Stream.intercalate (Stream.fromPure '.') Unfold.fromList === id
551{-# INLINE splitOn #-}
553 :: (IsStream t, Monad m)
554 => (a -> Bool) -> Fold m a b -> t m a -> t m b
556 -- We can express the infix splitting in terms of optional suffix split
557 -- fold. After applying a suffix split fold repeatedly if the last segment
558 -- ends with a suffix then we need to return the default output of the fold
559 -- after that to make it an infix split.
561 -- Alternately, we can also express it using an optional prefix split fold.
562 -- If the first segment starts with a prefix then we need to emit the
563 -- default output of the fold before that to make it an infix split, and
564 -- then apply prefix split fold repeatedly.
566 -- Since a suffix split fold can be easily expressed using a
567 -- non-backtracking fold, we use that.
568 foldManyPost (FL.takeEndBy_ predicate f)
570-- | Split on a suffixed separator element, dropping the separator. The
571-- supplied 'Fold' is applied on the split segments.
573-- >>> splitOnSuffix' p xs = Stream.toList $ Stream.splitOnSuffix p Fold.toList (Stream.fromList xs)
574-- >>> splitOnSuffix' (== '.') "a.b."
577-- >>> splitOnSuffix' (== '.') "a."
580-- An empty stream results in an empty output stream:
582-- >>> splitOnSuffix' (== '.') ""
585-- An empty segment consisting of only a suffix is folded to the default output
588-- >>> splitOnSuffix' (== '.') "."
591-- >>> splitOnSuffix' (== '.') "a..b.."
594-- A suffix is optional at the end of the stream:
596-- >>> splitOnSuffix' (== '.') "a"
599-- >>> splitOnSuffix' (== '.') ".a"
602-- >>> splitOnSuffix' (== '.') "a.b"
605-- > lines = splitOnSuffix (== '\n')
607-- 'splitOnSuffix' is an inverse of 'intercalateSuffix' with a single element:
609-- > Stream.intercalateSuffix (Stream.fromPure '.') Unfold.fromList . Stream.splitOnSuffix (== '.') Fold.toList === id
611-- Assuming the input stream does not contain the separator:
613-- > Stream.splitOnSuffix (== '.') Fold.toList . Stream.intercalateSuffix (Stream.fromPure '.') Unfold.fromList === id
617{-# INLINE splitOnSuffix #-}
619 :: (IsStream t, Monad m)
620 => (a -> Bool) -> Fold m a b -> t m a -> t m b
621splitOnSuffix predicate f = foldMany (FL.takeEndBy_ predicate f)
623-- | Split on a prefixed separator element, dropping the separator. The
624-- supplied 'Fold' is applied on the split segments.
627-- > splitOnPrefix' p xs = Stream.toList $ Stream.splitOnPrefix p (Fold.toList) (Stream.fromList xs)
628-- > splitOnPrefix' (== '.') ".a.b"
632-- An empty stream results in an empty output stream:
634-- > splitOnPrefix' (== '.') ""
638-- An empty segment consisting of only a prefix is folded to the default output
642-- > splitOnPrefix' (== '.') "."
645-- > splitOnPrefix' (== '.') ".a.b."
648-- > splitOnPrefix' (== '.') ".a..b"
653-- A prefix is optional at the beginning of the stream:
656-- > splitOnPrefix' (== '.') "a"
659-- > splitOnPrefix' (== '.') "a.b"
663-- 'splitOnPrefix' is an inverse of 'intercalatePrefix' with a single element:
665-- > Stream.intercalatePrefix (Stream.fromPure '.') Unfold.fromList . Stream.splitOnPrefix (== '.') Fold.toList === id
667-- Assuming the input stream does not contain the separator:
669-- > Stream.splitOnPrefix (== '.') Fold.toList . Stream.intercalatePrefix (Stream.fromPure '.') Unfold.fromList === id
673{-# INLINE splitOnPrefix #-}
674splitOnPrefix :: -- (IsStream t, MonadCatch m) =>
675 (a -> Bool) -> Fold m a b -> t m a -> t m b
676splitOnPrefix _predicate _f = undefined
677 -- parseMany (Parser.sliceBeginBy predicate f)
679-- | Like 'splitOn' after stripping leading, trailing, and repeated separators.
680-- Therefore, @".a..b."@ with '.' as the separator would be parsed as
681-- @["a","b"]@. In other words, its like parsing words from whitespace
684-- >>> wordsBy' p xs = Stream.toList $ Stream.wordsBy p Fold.toList (Stream.fromList xs)
686-- >>> wordsBy' (== ',') ""
689-- >>> wordsBy' (== ',') ","
692-- >>> wordsBy' (== ',') ",a,,b,"
695-- > words = wordsBy isSpace
699-- It is equivalent to splitting in any of the infix/prefix/suffix styles
700-- followed by removal of empty segments.
701{-# INLINE wordsBy #-}
703 :: (IsStream t, Monad m)
704 => (a -> Bool) -> Fold m a b -> t m a -> t m b
705wordsBy predicate f m =
706 fromStreamD $ D.wordsBy predicate f (toStreamD m)
708-- | Like 'splitOnSuffix' but keeps the suffix attached to the resulting
711-- >>> splitWithSuffix' p xs = Stream.toList $ splitWithSuffix p Fold.toList (Stream.fromList xs)
713-- >>> splitWithSuffix' (== '.') ""
716-- >>> splitWithSuffix' (== '.') "."
719-- >>> splitWithSuffix' (== '.') "a"
722-- >>> splitWithSuffix' (== '.') ".a"
725-- >>> splitWithSuffix' (== '.') "a."
728-- >>> splitWithSuffix' (== '.') "a.b"
731-- >>> splitWithSuffix' (== '.') "a.b."
734-- >>> splitWithSuffix' (== '.') "a..b.."
735-- ["a.",".","b.","."]
739{-# INLINE splitWithSuffix #-}
741 :: (IsStream t, Monad m)
742 => (a -> Bool) -> Fold m a b -> t m a -> t m b
743splitWithSuffix predicate f = foldMany (FL.takeEndBy predicate f)
745------------------------------------------------------------------------------
746-- Splitting - on a delimiter sequence
747------------------------------------------------------------------------------
749-- Int list examples for splitOn:
751-- >>> splitList [] [1,2,3,3,4]
752-- > [[1],[2],[3],[3],[4]]
754-- >>> splitList [5] [1,2,3,3,4]
757-- >>> splitList [1] [1,2,3,3,4]
760-- >>> splitList [4] [1,2,3,3,4]
763-- >>> splitList [2] [1,2,3,3,4]
766-- >>> splitList [3] [1,2,3,3,4]
769-- >>> splitList [3,3] [1,2,3,3,4]
772-- >>> splitList [1,2,3,3,4] [1,2,3,3,4]
775-- This can be implemented easily using Rabin Karp
776-- | Split on any one of the given patterns.
780{-# INLINE splitOnAny #-}
781splitOnAny :: -- (IsStream t, Monad m, Storable a, Integral a) =>
782 [Array a] -> Fold m a b -> t m a -> t m b
783splitOnAny _subseq _f _m =
784 undefined -- D.fromStreamD $ D.splitOnAny f subseq (D.toStreamD m)
786-- XXX use a non-monadic intersperse to remove the MonadAsync constraint.
787-- XXX Use two folds, one ring buffer fold for separator sequence and the other
788-- split consumer fold. The input is fed to the ring fold first and the
789-- rejected input is fed to the split fold. If the separator matches, the ring
790-- fold would consume all.
792-- | Like 'splitOnSeq' but splits the separator as well, as an infix token.
794-- >>> splitOn'_ pat xs = Stream.toList $ Stream.splitBySeq (Array.fromList pat) Fold.toList (Stream.fromList xs)
796-- >>> splitOn'_ "" "hello"
797-- ["h","","e","","l","","l","","o"]
799-- >>> splitOn'_ "hello" ""
802-- >>> splitOn'_ "hello" "hello"
805-- >>> splitOn'_ "x" "hello"
808-- >>> splitOn'_ "h" "hello"
811-- >>> splitOn'_ "o" "hello"
814-- >>> splitOn'_ "e" "hello"
817-- >>> splitOn'_ "l" "hello"
818-- ["he","l","","l","o"]
820-- >>> splitOn'_ "ll" "hello"
824{-# INLINE splitBySeq #-}
826 :: (IsStream t, MonadAsync m, Storable a, Enum a, Eq a)
827 => Array a -> Fold m a b -> t m a -> t m b
829 intersperseM (fold f (A.toStream patt)) $ splitOnSeq patt f m
831-- | Like 'splitSuffixBy' but the separator is a sequence of elements, instead
832-- of a predicate for a single element.
834-- >>> splitOnSuffixSeq_ pat xs = Stream.toList $ Stream.splitOnSuffixSeq (Array.fromList pat) Fold.toList (Stream.fromList xs)
836-- >>> splitOnSuffixSeq_ "." ""
839-- >>> splitOnSuffixSeq_ "." "."
842-- >>> splitOnSuffixSeq_ "." "a"
845-- >>> splitOnSuffixSeq_ "." ".a"
848-- >>> splitOnSuffixSeq_ "." "a."
851-- >>> splitOnSuffixSeq_ "." "a.b"
854-- >>> splitOnSuffixSeq_ "." "a.b."
857-- >>> splitOnSuffixSeq_ "." "a..b.."
860-- > lines = splitOnSuffixSeq "\n"
862-- 'splitOnSuffixSeq' is an inverse of 'intercalateSuffix'. The following law
865-- > intercalateSuffix . splitOnSuffixSeq == id
867-- The following law holds when the separator is non-empty and contains none of
868-- the elements present in the input lists:
870-- > splitSuffixOn . intercalateSuffix == id
872-- >>> splitOnSuffixSeq pat f = Stream.foldMany (Fold.takeEndBySeq_ pat f)
875{-# INLINE splitOnSuffixSeq #-}
877 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
878 => Array a -> Fold m a b -> t m a -> t m b
879splitOnSuffixSeq patt f m =
880 fromStreamD $ D.splitOnSuffixSeq False patt f (toStreamD m)
881-- XXX This may have a problem if the stream terminates and we start extracting
882-- the fold and then the fold terminates before consuming all the buffered
883-- input. We have no way to supply the buffered input back to the driver.
884-- splitOnSuffixSeq patt f =
885-- foldMany (FL.takeEndBySeq_ patt f)
887-- | Like 'splitOn' but drops any empty splits.
890{-# INLINE wordsOn #-}
891wordsOn :: -- (IsStream t, Monad m, Storable a, Eq a) =>
892 Array a -> Fold m a b -> t m a -> t m b
893wordsOn _subseq _f _m =
894 undefined -- D.fromStreamD $ D.wordsOn f subseq (D.toStreamD m)
896-- | Like 'splitOnSuffixSeq' but keeps the suffix intact in the splits.
898-- >>> splitWithSuffixSeq' pat xs = Stream.toList $ Stream.splitWithSuffixSeq (Array.fromList pat) Fold.toList (Stream.fromList xs)
900-- >>> splitWithSuffixSeq' "." ""
903-- >>> splitWithSuffixSeq' "." "."
906-- >>> splitWithSuffixSeq' "." "a"
909-- >>> splitWithSuffixSeq' "." ".a"
912-- >>> splitWithSuffixSeq' "." "a."
915-- >>> splitWithSuffixSeq' "." "a.b"
918-- >>> splitWithSuffixSeq' "." "a.b."
921-- >>> splitWithSuffixSeq' "." "a..b.."
922-- ["a.",".","b.","."]
924-- >>> splitWithSuffixSeq pat f = Stream.foldMany (Fold.takeEndBySeq pat f)
927{-# INLINE splitWithSuffixSeq #-}
929 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
930 => Array a -> Fold m a b -> t m a -> t m b
931splitWithSuffixSeq patt f m =
932 fromStreamD $ D.splitOnSuffixSeq True patt f (toStreamD m)
933-- splitWithSuffixSeq patt f =
934-- foldMany (FL.takeEndBySeq patt f)
936-- This can be implemented easily using Rabin Karp
937-- | Split post any one of the given patterns.
940{-# INLINE splitOnSuffixSeqAny #-}
941splitOnSuffixSeqAny :: -- (IsStream t, Monad m, Storable a, Integral a) =>
942 [Array a] -> Fold m a b -> t m a -> t m b
943splitOnSuffixSeqAny _subseq _f _m = undefined
944 -- D.fromStreamD $ D.splitPostAny f subseq (D.toStreamD m)
946------------------------------------------------------------------------------
948------------------------------------------------------------------------------
950-- | Group the input stream into groups of @n@ elements each and then fold each
951-- group using the provided fold function.
953-- >>> Stream.toList $ Stream.chunksOf 2 Fold.sum (Stream.enumerateFromTo 1 10)
956-- This can be considered as an n-fold version of 'take' where we apply
957-- 'take' repeatedly on the leftover stream until the stream exhausts.
959-- @chunksOf n f = foldMany (FL.take n f)@
962{-# INLINE chunksOf #-}
964 :: (IsStream t, Monad m)
965 => Int -> Fold m a b -> t m a -> t m b
966chunksOf n f = fromStreamD . D.chunksOf n f . toStreamD
968-- | @arraysOf n stream@ groups the elements in the input stream into arrays of
971-- Same as the following but may be more efficient:
973-- > arraysOf n = Stream.foldMany (A.writeN n)
976{-# INLINE arraysOf #-}
977arraysOf :: (IsStream t, MonadIO m, Storable a)
978 => Int -> t m a -> t m (Array a)
979arraysOf n = fromStreamD . A.arraysOf n . toStreamD
981-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
982-- XXX add this example after fixing the serial stream rate control
984-- | Group the input stream into windows of @n@ second each and then fold each
985-- group using the provided fold function.
987-- >>> Stream.toList $ Stream.take 5 $ Stream.intervalsOf 1 Fold.sum $ Stream.constRate 2 $ Stream.enumerateFrom 1
988-- [...,...,...,...,...]
991{-# INLINE intervalsOf #-}
993 :: (IsStream t, MonadAsync m)
994 => Double -> Fold m a b -> t m a -> t m b
996 splitWithSuffix isNothing (FL.catMaybes f)
997 (interjectSuffix n (return Nothing) (map Just xs))
999-- XXX This can be implemented more efficiently by sharing a Clock.Timer across
1000-- parallel threads and resetting it whenever a span is emitted.
1002-- | Like 'chunksOf' but if the chunk is not completed within the specified
1003-- time interval then emit whatever we have collected till now. The chunk
1004-- timeout is reset whenever a chunk is emitted.
1006-- >>> s = Stream.delayPost 0.3 $ Stream.fromList [1..1000]
1007-- >>> f = Stream.mapM_ print $ Stream.chunksOfTimeout 5 1 Fold.toList s
1010{-# INLINE chunksOfTimeout #-}
1011chunksOfTimeout :: (IsStream t, MonadAsync m, Functor (t m))
1012 => Int -> Double -> FL.Fold m a b -> t m a -> t m b
1013chunksOfTimeout n timeout f =
1015 . classifySessionsBy
1016 timeout False (const (return False)) timeout (FL.take n f)
1017 . Transform.timestamped
1020------------------------------------------------------------------------------
1021-- Windowed classification
1022------------------------------------------------------------------------------
1024-- TODO: To mark the position in space or time we can have Indexed or
1025-- TimeStamped types. That can make it easy to deal with the position indices
1028------------------------------------------------------------------------------
1029-- Keyed Sliding Windows
1030------------------------------------------------------------------------------
1033{-# INLINABLE classifySlidingChunks #-}
1034classifySlidingChunks
1035 :: (IsStream t, MonadAsync m, Ord k)
1036 => Int -- ^ window size
1037 -> Int -- ^ window slide
1038 -> Fold m a b -- ^ Fold to be applied to window events
1039 -> t m (k, a, Bool) -- ^ window key, data, close event
1041classifySlidingChunks wsize wslide (Fold step initial extract) str
1044-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
1045-- slide the send window when an ack is received and we slide the receive
1046-- window when a sequence is complete. Sliding is stateful in case of TCP,
1047-- sliding releases the send buffer or makes data available to the user from
1048-- the receive buffer.
1049{-# INLINABLE classifySlidingSessions #-}
1050classifySlidingSessions
1051 :: (IsStream t, MonadAsync m, Ord k)
1052 => Double -- ^ timer tick in seconds
1053 -> Double -- ^ time window size
1054 -> Double -- ^ window slide
1055 -> Fold m a b -- ^ Fold to be applied to window events
1056 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
1058classifySlidingSessions tick interval slide (Fold step initial extract) str
1062------------------------------------------------------------------------------
1063-- Sliding Window Buffers
1064------------------------------------------------------------------------------
1066-- These buffered versions could be faster than concurrent incremental folds of
1067-- all overlapping windows as in many cases we may not need all the values to
1068-- compute the fold, we can just compute the result using the old value and new
1069-- value. However, we may need the buffer once in a while, for example for
1070-- string search we usually compute the hash incrementally but when the hash
1071-- matches the hash of the pattern we need to compare the whole string.
1073-- XXX we should be able to implement sequence based splitting combinators
1074-- using this combinator.
1077-- | Buffer n elements of the input in a ring buffer. When t new elements are
1078-- collected, slide the window to remove the same number of oldest elements,
1079-- insert the new elements, and apply an incremental fold on the sliding
1080-- window, supplying the outgoing elements, the new ring buffer as arguments.
1082 :: (IsStream t, Monad m, Ord a, Storable a)
1083 => Int -- window size
1084 -> Int -- window slide
1085 -> Fold m (Ring a, Array a) b
1088slidingChunkBuffer = undefined
1090-- Buffer n seconds worth of stream elements of the input in a radix tree.
1091-- Every t seconds, remove the items that are older than n seconds, and apply
1092-- an incremental fold on the sliding window, supplying the outgoing elements,
1093-- and the new radix tree buffer as arguments.
1095 :: (IsStream t, Monad m, Ord a, Storable a)
1096 => Int -- window size
1098 -> Fold m (RTree a, Array a) b
1101slidingSessionBuffer = undefined
1104------------------------------------------------------------------------------
1105-- Keyed Session Windows
1106------------------------------------------------------------------------------
1109-- | Keyed variable size space windows. Close the window if we do not receive a
1110-- window event in the next "spaceout" elements.
1111{-# INLINABLE classifyChunksBy #-}
1113 :: (IsStream t, MonadAsync m, Ord k)
1114 => Int -- ^ window spaceout (spread)
1115 -> Bool -- ^ reset the spaceout when a chunk window element is received
1116 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1117 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1119classifyChunksBy spanout reset (Fold step initial extract) str = undefined
1121-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
1122-- received within the chunk size window. The chunk gets closed only if no
1123-- element is received within the chunk window.
1125{-# INLINABLE classifyKeepAliveChunks #-}
1126classifyKeepAliveChunks
1127 :: (IsStream t, MonadAsync m, Ord k)
1128 => Int -- ^ window spaceout (spread)
1129 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1130 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1132classifyKeepAliveChunks spanout = classifyChunksBy spanout True
1135data SessionState t m f s b = SessionState
1136 { sessionCurTime :: !AbsTime -- ^ time since last event
1137 , sessionEventTime :: !AbsTime -- ^ time as per last event
1138 -- We can use the Map size instead of maintaining a count, but if we have
1139 -- to switch to HashMap then it can be useful.
1140 , sessionCount :: !Int -- ^ total number of sessions in progress
1141 , sessionTimerHeap :: H.Heap (H.Entry AbsTime (Key f)) -- ^ heap for timeouts
1142 , sessionKeyValueMap :: f s -- ^ Stored sessions for keys
1143 , sessionOutputStream :: t (m :: Type -> Type) (Key f, b) -- ^ Completed sessions
1146data SessionEntry s = LiveSession !AbsTime !s | ZombieSession
1148-- delete from map and output the fold accumulator
1149ejectEntry :: (Monad m, IsStream t, IsMap f) =>
1157 -> m (heap, f entry, t m (Key f, b), Int)
1158ejectEntry extract hp mp out cnt acc key = do
1160 let out1 = (key, sess) `cons` out
1161 let mp1 = IsMap.mapDelete key mp
1162 return (hp, mp1, out1, cnt - 1)
1164{-# NOINLINE flush #-}
1165flush :: (Monad m, IsMap f, IsStream t) =>
1167 -> SessionState t m f (SessionEntry s) b
1168 -> m (SessionState t m f (SessionEntry s) b)
1169flush extract session@SessionState{..} = do
1170 (hp', mp', out, count) <-
1173 , sessionKeyValueMap
1178 { sessionCount = count
1179 , sessionTimerHeap = hp'
1180 , sessionKeyValueMap = mp'
1181 , sessionOutputStream = out
1186 ejectAll (hp, mp, out, !cnt) = do
1187 let hres = H.uncons hp
1189 Just (Entry _ key, hp1) -> do
1190 r <- case IsMap.mapLookup key mp of
1191 Nothing -> return (hp1, mp, out, cnt)
1192 Just ZombieSession ->
1193 return (hp1, IsMap.mapDelete key mp, out, cnt)
1194 Just (LiveSession _ acc) ->
1195 ejectEntry extract hp1 mp out cnt acc key
1198 assert (IsMap.mapNull mp) (return ())
1199 return (hp, mp, out, cnt)
1201{-# NOINLINE ejectOne #-}
1202ejectOne :: (IsMap f, IsStream t, Monad m) =>
1205 -> ( H.Heap (Entry AbsTime (Key f))
1206 , f (SessionEntry acc)
1210 -> m ( H.Heap (Entry AbsTime (Key f))
1211 , f (SessionEntry acc)
1212 , t m (Key f, b), Int
1214ejectOne reset extract = go
1218 go (hp, mp, out, !cnt) = do
1219 let hres = H.uncons hp
1221 Just (Entry expiry key, hp1) ->
1222 case IsMap.mapLookup key mp of
1223 Nothing -> go (hp1, mp, out, cnt)
1224 Just ZombieSession ->
1225 go (hp1, IsMap.mapDelete key mp, out, cnt)
1226 Just (LiveSession expiry1 acc) -> do
1227 if not reset || expiry1 <= expiry
1228 then ejectEntry extract hp1 mp out cnt acc key
1230 -- reset the session timeout and continue
1231 let hp2 = H.insert (Entry expiry1 key) hp1
1232 in go (hp2, mp, out, cnt)
1234 assert (IsMap.mapNull mp) (return ())
1235 return (hp, mp, out, cnt)
1237{-# NOINLINE ejectExpired #-}
1238ejectExpired :: (IsMap f, IsStream t, Monad m) =>
1242 -> SessionState t m f (SessionEntry acc) b
1244 -> m (SessionState t m f (SessionEntry acc) b)
1245ejectExpired reset ejectPred extract session@SessionState{..} curTime = do
1246 (hp', mp', out, count) <-
1248 sessionTimerHeap sessionKeyValueMap IsStream.nil sessionCount
1250 { sessionCurTime = curTime
1251 , sessionCount = count
1252 , sessionTimerHeap = hp'
1253 , sessionKeyValueMap = mp'
1254 , sessionOutputStream = out
1259 ejectLoop hp mp out !cnt = do
1260 let hres = H.uncons hp
1262 Just (Entry expiry key, hp1) -> do
1264 if curTime >= expiry
1265 then return (True, False)
1271 case IsMap.mapLookup key mp of
1272 Nothing -> ejectLoop hp1 mp out cnt
1273 Just ZombieSession ->
1274 ejectLoop hp1 (IsMap.mapDelete key mp) out cnt
1275 Just (LiveSession expiry1 acc) -> do
1276 if expiry1 <= curTime || not reset || force
1278 (hp2,mp1,out1,cnt1) <-
1279 ejectEntry extract hp1 mp out cnt acc key
1280 ejectLoop hp2 mp1 out1 cnt1
1282 -- reset the session timeout and continue
1283 let hp2 = H.insert (Entry expiry1 key) hp1
1284 in ejectLoop hp2 mp out cnt
1285 else return (hp, mp, out, cnt)
1287 assert (IsMap.mapNull mp) (return ())
1288 return (hp, mp, out, cnt)
1290-- XXX Use mutable IORef in accumulator
1291{-# INLINE classifySessionsByGeneric #-}
1292classifySessionsByGeneric
1293 :: forall t m f a b. (IsStream t, MonadAsync m, IsMap f)
1294 => Proxy (f :: (Type -> Type))
1295 -> Double -- ^ timer tick in seconds
1296 -> Bool -- ^ reset the timer when an event is received
1297 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
1298 -> Double -- ^ session timeout in seconds
1299 -> Fold m a b -- ^ Fold to be applied to session data
1300 -> t m (AbsTime, (Key f, a)) -- ^ timestamp, (session key, session data)
1301 -> t m (Key f, b) -- ^ session key, fold result
1302classifySessionsByGeneric _ tick reset ejectPred tmout
1303 (Fold step initial extract) input =
1304 Expand.unfoldMany (Unfold.lmap sessionOutputStream Unfold.fromStream)
1305 $ scanlMAfter' sstep (return szero) (flush extract)
1306 $ interjectSuffix tick (return Nothing)
1311 timeoutMs = toRelTime (round (tmout * 1000) :: MilliSecond64)
1312 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
1313 szero = SessionState
1314 { sessionCurTime = toAbsTime (0 :: MilliSecond64)
1315 , sessionEventTime = toAbsTime (0 :: MilliSecond64)
1317 , sessionTimerHeap = H.empty
1318 , sessionKeyValueMap = IsMap.mapEmpty :: f s
1319 , sessionOutputStream = IsStream.nil
1322 -- We can eject sessions based on the current session count to limit
1323 -- memory consumption. There are two possible strategies:
1325 -- 1) Eject old sessions or sessions beyond a certain/lower timeout
1326 -- threshold even before timeout, effectively reduce the timeout.
1327 -- 2) Drop creation of new sessions but keep accepting new events for the
1330 -- We use the first strategy as of now.
1332 -- Got a new stream input element
1333 sstep session@SessionState{..} (Just (timestamp, (key, value))) = do
1334 -- XXX instead of a heap we could use a timer wheel.
1336 -- XXX if the key is an Int, we can also use an IntMap for slightly
1337 -- better performance.
1341 -- Values for each key are collected in a map using the supplied fold.
1342 -- When we insert a key in the Map we insert an entry into the heap as
1343 -- well with the session expiry as the sort key. The Map entry
1344 -- consists of the fold result, and the expiry time of the session. If
1345 -- "reset" is True the expiry time is readjusted whenever a new event
1346 -- is processed. If the fold terminates and a new session is started
1347 -- for the same key the expiry time is set to the first timestamp of
1350 -- The heap must have at most one entry for any given key. The heap is
1351 -- processed periodically to remove the expired entries. We pick up an
1352 -- expired entry from the top of the heap and if the session has
1353 -- expired based on the expiry time in the Map entry then we remove the
1354 -- session from the Map and yield its fold output. Otherwise, we
1355 -- reinsert the entry into the heap based on the current expiry in the
1358 -- If an entry is removed from the Map and not removed from the heap
1359 -- and in the meantime it is inserted again in the Map (using the same
1360 -- key) then how do we avoid inserting multiple entries in the heap?
1361 -- For this reason we maintain the invariant that the Map entry is
1362 -- removed only when the heap entry is removed. Even if the fold has
1363 -- finished we still keep a dummy Map entry (ZombieSession) until the
1364 -- heap entry is removed. That way if we have a Map entry we do not
1365 -- insert a heap entry because we know it is already there.
1366 -- XXX The ZombieSession mechanism does not work as expected as we
1367 -- ignore ZombieSession when inserting a new entry. Anyway, we can
1368 -- remove this mechanism as at most only two heap entries may be
1369 -- created and they will be ultimately cleaned up.
1371 -- Heap processing needs the map and map processing needs the heap,
1372 -- therefore we cannot separate the two for modularity unless we have a
1373 -- way to achieve mutual recursion.
1375 let curTime = max sessionEventTime timestamp
1376 mOld = IsMap.mapLookup key sessionKeyValueMap
1378 -- deleting a key from the heap is expensive, so we never
1379 -- delete a key from heap, we just purge it from the Map and it
1380 -- gets purged from the heap on timeout. We just need an extra
1381 -- lookup in the Map when the key is purged from the heap, that
1382 -- should not be expensive.
1384 let (mp, cnt) = case mOld of
1385 Just (LiveSession _ _) ->
1387 key ZombieSession sessionKeyValueMap
1390 _ -> (sessionKeyValueMap, sessionCount)
1392 { sessionCurTime = curTime
1393 , sessionEventTime = curTime
1394 , sessionCount = cnt
1395 , sessionKeyValueMap = mp
1396 , sessionOutputStream = fromPure (key, fb)
1399 let expiry = addToAbsTime timestamp timeoutMs
1400 (hp1, mp1, out1, cnt1) <- do
1401 let vars = (sessionTimerHeap, sessionKeyValueMap,
1402 IsStream.nil, sessionCount)
1404 -- inserting new entry
1406 -- Eject a session from heap and map if needed
1407 eject <- ejectPred sessionCount
1408 (hp, mp, out, cnt) <-
1410 then ejectOne reset extract vars
1413 -- Insert the new session in heap
1414 let hp' = H.insert (Entry expiry key) hp
1415 in return (hp', mp, out, cnt + 1)
1416 -- updating old entry
1417 Just _ -> return vars
1419 let acc = LiveSession expiry fs1
1420 mp2 = IsMap.mapInsert key acc mp1
1421 return $ SessionState
1422 { sessionCurTime = curTime
1423 , sessionEventTime = curTime
1424 , sessionCount = cnt1
1425 , sessionTimerHeap = hp1
1426 , sessionKeyValueMap = mp2
1427 , sessionOutputStream = out1
1431 Just (LiveSession _ acc) -> return $ FL.Partial acc
1435 error $ "classifySessionsBy: "
1436 ++ "The supplied fold must consume at least one input"
1438 res <- step fs value
1440 FL.Done fb -> done fb
1441 FL.Partial fs1 -> partial fs1
1443 -- Got a timer tick event
1444 sstep sessionState@SessionState{..} Nothing =
1445 let curTime = addToAbsTime sessionCurTime tickMs
1446 in ejectExpired reset ejectPred extract sessionState curTime
1448-- | @classifySessionsBy tick keepalive predicate timeout fold stream@
1449-- classifies an input event @stream@ consisting of @(timestamp, (key,
1450-- value))@ into sessions based on the @key@, folding all the values
1451-- corresponding to the same key into a session using the supplied @fold@.
1453-- When the fold terminates or a @timeout@ occurs, a tuple consisting of the
1454-- session key and the folded value is emitted in the output stream. The
1455-- timeout is measured from the first event in the session. If the @keepalive@
1456-- option is set to 'True' the timeout is reset to 0 whenever an event is
1459-- The @timestamp@ in the input stream is an absolute time from some epoch,
1460-- characterizing the time when the input event was generated. The notion of
1461-- current time is maintained by a monotonic event time clock using the
1462-- timestamps seen in the input stream. The latest timestamp seen till now is
1463-- used as the base for the current time. When no new events are seen, a timer
1464-- is started with a clock resolution of @tick@ seconds. This timer is used to
1465-- detect session timeouts in the absence of new events.
1467-- To ensure an upper bound on the memory used the number of sessions can be
1468-- limited to an upper bound. If the ejection @predicate@ returns 'True', the
1469-- oldest session is ejected before inserting a new session.
1471-- When the stream ends any buffered sessions are ejected immediately.
1473-- If a session key is received even after a session has finished, another
1474-- session is created for that key.
1477-- Stream.mapM_ print
1478-- $ Stream.classifySessionsBy 1 False (const (return False)) 3 (Fold.take 3 Fold.toList)
1479-- $ Stream.timestamped
1480-- $ Stream.delay 0.1
1481-- $ (,) <$> Stream.fromList [1,2,3] <*> Stream.fromList ['a','b','c']
1488{-# INLINE classifySessionsBy #-}
1490 :: (IsStream t, MonadAsync m, Ord k)
1491 => Double -- ^ timer tick in seconds
1492 -> Bool -- ^ reset the timer when an event is received
1493 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
1494 -> Double -- ^ session timeout in seconds
1495 -> Fold m a b -- ^ Fold to be applied to session data
1496 -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
1497 -> t m (k, b) -- ^ session key, fold result
1498classifySessionsBy = classifySessionsByGeneric (Proxy :: Proxy (Map k))
1500-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
1501-- option set to 'True'.
1504-- classifyKeepAliveSessions = classifySessionsBy 1 True
1509{-# INLINE classifyKeepAliveSessions #-}
1510classifyKeepAliveSessions ::
1511 (IsStream t, MonadAsync m, Ord k)
1512 => (Int -> m Bool) -- ^ predicate to eject sessions on session count
1513 -> Double -- ^ session inactive timeout
1514 -> Fold m a b -- ^ Fold to be applied to session payload data
1515 -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
1517classifyKeepAliveSessions = classifySessionsBy 1 True
1519------------------------------------------------------------------------------
1520-- Keyed tumbling windows
1521------------------------------------------------------------------------------
1523-- Tumbling windows is a special case of sliding windows where the window slide
1524-- is the same as the window size. Or it can be a special case of session
1525-- windows where the reset flag is set to False.
1527-- XXX instead of using the early termination flag in the stream, we can use an
1528-- early terminating fold instead.
1531-- | Split the stream into fixed size chunks of specified size. Within each
1532-- such chunk fold the elements in buckets identified by the keys. A particular
1533-- bucket fold can be terminated early if a closing flag is encountered in an
1534-- element for that key.
1537{-# INLINABLE classifyChunksOf #-}
1539 :: (IsStream t, MonadAsync m, Ord k)
1540 => Int -- ^ window size
1541 -> Fold m a b -- ^ Fold to be applied to window events
1542 -> t m (k, a, Bool) -- ^ window key, data, close event
1544classifyChunksOf wsize = classifyChunksBy wsize False
1547-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
1548-- option set to 'False'.
1551-- classifySessionsOf = classifySessionsBy 1 False
1556{-# INLINE classifySessionsOf #-}
1557classifySessionsOf ::
1558 (IsStream t, MonadAsync m, Ord k)
1559 => (Int -> m Bool) -- ^ predicate to eject sessions on session count
1560 -> Double -- ^ time window size
1561 -> Fold m a b -- ^ Fold to be applied to session data
1562 -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
1564classifySessionsOf = classifySessionsBy 1 False
1566------------------------------------------------------------------------------
1568------------------------------------------------------------------------------
1570-- | @splitInnerBy splitter joiner stream@ splits the inner containers @f a@ of
1571-- an input stream @t m (f a)@ using the @splitter@ function. Container
1572-- elements @f a@ are collected until a split occurs, then all the elements
1573-- before the split are joined using the @joiner@ function.
1575-- For example, if we have a stream of @Array Word8@, we may want to split the
1576-- stream into arrays representing lines separated by '\n' byte such that the
1577-- resulting stream after a split would be one array for each line.
1579-- CAUTION! This is not a true streaming function as the container size after
1580-- the split and merge may not be bounded.
1583{-# INLINE splitInnerBy #-}
1585 :: (IsStream t, Monad m)
1586 => (f a -> m (f a, Maybe (f a))) -- splitter
1587 -> (f a -> f a -> m (f a)) -- joiner
1590splitInnerBy splitter joiner xs =
1591 fromStreamD $ D.splitInnerBy splitter joiner $ toStreamD xs
1593-- | Like 'splitInnerBy' but splits assuming the separator joins the segment in
1597{-# INLINE splitInnerBySuffix #-}
1599 :: (IsStream t, Monad m, Eq (f a), Monoid (f a))
1600 => (f a -> m (f a, Maybe (f a))) -- splitter
1601 -> (f a -> f a -> m (f a)) -- joiner
1604splitInnerBySuffix splitter joiner xs =
1605 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