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
58 -- -- *** Splitting By Sequences
64 -- Keeping the delimiters
69 -- Splitting using multiple sequence separators
71 -- , splitOnAnySuffixSeq
72 -- , splitOnAnyPrefixSeq
74 -- -- *** Splitting By Streams
75 -- -- | Splitting a stream using another stream as separator.
77 -- ** Keyed Window Classification
79 -- | Split the stream into chunks or windows by position or time. Each
80 -- window can be associated with a key, all events associated with a
81 -- particular key in the window can be folded to a single result. The
82 -- window termination can be dynamically controlled by the fold.
84 -- The term "chunk" is used for a window defined by position of elements
85 -- and the term "session" is used for a time window.
87 -- *** Tumbling Windows
88 -- | A new window starts after the previous window is finished.
94 -- *** Keep Alive Windows
95 -- | The window size is extended if an event arrives within the specified
96 -- window size. This can represent sessions with idle or inactive timeout.
98 -- , classifyKeepAliveChunks
99 , classifyKeepAliveSessions
102 -- *** Sliding Windows
103 -- | A new window starts after the specified slide from the previous
104 -- window. Therefore windows can overlap.
105 , classifySlidingChunks
106 , classifySlidingSessions
107 -- *** Sliding Window Buffers
108 -- , slidingChunkBuffer
109 -- , slidingSessionBuffer
112 -- * Reduce By Parsers
113 -- ** Generic Parsing
114 -- | Apply parsers on a stream.
122 -- In imperative terms, grouped folding can be considered as a nested loop
123 -- where we loop over the stream to group elements and then loop over
124 -- individual groups to fold them to a single value that is yielded in the
127 , wordsBy -- stripAndCompactBy
132 -- -- *** Searching Sequences
133 -- , seqIndices -- search a sequence in the stream
135 -- -- *** Searching Multiple Sequences
136 -- , seqIndicesAny -- search any of the given sequence in the stream
138 -- -- -- ** Searching Streams
139 -- -- | Finding a stream within another stream.
141 -- * Nested splitting
149import Control.Concurrent (threadDelay)
150import Control.Exception (assert)
151import Control.Monad.Catch (MonadThrow)
152import Control.Monad.IO.Class (MonadIO(..))
153import Data.Heap (Entry(..))
154import Data.Kind (Type)
155import Data.Maybe (isNothing)
156import Foreign.Storable (Storable)
157import Streamly.Internal.Control.Concurrent (MonadAsync)
158import Streamly.Internal.Data.Fold.Type (Fold (..))
159import Streamly.Internal.Data.Refold.Type (Refold (..))
160import Streamly.Internal.Data.Parser (Parser (..))
161import Streamly.Internal.Data.Array.Foreign.Type (Array)
162import Streamly.Internal.Data.Stream.IsStream.Common
173import Streamly.Internal.Data.Stream.IsStream.Type
174 (IsStream(..), fromStreamD, toStreamD, cons)
175import Streamly.Internal.Data.Time.Units
176 ( AbsTime, MilliSecond64(..), addToAbsTime, toRelTime
179import qualified Data.Heap as H
180import qualified Data.Map.Strict as Map
181import qualified Streamly.Internal.Data.Array.Foreign.Type as A
182import qualified Streamly.Internal.Data.Fold as FL
183import qualified Streamly.Internal.Data.Parser.ParserK.Type as PRK
184import qualified Streamly.Internal.Data.Parser.ParserD as PRD
185import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
186import qualified Streamly.Internal.Data.Stream.StreamD as D
187import qualified Streamly.Internal.Data.Stream.IsStream.Transform as Transform
189import Prelude hiding (concatMap, map)
193-- >>> import Prelude hiding (zipWith, concatMap, concat)
194-- >>> import qualified Streamly.Prelude as Stream
195-- >>> import Streamly.Internal.Data.Stream.IsStream as Stream
196-- >>> import qualified Streamly.Data.Fold as Fold
197-- >>> import qualified Streamly.Internal.Data.Fold as Fold
198-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
199-- >>> import qualified Streamly.Internal.Data.Parser as Parser
200-- >>> import qualified Streamly.Data.Array.Foreign as Array
202------------------------------------------------------------------------------
204------------------------------------------------------------------------------
206-- | Drop prefix from the input stream if present.
210-- /Unimplemented/ - Help wanted.
211{-# INLINE dropPrefix #-}
213 -- (Eq a, IsStream t, Monad m) =>
214 t m a -> t m a -> t m a
215dropPrefix = error "Not implemented yet!"
217-- | Drop all matching infix from the input stream if present. Infix stream
218-- may be consumed multiple times.
220-- Space: @O(n)@ where n is the length of the infix.
222-- /Unimplemented/ - Help wanted.
223{-# INLINE dropInfix #-}
225 -- (Eq a, IsStream t, Monad m) =>
226 t m a -> t m a -> t m a
227dropInfix = error "Not implemented yet!"
229-- | Drop suffix from the input stream if present. Suffix stream may be
230-- consumed multiple times.
232-- Space: @O(n)@ where n is the length of the suffix.
234-- /Unimplemented/ - Help wanted.
235{-# INLINE dropSuffix #-}
237 -- (Eq a, IsStream t, Monad m) =>
238 t m a -> t m a -> t m a
239dropSuffix = error "Not implemented yet!"
241------------------------------------------------------------------------------
243------------------------------------------------------------------------------
245-- Splitting operations that take a predicate and a Fold can be
246-- expressed using parseMany. Operations like chunksOf, intervalsOf, split*,
247-- can be expressed using parseMany when used with an appropriate Parser.
249-- XXX We need takeGE/takeBetween to implement "some" using "many".
251-- | Like 'foldMany' but appends empty fold output if the fold and stream
252-- termination aligns:
254-- >>> f = Fold.take 2 Fold.sum
255-- >>> Stream.toList $ Stream.foldManyPost f $ Stream.fromList []
257-- >>> Stream.toList $ Stream.foldManyPost f $ Stream.fromList [1..9]
259-- >>> Stream.toList $ Stream.foldManyPost f $ Stream.fromList [1..10]
264{-# INLINE foldManyPost #-}
266 :: (IsStream t, Monad m)
270foldManyPost f m = fromStreamD $ D.foldManyPost f (toStreamD m)
272-- | Apply a 'Fold' repeatedly on a stream and emit the fold outputs in the
275-- To sum every two contiguous elements in a stream:
277-- >>> f = Fold.take 2 Fold.sum
278-- >>> Stream.toList $ Stream.foldMany f $ Stream.fromList [1..10]
281-- On an empty stream the output is empty:
283-- >>> Stream.toList $ Stream.foldMany f $ Stream.fromList []
286-- Note @Stream.foldMany (Fold.take 0)@ would result in an infinite loop in a
291{-# INLINE foldMany #-}
293 :: (IsStream t, Monad m)
297foldMany f m = fromStreamD $ D.foldMany f (toStreamD m)
299-- | Like 'foldMany' but using the 'Refold' type instead of 'Fold'.
302{-# INLINE refoldMany #-}
303refoldMany :: (IsStream t, Monad m) =>
304 Refold m c a b -> m c -> t m a -> t m b
305refoldMany f action = fromStreamD . D.refoldMany f action . toStreamD
307-- | Apply a stream of folds to an input stream and emit the results in the
312{-# INLINE foldSequence #-}
314 :: -- (IsStream t, Monad m) =>
318foldSequence _f _m = undefined
320-- | Iterate a fold generator on a stream. The initial value @b@ is used to
321-- generate the first fold, the fold is applied on the stream and the result of
322-- the fold is used to generate the next fold and so on.
325-- >>> import Data.Monoid (Sum(..))
326-- >>> f x = return (Fold.take 2 (Fold.sconcat x))
327-- >>> s = Stream.map Sum $ Stream.fromList [1..10]
328-- >>> Stream.toList $ Stream.map getSum $ Stream.foldIterateM f (pure 0) s
333-- This is the streaming equivalent of monad like sequenced application of
334-- folds where next fold is dependent on the previous fold.
338{-# INLINE foldIterateM #-}
340 (IsStream t, Monad m) => (b -> m (Fold m a b)) -> m b -> t m a -> t m b
341foldIterateM f i m = fromStreamD $ D.foldIterateM f i (toStreamD m)
343-- | Like 'foldIterateM' but using the 'Refold' type instead. This could be
344-- much more efficient due to stream fusion.
347{-# INLINE refoldIterateM #-}
348refoldIterateM :: (IsStream t, Monad m) =>
349 Refold m b a b -> m b -> t m a -> t m b
350refoldIterateM c i m = fromStreamD $ D.refoldIterateM c i (toStreamD m)
352------------------------------------------------------------------------------
354------------------------------------------------------------------------------
356-- | Apply a 'Parser' repeatedly on a stream and emit the parsed values in the
359-- This is the streaming equivalent of the 'Streamly.Internal.Data.Parser.many'
362-- >>> Stream.toList $ Stream.parseMany (Parser.takeBetween 0 2 Fold.sum) $ Stream.fromList [1..10]
366-- > Stream.toList $ Stream.parseMany (Parser.line Fold.toList) $ Stream.fromList "hello\\nworld"
367-- ["hello\\n","world"]
372-- foldMany f = parseMany (fromFold f)
375-- Known Issues: When the parser fails there is no way to get the remaining
380{-# INLINE parseMany #-}
382 :: (IsStream t, MonadThrow m)
387 fromStreamD $ D.parseMany (PRK.fromParserK p) (toStreamD m)
389{-# INLINE parseManyD #-}
391 :: (IsStream t, MonadThrow m)
396 fromStreamD $ D.parseMany p (toStreamD m)
398-- | Apply a stream of parsers to an input stream and emit the results in the
403{-# INLINE parseSequence #-}
405 :: -- (IsStream t, Monad m) =>
409parseSequence _f _m = undefined
411-- XXX Change the parser arguments' order
413-- | @parseManyTill collect test stream@ tries the parser @test@ on the input,
414-- if @test@ fails it backtracks and tries @collect@, after @collect@ succeeds
415-- @test@ is tried again and so on. The parser stops when @test@ succeeds. The
416-- output of @test@ is discarded and the output of @collect@ is emitted in the
417-- output stream. The parser fails if @collect@ fails.
421{-# INLINE parseManyTill #-}
423 -- (IsStream t, MonadThrow m) =>
428parseManyTill = undefined
430-- | Iterate a parser generating function on a stream. The initial value @b@ is
431-- used to generate the first parser, the parser is applied on the stream and
432-- the result is used to generate the next parser and so on.
434-- >>> import Data.Monoid (Sum(..))
435-- >>> Stream.toList $ Stream.map getSum $ Stream.parseIterate (\b -> Parser.takeBetween 0 2 (Fold.sconcat b)) 0 $ Stream.map Sum $ Stream.fromList [1..10]
438-- This is the streaming equivalent of monad like sequenced application of
439-- parsers where next parser is dependent on the previous parser.
443{-# INLINE parseIterate #-}
445 :: (IsStream t, MonadThrow m)
446 => (b -> Parser m a b)
450parseIterate f i m = fromStreamD $
451 D.parseIterate (PRK.fromParserK . f) i (toStreamD m)
453------------------------------------------------------------------------------
454-- Generalized grouping
455------------------------------------------------------------------------------
457-- This combinator is the most general grouping combinator and can be used to
458-- implement all other grouping combinators.
460-- XXX check if this can implement the splitOn combinator i.e. we can slide in
461-- new elements, slide out old elements and incrementally compute the hash.
462-- Also, can we implement the windowed classification combinators using this?
464-- In fact this is a parse. Instead of using a special return value in the fold
465-- we are using a mapping function.
467-- Note that 'scanl'' (usually followed by a map to extract the desired value
468-- from the accumulator) can be used to realize many implementations e.g. a
469-- sliding window implementation. A scan followed by a mapMaybe is also a good
470-- pattern to express many problems where we want to emit a filtered output and
471-- not emit an output on every input.
473-- Passing on of the initial accumulator value to the next fold is equivalent
474-- to returning the leftover concept.
477-- | @groupScan splitter fold stream@ folds the input stream using @fold@.
478-- @splitter@ is applied on the accumulator of the fold every time an item is
479-- consumed by the fold. The fold continues until @splitter@ returns a 'Just'
480-- value. A 'Just' result from the @splitter@ specifies a result to be emitted
481-- in the output stream and the initial value of the accumulator for the next
482-- group's fold. This allows us to control whether to start fresh for the next
483-- fold or to continue from the previous fold's output.
485{-# INLINE groupScan #-}
487 :: (IsStream t, Monad m)
488 => (x -> m (Maybe (b, x))) -> Fold m a x -> t m a -> t m b
489groupScan split fold m = undefined
492------------------------------------------------------------------------------
494------------------------------------------------------------------------------
496-- | @groupsBy cmp f $ S.fromList [a,b,c,...]@ assigns the element @a@ to the
497-- first group, if @b \`cmp` a@ is 'True' then @b@ is also assigned to the same
498-- group. If @c \`cmp` a@ is 'True' then @c@ is also assigned to the same
499-- group and so on. When the comparison fails a new group is started. Each
500-- group is folded using the fold @f@ and the result of the fold is emitted in
503-- >>> Stream.toList $ Stream.groupsBy (>) Fold.toList $ Stream.fromList [1,3,7,0,2,5]
507{-# INLINE groupsBy #-}
509 :: (IsStream t, Monad m)
514groupsBy cmp f m = fromStreamD $ D.groupsBy cmp f (toStreamD m)
516-- | Unlike @groupsBy@ this function performs a rolling comparison of two
517-- successive elements in the input stream. @groupsByRolling cmp f $ S.fromList
518-- [a,b,c,...]@ assigns the element @a@ to the first group, if @a \`cmp` b@ is
519-- 'True' then @b@ is also assigned to the same group. If @b \`cmp` c@ is
520-- 'True' then @c@ is also assigned to the same group and so on. When the
521-- comparison fails a new group is started. Each group is folded using the fold
524-- >>> Stream.toList $ Stream.groupsByRolling (\a b -> a + 1 == b) Fold.toList $ Stream.fromList [1,2,3,7,8,9]
528{-# INLINE groupsByRolling #-}
530 :: (IsStream t, Monad m)
535groupsByRolling cmp f m = fromStreamD $ D.groupsRollingBy cmp f (toStreamD m)
538-- > groups = groupsBy (==)
539-- > groups = groupsByRolling (==)
541-- Groups contiguous spans of equal elements together in individual groups.
543-- >>> Stream.toList $ Stream.groups Fold.toList $ Stream.fromList [1,1,2,2]
548groups :: (IsStream t, Monad m, Eq a) => Fold m a b -> t m a -> t m b
549groups = groupsBy (==)
551------------------------------------------------------------------------------
552-- Splitting - by a predicate
553------------------------------------------------------------------------------
555-- In general we can use deintercalate for splitting. Then we can also use
556-- uniqBy to condense the separators. One way to generalize splitting is to
559-- data Segment a b = Empty | Segment b | Separator a
561-- XXX splitOn and splitOnSuffix have a different behavior on an empty stream,
564-- | Split on an infixed separator element, dropping the separator. The
565-- supplied 'Fold' is applied on the split segments. Splits the stream on
566-- separator elements determined by the supplied predicate, separator is
567-- considered as infixed between two segments:
569-- >>> splitOn' p xs = Stream.toList $ Stream.splitOn p Fold.toList (Stream.fromList xs)
570-- >>> splitOn' (== '.') "a.b"
573-- An empty stream is folded to the default value of the fold:
575-- >>> splitOn' (== '.') ""
578-- If one or both sides of the separator are missing then the empty segment on
579-- that side is folded to the default output of the fold:
581-- >>> splitOn' (== '.') "."
584-- >>> splitOn' (== '.') ".a"
587-- >>> splitOn' (== '.') "a."
590-- >>> splitOn' (== '.') "a..b"
593-- splitOn is an inverse of intercalating single element:
595-- > Stream.intercalate (Stream.fromPure '.') Unfold.fromList . Stream.splitOn (== '.') Fold.toList === id
597-- Assuming the input stream does not contain the separator:
599-- > Stream.splitOn (== '.') Fold.toList . Stream.intercalate (Stream.fromPure '.') Unfold.fromList === id
603{-# INLINE splitOn #-}
605 :: (IsStream t, Monad m)
606 => (a -> Bool) -> Fold m a b -> t m a -> t m b
608 -- We can express the infix splitting in terms of optional suffix split
609 -- fold. After applying a suffix split fold repeatedly if the last segment
610 -- ends with a suffix then we need to return the default output of the fold
611 -- after that to make it an infix split.
613 -- Alternately, we can also express it using an optional prefix split fold.
614 -- If the first segment starts with a prefix then we need to emit the
615 -- default output of the fold before that to make it an infix split, and
616 -- then apply prefix split fold repeatedly.
618 -- Since a suffix split fold can be easily expressed using a
619 -- non-backtracking fold, we use that.
620 foldManyPost (FL.takeEndBy_ predicate f)
622-- | Split on a suffixed separator element, dropping the separator. The
623-- supplied 'Fold' is applied on the split segments.
625-- >>> splitOnSuffix' p xs = Stream.toList $ Stream.splitOnSuffix p Fold.toList (Stream.fromList xs)
626-- >>> splitOnSuffix' (== '.') "a.b."
629-- >>> splitOnSuffix' (== '.') "a."
632-- An empty stream results in an empty output stream:
634-- >>> splitOnSuffix' (== '.') ""
637-- An empty segment consisting of only a suffix is folded to the default output
640-- >>> splitOnSuffix' (== '.') "."
643-- >>> splitOnSuffix' (== '.') "a..b.."
646-- A suffix is optional at the end of the stream:
648-- >>> splitOnSuffix' (== '.') "a"
651-- >>> splitOnSuffix' (== '.') ".a"
654-- >>> splitOnSuffix' (== '.') "a.b"
657-- > lines = splitOnSuffix (== '\n')
659-- 'splitOnSuffix' is an inverse of 'intercalateSuffix' with a single element:
661-- > Stream.intercalateSuffix (Stream.fromPure '.') Unfold.fromList . Stream.splitOnSuffix (== '.') Fold.toList === id
663-- Assuming the input stream does not contain the separator:
665-- > Stream.splitOnSuffix (== '.') Fold.toList . Stream.intercalateSuffix (Stream.fromPure '.') Unfold.fromList === id
669{-# INLINE splitOnSuffix #-}
671 :: (IsStream t, Monad m)
672 => (a -> Bool) -> Fold m a b -> t m a -> t m b
673splitOnSuffix predicate f = foldMany (FL.takeEndBy_ predicate f)
675-- | Split on a prefixed separator element, dropping the separator. The
676-- supplied 'Fold' is applied on the split segments.
679-- > splitOnPrefix' p xs = Stream.toList $ Stream.splitOnPrefix p (Fold.toList) (Stream.fromList xs)
680-- > splitOnPrefix' (== '.') ".a.b"
684-- An empty stream results in an empty output stream:
686-- > splitOnPrefix' (== '.') ""
690-- An empty segment consisting of only a prefix is folded to the default output
694-- > splitOnPrefix' (== '.') "."
697-- > splitOnPrefix' (== '.') ".a.b."
700-- > splitOnPrefix' (== '.') ".a..b"
705-- A prefix is optional at the beginning of the stream:
708-- > splitOnPrefix' (== '.') "a"
711-- > splitOnPrefix' (== '.') "a.b"
715-- 'splitOnPrefix' is an inverse of 'intercalatePrefix' with a single element:
717-- > Stream.intercalatePrefix (Stream.fromPure '.') Unfold.fromList . Stream.splitOnPrefix (== '.') Fold.toList === id
719-- Assuming the input stream does not contain the separator:
721-- > Stream.splitOnPrefix (== '.') Fold.toList . Stream.intercalatePrefix (Stream.fromPure '.') Unfold.fromList === id
725{-# INLINE splitOnPrefix #-}
726splitOnPrefix :: -- (IsStream t, MonadCatch m) =>
727 (a -> Bool) -> Fold m a b -> t m a -> t m b
728splitOnPrefix _predicate _f = undefined
729 -- parseMany (Parser.sliceBeginBy predicate f)
731-- | Like 'splitOn' after stripping leading, trailing, and repeated separators.
732-- Therefore, @".a..b."@ with '.' as the separator would be parsed as
733-- @["a","b"]@. In other words, its like parsing words from whitespace
736-- >>> wordsBy' p xs = Stream.toList $ Stream.wordsBy p Fold.toList (Stream.fromList xs)
738-- >>> wordsBy' (== ',') ""
741-- >>> wordsBy' (== ',') ","
744-- >>> wordsBy' (== ',') ",a,,b,"
747-- > words = wordsBy isSpace
751-- It is equivalent to splitting in any of the infix/prefix/suffix styles
752-- followed by removal of empty segments.
753{-# INLINE wordsBy #-}
755 :: (IsStream t, Monad m)
756 => (a -> Bool) -> Fold m a b -> t m a -> t m b
757wordsBy predicate f m =
758 fromStreamD $ D.wordsBy predicate f (toStreamD m)
760-- | Like 'splitOnSuffix' but keeps the suffix attached to the resulting
763-- >>> splitWithSuffix' p xs = Stream.toList $ splitWithSuffix p Fold.toList (Stream.fromList xs)
765-- >>> splitWithSuffix' (== '.') ""
768-- >>> splitWithSuffix' (== '.') "."
771-- >>> splitWithSuffix' (== '.') "a"
774-- >>> splitWithSuffix' (== '.') ".a"
777-- >>> splitWithSuffix' (== '.') "a."
780-- >>> splitWithSuffix' (== '.') "a.b"
783-- >>> splitWithSuffix' (== '.') "a.b."
786-- >>> splitWithSuffix' (== '.') "a..b.."
787-- ["a.",".","b.","."]
791{-# INLINE splitWithSuffix #-}
793 :: (IsStream t, Monad m)
794 => (a -> Bool) -> Fold m a b -> t m a -> t m b
795splitWithSuffix predicate f = foldMany (FL.takeEndBy predicate f)
797------------------------------------------------------------------------------
798-- Splitting - on a delimiter sequence
799------------------------------------------------------------------------------
801-- Int list examples for splitOn:
803-- >>> splitList [] [1,2,3,3,4]
804-- > [[1],[2],[3],[3],[4]]
806-- >>> splitList [5] [1,2,3,3,4]
809-- >>> splitList [1] [1,2,3,3,4]
812-- >>> splitList [4] [1,2,3,3,4]
815-- >>> splitList [2] [1,2,3,3,4]
818-- >>> splitList [3] [1,2,3,3,4]
821-- >>> splitList [3,3] [1,2,3,3,4]
824-- >>> splitList [1,2,3,3,4] [1,2,3,3,4]
828-- This can be implemented easily using Rabin Karp
829-- | Split on any one of the given patterns.
830{-# INLINE splitOnAny #-}
832 :: (IsStream t, Monad m, Storable a, Integral a)
833 => [Array a] -> Fold m a b -> t m a -> t m b
834splitOnAny subseq f m = undefined
835 -- fromStreamD $ D.splitOnAny f subseq (toStreamD m)
838-- XXX use a non-monadic intersperse to remove the MonadAsync constraint.
839-- XXX Use two folds, one ring buffer fold for separator sequence and the other
840-- split consumer fold. The input is fed to the ring fold first and the
841-- rejected input is fed to the split fold. If the separator matches, the ring
842-- fold would consume all.
844-- | Like 'splitOnSeq' but splits the separator as well, as an infix token.
846-- >>> splitOn'_ pat xs = Stream.toList $ Stream.splitBySeq (Array.fromList pat) Fold.toList (Stream.fromList xs)
848-- >>> splitOn'_ "" "hello"
849-- ["h","","e","","l","","l","","o"]
851-- >>> splitOn'_ "hello" ""
854-- >>> splitOn'_ "hello" "hello"
857-- >>> splitOn'_ "x" "hello"
860-- >>> splitOn'_ "h" "hello"
863-- >>> splitOn'_ "o" "hello"
866-- >>> splitOn'_ "e" "hello"
869-- >>> splitOn'_ "l" "hello"
870-- ["he","l","","l","o"]
872-- >>> splitOn'_ "ll" "hello"
876{-# INLINE splitBySeq #-}
878 :: (IsStream t, MonadAsync m, Storable a, Enum a, Eq a)
879 => Array a -> Fold m a b -> t m a -> t m b
881 intersperseM (fold f (A.toStream patt)) $ splitOnSeq patt f m
883-- | Like 'splitSuffixBy' but the separator is a sequence of elements, instead
884-- of a predicate for a single element.
886-- >>> splitOnSuffixSeq_ pat xs = Stream.toList $ Stream.splitOnSuffixSeq (Array.fromList pat) Fold.toList (Stream.fromList xs)
888-- >>> splitOnSuffixSeq_ "." ""
891-- >>> splitOnSuffixSeq_ "." "."
894-- >>> splitOnSuffixSeq_ "." "a"
897-- >>> splitOnSuffixSeq_ "." ".a"
900-- >>> splitOnSuffixSeq_ "." "a."
903-- >>> splitOnSuffixSeq_ "." "a.b"
906-- >>> splitOnSuffixSeq_ "." "a.b."
909-- >>> splitOnSuffixSeq_ "." "a..b.."
912-- > lines = splitOnSuffixSeq "\n"
914-- 'splitOnSuffixSeq' is an inverse of 'intercalateSuffix'. The following law
917-- > intercalateSuffix . splitOnSuffixSeq == id
919-- The following law holds when the separator is non-empty and contains none of
920-- the elements present in the input lists:
922-- > splitSuffixOn . intercalateSuffix == id
925{-# INLINE splitOnSuffixSeq #-}
927 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
928 => Array a -> Fold m a b -> t m a -> t m b
929splitOnSuffixSeq patt f m =
930 fromStreamD $ D.splitOnSuffixSeq False patt f (toStreamD m)
933-- | Like 'splitOn' but drops any empty splits.
935{-# INLINE wordsOn #-}
937 :: (IsStream t, Monad m, Storable a, Eq a)
938 => Array a -> Fold m a b -> t m a -> t m b
939wordsOn subseq f m = undefined
940 -- fromStreamD $ D.wordsOn f subseq (toStreamD m)
943-- | Like 'splitOnSuffixSeq' but keeps the suffix intact in the splits.
945-- >>> splitWithSuffixSeq' pat xs = Stream.toList $ Stream.splitWithSuffixSeq (Array.fromList pat) Fold.toList (Stream.fromList xs)
947-- >>> splitWithSuffixSeq' "." ""
950-- >>> splitWithSuffixSeq' "." "."
953-- >>> splitWithSuffixSeq' "." "a"
956-- >>> splitWithSuffixSeq' "." ".a"
959-- >>> splitWithSuffixSeq' "." "a."
962-- >>> splitWithSuffixSeq' "." "a.b"
965-- >>> splitWithSuffixSeq' "." "a.b."
968-- >>> splitWithSuffixSeq' "." "a..b.."
969-- ["a.",".","b.","."]
972{-# INLINE splitWithSuffixSeq #-}
974 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
975 => Array a -> Fold m a b -> t m a -> t m b
976splitWithSuffixSeq patt f m =
977 fromStreamD $ D.splitOnSuffixSeq True patt f (toStreamD m)
980-- This can be implemented easily using Rabin Karp
981-- | Split post any one of the given patterns.
982{-# INLINE splitOnSuffixSeqAny #-}
984 :: (IsStream t, Monad m, Storable a, Integral a)
985 => [Array a] -> Fold m a b -> t m a -> t m b
986splitOnSuffixSeqAny subseq f m = undefined
987 -- fromStreamD $ D.splitPostAny f subseq (toStreamD m)
990------------------------------------------------------------------------------
992------------------------------------------------------------------------------
994-- | Group the input stream into groups of @n@ elements each and then fold each
995-- group using the provided fold function.
997-- >>> Stream.toList $ Stream.chunksOf 2 Fold.sum (Stream.enumerateFromTo 1 10)
1000-- This can be considered as an n-fold version of 'take' where we apply
1001-- 'take' repeatedly on the leftover stream until the stream exhausts.
1003-- @chunksOf n f = foldMany (FL.take n f)@
1006{-# INLINE chunksOf #-}
1008 :: (IsStream t, Monad m)
1009 => Int -> Fold m a b -> t m a -> t m b
1010chunksOf n f = fromStreamD . D.chunksOf n f . toStreamD
1012-- | @arraysOf n stream@ groups the elements in the input stream into arrays of
1013-- @n@ elements each.
1015-- Same as the following but may be more efficient:
1017-- > arraysOf n = Stream.foldMany (A.writeN n)
1020{-# INLINE arraysOf #-}
1021arraysOf :: (IsStream t, MonadIO m, Storable a)
1022 => Int -> t m a -> t m (Array a)
1023arraysOf n = fromStreamD . A.arraysOf n . toStreamD
1025-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
1026-- XXX add this example after fixing the serial stream rate control
1028-- | Group the input stream into windows of @n@ second each and then fold each
1029-- group using the provided fold function.
1031-- >>> Stream.toList $ Stream.take 5 $ Stream.intervalsOf 1 Fold.sum $ Stream.constRate 2 $ Stream.enumerateFrom 1
1032-- [...,...,...,...,...]
1035{-# INLINE intervalsOf #-}
1037 :: (IsStream t, MonadAsync m)
1038 => Double -> Fold m a b -> t m a -> t m b
1040 splitWithSuffix isNothing (FL.catMaybes f)
1041 (interjectSuffix n (return Nothing) (map Just xs))
1043-- XXX This can be implemented more efficiently by sharing a Clock.Timer across
1044-- parallel threads and resetting it whenever a span is emitted.
1046-- | Like 'chunksOf' but if the chunk is not completed within the specified
1047-- time interval then emit whatever we have collected till now. The chunk
1048-- timeout is reset whenever a chunk is emitted.
1050-- >>> s = Stream.delayPost 0.3 $ Stream.fromList [1..1000]
1051-- >>> f = Stream.mapM_ print $ Stream.chunksOfTimeout 5 1 Fold.toList s
1054{-# INLINE chunksOfTimeout #-}
1055chunksOfTimeout :: (IsStream t, MonadAsync m, Functor (t m))
1056 => Int -> Double -> FL.Fold m a b -> t m a -> t m b
1057chunksOfTimeout n timeout f =
1059 . classifySessionsBy
1060 timeout False (const (return False)) timeout (FL.take n f)
1061 . Transform.timestamped
1064------------------------------------------------------------------------------
1065-- Windowed classification
1066------------------------------------------------------------------------------
1068-- We divide the stream into windows or chunks in space or time and each window
1069-- can be associated with a key, all events associated with a particular key in
1070-- the window can be folded to a single result. The stream can be split into
1071-- windows by size or by using a split predicate on the elements in the stream.
1072-- For example, when we receive a closing flag, we can close the window.
1074-- A "chunk" is a space window and a "session" is a time window. Are there any
1075-- other better short words to describe them. An alternative is to use
1076-- "swindow" and "twindow". Another word for "session" could be "spell".
1078-- TODO: To mark the position in space or time we can have Indexed or
1079-- TimeStamped types. That can make it easy to deal with the position indices
1082------------------------------------------------------------------------------
1083-- Keyed Sliding Windows
1084------------------------------------------------------------------------------
1087{-# INLINABLE classifySlidingChunks #-}
1088classifySlidingChunks
1089 :: (IsStream t, MonadAsync m, Ord k)
1090 => Int -- ^ window size
1091 -> Int -- ^ window slide
1092 -> Fold m a b -- ^ Fold to be applied to window events
1093 -> t m (k, a, Bool) -- ^ window key, data, close event
1095classifySlidingChunks wsize wslide (Fold step initial extract) str
1098-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
1099-- slide the send window when an ack is received and we slide the receive
1100-- window when a sequence is complete. Sliding is stateful in case of TCP,
1101-- sliding releases the send buffer or makes data available to the user from
1102-- the receive buffer.
1103{-# INLINABLE classifySlidingSessions #-}
1104classifySlidingSessions
1105 :: (IsStream t, MonadAsync m, Ord k)
1106 => Double -- ^ timer tick in seconds
1107 -> Double -- ^ time window size
1108 -> Double -- ^ window slide
1109 -> Fold m a b -- ^ Fold to be applied to window events
1110 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
1112classifySlidingSessions tick interval slide (Fold step initial extract) str
1116------------------------------------------------------------------------------
1117-- Sliding Window Buffers
1118------------------------------------------------------------------------------
1120-- These buffered versions could be faster than concurrent incremental folds of
1121-- all overlapping windows as in many cases we may not need all the values to
1122-- compute the fold, we can just compute the result using the old value and new
1123-- value. However, we may need the buffer once in a while, for example for
1124-- string search we usually compute the hash incrementally but when the hash
1125-- matches the hash of the pattern we need to compare the whole string.
1127-- XXX we should be able to implement sequence based splitting combinators
1128-- using this combinator.
1131-- | Buffer n elements of the input in a ring buffer. When t new elements are
1132-- collected, slide the window to remove the same number of oldest elements,
1133-- insert the new elements, and apply an incremental fold on the sliding
1134-- window, supplying the outgoing elements, the new ring buffer as arguments.
1136 :: (IsStream t, Monad m, Ord a, Storable a)
1137 => Int -- window size
1138 -> Int -- window slide
1139 -> Fold m (Ring a, Array a) b
1142slidingChunkBuffer = undefined
1144-- Buffer n seconds worth of stream elements of the input in a radix tree.
1145-- Every t seconds, remove the items that are older than n seconds, and apply
1146-- an incremental fold on the sliding window, supplying the outgoing elements,
1147-- and the new radix tree buffer as arguments.
1149 :: (IsStream t, Monad m, Ord a, Storable a)
1150 => Int -- window size
1152 -> Fold m (RTree a, Array a) b
1155slidingSessionBuffer = undefined
1158------------------------------------------------------------------------------
1159-- Keyed Session Windows
1160------------------------------------------------------------------------------
1163-- | Keyed variable size space windows. Close the window if we do not receive a
1164-- window event in the next "spaceout" elements.
1165{-# INLINABLE classifyChunksBy #-}
1167 :: (IsStream t, MonadAsync m, Ord k)
1168 => Int -- ^ window spaceout (spread)
1169 -> Bool -- ^ reset the spaceout when a chunk window element is received
1170 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1171 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1173classifyChunksBy spanout reset (Fold step initial extract) str = undefined
1175-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
1176-- received within the chunk size window. The chunk gets closed only if no
1177-- element is received within the chunk window.
1179{-# INLINABLE classifyKeepAliveChunks #-}
1180classifyKeepAliveChunks
1181 :: (IsStream t, MonadAsync m, Ord k)
1182 => Int -- ^ window spaceout (spread)
1183 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1184 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1186classifyKeepAliveChunks spanout = classifyChunksBy spanout True
1189data SessionState t m k a b = SessionState
1190 { sessionCurTime :: !AbsTime -- ^ time since last event
1191 , sessionEventTime :: !AbsTime -- ^ time as per last event
1192 , sessionCount :: !Int -- ^ total number sessions in progress
1193 , sessionTimerHeap :: H.Heap (H.Entry AbsTime k) -- ^ heap for timeouts
1194 , sessionKeyValueMap :: Map.Map k a -- ^ Stored sessions for keys
1195 , sessionOutputStream :: t (m :: Type -> Type) (k, b) -- ^ Completed sessions
1198data SessionEntry a b = LiveSession !a !b | ZombieSession
1200-- | @classifySessionsBy tick keepalive predicate timeout fold stream@
1201-- classifies an input event @stream@ consisting of @(timestamp, (key,
1202-- value))@ into sessions based on the @key@, folding all the values
1203-- corresponding to the same key into a session using the supplied @fold@.
1205-- When the fold terminates or a @timeout@ occurs, a tuple consisting of the
1206-- session key and the folded value is emitted in the output stream. The
1207-- timeout is measured from the first event in the session. If the @keepalive@
1208-- option is set to 'True' the timeout is reset to 0 whenever an event is
1211-- The @timestamp@ in the input stream is an absolute time from some epoch,
1212-- characterizing the time when the input event was generated. The notion of
1213-- current time is maintained by a monotonic event time clock using the
1214-- timestamps seen in the input stream. The latest timestamp seen till now is
1215-- used as the base for the current time. When no new events are seen, a timer
1216-- is started with a clock resolution of @tick@ seconds. This timer is used to
1217-- detect session timeouts in the absence of new events.
1219-- To ensure an upper bound on the memory used the number of sessions can be
1220-- limited to an upper bound. If the ejection @predicate@ returns 'True', the
1221-- oldest session is ejected before inserting a new session.
1224-- Stream.mapM_ print
1225-- $ Stream.classifySessionsBy 1 False (const (return False)) 3 (Fold.take 3 Fold.toList)
1226-- $ Stream.timestamped
1227-- $ Stream.delay 0.1
1228-- $ (,) <$> Stream.fromList [1,2,3] <*> Stream.fromList ['a','b','c']
1236{-# INLINABLE classifySessionsBy #-}
1238 :: (IsStream t, MonadAsync m, Ord k)
1239 => Double -- ^ timer tick in seconds
1240 -> Bool -- ^ reset the timer when an event is received
1241 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
1242 -> Double -- ^ session timeout in seconds
1243 -> Fold m a b -- ^ Fold to be applied to session data
1244 -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
1245 -> t m (k, b) -- ^ session key, fold result
1246classifySessionsBy tick reset ejectPred tmout
1247 (Fold step initial extract) str =
1248 concatMap sessionOutputStream $
1249 scanlMAfter' sstep (return szero) flush stream
1253 timeoutMs = toRelTime (round (tmout * 1000) :: MilliSecond64)
1254 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
1255 szero = SessionState
1256 { sessionCurTime = toAbsTime (0 :: MilliSecond64)
1257 , sessionEventTime = toAbsTime (0 :: MilliSecond64)
1259 , sessionTimerHeap = H.empty
1260 , sessionKeyValueMap = Map.empty
1261 , sessionOutputStream = IsStream.nil
1264 -- We can eject sessions based on the current session count to limit
1265 -- memory consumption. There are two possible strategies:
1267 -- 1) Eject old sessions or sessions beyond a certain/lower timeout
1268 -- threshold even before timeout, effectively reduce the timeout.
1269 -- 2) Drop creation of new sessions but keep accepting new events for the
1272 -- We use the first strategy as of now.
1274 -- Got a new stream input element
1275 sstep session@SessionState{..} (Just (timestamp, (key, value))) = do
1276 -- XXX we should use a heap in pinned memory to scale it to a large
1279 -- XXX if the key is an Int, we can also use an IntMap for slightly
1280 -- better performance.
1282 -- When we insert a key in the Map we insert an entry into the heap as
1283 -- well with the session expiry as the sort key. The Map entry
1284 -- consists of the fold result, and the expiry time of the session. If
1285 -- "reset" is True the expiry time is readjusted whenever a new event
1286 -- is processed. If the fold terminates and a new session is started
1287 -- for the same key the expiry time is set to the first timestamp of
1290 -- The heap must have at most one entry for any given key. The heap is
1291 -- processed periodically to remove the expired entries. We pick up an
1292 -- expired entry from the top of the heap and if the session has
1293 -- expired based on the expiry time in the Map entry then we remove the
1294 -- session from the Map and yield its fold output. Otherwise, we
1295 -- reinsert the entry into the heap based on the current expiry in the
1298 -- If an entry is removed from the Map and not removed from the heap
1299 -- and in the meantime it is inserted again in the Map (using the same
1300 -- key) then how do we avoid inserting multiple entries in the heap?
1301 -- For this reason we maintain the invariant that the Map entry is
1302 -- removed only when the heap entry is removed. Even if the fold has
1303 -- finished we still keep a dummy Map entry (ZombieSession) until the
1304 -- heap entry is removed. That way if we have a Map entry we do not
1305 -- insert a heap entry because we know it is already there.
1307 let curTime = max sessionEventTime timestamp
1308 mOld = Map.lookup key sessionKeyValueMap
1310 -- deleting a key from the heap is expensive, so we never
1311 -- delete a key from heap, we just purge it from the Map and it
1312 -- gets purged from the heap on timeout. We just need an extra
1313 -- lookup in the Map when the key is purged from the heap, that
1314 -- should not be expensive.
1316 let (mp, cnt) = case mOld of
1317 Just (LiveSession _ _) ->
1318 ( Map.insert key ZombieSession sessionKeyValueMap
1321 _ -> (sessionKeyValueMap, sessionCount)
1323 { sessionCurTime = curTime
1324 , sessionEventTime = curTime
1325 , sessionCount = cnt
1326 , sessionKeyValueMap = mp
1327 , sessionOutputStream = fromPure (key, fb)
1330 let expiry = addToAbsTime timestamp timeoutMs
1331 (hp1, mp1, out1, cnt1) <- do
1332 let vars = (sessionTimerHeap, sessionKeyValueMap,
1333 IsStream.nil, sessionCount)
1335 -- inserting new entry
1337 -- Eject a session from heap and map if needed
1338 eject <- ejectPred sessionCount
1339 (hp, mp, out, cnt) <-
1344 -- Insert the new session in heap
1345 let hp' = H.insert (Entry expiry key) hp
1346 in return (hp', mp, out, cnt + 1)
1347 -- updating old entry
1348 Just _ -> return vars
1350 let acc = LiveSession expiry fs1
1351 mp2 = Map.insert key acc mp1
1352 return $ SessionState
1353 { sessionCurTime = curTime
1354 , sessionEventTime = curTime
1355 , sessionCount = cnt1
1356 , sessionTimerHeap = hp1
1357 , sessionKeyValueMap = mp2
1358 , sessionOutputStream = out1
1362 Just (LiveSession _ acc) -> return $ FL.Partial acc
1366 error $ "classifySessionsBy: "
1367 ++ "The supplied fold must consume at least one input"
1369 res <- step fs value
1371 FL.Done fb -> done fb
1372 FL.Partial fs1 -> partial fs1
1374 -- Got a timer tick event
1375 sstep sessionState@SessionState{..} Nothing =
1376 let curTime = addToAbsTime sessionCurTime tickMs
1377 in ejectExpired sessionState curTime
1379 flush session@SessionState{..} = do
1380 (hp', mp', out, count) <-
1383 , sessionKeyValueMap
1388 { sessionCount = count
1389 , sessionTimerHeap = hp'
1390 , sessionKeyValueMap = mp'
1391 , sessionOutputStream = out
1394 -- delete from map and output the fold accumulator
1395 ejectEntry hp mp out cnt acc key = do
1397 let out1 = (key, sess) `cons` out
1398 let mp1 = Map.delete key mp
1399 return (hp, mp1, out1, cnt - 1)
1401 ejectAll (hp, mp, out, !cnt) = do
1402 let hres = H.uncons hp
1404 Just (Entry _ key, hp1) -> do
1405 r <- case Map.lookup key mp of
1406 Nothing -> return (hp1, mp, out, cnt)
1407 Just ZombieSession ->
1408 return (hp1, Map.delete key mp, out, cnt)
1409 Just (LiveSession _ acc) ->
1410 ejectEntry hp1 mp out cnt acc key
1413 assert (Map.null mp) (return ())
1414 return (hp, mp, out, cnt)
1416 ejectOne (hp, mp, out, !cnt) = do
1417 let hres = H.uncons hp
1419 Just (Entry expiry key, hp1) ->
1420 case Map.lookup key mp of
1421 Nothing -> ejectOne (hp1, mp, out, cnt)
1422 Just ZombieSession ->
1423 ejectOne (hp1, Map.delete key mp, out, cnt)
1424 Just (LiveSession expiry1 acc) -> do
1425 if not reset || expiry1 <= expiry
1426 then ejectEntry hp1 mp out cnt acc key
1428 -- reset the session timeout and continue
1429 let hp2 = H.insert (Entry expiry1 key) hp1
1430 in ejectOne (hp2, mp, out, cnt)
1432 assert (Map.null mp) (return ())
1433 return (hp, mp, out, cnt)
1435 ejectExpired session@SessionState{..} curTime = do
1436 (hp', mp', out, count) <-
1438 sessionTimerHeap sessionKeyValueMap IsStream.nil sessionCount
1440 { sessionCurTime = curTime
1441 , sessionCount = count
1442 , sessionTimerHeap = hp'
1443 , sessionKeyValueMap = mp'
1444 , sessionOutputStream = out
1449 ejectLoop hp mp out !cnt = do
1450 let hres = H.uncons hp
1452 Just (Entry expiry key, hp1) -> do
1454 if curTime >= expiry
1455 then return (True, False)
1461 case Map.lookup key mp of
1462 Nothing -> ejectLoop hp1 mp out cnt
1463 Just ZombieSession ->
1464 ejectLoop hp1 (Map.delete key mp) out cnt
1465 Just (LiveSession expiry1 acc) -> do
1466 if expiry1 <= curTime || not reset || force
1468 (hp2,mp1,out1,cnt1) <-
1469 ejectEntry hp1 mp out cnt acc key
1470 ejectLoop hp2 mp1 out1 cnt1
1472 -- reset the session timeout and continue
1473 let hp2 = H.insert (Entry expiry1 key) hp1
1474 in ejectLoop hp2 mp out cnt
1475 else return (hp, mp, out, cnt)
1477 assert (Map.null mp) (return ())
1478 return (hp, mp, out, cnt)
1480 -- merge timer events in the stream
1481 stream = map Just str `parallelFst` repeatM timer
1483 liftIO $ threadDelay (round $ tick * 1000000)
1486-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
1487-- option set to 'True'.
1490-- classifyKeepAliveSessions = classifySessionsBy 1 True
1495{-# INLINABLE classifyKeepAliveSessions #-}
1496classifyKeepAliveSessions ::
1497 (IsStream t, MonadAsync m, Ord k)
1498 => (Int -> m Bool) -- ^ predicate to eject sessions on session count
1499 -> Double -- ^ session inactive timeout
1500 -> Fold m a b -- ^ Fold to be applied to session payload data
1501 -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
1503classifyKeepAliveSessions = classifySessionsBy 1 True
1505------------------------------------------------------------------------------
1506-- Keyed tumbling windows
1507------------------------------------------------------------------------------
1509-- Tumbling windows is a special case of sliding windows where the window slide
1510-- is the same as the window size. Or it can be a special case of session
1511-- windows where the reset flag is set to False.
1513-- XXX instead of using the early termination flag in the stream, we can use an
1514-- early terminating fold instead.
1517-- | Split the stream into fixed size chunks of specified size. Within each
1518-- such chunk fold the elements in buckets identified by the keys. A particular
1519-- bucket fold can be terminated early if a closing flag is encountered in an
1520-- element for that key.
1523{-# INLINABLE classifyChunksOf #-}
1525 :: (IsStream t, MonadAsync m, Ord k)
1526 => Int -- ^ window size
1527 -> Fold m a b -- ^ Fold to be applied to window events
1528 -> t m (k, a, Bool) -- ^ window key, data, close event
1530classifyChunksOf wsize = classifyChunksBy wsize False
1533-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
1534-- option set to 'False'.
1537-- classifySessionsOf = classifySessionsBy 1 False
1542{-# INLINABLE classifySessionsOf #-}
1543classifySessionsOf ::
1544 (IsStream t, MonadAsync m, Ord k)
1545 => (Int -> m Bool) -- ^ predicate to eject sessions on session count
1546 -> Double -- ^ time window size
1547 -> Fold m a b -- ^ Fold to be applied to session data
1548 -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
1550classifySessionsOf = classifySessionsBy 1 False
1552------------------------------------------------------------------------------
1554------------------------------------------------------------------------------
1556-- | @splitInnerBy splitter joiner stream@ splits the inner containers @f a@ of
1557-- an input stream @t m (f a)@ using the @splitter@ function. Container
1558-- elements @f a@ are collected until a split occurs, then all the elements
1559-- before the split are joined using the @joiner@ function.
1561-- For example, if we have a stream of @Array Word8@, we may want to split the
1562-- stream into arrays representing lines separated by '\n' byte such that the
1563-- resulting stream after a split would be one array for each line.
1565-- CAUTION! This is not a true streaming function as the container size after
1566-- the split and merge may not be bounded.
1569{-# INLINE splitInnerBy #-}
1571 :: (IsStream t, Monad m)
1572 => (f a -> m (f a, Maybe (f a))) -- splitter
1573 -> (f a -> f a -> m (f a)) -- joiner
1576splitInnerBy splitter joiner xs =
1577 fromStreamD $ D.splitInnerBy splitter joiner $ toStreamD xs
1579-- | Like 'splitInnerBy' but splits assuming the separator joins the segment in
1583{-# INLINE splitInnerBySuffix #-}
1585 :: (IsStream t, Monad m, Eq (f a), Monoid (f a))
1586 => (f a -> m (f a, Maybe (f a))) -- splitter
1587 -> (f a -> f a -> m (f a)) -- joiner
1590splitInnerBySuffix splitter joiner xs =
1591 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