1{-# OPTIONS_HADDOCK hide #-}
2{-# LANGUAGE BangPatterns #-}
4{-# LANGUAGE ExistentialQuantification #-}
5{-# LANGUAGE FlexibleContexts #-}
6{-# LANGUAGE RankNTypes #-}
7{-# LANGUAGE RecordWildCards #-}
8{-# LANGUAGE ScopedTypeVariables #-}
11-- Module : Streamly.Internal.Data.Pipe
12-- Copyright : (c) 2019 Composewell Technologies
14-- Maintainer : streamly@composewell.com
15-- Stability : experimental
18-- There are three fundamental types in streamly. They are streams
19-- ("Streamly.Prelude"), pipes ("Streamly.Internal.Data.Pipe") and folds ("Streamly.Data.Fold").
20-- Streams are sources or producers of values, multiple sources can be merged
21-- into a single source but a source cannot be split into multiple stream
22-- sources. Folds are sinks or consumers, a stream can be split and
23-- distributed to multiple folds but the results cannot be merged back into a
24-- stream source again. Pipes are transformations, a stream source can be split
25-- and distributed to multiple pipes each pipe can apply its own transform on
26-- the stream and the results can be merged back into a single pipe. Pipes can
27-- be attached to a source to produce a source or they can be attached to a
28-- fold to produce a fold, or multiple pipes can be merged or zipped into a
31-- > import qualified Streamly.Internal.Data.Pipe as P
33module Streamly.Internal.Data.Pipe
55 -- ** Scanning Filters
60 -- | Insertion adds more elements to the stream.
82 -- | Streams can be split into segments in space or in time. We use the
83 -- term @chunk@ to refer to a spatial length of the stream (spatial window)
84 -- and the term @session@ to refer to a length in time (time window).
86 -- In imperative terms, grouped folding can be considered as a nested loop
87 -- where we loop over the stream to group elements and then loop over
88 -- individual groups to fold them to a single value that is yielded in the
108 -- Keeping the delimiters
113 -- Splitting by multiple sequences
115 -- , splitSuffixOnAny
116 -- , splitPrefixOnAny
132 -- The 'Applicative' instance of a distributing 'Fold' distributes one copy
133 -- of the stream to each fold and combines the results using a function.
137 -- |-------Fold m a b--------|
138 -- ---stream m a---| |---m (b,c,...)
139 -- |-------Fold m a c--------|
144 -- To compute the average of numbers in a stream without going throught he
147 -- >>> let avg = (/) <$> FL.sum <*> fmap fromIntegral FL.length
148 -- >>> FL.foldl' avg (S.enumerateFromTo 1.0 100.0)
151 -- The 'Semigroup' and 'Monoid' instances of a distributing fold distribute
152 -- the input to both the folds and combines the outputs using Monoid or
153 -- Semigroup instances of the output types:
155 -- >>> import Data.Monoid (Sum)
156 -- >>> FL.foldl' (FL.head <> FL.last) (fmap Sum $ S.enumerateFromTo 1.0 100.0)
157 -- Just (Sum {getSum = 101.0})
159 -- The 'Num', 'Floating', and 'Fractional' instances work in the same way.
166 -- Direct items in the input stream to different folds using a function to
167 -- select the fold. This is useful to demultiplex the input stream.
184 -- These can be expressed using lmap/lmapM and unzip
191 , duplicate -- experimental
193 -- * Windowed Classification
194 -- | Split the stream into windows or chunks in space or time. Each window
195 -- can be associated with a key, all events associated with a particular
196 -- key in the window can be folded to a single result. The stream is split
197 -- into windows of specified size, the window can be terminated early if
198 -- the closing flag is specified in the input stream.
200 -- The term "chunk" is used for a space window and the term "session" is
201 -- used for a time window.
203 -- ** Tumbling Windows
204 -- | A new window starts after the previous window is finished.
205 -- , classifyChunksOf
208 -- ** Keep Alive Windows
209 -- | The window size is extended if an event arrives within the specified
210 -- window size. This can represent sessions with idle or inactive timeout.
211 -- , classifyKeepAliveChunks
212 , classifyKeepAliveSessions
215 -- ** Sliding Windows
216 -- | A new window starts after the specified slide from the previous
217 -- window. Therefore windows can overlap.
218 , classifySlidingChunks
219 , classifySlidingSessions
221 -- ** Sliding Window Buffers
222 -- , slidingChunkBuffer
223 -- , slidingSessionBuffer
228-- import Control.Concurrent (threadDelay, forkIO, killThread)
229-- import Control.Concurrent.MVar (MVar, newMVar, takeMVar, putMVar)
230-- import Control.Exception (SomeException(..), catch, mask)
231-- import Control.Monad (void)
232-- import Control.Monad.Catch (throwM)
233-- import Control.Monad.IO.Class (MonadIO(..))
234-- import Control.Monad.Trans (lift)
235-- import Control.Monad.Trans.Control (control)
236-- import Data.Functor.Identity (Identity)
237-- import Data.Heap (Entry(..))
238-- import Data.Map.Strict (Map)
239-- import Data.Maybe (fromJust, isJust, isNothing)
241-- import Foreign.Storable (Storable(..))
243 hiding (id, filter, drop, dropWhile, take, takeWhile, zipWith, foldr,
244 foldl, map, mapM_, sequence, all, any, sum, product, elem,
245 notElem, maximum, minimum, head, last, tail, length, null,
246 reverse, iterate, init, and, or, lookup, foldr1, (!!),
247 scanl, scanl1, replicate, concatMap, mconcat, foldMap, unzip,
248 span, splitAt, break, mapM)
250-- import qualified Data.Heap as H
251-- import qualified Data.Map.Strict as Map
252-- import qualified Prelude
254-- import Streamly (MonadAsync, parallel)
255-- import Streamly.Data.Fold.Types (Fold(..))
256import Streamly.Internal.Data.Pipe.Types
257 (Pipe(..), PipeState(..), Step(..), zipWith, tee, map, compose)
258-- import Streamly.Internal.Memory.Array.Types (Array)
259-- import Streamly.Memory.Ring (Ring)
260-- import Streamly.Streams.Serial (SerialT)
261-- import Streamly.Streams.StreamK (IsStream())
262-- import Streamly.Internal.Data.Time.Units
263-- (AbsTime, MilliSecond64(..), addToAbsTime, diffAbsTime, toRelTime,
266-- import Streamly.Internal.Data.Strict
268-- import qualified Streamly.Internal.Memory.Array.Types as A
269-- import qualified Streamly.Prelude as S
270-- import qualified Streamly.Streams.StreamD as D
271-- import qualified Streamly.Streams.StreamK as K
272-- import qualified Streamly.Streams.Prelude as P
274------------------------------------------------------------------------------
276------------------------------------------------------------------------------
278-- | Lift a monadic function to a 'Pipe'.
282mapM :: Monad m => (a -> m b) -> Pipe m a b
283mapM f = Pipe consume undefined ()
287 return $ Yield r (Consume ())
289------------------------------------------------------------------------------
291------------------------------------------------------------------------------
293-- | Include only those elements that pass a predicate.
295-- >>> FL.foldl (lfilter (> 5) FL.sum) [1..10]
299{-# INLINABLE lfilter #-}
300lfilter :: Monad m => (a -> Bool) -> Fold m a r -> Fold m a r
301lfilter f (Fold step begin done) = Fold step' begin done
303 step' x a = if f a then step x a else return x
305-- | Like 'lfilter' but with a monadic predicate.
308{-# INLINABLE lfilterM #-}
309lfilterM :: Monad m => (a -> m Bool) -> Fold m a r -> Fold m a r
310lfilterM f (Fold step begin done) = Fold step' begin done
314 if use then step x a else return x
316-- | Take first 'n' elements from the stream and discard the rest.
319{-# INLINABLE ltake #-}
320ltake :: Monad m => Int -> Fold m a b -> Fold m a b
321ltake n (Fold step initial done) = Fold step' initial' done'
323 initial' = fmap (Tuple' 0) initial
324 step' (Tuple' i r) a = do
328 return $ Tuple' (i + 1) res
329 else return $ Tuple' i r
330 done' (Tuple' _ r) = done r
332-- | Takes elements from the input as long as the predicate succeeds.
335{-# INLINABLE ltakeWhile #-}
336ltakeWhile :: Monad m => (a -> Bool) -> Fold m a b -> Fold m a b
337ltakeWhile predicate (Fold step initial done) = Fold step' initial' done'
339 initial' = fmap Left' initial
340 step' (Left' r) a = do
342 then fmap Left' $ step r a
343 else return (Right' r)
345 done' (Left' r) = done r
346 done' (Right' r) = done r
348------------------------------------------------------------------------------
350------------------------------------------------------------------------------
352------------------------------------------------------------------------------
353-- Grouping without looking at elements
354------------------------------------------------------------------------------
356------------------------------------------------------------------------------
358------------------------------------------------------------------------------
361-- | @splitAt n f1 f2@ composes folds @f1@ and @f2@ such that first @n@
362-- elements of its input are sent to fold @f1@ and the rest of the stream is
365-- > let splitAt_ n xs = FL.foldl' (FL.splitAt n FL.toList FL.toList) $ S.fromList xs
367-- >>> splitAt_ 6 "Hello World!"
368-- > ("Hello ","World!")
370-- >>> splitAt_ (-1) [1,2,3]
373-- >>> splitAt_ 0 [1,2,3]
376-- >>> splitAt_ 1 [1,2,3]
379-- >>> splitAt_ 3 [1,2,3]
382-- >>> splitAt_ 4 [1,2,3]
385-- This can be considered as a two-fold version of 'ltake' where we take both
386-- the segments instead of discarding the leftover.
389{-# INLINE splitAt #-}
396splitAt n (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
397 Fold step init extract
399 init = Tuple3' <$> return n <*> initialL <*> initialR
401 step (Tuple3' i xL xR) input =
403 then stepL xL input >>= (\a -> return (Tuple3' (i - 1) a xR))
404 else stepR xR input >>= (\b -> return (Tuple3' i xL b))
406 extract (Tuple3' _ a b) = (,) <$> extractL a <*> extractR b
408------------------------------------------------------------------------------
410------------------------------------------------------------------------------
412------------------------------------------------------------------------------
413-- Generalized grouping
414------------------------------------------------------------------------------
416-- This combinator is the most general grouping combinator and can be used to
417-- implement all other grouping combinators.
419-- XXX check if this can implement the splitOn combinator i.e. we can slide in
420-- new elements, slide out old elements and incrementally compute the hash.
421-- Also, can we implement the windowed classification combinators using this?
423-- In fact this is a parse. Instead of using a special return value in the fold
424-- we are using a mapping function.
426-- Note that 'scanl'' (usually followed by a map to extract the desired value
427-- from the accumulator) can be used to realize many implementations e.g. a
428-- sliding window implementation. A scan followed by a mapMaybe is also a good
429-- pattern to express many problems where we want to emit a filtered output and
430-- not emit an output on every input.
432-- Passing on of the initial accumulator value to the next fold is equivalent
433-- to returning the leftover concept.
436-- | @groupScan splitter fold stream@ folds the input stream using @fold@.
437-- @splitter@ is applied on the accumulator of the fold every time an item is
438-- consumed by the fold. The fold continues until @splitter@ returns a 'Just'
439-- value. A 'Just' result from the @splitter@ specifies a result to be emitted
440-- in the output stream and the initial value of the accumulator for the next
441-- group's fold. This allows us to control whether to start fresh for the next
442-- fold or to continue from the previous fold's output.
444{-# INLINE groupScan #-}
446 :: (IsStream t, Monad m)
447 => (x -> m (Maybe (b, x))) -> Fold m a x -> t m a -> t m b
448groupScan split fold m = undefined
451-- | Group the input stream into groups of @n@ elements each and then fold each
452-- group using the provided fold function.
454-- >> S.toList $ FL.chunksOf 2 FL.sum (S.enumerateFromTo 1 10)
457-- This can be considered as an n-fold version of 'ltake' where we apply
458-- 'ltake' repeatedly on the leftover stream until the stream exhausts.
461{-# INLINE chunksOf #-}
463 :: (IsStream t, Monad m)
464 => Int -> Fold m a b -> t m a -> t m b
465chunksOf n f m = D.fromStreamD $ D.groupsOf n f (D.toStreamD m)
467-- | Transform a fold from a pure input to a 'Maybe' input, consuming only
469{-# INLINE lcatMaybes #-}
470lcatMaybes :: Monad m => Fold m a b -> Fold m (Maybe a) b
471lcatMaybes = lfilter isJust . lmap fromJust
473-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
474-- XXX add this example after fixing the serial stream rate control
475-- >>> S.toList $ S.take 5 $ sessionsOf 1 FL.sum $ constRate 2 $ S.enumerateFrom 1
478-- | Group the input stream into windows of @n@ second each and then fold each
479-- group using the provided fold function.
482{-# INLINE sessionsOf #-}
484 :: (IsStream t, MonadAsync m)
485 => Double -> Fold m a b -> t m a -> t m b
487 splitSuffixBy' isNothing (lcatMaybes f)
488 (S.intersperseByTime n (return Nothing) (S.map Just xs))
490------------------------------------------------------------------------------
492------------------------------------------------------------------------------
494------------------------------------------------------------------------------
496------------------------------------------------------------------------------
498-- | Break the input stream into two groups, the first group takes the input as
499-- long as the predicate applied to the first element of the stream and next
500-- input element holds 'True', the second group takes the rest of the input.
508_spanBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
509 Fold step init extract
512 init = Tuple3' <$> initialL <*> initialR <*> return (Tuple' Nothing True)
514 step (Tuple3' a b (Tuple' (Just frst) isFirstG)) input =
515 if cmp frst input && isFirstG
517 >>= (\a' -> return (Tuple3' a' b (Tuple' (Just frst) isFirstG)))
519 >>= (\a' -> return (Tuple3' a a' (Tuple' Nothing False)))
521 step (Tuple3' a b (Tuple' Nothing isFirstG)) input =
524 >>= (\a' -> return (Tuple3' a' b (Tuple' (Just input) isFirstG)))
526 >>= (\a' -> return (Tuple3' a a' (Tuple' Nothing False)))
528 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
530-- | Span as long as the predicate is 'True'. @span p f1 f2@ composes folds
531-- @f1@ and @f2@ such that the composed fold continues sending the input to
532-- @f1@ as long as the predicate @p@ is 'True'. The rest of the input is sent
535-- > let span_ p xs = FL.foldl' (FL.span p FL.toList FL.toList) $ S.fromList xs
537-- >>> span_ (< 1) [1,2,3]
540-- >>> span_ (< 2) [1,2,3]
543-- >>> span_ (< 4) [1,2,3]
546-- This can be considered as a two-fold version of 'ltakeWhile' where we take
547-- both the segments instead of discarding the leftover.
557span p (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
558 Fold step init extract
561 init = Tuple3' <$> initialL <*> initialR <*> return True
563 step (Tuple3' a b isFirstG) input =
564 if isFirstG && p input
565 then stepL a input >>= (\a' -> return (Tuple3' a' b True))
566 else stepR b input >>= (\a' -> return (Tuple3' a a' False))
568 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
571-- > break p = span (not . p)
573-- Break as soon as the predicate becomes 'True'. @break p f1 f2@ composes
574-- folds @f1@ and @f2@ such that @f1@ stops receiving input as soon as the
575-- predicate @p@ becomes 'True'. The rest of the input is sent to @f2@.
577-- This is the binary version of 'splitBy'.
579-- > let break_ p xs = FL.foldl' (FL.break p FL.toList FL.toList) $ S.fromList xs
581-- >>> break_ (< 1) [3,2,1]
584-- >>> break_ (< 2) [3,2,1]
587-- >>> break_ (< 4) [3,2,1]
598break p = span (not . p)
600-- | Like 'spanBy' but applies the predicate in a rolling fashion i.e.
601-- predicate is applied to the previous and the next input elements.
602{-# INLINE _spanRollingBy #-}
609_spanRollingBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
610 Fold step init extract
613 init = Tuple3' <$> initialL <*> initialR <*> return Nothing
615 step (Tuple3' a b (Just frst)) input =
617 then stepL a input >>= (\a' -> return (Tuple3' a' b (Just input)))
618 else stepR b input >>= (\b' -> return (Tuple3' a b' (Just input)))
620 step (Tuple3' a b Nothing) input =
621 stepL a input >>= (\a' -> return (Tuple3' a' b (Just input)))
623 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
625------------------------------------------------------------------------------
627------------------------------------------------------------------------------
629-- | @groupsBy cmp f $ S.fromList [a,b,c,...]@ assigns the element @a@ to the
630-- first group, if @a \`cmp` b@ is 'True' then @b@ is also assigned to the same
631-- group. If @a \`cmp` c@ is 'True' then @c@ is also assigned to the same
632-- group and so on. When the comparison fails a new group is started. Each
633-- group is folded using the fold @f@.
635-- >>> S.toList $ FL.groupsBy (>) FL.toList $ S.fromList [1,3,7,0,2,5]
636-- > [[1,3,7],[0,2,5]]
639{-# INLINE groupsBy #-}
641 :: (IsStream t, Monad m)
646groupsBy cmp f m = D.fromStreamD $ D.groupsBy cmp f (D.toStreamD m)
648-- | Unlike @groupsBy@ this function performs a rolling comparison of two
649-- successive elements in the input stream. @groupsRollingBy cmp f $ S.fromList
650-- [a,b,c,...]@ assigns the element @a@ to the first group, if @a \`cmp` b@ is
651-- 'True' then @b@ is also assigned to the same group. If @b \`cmp` c@ is
652-- 'True' then @c@ is also assigned to the same group and so on. When the
653-- comparison fails a new group is started. Each group is folded using the fold
656-- >>> S.toList $ FL.groupsRollingBy (\a b -> a + 1 == b) FL.toList $ S.fromList [1,2,3,7,8,9]
657-- > [[1,2,3],[7,8,9]]
660{-# INLINE groupsRollingBy #-}
662 :: (IsStream t, Monad m)
667groupsRollingBy cmp f m = D.fromStreamD $ D.groupsRollingBy cmp f (D.toStreamD m)
670-- > groups = groupsBy (==)
671-- > groups = groupsRollingBy (==)
673-- Groups a contiguous span of equal elements together in one group.
675-- >>> S.toList $ FL.groups FL.toList $ S.fromList [1,1,2,2]
679groups :: (IsStream t, Monad m, Eq a) => Fold m a b -> t m a -> t m b
680groups = groupsBy (==)
682------------------------------------------------------------------------------
683-- Binary splitting on a separator
684------------------------------------------------------------------------------
687-- | Find the first occurrence of the specified sequence in the input stream
688-- and break the input stream into two parts, the first part consisting of the
689-- stream before the sequence and the second part consisting of the sequence
690-- and the rest of the stream.
692-- > let breakOn_ pat xs = FL.foldl' (FL.breakOn pat FL.toList FL.toList) $ S.fromList xs
694-- >>> breakOn_ "dear" "Hello dear world!"
695-- > ("Hello ","dear world!")
697{-# INLINE breakOn #-}
698breakOn :: Monad m => Array a -> Fold m a b -> Fold m a c -> Fold m a (b,c)
699breakOn pat f m = undefined
702------------------------------------------------------------------------------
703-- N-ary split on a predicate
704------------------------------------------------------------------------------
706-- TODO: Use a Splitter configuration similar to the "split" package to make it
707-- possible to express all splitting combinations. In general, we can have
708-- infix/suffix/prefix/condensing of separators, dropping both leading/trailing
709-- separators. We can have a single split operation taking the splitter config
712-- | Split a stream on separator elements determined by a predicate, dropping
713-- the separator. Separators are not considered part of stream segments on
714-- either side of it instead they are treated as infixed between two stream
715-- segments. For example, with @.@ as separator, @"a.b.c"@ would be parsed as
716-- @["a","b","c"]@. When @.@ is in leading or trailing position it is still
717-- considered as infixed, treating the first or the last segment as empty. For
718-- example, @".a."@ would be parsed as @["","a",""]@. This operation is
719-- opposite of 'intercalate'.
721-- Let's use the following definition for illustration:
723-- > splitBy_ p xs = S.toList $ FL.splitBy p (FL.toList) (S.fromList xs)
725-- >>> splitBy_ (== '.') ""
728-- >>> splitBy_ (== '.') "."
731-- >>> splitBy_ (== '.') ".a"
734-- >>> splitBy_ (== '.') "a."
737-- >>> splitBy_ (== '.') "a.b"
740-- >>> splitBy_ (== '.') "a..b"
743-- This can be considered as an n-fold version of 'break' where we apply
744-- 'break' successively on the input stream, dropping the first element
745-- of the second segment after each break.
748{-# INLINE splitBy #-}
750 :: (IsStream t, Monad m)
751 => (a -> Bool) -> Fold m a b -> t m a -> t m b
752splitBy predicate f m =
753 D.fromStreamD $ D.splitBy predicate f (D.toStreamD m)
755-- | Like 'splitBy' but the separator is treated as part of the previous
756-- stream segment (suffix). Therefore, when the separator is in trailing
757-- position, no empty segment is considered to follow it. For example, @"a.b."@
758-- would be parsed as @["a","b"]@ instead of @["a","b",""]@ as in the case of
761-- > splitSuffixBy_ p xs = S.toList $ FL.splitSuffixBy p (FL.toList) (S.fromList xs)
763-- >>> splitSuffixBy_ (== '.') ""
766-- >>> splitSuffixBy_ (== '.') "."
769-- >>> splitSuffixBy_ (== '.') "a"
772-- >>> splitSuffixBy_ (== '.') ".a"
775-- >>> splitSuffixBy_ (== '.') "a."
778-- >>> splitSuffixBy_ (== '.') "a.b"
781-- >>> splitSuffixBy_ (== '.') "a.b."
784-- >>> splitSuffixBy_ (== '.') "a..b.."
787-- > lines = splitSuffixBy (== '\n')
789-- This can be considered as an n-fold version of 'breakPost' where we apply
790-- 'breakPost' successively on the input stream, dropping the first element
791-- of the second segment after each break.
794{-# INLINE splitSuffixBy #-}
796 :: (IsStream t, Monad m)
797 => (a -> Bool) -> Fold m a b -> t m a -> t m b
798splitSuffixBy predicate f m =
799 D.fromStreamD $ D.splitSuffixBy predicate f (D.toStreamD m)
801-- | Like 'splitBy' but ignores repeated separators or separators in leading
802-- or trailing position. Therefore, @"..a..b.."@ would be parsed as
803-- @["a","b"]@. In other words, it treats the input like words separated by
804-- whitespace elements determined by the predicate.
806-- > wordsBy' p xs = S.toList $ FL.wordsBy p (FL.toList) (S.fromList xs)
808-- >>> wordsBy' (== ',') ""
811-- >>> wordsBy' (== ',') ","
814-- >>> wordsBy' (== ',') ",a,,b,"
817-- > words = wordsBy isSpace
820{-# INLINE wordsBy #-}
822 :: (IsStream t, Monad m)
823 => (a -> Bool) -> Fold m a b -> t m a -> t m b
824wordsBy predicate f m =
825 D.fromStreamD $ D.wordsBy predicate f (D.toStreamD m)
827-- XXX we should express this using the Splitter config.
829-- We can get splitSuffixBy' by appending the suffix to the output segments
830-- produced by splitSuffixBy. However, it may add an additional suffix if the last
831-- fragment did not have a suffix in the first place.
833-- | Like 'splitSuffixBy' but keeps the suffix in the splits.
835-- > splitSuffixBy'_ p xs = S.toList $ FL.splitSuffixBy' p (FL.toList) (S.fromList xs)
837-- >>> splitSuffixBy'_ (== '.') ""
840-- >>> splitSuffixBy'_ (== '.') "."
843-- >>> splitSuffixBy'_ (== '.') "a"
846-- >>> splitSuffixBy'_ (== '.') ".a"
849-- >>> splitSuffixBy'_ (== '.') "a."
852-- >>> splitSuffixBy'_ (== '.') "a.b"
855-- >>> splitSuffixBy'_ (== '.') "a.b."
858-- >>> splitSuffixBy'_ (== '.') "a..b.."
859-- > ["a.",".","b.","."]
861-- This can be considered as an n-fold version of 'breakPost' where we apply
862-- 'breakPost' successively on the input stream.
865{-# INLINE splitSuffixBy' #-}
867 :: (IsStream t, Monad m)
868 => (a -> Bool) -> Fold m a b -> t m a -> t m b
869splitSuffixBy' predicate f m =
870 D.fromStreamD $ D.splitSuffixBy' predicate f (D.toStreamD m)
872------------------------------------------------------------------------------
873-- Split on a delimiter sequence
874------------------------------------------------------------------------------
876-- Int list examples for splitOn:
878-- >>> splitList [] [1,2,3,3,4]
879-- > [[1],[2],[3],[3],[4]]
881-- >>> splitList [5] [1,2,3,3,4]
884-- >>> splitList [1] [1,2,3,3,4]
887-- >>> splitList [4] [1,2,3,3,4]
890-- >>> splitList [2] [1,2,3,3,4]
893-- >>> splitList [3] [1,2,3,3,4]
896-- >>> splitList [3,3] [1,2,3,3,4]
899-- >>> splitList [1,2,3,3,4] [1,2,3,3,4]
902-- | Split the stream on both sides of a separator sequence, dropping the
905-- For illustration, let's define a function that operates on pure lists:
908-- splitOn_ pat xs = S.toList $ FL.splitOn (A.fromList pat) (FL.toList) (S.fromList xs)
911-- >>> splitOn_ "" "hello"
912-- > ["h","e","l","l","o"]
914-- >>> splitOn_ "hello" ""
917-- >>> splitOn_ "hello" "hello"
920-- >>> splitOn_ "x" "hello"
923-- >>> splitOn_ "h" "hello"
926-- >>> splitOn_ "o" "hello"
929-- >>> splitOn_ "e" "hello"
932-- >>> splitOn_ "l" "hello"
935-- >>> splitOn_ "ll" "hello"
938-- 'splitOn' is an inverse of 'intercalate'. The following law always holds:
940-- > intercalate . splitOn == id
942-- The following law holds when the separator is non-empty and contains none of
943-- the elements present in the input lists:
945-- > splitOn . intercalate == id
947-- The following law always holds:
949-- > concat . splitOn . intercalate == concat
952{-# INLINE splitOn #-}
954 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
955 => Array a -> Fold m a b -> t m a -> t m b
956splitOn patt f m = D.fromStreamD $ D.splitOn patt f (D.toStreamD m)
959-- This can be implemented easily using Rabin Karp
960-- | Split on any one of the given patterns.
961{-# INLINE splitOnAny #-}
963 :: (IsStream t, Monad m, Storable a, Integral a)
964 => [Array a] -> Fold m a b -> t m a -> t m b
965splitOnAny subseq f m = undefined -- D.fromStreamD $ D.splitOnAny f subseq (D.toStreamD m)
968-- | Like 'splitSuffixBy' but the separator is a sequence of elements, instead
969-- of a predicate for a single element.
971-- > splitSuffixOn_ pat xs = S.toList $ FL.splitSuffixOn (A.fromList pat) (FL.toList) (S.fromList xs)
973-- >>> splitSuffixOn_ "." ""
976-- >>> splitSuffixOn_ "." "."
979-- >>> splitSuffixOn_ "." "a"
982-- >>> splitSuffixOn_ "." ".a"
985-- >>> splitSuffixOn_ "." "a."
988-- >>> splitSuffixOn_ "." "a.b"
991-- >>> splitSuffixOn_ "." "a.b."
994-- >>> splitSuffixOn_ "." "a..b.."
997-- > lines = splitSuffixOn "\n"
1000{-# INLINE splitSuffixOn #-}
1002 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
1003 => Array a -> Fold m a b -> t m a -> t m b
1004splitSuffixOn patt f m =
1005 D.fromStreamD $ D.splitSuffixOn False patt f (D.toStreamD m)
1008-- | Like 'splitOn' but drops any empty splits.
1010{-# INLINE wordsOn #-}
1012 :: (IsStream t, Monad m, Storable a, Eq a)
1013 => Array a -> Fold m a b -> t m a -> t m b
1014wordsOn subseq f m = undefined -- D.fromStreamD $ D.wordsOn f subseq (D.toStreamD m)
1017-- XXX use a non-monadic intersperse to remove the MonadAsync constraint.
1019-- | Like 'splitOn' but splits the separator as well, as an infix token.
1021-- > splitOn'_ pat xs = S.toList $ FL.splitOn' (A.fromList pat) (FL.toList) (S.fromList xs)
1023-- >>> splitOn'_ "" "hello"
1024-- > ["h","","e","","l","","l","","o"]
1026-- >>> splitOn'_ "hello" ""
1029-- >>> splitOn'_ "hello" "hello"
1032-- >>> splitOn'_ "x" "hello"
1035-- >>> splitOn'_ "h" "hello"
1038-- >>> splitOn'_ "o" "hello"
1041-- >>> splitOn'_ "e" "hello"
1044-- >>> splitOn'_ "l" "hello"
1045-- > ["he","l","","l","o"]
1047-- >>> splitOn'_ "ll" "hello"
1051{-# INLINE splitOn' #-}
1053 :: (IsStream t, MonadAsync m, Storable a, Enum a, Eq a)
1054 => Array a -> Fold m a b -> t m a -> t m b
1055splitOn' patt f m = S.intersperseM (foldl' f (A.read patt)) $ splitOn patt f m
1057-- | Like 'splitSuffixOn' but keeps the suffix intact in the splits.
1059-- > splitSuffixOn'_ pat xs = S.toList $ FL.splitSuffixOn' (A.fromList pat) (FL.toList) (S.fromList xs)
1061-- >>> splitSuffixOn'_ "." ""
1064-- >>> splitSuffixOn'_ "." "."
1067-- >>> splitSuffixOn'_ "." "a"
1070-- >>> splitSuffixOn'_ "." ".a"
1073-- >>> splitSuffixOn'_ "." "a."
1076-- >>> splitSuffixOn'_ "." "a.b"
1079-- >>> splitSuffixOn'_ "." "a.b."
1082-- >>> splitSuffixOn'_ "." "a..b.."
1083-- > ["a.",".","b.","."]
1086{-# INLINE splitSuffixOn' #-}
1088 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
1089 => Array a -> Fold m a b -> t m a -> t m b
1090splitSuffixOn' patt f m =
1091 D.fromStreamD $ D.splitSuffixOn True patt f (D.toStreamD m)
1094-- This can be implemented easily using Rabin Karp
1095-- | Split post any one of the given patterns.
1096{-# INLINE splitSuffixOnAny #-}
1098 :: (IsStream t, Monad m, Storable a, Integral a)
1099 => [Array a] -> Fold m a b -> t m a -> t m b
1100splitSuffixOnAny subseq f m = undefined
1101 -- D.fromStreamD $ D.splitPostAny f subseq (D.toStreamD m)
1104------------------------------------------------------------------------------
1105-- Reorder in sequence
1106------------------------------------------------------------------------------
1109-- Buffer until the next element in sequence arrives. The function argument
1110-- determines the difference in sequence numbers. This could be useful in
1111-- implementing sequenced streams, for example, TCP reassembly.
1112{-# INLINE reassembleBy #-}
1114 :: (IsStream t, Monad m)
1119reassembleBy = undefined
1122------------------------------------------------------------------------------
1124------------------------------------------------------------------------------
1126-- | Distribute one copy of the stream to each fold and zip the results.
1129-- |-------Fold m a b--------|
1130-- ---stream m a---| |---m (b,c)
1131-- |-------Fold m a c--------|
1133-- >>> FL.foldl' (FL.tee FL.sum FL.length) (S.enumerateFromTo 1.0 100.0)
1138tee :: Monad m => Fold m a b -> Fold m a c -> Fold m a (b,c)
1139tee f1 f2 = (,) <$> f1 <*> f2
1141{-# INLINE foldNil #-}
1142foldNil :: Monad m => Fold m a [b]
1143foldNil = Fold step begin done where
1145 step _ _ = return []
1148{-# INLINE foldCons #-}
1149foldCons :: Monad m => Fold m a b -> Fold m a [b] -> Fold m a [b]
1150foldCons (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1151 Fold step begin done
1155 begin = Tuple' <$> beginL <*> beginR
1156 step (Tuple' xL xR) a = Tuple' <$> stepL xL a <*> stepR xR a
1157 done (Tuple' xL xR) = (:) <$> (doneL xL) <*> (doneR xR)
1159-- XXX use "List" instead of "[]"?, use Array for output to scale it to a large
1160-- number of consumers?
1162-- | Distribute one copy of the stream to each fold and collect the results in
1167-- |-------Fold m a b--------|
1168-- ---stream m a---| |---m [b]
1169-- |-------Fold m a b--------|
1174-- >>> FL.foldl' (FL.distribute [FL.sum, FL.length]) (S.enumerateFromTo 1 5)
1177-- This is the consumer side dual of the producer side 'sequence' operation.
1180{-# INLINE distribute #-}
1181distribute :: Monad m => [Fold m a b] -> Fold m a [b]
1182distribute [] = foldNil
1183distribute (x:xs) = foldCons x (distribute xs)
1185------------------------------------------------------------------------------
1187------------------------------------------------------------------------------
1189-- | Partition the input over two folds using an 'Either' partitioning
1194-- |-------Fold b x--------|
1195-- -----stream m a --> (Either b c)----| |----(x,y)
1196-- |-------Fold c y--------|
1199-- Send input to either fold randomly:
1201-- >>> import System.Random (randomIO)
1202-- >>> randomly a = randomIO >>= \x -> return $ if x then Left a else Right a
1203-- >>> FL.foldl' (FL.partitionByM randomly FL.length FL.length) (S.enumerateFromTo 1 100)
1206-- Send input to the two folds in a proportion of 2:1:
1209-- import Data.IORef (newIORef, readIORef, writeIORef)
1210-- proportionately m n = do
1211-- ref <- newIORef $ cycle $ concat [replicate m Left, replicate n Right]
1212-- return $ \\a -> do
1213-- r <- readIORef ref
1214-- writeIORef ref $ tail r
1218-- f <- proportionately 2 1
1219-- r <- FL.foldl' (FL.partitionByM f FL.length FL.length) (S.enumerateFromTo (1 :: Int) 100)
1226-- This is the consumer side dual of the producer side 'mergeBy' operation.
1229{-# INLINE partitionByM #-}
1230partitionByM :: Monad m
1231 => (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1232partitionByM f (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1234 Fold step begin done
1238 begin = Tuple' <$> beginL <*> beginR
1239 step (Tuple' xL xR) a = do
1242 Left b -> Tuple' <$> stepL xL b <*> return xR
1243 Right c -> Tuple' <$> return xL <*> stepR xR c
1244 done (Tuple' xL xR) = (,) <$> doneL xL <*> doneR xR
1246-- Note: we could use (a -> Bool) instead of (a -> Either b c), but the latter
1247-- makes the signature clearer as to which case belongs to which fold.
1248-- XXX need to check the performance in both cases.
1250-- | Same as 'partitionByM' but with a pure partition function.
1252-- Count even and odd numbers in a stream:
1255-- >>> let f = FL.partitionBy (\\n -> if even n then Left n else Right n)
1256-- (fmap (("Even " ++) . show) FL.length)
1257-- (fmap (("Odd " ++) . show) FL.length)
1258-- in FL.foldl' f (S.enumerateFromTo 1 100)
1259-- ("Even 50","Odd 50")
1263{-# INLINE partitionBy #-}
1264partitionBy :: Monad m
1265 => (a -> Either b c) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1266partitionBy f = partitionByM (return . f)
1268-- | Compose two folds such that the combined fold accepts a stream of 'Either'
1269-- and routes the 'Left' values to the first fold and 'Right' values to the
1272-- > partition = partitionBy id
1275{-# INLINE partition #-}
1277 => Fold m b x -> Fold m c y -> Fold m (Either b c) (x, y)
1278partition = partitionBy id
1281-- | Send one item to each fold in a round-robin fashion. This is the consumer
1282-- side dual of producer side 'mergeN' operation.
1284-- partitionN :: Monad m => [Fold m a b] -> Fold m a [b]
1285-- partitionN fs = Fold step begin done
1288-- Demultiplex an input element into a number of typed variants. We want to
1289-- statically restrict the target values within a set of predefined types, an
1290-- enumeration of a GADT. We also want to make sure that the Map contains only
1291-- those types and the full set of those types. Instead of Map it should
1292-- probably be a lookup-table using an array and not in GC memory.
1294-- This is the consumer side dual of the producer side 'mux' operation (XXX to
1297-- | Split the input stream based on a key field and fold each split using a
1298-- specific fold collecting the results in a map from the keys to the results.
1299-- Useful for cases like protocol handlers to handle different type of packets
1300-- using different handlers.
1304-- |-------Fold m a b
1305-- -----stream m a-----Map-----|
1306-- |-------Fold m a b
1312{-# INLINE demuxWith #-}
1313demuxWith :: (Monad m, Ord k)
1314 => (a -> k) -> Map k (Fold m a b) -> Fold m a (Map k b)
1315demuxWith f kv = Fold step initial extract
1321 -- XXX should we raise an exception in Nothing case?
1322 -- Ideally we should enforce that it is a total map over k so that look
1324 -- XXX we could use a monadic update function for a single lookup and
1325 -- update in the map.
1327 in case Map.lookup k mp of
1328 Nothing -> return mp
1329 Just (Fold step' acc extract') -> do
1330 !r <- acc >>= \x -> step' x a
1331 return $ Map.insert k (Fold step' (return r) extract') mp
1332 extract = Prelude.mapM (\(Fold _ acc e) -> acc >>= e)
1334-- | Fold a stream of key value pairs using a map of specific folds for each
1335-- key into a map from keys to the results of fold outputs of the corresponding
1339-- > let table = Data.Map.fromList [(\"SUM", FL.sum), (\"PRODUCT", FL.product)]
1340-- input = S.fromList [(\"SUM",1),(\"PRODUCT",2),(\"SUM",3),(\"PRODUCT",4)]
1341-- in FL.foldl' (FL.demux table) input
1348demux :: (Monad m, Ord k)
1349 => Map k (Fold m a b) -> Fold m (k, a) (Map k b)
1350demux fs = demuxWith fst (Map.map (lmap snd) fs)
1352-- | Split the input stream based on a key field and fold each split using a
1353-- specific fold without collecting the results. Useful for cases like protocol
1354-- handlers to handle different type of packets.
1358-- |-------Fold m a ()
1359-- -----stream m a-----Map-----|
1360-- |-------Fold m a ()
1368-- demuxWith_ can be slightly faster than demuxWith because we do not need to
1369-- update the Map in this case. This may be significant only if the map is
1371{-# INLINE demuxWith_ #-}
1372demuxWith_ :: (Monad m, Ord k)
1373 => (a -> k) -> Map k (Fold m a b) -> Fold m a ()
1374demuxWith_ f kv = Fold step initial extract
1379 Prelude.mapM (\(Fold s i e) ->
1380 i >>= \r -> return (Fold s (return r) e)) kv
1382 -- XXX should we raise an exception in Nothing case?
1383 -- Ideally we should enforce that it is a total map over k so that look
1385 case Map.lookup (f a) mp of
1386 Nothing -> return mp
1387 Just (Fold step' acc _) -> do
1388 _ <- acc >>= \x -> step' x a
1390 extract mp = Prelude.mapM (\(Fold _ acc e) -> acc >>= e) mp >> return ()
1392-- | Given a stream of key value pairs and a map from keys to folds, fold the
1393-- values for each key using the corresponding folds, discarding the outputs.
1396-- > let prn = FL.drainBy print
1397-- > let table = Data.Map.fromList [(\"ONE", prn), (\"TWO", prn)]
1398-- input = S.fromList [(\"ONE",1),(\"TWO",2)]
1399-- in FL.foldl' (FL.demux_ table) input
1405{-# INLINE demux_ #-}
1406demux_ :: (Monad m, Ord k) => Map k (Fold m a ()) -> Fold m (k, a) ()
1407demux_ fs = demuxWith_ fst (Map.map (lmap snd) fs)
1409-- XXX instead of a Map we could yield the results as a pure stream as they
1410-- complete. We could then concatMap the fold results to implement the
1411-- windowing combinators.
1413-- | Split the input stream based on a key field and fold each split using the
1414-- given fold. Useful for map/reduce, bucketizing the input in different bins
1415-- or for generating histograms.
1418-- > let input = S.fromList [(\"ONE",1),(\"ONE",1.1),(\"TWO",2), (\"TWO",2.2)]
1419-- in FL.foldl' (FL.classify FL.toListRev) input
1420-- fromList [(\"ONE",[1.1,1.0]),(\"TWO",[2.2,2.0])]
1424{-# INLINE classifyWith #-}
1425classifyWith :: (Monad m, Ord k) => (a -> k) -> Fold m a b -> Fold m a (Map k b)
1426classifyWith f (Fold step initial extract) = Fold step' initial' extract'
1430 initial' = return Map.empty
1433 in case Map.lookup k kv of
1437 return $ Map.insert k r kv
1440 return $ Map.insert k r kv
1441 extract' = Prelude.mapM extract
1443-- | Given an input stream of key value pairs and a fold for values, fold all
1444-- the values belonging to each key. Useful for map/reduce, bucketizing the
1445-- input in different bins or for generating histograms.
1448-- > let input = S.fromList [(\"ONE",1),(\"ONE",1.1),(\"TWO",2), (\"TWO",2.2)]
1449-- in FL.foldl' (FL.classify FL.toListRev) input
1450-- fromList [(\"ONE",[1.1,1.0]),(\"TWO",[2.2,2.0])]
1455-- > classify fld = classifyWith fst (lmap snd fld)
1458{-# INLINE classify #-}
1459classify :: (Monad m, Ord k) => Fold m a b -> Fold m (k, a) (Map k b)
1460classify fld = classifyWith fst (lmap snd fld)
1462------------------------------------------------------------------------------
1464------------------------------------------------------------------------------
1466-- | Like 'unzipWith' but with a monadic splitter function.
1469{-# INLINE unzipWithM #-}
1470unzipWithM :: Monad m
1471 => (a -> m (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
1472unzipWithM f (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1473 Fold step begin done
1477 step (Tuple' xL xR) a = do
1479 Tuple' <$> stepL xL b <*> stepR xR c
1480 begin = Tuple' <$> beginL <*> beginR
1481 done (Tuple' xL xR) = (,) <$> doneL xL <*> doneR xR
1483-- | Split elements in the input stream into two parts using a pure splitter
1484-- function, direct each part to a different fold and zip the results.
1487{-# INLINE unzipWith #-}
1489 => (a -> (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
1490unzipWith f = unzipWithM (return . f)
1492-- | Send the elements of tuples in a stream of tuples through two different
1497-- |-------Fold a x--------|
1498-- -----Stream m x----(a,b)--| |----m (x,y)
1499-- |-------Fold b y--------|
1503-- This is the consumer side dual of the producer side 'zip' operation.
1507unzip :: Monad m => Fold m a x -> Fold m b y -> Fold m (a,b) (x,y)
1510------------------------------------------------------------------------------
1512------------------------------------------------------------------------------
1514-- | Modify the fold such that when the fold is done, instead of returning the
1515-- accumulator, it returns a fold. The returned fold starts from where we left
1516-- i.e. it uses the last accumulator value as the initial value of the
1517-- accumulator. Thus we can resume the fold later and feed it more input.
1520-- > more <- FL.foldl (FL.duplicate FL.sum) (S.enumerateFromTo 1 10)
1521-- > evenMore <- FL.foldl (FL.duplicate more) (S.enumerateFromTo 11 20)
1522-- > FL.foldl evenMore (S.enumerateFromTo 21 30)
1526{-# INLINABLE duplicate #-}
1527duplicate :: Applicative m => Fold m a b -> Fold m a (Fold m a b)
1528duplicate (Fold step begin done) =
1529 Fold step begin (\x -> pure (Fold step (pure x) done))
1532-- All the stream flattening transformations can also be applied to a fold
1535-- | This can be used to apply all the stream generation operations on folds.
1536lconcatMap ::(IsStream t, Monad m) => (a -> t m b)
1539lconcatMap s f1 f2 = undefined
1542-- All the grouping transformation that we apply to a stream can also be
1543-- applied to a fold input stream.
1546-- | Group the input stream into groups of elements between @low@ and @high@.
1547-- Collection starts in chunks of @low@ and then keeps doubling until we reach
1548-- @high@. Each chunk is folded using the provided fold function.
1550-- This could be useful, for example, when we are folding a stream of unknown
1551-- size to a stream of arrays and we want to minimize the number of
1556-- XXX we should be able to implement it with parsers/terminating folds.
1558{-# INLINE lchunksInRange #-}
1559lchunksInRange :: Monad m
1560 => Int -> Int -> Fold m a b -> Fold m b c -> Fold m a c
1561lchunksInRange low high (Fold step1 initial1 extract1)
1562 (Fold step2 initial2 extract2) = undefined
1565-- | Group the input stream into groups of @n@ elements each and then fold each
1566-- group using the provided fold function.
1570-- -----Fold m a b----|-Fold n a c-|-Fold n a c-|-...-|----Fold m a c
1574{-# INLINE lchunksOf #-}
1575lchunksOf :: Monad m => Int -> Fold m a b -> Fold m b c -> Fold m a c
1576lchunksOf n (Fold step1 initial1 extract1) (Fold step2 initial2 extract2) =
1577 Fold step' initial' extract'
1581 initial' = (Tuple3' 0) <$> initial1 <*> initial2
1582 step' (Tuple3' i r1 r2) a = do
1586 return $ Tuple3' (i + 1) res r2
1589 acc2 <- step2 r2 res
1593 return $ Tuple3' 1 acc1 acc2
1594 extract' (Tuple3' _ _ r) = extract2 r
1596-- | Group the input stream into windows of n second each and then fold each
1597-- group using the provided fold function.
1599-- For example, we can copy and distribute a stream to multiple folds where
1600-- each fold can group the input differently e.g. by one second, one minute and
1601-- one hour windows respectively and fold each resulting stream of folds.
1605-- -----Fold m a b----|-Fold n a c-|-Fold n a c-|-...-|----Fold m a c
1608{-# INLINE lsessionsOf #-}
1609lsessionsOf :: MonadAsync m => Double -> Fold m a b -> Fold m b c -> Fold m a c
1610lsessionsOf n (Fold step1 initial1 extract1) (Fold step2 initial2 extract2) =
1611 Fold step' initial' extract'
1615 -- XXX MVar may be expensive we need a cheaper synch mechanism here
1619 mv1 <- liftIO $ newMVar i1
1620 mv2 <- liftIO $ newMVar (Right i2)
1621 t <- control $ \run ->
1622 mask $ \restore -> do
1623 tid <- forkIO $ catch (restore $ void $ run (timerThread mv1 mv2))
1624 (handleChildException mv2)
1626 return $ Tuple3' t mv1 mv2
1627 step' acc@(Tuple3' _ mv1 _) a = do
1628 r1 <- liftIO $ takeMVar mv1
1630 liftIO $ putMVar mv1 res
1632 extract' (Tuple3' tid _ mv2) = do
1633 r2 <- liftIO $ takeMVar mv2
1634 liftIO $ killThread tid
1637 Right x -> extract2 x
1639 timerThread mv1 mv2 = do
1640 liftIO $ threadDelay (round $ n * 1000000)
1642 r1 <- liftIO $ takeMVar mv1
1644 liftIO $ putMVar mv1 i1
1647 r2 <- liftIO $ takeMVar mv2
1650 Right x -> fmap Right $ step2 x res1
1651 liftIO $ putMVar mv2 res
1654 handleChildException ::
1655 MVar (Either SomeException a) -> SomeException -> IO ()
1656 handleChildException mv2 e = do
1663------------------------------------------------------------------------------
1664-- Windowed classification
1665------------------------------------------------------------------------------
1667-- We divide the stream into windows or chunks in space or time and each window
1668-- can be associated with a key, all events associated with a particular key in
1669-- the window can be folded to a single result. The stream can be split into
1670-- windows by size or by using a split predicate on the elements in the stream.
1671-- For example, when we receive a closing flag, we can close the window.
1673-- A "chunk" is a space window and a "session" is a time window. Are there any
1674-- other better short words to describe them. An alternative is to use
1675-- "swindow" and "twindow". Another word for "session" could be "spell".
1677-- TODO: To mark the position in space or time we can have Indexed or
1678-- TimeStamped types. That can make it easy to deal with the position indices
1681------------------------------------------------------------------------------
1682-- Keyed Sliding Windows
1683------------------------------------------------------------------------------
1686{-# INLINABLE classifySlidingChunks #-}
1687classifySlidingChunks
1688 :: (IsStream t, MonadAsync m, Ord k)
1689 => Int -- ^ window size
1690 -> Int -- ^ window slide
1691 -> Fold m a b -- ^ Fold to be applied to window events
1692 -> t m (k, a, Bool) -- ^ window key, data, close event
1694classifySlidingChunks wsize wslide (Fold step initial extract) str
1697-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
1698-- slide the send window when an ack is received and we slide the receive
1699-- window when a sequence is complete. Sliding is stateful in case of TCP,
1700-- sliding releases the send buffer or makes data available to the user from
1701-- the receive buffer.
1702{-# INLINABLE classifySlidingSessions #-}
1703classifySlidingSessions
1704 :: (IsStream t, MonadAsync m, Ord k)
1705 => Double -- ^ timer tick in seconds
1706 -> Double -- ^ time window size
1707 -> Double -- ^ window slide
1708 -> Fold m a b -- ^ Fold to be applied to window events
1709 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
1711classifySlidingSessions tick interval slide (Fold step initial extract) str
1715------------------------------------------------------------------------------
1716-- Sliding Window Buffers
1717------------------------------------------------------------------------------
1719-- These buffered versions could be faster than concurrent incremental folds of
1720-- all overlapping windows as in many cases we may not need all the values to
1721-- compute the fold, we can just compute the result using the old value and new
1722-- value. However, we may need the buffer once in a while, for example for
1723-- string search we usually compute the hash incrementally but when the hash
1724-- matches the hash of the pattern we need to compare the whole string.
1726-- XXX we should be able to implement sequence based splitting combinators
1727-- using this combinator.
1730-- | Buffer n elements of the input in a ring buffer. When t new elements are
1731-- collected, slide the window to remove the same number of oldest elements,
1732-- insert the new elements, and apply an incremental fold on the sliding
1733-- window, supplying the outgoing elements, the new ring buffer as arguments.
1735 :: (IsStream t, Monad m, Ord a, Storable a)
1736 => Int -- window size
1737 -> Int -- window slide
1738 -> Fold m (Ring a, Array a) b
1741slidingChunkBuffer = undefined
1743-- Buffer n seconds worth of stream elements of the input in a radix tree.
1744-- Every t seconds, remove the items that are older than n seconds, and apply
1745-- an incremental fold on the sliding window, supplying the outgoing elements,
1746-- and the new radix tree buffer as arguments.
1748 :: (IsStream t, Monad m, Ord a, Storable a)
1749 => Int -- window size
1751 -> Fold m (RTree a, Array a) b
1754slidingSessionBuffer = undefined
1757------------------------------------------------------------------------------
1758-- Keyed Session Windows
1759------------------------------------------------------------------------------
1762-- | Keyed variable size space windows. Close the window if we do not receive a
1763-- window event in the next "spaceout" elements.
1764{-# INLINABLE classifyChunksBy #-}
1766 :: (IsStream t, MonadAsync m, Ord k)
1767 => Int -- ^ window spaceout (spread)
1768 -> Bool -- ^ reset the spaceout when a chunk window element is received
1769 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1770 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1772classifyChunksBy spanout reset (Fold step initial extract) str = undefined
1774-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
1775-- received within the chunk size window. The chunk gets closed only if no
1776-- element is received within the chunk window.
1778{-# INLINABLE classifyKeepAliveChunks #-}
1779classifyKeepAliveChunks
1780 :: (IsStream t, MonadAsync m, Ord k)
1781 => Int -- ^ window spaceout (spread)
1782 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1783 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1785classifyKeepAliveChunks spanout = classifyChunksBy spanout True
1788-- | @classifySessionsBy tick timeout reset f stream@ groups together all input
1789-- stream elements that belong to the same session. @timeout@ is the maximum
1790-- lifetime of a session in seconds. All elements belonging to a session are
1791-- purged after this duration. If "reset" is 'Ture' then the timeout is reset
1792-- after every event received in the session. Session duration is measured
1793-- using the timestamp of the first element seen for that session. To detect
1794-- session timeouts, a monotonic event time clock is maintained using the
1795-- timestamps seen in the inputs and a timer with a tick duration specified by
1798-- @session key@ is a key that uniquely identifies the session for the given
1799-- element, @timestamp@ characterizes the time when the input element was
1800-- generated, this is an absolute time measured from some @Epoch@. @session
1801-- close@ is a boolean indicating whether this element marks the closing of the
1802-- session. When an input element with @session close@ set to @True@ is seen
1803-- the session is purged immediately.
1805-- All the input elements belonging to a session are collected using the fold
1806-- @f@. The session key and the fold result are emitted in the output stream
1807-- when the session is purged either via the session close event or via the
1808-- session liftime timeout.
1811{-# INLINABLE classifySessionsBy #-}
1813 :: (IsStream t, MonadAsync m, Ord k)
1814 => Double -- ^ timer tick in seconds
1815 -> Double -- ^ session timeout
1816 -> Bool -- ^ reset the timeout when an event is received
1817 -> Fold m a b -- ^ Fold to be applied to session events
1818 -> t m (k, a, Bool, AbsTime) -- ^ session key, timestamp, close event, data
1820classifySessionsBy tick timeout reset (Fold step initial extract) str =
1821 S.concatMap (\(Tuple4' _ _ _ s) -> s) $ S.scanlM' sstep szero stream
1825 timeoutMs = toRelTime (round (timeout * 1000) :: MilliSecond64)
1826 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
1827 szero = Tuple4' (toAbsTime (0 :: MilliSecond64)) H.empty Map.empty S.nil
1829 -- Got a new stream input element
1830 sstep (Tuple4' evTime hp mp _) (Just (key, a, closing, ts)) =
1831 -- XXX we should use a heap in pinned memory to scale it to a large
1834 -- deleting a key from the heap is expensive, so we never delete a
1835 -- key, we just purge it from the Map and it gets purged from the
1836 -- heap on timeout. We just need an extra lookup in the Map when
1837 -- the key is purged from the heap, that should not be expensive.
1839 -- To detect session inactivity we keep a timestamp of the latest event
1840 -- in the Map along with the fold result. When we purge the session
1841 -- from the heap we match the timestamp in the heap with the timestamp
1842 -- in the Map, if the latest timestamp is newer and has not expired we
1843 -- reinsert the key in the heap.
1845 -- XXX if the key is an Int, we can also use an IntMap for slightly
1846 -- better performance.
1848 let accumulate v = do
1849 Tuple' _ old <- maybe (initial >>= return . Tuple' ts) return v
1851 return $ Tuple' ts new
1854 let (r, mp') = Map.updateLookupWithKey (\_ _ -> Nothing) key mp
1855 Tuple' _ acc <- accumulate r
1857 return $ Tuple4' evTime hp mp' (S.yield (key, res))
1859 let r = Map.lookup key mp
1861 let mp' = Map.insert key acc mp
1865 let expiry = addToAbsTime ts timeoutMs
1866 in H.insert (Entry expiry key) hp
1868 -- Event time is maintained as monotonically increasing
1869 -- time. If we have lagged behind any of the timestamps
1870 -- seen then we increase it to match the latest time seen
1871 -- in the timestamps. We also increase it on timer ticks.
1872 return $ Tuple4' (max evTime ts) hp' mp' S.nil
1874 -- Got a timer tick event
1875 -- XXX can we yield the entries without accumulating them?
1876 sstep (Tuple4' evTime heap sessions _) Nothing = do
1877 (hp', mp', out) <- go heap sessions S.nil
1878 return $ Tuple4' curTime hp' mp' out
1882 curTime = addToAbsTime evTime tickMs
1884 let hres = H.uncons hp
1886 Just (Entry ts key, hp') -> do
1887 let duration = diffAbsTime curTime ts
1888 if duration >= timeoutMs
1890 let (r, mp') = Map.updateLookupWithKey
1891 (\_ _ -> Nothing) key mp
1893 Nothing -> go hp' mp' out
1894 Just (Tuple' latestTS acc) -> do
1895 let dur = diffAbsTime curTime latestTS
1896 if dur >= timeoutMs || not reset
1899 go hp' mp' ((key, sess) `S.cons` out)
1901 -- reset the session timeout
1902 let expiry = addToAbsTime latestTS timeoutMs
1903 hp'' = H.insert (Entry expiry key) hp'
1904 mp'' = Map.insert key (Tuple' latestTS acc) mp'
1906 else return (hp, mp, out)
1907 Nothing -> return (hp, mp, out)
1909 -- merge timer events in the stream
1910 stream = S.map Just str `parallel` S.repeatM timer
1912 liftIO $ threadDelay (round $ tick * 1000000)
1915-- | Like 'classifySessionsOf' but the session is kept alive if an event is
1916-- received within the session window. The session times out and gets closed
1917-- only if no event is received within the specified session window size.
1920{-# INLINABLE classifyKeepAliveSessions #-}
1921classifyKeepAliveSessions
1922 :: (IsStream t, MonadAsync m, Ord k)
1923 => Double -- ^ session inactive timeout
1924 -> Fold m a b -- ^ Fold to be applied to session payload data
1925 -> t m (k, a, Bool, AbsTime) -- ^ session key, data, close flag, timestamp
1927classifyKeepAliveSessions timeout = classifySessionsBy 1 timeout True
1929------------------------------------------------------------------------------
1930-- Keyed tumbling windows
1931------------------------------------------------------------------------------
1933-- Tumbling windows is a special case of sliding windows where the window slide
1934-- is the same as the window size. Or it can be a special case of session
1935-- windows where the reset flag is set to False.
1937-- XXX instead of using the early termination flag in the stream, we can use an
1938-- early terminating fold instead.
1941-- | Split the stream into fixed size chunks of specified size. Within each
1942-- such chunk fold the elements in buckets identified by the keys. A particular
1943-- bucket fold can be terminated early if a closing flag is encountered in an
1944-- element for that key.
1947{-# INLINABLE classifyChunksOf #-}
1949 :: (IsStream t, MonadAsync m, Ord k)
1950 => Int -- ^ window size
1951 -> Fold m a b -- ^ Fold to be applied to window events
1952 -> t m (k, a, Bool) -- ^ window key, data, close event
1954classifyChunksOf wsize = classifyChunksBy wsize False
1957-- | Split the stream into fixed size time windows of specified interval in
1958-- seconds. Within each such window, fold the elements in buckets identified by
1959-- the keys. A particular bucket fold can be terminated early if a closing flag
1960-- is encountered in an element for that key. Once a fold is terminated the key
1961-- and value for that bucket are emitted in the output stream.
1963-- Session @timestamp@ in the input stream is an absolute time from some epoch,
1964-- characterizing the time when the input element was generated. To detect
1965-- session window end, a monotonic event time clock is maintained synced with
1966-- the timestamps with a clock resolution of 1 second.
1969{-# INLINABLE classifySessionsOf #-}
1971 :: (IsStream t, MonadAsync m, Ord k)
1972 => Double -- ^ time window size
1973 -> Fold m a b -- ^ Fold to be applied to window events
1974 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
1976classifySessionsOf interval = classifySessionsBy 1 interval False
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