@@ -179,6 +179,11 @@ module Streamly.Prelude
179 , foldl1'
179 , foldl1'
180 , foldlM'
180 , foldlM'
182 -- ** Full Folds
187 -- ** Full Folds
183 -- | Folds that are guaranteed to evaluate the whole stream.
188 -- | Folds that are guaranteed to evaluate the whole stream.
@@ -320,6 +325,20 @@ module Streamly.Prelude
320 , scanl1'
325 , scanl1'
321 , scanl1M'
326 , scanl1M'
323 , indexed
342 , indexed
324 , indexedR
343 , indexedR
325 -- , timestamped
344 -- , timestamped
@@ -441,11 +460,115 @@ module Streamly.Prelude
441 -- , interposeBy
460 -- , interposeBy
442 -- , intercalate
461 -- , intercalate
466+ -- , splitAt -- spanN
467+ -- , splitIn -- sessionN
470+ -- , span -- spanWhile
471+ -- , break -- breakBefore
482+ -- | Streams can be split into segments in space or in time. We use the
483+ -- term @chunk@ to refer to a spatial length of the stream (spatial window)
484+ -- and the term @session@ to refer to a length in time (time window).
486+ -- In imperative terms, grouped folding can be considered as a nested loop
487+ -- where we loop over the stream to group elements and then loop over
488+ -- individual groups to fold them to a single value that is yielded in the
500+ -- *** Using Element Separators
501+ -- On == Dropping the separator
506+ -- By == Keeping the separator
510+ , wordsBy -- stripAndCompactBy
512+ -- -- *** Using Sequence Separators
514+ -- , splitOnSuffixSeq
515+ -- , splitOnPrefixSeq
517+ -- Keeping the delimiters
519+ -- , splitBySeqSuffix
520+ -- , splitBySeqPrefix
523+ -- Splitting using multiple sequence separators
525+ -- , splitOnAnySuffixSeq
526+ -- , splitOnAnyPrefixSeq
444 -- ** Distributing
533 -- ** Distributing
447 , Par.tee
536 , Par.tee
539+ -- * Windowed Classification
540+ -- | Split the stream into windows or chunks in space or time. Each window
541+ -- can be associated with a key, all events associated with a particular
542+ -- key in the window can be folded to a single result. The stream is split
543+ -- into windows of specified size, the window can be terminated early if
544+ -- the closing flag is specified in the input stream.
546+ -- The term "chunk" is used for a space window and the term "session" is
547+ -- used for a time window.
549+ -- ** Tumbling Windows
550+ -- | A new window starts after the previous window is finished.
551+ -- , classifyChunksOf
552+ -- , classifySessionsOf
554+ -- ** Keep Alive Windows
555+ -- | The window size is extended if an event arrives within the specified
556+ -- window size. This can represent sessions with idle or inactive timeout.
557+ -- , classifyKeepAliveChunks
558+ -- , classifyKeepAliveSessions
561+ -- ** Sliding Windows
562+ -- | A new window starts after the specified slide from the previous
563+ -- window. Therefore windows can overlap.
564+ , classifySlidingChunks
565+ , classifySlidingSessions
567+ -- ** Sliding Window Buffers
568+ -- , slidingChunkBuffer
569+ -- , slidingSessionBuffer
449 -- ** Containers of Streams
572 -- ** Containers of Streams
450 -- | These are variants of standard 'Foldable' fold functions that use a
573 -- | These are variants of standard 'Foldable' fold functions that use a
451 -- polymorphic stream sum operation (e.g. 'async' or 'wSerial') to fold a
574 -- polymorphic stream sum operation (e.g. 'async' or 'wSerial') to fold a
@@ -495,19 +618,25 @@ import Control.Monad (void)
495import Control.Monad.Catch (MonadCatch)
618import Control.Monad.Catch (MonadCatch)
496import Control.Monad.IO.Class (MonadIO(..))
619import Control.Monad.IO.Class (MonadIO(..))
497import Control.Monad.Trans (MonadTrans(..))
620import Control.Monad.Trans (MonadTrans(..))
621+import Data.Functor.Identity (Identity)
622+import Data.Heap (Entry(..))
498−import Data.Maybe (isJust, fromJust)
623+import Data.Maybe (isJust, fromJust, isNothing)
499import Foreign.Storable (Storable)
624import Foreign.Storable (Storable)
500import Prelude
625import Prelude
501 hiding (filter, drop, dropWhile, take, takeWhile, zipWith, foldr,
626 hiding (filter, drop, dropWhile, take, takeWhile, zipWith, foldr,
502− foldl, mapM, mapM_, sequence, all, any, sum, product, elem,
627+ foldl, map, mapM, mapM_, sequence, all, any, sum, product, elem,
503 notElem, maximum, minimum, head, last, tail, length, null,
628 notElem, maximum, minimum, head, last, tail, length, null,
504 reverse, iterate, init, and, or, lookup, foldr1, (!!),
629 reverse, iterate, init, and, or, lookup, foldr1, (!!),
505− scanl, scanl1, replicate, concatMap)
630+ scanl, scanl1, replicate, concatMap, span)
632+import qualified Data.Heap as H
633+import qualified Data.Map.Strict as Map
507import qualified Prelude
634import qualified Prelude
508import qualified System.IO as IO
635import qualified System.IO as IO
510import Streamly.Enumeration (Enumerable(..), enumerate, enumerateTo)
637import Streamly.Enumeration (Enumerable(..), enumerate, enumerateTo)
638+import Streamly.Fold.Types (Fold (..))
639+import Streamly.Mem.Array (Array)
511import Streamly.SVar (MonadAsync, defState)
640import Streamly.SVar (MonadAsync, defState)
512import Streamly.Streams.Async (mkAsync')
641import Streamly.Streams.Async (mkAsync')
513import Streamly.Streams.Combinators (maxYields)
642import Streamly.Streams.Combinators (maxYields)
@@ -517,8 +646,14 @@ import Streamly.Streams.StreamD (fromStreamD, toStreamD)
517import Streamly.Streams.StreamK (IsStream(..))
646import Streamly.Streams.StreamK (IsStream(..))
518import Streamly.Streams.Serial (SerialT)
647import Streamly.Streams.Serial (SerialT)
519import Streamly.Pipe.Types (Pipe (..))
648import Streamly.Pipe.Types (Pipe (..))
649+import Streamly.Time.Units
650+ (AbsTime, MilliSecond64(..), addToAbsTime, diffAbsTime, toRelTime,
653+import Streamly.Strict
521−import qualified Streamly.Fold.Types as FL
655+import qualified Streamly.Mem.Array as A
656+import qualified Streamly.Fold as FL
522import qualified Streamly.Streams.Prelude as P
657import qualified Streamly.Streams.Prelude as P
523import qualified Streamly.Streams.StreamK as K
658import qualified Streamly.Streams.StreamK as K
524import qualified Streamly.Streams.StreamD as D
659import qualified Streamly.Streams.StreamD as D
@@ -1205,6 +1340,75 @@ foldxM = P.foldlMx'
1205foldlM' :: Monad m => (b -> a -> m b) -> b -> SerialT m a -> m b
1340foldlM' :: Monad m => (b -> a -> m b) -> b -> SerialT m a -> m b
1206foldlM' step begin m = S.foldlM' step begin $ toStreamS m
1341foldlM' step begin m = S.foldlM' step begin $ toStreamS m
1343+------------------------------------------------------------------------------
1345+------------------------------------------------------------------------------
1349+-- We can use the left folds in this module instead of the folds in
1350+-- "Streamly.Prelude". For example the following two ways of folding are
1351+-- equivalent in functionality and performance,
1353+-- >>> FL.foldl' FL.sum (S.enumerateFromTo 1 100)
1355+-- >>> S.sum (S.enumerateFromTo 1 100)
1358+-- However, left folds are push type folds. That means we push the entire input
1359+-- to a fold before we can get the output. Therefore, the performance is
1360+-- equivalent only for full folds like 'sum' and 'length'. For partial folds
1361+-- like 'head' or 'any' the folds in "Streamly.Prelude" may be much more
1362+-- efficient because they are implemented as right folds that terminate as soon
1363+-- as we get the result. Note that when a full fold is composed with a partial
1364+-- fold in parallel the performance is not impacted as we anyway have to
1365+-- consume the whole stream due to the full fold.
1367+-- >>> S.head (1 `S.cons` undefined)
1369+-- >>> FL.foldl' FL.head (1 `S.cons` undefined)
1370+-- *** Exception: Prelude.undefined
1372+-- However, we can wrap the fold in a scan to convert it into a lazy stream of
1373+-- fold steps. We can then terminate the stream whenever we want. For example,
1375+-- >>> S.toList $ S.take 1 $ FL.scanl' FL.head (1 `S.cons` undefined)
1378+-- The following example extracts the input stream up to a point where the
1379+-- running average of elements is no more than 10:
1383+-- $ S.map (fromJust . fst)
1384+-- $ S.takeWhile (\\(_,x) -> x <= 10)
1385+-- $ FL.postscanl' ((,) \<$> FL.last \<*> avg) (S.enumerateFromTo 1.0 100.0)
1388+-- [1.0,2.0,3.0,4.0,5.0,6.0,7.0,8.0,9.0,10.0,11.0,12.0,13.0,14.0,15.0,16.0,17.0,18.0,19.0]
1391+-- | Fold a stream using the supplied monadic fold.
1393+-- >>> S.foldlStream FL.sum (S.enumerateFromTo 1 100)
1397+{-# INLINE runFold #-}
1398+runFold :: Monad m => Fold m a b -> SerialT m a -> m b
1401+------------------------------------------------------------------------------
1403+------------------------------------------------------------------------------
1406+-- | Drain a stream to a 'Sink'.
1407+{-# INLINE runSink #-}
1408+runSink :: Monad m => Sink m a -> SerialT m a -> m ()
1409+runSink = runFold . toFold
1208------------------------------------------------------------------------------
1412------------------------------------------------------------------------------
1209-- Specialized folds
1413-- Specialized folds
1210------------------------------------------------------------------------------
1414------------------------------------------------------------------------------
@@ -1601,6 +1805,34 @@ toHandle h m = go m
1601 yieldk a r = liftIO (IO.hPutStrLn h a) >> go r
1805 yieldk a r = liftIO (IO.hPutStrLn h a) >> go r
1602 in K.foldStream defState yieldk single stop m1
1806 in K.foldStream defState yieldk single stop m1
1808+-- | A fold that buffers its input to a pure stream.
1810+-- /Warning!/ working on large streams accumulated as buffers in memory could
1811+-- be very inefficient, consider using "Streamly.Array" instead.
1814+{-# INLINE _toStream #-}
1815+_toStream :: Monad m => Fold m a (SerialT Identity a)
1816+_toStream = Fold (\f x -> return $ f . (x `K.cons`))
1818+ (return . ($ K.nil))
1820+-- This is more efficient than 'toStream'. toStream is exactly the same as
1821+-- reversing the stream after toStreamRev.
1823+-- | Buffers the input stream to a pure stream in the reverse order of the
1826+-- /Warning!/ working on large streams accumulated as buffers in memory could
1827+-- be very inefficient, consider using "Streamly.Array" instead.
1831+-- xn : ... : x2 : x1 : []
1832+{-# INLINABLE _toStreamRev #-}
1833+_toStreamRev :: Monad m => Fold m a (SerialT Identity a)
1834+_toStreamRev = Fold (\xs x -> return $ x `K.cons` xs) (return K.nil) return
1604------------------------------------------------------------------------------
1836------------------------------------------------------------------------------
1605-- General Transformation
1837-- General Transformation
1606------------------------------------------------------------------------------
1838------------------------------------------------------------------------------
@@ -1751,6 +1983,26 @@ scanl1M' step m = fromStreamD $ D.scanl1M' step $ toStreamD m
1751scanl1' :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> t m a
1983scanl1' :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> t m a
1752scanl1' step m = fromStreamD $ D.scanl1' step $ toStreamD m
1984scanl1' step m = fromStreamD $ D.scanl1' step $ toStreamD m
1986+------------------------------------------------------------------------------
1987+-- Scanning with a Fold
1988+------------------------------------------------------------------------------
1990+-- | Scan a stream using the given monadic fold.
1993+{-# INLINE runScan #-}
1994+runScan :: Monad m => Fold m a b -> SerialT m a -> SerialT m b
1995+runScan (Fold step begin done) = P.scanlMx' step begin done
1997+-- | Postscan a stream using the given monadic fold.
2000+{-# INLINE runPostscan #-}
2001+runPostscan :: Monad m => Fold m a b -> SerialT m a -> SerialT m b
2002+runPostscan (Fold step begin done) = P.postscanlMx' step begin done
1754------------------------------------------------------------------------------
2006------------------------------------------------------------------------------
1755-- Transformation by Filtering
2007-- Transformation by Filtering
1756------------------------------------------------------------------------------
2008------------------------------------------------------------------------------
@@ -2017,11 +2269,11 @@ intersperseBySpan _n _f _xs = undefined
2019-- @since 0.7.0
2271-- @since 0.7.0
2020−{-# INLINE _intersperseByTime #-}
2272+{-# INLINE intersperseByTime #-}
2021−_intersperseByTime
2273+intersperseByTime
2022 :: (IsStream t, MonadAsync m)
2274 :: (IsStream t, MonadAsync m)
2023 => Double -> m a -> t m a -> t m a
2275 => Double -> m a -> t m a -> t m a
2024−_intersperseByTime n f xs = xs `Par.parallelEndByFirst` repeatM timed
2276+intersperseByTime n f xs = xs `Par.parallelEndByFirst` repeatM timed
2025 where timed = liftIO (threadDelay (round $ n * 1000000)) >> f
2277 where timed = liftIO (threadDelay (round $ n * 1000000)) >> f
2027-- | @insertBy cmp elem stream@ inserts @elem@ before the first element in
2279-- | @insertBy cmp elem stream@ inserts @elem@ before the first element in
@@ -2298,6 +2550,791 @@ interposeBy :: (IsStream t, Monad m)
2298intercalate :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
2550intercalate :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
2553+------------------------------------------------------------------------------
2554+-- Grouping/Splitting
2555+------------------------------------------------------------------------------
2557+------------------------------------------------------------------------------
2558+-- Grouping without looking at elements
2559+------------------------------------------------------------------------------
2561+------------------------------------------------------------------------------
2563+------------------------------------------------------------------------------
2566+-- | @splitAt n f1 f2@ composes folds @f1@ and @f2@ such that first @n@
2567+-- elements of its input are consumed by fold @f1@ and the rest of the stream
2568+-- is consumed by fold @f2@.
2570+-- > let splitAt_ n xs = FL.foldl' (FL.splitAt n FL.toList FL.toList) $ S.fromList xs
2572+-- >>> splitAt_ 6 "Hello World!"
2573+-- > ("Hello ","World!")
2575+-- >>> splitAt_ (-1) [1,2,3]
2578+-- >>> splitAt_ 0 [1,2,3]
2581+-- >>> splitAt_ 1 [1,2,3]
2584+-- >>> splitAt_ 3 [1,2,3]
2587+-- >>> splitAt_ 4 [1,2,3]
2590+-- This can be considered as a two-fold version of 'ltake' where we take both
2591+-- the segments instead of discarding the leftover.
2594+{-# INLINE _splitAt #-}
2601+_splitAt n (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
2602+ Fold step initial extract
2604+ initial = Tuple3' <$> return n <*> initialL <*> initialR
2606+ step (Tuple3' i xL xR) input =
2608+ then stepL xL input >>= (\a -> return (Tuple3' (i - 1) a xR))
2609+ else stepR xR input >>= (\b -> return (Tuple3' i xL b))
2611+ extract (Tuple3' _ a b) = (,) <$> extractL a <*> extractR b
2613+------------------------------------------------------------------------------
2615+------------------------------------------------------------------------------
2617+------------------------------------------------------------------------------
2618+-- Generalized grouping
2619+------------------------------------------------------------------------------
2621+-- This combinator is the most general grouping combinator and can be used to
2622+-- implement all other grouping combinators.
2624+-- XXX check if this can implement the splitOn combinator i.e. we can slide in
2625+-- new elements, slide out old elements and incrementally compute the hash.
2626+-- Also, can we implement the windowed classification combinators using this?
2628+-- In fact this is a parse. Instead of using a special return value in the fold
2629+-- we are using a mapping function.
2631+-- Note that 'scanl'' (usually followed by a map to extract the desired value
2632+-- from the accumulator) can be used to realize many implementations e.g. a
2633+-- sliding window implementation. A scan followed by a mapMaybe is also a good
2634+-- pattern to express many problems where we want to emit a filtered output and
2635+-- not emit an output on every input.
2637+-- Passing on of the initial accumulator value to the next fold is equivalent
2638+-- to returning the leftover concept.
2641+-- | @groupScan splitter fold stream@ folds the input stream using @fold@.
2642+-- @splitter@ is applied on the accumulator of the fold every time an item is
2643+-- consumed by the fold. The fold continues until @splitter@ returns a 'Just'
2644+-- value. A 'Just' result from the @splitter@ specifies a result to be emitted
2645+-- in the output stream and the initial value of the accumulator for the next
2646+-- group's fold. This allows us to control whether to start fresh for the next
2647+-- fold or to continue from the previous fold's output.
2649+{-# INLINE groupScan #-}
2651+ :: (IsStream t, Monad m)
2652+ => (x -> m (Maybe (b, x))) -> Fold m a x -> t m a -> t m b
2653+groupScan split fold m = undefined
2656+-- | Group the input stream into groups of @n@ elements each and then fold each
2657+-- group using the provided fold function.
2659+-- >> S.toList $ FL.chunksOf 2 FL.sum (S.enumerateFromTo 1 10)
2662+-- This can be considered as an n-fold version of 'ltake' where we apply
2663+-- 'ltake' repeatedly on the leftover stream until the stream exhausts.
2666+{-# INLINE chunksOf #-}
2668+ :: (IsStream t, Monad m)
2669+ => Int -> Fold m a b -> t m a -> t m b
2670+chunksOf n f m = D.fromStreamD $ D.groupsOf n f (D.toStreamD m)
2672+-- | Transform a fold from a pure input to a 'Maybe' input, consuming only
2674+{-# INLINE lcatMaybes #-}
2675+lcatMaybes :: Monad m => Fold m a b -> Fold m (Maybe a) b
2676+lcatMaybes = FL.lfilter isJust . FL.lmap fromJust
2678+-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
2679+-- XXX add this example after fixing the serial stream rate control
2680+-- >>> S.toList $ S.take 5 $ sessionsOf 1 FL.sum $ constRate 2 $ S.enumerateFrom 1
2683+-- | Group the input stream into windows of @n@ second each and then fold each
2684+-- group using the provided fold function.
2687+{-# INLINE sessionsOf #-}
2689+ :: (IsStream t, MonadAsync m)
2690+ => Double -> Fold m a b -> t m a -> t m b
2692+ splitBySuffix isNothing (lcatMaybes f)
2693+ (intersperseByTime n (return Nothing) (Serial.map Just xs))
2695+------------------------------------------------------------------------------
2696+-- Element Aware APIs
2697+------------------------------------------------------------------------------
2699+------------------------------------------------------------------------------
2701+------------------------------------------------------------------------------
2703+-- | Break the input stream into two groups, the first group takes the input as
2704+-- long as the predicate applied to the first element of the stream and next
2705+-- input element holds 'True', the second group takes the rest of the input.
2709+ => (a -> a -> Bool)
2713+_spanBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
2714+ Fold step initial extract
2717+ initial = Tuple3' <$> initialL <*> initialR <*> return (Tuple' Nothing True)
2719+ step (Tuple3' a b (Tuple' (Just frst) isFirstG)) input =
2720+ if cmp frst input && isFirstG
2722+ >>= (\a' -> return (Tuple3' a' b (Tuple' (Just frst) isFirstG)))
2724+ >>= (\a' -> return (Tuple3' a a' (Tuple' Nothing False)))
2726+ step (Tuple3' a b (Tuple' Nothing isFirstG)) input =
2729+ >>= (\a' -> return (Tuple3' a' b (Tuple' (Just input) isFirstG)))
2731+ >>= (\a' -> return (Tuple3' a a' (Tuple' Nothing False)))
2733+ extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
2735+-- | @span p f1 f2@ composes folds @f1@ and @f2@ such that @f1@ consumes the
2736+-- input as long as the predicate @p@ is 'True'. @f2@ consumes the rest of the
2739+-- > let span_ p xs = FL.foldl' (FL.span p FL.toList FL.toList) $ S.fromList xs
2741+-- >>> span_ (< 1) [1,2,3]
2744+-- >>> span_ (< 2) [1,2,3]
2747+-- >>> span_ (< 4) [1,2,3]
2752+-- This can be considered as a two-fold version of 'ltakeWhile' where we take
2753+-- both the segments instead of discarding the leftover.
2761+span p (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
2762+ Fold step initial extract
2766+ initial = Tuple3' <$> initialL <*> initialR <*> return True
2768+ step (Tuple3' a b isFirstG) input =
2769+ if isFirstG && p input
2770+ then stepL a input >>= (\a' -> return (Tuple3' a' b True))
2771+ else stepR b input >>= (\a' -> return (Tuple3' a a' False))
2773+ extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
2776+-- > break p = span (not . p)
2778+-- Break as soon as the predicate becomes 'True'. @break p f1 f2@ composes
2779+-- folds @f1@ and @f2@ such that @f1@ stops consuming input as soon as the
2780+-- predicate @p@ becomes 'True'. The rest of the input is consumed @f2@.
2782+-- This is the binary version of 'splitBy'.
2784+-- > let break_ p xs = FL.foldl' (FL.break p FL.toList FL.toList) $ S.fromList xs
2786+-- >>> break_ (< 1) [3,2,1]
2789+-- >>> break_ (< 2) [3,2,1]
2792+-- >>> break_ (< 4) [3,2,1]
2796+{-# INLINE _break #-}
2803+_break p = span (not . p)
2805+-- | Like 'spanBy' but applies the predicate in a rolling fashion i.e.
2806+-- predicate is applied to the previous and the next input elements.
2807+{-# INLINE _spanRollingBy #-}
2810+ => (a -> a -> Bool)
2814+_spanRollingBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
2815+ Fold step initial extract
2818+ initial = Tuple3' <$> initialL <*> initialR <*> return Nothing
2820+ step (Tuple3' a b (Just frst)) input =
2822+ then stepL a input >>= (\a' -> return (Tuple3' a' b (Just input)))
2823+ else stepR b input >>= (\b' -> return (Tuple3' a b' (Just input)))
2825+ step (Tuple3' a b Nothing) input =
2826+ stepL a input >>= (\a' -> return (Tuple3' a' b (Just input)))
2828+ extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
2830+------------------------------------------------------------------------------
2832+------------------------------------------------------------------------------
2834+-- | @groupsBy cmp f $ S.fromList [a,b,c,...]@ assigns the element @a@ to the
2835+-- first group, if @a \`cmp` b@ is 'True' then @b@ is also assigned to the same
2836+-- group. If @a \`cmp` c@ is 'True' then @c@ is also assigned to the same
2837+-- group and so on. When the comparison fails a new group is started. Each
2838+-- group is folded using the fold @f@ and the result of the fold is emitted in
2839+-- the output stream.
2841+-- >>> S.toList $ FL.groupsBy (>) FL.toList $ S.fromList [1,3,7,0,2,5]
2842+-- > [[1,3,7],[0,2,5]]
2845+{-# INLINE groupsBy #-}
2847+ :: (IsStream t, Monad m)
2848+ => (a -> a -> Bool)
2852+groupsBy cmp f m = D.fromStreamD $ D.groupsBy cmp f (D.toStreamD m)
2854+-- | Unlike @groupsBy@ this function performs a rolling comparison of two
2855+-- successive elements in the input stream. @groupsByRolling cmp f $ S.fromList
2856+-- [a,b,c,...]@ assigns the element @a@ to the first group, if @a \`cmp` b@ is
2857+-- 'True' then @b@ is also assigned to the same group. If @b \`cmp` c@ is
2858+-- 'True' then @c@ is also assigned to the same group and so on. When the
2859+-- comparison fails a new group is started. Each group is folded using the fold
2862+-- >>> S.toList $ FL.groupsByRolling (\a b -> a + 1 == b) FL.toList $ S.fromList [1,2,3,7,8,9]
2863+-- > [[1,2,3],[7,8,9]]
2866+{-# INLINE groupsByRolling #-}
2868+ :: (IsStream t, Monad m)
2869+ => (a -> a -> Bool)
2873+groupsByRolling cmp f m = D.fromStreamD $ D.groupsRollingBy cmp f (D.toStreamD m)
2876+-- > groups = groupsBy (==)
2877+-- > groups = groupsByRolling (==)
2879+-- Groups contiguous spans of equal elements together in individual groups.
2881+-- >>> S.toList $ FL.groups FL.toList $ S.fromList [1,1,2,2]
2885+groups :: (IsStream t, Monad m, Eq a) => Fold m a b -> t m a -> t m b
2886+groups = groupsBy (==)
2888+------------------------------------------------------------------------------
2889+-- Binary splitting on a separator
2890+------------------------------------------------------------------------------
2893+-- | Find the first occurrence of the specified sequence in the input stream
2894+-- and break the input stream into two parts, the first part consisting of the
2895+-- stream before the sequence and the second part consisting of the sequence
2896+-- and the rest of the stream.
2898+-- > let breakOn_ pat xs = FL.foldl' (FL.breakOn pat FL.toList FL.toList) $ S.fromList xs
2900+-- >>> breakOn_ "dear" "Hello dear world!"
2901+-- > ("Hello ","dear world!")
2903+{-# INLINE breakOn #-}
2904+breakOn :: Monad m => Array a -> Fold m a b -> Fold m a c -> Fold m a (b,c)
2905+breakOn pat f m = undefined
2908+------------------------------------------------------------------------------
2909+-- N-ary split on a predicate
2910+------------------------------------------------------------------------------
2912+-- TODO: Use a Splitter configuration similar to the "split" package to make it
2913+-- possible to express all splitting combinations. In general, we can have
2914+-- infix/suffix/prefix/condensing of separators, dropping both leading/trailing
2915+-- separators. We can have a single split operation taking the splitter config
2918+-- | Split a stream on separator elements determined by a predicate, dropping
2919+-- the separator. Separator is considered as infixed between two segments, if
2920+-- one side of the separator is missing then it is parsed as an empty stream.
2921+-- With '-' representing elements and '.' as separator, 'splitOn' splits as
2925+-- "--.--" => "--" "--"
2930+-- @splitOn (== x)@ is an inverse of @intercalate (S.yield x)@
2932+-- Let's use the following definition for illustration:
2934+-- > splitOn' p xs = S.toList $ FL.splitOn p (FL.toList) (S.fromList xs)
2936+-- >>> splitOn' (== '.') ""
2939+-- >>> splitOn' (== '.') "."
2942+-- >>> splitOn' (== '.') ".a"
2945+-- >>> splitOn' (== '.') "a."
2948+-- >>> splitOn' (== '.') "a.b"
2951+-- >>> splitOn' (== '.') "a..b"
2956+-- This can be considered as an n-fold version of 'breakOn' where we apply
2957+-- 'breakOn' successively on the input stream, dropping the first element
2958+-- of the second segment after each break.
2960+{-# INLINE splitOn #-}
2962+ :: (IsStream t, Monad m)
2963+ => (a -> Bool) -> Fold m a b -> t m a -> t m b
2964+splitOn predicate f m =
2965+ D.fromStreamD $ D.splitBy predicate f (D.toStreamD m)
2967+-- | Like 'splitOn' but the separator is considered as suffixed to the segments
2968+-- in the stream. A missing suffix at the end is allowed. A separator at the
2969+-- beginning is parsed as empty segment. With '-' representing elements and
2970+-- '.' as separator, 'splitOnSuffix' splits as follows:
2973+-- "--.--." => "--" "--"
2974+-- "--.--" => "--" "--"
2975+-- ".--." => "" "--"
2978+-- > splitOnSuffix' p xs = S.toList $ FL.splitSuffixBy p (FL.toList) (S.fromList xs)
2980+-- >>> splitOnSuffix' (== '.') ""
2983+-- >>> splitOnSuffix' (== '.') "."
2986+-- >>> splitOnSuffix' (== '.') "a"
2989+-- >>> splitOnSuffix' (== '.') ".a"
2992+-- >>> splitOnSuffix' (== '.') "a."
2995+-- >>> splitOnSuffix' (== '.') "a.b"
2998+-- >>> splitOnSuffix' (== '.') "a.b."
3001+-- >>> splitOnSuffix' (== '.') "a..b.."
3002+-- > ["a","","b",""]
3004+-- > lines = splitOnSuffix (== '\n')
3008+-- This can be considered as an n-fold version of 'breakPost' where we apply
3009+-- 'breakPost' successively on the input stream, dropping the first element
3010+-- of the second segment after each break.
3012+{-# INLINE splitOnSuffix #-}
3014+ :: (IsStream t, Monad m)
3015+ => (a -> Bool) -> Fold m a b -> t m a -> t m b
3016+splitOnSuffix predicate f m =
3017+ D.fromStreamD $ D.splitSuffixBy predicate f (D.toStreamD m)
3019+-- | Like 'splitBy' but ignores repeated separators or separators in leading
3020+-- or trailing position. Therefore, @"..a..b.."@ would be parsed as
3021+-- @["a","b"]@. In other words, it treats the input like words separated by
3022+-- whitespace elements determined by the predicate.
3024+-- > wordsBy' p xs = S.toList $ FL.wordsBy p (FL.toList) (S.fromList xs)
3026+-- >>> wordsBy' (== ',') ""
3029+-- >>> wordsBy' (== ',') ","
3032+-- >>> wordsBy' (== ',') ",a,,b,"
3035+-- > words = wordsBy isSpace
3038+{-# INLINE wordsBy #-}
3040+ :: (IsStream t, Monad m)
3041+ => (a -> Bool) -> Fold m a b -> t m a -> t m b
3042+wordsBy predicate f m =
3043+ D.fromStreamD $ D.wordsBy predicate f (D.toStreamD m)
3045+-- | Like 'splitOnSuffix' but keeps the suffix attached to the resulting
3048+-- > splitBySuffix' p xs = S.toList $ FL.splitBySuffix p (FL.toList) (S.fromList xs)
3050+-- >>> splitBySuffix' (== '.') ""
3053+-- >>> splitBySuffix' (== '.') "."
3056+-- >>> splitBySuffix' (== '.') "a"
3059+-- >>> splitBySuffix' (== '.') ".a"
3062+-- >>> splitBySuffix' (== '.') "a."
3065+-- >>> splitBySuffix' (== '.') "a.b"
3068+-- >>> splitBySuffix' (== '.') "a.b."
3071+-- >>> splitBySuffix' (== '.') "a..b.."
3072+-- > ["a.",".","b.","."]
3076+-- This can be considered as an n-fold version of 'breakPost' where we apply
3077+-- 'breakPost' successively on the input stream.
3079+{-# INLINE splitBySuffix #-}
3081+ :: (IsStream t, Monad m)
3082+ => (a -> Bool) -> Fold m a b -> t m a -> t m b
3083+splitBySuffix predicate f m =
3084+ D.fromStreamD $ D.splitSuffixBy' predicate f (D.toStreamD m)
3086+------------------------------------------------------------------------------
3087+-- Split on a delimiter sequence
3088+------------------------------------------------------------------------------
3090+-- Int list examples for splitOn:
3092+-- >>> splitList [] [1,2,3,3,4]
3093+-- > [[1],[2],[3],[3],[4]]
3095+-- >>> splitList [5] [1,2,3,3,4]
3098+-- >>> splitList [1] [1,2,3,3,4]
3101+-- >>> splitList [4] [1,2,3,3,4]
3104+-- >>> splitList [2] [1,2,3,3,4]
3107+-- >>> splitList [3] [1,2,3,3,4]
3110+-- >>> splitList [3,3] [1,2,3,3,4]
3113+-- >>> splitList [1,2,3,3,4] [1,2,3,3,4]
3116+-- | Like 'splitOn' but the separator is a sequence of elements instead of a
3119+-- For illustration, let's define a function that operates on pure lists:
3122+-- splitOnSeq' pat xs = S.toList $ FL.splitOnSeq (A.fromList pat) (FL.toList) (S.fromList xs)
3125+-- >>> splitOnSeq' "" "hello"
3126+-- > ["h","e","l","l","o"]
3128+-- >>> splitOnSeq' "hello" ""
3131+-- >>> splitOnSeq' "hello" "hello"
3134+-- >>> splitOnSeq' "x" "hello"
3137+-- >>> splitOnSeq' "h" "hello"
3140+-- >>> splitOnSeq' "o" "hello"
3143+-- >>> splitOnSeq' "e" "hello"
3146+-- >>> splitOnSeq' "l" "hello"
3149+-- >>> splitOnSeq' "ll" "hello"
3152+-- 'splitOnSeq' is an inverse of 'intercalate'. The following law always holds:
3154+-- > intercalate . splitOn == id
3156+-- The following law holds when the separator is non-empty and contains none of
3157+-- the elements present in the input lists:
3159+-- > splitOn . intercalate == id
3163+-- XXX We can use a polymorphic vector implemented by Array# to represent the
3164+-- sequence, that way we can avoid the Storable constraint. If we still need
3165+-- Storable Array for performance, we can use a separate splitOnArray API for
3166+-- that. We can also have an API where the sequence itself is a lazy stream, so
3167+-- that we can search files in files for example.
3168+{-# INLINE splitOnSeq #-}
3170+ :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
3171+ => Array a -> Fold m a b -> t m a -> t m b
3172+splitOnSeq patt f m = D.fromStreamD $ D.splitOn patt f (D.toStreamD m)
3175+-- This can be implemented easily using Rabin Karp
3176+-- | Split on any one of the given patterns.
3177+{-# INLINE splitOnAny #-}
3179+ :: (IsStream t, Monad m, Storable a, Integral a)
3180+ => [Array a] -> Fold m a b -> t m a -> t m b
3181+splitOnAny subseq f m = undefined -- D.fromStreamD $ D.splitOnAny f subseq (D.toStreamD m)
3184+-- | Like 'splitSuffixBy' but the separator is a sequence of elements, instead
3185+-- of a predicate for a single element.
3187+-- > splitSuffixOn_ pat xs = S.toList $ FL.splitSuffixOn (A.fromList pat) (FL.toList) (S.fromList xs)
3189+-- >>> splitSuffixOn_ "." ""
3192+-- >>> splitSuffixOn_ "." "."
3195+-- >>> splitSuffixOn_ "." "a"
3198+-- >>> splitSuffixOn_ "." ".a"
3201+-- >>> splitSuffixOn_ "." "a."
3204+-- >>> splitSuffixOn_ "." "a.b"
3207+-- >>> splitSuffixOn_ "." "a.b."
3210+-- >>> splitSuffixOn_ "." "a..b.."
3211+-- > ["a","","b",""]
3213+-- > lines = splitSuffixOn "\n"
3216+{-# INLINE _splitOnSuffixSeq #-}
3218+ :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
3219+ => Array a -> Fold m a b -> t m a -> t m b
3220+_splitOnSuffixSeq patt f m =
3221+ D.fromStreamD $ D.splitSuffixOn False patt f (D.toStreamD m)
3224+-- | Like 'splitOn' but drops any empty splits.
3226+{-# INLINE wordsOn #-}
3228+ :: (IsStream t, Monad m, Storable a, Eq a)
3229+ => Array a -> Fold m a b -> t m a -> t m b
3230+wordsOn subseq f m = undefined -- D.fromStreamD $ D.wordsOn f subseq (D.toStreamD m)
3233+-- XXX use a non-monadic intersperse to remove the MonadAsync constraint.
3235+-- | Like 'splitOnSeq' but splits the separator as well, as an infix token.
3237+-- > splitOn'_ pat xs = S.toList $ FL.splitOn' (A.fromList pat) (FL.toList) (S.fromList xs)
3239+-- >>> splitOn'_ "" "hello"
3240+-- > ["h","","e","","l","","l","","o"]
3242+-- >>> splitOn'_ "hello" ""
3245+-- >>> splitOn'_ "hello" "hello"
3246+-- > ["","hello",""]
3248+-- >>> splitOn'_ "x" "hello"
3251+-- >>> splitOn'_ "h" "hello"
3252+-- > ["","h","ello"]
3254+-- >>> splitOn'_ "o" "hello"
3255+-- > ["hell","o",""]
3257+-- >>> splitOn'_ "e" "hello"
3258+-- > ["h","e","llo"]
3260+-- >>> splitOn'_ "l" "hello"
3261+-- > ["he","l","","l","o"]
3263+-- >>> splitOn'_ "ll" "hello"
3264+-- > ["he","ll","o"]
3267+{-# INLINE _splitBySeq #-}
3269+ :: (IsStream t, MonadAsync m, Storable a, Enum a, Eq a)
3270+ => Array a -> Fold m a b -> t m a -> t m b
3271+_splitBySeq patt f m = intersperseM (runFold f (A.read patt)) $ splitOnSeq patt f m
3273+-- | Like 'splitSuffixOn' but keeps the suffix intact in the splits.
3275+-- > splitSuffixOn'_ pat xs = S.toList $ FL.splitSuffixOn' (A.fromList pat) (FL.toList) (S.fromList xs)
3277+-- >>> splitSuffixOn'_ "." ""
3280+-- >>> splitSuffixOn'_ "." "."
3283+-- >>> splitSuffixOn'_ "." "a"
3286+-- >>> splitSuffixOn'_ "." ".a"
3289+-- >>> splitSuffixOn'_ "." "a."
3292+-- >>> splitSuffixOn'_ "." "a.b"
3295+-- >>> splitSuffixOn'_ "." "a.b."
3298+-- >>> splitSuffixOn'_ "." "a..b.."
3299+-- > ["a.",".","b.","."]
3302+{-# INLINE _splitBySuffixSeq #-}
3304+ :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
3305+ => Array a -> Fold m a b -> t m a -> t m b
3306+_splitBySuffixSeq patt f m =
3307+ D.fromStreamD $ D.splitSuffixOn True patt f (D.toStreamD m)
3310+-- This can be implemented easily using Rabin Karp
3311+-- | Split post any one of the given patterns.
3312+{-# INLINE splitSuffixOnAny #-}
3314+ :: (IsStream t, Monad m, Storable a, Integral a)
3315+ => [Array a] -> Fold m a b -> t m a -> t m b
3316+splitSuffixOnAny subseq f m = undefined
3317+ -- D.fromStreamD $ D.splitPostAny f subseq (D.toStreamD m)
3320+------------------------------------------------------------------------------
3321+-- Reorder in sequence
3322+------------------------------------------------------------------------------
3325+-- Buffer until the next element in sequence arrives. The function argument
3326+-- determines the difference in sequence numbers. This could be useful in
3327+-- implementing sequenced streams, for example, TCP reassembly.
3328+{-# INLINE reassembleBy #-}
3330+ :: (IsStream t, Monad m)
3335+reassembleBy = undefined
2301------------------------------------------------------------------------------
3338------------------------------------------------------------------------------
2302-- Distributing
3339-- Distributing
2303------------------------------------------------------------------------------
3340------------------------------------------------------------------------------
@@ -2340,6 +3377,320 @@ tap f xs = D.fromStreamD $ D.tap f (D.toStreamD xs)
2340trace :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m a
3377trace :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m a
2341trace f = mapM (\x -> void (f x) >> return x)
3378trace f = mapM (\x -> void (f x) >> return x)
3380+------------------------------------------------------------------------------
3381+-- Windowed classification
3382+------------------------------------------------------------------------------
3384+-- We divide the stream into windows or chunks in space or time and each window
3385+-- can be associated with a key, all events associated with a particular key in
3386+-- the window can be folded to a single result. The stream can be split into
3387+-- windows by size or by using a split predicate on the elements in the stream.
3388+-- For example, when we receive a closing flag, we can close the window.
3390+-- A "chunk" is a space window and a "session" is a time window. Are there any
3391+-- other better short words to describe them. An alternative is to use
3392+-- "swindow" and "twindow". Another word for "session" could be "spell".
3394+-- TODO: To mark the position in space or time we can have Indexed or
3395+-- TimeStamped types. That can make it easy to deal with the position indices
3398+------------------------------------------------------------------------------
3399+-- Keyed Sliding Windows
3400+------------------------------------------------------------------------------
3403+{-# INLINABLE classifySlidingChunks #-}
3404+classifySlidingChunks
3405+ :: (IsStream t, MonadAsync m, Ord k)
3406+ => Int -- ^ window size
3407+ -> Int -- ^ window slide
3408+ -> Fold m a b -- ^ Fold to be applied to window events
3409+ -> t m (k, a, Bool) -- ^ window key, data, close event
3411+classifySlidingChunks wsize wslide (Fold step initial extract) str
3414+-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
3415+-- slide the send window when an ack is received and we slide the receive
3416+-- window when a sequence is complete. Sliding is stateful in case of TCP,
3417+-- sliding releases the send buffer or makes data available to the user from
3418+-- the receive buffer.
3419+{-# INLINABLE classifySlidingSessions #-}
3420+classifySlidingSessions
3421+ :: (IsStream t, MonadAsync m, Ord k)
3422+ => Double -- ^ timer tick in seconds
3423+ -> Double -- ^ time window size
3424+ -> Double -- ^ window slide
3425+ -> Fold m a b -- ^ Fold to be applied to window events
3426+ -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
3428+classifySlidingSessions tick interval slide (Fold step initial extract) str
3432+------------------------------------------------------------------------------
3433+-- Sliding Window Buffers
3434+------------------------------------------------------------------------------
3436+-- These buffered versions could be faster than concurrent incremental folds of
3437+-- all overlapping windows as in many cases we may not need all the values to
3438+-- compute the fold, we can just compute the result using the old value and new
3439+-- value. However, we may need the buffer once in a while, for example for
3440+-- string search we usually compute the hash incrementally but when the hash
3441+-- matches the hash of the pattern we need to compare the whole string.
3443+-- XXX we should be able to implement sequence based splitting combinators
3444+-- using this combinator.
3447+-- | Buffer n elements of the input in a ring buffer. When t new elements are
3448+-- collected, slide the window to remove the same number of oldest elements,
3449+-- insert the new elements, and apply an incremental fold on the sliding
3450+-- window, supplying the outgoing elements, the new ring buffer as arguments.
3452+ :: (IsStream t, Monad m, Ord a, Storable a)
3453+ => Int -- window size
3454+ -> Int -- window slide
3455+ -> Fold m (Ring a, Array a) b
3458+slidingChunkBuffer = undefined
3460+-- Buffer n seconds worth of stream elements of the input in a radix tree.
3461+-- Every t seconds, remove the items that are older than n seconds, and apply
3462+-- an incremental fold on the sliding window, supplying the outgoing elements,
3463+-- and the new radix tree buffer as arguments.
3464+slidingSessionBuffer
3465+ :: (IsStream t, Monad m, Ord a, Storable a)
3466+ => Int -- window size
3467+ -> Int -- tick size
3468+ -> Fold m (RTree a, Array a) b
3471+slidingSessionBuffer = undefined
3474+------------------------------------------------------------------------------
3475+-- Keyed Session Windows
3476+------------------------------------------------------------------------------
3479+-- | Keyed variable size space windows. Close the window if we do not receive a
3480+-- window event in the next "spaceout" elements.
3481+{-# INLINABLE classifyChunksBy #-}
3483+ :: (IsStream t, MonadAsync m, Ord k)
3484+ => Int -- ^ window spaceout (spread)
3485+ -> Bool -- ^ reset the spaceout when a chunk window element is received
3486+ -> Fold m a b -- ^ Fold to be applied to chunk window elements
3487+ -> t m (k, a, Bool) -- ^ chunk key, data, last element
3489+classifyChunksBy spanout reset (Fold step initial extract) str = undefined
3491+-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
3492+-- received within the chunk size window. The chunk gets closed only if no
3493+-- element is received within the chunk window.
3495+{-# INLINABLE classifyKeepAliveChunks #-}
3496+classifyKeepAliveChunks
3497+ :: (IsStream t, MonadAsync m, Ord k)
3498+ => Int -- ^ window spaceout (spread)
3499+ -> Fold m a b -- ^ Fold to be applied to chunk window elements
3500+ -> t m (k, a, Bool) -- ^ chunk key, data, last element
3502+classifyKeepAliveChunks spanout = classifyChunksBy spanout True
3505+-- | @classifySessionsBy tick timeout reset f stream@ groups together all input
3506+-- stream elements that belong to the same session. @timeout@ is the maximum
3507+-- lifetime of a session in seconds. All elements belonging to a session are
3508+-- purged after this duration. If "reset" is 'Ture' then the timeout is reset
3509+-- after every event received in the session. Session duration is measured
3510+-- using the timestamp of the first element seen for that session. To detect
3511+-- session timeouts, a monotonic event time clock is maintained using the
3512+-- timestamps seen in the inputs and a timer with a tick duration specified by
3515+-- @session key@ is a key that uniquely identifies the session for the given
3516+-- element, @timestamp@ characterizes the time when the input element was
3517+-- generated, this is an absolute time measured from some @Epoch@. @session
3518+-- close@ is a boolean indicating whether this element marks the closing of the
3519+-- session. When an input element with @session close@ set to @True@ is seen
3520+-- the session is purged immediately.
3522+-- All the input elements belonging to a session are collected using the fold
3523+-- @f@. The session key and the fold result are emitted in the output stream
3524+-- when the session is purged either via the session close event or via the
3525+-- session liftime timeout.
3528+{-# INLINABLE classifySessionsBy #-}
3530+ :: (IsStream t, MonadAsync m, Ord k)
3531+ => Double -- ^ timer tick in seconds
3532+ -> Double -- ^ session timeout
3533+ -> Bool -- ^ reset the timeout when an event is received
3534+ -> Fold m a b -- ^ Fold to be applied to session events
3535+ -> t m (k, a, Bool, AbsTime) -- ^ session key, timestamp, close event, data
3537+classifySessionsBy tick timeout reset (Fold step initial extract) str =
3538+ concatMap (\(Tuple4' _ _ _ s) -> s) $ scanlM' sstep szero stream
3542+ timeoutMs = toRelTime (round (timeout * 1000) :: MilliSecond64)
3543+ tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
3544+ szero = Tuple4' (toAbsTime (0 :: MilliSecond64)) H.empty Map.empty K.nil
3546+ -- Got a new stream input element
3547+ sstep (Tuple4' evTime hp mp _) (Just (key, a, closing, ts)) =
3548+ -- XXX we should use a heap in pinned memory to scale it to a large
3551+ -- deleting a key from the heap is expensive, so we never delete a
3552+ -- key, we just purge it from the Map and it gets purged from the
3553+ -- heap on timeout. We just need an extra lookup in the Map when
3554+ -- the key is purged from the heap, that should not be expensive.
3556+ -- To detect session inactivity we keep a timestamp of the latest event
3557+ -- in the Map along with the fold result. When we purge the session
3558+ -- from the heap we match the timestamp in the heap with the timestamp
3559+ -- in the Map, if the latest timestamp is newer and has not expired we
3560+ -- reinsert the key in the heap.
3562+ -- XXX if the key is an Int, we can also use an IntMap for slightly
3563+ -- better performance.
3565+ let accumulate v = do
3566+ Tuple' _ old <- maybe (initial >>= return . Tuple' ts) return v
3568+ return $ Tuple' ts new
3571+ let (r, mp') = Map.updateLookupWithKey (\_ _ -> Nothing) key mp
3572+ Tuple' _ acc <- accumulate r
3574+ return $ Tuple4' evTime hp mp' (yield (key, res))
3576+ let r = Map.lookup key mp
3577+ acc <- accumulate r
3578+ let mp' = Map.insert key acc mp
3582+ let expiry = addToAbsTime ts timeoutMs
3583+ in H.insert (Entry expiry key) hp
3585+ -- Event time is maintained as monotonically increasing
3586+ -- time. If we have lagged behind any of the timestamps
3587+ -- seen then we increase it to match the latest time seen
3588+ -- in the timestamps. We also increase it on timer ticks.
3589+ return $ Tuple4' (max evTime ts) hp' mp' K.nil
3591+ -- Got a timer tick event
3592+ -- XXX can we yield the entries without accumulating them?
3593+ sstep (Tuple4' evTime heap sessions _) Nothing = do
3594+ (hp', mp', out) <- go heap sessions K.nil
3595+ return $ Tuple4' curTime hp' mp' out
3599+ curTime = addToAbsTime evTime tickMs
3601+ let hres = H.uncons hp
3603+ Just (Entry ts key, hp') -> do
3604+ let duration = diffAbsTime curTime ts
3605+ if duration >= timeoutMs
3607+ let (r, mp') = Map.updateLookupWithKey
3608+ (\_ _ -> Nothing) key mp
3610+ Nothing -> go hp' mp' out
3611+ Just (Tuple' latestTS acc) -> do
3612+ let dur = diffAbsTime curTime latestTS
3613+ if dur >= timeoutMs || not reset
3615+ sess <- extract acc
3616+ go hp' mp' ((key, sess) `K.cons` out)
3618+ -- reset the session timeout
3619+ let expiry = addToAbsTime latestTS timeoutMs
3620+ hp'' = H.insert (Entry expiry key) hp'
3621+ mp'' = Map.insert key (Tuple' latestTS acc) mp'
3622+ in go hp'' mp'' out
3623+ else return (hp, mp, out)
3624+ Nothing -> return (hp, mp, out)
3626+ -- merge timer events in the stream
3627+ stream = Serial.map Just str `Par.parallel` repeatM timer
3629+ liftIO $ threadDelay (round $ tick * 1000000)
3632+-- | Like 'classifySessionsOf' but the session is kept alive if an event is
3633+-- received within the session window. The session times out and gets closed
3634+-- only if no event is received within the specified session window size.
3637+{-# INLINABLE _classifyKeepAliveSessions #-}
3638+_classifyKeepAliveSessions
3639+ :: (IsStream t, MonadAsync m, Ord k)
3640+ => Double -- ^ session inactive timeout
3641+ -> Fold m a b -- ^ Fold to be applied to session payload data
3642+ -> t m (k, a, Bool, AbsTime) -- ^ session key, data, close flag, timestamp
3644+_classifyKeepAliveSessions timeout = classifySessionsBy 1 timeout True
3646+------------------------------------------------------------------------------
3647+-- Keyed tumbling windows
3648+------------------------------------------------------------------------------
3650+-- Tumbling windows is a special case of sliding windows where the window slide
3651+-- is the same as the window size. Or it can be a special case of session
3652+-- windows where the reset flag is set to False.
3654+-- XXX instead of using the early termination flag in the stream, we can use an
3655+-- early terminating fold instead.
3658+-- | Split the stream into fixed size chunks of specified size. Within each
3659+-- such chunk fold the elements in buckets identified by the keys. A particular
3660+-- bucket fold can be terminated early if a closing flag is encountered in an
3661+-- element for that key.
3664+{-# INLINABLE classifyChunksOf #-}
3666+ :: (IsStream t, MonadAsync m, Ord k)
3667+ => Int -- ^ window size
3668+ -> Fold m a b -- ^ Fold to be applied to window events
3669+ -> t m (k, a, Bool) -- ^ window key, data, close event
3671+classifyChunksOf wsize = classifyChunksBy wsize False
3674+-- | Split the stream into fixed size time windows of specified interval in
3675+-- seconds. Within each such window, fold the elements in buckets identified by
3676+-- the keys. A particular bucket fold can be terminated early if a closing flag
3677+-- is encountered in an element for that key. Once a fold is terminated the key
3678+-- and value for that bucket are emitted in the output stream.
3680+-- Session @timestamp@ in the input stream is an absolute time from some epoch,
3681+-- characterizing the time when the input element was generated. To detect
3682+-- session window end, a monotonic event time clock is maintained synced with
3683+-- the timestamps with a clock resolution of 1 second.
3686+{-# INLINABLE _classifySessionsOf #-}
3688+ :: (IsStream t, MonadAsync m, Ord k)
3689+ => Double -- ^ time window size
3690+ -> Fold m a b -- ^ Fold to be applied to window events
3691+ -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
3693+_classifySessionsOf interval = classifySessionsBy 1 interval False
2343------------------------------------------------------------------------------
3694------------------------------------------------------------------------------
2344-- Exceptions
3695-- Exceptions
2345------------------------------------------------------------------------------
3696------------------------------------------------------------------------------