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
175import Streamly.Internal.Data.Stream.IsStream.Type
176 (IsStream(..), fromStreamD, toStreamD, cons)
177import Streamly.Internal.Data.Time.Units
178 ( AbsTime, MilliSecond64(..), addToAbsTime, toRelTime
181import qualified Data.Heap as H
182import qualified Streamly.Internal.Data.Array.Foreign.Type as A
183import qualified Streamly.Internal.Data.Fold as FL
184import qualified Streamly.Internal.Data.IsMap as IsMap
185import qualified Streamly.Internal.Data.Parser.ParserD as PRD
186import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
187import qualified Streamly.Internal.Data.Stream.StreamD as D
188import qualified Streamly.Internal.Data.Stream.IsStream.Expand as Expand
189import qualified Streamly.Internal.Data.Stream.IsStream.Transform as Transform
190import qualified Streamly.Internal.Data.Unfold as Unfold
192import Prelude hiding (concatMap, map)
196-- >>> import Prelude hiding (zipWith, concatMap, concat)
197-- >>> import qualified Streamly.Prelude as Stream
198-- >>> import Streamly.Internal.Data.Stream.IsStream as Stream
199-- >>> import qualified Streamly.Data.Fold as Fold
200-- >>> import qualified Streamly.Internal.Data.Fold as Fold
201-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
202-- >>> import qualified Streamly.Internal.Data.Parser as Parser
203-- >>> import qualified Streamly.Data.Array.Foreign as Array
205------------------------------------------------------------------------------
207------------------------------------------------------------------------------
209-- | Drop prefix from the input stream if present.
214{-# INLINE dropPrefix #-}
216 -- (Eq a, IsStream t, Monad m) =>
217 t m a -> t m a -> t m a
218dropPrefix = error "Not implemented yet!"
220-- | Drop all matching infix from the input stream if present. Infix stream
221-- may be consumed multiple times.
223-- Space: @O(n)@ where n is the length of the infix.
226{-# INLINE dropInfix #-}
228 -- (Eq a, IsStream t, Monad m) =>
229 t m a -> t m a -> t m a
230dropInfix = error "Not implemented yet!"
232-- | Drop suffix from the input stream if present. Suffix stream may be
233-- consumed multiple times.
235-- Space: @O(n)@ where n is the length of the suffix.
238{-# INLINE dropSuffix #-}
240 -- (Eq a, IsStream t, Monad m) =>
241 t m a -> t m a -> t m a
242dropSuffix = error "Not implemented yet!"
244------------------------------------------------------------------------------
246------------------------------------------------------------------------------
248-- Splitting operations that take a predicate and a Fold can be
249-- expressed using parseMany. Operations like chunksOf, intervalsOf, split*,
250-- can be expressed using parseMany when used with an appropriate Parser.
252-- XXX We need takeGE/takeBetween to implement "some" using "many".
254-- | Like 'foldMany' but appends empty fold output if the fold and stream
255-- termination aligns:
257-- >>> f = Fold.take 2 Fold.sum
258-- >>> Stream.toList $ Stream.foldManyPost f $ Stream.fromList []
260-- >>> Stream.toList $ Stream.foldManyPost f $ Stream.fromList [1..9]
262-- >>> Stream.toList $ Stream.foldManyPost f $ Stream.fromList [1..10]
267{-# INLINE foldManyPost #-}
269 :: (IsStream t, Monad m)
273foldManyPost f m = fromStreamD $ D.foldManyPost f (toStreamD m)
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]
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, MonadThrow m)
390 fromStreamD $ D.parseMany (PRD.fromParserK p) (toStreamD m)
392-- | Same as parseMany but for StreamD streams.
396{-# INLINE parseManyD #-}
398 :: (IsStream t, MonadThrow m)
403 fromStreamD $ D.parseMany 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 $ Stream.map getSum $ Stream.parseIterate (\b -> Parser.takeBetween 0 2 (Fold.sconcat b)) 0 $ Stream.map 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, MonadThrow m)
453 => (b -> Parser m a b)
457parseIterate f i m = fromStreamD $
458 D.parseIterate (PRD.fromParserK . 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, Storable 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, Enum a, Eq a)
847 => Array a -> Fold m a b -> t m a -> t m b
849 intersperseM (fold f (A.toStream 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
893{-# INLINE splitOnSuffixSeq #-}
895 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
896 => Array a -> Fold m a b -> t m a -> t m b
897splitOnSuffixSeq patt f m =
898 fromStreamD $ D.splitOnSuffixSeq False patt f (toStreamD m)
900-- | Like 'splitOn' but drops any empty splits.
903{-# INLINE wordsOn #-}
904wordsOn :: -- (IsStream t, Monad m, Storable a, Eq a) =>
905 Array a -> Fold m a b -> t m a -> t m b
906wordsOn _subseq _f _m =
907 undefined -- D.fromStreamD $ D.wordsOn f subseq (D.toStreamD m)
909-- | Like 'splitOnSuffixSeq' but keeps the suffix intact in the splits.
911-- >>> splitWithSuffixSeq' pat xs = Stream.toList $ Stream.splitWithSuffixSeq (Array.fromList pat) Fold.toList (Stream.fromList xs)
913-- >>> splitWithSuffixSeq' "." ""
916-- >>> splitWithSuffixSeq' "." "."
919-- >>> splitWithSuffixSeq' "." "a"
922-- >>> splitWithSuffixSeq' "." ".a"
925-- >>> splitWithSuffixSeq' "." "a."
928-- >>> splitWithSuffixSeq' "." "a.b"
931-- >>> splitWithSuffixSeq' "." "a.b."
934-- >>> splitWithSuffixSeq' "." "a..b.."
935-- ["a.",".","b.","."]
938{-# INLINE splitWithSuffixSeq #-}
940 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
941 => Array a -> Fold m a b -> t m a -> t m b
942splitWithSuffixSeq patt f m =
943 fromStreamD $ D.splitOnSuffixSeq True patt f (toStreamD m)
945-- This can be implemented easily using Rabin Karp
946-- | Split post any one of the given patterns.
949{-# INLINE splitOnSuffixSeqAny #-}
950splitOnSuffixSeqAny :: -- (IsStream t, Monad m, Storable a, Integral a) =>
951 [Array a] -> Fold m a b -> t m a -> t m b
952splitOnSuffixSeqAny _subseq _f _m = undefined
953 -- D.fromStreamD $ D.splitPostAny f subseq (D.toStreamD m)
955------------------------------------------------------------------------------
957------------------------------------------------------------------------------
959-- | Group the input stream into groups of @n@ elements each and then fold each
960-- group using the provided fold function.
962-- >>> Stream.toList $ Stream.chunksOf 2 Fold.sum (Stream.enumerateFromTo 1 10)
965-- This can be considered as an n-fold version of 'take' where we apply
966-- 'take' repeatedly on the leftover stream until the stream exhausts.
968-- @chunksOf n f = foldMany (FL.take n f)@
971{-# INLINE chunksOf #-}
973 :: (IsStream t, Monad m)
974 => Int -> Fold m a b -> t m a -> t m b
975chunksOf n f = fromStreamD . D.chunksOf n f . toStreamD
977-- | @arraysOf n stream@ groups the elements in the input stream into arrays of
980-- Same as the following but may be more efficient:
982-- > arraysOf n = Stream.foldMany (A.writeN n)
985{-# INLINE arraysOf #-}
986arraysOf :: (IsStream t, MonadIO m, Storable a)
987 => Int -> t m a -> t m (Array a)
988arraysOf n = fromStreamD . A.arraysOf n . toStreamD
990-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
991-- XXX add this example after fixing the serial stream rate control
993-- | Group the input stream into windows of @n@ second each and then fold each
994-- group using the provided fold function.
996-- >>> Stream.toList $ Stream.take 5 $ Stream.intervalsOf 1 Fold.sum $ Stream.constRate 2 $ Stream.enumerateFrom 1
997-- [...,...,...,...,...]
1000{-# INLINE intervalsOf #-}
1002 :: (IsStream t, MonadAsync m)
1003 => Double -> Fold m a b -> t m a -> t m b
1005 splitWithSuffix isNothing (FL.catMaybes f)
1006 (interjectSuffix n (return Nothing) (map Just xs))
1008-- XXX This can be implemented more efficiently by sharing a Clock.Timer across
1009-- parallel threads and resetting it whenever a span is emitted.
1011-- | Like 'chunksOf' but if the chunk is not completed within the specified
1012-- time interval then emit whatever we have collected till now. The chunk
1013-- timeout is reset whenever a chunk is emitted.
1015-- >>> s = Stream.delayPost 0.3 $ Stream.fromList [1..1000]
1016-- >>> f = Stream.mapM_ print $ Stream.chunksOfTimeout 5 1 Fold.toList s
1019{-# INLINE chunksOfTimeout #-}
1020chunksOfTimeout :: (IsStream t, MonadAsync m, Functor (t m))
1021 => Int -> Double -> FL.Fold m a b -> t m a -> t m b
1022chunksOfTimeout n timeout f =
1024 . classifySessionsBy
1025 timeout False (const (return False)) timeout (FL.take n f)
1026 . Transform.timestamped
1029------------------------------------------------------------------------------
1030-- Windowed classification
1031------------------------------------------------------------------------------
1033-- TODO: To mark the position in space or time we can have Indexed or
1034-- TimeStamped types. That can make it easy to deal with the position indices
1037------------------------------------------------------------------------------
1038-- Keyed Sliding Windows
1039------------------------------------------------------------------------------
1042{-# INLINABLE classifySlidingChunks #-}
1043classifySlidingChunks
1044 :: (IsStream t, MonadAsync m, Ord k)
1045 => Int -- ^ window size
1046 -> Int -- ^ window slide
1047 -> Fold m a b -- ^ Fold to be applied to window events
1048 -> t m (k, a, Bool) -- ^ window key, data, close event
1050classifySlidingChunks wsize wslide (Fold step initial extract) str
1053-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
1054-- slide the send window when an ack is received and we slide the receive
1055-- window when a sequence is complete. Sliding is stateful in case of TCP,
1056-- sliding releases the send buffer or makes data available to the user from
1057-- the receive buffer.
1058{-# INLINABLE classifySlidingSessions #-}
1059classifySlidingSessions
1060 :: (IsStream t, MonadAsync m, Ord k)
1061 => Double -- ^ timer tick in seconds
1062 -> Double -- ^ time window size
1063 -> Double -- ^ window slide
1064 -> Fold m a b -- ^ Fold to be applied to window events
1065 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
1067classifySlidingSessions tick interval slide (Fold step initial extract) str
1071------------------------------------------------------------------------------
1072-- Sliding Window Buffers
1073------------------------------------------------------------------------------
1075-- These buffered versions could be faster than concurrent incremental folds of
1076-- all overlapping windows as in many cases we may not need all the values to
1077-- compute the fold, we can just compute the result using the old value and new
1078-- value. However, we may need the buffer once in a while, for example for
1079-- string search we usually compute the hash incrementally but when the hash
1080-- matches the hash of the pattern we need to compare the whole string.
1082-- XXX we should be able to implement sequence based splitting combinators
1083-- using this combinator.
1086-- | Buffer n elements of the input in a ring buffer. When t new elements are
1087-- collected, slide the window to remove the same number of oldest elements,
1088-- insert the new elements, and apply an incremental fold on the sliding
1089-- window, supplying the outgoing elements, the new ring buffer as arguments.
1091 :: (IsStream t, Monad m, Ord a, Storable a)
1092 => Int -- window size
1093 -> Int -- window slide
1094 -> Fold m (Ring a, Array a) b
1097slidingChunkBuffer = undefined
1099-- Buffer n seconds worth of stream elements of the input in a radix tree.
1100-- Every t seconds, remove the items that are older than n seconds, and apply
1101-- an incremental fold on the sliding window, supplying the outgoing elements,
1102-- and the new radix tree buffer as arguments.
1104 :: (IsStream t, Monad m, Ord a, Storable a)
1105 => Int -- window size
1107 -> Fold m (RTree a, Array a) b
1110slidingSessionBuffer = undefined
1113------------------------------------------------------------------------------
1114-- Keyed Session Windows
1115------------------------------------------------------------------------------
1118-- | Keyed variable size space windows. Close the window if we do not receive a
1119-- window event in the next "spaceout" elements.
1120{-# INLINABLE classifyChunksBy #-}
1122 :: (IsStream t, MonadAsync m, Ord k)
1123 => Int -- ^ window spaceout (spread)
1124 -> Bool -- ^ reset the spaceout when a chunk window element is received
1125 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1126 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1128classifyChunksBy spanout reset (Fold step initial extract) str = undefined
1130-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
1131-- received within the chunk size window. The chunk gets closed only if no
1132-- element is received within the chunk window.
1134{-# INLINABLE classifyKeepAliveChunks #-}
1135classifyKeepAliveChunks
1136 :: (IsStream t, MonadAsync m, Ord k)
1137 => Int -- ^ window spaceout (spread)
1138 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1139 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1141classifyKeepAliveChunks spanout = classifyChunksBy spanout True
1144data SessionState t m f s b = SessionState
1145 { sessionCurTime :: !AbsTime -- ^ time since last event
1146 , sessionEventTime :: !AbsTime -- ^ time as per last event
1147 -- We can use the Map size instead of maintaining a count, but if we have
1148 -- to switch to HashMap then it can be useful.
1149 , sessionCount :: !Int -- ^ total number of sessions in progress
1150 , sessionTimerHeap :: H.Heap (H.Entry AbsTime (Key f)) -- ^ heap for timeouts
1151 , sessionKeyValueMap :: f s -- ^ Stored sessions for keys
1152 , sessionOutputStream :: t (m :: Type -> Type) (Key f, b) -- ^ Completed sessions
1155data SessionEntry s = LiveSession !AbsTime !s | ZombieSession
1157-- delete from map and output the fold accumulator
1158ejectEntry :: (Monad m, IsStream t, IsMap f) =>
1166 -> m (heap, f entry, t m (Key f, b), Int)
1167ejectEntry extract hp mp out cnt acc key = do
1169 let out1 = (key, sess) `cons` out
1170 let mp1 = IsMap.mapDelete key mp
1171 return (hp, mp1, out1, cnt - 1)
1173{-# NOINLINE flush #-}
1174flush :: (Monad m, IsMap f, IsStream t) =>
1176 -> SessionState t m f (SessionEntry s) b
1177 -> m (SessionState t m f (SessionEntry s) b)
1178flush extract session@SessionState{..} = do
1179 (hp', mp', out, count) <-
1182 , sessionKeyValueMap
1187 { sessionCount = count
1188 , sessionTimerHeap = hp'
1189 , sessionKeyValueMap = mp'
1190 , sessionOutputStream = out
1195 ejectAll (hp, mp, out, !cnt) = do
1196 let hres = H.uncons hp
1198 Just (Entry _ key, hp1) -> do
1199 r <- case IsMap.mapLookup key mp of
1200 Nothing -> return (hp1, mp, out, cnt)
1201 Just ZombieSession ->
1202 return (hp1, IsMap.mapDelete key mp, out, cnt)
1203 Just (LiveSession _ acc) ->
1204 ejectEntry extract hp1 mp out cnt acc key
1207 assert (IsMap.mapNull mp) (return ())
1208 return (hp, mp, out, cnt)
1210{-# NOINLINE ejectOne #-}
1211ejectOne :: (IsMap f, IsStream t, Monad m) =>
1214 -> ( H.Heap (Entry AbsTime (Key f))
1215 , f (SessionEntry acc)
1219 -> m ( H.Heap (Entry AbsTime (Key f))
1220 , f (SessionEntry acc)
1221 , t m (Key f, b), Int
1223ejectOne reset extract = go
1227 go (hp, mp, out, !cnt) = do
1228 let hres = H.uncons hp
1230 Just (Entry expiry key, hp1) ->
1231 case IsMap.mapLookup key mp of
1232 Nothing -> go (hp1, mp, out, cnt)
1233 Just ZombieSession ->
1234 go (hp1, IsMap.mapDelete key mp, out, cnt)
1235 Just (LiveSession expiry1 acc) -> do
1236 if not reset || expiry1 <= expiry
1237 then ejectEntry extract hp1 mp out cnt acc key
1239 -- reset the session timeout and continue
1240 let hp2 = H.insert (Entry expiry1 key) hp1
1241 in go (hp2, mp, out, cnt)
1243 assert (IsMap.mapNull mp) (return ())
1244 return (hp, mp, out, cnt)
1246{-# NOINLINE ejectExpired #-}
1247ejectExpired :: (IsMap f, IsStream t, Monad m) =>
1251 -> SessionState t m f (SessionEntry acc) b
1253 -> m (SessionState t m f (SessionEntry acc) b)
1254ejectExpired reset ejectPred extract session@SessionState{..} curTime = do
1255 (hp', mp', out, count) <-
1257 sessionTimerHeap sessionKeyValueMap IsStream.nil sessionCount
1259 { sessionCurTime = curTime
1260 , sessionCount = count
1261 , sessionTimerHeap = hp'
1262 , sessionKeyValueMap = mp'
1263 , sessionOutputStream = out
1268 ejectLoop hp mp out !cnt = do
1269 let hres = H.uncons hp
1271 Just (Entry expiry key, hp1) -> do
1273 if curTime >= expiry
1274 then return (True, False)
1280 case IsMap.mapLookup key mp of
1281 Nothing -> ejectLoop hp1 mp out cnt
1282 Just ZombieSession ->
1283 ejectLoop hp1 (IsMap.mapDelete key mp) out cnt
1284 Just (LiveSession expiry1 acc) -> do
1285 if expiry1 <= curTime || not reset || force
1287 (hp2,mp1,out1,cnt1) <-
1288 ejectEntry extract hp1 mp out cnt acc key
1289 ejectLoop hp2 mp1 out1 cnt1
1291 -- reset the session timeout and continue
1292 let hp2 = H.insert (Entry expiry1 key) hp1
1293 in ejectLoop hp2 mp out cnt
1294 else return (hp, mp, out, cnt)
1296 assert (IsMap.mapNull mp) (return ())
1297 return (hp, mp, out, cnt)
1299-- XXX Use mutable IORef in accumulator
1300{-# INLINE classifySessionsByGeneric #-}
1301classifySessionsByGeneric
1302 :: forall t m f a b. (IsStream t, MonadAsync m, IsMap f)
1303 => Proxy (f :: (Type -> Type))
1304 -> Double -- ^ timer tick in seconds
1305 -> Bool -- ^ reset the timer when an event is received
1306 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
1307 -> Double -- ^ session timeout in seconds
1308 -> Fold m a b -- ^ Fold to be applied to session data
1309 -> t m (AbsTime, (Key f, a)) -- ^ timestamp, (session key, session data)
1310 -> t m (Key f, b) -- ^ session key, fold result
1311classifySessionsByGeneric _ tick reset ejectPred tmout
1312 (Fold step initial extract) input =
1313 Expand.unfoldMany (Unfold.lmap sessionOutputStream Unfold.fromStream)
1314 $ scanlMAfter' sstep (return szero) (flush extract)
1315 $ interjectSuffix tick (return Nothing)
1320 timeoutMs = toRelTime (round (tmout * 1000) :: MilliSecond64)
1321 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
1322 szero = SessionState
1323 { sessionCurTime = toAbsTime (0 :: MilliSecond64)
1324 , sessionEventTime = toAbsTime (0 :: MilliSecond64)
1326 , sessionTimerHeap = H.empty
1327 , sessionKeyValueMap = IsMap.mapEmpty :: f s
1328 , sessionOutputStream = IsStream.nil
1331 -- We can eject sessions based on the current session count to limit
1332 -- memory consumption. There are two possible strategies:
1334 -- 1) Eject old sessions or sessions beyond a certain/lower timeout
1335 -- threshold even before timeout, effectively reduce the timeout.
1336 -- 2) Drop creation of new sessions but keep accepting new events for the
1339 -- We use the first strategy as of now.
1341 -- Got a new stream input element
1342 sstep session@SessionState{..} (Just (timestamp, (key, value))) = do
1343 -- XXX instead of a heap we could use a timer wheel.
1345 -- XXX if the key is an Int, we can also use an IntMap for slightly
1346 -- better performance.
1350 -- Values for each key are collected in a map using the supplied fold.
1351 -- When we insert a key in the Map we insert an entry into the heap as
1352 -- well with the session expiry as the sort key. The Map entry
1353 -- consists of the fold result, and the expiry time of the session. If
1354 -- "reset" is True the expiry time is readjusted whenever a new event
1355 -- is processed. If the fold terminates and a new session is started
1356 -- for the same key the expiry time is set to the first timestamp of
1359 -- The heap must have at most one entry for any given key. The heap is
1360 -- processed periodically to remove the expired entries. We pick up an
1361 -- expired entry from the top of the heap and if the session has
1362 -- expired based on the expiry time in the Map entry then we remove the
1363 -- session from the Map and yield its fold output. Otherwise, we
1364 -- reinsert the entry into the heap based on the current expiry in the
1367 -- If an entry is removed from the Map and not removed from the heap
1368 -- and in the meantime it is inserted again in the Map (using the same
1369 -- key) then how do we avoid inserting multiple entries in the heap?
1370 -- For this reason we maintain the invariant that the Map entry is
1371 -- removed only when the heap entry is removed. Even if the fold has
1372 -- finished we still keep a dummy Map entry (ZombieSession) until the
1373 -- heap entry is removed. That way if we have a Map entry we do not
1374 -- insert a heap entry because we know it is already there.
1375 -- XXX The ZombieSession mechanism does not work as expected as we
1376 -- ignore ZombieSession when inserting a new entry. Anyway, we can
1377 -- remove this mechanism as at most only two heap entries may be
1378 -- created and they will be ultimately cleaned up.
1380 -- Heap processing needs the map and map processing needs the heap,
1381 -- therefore we cannot separate the two for modularity unless we have a
1382 -- way to achieve mutual recursion.
1384 let curTime = max sessionEventTime timestamp
1385 mOld = IsMap.mapLookup key sessionKeyValueMap
1387 -- deleting a key from the heap is expensive, so we never
1388 -- delete a key from heap, we just purge it from the Map and it
1389 -- gets purged from the heap on timeout. We just need an extra
1390 -- lookup in the Map when the key is purged from the heap, that
1391 -- should not be expensive.
1393 let (mp, cnt) = case mOld of
1394 Just (LiveSession _ _) ->
1396 key ZombieSession sessionKeyValueMap
1399 _ -> (sessionKeyValueMap, sessionCount)
1401 { sessionCurTime = curTime
1402 , sessionEventTime = curTime
1403 , sessionCount = cnt
1404 , sessionKeyValueMap = mp
1405 , sessionOutputStream = fromPure (key, fb)
1408 let expiry = addToAbsTime timestamp timeoutMs
1409 (hp1, mp1, out1, cnt1) <- do
1410 let vars = (sessionTimerHeap, sessionKeyValueMap,
1411 IsStream.nil, sessionCount)
1413 -- inserting new entry
1415 -- Eject a session from heap and map if needed
1416 eject <- ejectPred sessionCount
1417 (hp, mp, out, cnt) <-
1419 then ejectOne reset extract vars
1422 -- Insert the new session in heap
1423 let hp' = H.insert (Entry expiry key) hp
1424 in return (hp', mp, out, cnt + 1)
1425 -- updating old entry
1426 Just _ -> return vars
1428 let acc = LiveSession expiry fs1
1429 mp2 = IsMap.mapInsert key acc mp1
1430 return $ SessionState
1431 { sessionCurTime = curTime
1432 , sessionEventTime = curTime
1433 , sessionCount = cnt1
1434 , sessionTimerHeap = hp1
1435 , sessionKeyValueMap = mp2
1436 , sessionOutputStream = out1
1440 Just (LiveSession _ acc) -> return $ FL.Partial acc
1444 error $ "classifySessionsBy: "
1445 ++ "The supplied fold must consume at least one input"
1447 res <- step fs value
1449 FL.Done fb -> done fb
1450 FL.Partial fs1 -> partial fs1
1452 -- Got a timer tick event
1453 sstep sessionState@SessionState{..} Nothing =
1454 let curTime = addToAbsTime sessionCurTime tickMs
1455 in ejectExpired reset ejectPred extract sessionState curTime
1457-- | @classifySessionsBy tick keepalive predicate timeout fold stream@
1458-- classifies an input event @stream@ consisting of @(timestamp, (key,
1459-- value))@ into sessions based on the @key@, folding all the values
1460-- corresponding to the same key into a session using the supplied @fold@.
1462-- When the fold terminates or a @timeout@ occurs, a tuple consisting of the
1463-- session key and the folded value is emitted in the output stream. The
1464-- timeout is measured from the first event in the session. If the @keepalive@
1465-- option is set to 'True' the timeout is reset to 0 whenever an event is
1468-- The @timestamp@ in the input stream is an absolute time from some epoch,
1469-- characterizing the time when the input event was generated. The notion of
1470-- current time is maintained by a monotonic event time clock using the
1471-- timestamps seen in the input stream. The latest timestamp seen till now is
1472-- used as the base for the current time. When no new events are seen, a timer
1473-- is started with a clock resolution of @tick@ seconds. This timer is used to
1474-- detect session timeouts in the absence of new events.
1476-- To ensure an upper bound on the memory used the number of sessions can be
1477-- limited to an upper bound. If the ejection @predicate@ returns 'True', the
1478-- oldest session is ejected before inserting a new session.
1480-- When the stream ends any buffered sessions are ejected immediately.
1482-- If a session key is received even after a session has finished, another
1483-- session is created for that key.
1486-- Stream.mapM_ print
1487-- $ Stream.classifySessionsBy 1 False (const (return False)) 3 (Fold.take 3 Fold.toList)
1488-- $ Stream.timestamped
1489-- $ Stream.delay 0.1
1490-- $ (,) <$> Stream.fromList [1,2,3] <*> Stream.fromList ['a','b','c']
1497{-# INLINE classifySessionsBy #-}
1499 :: (IsStream t, MonadAsync m, Ord k)
1500 => Double -- ^ timer tick in seconds
1501 -> Bool -- ^ reset the timer when an event is received
1502 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
1503 -> Double -- ^ session timeout in seconds
1504 -> Fold m a b -- ^ Fold to be applied to session data
1505 -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
1506 -> t m (k, b) -- ^ session key, fold result
1507classifySessionsBy = classifySessionsByGeneric (Proxy :: Proxy (Map k))
1509-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
1510-- option set to 'True'.
1513-- classifyKeepAliveSessions = classifySessionsBy 1 True
1518{-# INLINE classifyKeepAliveSessions #-}
1519classifyKeepAliveSessions ::
1520 (IsStream t, MonadAsync m, Ord k)
1521 => (Int -> m Bool) -- ^ predicate to eject sessions on session count
1522 -> Double -- ^ session inactive timeout
1523 -> Fold m a b -- ^ Fold to be applied to session payload data
1524 -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
1526classifyKeepAliveSessions = classifySessionsBy 1 True
1528------------------------------------------------------------------------------
1529-- Keyed tumbling windows
1530------------------------------------------------------------------------------
1532-- Tumbling windows is a special case of sliding windows where the window slide
1533-- is the same as the window size. Or it can be a special case of session
1534-- windows where the reset flag is set to False.
1536-- XXX instead of using the early termination flag in the stream, we can use an
1537-- early terminating fold instead.
1540-- | Split the stream into fixed size chunks of specified size. Within each
1541-- such chunk fold the elements in buckets identified by the keys. A particular
1542-- bucket fold can be terminated early if a closing flag is encountered in an
1543-- element for that key.
1546{-# INLINABLE classifyChunksOf #-}
1548 :: (IsStream t, MonadAsync m, Ord k)
1549 => Int -- ^ window size
1550 -> Fold m a b -- ^ Fold to be applied to window events
1551 -> t m (k, a, Bool) -- ^ window key, data, close event
1553classifyChunksOf wsize = classifyChunksBy wsize False
1556-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
1557-- option set to 'False'.
1560-- classifySessionsOf = classifySessionsBy 1 False
1565{-# INLINE classifySessionsOf #-}
1566classifySessionsOf ::
1567 (IsStream t, MonadAsync m, Ord k)
1568 => (Int -> m Bool) -- ^ predicate to eject sessions on session count
1569 -> Double -- ^ time window size
1570 -> Fold m a b -- ^ Fold to be applied to session data
1571 -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
1573classifySessionsOf = classifySessionsBy 1 False
1575------------------------------------------------------------------------------
1577------------------------------------------------------------------------------
1579-- | @splitInnerBy splitter joiner stream@ splits the inner containers @f a@ of
1580-- an input stream @t m (f a)@ using the @splitter@ function. Container
1581-- elements @f a@ are collected until a split occurs, then all the elements
1582-- before the split are joined using the @joiner@ function.
1584-- For example, if we have a stream of @Array Word8@, we may want to split the
1585-- stream into arrays representing lines separated by '\n' byte such that the
1586-- resulting stream after a split would be one array for each line.
1588-- CAUTION! This is not a true streaming function as the container size after
1589-- the split and merge may not be bounded.
1592{-# INLINE splitInnerBy #-}
1594 :: (IsStream t, Monad m)
1595 => (f a -> m (f a, Maybe (f a))) -- splitter
1596 -> (f a -> f a -> m (f a)) -- joiner
1599splitInnerBy splitter joiner xs =
1600 fromStreamD $ D.splitInnerBy splitter joiner $ toStreamD xs
1602-- | Like 'splitInnerBy' but splits assuming the separator joins the segment in
1606{-# INLINE splitInnerBySuffix #-}
1608 :: (IsStream t, Monad m, Eq (f a), Monoid (f a))
1609 => (f a -> m (f a, Maybe (f a))) -- splitter
1610 -> (f a -> f a -> m (f a)) -- joiner
1613splitInnerBySuffix splitter joiner xs =
1614 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