@@ -179,6 +179,11 @@ module Streamly.Prelude
179179 , foldl1'
180180 , foldlM'
181181
182+ -- * Running Folds
183+ -- $runningfolds
184+
185+ , runFold
186+
182187 -- ** Full Folds
183188 -- | Folds that are guaranteed to evaluate the whole stream.
184189
@@ -320,6 +325,20 @@ module Streamly.Prelude
320325 , scanl1'
321326 , scanl1M'
322327
328+ -- ** Running Scans
329+ , runScan
330+ , runPostscan
331+
332+ -- , lscanl'
333+ -- , lscanlM'
334+ -- , lscanl1'
335+ -- , lscanl1M'
336+ --
337+ -- , lpostscanl'
338+ -- , lpostscanlM'
339+ -- , lprescanl'
340+ -- , lprescanlM'
341+
323342 , indexed
324343 , indexedR
325344 -- , timestamped
@@ -441,11 +460,115 @@ module Streamly.Prelude
441460 -- , interposeBy
442461 -- , intercalate
443462
463+ -- ** Breaking
464+
465+ -- By chunks
466+ -- , splitAt -- spanN
467+ -- , splitIn -- sessionN
468+
469+ -- By elements
470+ -- , span -- spanWhile
471+ -- , break -- breakBefore
472+ -- , breakAfter
473+ -- , breakOn
474+ -- , breakAround
475+ -- , spanBy
476+ -- , spanByRolling
477+
478+ -- By sequences
479+ -- breakOnSeq
480+
481+ -- ** Splitting
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).
485+
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
489+ -- output stream.
490+
491+ -- , groupScan
492+
493+ -- *** Chunks
494+ , chunksOf
495+ , sessionsOf
496+
497+ -- , lchunksOf
498+ -- , lsessionsOf
499+
500+ -- *** Using Element Separators
501+ -- On == Dropping the separator
502+ , splitOn
503+ , splitOnSuffix
504+ -- , splitOnPrefix
505+
506+ -- By == Keeping the separator
507+ -- , splitBy
508+ , splitBySuffix
509+ -- , splitByPrefix
510+ , wordsBy -- stripAndCompactBy
511+
512+ -- -- *** Using Sequence Separators
513+ -- , splitOnSeq
514+ -- , splitOnSuffixSeq
515+ -- , splitOnPrefixSeq
516+
517+ -- Keeping the delimiters
518+ -- , splitBySeq
519+ -- , splitBySeqSuffix
520+ -- , splitBySeqPrefix
521+ -- , wordsBySeq
522+
523+ -- Splitting using multiple sequence separators
524+ -- , splitOnAnySeq
525+ -- , splitOnAnySuffixSeq
526+ -- , splitOnAnyPrefixSeq
527+
528+ -- ** Grouping
529+ , groups
530+ , groupsBy
531+ , groupsByRolling
532+
444533 -- ** Distributing
445534 , trace
446535 , tap
447536 , Par.tee
448537
538+ {-
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.
545+ --
546+ -- The term "chunk" is used for a space window and the term "session" is
547+ -- used for a time window.
548+
549+ -- ** Tumbling Windows
550+ -- | A new window starts after the previous window is finished.
551+ -- , classifyChunksOf
552+ -- , classifySessionsOf
553+
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
559+
560+ {-
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
566+ -}
567+ -- ** Sliding Window Buffers
568+ -- , slidingChunkBuffer
569+ -- , slidingSessionBuffer
570+ -}
571+
449572 -- ** Containers of Streams
450573 -- | These are variants of standard 'Foldable' fold functions that use a
451574 -- polymorphic stream sum operation (e.g. 'async' or 'wSerial') to fold a
@@ -495,19 +618,25 @@ import Control.Monad (void)
495618import Control.Monad.Catch (MonadCatch)
496619import Control.Monad.IO.Class (MonadIO(..))
497620import Control.Monad.Trans (MonadTrans(..))
498−import Data.Maybe (isJust, fromJust)
621+import Data.Functor.Identity (Identity)
622+import Data.Heap (Entry(..))
623+import Data.Maybe (isJust, fromJust, isNothing)
499624import Foreign.Storable (Storable)
500625import Prelude
501626 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,
503628 notElem, maximum, minimum, head, last, tail, length, null,
504629 reverse, iterate, init, and, or, lookup, foldr1, (!!),
505− scanl, scanl1, replicate, concatMap)
630+ scanl, scanl1, replicate, concatMap, span)
506631
632+import qualified Data.Heap as H
633+import qualified Data.Map.Strict as Map
507634import qualified Prelude
508635import qualified System.IO as IO
509636
510637import Streamly.Enumeration (Enumerable(..), enumerate, enumerateTo)
638+import Streamly.Fold.Types (Fold (..))
639+import Streamly.Mem.Array (Array)
511640import Streamly.SVar (MonadAsync, defState)
512641import Streamly.Streams.Async (mkAsync')
513642import Streamly.Streams.Combinators (maxYields)
@@ -517,8 +646,14 @@ import Streamly.Streams.StreamD (fromStreamD, toStreamD)
517646import Streamly.Streams.StreamK (IsStream(..))
518647import Streamly.Streams.Serial (SerialT)
519648import Streamly.Pipe.Types (Pipe (..))
649+import Streamly.Time.Units
650+ (AbsTime, MilliSecond64(..), addToAbsTime, diffAbsTime, toRelTime,
651+ toAbsTime)
652+
653+import Streamly.Strict
520654
521−import qualified Streamly.Fold.Types as FL
655+import qualified Streamly.Mem.Array as A
656+import qualified Streamly.Fold as FL
522657import qualified Streamly.Streams.Prelude as P
523658import qualified Streamly.Streams.StreamK as K
524659import qualified Streamly.Streams.StreamD as D
@@ -1205,6 +1340,75 @@ foldxM = P.foldlMx'
12051340foldlM' :: Monad m => (b -> a -> m b) -> b -> SerialT m a -> m b
12061341foldlM' step begin m = S.foldlM' step begin $ toStreamS m
12071342
1343+------------------------------------------------------------------------------
1344+-- Running a Fold
1345+------------------------------------------------------------------------------
1346+
1347+-- $runningfolds
1348+--
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,
1352+--
1353+-- >>> FL.foldl' FL.sum (S.enumerateFromTo 1 100)
1354+-- 5050
1355+-- >>> S.sum (S.enumerateFromTo 1 100)
1356+-- 5050
1357+--
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.
1366+--
1367+-- >>> S.head (1 `S.cons` undefined)
1368+-- Just 1
1369+-- >>> FL.foldl' FL.head (1 `S.cons` undefined)
1370+-- *** Exception: Prelude.undefined
1371+--
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,
1374+--
1375+-- >>> S.toList $ S.take 1 $ FL.scanl' FL.head (1 `S.cons` undefined)
1376+-- [Nothing]
1377+--
1378+-- The following example extracts the input stream up to a point where the
1379+-- running average of elements is no more than 10:
1380+--
1381+-- @
1382+-- > S.toList
1383+-- $ S.map (fromJust . fst)
1384+-- $ S.takeWhile (\\(_,x) -> x <= 10)
1385+-- $ FL.postscanl' ((,) \<$> FL.last \<*> avg) (S.enumerateFromTo 1.0 100.0)
1386+-- @
1387+-- @
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]
1389+-- @
1390+
1391+-- | Fold a stream using the supplied monadic fold.
1392+--
1393+-- >>> S.foldlStream FL.sum (S.enumerateFromTo 1 100)
1394+-- 5050
1395+--
1396+-- @since 0.7.0
1397+{-# INLINE runFold #-}
1398+runFold :: Monad m => Fold m a b -> SerialT m a -> m b
1399+runFold = P.runFold
1400+
1401+------------------------------------------------------------------------------
1402+-- Running a sink
1403+------------------------------------------------------------------------------
1404+
1405+{-
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
1410+-}
1411+
12081412------------------------------------------------------------------------------
12091413-- Specialized folds
12101414------------------------------------------------------------------------------
@@ -1601,6 +1805,34 @@ toHandle h m = go m
16011805 yieldk a r = liftIO (IO.hPutStrLn h a) >> go r
16021806 in K.foldStream defState yieldk single stop m1
16031807
1808+-- | A fold that buffers its input to a pure stream.
1809+--
1810+-- /Warning!/ working on large streams accumulated as buffers in memory could
1811+-- be very inefficient, consider using "Streamly.Array" instead.
1812+--
1813+-- @since 0.7.0
1814+{-# INLINE _toStream #-}
1815+_toStream :: Monad m => Fold m a (SerialT Identity a)
1816+_toStream = Fold (\f x -> return $ f . (x `K.cons`))
1817+ (return id)
1818+ (return . ($ K.nil))
1819+
1820+-- This is more efficient than 'toStream'. toStream is exactly the same as
1821+-- reversing the stream after toStreamRev.
1822+--
1823+-- | Buffers the input stream to a pure stream in the reverse order of the
1824+-- input.
1825+--
1826+-- /Warning!/ working on large streams accumulated as buffers in memory could
1827+-- be very inefficient, consider using "Streamly.Array" instead.
1828+--
1829+-- @since 0.7.0
1830+
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
1835+
16041836------------------------------------------------------------------------------
16051837-- General Transformation
16061838------------------------------------------------------------------------------
@@ -1751,6 +1983,26 @@ scanl1M' step m = fromStreamD $ D.scanl1M' step $ toStreamD m
17511983scanl1' :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> t m a
17521984scanl1' step m = fromStreamD $ D.scanl1' step $ toStreamD m
17531985
1986+------------------------------------------------------------------------------
1987+-- Scanning with a Fold
1988+------------------------------------------------------------------------------
1989+
1990+-- | Scan a stream using the given monadic fold.
1991+--
1992+-- @since 0.7.0
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
1996+
1997+-- | Postscan a stream using the given monadic fold.
1998+--
1999+-- @since 0.7.0
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
2003+
2004+-- XXX runPrescan
2005+
17542006------------------------------------------------------------------------------
17552007-- Transformation by Filtering
17562008------------------------------------------------------------------------------
@@ -2017,11 +2269,11 @@ intersperseBySpan _n _f _xs = undefined
20172269-- @
20182270--
20192271-- @since 0.7.0
2020−{-# INLINE _intersperseByTime #-}
2021−_intersperseByTime
2272+{-# INLINE intersperseByTime #-}
2273+intersperseByTime
20222274 :: (IsStream t, MonadAsync m)
20232275 => 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
20252277 where timed = liftIO (threadDelay (round $ n * 1000000)) >> f
20262278
20272279-- | @insertBy cmp elem stream@ inserts @elem@ before the first element in
@@ -2298,6 +2550,791 @@ interposeBy :: (IsStream t, Monad m)
22982550intercalate :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
22992551-}
23002552
2553+------------------------------------------------------------------------------
2554+-- Grouping/Splitting
2555+------------------------------------------------------------------------------
2556+
2557+------------------------------------------------------------------------------
2558+-- Grouping without looking at elements
2559+------------------------------------------------------------------------------
2560+--
2561+------------------------------------------------------------------------------
2562+-- Binary APIs
2563+------------------------------------------------------------------------------
2564+--
2565+
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@.
2569+--
2570+-- > let splitAt_ n xs = FL.foldl' (FL.splitAt n FL.toList FL.toList) $ S.fromList xs
2571+--
2572+-- >>> splitAt_ 6 "Hello World!"
2573+-- > ("Hello ","World!")
2574+--
2575+-- >>> splitAt_ (-1) [1,2,3]
2576+-- > ([],[1,2,3])
2577+--
2578+-- >>> splitAt_ 0 [1,2,3]
2579+-- > ([],[1,2,3])
2580+--
2581+-- >>> splitAt_ 1 [1,2,3]
2582+-- > ([1],[2,3])
2583+--
2584+-- >>> splitAt_ 3 [1,2,3]
2585+-- > ([1,2,3],[])
2586+--
2587+-- >>> splitAt_ 4 [1,2,3]
2588+-- > ([1,2,3],[])
2589+--
2590+-- This can be considered as a two-fold version of 'ltake' where we take both
2591+-- the segments instead of discarding the leftover.
2592+--
2593+-- @since 0.7.0
2594+{-# INLINE _splitAt #-}
2595+_splitAt
2596+ :: Monad m
2597+ => Int
2598+ -> Fold m a b
2599+ -> Fold m a c
2600+ -> Fold m a (b, c)
2601+_splitAt n (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
2602+ Fold step initial extract
2603+ where
2604+ initial = Tuple3' <$> return n <*> initialL <*> initialR
2605+
2606+ step (Tuple3' i xL xR) input =
2607+ if i > 0
2608+ then stepL xL input >>= (\a -> return (Tuple3' (i - 1) a xR))
2609+ else stepR xR input >>= (\b -> return (Tuple3' i xL b))
2610+
2611+ extract (Tuple3' _ a b) = (,) <$> extractL a <*> extractR b
2612+
2613+------------------------------------------------------------------------------
2614+-- N-ary APIs
2615+------------------------------------------------------------------------------
2616+
2617+------------------------------------------------------------------------------
2618+-- Generalized grouping
2619+------------------------------------------------------------------------------
2620+
2621+-- This combinator is the most general grouping combinator and can be used to
2622+-- implement all other grouping combinators.
2623+--
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?
2627+--
2628+-- In fact this is a parse. Instead of using a special return value in the fold
2629+-- we are using a mapping function.
2630+--
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.
2636+--
2637+-- Passing on of the initial accumulator value to the next fold is equivalent
2638+-- to returning the leftover concept.
2639+
2640+{-
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.
2648+--
2649+{-# INLINE groupScan #-}
2650+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
2654+-}
2655+
2656+-- | Group the input stream into groups of @n@ elements each and then fold each
2657+-- group using the provided fold function.
2658+--
2659+-- >> S.toList $ FL.chunksOf 2 FL.sum (S.enumerateFromTo 1 10)
2660+-- > [3,7,11,15,19]
2661+--
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.
2664+--
2665+-- @since 0.7.0
2666+{-# INLINE chunksOf #-}
2667+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)
2671+
2672+-- | Transform a fold from a pure input to a 'Maybe' input, consuming only
2673+-- 'Just' values.
2674+{-# INLINE lcatMaybes #-}
2675+lcatMaybes :: Monad m => Fold m a b -> Fold m (Maybe a) b
2676+lcatMaybes = FL.lfilter isJust . FL.lmap fromJust
2677+
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
2681+-- > [3,7,11,15,19]
2682+--
2683+-- | Group the input stream into windows of @n@ second each and then fold each
2684+-- group using the provided fold function.
2685+--
2686+-- @since 0.7.0
2687+{-# INLINE sessionsOf #-}
2688+sessionsOf
2689+ :: (IsStream t, MonadAsync m)
2690+ => Double -> Fold m a b -> t m a -> t m b
2691+sessionsOf n f xs =
2692+ splitBySuffix isNothing (lcatMaybes f)
2693+ (intersperseByTime n (return Nothing) (Serial.map Just xs))
2694+
2695+------------------------------------------------------------------------------
2696+-- Element Aware APIs
2697+------------------------------------------------------------------------------
2698+--
2699+------------------------------------------------------------------------------
2700+-- Binary APIs
2701+------------------------------------------------------------------------------
2702+
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.
2706+--
2707+_spanBy
2708+ :: Monad m
2709+ => (a -> a -> Bool)
2710+ -> Fold m a b
2711+ -> Fold m a c
2712+ -> Fold m a (b, c)
2713+_spanBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
2714+ Fold step initial extract
2715+
2716+ where
2717+ initial = Tuple3' <$> initialL <*> initialR <*> return (Tuple' Nothing True)
2718+
2719+ step (Tuple3' a b (Tuple' (Just frst) isFirstG)) input =
2720+ if cmp frst input && isFirstG
2721+ then stepL a input
2722+ >>= (\a' -> return (Tuple3' a' b (Tuple' (Just frst) isFirstG)))
2723+ else stepR b input
2724+ >>= (\a' -> return (Tuple3' a a' (Tuple' Nothing False)))
2725+
2726+ step (Tuple3' a b (Tuple' Nothing isFirstG)) input =
2727+ if isFirstG
2728+ then stepL a input
2729+ >>= (\a' -> return (Tuple3' a' b (Tuple' (Just input) isFirstG)))
2730+ else stepR b input
2731+ >>= (\a' -> return (Tuple3' a a' (Tuple' Nothing False)))
2732+
2733+ extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
2734+
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
2737+-- input.
2738+--
2739+-- > let span_ p xs = FL.foldl' (FL.span p FL.toList FL.toList) $ S.fromList xs
2740+--
2741+-- >>> span_ (< 1) [1,2,3]
2742+-- > ([],[1,2,3])
2743+--
2744+-- >>> span_ (< 2) [1,2,3]
2745+-- > ([1],[2,3])
2746+--
2747+-- >>> span_ (< 4) [1,2,3]
2748+-- > ([1,2,3],[])
2749+--
2750+-- @since 0.7.0
2751+
2752+-- This can be considered as a two-fold version of 'ltakeWhile' where we take
2753+-- both the segments instead of discarding the leftover.
2754+{-# INLINE span #-}
2755+span
2756+ :: Monad m
2757+ => (a -> Bool)
2758+ -> Fold m a b
2759+ -> Fold m a c
2760+ -> Fold m a (b, c)
2761+span p (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
2762+ Fold step initial extract
2763+
2764+ where
2765+
2766+ initial = Tuple3' <$> initialL <*> initialR <*> return True
2767+
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))
2772+
2773+ extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
2774+
2775+-- |
2776+-- > break p = span (not . p)
2777+--
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@.
2781+--
2782+-- This is the binary version of 'splitBy'.
2783+--
2784+-- > let break_ p xs = FL.foldl' (FL.break p FL.toList FL.toList) $ S.fromList xs
2785+--
2786+-- >>> break_ (< 1) [3,2,1]
2787+-- > ([3,2,1],[])
2788+--
2789+-- >>> break_ (< 2) [3,2,1]
2790+-- > ([3,2],[1])
2791+--
2792+-- >>> break_ (< 4) [3,2,1]
2793+-- > ([],[3,2,1])
2794+--
2795+-- @since 0.7.0
2796+{-# INLINE _break #-}
2797+_break
2798+ :: Monad m
2799+ => (a -> Bool)
2800+ -> Fold m a b
2801+ -> Fold m a c
2802+ -> Fold m a (b, c)
2803+_break p = span (not . p)
2804+
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 #-}
2808+_spanRollingBy
2809+ :: Monad m
2810+ => (a -> a -> Bool)
2811+ -> Fold m a b
2812+ -> Fold m a c
2813+ -> Fold m a (b, c)
2814+_spanRollingBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
2815+ Fold step initial extract
2816+
2817+ where
2818+ initial = Tuple3' <$> initialL <*> initialR <*> return Nothing
2819+
2820+ step (Tuple3' a b (Just frst)) input =
2821+ if cmp input frst
2822+ then stepL a input >>= (\a' -> return (Tuple3' a' b (Just input)))
2823+ else stepR b input >>= (\b' -> return (Tuple3' a b' (Just input)))
2824+
2825+ step (Tuple3' a b Nothing) input =
2826+ stepL a input >>= (\a' -> return (Tuple3' a' b (Just input)))
2827+
2828+ extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
2829+
2830+------------------------------------------------------------------------------
2831+-- N-ary APIs
2832+------------------------------------------------------------------------------
2833+--
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.
2840+--
2841+-- >>> S.toList $ FL.groupsBy (>) FL.toList $ S.fromList [1,3,7,0,2,5]
2842+-- > [[1,3,7],[0,2,5]]
2843+--
2844+-- @since 0.7.0
2845+{-# INLINE groupsBy #-}
2846+groupsBy
2847+ :: (IsStream t, Monad m)
2848+ => (a -> a -> Bool)
2849+ -> Fold m a b
2850+ -> t m a
2851+ -> t m b
2852+groupsBy cmp f m = D.fromStreamD $ D.groupsBy cmp f (D.toStreamD m)
2853+
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
2860+-- @f@.
2861+--
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]]
2864+--
2865+-- @since 0.7.0
2866+{-# INLINE groupsByRolling #-}
2867+groupsByRolling
2868+ :: (IsStream t, Monad m)
2869+ => (a -> a -> Bool)
2870+ -> Fold m a b
2871+ -> t m a
2872+ -> t m b
2873+groupsByRolling cmp f m = D.fromStreamD $ D.groupsRollingBy cmp f (D.toStreamD m)
2874+
2875+-- |
2876+-- > groups = groupsBy (==)
2877+-- > groups = groupsByRolling (==)
2878+--
2879+-- Groups contiguous spans of equal elements together in individual groups.
2880+--
2881+-- >>> S.toList $ FL.groups FL.toList $ S.fromList [1,1,2,2]
2882+-- > [[1,1],[2,2]]
2883+--
2884+-- @since 0.7.0
2885+groups :: (IsStream t, Monad m, Eq a) => Fold m a b -> t m a -> t m b
2886+groups = groupsBy (==)
2887+
2888+------------------------------------------------------------------------------
2889+-- Binary splitting on a separator
2890+------------------------------------------------------------------------------
2891+
2892+{-
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.
2897+--
2898+-- > let breakOn_ pat xs = FL.foldl' (FL.breakOn pat FL.toList FL.toList) $ S.fromList xs
2899+--
2900+-- >>> breakOn_ "dear" "Hello dear world!"
2901+-- > ("Hello ","dear world!")
2902+--
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
2906+-}
2907+
2908+------------------------------------------------------------------------------
2909+-- N-ary split on a predicate
2910+------------------------------------------------------------------------------
2911+
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
2916+-- as argument.
2917+
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
2922+-- follows:
2923+--
2924+-- @
2925+-- "--.--" => "--" "--"
2926+-- "--." => "--" ""
2927+-- ".--" => "" "--"
2928+-- @
2929+--
2930+-- @splitOn (== x)@ is an inverse of @intercalate (S.yield x)@
2931+--
2932+-- Let's use the following definition for illustration:
2933+--
2934+-- > splitOn' p xs = S.toList $ FL.splitOn p (FL.toList) (S.fromList xs)
2935+--
2936+-- >>> splitOn' (== '.') ""
2937+-- [""]
2938+--
2939+-- >>> splitOn' (== '.') "."
2940+-- ["",""]
2941+--
2942+-- >>> splitOn' (== '.') ".a"
2943+-- > ["","a"]
2944+--
2945+-- >>> splitOn' (== '.') "a."
2946+-- > ["a",""]
2947+--
2948+-- >>> splitOn' (== '.') "a.b"
2949+-- > ["a","b"]
2950+--
2951+-- >>> splitOn' (== '.') "a..b"
2952+-- > ["a","","b"]
2953+--
2954+-- @since 0.7.0
2955+
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.
2959+--
2960+{-# INLINE splitOn #-}
2961+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)
2966+
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:
2971+--
2972+-- @
2973+-- "--.--." => "--" "--"
2974+-- "--.--" => "--" "--"
2975+-- ".--." => "" "--"
2976+-- @
2977+--
2978+-- > splitOnSuffix' p xs = S.toList $ FL.splitSuffixBy p (FL.toList) (S.fromList xs)
2979+--
2980+-- >>> splitOnSuffix' (== '.') ""
2981+-- []
2982+--
2983+-- >>> splitOnSuffix' (== '.') "."
2984+-- [""]
2985+--
2986+-- >>> splitOnSuffix' (== '.') "a"
2987+-- ["a"]
2988+--
2989+-- >>> splitOnSuffix' (== '.') ".a"
2990+-- > ["","a"]
2991+--
2992+-- >>> splitOnSuffix' (== '.') "a."
2993+-- > ["a"]
2994+--
2995+-- >>> splitOnSuffix' (== '.') "a.b"
2996+-- > ["a","b"]
2997+--
2998+-- >>> splitOnSuffix' (== '.') "a.b."
2999+-- > ["a","b"]
3000+--
3001+-- >>> splitOnSuffix' (== '.') "a..b.."
3002+-- > ["a","","b",""]
3003+--
3004+-- > lines = splitOnSuffix (== '\n')
3005+--
3006+-- @since 0.7.0
3007+
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.
3011+--
3012+{-# INLINE splitOnSuffix #-}
3013+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)
3018+
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.
3023+--
3024+-- > wordsBy' p xs = S.toList $ FL.wordsBy p (FL.toList) (S.fromList xs)
3025+--
3026+-- >>> wordsBy' (== ',') ""
3027+-- > []
3028+--
3029+-- >>> wordsBy' (== ',') ","
3030+-- > []
3031+--
3032+-- >>> wordsBy' (== ',') ",a,,b,"
3033+-- > ["a","b"]
3034+--
3035+-- > words = wordsBy isSpace
3036+--
3037+-- @since 0.7.0
3038+{-# INLINE wordsBy #-}
3039+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)
3044+
3045+-- | Like 'splitOnSuffix' but keeps the suffix attached to the resulting
3046+-- splits.
3047+--
3048+-- > splitBySuffix' p xs = S.toList $ FL.splitBySuffix p (FL.toList) (S.fromList xs)
3049+--
3050+-- >>> splitBySuffix' (== '.') ""
3051+-- []
3052+--
3053+-- >>> splitBySuffix' (== '.') "."
3054+-- ["."]
3055+--
3056+-- >>> splitBySuffix' (== '.') "a"
3057+-- ["a"]
3058+--
3059+-- >>> splitBySuffix' (== '.') ".a"
3060+-- > [".","a"]
3061+--
3062+-- >>> splitBySuffix' (== '.') "a."
3063+-- > ["a."]
3064+--
3065+-- >>> splitBySuffix' (== '.') "a.b"
3066+-- > ["a.","b"]
3067+--
3068+-- >>> splitBySuffix' (== '.') "a.b."
3069+-- > ["a.","b."]
3070+--
3071+-- >>> splitBySuffix' (== '.') "a..b.."
3072+-- > ["a.",".","b.","."]
3073+--
3074+-- @since 0.7.0
3075+
3076+-- This can be considered as an n-fold version of 'breakPost' where we apply
3077+-- 'breakPost' successively on the input stream.
3078+--
3079+{-# INLINE splitBySuffix #-}
3080+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)
3085+
3086+------------------------------------------------------------------------------
3087+-- Split on a delimiter sequence
3088+------------------------------------------------------------------------------
3089+
3090+-- Int list examples for splitOn:
3091+--
3092+-- >>> splitList [] [1,2,3,3,4]
3093+-- > [[1],[2],[3],[3],[4]]
3094+--
3095+-- >>> splitList [5] [1,2,3,3,4]
3096+-- > [[1,2,3,3,4]]
3097+--
3098+-- >>> splitList [1] [1,2,3,3,4]
3099+-- > [[],[2,3,3,4]]
3100+--
3101+-- >>> splitList [4] [1,2,3,3,4]
3102+-- > [[1,2,3,3],[]]
3103+--
3104+-- >>> splitList [2] [1,2,3,3,4]
3105+-- > [[1],[3,3,4]]
3106+--
3107+-- >>> splitList [3] [1,2,3,3,4]
3108+-- > [[1,2],[],[4]]
3109+--
3110+-- >>> splitList [3,3] [1,2,3,3,4]
3111+-- > [[1,2],[4]]
3112+--
3113+-- >>> splitList [1,2,3,3,4] [1,2,3,3,4]
3114+-- > [[],[]]
3115+
3116+-- | Like 'splitOn' but the separator is a sequence of elements instead of a
3117+-- single element.
3118+--
3119+-- For illustration, let's define a function that operates on pure lists:
3120+--
3121+-- @
3122+-- splitOnSeq' pat xs = S.toList $ FL.splitOnSeq (A.fromList pat) (FL.toList) (S.fromList xs)
3123+-- @
3124+--
3125+-- >>> splitOnSeq' "" "hello"
3126+-- > ["h","e","l","l","o"]
3127+--
3128+-- >>> splitOnSeq' "hello" ""
3129+-- > [""]
3130+--
3131+-- >>> splitOnSeq' "hello" "hello"
3132+-- > ["",""]
3133+--
3134+-- >>> splitOnSeq' "x" "hello"
3135+-- > ["hello"]
3136+--
3137+-- >>> splitOnSeq' "h" "hello"
3138+-- > ["","ello"]
3139+--
3140+-- >>> splitOnSeq' "o" "hello"
3141+-- > ["hell",""]
3142+--
3143+-- >>> splitOnSeq' "e" "hello"
3144+-- > ["h","llo"]
3145+--
3146+-- >>> splitOnSeq' "l" "hello"
3147+-- > ["he","","o"]
3148+--
3149+-- >>> splitOnSeq' "ll" "hello"
3150+-- > ["he","o"]
3151+--
3152+-- 'splitOnSeq' is an inverse of 'intercalate'. The following law always holds:
3153+--
3154+-- > intercalate . splitOn == id
3155+--
3156+-- The following law holds when the separator is non-empty and contains none of
3157+-- the elements present in the input lists:
3158+--
3159+-- > splitOn . intercalate == id
3160+--
3161+-- @since 0.7.0
3162+
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 #-}
3169+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)
3173+
3174+{-
3175+-- This can be implemented easily using Rabin Karp
3176+-- | Split on any one of the given patterns.
3177+{-# INLINE splitOnAny #-}
3178+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)
3182+-}
3183+
3184+-- | Like 'splitSuffixBy' but the separator is a sequence of elements, instead
3185+-- of a predicate for a single element.
3186+--
3187+-- > splitSuffixOn_ pat xs = S.toList $ FL.splitSuffixOn (A.fromList pat) (FL.toList) (S.fromList xs)
3188+--
3189+-- >>> splitSuffixOn_ "." ""
3190+-- [""]
3191+--
3192+-- >>> splitSuffixOn_ "." "."
3193+-- [""]
3194+--
3195+-- >>> splitSuffixOn_ "." "a"
3196+-- ["a"]
3197+--
3198+-- >>> splitSuffixOn_ "." ".a"
3199+-- > ["","a"]
3200+--
3201+-- >>> splitSuffixOn_ "." "a."
3202+-- > ["a"]
3203+--
3204+-- >>> splitSuffixOn_ "." "a.b"
3205+-- > ["a","b"]
3206+--
3207+-- >>> splitSuffixOn_ "." "a.b."
3208+-- > ["a","b"]
3209+--
3210+-- >>> splitSuffixOn_ "." "a..b.."
3211+-- > ["a","","b",""]
3212+--
3213+-- > lines = splitSuffixOn "\n"
3214+--
3215+-- @since 0.7.0
3216+{-# INLINE _splitOnSuffixSeq #-}
3217+_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)
3222+
3223+{-
3224+-- | Like 'splitOn' but drops any empty splits.
3225+--
3226+{-# INLINE wordsOn #-}
3227+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)
3231+-}
3232+
3233+-- XXX use a non-monadic intersperse to remove the MonadAsync constraint.
3234+--
3235+-- | Like 'splitOnSeq' but splits the separator as well, as an infix token.
3236+--
3237+-- > splitOn'_ pat xs = S.toList $ FL.splitOn' (A.fromList pat) (FL.toList) (S.fromList xs)
3238+--
3239+-- >>> splitOn'_ "" "hello"
3240+-- > ["h","","e","","l","","l","","o"]
3241+--
3242+-- >>> splitOn'_ "hello" ""
3243+-- > [""]
3244+--
3245+-- >>> splitOn'_ "hello" "hello"
3246+-- > ["","hello",""]
3247+--
3248+-- >>> splitOn'_ "x" "hello"
3249+-- > ["hello"]
3250+--
3251+-- >>> splitOn'_ "h" "hello"
3252+-- > ["","h","ello"]
3253+--
3254+-- >>> splitOn'_ "o" "hello"
3255+-- > ["hell","o",""]
3256+--
3257+-- >>> splitOn'_ "e" "hello"
3258+-- > ["h","e","llo"]
3259+--
3260+-- >>> splitOn'_ "l" "hello"
3261+-- > ["he","l","","l","o"]
3262+--
3263+-- >>> splitOn'_ "ll" "hello"
3264+-- > ["he","ll","o"]
3265+--
3266+-- @since 0.7.0
3267+{-# INLINE _splitBySeq #-}
3268+_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
3272+
3273+-- | Like 'splitSuffixOn' but keeps the suffix intact in the splits.
3274+--
3275+-- > splitSuffixOn'_ pat xs = S.toList $ FL.splitSuffixOn' (A.fromList pat) (FL.toList) (S.fromList xs)
3276+--
3277+-- >>> splitSuffixOn'_ "." ""
3278+-- [""]
3279+--
3280+-- >>> splitSuffixOn'_ "." "."
3281+-- ["."]
3282+--
3283+-- >>> splitSuffixOn'_ "." "a"
3284+-- ["a"]
3285+--
3286+-- >>> splitSuffixOn'_ "." ".a"
3287+-- > [".","a"]
3288+--
3289+-- >>> splitSuffixOn'_ "." "a."
3290+-- > ["a."]
3291+--
3292+-- >>> splitSuffixOn'_ "." "a.b"
3293+-- > ["a.","b"]
3294+--
3295+-- >>> splitSuffixOn'_ "." "a.b."
3296+-- > ["a.","b."]
3297+--
3298+-- >>> splitSuffixOn'_ "." "a..b.."
3299+-- > ["a.",".","b.","."]
3300+--
3301+-- @since 0.7.0
3302+{-# INLINE _splitBySuffixSeq #-}
3303+_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)
3308+
3309+{-
3310+-- This can be implemented easily using Rabin Karp
3311+-- | Split post any one of the given patterns.
3312+{-# INLINE splitSuffixOnAny #-}
3313+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)
3318+-}
3319+
3320+------------------------------------------------------------------------------
3321+-- Reorder in sequence
3322+------------------------------------------------------------------------------
3323+
3324+{-
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 #-}
3329+reassembleBy
3330+ :: (IsStream t, Monad m)
3331+ => Fold m a b
3332+ -> (a -> a -> Int)
3333+ -> t m a
3334+ -> t m b
3335+reassembleBy = undefined
3336+-}
3337+
23013338------------------------------------------------------------------------------
23023339-- Distributing
23033340------------------------------------------------------------------------------
@@ -2340,6 +3377,320 @@ tap f xs = D.fromStreamD $ D.tap f (D.toStreamD xs)
23403377trace :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m a
23413378trace f = mapM (\x -> void (f x) >> return x)
23423379
3380+------------------------------------------------------------------------------
3381+-- Windowed classification
3382+------------------------------------------------------------------------------
3383+
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.
3389+--
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".
3393+--
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
3396+-- or timestamps.
3397+
3398+------------------------------------------------------------------------------
3399+-- Keyed Sliding Windows
3400+------------------------------------------------------------------------------
3401+
3402+{-
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
3410+ -> t m (k, b)
3411+classifySlidingChunks wsize wslide (Fold step initial extract) str
3412+ = undefined
3413+
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
3427+ -> t m (k, b)
3428+classifySlidingSessions tick interval slide (Fold step initial extract) str
3429+ = undefined
3430+-}
3431+
3432+------------------------------------------------------------------------------
3433+-- Sliding Window Buffers
3434+------------------------------------------------------------------------------
3435+
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.
3442+--
3443+-- XXX we should be able to implement sequence based splitting combinators
3444+-- using this combinator.
3445+
3446+{-
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.
3451+slidingChunkBuffer
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
3456+ -> t m a
3457+ -> t m b
3458+slidingChunkBuffer = undefined
3459+
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
3469+ -> t m a
3470+ -> t m b
3471+slidingSessionBuffer = undefined
3472+-}
3473+
3474+------------------------------------------------------------------------------
3475+-- Keyed Session Windows
3476+------------------------------------------------------------------------------
3477+
3478+{-
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 #-}
3482+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
3488+ -> t m (k, b)
3489+classifyChunksBy spanout reset (Fold step initial extract) str = undefined
3490+
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.
3494+--
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
3501+ -> t m (k, b)
3502+classifyKeepAliveChunks spanout = classifyChunksBy spanout True
3503+-}
3504+
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
3513+-- @tick@.
3514+--
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.
3521+--
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.
3526+--
3527+-- @since 0.7.0
3528+{-# INLINABLE classifySessionsBy #-}
3529+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
3536+ -> t m (k, b)
3537+classifySessionsBy tick timeout reset (Fold step initial extract) str =
3538+ concatMap (\(Tuple4' _ _ _ s) -> s) $ scanlM' sstep szero stream
3539+
3540+ where
3541+
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
3545+
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
3549+ -- size
3550+ --
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.
3555+ --
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.
3561+ --
3562+ -- XXX if the key is an Int, we can also use an IntMap for slightly
3563+ -- better performance.
3564+ --
3565+ let accumulate v = do
3566+ Tuple' _ old <- maybe (initial >>= return . Tuple' ts) return v
3567+ new <- step old a
3568+ return $ Tuple' ts new
3569+ in if closing
3570+ then do
3571+ let (r, mp') = Map.updateLookupWithKey (\_ _ -> Nothing) key mp
3572+ Tuple' _ acc <- accumulate r
3573+ res <- extract acc
3574+ return $ Tuple4' evTime hp mp' (yield (key, res))
3575+ else do
3576+ let r = Map.lookup key mp
3577+ acc <- accumulate r
3578+ let mp' = Map.insert key acc mp
3579+ let hp' =
3580+ case r of
3581+ Nothing ->
3582+ let expiry = addToAbsTime ts timeoutMs
3583+ in H.insert (Entry expiry key) hp
3584+ Just _ -> 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
3590+
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
3596+
3597+ where
3598+
3599+ curTime = addToAbsTime evTime tickMs
3600+ go hp mp out = do
3601+ let hres = H.uncons hp
3602+ case hres of
3603+ Just (Entry ts key, hp') -> do
3604+ let duration = diffAbsTime curTime ts
3605+ if duration >= timeoutMs
3606+ then do
3607+ let (r, mp') = Map.updateLookupWithKey
3608+ (\_ _ -> Nothing) key mp
3609+ case r of
3610+ Nothing -> go hp' mp' out
3611+ Just (Tuple' latestTS acc) -> do
3612+ let dur = diffAbsTime curTime latestTS
3613+ if dur >= timeoutMs || not reset
3614+ then do
3615+ sess <- extract acc
3616+ go hp' mp' ((key, sess) `K.cons` out)
3617+ else
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)
3625+
3626+ -- merge timer events in the stream
3627+ stream = Serial.map Just str `Par.parallel` repeatM timer
3628+ timer = do
3629+ liftIO $ threadDelay (round $ tick * 1000000)
3630+ return Nothing
3631+
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.
3635+--
3636+-- @since 0.7.0
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
3643+ -> t m (k, b)
3644+_classifyKeepAliveSessions timeout = classifySessionsBy 1 timeout True
3645+
3646+------------------------------------------------------------------------------
3647+-- Keyed tumbling windows
3648+------------------------------------------------------------------------------
3649+
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.
3653+
3654+-- XXX instead of using the early termination flag in the stream, we can use an
3655+-- early terminating fold instead.
3656+
3657+{-
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.
3662+--
3663+-- @since 0.7.0
3664+{-# INLINABLE classifyChunksOf #-}
3665+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
3670+ -> t m (k, b)
3671+classifyChunksOf wsize = classifyChunksBy wsize False
3672+-}
3673+
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.
3679+--
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.
3684+--
3685+-- @since 0.7.0
3686+{-# INLINABLE _classifySessionsOf #-}
3687+_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
3692+ -> t m (k, b)
3693+_classifySessionsOf interval = classifySessionsBy 1 interval False
23433694------------------------------------------------------------------------------
23443695-- Exceptions
23453696------------------------------------------------------------------------------