1{-# LANGUAGE BangPatterns #-}
3{-# LANGUAGE ExistentialQuantification #-}
4{-# LANGUAGE FlexibleContexts #-}
5{-# LANGUAGE RankNTypes #-}
6{-# LANGUAGE RecordWildCards #-}
7{-# LANGUAGE ScopedTypeVariables #-}
10-- Module : Streamly.Pipe
11-- Copyright : (c) 2019 Composewell Technologies
13-- Maintainer : harendra.kumar@gmail.com
14-- Stability : experimental
17-- There are three fundamental types in streamly. They are streams
18-- ("Streamly.Prelude"), pipes ("Streamly.Pipe") and folds ("Streamly.Fold").
19-- Streams are sources or producers of values, multiple sources can be merged
20-- into a single source but a source cannot be split into multiple stream
21-- sources. Folds are sinks or consumers, a stream can be split and
22-- distributed to multiple folds but the results cannot be merged back into a
23-- stream source again. Pipes are transformations, a stream source can be split
24-- and distributed to multiple pipes each pipe can apply its own transform on
25-- the stream and the results can be merged back into a single pipe. Pipes can
26-- be attached to a source to produce a source or they can be attached to a
27-- fold to produce a fold, or multiple pipes can be merged or zipped into a
30-- > import qualified Streamly.Pipe as P
54 -- ** Scanning Filters
59 -- | Insertion adds more elements to the stream.
81 -- | Streams can be split into segments in space or in time. We use the
82 -- term @chunk@ to refer to a spatial length of the stream (spatial window)
83 -- and the term @session@ to refer to a length in time (time window).
85 -- In imperative terms, grouped folding can be considered as a nested loop
86 -- where we loop over the stream to group elements and then loop over
87 -- individual groups to fold them to a single value that is yielded in the
107 -- Keeping the delimiters
112 -- Splitting by multiple sequences
114 -- , splitSuffixOnAny
115 -- , splitPrefixOnAny
131 -- The 'Applicative' instance of a distributing 'Fold' distributes one copy
132 -- of the stream to each fold and combines the results using a function.
136 -- |-------Fold m a b--------|
137 -- ---stream m a---| |---m (b,c,...)
138 -- |-------Fold m a c--------|
143 -- To compute the average of numbers in a stream without going throught he
146 -- >>> let avg = (/) <$> FL.sum <*> fmap fromIntegral FL.length
147 -- >>> FL.foldl' avg (S.enumerateFromTo 1.0 100.0)
150 -- The 'Semigroup' and 'Monoid' instances of a distributing fold distribute
151 -- the input to both the folds and combines the outputs using Monoid or
152 -- Semigroup instances of the output types:
154 -- >>> import Data.Monoid (Sum)
155 -- >>> FL.foldl' (FL.head <> FL.last) (fmap Sum $ S.enumerateFromTo 1.0 100.0)
156 -- Just (Sum {getSum = 101.0})
158 -- The 'Num', 'Floating', and 'Fractional' instances work in the same way.
165 -- Direct items in the input stream to different folds using a function to
166 -- select the fold. This is useful to demultiplex the input stream.
183 -- These can be expressed using lmap/lmapM and unzip
190 , duplicate -- experimental
192 -- * Windowed Classification
193 -- | Split the stream into windows or chunks in space or time. Each window
194 -- can be associated with a key, all events associated with a particular
195 -- key in the window can be folded to a single result. The stream is split
196 -- into windows of specified size, the window can be terminated early if
197 -- the closing flag is specified in the input stream.
199 -- The term "chunk" is used for a space window and the term "session" is
200 -- used for a time window.
202 -- ** Tumbling Windows
203 -- | A new window starts after the previous window is finished.
204 -- , classifyChunksOf
207 -- ** Keep Alive Windows
208 -- | The window size is extended if an event arrives within the specified
209 -- window size. This can represent sessions with idle or inactive timeout.
210 -- , classifyKeepAliveChunks
211 , classifyKeepAliveSessions
214 -- ** Sliding Windows
215 -- | A new window starts after the specified slide from the previous
216 -- window. Therefore windows can overlap.
217 , classifySlidingChunks
218 , classifySlidingSessions
220 -- ** Sliding Window Buffers
221 -- , slidingChunkBuffer
222 -- , slidingSessionBuffer
227-- import Control.Concurrent (threadDelay, forkIO, killThread)
228-- import Control.Concurrent.MVar (MVar, newMVar, takeMVar, putMVar)
229-- import Control.Exception (SomeException(..), catch, mask)
230-- import Control.Monad (void)
231-- import Control.Monad.Catch (throwM)
232-- import Control.Monad.IO.Class (MonadIO(..))
233-- import Control.Monad.Trans (lift)
234-- import Control.Monad.Trans.Control (control)
235-- import Data.Functor.Identity (Identity)
236-- import Data.Heap (Entry(..))
237-- import Data.Map.Strict (Map)
238-- import Data.Maybe (fromJust, isJust, isNothing)
240-- import Foreign.Storable (Storable(..))
242 hiding (id, filter, drop, dropWhile, take, takeWhile, zipWith, foldr,
243 foldl, map, mapM_, sequence, all, any, sum, product, elem,
244 notElem, maximum, minimum, head, last, tail, length, null,
245 reverse, iterate, init, and, or, lookup, foldr1, (!!),
246 scanl, scanl1, replicate, concatMap, mconcat, foldMap, unzip,
247 span, splitAt, break, mapM)
249-- import qualified Data.Heap as H
250-- import qualified Data.Map.Strict as Map
251-- import qualified Prelude
253-- import Streamly (MonadAsync, parallel)
254-- import Streamly.Fold.Types (Fold(..))
255import Streamly.Pipe.Types
256 (Pipe(..), PipeState(..), Step(..), zipWith, tee, map, compose)
257-- import Streamly.Mem.Array.Types (Array)
258-- import Streamly.Mem.Ring (Ring)
259-- import Streamly.Streams.Serial (SerialT)
260-- import Streamly.Streams.StreamK (IsStream())
261-- import Streamly.Time.Units
262-- (AbsTime, MilliSecond64(..), addToAbsTime, diffAbsTime, toRelTime,
265-- import Streamly.Strict
267-- import qualified Streamly.Mem.Array.Types as A
268-- import qualified Streamly.Prelude as S
269-- import qualified Streamly.Streams.StreamD as D
270-- import qualified Streamly.Streams.StreamK as K
271-- import qualified Streamly.Streams.Prelude as P
273------------------------------------------------------------------------------
275------------------------------------------------------------------------------
277-- | Lift a monadic function to a 'Pipe'.
281mapM :: Monad m => (a -> m b) -> Pipe m a b
282mapM f = Pipe consume undefined ()
286 return $ Yield r (Consume ())
288------------------------------------------------------------------------------
290------------------------------------------------------------------------------
292-- | Include only those elements that pass a predicate.
294-- >>> FL.foldl (lfilter (> 5) FL.sum) [1..10]
298{-# INLINABLE lfilter #-}
299lfilter :: Monad m => (a -> Bool) -> Fold m a r -> Fold m a r
300lfilter f (Fold step begin done) = Fold step' begin done
302 step' x a = if f a then step x a else return x
304-- | Like 'lfilter' but with a monadic predicate.
307{-# INLINABLE lfilterM #-}
308lfilterM :: Monad m => (a -> m Bool) -> Fold m a r -> Fold m a r
309lfilterM f (Fold step begin done) = Fold step' begin done
313 if use then step x a else return x
315-- | Take first 'n' elements from the stream and discard the rest.
318{-# INLINABLE ltake #-}
319ltake :: Monad m => Int -> Fold m a b -> Fold m a b
320ltake n (Fold step initial done) = Fold step' initial' done'
322 initial' = fmap (Tuple' 0) initial
323 step' (Tuple' i r) a = do
327 return $ Tuple' (i + 1) res
328 else return $ Tuple' i r
329 done' (Tuple' _ r) = done r
331-- | Takes elements from the input as long as the predicate succeeds.
334{-# INLINABLE ltakeWhile #-}
335ltakeWhile :: Monad m => (a -> Bool) -> Fold m a b -> Fold m a b
336ltakeWhile predicate (Fold step initial done) = Fold step' initial' done'
338 initial' = fmap Left' initial
339 step' (Left' r) a = do
341 then fmap Left' $ step r a
342 else return (Right' r)
344 done' (Left' r) = done r
345 done' (Right' r) = done r
347------------------------------------------------------------------------------
349------------------------------------------------------------------------------
351------------------------------------------------------------------------------
352-- Grouping without looking at elements
353------------------------------------------------------------------------------
355------------------------------------------------------------------------------
357------------------------------------------------------------------------------
360-- | @splitAt n f1 f2@ composes folds @f1@ and @f2@ such that first @n@
361-- elements of its input are sent to fold @f1@ and the rest of the stream is
364-- > let splitAt_ n xs = FL.foldl' (FL.splitAt n FL.toList FL.toList) $ S.fromList xs
366-- >>> splitAt_ 6 "Hello World!"
367-- > ("Hello ","World!")
369-- >>> splitAt_ (-1) [1,2,3]
372-- >>> splitAt_ 0 [1,2,3]
375-- >>> splitAt_ 1 [1,2,3]
378-- >>> splitAt_ 3 [1,2,3]
381-- >>> splitAt_ 4 [1,2,3]
384-- This can be considered as a two-fold version of 'ltake' where we take both
385-- the segments instead of discarding the leftover.
388{-# INLINE splitAt #-}
395splitAt n (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
396 Fold step init extract
398 init = Tuple3' <$> return n <*> initialL <*> initialR
400 step (Tuple3' i xL xR) input =
402 then stepL xL input >>= (\a -> return (Tuple3' (i - 1) a xR))
403 else stepR xR input >>= (\b -> return (Tuple3' i xL b))
405 extract (Tuple3' _ a b) = (,) <$> extractL a <*> extractR b
407------------------------------------------------------------------------------
409------------------------------------------------------------------------------
411------------------------------------------------------------------------------
412-- Generalized grouping
413------------------------------------------------------------------------------
415-- This combinator is the most general grouping combinator and can be used to
416-- implement all other grouping combinators.
418-- XXX check if this can implement the splitOn combinator i.e. we can slide in
419-- new elements, slide out old elements and incrementally compute the hash.
420-- Also, can we implement the windowed classification combinators using this?
422-- In fact this is a parse. Instead of using a special return value in the fold
423-- we are using a mapping function.
425-- Note that 'scanl'' (usually followed by a map to extract the desired value
426-- from the accumulator) can be used to realize many implementations e.g. a
427-- sliding window implementation. A scan followed by a mapMaybe is also a good
428-- pattern to express many problems where we want to emit a filtered output and
429-- not emit an output on every input.
431-- Passing on of the initial accumulator value to the next fold is equivalent
432-- to returning the leftover concept.
435-- | @groupScan splitter fold stream@ folds the input stream using @fold@.
436-- @splitter@ is applied on the accumulator of the fold every time an item is
437-- consumed by the fold. The fold continues until @splitter@ returns a 'Just'
438-- value. A 'Just' result from the @splitter@ specifies a result to be emitted
439-- in the output stream and the initial value of the accumulator for the next
440-- group's fold. This allows us to control whether to start fresh for the next
441-- fold or to continue from the previous fold's output.
443{-# INLINE groupScan #-}
445 :: (IsStream t, Monad m)
446 => (x -> m (Maybe (b, x))) -> Fold m a x -> t m a -> t m b
447groupScan split fold m = undefined
450-- | Group the input stream into groups of @n@ elements each and then fold each
451-- group using the provided fold function.
453-- >> S.toList $ FL.chunksOf 2 FL.sum (S.enumerateFromTo 1 10)
456-- This can be considered as an n-fold version of 'ltake' where we apply
457-- 'ltake' repeatedly on the leftover stream until the stream exhausts.
460{-# INLINE chunksOf #-}
462 :: (IsStream t, Monad m)
463 => Int -> Fold m a b -> t m a -> t m b
464chunksOf n f m = D.fromStreamD $ D.groupsOf n f (D.toStreamD m)
466-- | Transform a fold from a pure input to a 'Maybe' input, consuming only
468{-# INLINE lcatMaybes #-}
469lcatMaybes :: Monad m => Fold m a b -> Fold m (Maybe a) b
470lcatMaybes = lfilter isJust . lmap fromJust
472-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
473-- XXX add this example after fixing the serial stream rate control
474-- >>> S.toList $ S.take 5 $ sessionsOf 1 FL.sum $ constRate 2 $ S.enumerateFrom 1
477-- | Group the input stream into windows of @n@ second each and then fold each
478-- group using the provided fold function.
481{-# INLINE sessionsOf #-}
483 :: (IsStream t, MonadAsync m)
484 => Double -> Fold m a b -> t m a -> t m b
486 splitSuffixBy' isNothing (lcatMaybes f)
487 (S.intersperseByTime n (return Nothing) (S.map Just xs))
489------------------------------------------------------------------------------
491------------------------------------------------------------------------------
493------------------------------------------------------------------------------
495------------------------------------------------------------------------------
497-- | Break the input stream into two groups, the first group takes the input as
498-- long as the predicate applied to the first element of the stream and next
499-- input element holds 'True', the second group takes the rest of the input.
507_spanBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
508 Fold step init extract
511 init = Tuple3' <$> initialL <*> initialR <*> return (Tuple' Nothing True)
513 step (Tuple3' a b (Tuple' (Just frst) isFirstG)) input =
514 if cmp frst input && isFirstG
516 >>= (\a' -> return (Tuple3' a' b (Tuple' (Just frst) isFirstG)))
518 >>= (\a' -> return (Tuple3' a a' (Tuple' Nothing False)))
520 step (Tuple3' a b (Tuple' Nothing isFirstG)) input =
523 >>= (\a' -> return (Tuple3' a' b (Tuple' (Just input) isFirstG)))
525 >>= (\a' -> return (Tuple3' a a' (Tuple' Nothing False)))
527 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
529-- | Span as long as the predicate is 'True'. @span p f1 f2@ composes folds
530-- @f1@ and @f2@ such that the composed fold continues sending the input to
531-- @f1@ as long as the predicate @p@ is 'True'. The rest of the input is sent
534-- > let span_ p xs = FL.foldl' (FL.span p FL.toList FL.toList) $ S.fromList xs
536-- >>> span_ (< 1) [1,2,3]
539-- >>> span_ (< 2) [1,2,3]
542-- >>> span_ (< 4) [1,2,3]
545-- This can be considered as a two-fold version of 'ltakeWhile' where we take
546-- both the segments instead of discarding the leftover.
556span p (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
557 Fold step init extract
560 init = Tuple3' <$> initialL <*> initialR <*> return True
562 step (Tuple3' a b isFirstG) input =
563 if isFirstG && p input
564 then stepL a input >>= (\a' -> return (Tuple3' a' b True))
565 else stepR b input >>= (\a' -> return (Tuple3' a a' False))
567 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
570-- > break p = span (not . p)
572-- Break as soon as the predicate becomes 'True'. @break p f1 f2@ composes
573-- folds @f1@ and @f2@ such that @f1@ stops receiving input as soon as the
574-- predicate @p@ becomes 'True'. The rest of the input is sent to @f2@.
576-- This is the binary version of 'splitBy'.
578-- > let break_ p xs = FL.foldl' (FL.break p FL.toList FL.toList) $ S.fromList xs
580-- >>> break_ (< 1) [3,2,1]
583-- >>> break_ (< 2) [3,2,1]
586-- >>> break_ (< 4) [3,2,1]
597break p = span (not . p)
599-- | Like 'spanBy' but applies the predicate in a rolling fashion i.e.
600-- predicate is applied to the previous and the next input elements.
601{-# INLINE _spanRollingBy #-}
608_spanRollingBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
609 Fold step init extract
612 init = Tuple3' <$> initialL <*> initialR <*> return Nothing
614 step (Tuple3' a b (Just frst)) input =
616 then stepL a input >>= (\a' -> return (Tuple3' a' b (Just input)))
617 else stepR b input >>= (\b' -> return (Tuple3' a b' (Just input)))
619 step (Tuple3' a b Nothing) input =
620 stepL a input >>= (\a' -> return (Tuple3' a' b (Just input)))
622 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
624------------------------------------------------------------------------------
626------------------------------------------------------------------------------
628-- | @groupsBy cmp f $ S.fromList [a,b,c,...]@ assigns the element @a@ to the
629-- first group, if @a \`cmp` b@ is 'True' then @b@ is also assigned to the same
630-- group. If @a \`cmp` c@ is 'True' then @c@ is also assigned to the same
631-- group and so on. When the comparison fails a new group is started. Each
632-- group is folded using the fold @f@.
634-- >>> S.toList $ FL.groupsBy (>) FL.toList $ S.fromList [1,3,7,0,2,5]
635-- > [[1,3,7],[0,2,5]]
638{-# INLINE groupsBy #-}
640 :: (IsStream t, Monad m)
645groupsBy cmp f m = D.fromStreamD $ D.groupsBy cmp f (D.toStreamD m)
647-- | Unlike @groupsBy@ this function performs a rolling comparison of two
648-- successive elements in the input stream. @groupsRollingBy cmp f $ S.fromList
649-- [a,b,c,...]@ assigns the element @a@ to the first group, if @a \`cmp` b@ is
650-- 'True' then @b@ is also assigned to the same group. If @b \`cmp` c@ is
651-- 'True' then @c@ is also assigned to the same group and so on. When the
652-- comparison fails a new group is started. Each group is folded using the fold
655-- >>> S.toList $ FL.groupsRollingBy (\a b -> a + 1 == b) FL.toList $ S.fromList [1,2,3,7,8,9]
656-- > [[1,2,3],[7,8,9]]
659{-# INLINE groupsRollingBy #-}
661 :: (IsStream t, Monad m)
666groupsRollingBy cmp f m = D.fromStreamD $ D.groupsRollingBy cmp f (D.toStreamD m)
669-- > groups = groupsBy (==)
670-- > groups = groupsRollingBy (==)
672-- Groups a contiguous span of equal elements together in one group.
674-- >>> S.toList $ FL.groups FL.toList $ S.fromList [1,1,2,2]
678groups :: (IsStream t, Monad m, Eq a) => Fold m a b -> t m a -> t m b
679groups = groupsBy (==)
681------------------------------------------------------------------------------
682-- Binary splitting on a separator
683------------------------------------------------------------------------------
686-- | Find the first occurrence of the specified sequence in the input stream
687-- and break the input stream into two parts, the first part consisting of the
688-- stream before the sequence and the second part consisting of the sequence
689-- and the rest of the stream.
691-- > let breakOn_ pat xs = FL.foldl' (FL.breakOn pat FL.toList FL.toList) $ S.fromList xs
693-- >>> breakOn_ "dear" "Hello dear world!"
694-- > ("Hello ","dear world!")
696{-# INLINE breakOn #-}
697breakOn :: Monad m => Array a -> Fold m a b -> Fold m a c -> Fold m a (b,c)
698breakOn pat f m = undefined
701------------------------------------------------------------------------------
702-- N-ary split on a predicate
703------------------------------------------------------------------------------
705-- TODO: Use a Splitter configuration similar to the "split" package to make it
706-- possible to express all splitting combinations. In general, we can have
707-- infix/suffix/prefix/condensing of separators, dropping both leading/trailing
708-- separators. We can have a single split operation taking the splitter config
711-- | Split a stream on separator elements determined by a predicate, dropping
712-- the separator. Separators are not considered part of stream segments on
713-- either side of it instead they are treated as infixed between two stream
714-- segments. For example, with @.@ as separator, @"a.b.c"@ would be parsed as
715-- @["a","b","c"]@. When @.@ is in leading or trailing position it is still
716-- considered as infixed, treating the first or the last segment as empty. For
717-- example, @".a."@ would be parsed as @["","a",""]@. This operation is
718-- opposite of 'intercalate'.
720-- Let's use the following definition for illustration:
722-- > splitBy_ p xs = S.toList $ FL.splitBy p (FL.toList) (S.fromList xs)
724-- >>> splitBy_ (== '.') ""
727-- >>> splitBy_ (== '.') "."
730-- >>> splitBy_ (== '.') ".a"
733-- >>> splitBy_ (== '.') "a."
736-- >>> splitBy_ (== '.') "a.b"
739-- >>> splitBy_ (== '.') "a..b"
742-- This can be considered as an n-fold version of 'break' where we apply
743-- 'break' successively on the input stream, dropping the first element
744-- of the second segment after each break.
747{-# INLINE splitBy #-}
749 :: (IsStream t, Monad m)
750 => (a -> Bool) -> Fold m a b -> t m a -> t m b
751splitBy predicate f m =
752 D.fromStreamD $ D.splitBy predicate f (D.toStreamD m)
754-- | Like 'splitBy' but the separator is treated as part of the previous
755-- stream segment (suffix). Therefore, when the separator is in trailing
756-- position, no empty segment is considered to follow it. For example, @"a.b."@
757-- would be parsed as @["a","b"]@ instead of @["a","b",""]@ as in the case of
760-- > splitSuffixBy_ p xs = S.toList $ FL.splitSuffixBy p (FL.toList) (S.fromList xs)
762-- >>> splitSuffixBy_ (== '.') ""
765-- >>> splitSuffixBy_ (== '.') "."
768-- >>> splitSuffixBy_ (== '.') "a"
771-- >>> splitSuffixBy_ (== '.') ".a"
774-- >>> splitSuffixBy_ (== '.') "a."
777-- >>> splitSuffixBy_ (== '.') "a.b"
780-- >>> splitSuffixBy_ (== '.') "a.b."
783-- >>> splitSuffixBy_ (== '.') "a..b.."
786-- > lines = splitSuffixBy (== '\n')
788-- This can be considered as an n-fold version of 'breakPost' where we apply
789-- 'breakPost' successively on the input stream, dropping the first element
790-- of the second segment after each break.
793{-# INLINE splitSuffixBy #-}
795 :: (IsStream t, Monad m)
796 => (a -> Bool) -> Fold m a b -> t m a -> t m b
797splitSuffixBy predicate f m =
798 D.fromStreamD $ D.splitSuffixBy predicate f (D.toStreamD m)
800-- | Like 'splitBy' but ignores repeated separators or separators in leading
801-- or trailing position. Therefore, @"..a..b.."@ would be parsed as
802-- @["a","b"]@. In other words, it treats the input like words separated by
803-- whitespace elements determined by the predicate.
805-- > wordsBy' p xs = S.toList $ FL.wordsBy p (FL.toList) (S.fromList xs)
807-- >>> wordsBy' (== ',') ""
810-- >>> wordsBy' (== ',') ","
813-- >>> wordsBy' (== ',') ",a,,b,"
816-- > words = wordsBy isSpace
819{-# INLINE wordsBy #-}
821 :: (IsStream t, Monad m)
822 => (a -> Bool) -> Fold m a b -> t m a -> t m b
823wordsBy predicate f m =
824 D.fromStreamD $ D.wordsBy predicate f (D.toStreamD m)
826-- XXX we should express this using the Splitter config.
828-- We can get splitSuffixBy' by appending the suffix to the output segments
829-- produced by splitSuffixBy. However, it may add an additional suffix if the last
830-- fragment did not have a suffix in the first place.
832-- | Like 'splitSuffixBy' but keeps the suffix in the splits.
834-- > splitSuffixBy'_ p xs = S.toList $ FL.splitSuffixBy' p (FL.toList) (S.fromList xs)
836-- >>> splitSuffixBy'_ (== '.') ""
839-- >>> splitSuffixBy'_ (== '.') "."
842-- >>> splitSuffixBy'_ (== '.') "a"
845-- >>> splitSuffixBy'_ (== '.') ".a"
848-- >>> splitSuffixBy'_ (== '.') "a."
851-- >>> splitSuffixBy'_ (== '.') "a.b"
854-- >>> splitSuffixBy'_ (== '.') "a.b."
857-- >>> splitSuffixBy'_ (== '.') "a..b.."
858-- > ["a.",".","b.","."]
860-- This can be considered as an n-fold version of 'breakPost' where we apply
861-- 'breakPost' successively on the input stream.
864{-# INLINE splitSuffixBy' #-}
866 :: (IsStream t, Monad m)
867 => (a -> Bool) -> Fold m a b -> t m a -> t m b
868splitSuffixBy' predicate f m =
869 D.fromStreamD $ D.splitSuffixBy' predicate f (D.toStreamD m)
871------------------------------------------------------------------------------
872-- Split on a delimiter sequence
873------------------------------------------------------------------------------
875-- Int list examples for splitOn:
877-- >>> splitList [] [1,2,3,3,4]
878-- > [[1],[2],[3],[3],[4]]
880-- >>> splitList [5] [1,2,3,3,4]
883-- >>> splitList [1] [1,2,3,3,4]
886-- >>> splitList [4] [1,2,3,3,4]
889-- >>> splitList [2] [1,2,3,3,4]
892-- >>> splitList [3] [1,2,3,3,4]
895-- >>> splitList [3,3] [1,2,3,3,4]
898-- >>> splitList [1,2,3,3,4] [1,2,3,3,4]
901-- | Split the stream on both sides of a separator sequence, dropping the
904-- For illustration, let's define a function that operates on pure lists:
907-- splitOn_ pat xs = S.toList $ FL.splitOn (A.fromList pat) (FL.toList) (S.fromList xs)
910-- >>> splitOn_ "" "hello"
911-- > ["h","e","l","l","o"]
913-- >>> splitOn_ "hello" ""
916-- >>> splitOn_ "hello" "hello"
919-- >>> splitOn_ "x" "hello"
922-- >>> splitOn_ "h" "hello"
925-- >>> splitOn_ "o" "hello"
928-- >>> splitOn_ "e" "hello"
931-- >>> splitOn_ "l" "hello"
934-- >>> splitOn_ "ll" "hello"
937-- 'splitOn' is an inverse of 'intercalate'. The following law always holds:
939-- > intercalate . splitOn == id
941-- The following law holds when the separator is non-empty and contains none of
942-- the elements present in the input lists:
944-- > splitOn . intercalate == id
946-- The following law always holds:
948-- > concat . splitOn . intercalate == concat
951{-# INLINE splitOn #-}
953 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
954 => Array a -> Fold m a b -> t m a -> t m b
955splitOn patt f m = D.fromStreamD $ D.splitOn patt f (D.toStreamD m)
958-- This can be implemented easily using Rabin Karp
959-- | Split on any one of the given patterns.
960{-# INLINE splitOnAny #-}
962 :: (IsStream t, Monad m, Storable a, Integral a)
963 => [Array a] -> Fold m a b -> t m a -> t m b
964splitOnAny subseq f m = undefined -- D.fromStreamD $ D.splitOnAny f subseq (D.toStreamD m)
967-- | Like 'splitSuffixBy' but the separator is a sequence of elements, instead
968-- of a predicate for a single element.
970-- > splitSuffixOn_ pat xs = S.toList $ FL.splitSuffixOn (A.fromList pat) (FL.toList) (S.fromList xs)
972-- >>> splitSuffixOn_ "." ""
975-- >>> splitSuffixOn_ "." "."
978-- >>> splitSuffixOn_ "." "a"
981-- >>> splitSuffixOn_ "." ".a"
984-- >>> splitSuffixOn_ "." "a."
987-- >>> splitSuffixOn_ "." "a.b"
990-- >>> splitSuffixOn_ "." "a.b."
993-- >>> splitSuffixOn_ "." "a..b.."
996-- > lines = splitSuffixOn "\n"
999{-# INLINE splitSuffixOn #-}
1001 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
1002 => Array a -> Fold m a b -> t m a -> t m b
1003splitSuffixOn patt f m =
1004 D.fromStreamD $ D.splitSuffixOn False patt f (D.toStreamD m)
1007-- | Like 'splitOn' but drops any empty splits.
1009{-# INLINE wordsOn #-}
1011 :: (IsStream t, Monad m, Storable a, Eq a)
1012 => Array a -> Fold m a b -> t m a -> t m b
1013wordsOn subseq f m = undefined -- D.fromStreamD $ D.wordsOn f subseq (D.toStreamD m)
1016-- XXX use a non-monadic intersperse to remove the MonadAsync constraint.
1018-- | Like 'splitOn' but splits the separator as well, as an infix token.
1020-- > splitOn'_ pat xs = S.toList $ FL.splitOn' (A.fromList pat) (FL.toList) (S.fromList xs)
1022-- >>> splitOn'_ "" "hello"
1023-- > ["h","","e","","l","","l","","o"]
1025-- >>> splitOn'_ "hello" ""
1028-- >>> splitOn'_ "hello" "hello"
1031-- >>> splitOn'_ "x" "hello"
1034-- >>> splitOn'_ "h" "hello"
1037-- >>> splitOn'_ "o" "hello"
1040-- >>> splitOn'_ "e" "hello"
1043-- >>> splitOn'_ "l" "hello"
1044-- > ["he","l","","l","o"]
1046-- >>> splitOn'_ "ll" "hello"
1050{-# INLINE splitOn' #-}
1052 :: (IsStream t, MonadAsync m, Storable a, Enum a, Eq a)
1053 => Array a -> Fold m a b -> t m a -> t m b
1054splitOn' patt f m = S.intersperseM (foldl' f (A.read patt)) $ splitOn patt f m
1056-- | Like 'splitSuffixOn' but keeps the suffix intact in the splits.
1058-- > splitSuffixOn'_ pat xs = S.toList $ FL.splitSuffixOn' (A.fromList pat) (FL.toList) (S.fromList xs)
1060-- >>> splitSuffixOn'_ "." ""
1063-- >>> splitSuffixOn'_ "." "."
1066-- >>> splitSuffixOn'_ "." "a"
1069-- >>> splitSuffixOn'_ "." ".a"
1072-- >>> splitSuffixOn'_ "." "a."
1075-- >>> splitSuffixOn'_ "." "a.b"
1078-- >>> splitSuffixOn'_ "." "a.b."
1081-- >>> splitSuffixOn'_ "." "a..b.."
1082-- > ["a.",".","b.","."]
1085{-# INLINE splitSuffixOn' #-}
1087 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
1088 => Array a -> Fold m a b -> t m a -> t m b
1089splitSuffixOn' patt f m =
1090 D.fromStreamD $ D.splitSuffixOn True patt f (D.toStreamD m)
1093-- This can be implemented easily using Rabin Karp
1094-- | Split post any one of the given patterns.
1095{-# INLINE splitSuffixOnAny #-}
1097 :: (IsStream t, Monad m, Storable a, Integral a)
1098 => [Array a] -> Fold m a b -> t m a -> t m b
1099splitSuffixOnAny subseq f m = undefined
1100 -- D.fromStreamD $ D.splitPostAny f subseq (D.toStreamD m)
1103------------------------------------------------------------------------------
1104-- Reorder in sequence
1105------------------------------------------------------------------------------
1108-- Buffer until the next element in sequence arrives. The function argument
1109-- determines the difference in sequence numbers. This could be useful in
1110-- implementing sequenced streams, for example, TCP reassembly.
1111{-# INLINE reassembleBy #-}
1113 :: (IsStream t, Monad m)
1118reassembleBy = undefined
1121------------------------------------------------------------------------------
1123------------------------------------------------------------------------------
1125-- | Distribute one copy of the stream to each fold and zip the results.
1128-- |-------Fold m a b--------|
1129-- ---stream m a---| |---m (b,c)
1130-- |-------Fold m a c--------|
1132-- >>> FL.foldl' (FL.tee FL.sum FL.length) (S.enumerateFromTo 1.0 100.0)
1137tee :: Monad m => Fold m a b -> Fold m a c -> Fold m a (b,c)
1138tee f1 f2 = (,) <$> f1 <*> f2
1140{-# INLINE foldNil #-}
1141foldNil :: Monad m => Fold m a [b]
1142foldNil = Fold step begin done where
1144 step _ _ = return []
1147{-# INLINE foldCons #-}
1148foldCons :: Monad m => Fold m a b -> Fold m a [b] -> Fold m a [b]
1149foldCons (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1150 Fold step begin done
1154 begin = Tuple' <$> beginL <*> beginR
1155 step (Tuple' xL xR) a = Tuple' <$> stepL xL a <*> stepR xR a
1156 done (Tuple' xL xR) = (:) <$> (doneL xL) <*> (doneR xR)
1158-- XXX use "List" instead of "[]"?, use Array for output to scale it to a large
1159-- number of consumers?
1161-- | Distribute one copy of the stream to each fold and collect the results in
1166-- |-------Fold m a b--------|
1167-- ---stream m a---| |---m [b]
1168-- |-------Fold m a b--------|
1173-- >>> FL.foldl' (FL.distribute [FL.sum, FL.length]) (S.enumerateFromTo 1 5)
1176-- This is the consumer side dual of the producer side 'sequence' operation.
1179{-# INLINE distribute #-}
1180distribute :: Monad m => [Fold m a b] -> Fold m a [b]
1181distribute [] = foldNil
1182distribute (x:xs) = foldCons x (distribute xs)
1184------------------------------------------------------------------------------
1186------------------------------------------------------------------------------
1188-- | Partition the input over two folds using an 'Either' partitioning
1193-- |-------Fold b x--------|
1194-- -----stream m a --> (Either b c)----| |----(x,y)
1195-- |-------Fold c y--------|
1198-- Send input to either fold randomly:
1200-- >>> import System.Random (randomIO)
1201-- >>> randomly a = randomIO >>= \x -> return $ if x then Left a else Right a
1202-- >>> FL.foldl' (FL.partitionByM randomly FL.length FL.length) (S.enumerateFromTo 1 100)
1205-- Send input to the two folds in a proportion of 2:1:
1208-- import Data.IORef (newIORef, readIORef, writeIORef)
1209-- proportionately m n = do
1210-- ref <- newIORef $ cycle $ concat [replicate m Left, replicate n Right]
1211-- return $ \\a -> do
1212-- r <- readIORef ref
1213-- writeIORef ref $ tail r
1217-- f <- proportionately 2 1
1218-- r <- FL.foldl' (FL.partitionByM f FL.length FL.length) (S.enumerateFromTo (1 :: Int) 100)
1225-- This is the consumer side dual of the producer side 'mergeBy' operation.
1228{-# INLINE partitionByM #-}
1229partitionByM :: Monad m
1230 => (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1231partitionByM f (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1233 Fold step begin done
1237 begin = Tuple' <$> beginL <*> beginR
1238 step (Tuple' xL xR) a = do
1241 Left b -> Tuple' <$> stepL xL b <*> return xR
1242 Right c -> Tuple' <$> return xL <*> stepR xR c
1243 done (Tuple' xL xR) = (,) <$> doneL xL <*> doneR xR
1245-- Note: we could use (a -> Bool) instead of (a -> Either b c), but the latter
1246-- makes the signature clearer as to which case belongs to which fold.
1247-- XXX need to check the performance in both cases.
1249-- | Same as 'partitionByM' but with a pure partition function.
1251-- Count even and odd numbers in a stream:
1254-- >>> let f = FL.partitionBy (\\n -> if even n then Left n else Right n)
1255-- (fmap (("Even " ++) . show) FL.length)
1256-- (fmap (("Odd " ++) . show) FL.length)
1257-- in FL.foldl' f (S.enumerateFromTo 1 100)
1258-- ("Even 50","Odd 50")
1262{-# INLINE partitionBy #-}
1263partitionBy :: Monad m
1264 => (a -> Either b c) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1265partitionBy f = partitionByM (return . f)
1267-- | Compose two folds such that the combined fold accepts a stream of 'Either'
1268-- and routes the 'Left' values to the first fold and 'Right' values to the
1271-- > partition = partitionBy id
1274{-# INLINE partition #-}
1276 => Fold m b x -> Fold m c y -> Fold m (Either b c) (x, y)
1277partition = partitionBy id
1280-- | Send one item to each fold in a round-robin fashion. This is the consumer
1281-- side dual of producer side 'mergeN' operation.
1283-- partitionN :: Monad m => [Fold m a b] -> Fold m a [b]
1284-- partitionN fs = Fold step begin done
1287-- Demultiplex an input element into a number of typed variants. We want to
1288-- statically restrict the target values within a set of predefined types, an
1289-- enumeration of a GADT. We also want to make sure that the Map contains only
1290-- those types and the full set of those types. Instead of Map it should
1291-- probably be a lookup-table using an array and not in GC memory.
1293-- This is the consumer side dual of the producer side 'mux' operation (XXX to
1296-- | Split the input stream based on a key field and fold each split using a
1297-- specific fold collecting the results in a map from the keys to the results.
1298-- Useful for cases like protocol handlers to handle different type of packets
1299-- using different handlers.
1303-- |-------Fold m a b
1304-- -----stream m a-----Map-----|
1305-- |-------Fold m a b
1311{-# INLINE demuxWith #-}
1312demuxWith :: (Monad m, Ord k)
1313 => (a -> k) -> Map k (Fold m a b) -> Fold m a (Map k b)
1314demuxWith f kv = Fold step initial extract
1320 -- XXX should we raise an exception in Nothing case?
1321 -- Ideally we should enforce that it is a total map over k so that look
1323 -- XXX we could use a monadic update function for a single lookup and
1324 -- update in the map.
1326 in case Map.lookup k mp of
1327 Nothing -> return mp
1328 Just (Fold step' acc extract') -> do
1329 !r <- acc >>= \x -> step' x a
1330 return $ Map.insert k (Fold step' (return r) extract') mp
1331 extract = Prelude.mapM (\(Fold _ acc e) -> acc >>= e)
1333-- | Fold a stream of key value pairs using a map of specific folds for each
1334-- key into a map from keys to the results of fold outputs of the corresponding
1338-- > let table = Data.Map.fromList [(\"SUM", FL.sum), (\"PRODUCT", FL.product)]
1339-- input = S.fromList [(\"SUM",1),(\"PRODUCT",2),(\"SUM",3),(\"PRODUCT",4)]
1340-- in FL.foldl' (FL.demux table) input
1347demux :: (Monad m, Ord k)
1348 => Map k (Fold m a b) -> Fold m (k, a) (Map k b)
1349demux fs = demuxWith fst (Map.map (lmap snd) fs)
1351-- | Split the input stream based on a key field and fold each split using a
1352-- specific fold without collecting the results. Useful for cases like protocol
1353-- handlers to handle different type of packets.
1357-- |-------Fold m a ()
1358-- -----stream m a-----Map-----|
1359-- |-------Fold m a ()
1367-- demuxWith_ can be slightly faster than demuxWith because we do not need to
1368-- update the Map in this case. This may be significant only if the map is
1370{-# INLINE demuxWith_ #-}
1371demuxWith_ :: (Monad m, Ord k)
1372 => (a -> k) -> Map k (Fold m a b) -> Fold m a ()
1373demuxWith_ f kv = Fold step initial extract
1378 Prelude.mapM (\(Fold s i e) ->
1379 i >>= \r -> return (Fold s (return r) e)) kv
1381 -- XXX should we raise an exception in Nothing case?
1382 -- Ideally we should enforce that it is a total map over k so that look
1384 case Map.lookup (f a) mp of
1385 Nothing -> return mp
1386 Just (Fold step' acc _) -> do
1387 _ <- acc >>= \x -> step' x a
1389 extract mp = Prelude.mapM (\(Fold _ acc e) -> acc >>= e) mp >> return ()
1391-- | Given a stream of key value pairs and a map from keys to folds, fold the
1392-- values for each key using the corresponding folds, discarding the outputs.
1395-- > let prn = FL.drainBy print
1396-- > let table = Data.Map.fromList [(\"ONE", prn), (\"TWO", prn)]
1397-- input = S.fromList [(\"ONE",1),(\"TWO",2)]
1398-- in FL.foldl' (FL.demux_ table) input
1404{-# INLINE demux_ #-}
1405demux_ :: (Monad m, Ord k) => Map k (Fold m a ()) -> Fold m (k, a) ()
1406demux_ fs = demuxWith_ fst (Map.map (lmap snd) fs)
1408-- XXX instead of a Map we could yield the results as a pure stream as they
1409-- complete. We could then concatMap the fold results to implement the
1410-- windowing combinators.
1412-- | Split the input stream based on a key field and fold each split using the
1413-- given fold. Useful for map/reduce, bucketizing the input in different bins
1414-- or for generating histograms.
1417-- > let input = S.fromList [(\"ONE",1),(\"ONE",1.1),(\"TWO",2), (\"TWO",2.2)]
1418-- in FL.foldl' (FL.classify FL.toListRev) input
1419-- fromList [(\"ONE",[1.1,1.0]),(\"TWO",[2.2,2.0])]
1423{-# INLINE classifyWith #-}
1424classifyWith :: (Monad m, Ord k) => (a -> k) -> Fold m a b -> Fold m a (Map k b)
1425classifyWith f (Fold step initial extract) = Fold step' initial' extract'
1429 initial' = return Map.empty
1432 in case Map.lookup k kv of
1436 return $ Map.insert k r kv
1439 return $ Map.insert k r kv
1440 extract' = Prelude.mapM extract
1442-- | Given an input stream of key value pairs and a fold for values, fold all
1443-- the values belonging to each key. Useful for map/reduce, bucketizing the
1444-- input in different bins or for generating histograms.
1447-- > let input = S.fromList [(\"ONE",1),(\"ONE",1.1),(\"TWO",2), (\"TWO",2.2)]
1448-- in FL.foldl' (FL.classify FL.toListRev) input
1449-- fromList [(\"ONE",[1.1,1.0]),(\"TWO",[2.2,2.0])]
1454-- > classify fld = classifyWith fst (lmap snd fld)
1457{-# INLINE classify #-}
1458classify :: (Monad m, Ord k) => Fold m a b -> Fold m (k, a) (Map k b)
1459classify fld = classifyWith fst (lmap snd fld)
1461------------------------------------------------------------------------------
1463------------------------------------------------------------------------------
1465-- | Like 'unzipWith' but with a monadic splitter function.
1468{-# INLINE unzipWithM #-}
1469unzipWithM :: Monad m
1470 => (a -> m (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
1471unzipWithM f (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1472 Fold step begin done
1476 step (Tuple' xL xR) a = do
1478 Tuple' <$> stepL xL b <*> stepR xR c
1479 begin = Tuple' <$> beginL <*> beginR
1480 done (Tuple' xL xR) = (,) <$> doneL xL <*> doneR xR
1482-- | Split elements in the input stream into two parts using a pure splitter
1483-- function, direct each part to a different fold and zip the results.
1486{-# INLINE unzipWith #-}
1488 => (a -> (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
1489unzipWith f = unzipWithM (return . f)
1491-- | Send the elements of tuples in a stream of tuples through two different
1496-- |-------Fold a x--------|
1497-- -----Stream m x----(a,b)--| |----m (x,y)
1498-- |-------Fold b y--------|
1502-- This is the consumer side dual of the producer side 'zip' operation.
1506unzip :: Monad m => Fold m a x -> Fold m b y -> Fold m (a,b) (x,y)
1509------------------------------------------------------------------------------
1511------------------------------------------------------------------------------
1513-- | Modify the fold such that when the fold is done, instead of returning the
1514-- accumulator, it returns a fold. The returned fold starts from where we left
1515-- i.e. it uses the last accumulator value as the initial value of the
1516-- accumulator. Thus we can resume the fold later and feed it more input.
1519-- > more <- FL.foldl (FL.duplicate FL.sum) (S.enumerateFromTo 1 10)
1520-- > evenMore <- FL.foldl (FL.duplicate more) (S.enumerateFromTo 11 20)
1521-- > FL.foldl evenMore (S.enumerateFromTo 21 30)
1525{-# INLINABLE duplicate #-}
1526duplicate :: Applicative m => Fold m a b -> Fold m a (Fold m a b)
1527duplicate (Fold step begin done) =
1528 Fold step begin (\x -> pure (Fold step (pure x) done))
1531-- All the stream flattening transformations can also be applied to a fold
1534-- | This can be used to apply all the stream generation operations on folds.
1535lconcatMap ::(IsStream t, Monad m) => (a -> t m b)
1538lconcatMap s f1 f2 = undefined
1541-- All the grouping transformation that we apply to a stream can also be
1542-- applied to a fold input stream.
1545-- | Group the input stream into groups of elements between @low@ and @high@.
1546-- Collection starts in chunks of @low@ and then keeps doubling until we reach
1547-- @high@. Each chunk is folded using the provided fold function.
1549-- This could be useful, for example, when we are folding a stream of unknown
1550-- size to a stream of arrays and we want to minimize the number of
1555-- XXX we should be able to implement it with parsers/terminating folds.
1557{-# INLINE lchunksInRange #-}
1558lchunksInRange :: Monad m
1559 => Int -> Int -> Fold m a b -> Fold m b c -> Fold m a c
1560lchunksInRange low high (Fold step1 initial1 extract1)
1561 (Fold step2 initial2 extract2) = undefined
1564-- | Group the input stream into groups of @n@ elements each and then fold each
1565-- group using the provided fold function.
1569-- -----Fold m a b----|-Fold n a c-|-Fold n a c-|-...-|----Fold m a c
1573{-# INLINE lchunksOf #-}
1574lchunksOf :: Monad m => Int -> Fold m a b -> Fold m b c -> Fold m a c
1575lchunksOf n (Fold step1 initial1 extract1) (Fold step2 initial2 extract2) =
1576 Fold step' initial' extract'
1580 initial' = (Tuple3' 0) <$> initial1 <*> initial2
1581 step' (Tuple3' i r1 r2) a = do
1585 return $ Tuple3' (i + 1) res r2
1588 acc2 <- step2 r2 res
1592 return $ Tuple3' 1 acc1 acc2
1593 extract' (Tuple3' _ _ r) = extract2 r
1595-- | Group the input stream into windows of n second each and then fold each
1596-- group using the provided fold function.
1598-- For example, we can copy and distribute a stream to multiple folds where
1599-- each fold can group the input differently e.g. by one second, one minute and
1600-- one hour windows respectively and fold each resulting stream of folds.
1604-- -----Fold m a b----|-Fold n a c-|-Fold n a c-|-...-|----Fold m a c
1607{-# INLINE lsessionsOf #-}
1608lsessionsOf :: MonadAsync m => Double -> Fold m a b -> Fold m b c -> Fold m a c
1609lsessionsOf n (Fold step1 initial1 extract1) (Fold step2 initial2 extract2) =
1610 Fold step' initial' extract'
1614 -- XXX MVar may be expensive we need a cheaper synch mechanism here
1618 mv1 <- liftIO $ newMVar i1
1619 mv2 <- liftIO $ newMVar (Right i2)
1620 t <- control $ \run ->
1621 mask $ \restore -> do
1622 tid <- forkIO $ catch (restore $ void $ run (timerThread mv1 mv2))
1623 (handleChildException mv2)
1625 return $ Tuple3' t mv1 mv2
1626 step' acc@(Tuple3' _ mv1 _) a = do
1627 r1 <- liftIO $ takeMVar mv1
1629 liftIO $ putMVar mv1 res
1631 extract' (Tuple3' tid _ mv2) = do
1632 r2 <- liftIO $ takeMVar mv2
1633 liftIO $ killThread tid
1636 Right x -> extract2 x
1638 timerThread mv1 mv2 = do
1639 liftIO $ threadDelay (round $ n * 1000000)
1641 r1 <- liftIO $ takeMVar mv1
1643 liftIO $ putMVar mv1 i1
1646 r2 <- liftIO $ takeMVar mv2
1649 Right x -> fmap Right $ step2 x res1
1650 liftIO $ putMVar mv2 res
1653 handleChildException ::
1654 MVar (Either SomeException a) -> SomeException -> IO ()
1655 handleChildException mv2 e = do
1662------------------------------------------------------------------------------
1663-- Windowed classification
1664------------------------------------------------------------------------------
1666-- We divide the stream into windows or chunks in space or time and each window
1667-- can be associated with a key, all events associated with a particular key in
1668-- the window can be folded to a single result. The stream can be split into
1669-- windows by size or by using a split predicate on the elements in the stream.
1670-- For example, when we receive a closing flag, we can close the window.
1672-- A "chunk" is a space window and a "session" is a time window. Are there any
1673-- other better short words to describe them. An alternative is to use
1674-- "swindow" and "twindow". Another word for "session" could be "spell".
1676-- TODO: To mark the position in space or time we can have Indexed or
1677-- TimeStamped types. That can make it easy to deal with the position indices
1680------------------------------------------------------------------------------
1681-- Keyed Sliding Windows
1682------------------------------------------------------------------------------
1685{-# INLINABLE classifySlidingChunks #-}
1686classifySlidingChunks
1687 :: (IsStream t, MonadAsync m, Ord k)
1688 => Int -- ^ window size
1689 -> Int -- ^ window slide
1690 -> Fold m a b -- ^ Fold to be applied to window events
1691 -> t m (k, a, Bool) -- ^ window key, data, close event
1693classifySlidingChunks wsize wslide (Fold step initial extract) str
1696-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
1697-- slide the send window when an ack is received and we slide the receive
1698-- window when a sequence is complete. Sliding is stateful in case of TCP,
1699-- sliding releases the send buffer or makes data available to the user from
1700-- the receive buffer.
1701{-# INLINABLE classifySlidingSessions #-}
1702classifySlidingSessions
1703 :: (IsStream t, MonadAsync m, Ord k)
1704 => Double -- ^ timer tick in seconds
1705 -> Double -- ^ time window size
1706 -> Double -- ^ window slide
1707 -> Fold m a b -- ^ Fold to be applied to window events
1708 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
1710classifySlidingSessions tick interval slide (Fold step initial extract) str
1714------------------------------------------------------------------------------
1715-- Sliding Window Buffers
1716------------------------------------------------------------------------------
1718-- These buffered versions could be faster than concurrent incremental folds of
1719-- all overlapping windows as in many cases we may not need all the values to
1720-- compute the fold, we can just compute the result using the old value and new
1721-- value. However, we may need the buffer once in a while, for example for
1722-- string search we usually compute the hash incrementally but when the hash
1723-- matches the hash of the pattern we need to compare the whole string.
1725-- XXX we should be able to implement sequence based splitting combinators
1726-- using this combinator.
1729-- | Buffer n elements of the input in a ring buffer. When t new elements are
1730-- collected, slide the window to remove the same number of oldest elements,
1731-- insert the new elements, and apply an incremental fold on the sliding
1732-- window, supplying the outgoing elements, the new ring buffer as arguments.
1734 :: (IsStream t, Monad m, Ord a, Storable a)
1735 => Int -- window size
1736 -> Int -- window slide
1737 -> Fold m (Ring a, Array a) b
1740slidingChunkBuffer = undefined
1742-- Buffer n seconds worth of stream elements of the input in a radix tree.
1743-- Every t seconds, remove the items that are older than n seconds, and apply
1744-- an incremental fold on the sliding window, supplying the outgoing elements,
1745-- and the new radix tree buffer as arguments.
1747 :: (IsStream t, Monad m, Ord a, Storable a)
1748 => Int -- window size
1750 -> Fold m (RTree a, Array a) b
1753slidingSessionBuffer = undefined
1756------------------------------------------------------------------------------
1757-- Keyed Session Windows
1758------------------------------------------------------------------------------
1761-- | Keyed variable size space windows. Close the window if we do not receive a
1762-- window event in the next "spaceout" elements.
1763{-# INLINABLE classifyChunksBy #-}
1765 :: (IsStream t, MonadAsync m, Ord k)
1766 => Int -- ^ window spaceout (spread)
1767 -> Bool -- ^ reset the spaceout when a chunk window element is received
1768 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1769 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1771classifyChunksBy spanout reset (Fold step initial extract) str = undefined
1773-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
1774-- received within the chunk size window. The chunk gets closed only if no
1775-- element is received within the chunk window.
1777{-# INLINABLE classifyKeepAliveChunks #-}
1778classifyKeepAliveChunks
1779 :: (IsStream t, MonadAsync m, Ord k)
1780 => Int -- ^ window spaceout (spread)
1781 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1782 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1784classifyKeepAliveChunks spanout = classifyChunksBy spanout True
1787-- | @classifySessionsBy tick timeout reset f stream@ groups together all input
1788-- stream elements that belong to the same session. @timeout@ is the maximum
1789-- lifetime of a session in seconds. All elements belonging to a session are
1790-- purged after this duration. If "reset" is 'Ture' then the timeout is reset
1791-- after every event received in the session. Session duration is measured
1792-- using the timestamp of the first element seen for that session. To detect
1793-- session timeouts, a monotonic event time clock is maintained using the
1794-- timestamps seen in the inputs and a timer with a tick duration specified by
1797-- @session key@ is a key that uniquely identifies the session for the given
1798-- element, @timestamp@ characterizes the time when the input element was
1799-- generated, this is an absolute time measured from some @Epoch@. @session
1800-- close@ is a boolean indicating whether this element marks the closing of the
1801-- session. When an input element with @session close@ set to @True@ is seen
1802-- the session is purged immediately.
1804-- All the input elements belonging to a session are collected using the fold
1805-- @f@. The session key and the fold result are emitted in the output stream
1806-- when the session is purged either via the session close event or via the
1807-- session liftime timeout.
1810{-# INLINABLE classifySessionsBy #-}
1812 :: (IsStream t, MonadAsync m, Ord k)
1813 => Double -- ^ timer tick in seconds
1814 -> Double -- ^ session timeout
1815 -> Bool -- ^ reset the timeout when an event is received
1816 -> Fold m a b -- ^ Fold to be applied to session events
1817 -> t m (k, a, Bool, AbsTime) -- ^ session key, timestamp, close event, data
1819classifySessionsBy tick timeout reset (Fold step initial extract) str =
1820 S.concatMap (\(Tuple4' _ _ _ s) -> s) $ S.scanlM' sstep szero stream
1824 timeoutMs = toRelTime (round (timeout * 1000) :: MilliSecond64)
1825 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
1826 szero = Tuple4' (toAbsTime (0 :: MilliSecond64)) H.empty Map.empty S.nil
1828 -- Got a new stream input element
1829 sstep (Tuple4' evTime hp mp _) (Just (key, a, closing, ts)) =
1830 -- XXX we should use a heap in pinned memory to scale it to a large
1833 -- deleting a key from the heap is expensive, so we never delete a
1834 -- key, we just purge it from the Map and it gets purged from the
1835 -- heap on timeout. We just need an extra lookup in the Map when
1836 -- the key is purged from the heap, that should not be expensive.
1838 -- To detect session inactivity we keep a timestamp of the latest event
1839 -- in the Map along with the fold result. When we purge the session
1840 -- from the heap we match the timestamp in the heap with the timestamp
1841 -- in the Map, if the latest timestamp is newer and has not expired we
1842 -- reinsert the key in the heap.
1844 -- XXX if the key is an Int, we can also use an IntMap for slightly
1845 -- better performance.
1847 let accumulate v = do
1848 Tuple' _ old <- maybe (initial >>= return . Tuple' ts) return v
1850 return $ Tuple' ts new
1853 let (r, mp') = Map.updateLookupWithKey (\_ _ -> Nothing) key mp
1854 Tuple' _ acc <- accumulate r
1856 return $ Tuple4' evTime hp mp' (S.yield (key, res))
1858 let r = Map.lookup key mp
1860 let mp' = Map.insert key acc mp
1864 let expiry = addToAbsTime ts timeoutMs
1865 in H.insert (Entry expiry key) hp
1867 -- Event time is maintained as monotonically increasing
1868 -- time. If we have lagged behind any of the timestamps
1869 -- seen then we increase it to match the latest time seen
1870 -- in the timestamps. We also increase it on timer ticks.
1871 return $ Tuple4' (max evTime ts) hp' mp' S.nil
1873 -- Got a timer tick event
1874 -- XXX can we yield the entries without accumulating them?
1875 sstep (Tuple4' evTime heap sessions _) Nothing = do
1876 (hp', mp', out) <- go heap sessions S.nil
1877 return $ Tuple4' curTime hp' mp' out
1881 curTime = addToAbsTime evTime tickMs
1883 let hres = H.uncons hp
1885 Just (Entry ts key, hp') -> do
1886 let duration = diffAbsTime curTime ts
1887 if duration >= timeoutMs
1889 let (r, mp') = Map.updateLookupWithKey
1890 (\_ _ -> Nothing) key mp
1892 Nothing -> go hp' mp' out
1893 Just (Tuple' latestTS acc) -> do
1894 let dur = diffAbsTime curTime latestTS
1895 if dur >= timeoutMs || not reset
1898 go hp' mp' ((key, sess) `S.cons` out)
1900 -- reset the session timeout
1901 let expiry = addToAbsTime latestTS timeoutMs
1902 hp'' = H.insert (Entry expiry key) hp'
1903 mp'' = Map.insert key (Tuple' latestTS acc) mp'
1905 else return (hp, mp, out)
1906 Nothing -> return (hp, mp, out)
1908 -- merge timer events in the stream
1909 stream = S.map Just str `parallel` S.repeatM timer
1911 liftIO $ threadDelay (round $ tick * 1000000)
1914-- | Like 'classifySessionsOf' but the session is kept alive if an event is
1915-- received within the session window. The session times out and gets closed
1916-- only if no event is received within the specified session window size.
1919{-# INLINABLE classifyKeepAliveSessions #-}
1920classifyKeepAliveSessions
1921 :: (IsStream t, MonadAsync m, Ord k)
1922 => Double -- ^ session inactive timeout
1923 -> Fold m a b -- ^ Fold to be applied to session payload data
1924 -> t m (k, a, Bool, AbsTime) -- ^ session key, data, close flag, timestamp
1926classifyKeepAliveSessions timeout = classifySessionsBy 1 timeout True
1928------------------------------------------------------------------------------
1929-- Keyed tumbling windows
1930------------------------------------------------------------------------------
1932-- Tumbling windows is a special case of sliding windows where the window slide
1933-- is the same as the window size. Or it can be a special case of session
1934-- windows where the reset flag is set to False.
1936-- XXX instead of using the early termination flag in the stream, we can use an
1937-- early terminating fold instead.
1940-- | Split the stream into fixed size chunks of specified size. Within each
1941-- such chunk fold the elements in buckets identified by the keys. A particular
1942-- bucket fold can be terminated early if a closing flag is encountered in an
1943-- element for that key.
1946{-# INLINABLE classifyChunksOf #-}
1948 :: (IsStream t, MonadAsync m, Ord k)
1949 => Int -- ^ window size
1950 -> Fold m a b -- ^ Fold to be applied to window events
1951 -> t m (k, a, Bool) -- ^ window key, data, close event
1953classifyChunksOf wsize = classifyChunksBy wsize False
1956-- | Split the stream into fixed size time windows of specified interval in
1957-- seconds. Within each such window, fold the elements in buckets identified by
1958-- the keys. A particular bucket fold can be terminated early if a closing flag
1959-- is encountered in an element for that key. Once a fold is terminated the key
1960-- and value for that bucket are emitted in the output stream.
1962-- Session @timestamp@ in the input stream is an absolute time from some epoch,
1963-- characterizing the time when the input element was generated. To detect
1964-- session window end, a monotonic event time clock is maintained synced with
1965-- the timestamps with a clock resolution of 1 second.
1968{-# INLINABLE classifySessionsOf #-}
1970 :: (IsStream t, MonadAsync m, Ord k)
1971 => Double -- ^ time window size
1972 -> Fold m a b -- ^ Fold to be applied to window events
1973 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
1975classifySessionsOf 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