1{-# LANGUAGE BangPatterns #-}
3{-# LANGUAGE ExistentialQuantification #-}
4{-# LANGUAGE FlexibleContexts #-}
5{-# LANGUAGE RankNTypes #-}
6{-# LANGUAGE RecordWildCards #-}
7{-# LANGUAGE ScopedTypeVariables #-}
10-- Module : Streamly.Fold
11-- Copyright : (c) 2019 Composewell Technologies
12-- (c) 2013 Gabriel Gonzalez
14-- Maintainer : harendra.kumar@gmail.com
15-- Stability : experimental
18-- This module is the consumer side dual of the "Streamly.Prelude" module. The
19-- "Streamly.Prelude" module contains producer or stream generation side
20-- compositions while this module contains consumer side or stream fold
21-- composition. Both the modules are organized in the same way so that you can
22-- easily find the corresponding operations.
24-- Stream consumers are essentially left folds that can be composed in many
25-- interesting ways. For example, a parsing applicative composition builds an
26-- applicative parser, a distributing applicative fold composition distributes
27-- the input to the constituent folds and then combines the outputs.
28-- Similarly, a partitioning applicative can partition the input among the
29-- constituent folds. There are many other ways to combine folds.
31-- All combinators in this module are of true streaming nature. For example,
32-- the splitting combinators accept a fold which consumes the elements in the
33-- split segments as soon as they are generated by the producer, segments are
34-- not buffered in memory, resulting in constant memory usage. The folds used
35-- in these combinators can be considered as continuations.
37-- > import qualified Streamly.Fold as FL
40-- A left fold is represented by the type 'Fold'. @Fold m a b@ folds an
41-- input stream consisting of values of type @a@ to a structure of type
42-- @b@. The fold can be run using 'foldl'' and an input stream.
44-- >>> FL.foldl' FL.sum (S.enumerateFromTo 1 100)
47-- Also see the "Streamly.Sink" module that provides specialized left folds
48-- that discard the outputs.
49-- IMPORTANT: keep the signatures consistent with the folds in Streamly.Prelude
80 -- ** Full Folds (Monoidal)
85 -- ** Full Folds (To Containers)
86 -- | Avoid using these folds in scalable or performance critical
87 -- applications, they buffer all the input in memory.
89 , toStream -- experimental
90 , toStreamRev -- experimental
93 , toListRev -- experimental
110 -- XXX these are slower than right folds even when full input is used
117 -- | Unlike stream producer types (e.g. @SerialT m a@) which have only
118 -- output side, folds have an input side as well as an output side. In the
119 -- type @Fold m a b@, the input type is @a@ and the output type is @b@.
120 -- Transformations can be applied either on the input side or on the output
121 -- side. The 'Functor' instance of a fold maps on the output of the fold:
123 -- >>> FL.foldl' (fmap show FL.sum) (S.enumerateFromTo 1 100)
126 -- However, the input side or contravariant transformations are more
127 -- interesting for folds. The following sections describe the input
128 -- transformation operations on a fold. The names of the operations are
129 -- consistent with their covariant counterparts in "Streamly.Prelude", the
130 -- only difference is that they are prefixed with 'l' which stands for
131 -- 'left' assuming left side is the input side, notice that in @Fold m a b@
132 -- the type variable @a@ is on the left side.
134 -- ** Covariant Operations
164 -- ** Mapping Filters
168 -- ** Scanning Filters
173 -- | Insertion adds more elements to the stream.
213 -- | Streams can be split into segments in space or in time. We use the
214 -- term @chunk@ to refer to a spatial length of the stream (spatial window)
215 -- and the term @session@ to refer to a length in time (time window).
217 -- In imperative terms, grouped folding can be considered as a nested loop
218 -- where we loop over the stream to group elements and then loop over
219 -- individual groups to fold them to a single value that is yielded in the
244 -- Keeping the delimiters
249 -- Splitting by multiple sequences
251 -- , splitSuffixOnAny
252 -- , splitPrefixOnAny
261 -- The 'Applicative' instance of a distributing 'Fold' distributes one copy
262 -- of the stream to each fold and combines the results using a function.
266 -- |-------Fold m a b--------|
267 -- ---stream m a---| |---m (b,c,...)
268 -- |-------Fold m a c--------|
273 -- To compute the average of numbers in a stream without going throught he
276 -- >>> let avg = (/) <$> FL.sum <*> fmap fromIntegral FL.length
277 -- >>> FL.foldl' avg (S.enumerateFromTo 1.0 100.0)
280 -- The 'Semigroup' and 'Monoid' instances of a distributing fold distribute
281 -- the input to both the folds and combines the outputs using Monoid or
282 -- Semigroup instances of the output types:
284 -- >>> import Data.Monoid (Sum)
285 -- >>> FL.foldl' (FL.head <> FL.last) (fmap Sum $ S.enumerateFromTo 1.0 100.0)
286 -- Just (Sum {getSum = 101.0})
288 -- The 'Num', 'Floating', and 'Fractional' instances work in the same way.
295 -- Direct items in the input stream to different folds using a function to
296 -- select the fold. This is useful to demultiplex the input stream.
313 -- These can be expressed using lmap/lmapM and unzip
320 , duplicate -- experimental
322 -- * Windowed Classification
323 -- | Split the stream into windows or chunks in space or time. Each window
324 -- can be associated with a key, all events associated with a particular
325 -- key in the window can be folded to a single result. The stream is split
326 -- into windows of specified size, the window can be terminated early if
327 -- the closing flag is specified in the input stream.
329 -- The term "chunk" is used for a space window and the term "session" is
330 -- used for a time window.
332 -- ** Tumbling Windows
333 -- | A new window starts after the previous window is finished.
334 -- , classifyChunksOf
337 -- ** Keep Alive Windows
338 -- | The window size is extended if an event arrives within the specified
339 -- window size. This can represent sessions with idle or inactive timeout.
340 -- , classifyKeepAliveChunks
341 , classifyKeepAliveSessions
344 -- ** Sliding Windows
345 -- | A new window starts after the specified slide from the previous
346 -- window. Therefore windows can overlap.
347 , classifySlidingChunks
348 , classifySlidingSessions
350 -- ** Sliding Window Buffers
351 -- , slidingChunkBuffer
352 -- , slidingSessionBuffer
356import Control.Concurrent (threadDelay, forkIO, killThread)
357import Control.Concurrent.MVar (MVar, newMVar, takeMVar, putMVar)
358import Control.Exception (SomeException(..), catch, mask)
359import Control.Monad (void)
360import Control.Monad.Catch (throwM)
361import Control.Monad.IO.Class (MonadIO(..))
362import Control.Monad.Trans.Control (control)
363import Data.Functor.Identity (Identity)
364import Data.Heap (Entry(..))
365import Data.Map.Strict (Map)
366import Data.Maybe (fromJust, isJust, isNothing)
368import Foreign.Storable (Storable(..))
370 hiding (filter, drop, dropWhile, take, takeWhile, zipWith, foldr,
371 foldl, map, mapM_, sequence, all, any, sum, product, elem,
372 notElem, maximum, minimum, head, last, tail, length, null,
373 reverse, iterate, init, and, or, lookup, foldr1, (!!),
374 scanl, scanl1, replicate, concatMap, mconcat, foldMap, unzip,
375 span, splitAt, break, mapM)
377import qualified Data.Heap as H
378import qualified Data.Map.Strict as Map
379import qualified Prelude
381import Streamly (MonadAsync, parallel)
382import Streamly.Fold.Types (Fold(..))
383import Streamly.Mem.Array.Types (Array)
384-- import Streamly.Mem.Ring (Ring)
385import Streamly.Streams.Serial (SerialT)
386import Streamly.Streams.StreamK (IsStream())
387import Streamly.Time.Units
388 (AbsTime, MilliSecond64(..), addToAbsTime, diffAbsTime, toRelTime,
391import Streamly.Strict
393import qualified Streamly.Mem.Array.Types as A
394import qualified Streamly.Prelude as S
395import qualified Streamly.Streams.StreamD as D
396import qualified Streamly.Streams.StreamK as K
397import qualified Streamly.Streams.Prelude as P
401-- We can use the left folds in this module instead of the folds in
402-- "Streamly.Prelude". For example the following two ways of folding are
403-- equivalent in functionality and performance,
405-- >>> FL.foldl' FL.sum (S.enumerateFromTo 1 100)
407-- >>> S.sum (S.enumerateFromTo 1 100)
410-- However, left folds are push type folds. That means we push the entire input
411-- to a fold before we can get the output. Therefore, the performance is
412-- equivalent only for full folds like 'sum' and 'length'. For partial folds
413-- like 'head' or 'any' the folds in "Streamly.Prelude" may be much more
414-- efficient because they are implemented as right folds that terminate as soon
415-- as we get the result. Note that when a full fold is composed with a partial
416-- fold in parallel the performance is not impacted as we anyway have to
417-- consume the whole stream due to the full fold.
419-- >>> S.head (1 `S.cons` undefined)
421-- >>> FL.foldl' FL.head (1 `S.cons` undefined)
422-- *** Exception: Prelude.undefined
424-- However, we can wrap the fold in a scan to convert it into a lazy stream of
425-- fold steps. We can then terminate the stream whenever we want. For example,
427-- >>> S.toList $ S.take 1 $ FL.scanl' FL.head (1 `S.cons` undefined)
430-- The following example extracts the input stream up to a point where the
431-- running average of elements is no more than 10:
435-- $ S.map (fromJust . fst)
436-- $ S.takeWhile (\\(_,x) -> x <= 10)
437-- $ FL.postscanl' ((,) \<$> FL.last \<*> avg) (S.enumerateFromTo 1.0 100.0)
440-- [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]
443------------------------------------------------------------------------------
445------------------------------------------------------------------------------
447-- | Fold a stream using the supplied monadic fold.
449-- >>> FL.foldl' FL.sum (S.enumerateFromTo 1 100)
454foldl' :: Monad m => Fold m a b -> SerialT m a -> m b
455foldl' (Fold step begin done) = P.foldlMx' step begin done
457------------------------------------------------------------------------------
458-- Transformations on fold inputs
459------------------------------------------------------------------------------
461-- | Convert a fold that generates a monadic action to a fold that generates
462-- the output of that monadic action.
465{-# INLINE sequence #-}
466sequence :: Monad m => Fold m a (m b) -> Fold m a b
467sequence (Fold step initial extract) = Fold step initial extract'
473-- | Convert a fold that generates a monadic action to a fold that generates
474-- the output of that monadic action.
478mapM :: Monad m => (b -> m c) -> Fold m a b -> Fold m a c
479mapM f = sequence . fmap f
481------------------------------------------------------------------------------
482-- Transformations on fold inputs
483------------------------------------------------------------------------------
485-- | @(lmap f fold)@ maps the function @f@ on the input of the fold.
487-- >>> FL.foldl' (FL.lmap (\x -> x * x) FL.sum) (S.enumerateFromTo 1 100)
491{-# INLINABLE lmap #-}
492lmap :: (a -> b) -> Fold m b r -> Fold m a r
493lmap f (Fold step begin done) = Fold step' begin done
495 step' x a = step x (f a)
497-- | @(lmapM f fold)@ maps the monadic function @f@ on the input of the fold.
500{-# INLINABLE lmapM #-}
501lmapM :: Monad m => (a -> m b) -> Fold m b r -> Fold m a r
502lmapM f (Fold step begin done) = Fold step' begin done
504 step' x a = f a >>= step x
506------------------------------------------------------------------------------
508------------------------------------------------------------------------------
510-- | @_Fold1 step@ returns a new 'Fold' using just a step function that has the
511-- same type for the accumulator and the element. The result type is the
512-- accumulator type wrapped in 'Maybe'. The initial accumulator is retrieved
513-- from the 'Foldable', the result is 'None' for empty containers.
514{-# INLINABLE _Fold1 #-}
515_Fold1 :: Monad m => (a -> a -> a) -> Fold m a (Maybe a)
516_Fold1 step = Fold step_ (return Nothing') fromStrictMaybe
518 step_ mx a = return $ Just' $
523------------------------------------------------------------------------------
525------------------------------------------------------------------------------
527------------------------------------------------------------------------------
529------------------------------------------------------------------------------
531-- | A fold that drains all its input, running the effects and discarding the
535{-# INLINABLE drain #-}
536drain :: Monad m => Fold m a ()
537drain = Fold step begin done
544-- > drainBy f = lmapM f drain
546-- Drain all input after passing it through a monadic function. This is the
547-- dual of mapM_ on stream producers.
548{-# INLINABLE drainBy #-}
549drainBy :: Monad m => (a -> m b) -> Fold m a ()
550drainBy f = Fold (const (void . f)) (return ()) return
552-- | Extract the last element of the input stream, if any.
555{-# INLINABLE last #-}
556last :: Monad m => Fold m a (Maybe a)
557last = _Fold1 (flip const)
559------------------------------------------------------------------------------
561------------------------------------------------------------------------------
563-- | Like 'length', except with a more general 'Num' return value
566{-# INLINABLE genericLength #-}
567genericLength :: (Monad m, Num b) => Fold m a b
568genericLength = Fold (\n _ -> return $ n + 1) (return 0) return
570-- | Determine the length of the input stream.
573{-# INLINABLE length #-}
574length :: Monad m => Fold m a Int
575length = genericLength
577-- | Determine the sum of all elements of a stream of numbers. Returns additive
578-- identity (@0@) when the stream is empty. Note that this is not numerically
579-- stable for floating point numbers.
583sum :: (Monad m, Num a) => Fold m a a
584sum = Fold (\x a -> return $ x + a) (return 0) return
586-- | Determine the product of all elements of a stream of numbers. Returns
587-- multiplicative identity (@1@) when the stream is empty.
590{-# INLINABLE product #-}
591product :: (Monad m, Num a) => Fold m a a
592product = Fold (\x a -> return $ x * a) (return 1) return
594------------------------------------------------------------------------------
596------------------------------------------------------------------------------
598-- | Determine the maximum element in a stream using the supplied comparison
602{-# INLINABLE maximumBy #-}
603maximumBy :: Monad m => (a -> a -> Ordering) -> Fold m a (Maybe a)
604maximumBy cmp = _Fold1 max'
606 max' x y = case cmp x y of
612-- maximum = 'maximumBy' compare
615-- Determine the maximum element in a stream.
618{-# INLINABLE maximum #-}
619maximum :: (Monad m, Ord a) => Fold m a (Maybe a)
622-- | Computes the minimum element with respect to the given comparison function
623{-# INLINABLE minimumBy #-}
624minimumBy :: Monad m => (a -> a -> Ordering) -> Fold m a (Maybe a)
625minimumBy cmp = _Fold1 min'
627 min' x y = case cmp x y of
631-- | Determine the minimum element in a stream using the supplied comparison
635{-# INLINABLE minimum #-}
636minimum :: (Monad m, Ord a) => Fold m a (Maybe a)
639------------------------------------------------------------------------------
640-- To Summary (Statistical)
641------------------------------------------------------------------------------
643-- | Compute a numerically stable arithmetic mean of all elements in the input
647{-# INLINABLE mean #-}
648mean :: (Monad m, Fractional a) => Fold m a a
649mean = Fold step (return begin) (return . done)
652 step (Tuple' x n) y = return $
654 in Tuple' (x + (y - x) / n') n'
655 done (Tuple' x _) = x
657-- | Compute a numerically stable (population) variance over all elements in
661{-# INLINABLE variance #-}
662variance :: (Monad m, Fractional a) => Fold m a a
663variance = Fold step (return begin) (return . done)
665 begin = Tuple3' 0 0 0
667 step (Tuple3' n mean_ m2) x = return $ Tuple3' n' mean' m2'
670 mean' = (n * mean_ + x) / (n + 1)
672 m2' = m2 + delta * delta * n / (n + 1)
674 done (Tuple3' n _ m2) = m2 / n
676-- | Compute a numerically stable (population) standard deviation over all
677-- elements in the input stream.
680{-# INLINABLE stdDev #-}
681stdDev :: (Monad m, Floating a) => Fold m a a
682stdDev = sqrt variance
684------------------------------------------------------------------------------
685-- Monoidal left folds
686------------------------------------------------------------------------------
688-- | Fold an input stream consisting of monoidal elements using 'mappend'
691-- > FL.foldl FL.mconcat (S.map Sum $ S.enumerateFromTo 1 10)
694{-# INLINABLE mconcat #-}
695mconcat :: (Monad m, Monoid a) => Fold m a a
696mconcat = Fold (\x a -> return $ mappend x a) (return mempty) return
699-- > foldMap f = map f mconcat
701-- Make a fold from a pure function that folds the output of the function
702-- using 'mappend' and 'mempty'.
704-- > FL.foldl (FL.foldMap Sum) $ S.enumerateFromTo 1 10
707{-# INLINABLE foldMap #-}
708foldMap :: (Monad m, Monoid b) => (a -> b) -> Fold m a b
709foldMap f = lmap f mconcat
712-- > foldMapM f = mapM f mconcat
714-- Make a fold from a monadic function that folds the output of the function
715-- using 'mappend' and 'mempty'.
717-- > FL.foldM (FL.foldMapM (return . Sum)) $ S.enumerateFromTo 1 10
720{-# INLINABLE foldMapM #-}
721foldMapM :: (Monad m, Monoid b) => (a -> m b) -> Fold m a b
722foldMapM act = Fold step begin done
725 begin = return mempty
728 return $! mappend m m'
730------------------------------------------------------------------------------
732------------------------------------------------------------------------------
734-- | Folds the input stream to a list.
736-- /Warning!/ working on large lists accumulated as buffers in memory could be
737-- very inefficient, consider using "Streamly.Array" instead.
741-- id . (x1 :) . (x2 :) . (x3 :) . ... . (xn :) $ []
742{-# INLINABLE toList #-}
743toList :: Monad m => Fold m a [a]
744toList = Fold (\f x -> return $ f . (x :))
748-- This is more efficient than 'toList'. toList is exactly the same as
749-- reversing the list after toListRev.
751-- | Buffers the input stream to a list in the reverse order of the input.
753-- /Warning!/ working on large lists accumulated as buffers in memory could be
754-- very inefficient, consider using "Streamly.Array" instead.
758-- xn : ... : x2 : x1 : []
759{-# INLINABLE toListRev #-}
760toListRev :: Monad m => Fold m a [a]
761toListRev = Fold (\xs x -> return $ x:xs) (return []) return
763-- | A fold that buffers its input to a pure stream.
765-- /Warning!/ working on large streams accumulated as buffers in memory could
766-- be very inefficient, consider using "Streamly.Array" instead.
769{-# INLINE toStream #-}
770toStream :: Monad m => Fold m a (SerialT Identity a)
771toStream = Fold (\f x -> return $ f . (x `K.cons`))
775-- This is more efficient than 'toStream'. toStream is exactly the same as
776-- reversing the stream after toStreamRev.
778-- | Buffers the input stream to a pure stream in the reverse order of the
781-- /Warning!/ working on large streams accumulated as buffers in memory could
782-- be very inefficient, consider using "Streamly.Array" instead.
786-- xn : ... : x2 : x1 : []
787{-# INLINABLE toStreamRev #-}
788toStreamRev :: Monad m => Fold m a (SerialT Identity a)
789toStreamRev = Fold (\xs x -> return $ x `S.cons` xs) (return S.nil) return
791------------------------------------------------------------------------------
793------------------------------------------------------------------------------
795------------------------------------------------------------------------------
797------------------------------------------------------------------------------
799-- | Like 'index', except with a more general 'Integral' argument
802{-# INLINABLE genericIndex #-}
803genericIndex :: (Integral i, Monad m) => i -> Fold m a (Maybe a)
804genericIndex i = Fold step (return $ Left' 0) done
817-- | Lookup the element at the given index.
820{-# INLINABLE index #-}
821index :: Monad m => Int -> Fold m a (Maybe a)
824-- | Extract the first element of the stream, if any.
827{-# INLINABLE head #-}
828head :: Monad m => Fold m a (Maybe a)
831-- | Returns the first element that satisfies the given predicate.
834{-# INLINABLE find #-}
835find :: Monad m => (a -> Bool) -> Fold m a (Maybe a)
836find predicate = Fold step (return Nothing') fromStrictMaybe
840 Nothing' -> if predicate a
845-- | In a stream of (key-value) pairs @(a, b)@, return the value @b@ of the
846-- first pair where the key equals the given value @a@.
849{-# INLINABLE lookup #-}
850lookup :: (Eq a, Monad m) => a -> Fold m (a,b) (Maybe b)
851lookup a0 = Fold step (return Nothing') fromStrictMaybe
853 step x (a,b) = return $
855 Nothing' -> if a == a0
860-- | Convert strict 'Either'' to lazy 'Maybe'
861{-# INLINABLE hush #-}
862hush :: Either' a b -> Maybe b
863hush (Left' _) = Nothing
864hush (Right' b) = Just b
866-- | Returns the first index that satisfies the given predicate.
869{-# INLINABLE findIndex #-}
870findIndex :: Monad m => (a -> Bool) -> Fold m a (Maybe Int)
871findIndex predicate = Fold step (return $ Left' 0) (return . hush)
881-- | Returns the first index where a given value is found in the stream.
884{-# INLINABLE elemIndex #-}
885elemIndex :: (Eq a, Monad m) => a -> Fold m a (Maybe Int)
886elemIndex a = findIndex (a ==)
888------------------------------------------------------------------------------
890------------------------------------------------------------------------------
892-- | Return 'True' if the input stream is empty.
895{-# INLINABLE null #-}
896null :: Monad m => Fold m a Bool
897null = Fold (\_ _ -> return False) (return True) return
900-- > any p = lmap p or
902-- | Returns 'True' if any of the elements of a stream satisfies a predicate.
906any :: Monad m => (a -> Bool) -> Fold m a Bool
907any predicate = Fold (\x a -> return $ x || predicate a) (return False) return
909-- | Return 'True' if the given element is present in the stream.
912{-# INLINABLE elem #-}
913elem :: (Eq a, Monad m) => a -> Fold m a Bool
917-- > all p = lmap p and
919-- | Returns 'True' if all elements of a stream satisfy a predicate.
923all :: Monad m => (a -> Bool) -> Fold m a Bool
924all predicate = Fold (\x a -> return $ x && predicate a) (return True) return
926-- | Returns 'True' if the given element is not present in the stream.
929{-# INLINABLE notElem #-}
930notElem :: (Eq a, Monad m) => a -> Fold m a Bool
931notElem a = all (a /=)
933-- | Returns 'True' if all elements are 'True', 'False' otherwise
937and :: Monad m => Fold m Bool Bool
938and = Fold (\x a -> return $ x && a) (return True) return
940-- | Returns 'True' if any element is 'True', 'False' otherwise
944or :: Monad m => Fold m Bool Bool
945or = Fold (\x a -> return $ x || a) (return False) return
947------------------------------------------------------------------------------
948-- Scanning with a Fold
949------------------------------------------------------------------------------
951-- | Scan a stream using the given monadic fold.
955scanl' :: Monad m => Fold m a b -> SerialT m a -> SerialT m b
956scanl' (Fold step begin done) = P.scanlMx' step begin done
958-- | Postscan a stream using the given monadic fold.
961{-# INLINE postscanl' #-}
962postscanl' :: Monad m => Fold m a b -> SerialT m a -> SerialT m b
963postscanl' (Fold step begin done) = P.postscanlMx' step begin done
967------------------------------------------------------------------------------
969------------------------------------------------------------------------------
971-- | Include only those elements that pass a predicate.
973-- >>> FL.foldl (lfilter (> 5) FL.sum) [1..10]
977{-# INLINABLE lfilter #-}
978lfilter :: Monad m => (a -> Bool) -> Fold m a r -> Fold m a r
979lfilter f (Fold step begin done) = Fold step' begin done
981 step' x a = if f a then step x a else return x
983-- | Like 'lfilter' but with a monadic predicate.
986{-# INLINABLE lfilterM #-}
987lfilterM :: Monad m => (a -> m Bool) -> Fold m a r -> Fold m a r
988lfilterM f (Fold step begin done) = Fold step' begin done
992 if use then step x a else return x
994-- | Take first 'n' elements from the stream and discard the rest.
997{-# INLINABLE ltake #-}
998ltake :: Monad m => Int -> Fold m a b -> Fold m a b
999ltake n (Fold step initial done) = Fold step' initial' done'
1001 initial' = fmap (Tuple' 0) initial
1002 step' (Tuple' i r) a = do
1006 return $ Tuple' (i + 1) res
1007 else return $ Tuple' i r
1008 done' (Tuple' _ r) = done r
1010-- | Takes elements from the input as long as the predicate succeeds.
1013{-# INLINABLE ltakeWhile #-}
1014ltakeWhile :: Monad m => (a -> Bool) -> Fold m a b -> Fold m a b
1015ltakeWhile predicate (Fold step initial done) = Fold step' initial' done'
1017 initial' = fmap Left' initial
1018 step' (Left' r) a = do
1020 then fmap Left' $ step r a
1021 else return (Right' r)
1022 step' r _ = return r
1023 done' (Left' r) = done r
1024 done' (Right' r) = done r
1026------------------------------------------------------------------------------
1027-- Grouping/Splitting
1028------------------------------------------------------------------------------
1030------------------------------------------------------------------------------
1031-- Grouping without looking at elements
1032------------------------------------------------------------------------------
1034------------------------------------------------------------------------------
1036------------------------------------------------------------------------------
1039-- | @splitAt n f1 f2@ composes folds @f1@ and @f2@ such that first @n@
1040-- elements of its input are sent to fold @f1@ and the rest of the stream is
1041-- sent to fold @f2@.
1043-- > let splitAt_ n xs = FL.foldl' (FL.splitAt n FL.toList FL.toList) $ S.fromList xs
1045-- >>> splitAt_ 6 "Hello World!"
1046-- > ("Hello ","World!")
1048-- >>> splitAt_ (-1) [1,2,3]
1051-- >>> splitAt_ 0 [1,2,3]
1054-- >>> splitAt_ 1 [1,2,3]
1057-- >>> splitAt_ 3 [1,2,3]
1060-- >>> splitAt_ 4 [1,2,3]
1063-- This can be considered as a two-fold version of 'ltake' where we take both
1064-- the segments instead of discarding the leftover.
1067{-# INLINE splitAt #-}
1074splitAt n (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
1075 Fold step init extract
1077 init = Tuple3' <$> return n <*> initialL <*> initialR
1079 step (Tuple3' i xL xR) input =
1081 then stepL xL input >>= (\a -> return (Tuple3' (i - 1) a xR))
1082 else stepR xR input >>= (\b -> return (Tuple3' i xL b))
1084 extract (Tuple3' _ a b) = (,) <$> extractL a <*> extractR b
1086------------------------------------------------------------------------------
1088------------------------------------------------------------------------------
1090------------------------------------------------------------------------------
1091-- Generalized grouping
1092------------------------------------------------------------------------------
1094-- This combinator is the most general grouping combinator and can be used to
1095-- implement all other grouping combinators.
1097-- XXX check if this can implement the splitOn combinator i.e. we can slide in
1098-- new elements, slide out old elements and incrementally compute the hash.
1099-- Also, can we implement the windowed classification combinators using this?
1101-- In fact this is a parse. Instead of using a special return value in the fold
1102-- we are using a mapping function.
1104-- Note that 'scanl'' (usually followed by a map to extract the desired value
1105-- from the accumulator) can be used to realize many implementations e.g. a
1106-- sliding window implementation. A scan followed by a mapMaybe is also a good
1107-- pattern to express many problems where we want to emit a filtered output and
1108-- not emit an output on every input.
1110-- Passing on of the initial accumulator value to the next fold is equivalent
1111-- to returning the leftover concept.
1114-- | @groupScan splitter fold stream@ folds the input stream using @fold@.
1115-- @splitter@ is applied on the accumulator of the fold every time an item is
1116-- consumed by the fold. The fold continues until @splitter@ returns a 'Just'
1117-- value. A 'Just' result from the @splitter@ specifies a result to be emitted
1118-- in the output stream and the initial value of the accumulator for the next
1119-- group's fold. This allows us to control whether to start fresh for the next
1120-- fold or to continue from the previous fold's output.
1122{-# INLINE groupScan #-}
1124 :: (IsStream t, Monad m)
1125 => (x -> m (Maybe (b, x))) -> Fold m a x -> t m a -> t m b
1126groupScan split fold m = undefined
1129-- | Group the input stream into groups of @n@ elements each and then fold each
1130-- group using the provided fold function.
1132-- >> S.toList $ FL.chunksOf 2 FL.sum (S.enumerateFromTo 1 10)
1135-- This can be considered as an n-fold version of 'ltake' where we apply
1136-- 'ltake' repeatedly on the leftover stream until the stream exhausts.
1139{-# INLINE chunksOf #-}
1141 :: (IsStream t, Monad m)
1142 => Int -> Fold m a b -> t m a -> t m b
1143chunksOf n f m = D.fromStreamD $ D.groupsOf n f (D.toStreamD m)
1145-- | Transform a fold from a pure input to a 'Maybe' input, consuming only
1147{-# INLINE lcatMaybes #-}
1148lcatMaybes :: Monad m => Fold m a b -> Fold m (Maybe a) b
1149lcatMaybes = lfilter isJust . lmap fromJust
1151-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
1152-- XXX add this example after fixing the serial stream rate control
1153-- >>> S.toList $ S.take 5 $ sessionsOf 1 FL.sum $ constRate 2 $ S.enumerateFrom 1
1156-- | Group the input stream into windows of @n@ second each and then fold each
1157-- group using the provided fold function.
1160{-# INLINE sessionsOf #-}
1162 :: (IsStream t, MonadAsync m)
1163 => Double -> Fold m a b -> t m a -> t m b
1165 splitSuffixBy' isNothing (lcatMaybes f)
1166 (S.intersperseByTime n (return Nothing) (S.map Just xs))
1168------------------------------------------------------------------------------
1169-- Element Aware APIs
1170------------------------------------------------------------------------------
1172------------------------------------------------------------------------------
1174------------------------------------------------------------------------------
1176-- | Break the input stream into two groups, the first group takes the input as
1177-- long as the predicate applied to the first element of the stream and next
1178-- input element holds 'True', the second group takes the rest of the input.
1186_spanBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
1187 Fold step init extract
1190 init = Tuple3' <$> initialL <*> initialR <*> return (Tuple' Nothing True)
1192 step (Tuple3' a b (Tuple' (Just frst) isFirstG)) input =
1193 if cmp frst input && isFirstG
1195 >>= (\a' -> return (Tuple3' a' b (Tuple' (Just frst) isFirstG)))
1197 >>= (\a' -> return (Tuple3' a a' (Tuple' Nothing False)))
1199 step (Tuple3' a b (Tuple' Nothing isFirstG)) input =
1202 >>= (\a' -> return (Tuple3' a' b (Tuple' (Just input) isFirstG)))
1204 >>= (\a' -> return (Tuple3' a a' (Tuple' Nothing False)))
1206 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
1208-- | Span as long as the predicate is 'True'. @span p f1 f2@ composes folds
1209-- @f1@ and @f2@ such that the composed fold continues sending the input to
1210-- @f1@ as long as the predicate @p@ is 'True'. The rest of the input is sent
1213-- > let span_ p xs = FL.foldl' (FL.span p FL.toList FL.toList) $ S.fromList xs
1215-- >>> span_ (< 1) [1,2,3]
1218-- >>> span_ (< 2) [1,2,3]
1221-- >>> span_ (< 4) [1,2,3]
1224-- This can be considered as a two-fold version of 'ltakeWhile' where we take
1225-- both the segments instead of discarding the leftover.
1235span p (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
1236 Fold step init extract
1239 init = Tuple3' <$> initialL <*> initialR <*> return True
1241 step (Tuple3' a b isFirstG) input =
1242 if isFirstG && p input
1243 then stepL a input >>= (\a' -> return (Tuple3' a' b True))
1244 else stepR b input >>= (\a' -> return (Tuple3' a a' False))
1246 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
1249-- > break p = span (not . p)
1251-- Break as soon as the predicate becomes 'True'. @break p f1 f2@ composes
1252-- folds @f1@ and @f2@ such that @f1@ stops receiving input as soon as the
1253-- predicate @p@ becomes 'True'. The rest of the input is sent to @f2@.
1255-- This is the binary version of 'splitBy'.
1257-- > let break_ p xs = FL.foldl' (FL.break p FL.toList FL.toList) $ S.fromList xs
1259-- >>> break_ (< 1) [3,2,1]
1262-- >>> break_ (< 2) [3,2,1]
1265-- >>> break_ (< 4) [3,2,1]
1276break p = span (not . p)
1278-- | Like 'spanBy' but applies the predicate in a rolling fashion i.e.
1279-- predicate is applied to the previous and the next input elements.
1280{-# INLINE _spanRollingBy #-}
1287_spanRollingBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
1288 Fold step init extract
1291 init = Tuple3' <$> initialL <*> initialR <*> return Nothing
1293 step (Tuple3' a b (Just frst)) input =
1295 then stepL a input >>= (\a' -> return (Tuple3' a' b (Just input)))
1296 else stepR b input >>= (\b' -> return (Tuple3' a b' (Just input)))
1298 step (Tuple3' a b Nothing) input =
1299 stepL a input >>= (\a' -> return (Tuple3' a' b (Just input)))
1301 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
1303------------------------------------------------------------------------------
1305------------------------------------------------------------------------------
1307-- | @groupsBy cmp f $ S.fromList [a,b,c,...]@ assigns the element @a@ to the
1308-- first group, if @a \`cmp` b@ is 'True' then @b@ is also assigned to the same
1309-- group. If @a \`cmp` c@ is 'True' then @c@ is also assigned to the same
1310-- group and so on. When the comparison fails a new group is started. Each
1311-- group is folded using the fold @f@.
1313-- >>> S.toList $ FL.groupsBy (>) FL.toList $ S.fromList [1,3,7,0,2,5]
1314-- > [[1,3,7],[0,2,5]]
1317{-# INLINE groupsBy #-}
1319 :: (IsStream t, Monad m)
1324groupsBy cmp f m = D.fromStreamD $ D.groupsBy cmp f (D.toStreamD m)
1326-- | Unlike @groupsBy@ this function performs a rolling comparison of two
1327-- successive elements in the input stream. @groupsRollingBy cmp f $ S.fromList
1328-- [a,b,c,...]@ assigns the element @a@ to the first group, if @a \`cmp` b@ is
1329-- 'True' then @b@ is also assigned to the same group. If @b \`cmp` c@ is
1330-- 'True' then @c@ is also assigned to the same group and so on. When the
1331-- comparison fails a new group is started. Each group is folded using the fold
1334-- >>> S.toList $ FL.groupsRollingBy (\a b -> a + 1 == b) FL.toList $ S.fromList [1,2,3,7,8,9]
1335-- > [[1,2,3],[7,8,9]]
1338{-# INLINE groupsRollingBy #-}
1340 :: (IsStream t, Monad m)
1345groupsRollingBy cmp f m = D.fromStreamD $ D.groupsRollingBy cmp f (D.toStreamD m)
1348-- > groups = groupsBy (==)
1349-- > groups = groupsRollingBy (==)
1351-- Groups a contiguous span of equal elements together in one group.
1353-- >>> S.toList $ FL.groups FL.toList $ S.fromList [1,1,2,2]
1357groups :: (IsStream t, Monad m, Eq a) => Fold m a b -> t m a -> t m b
1358groups = groupsBy (==)
1360------------------------------------------------------------------------------
1361-- Binary splitting on a separator
1362------------------------------------------------------------------------------
1365-- | Find the first occurrence of the specified sequence in the input stream
1366-- and break the input stream into two parts, the first part consisting of the
1367-- stream before the sequence and the second part consisting of the sequence
1368-- and the rest of the stream.
1370-- > let breakOn_ pat xs = FL.foldl' (FL.breakOn pat FL.toList FL.toList) $ S.fromList xs
1372-- >>> breakOn_ "dear" "Hello dear world!"
1373-- > ("Hello ","dear world!")
1375{-# INLINE breakOn #-}
1376breakOn :: Monad m => Array a -> Fold m a b -> Fold m a c -> Fold m a (b,c)
1377breakOn pat f m = undefined
1380------------------------------------------------------------------------------
1381-- N-ary split on a predicate
1382------------------------------------------------------------------------------
1384-- TODO: Use a Splitter configuration similar to the "split" package to make it
1385-- possible to express all splitting combinations. In general, we can have
1386-- infix/suffix/prefix/condensing of separators, dropping both leading/trailing
1387-- separators. We can have a single split operation taking the splitter config
1390-- | Split a stream on separator elements determined by a predicate, dropping
1391-- the separator. Separators are not considered part of stream segments on
1392-- either side of it instead they are treated as infixed between two stream
1393-- segments. For example, with @.@ as separator, @"a.b.c"@ would be parsed as
1394-- @["a","b","c"]@. When @.@ is in leading or trailing position it is still
1395-- considered as infixed, treating the first or the last segment as empty. For
1396-- example, @".a."@ would be parsed as @["","a",""]@. This operation is
1397-- opposite of 'intercalate'.
1399-- Let's use the following definition for illustration:
1401-- > splitBy_ p xs = S.toList $ FL.splitBy p (FL.toList) (S.fromList xs)
1403-- >>> splitBy_ (== '.') ""
1406-- >>> splitBy_ (== '.') "."
1409-- >>> splitBy_ (== '.') ".a"
1412-- >>> splitBy_ (== '.') "a."
1415-- >>> splitBy_ (== '.') "a.b"
1418-- >>> splitBy_ (== '.') "a..b"
1421-- This can be considered as an n-fold version of 'break' where we apply
1422-- 'break' successively on the input stream, dropping the first element
1423-- of the second segment after each break.
1426{-# INLINE splitBy #-}
1428 :: (IsStream t, Monad m)
1429 => (a -> Bool) -> Fold m a b -> t m a -> t m b
1430splitBy predicate f m =
1431 D.fromStreamD $ D.splitBy predicate f (D.toStreamD m)
1433-- | Like 'splitBy' but the separator is treated as part of the previous
1434-- stream segment (suffix). Therefore, when the separator is in trailing
1435-- position, no empty segment is considered to follow it. For example, @"a.b."@
1436-- would be parsed as @["a","b"]@ instead of @["a","b",""]@ as in the case of
1439-- > splitSuffixBy_ p xs = S.toList $ FL.splitSuffixBy p (FL.toList) (S.fromList xs)
1441-- >>> splitSuffixBy_ (== '.') ""
1444-- >>> splitSuffixBy_ (== '.') "."
1447-- >>> splitSuffixBy_ (== '.') "a"
1450-- >>> splitSuffixBy_ (== '.') ".a"
1453-- >>> splitSuffixBy_ (== '.') "a."
1456-- >>> splitSuffixBy_ (== '.') "a.b"
1459-- >>> splitSuffixBy_ (== '.') "a.b."
1462-- >>> splitSuffixBy_ (== '.') "a..b.."
1465-- > lines = splitSuffixBy (== '\n')
1467-- This can be considered as an n-fold version of 'breakPost' where we apply
1468-- 'breakPost' successively on the input stream, dropping the first element
1469-- of the second segment after each break.
1472{-# INLINE splitSuffixBy #-}
1474 :: (IsStream t, Monad m)
1475 => (a -> Bool) -> Fold m a b -> t m a -> t m b
1476splitSuffixBy predicate f m =
1477 D.fromStreamD $ D.splitSuffixBy predicate f (D.toStreamD m)
1479-- | Like 'splitBy' but ignores repeated separators or separators in leading
1480-- or trailing position. Therefore, @"..a..b.."@ would be parsed as
1481-- @["a","b"]@. In other words, it treats the input like words separated by
1482-- whitespace elements determined by the predicate.
1484-- > wordsBy' p xs = S.toList $ FL.wordsBy p (FL.toList) (S.fromList xs)
1486-- >>> wordsBy' (== ',') ""
1489-- >>> wordsBy' (== ',') ","
1492-- >>> wordsBy' (== ',') ",a,,b,"
1495-- > words = wordsBy isSpace
1498{-# INLINE wordsBy #-}
1500 :: (IsStream t, Monad m)
1501 => (a -> Bool) -> Fold m a b -> t m a -> t m b
1502wordsBy predicate f m =
1503 D.fromStreamD $ D.wordsBy predicate f (D.toStreamD m)
1505-- XXX we should express this using the Splitter config.
1507-- We can get splitSuffixBy' by appending the suffix to the output segments
1508-- produced by splitSuffixBy. However, it may add an additional suffix if the last
1509-- fragment did not have a suffix in the first place.
1511-- | Like 'splitSuffixBy' but keeps the suffix in the splits.
1513-- > splitSuffixBy'_ p xs = S.toList $ FL.splitSuffixBy' p (FL.toList) (S.fromList xs)
1515-- >>> splitSuffixBy'_ (== '.') ""
1518-- >>> splitSuffixBy'_ (== '.') "."
1521-- >>> splitSuffixBy'_ (== '.') "a"
1524-- >>> splitSuffixBy'_ (== '.') ".a"
1527-- >>> splitSuffixBy'_ (== '.') "a."
1530-- >>> splitSuffixBy'_ (== '.') "a.b"
1533-- >>> splitSuffixBy'_ (== '.') "a.b."
1536-- >>> splitSuffixBy'_ (== '.') "a..b.."
1537-- > ["a.",".","b.","."]
1539-- This can be considered as an n-fold version of 'breakPost' where we apply
1540-- 'breakPost' successively on the input stream.
1543{-# INLINE splitSuffixBy' #-}
1545 :: (IsStream t, Monad m)
1546 => (a -> Bool) -> Fold m a b -> t m a -> t m b
1547splitSuffixBy' predicate f m =
1548 D.fromStreamD $ D.splitSuffixBy' predicate f (D.toStreamD m)
1550------------------------------------------------------------------------------
1551-- Split on a delimiter sequence
1552------------------------------------------------------------------------------
1554-- Int list examples for splitOn:
1556-- >>> splitList [] [1,2,3,3,4]
1557-- > [[1],[2],[3],[3],[4]]
1559-- >>> splitList [5] [1,2,3,3,4]
1562-- >>> splitList [1] [1,2,3,3,4]
1565-- >>> splitList [4] [1,2,3,3,4]
1568-- >>> splitList [2] [1,2,3,3,4]
1571-- >>> splitList [3] [1,2,3,3,4]
1574-- >>> splitList [3,3] [1,2,3,3,4]
1577-- >>> splitList [1,2,3,3,4] [1,2,3,3,4]
1580-- | Split the stream on both sides of a separator sequence, dropping the
1583-- For illustration, let's define a function that operates on pure lists:
1586-- splitOn_ pat xs = S.toList $ FL.splitOn (A.fromList pat) (FL.toList) (S.fromList xs)
1589-- >>> splitOn_ "" "hello"
1590-- > ["h","e","l","l","o"]
1592-- >>> splitOn_ "hello" ""
1595-- >>> splitOn_ "hello" "hello"
1598-- >>> splitOn_ "x" "hello"
1601-- >>> splitOn_ "h" "hello"
1604-- >>> splitOn_ "o" "hello"
1607-- >>> splitOn_ "e" "hello"
1610-- >>> splitOn_ "l" "hello"
1613-- >>> splitOn_ "ll" "hello"
1616-- 'splitOn' is an inverse of 'intercalate'. The following law always holds:
1618-- > intercalate . splitOn == id
1620-- The following law holds when the separator is non-empty and contains none of
1621-- the elements present in the input lists:
1623-- > splitOn . intercalate == id
1625-- The following law always holds:
1627-- > concat . splitOn . intercalate == concat
1630{-# INLINE splitOn #-}
1632 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
1633 => Array a -> Fold m a b -> t m a -> t m b
1634splitOn patt f m = D.fromStreamD $ D.splitOn patt f (D.toStreamD m)
1637-- This can be implemented easily using Rabin Karp
1638-- | Split on any one of the given patterns.
1639{-# INLINE splitOnAny #-}
1641 :: (IsStream t, Monad m, Storable a, Integral a)
1642 => [Array a] -> Fold m a b -> t m a -> t m b
1643splitOnAny subseq f m = undefined -- D.fromStreamD $ D.splitOnAny f subseq (D.toStreamD m)
1646-- | Like 'splitSuffixBy' but the separator is a sequence of elements, instead
1647-- of a predicate for a single element.
1649-- > splitSuffixOn_ pat xs = S.toList $ FL.splitSuffixOn (A.fromList pat) (FL.toList) (S.fromList xs)
1651-- >>> splitSuffixOn_ "." ""
1654-- >>> splitSuffixOn_ "." "."
1657-- >>> splitSuffixOn_ "." "a"
1660-- >>> splitSuffixOn_ "." ".a"
1663-- >>> splitSuffixOn_ "." "a."
1666-- >>> splitSuffixOn_ "." "a.b"
1669-- >>> splitSuffixOn_ "." "a.b."
1672-- >>> splitSuffixOn_ "." "a..b.."
1675-- > lines = splitSuffixOn "\n"
1678{-# INLINE splitSuffixOn #-}
1680 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
1681 => Array a -> Fold m a b -> t m a -> t m b
1682splitSuffixOn patt f m =
1683 D.fromStreamD $ D.splitSuffixOn False patt f (D.toStreamD m)
1686-- | Like 'splitOn' but drops any empty splits.
1688{-# INLINE wordsOn #-}
1690 :: (IsStream t, Monad m, Storable a, Eq a)
1691 => Array a -> Fold m a b -> t m a -> t m b
1692wordsOn subseq f m = undefined -- D.fromStreamD $ D.wordsOn f subseq (D.toStreamD m)
1695-- XXX use a non-monadic intersperse to remove the MonadAsync constraint.
1697-- | Like 'splitOn' but splits the separator as well, as an infix token.
1699-- > splitOn'_ pat xs = S.toList $ FL.splitOn' (A.fromList pat) (FL.toList) (S.fromList xs)
1701-- >>> splitOn'_ "" "hello"
1702-- > ["h","","e","","l","","l","","o"]
1704-- >>> splitOn'_ "hello" ""
1707-- >>> splitOn'_ "hello" "hello"
1710-- >>> splitOn'_ "x" "hello"
1713-- >>> splitOn'_ "h" "hello"
1716-- >>> splitOn'_ "o" "hello"
1719-- >>> splitOn'_ "e" "hello"
1722-- >>> splitOn'_ "l" "hello"
1723-- > ["he","l","","l","o"]
1725-- >>> splitOn'_ "ll" "hello"
1729{-# INLINE splitOn' #-}
1731 :: (IsStream t, MonadAsync m, Storable a, Enum a, Eq a)
1732 => Array a -> Fold m a b -> t m a -> t m b
1733splitOn' patt f m = S.intersperseM (foldl' f (A.read patt)) $ splitOn patt f m
1735-- | Like 'splitSuffixOn' but keeps the suffix intact in the splits.
1737-- > splitSuffixOn'_ pat xs = S.toList $ FL.splitSuffixOn' (A.fromList pat) (FL.toList) (S.fromList xs)
1739-- >>> splitSuffixOn'_ "." ""
1742-- >>> splitSuffixOn'_ "." "."
1745-- >>> splitSuffixOn'_ "." "a"
1748-- >>> splitSuffixOn'_ "." ".a"
1751-- >>> splitSuffixOn'_ "." "a."
1754-- >>> splitSuffixOn'_ "." "a.b"
1757-- >>> splitSuffixOn'_ "." "a.b."
1760-- >>> splitSuffixOn'_ "." "a..b.."
1761-- > ["a.",".","b.","."]
1764{-# INLINE splitSuffixOn' #-}
1766 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
1767 => Array a -> Fold m a b -> t m a -> t m b
1768splitSuffixOn' patt f m =
1769 D.fromStreamD $ D.splitSuffixOn True patt f (D.toStreamD m)
1772-- This can be implemented easily using Rabin Karp
1773-- | Split post any one of the given patterns.
1774{-# INLINE splitSuffixOnAny #-}
1776 :: (IsStream t, Monad m, Storable a, Integral a)
1777 => [Array a] -> Fold m a b -> t m a -> t m b
1778splitSuffixOnAny subseq f m = undefined
1779 -- D.fromStreamD $ D.splitPostAny f subseq (D.toStreamD m)
1782------------------------------------------------------------------------------
1783-- Reorder in sequence
1784------------------------------------------------------------------------------
1787-- Buffer until the next element in sequence arrives. The function argument
1788-- determines the difference in sequence numbers. This could be useful in
1789-- implementing sequenced streams, for example, TCP reassembly.
1790{-# INLINE reassembleBy #-}
1792 :: (IsStream t, Monad m)
1797reassembleBy = undefined
1800------------------------------------------------------------------------------
1802------------------------------------------------------------------------------
1804-- | Distribute one copy of the stream to each fold and zip the results.
1807-- |-------Fold m a b--------|
1808-- ---stream m a---| |---m (b,c)
1809-- |-------Fold m a c--------|
1811-- >>> FL.foldl' (FL.tee FL.sum FL.length) (S.enumerateFromTo 1.0 100.0)
1816tee :: Monad m => Fold m a b -> Fold m a c -> Fold m a (b,c)
1817tee f1 f2 = (,) <$> f1 <*> f2
1819{-# INLINE foldNil #-}
1820foldNil :: Monad m => Fold m a [b]
1821foldNil = Fold step begin done where
1823 step _ _ = return []
1826{-# INLINE foldCons #-}
1827foldCons :: Monad m => Fold m a b -> Fold m a [b] -> Fold m a [b]
1828foldCons (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1829 Fold step begin done
1833 begin = Tuple' <$> beginL <*> beginR
1834 step (Tuple' xL xR) a = Tuple' <$> stepL xL a <*> stepR xR a
1835 done (Tuple' xL xR) = (:) <$> (doneL xL) <*> (doneR xR)
1837-- XXX use "List" instead of "[]"?, use Array for output to scale it to a large
1838-- number of consumers?
1840-- | Distribute one copy of the stream to each fold and collect the results in
1845-- |-------Fold m a b--------|
1846-- ---stream m a---| |---m [b]
1847-- |-------Fold m a b--------|
1852-- >>> FL.foldl' (FL.distribute [FL.sum, FL.length]) (S.enumerateFromTo 1 5)
1855-- This is the consumer side dual of the producer side 'sequence' operation.
1858{-# INLINE distribute #-}
1859distribute :: Monad m => [Fold m a b] -> Fold m a [b]
1860distribute [] = foldNil
1861distribute (x:xs) = foldCons x (distribute xs)
1863------------------------------------------------------------------------------
1865------------------------------------------------------------------------------
1867-- | Partition the input over two folds using an 'Either' partitioning
1872-- |-------Fold b x--------|
1873-- -----stream m a --> (Either b c)----| |----(x,y)
1874-- |-------Fold c y--------|
1877-- Send input to either fold randomly:
1879-- >>> import System.Random (randomIO)
1880-- >>> randomly a = randomIO >>= \x -> return $ if x then Left a else Right a
1881-- >>> FL.foldl' (FL.partitionByM randomly FL.length FL.length) (S.enumerateFromTo 1 100)
1884-- Send input to the two folds in a proportion of 2:1:
1887-- import Data.IORef (newIORef, readIORef, writeIORef)
1888-- proportionately m n = do
1889-- ref <- newIORef $ cycle $ concat [replicate m Left, replicate n Right]
1890-- return $ \\a -> do
1891-- r <- readIORef ref
1892-- writeIORef ref $ tail r
1896-- f <- proportionately 2 1
1897-- r <- FL.foldl' (FL.partitionByM f FL.length FL.length) (S.enumerateFromTo (1 :: Int) 100)
1904-- This is the consumer side dual of the producer side 'mergeBy' operation.
1907{-# INLINE partitionByM #-}
1908partitionByM :: Monad m
1909 => (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1910partitionByM f (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1912 Fold step begin done
1916 begin = Tuple' <$> beginL <*> beginR
1917 step (Tuple' xL xR) a = do
1920 Left b -> Tuple' <$> stepL xL b <*> return xR
1921 Right c -> Tuple' <$> return xL <*> stepR xR c
1922 done (Tuple' xL xR) = (,) <$> doneL xL <*> doneR xR
1924-- Note: we could use (a -> Bool) instead of (a -> Either b c), but the latter
1925-- makes the signature clearer as to which case belongs to which fold.
1926-- XXX need to check the performance in both cases.
1928-- | Same as 'partitionByM' but with a pure partition function.
1930-- Count even and odd numbers in a stream:
1933-- >>> let f = FL.partitionBy (\\n -> if even n then Left n else Right n)
1934-- (fmap (("Even " ++) . show) FL.length)
1935-- (fmap (("Odd " ++) . show) FL.length)
1936-- in FL.foldl' f (S.enumerateFromTo 1 100)
1937-- ("Even 50","Odd 50")
1941{-# INLINE partitionBy #-}
1942partitionBy :: Monad m
1943 => (a -> Either b c) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1944partitionBy f = partitionByM (return . f)
1946-- | Compose two folds such that the combined fold accepts a stream of 'Either'
1947-- and routes the 'Left' values to the first fold and 'Right' values to the
1950-- > partition = partitionBy id
1953{-# INLINE partition #-}
1955 => Fold m b x -> Fold m c y -> Fold m (Either b c) (x, y)
1956partition = partitionBy id
1959-- | Send one item to each fold in a round-robin fashion. This is the consumer
1960-- side dual of producer side 'mergeN' operation.
1962-- partitionN :: Monad m => [Fold m a b] -> Fold m a [b]
1963-- partitionN fs = Fold step begin done
1966-- Demultiplex an input element into a number of typed variants. We want to
1967-- statically restrict the target values within a set of predefined types, an
1968-- enumeration of a GADT. We also want to make sure that the Map contains only
1969-- those types and the full set of those types. Instead of Map it should
1970-- probably be a lookup-table using an array and not in GC memory.
1972-- This is the consumer side dual of the producer side 'mux' operation (XXX to
1975-- | Split the input stream based on a key field and fold each split using a
1976-- specific fold collecting the results in a map from the keys to the results.
1977-- Useful for cases like protocol handlers to handle different type of packets
1978-- using different handlers.
1982-- |-------Fold m a b
1983-- -----stream m a-----Map-----|
1984-- |-------Fold m a b
1990{-# INLINE demuxWith #-}
1991demuxWith :: (Monad m, Ord k)
1992 => (a -> k) -> Map k (Fold m a b) -> Fold m a (Map k b)
1993demuxWith f kv = Fold step initial extract
1999 -- XXX should we raise an exception in Nothing case?
2000 -- Ideally we should enforce that it is a total map over k so that look
2002 -- XXX we could use a monadic update function for a single lookup and
2003 -- update in the map.
2005 in case Map.lookup k mp of
2006 Nothing -> return mp
2007 Just (Fold step' acc extract') -> do
2008 !r <- acc >>= \x -> step' x a
2009 return $ Map.insert k (Fold step' (return r) extract') mp
2010 extract = Prelude.mapM (\(Fold _ acc e) -> acc >>= e)
2012-- | Fold a stream of key value pairs using a map of specific folds for each
2013-- key into a map from keys to the results of fold outputs of the corresponding
2017-- > let table = Data.Map.fromList [(\"SUM", FL.sum), (\"PRODUCT", FL.product)]
2018-- input = S.fromList [(\"SUM",1),(\"PRODUCT",2),(\"SUM",3),(\"PRODUCT",4)]
2019-- in FL.foldl' (FL.demux table) input
2026demux :: (Monad m, Ord k)
2027 => Map k (Fold m a b) -> Fold m (k, a) (Map k b)
2028demux fs = demuxWith fst (Map.map (lmap snd) fs)
2030-- | Split the input stream based on a key field and fold each split using a
2031-- specific fold without collecting the results. Useful for cases like protocol
2032-- handlers to handle different type of packets.
2036-- |-------Fold m a ()
2037-- -----stream m a-----Map-----|
2038-- |-------Fold m a ()
2046-- demuxWith_ can be slightly faster than demuxWith because we do not need to
2047-- update the Map in this case. This may be significant only if the map is
2049{-# INLINE demuxWith_ #-}
2050demuxWith_ :: (Monad m, Ord k)
2051 => (a -> k) -> Map k (Fold m a b) -> Fold m a ()
2052demuxWith_ f kv = Fold step initial extract
2057 Prelude.mapM (\(Fold s i e) ->
2058 i >>= \r -> return (Fold s (return r) e)) kv
2060 -- XXX should we raise an exception in Nothing case?
2061 -- Ideally we should enforce that it is a total map over k so that look
2063 case Map.lookup (f a) mp of
2064 Nothing -> return mp
2065 Just (Fold step' acc _) -> do
2066 _ <- acc >>= \x -> step' x a
2068 extract mp = Prelude.mapM (\(Fold _ acc e) -> acc >>= e) mp >> return ()
2070-- | Given a stream of key value pairs and a map from keys to folds, fold the
2071-- values for each key using the corresponding folds, discarding the outputs.
2074-- > let prn = FL.drainBy print
2075-- > let table = Data.Map.fromList [(\"ONE", prn), (\"TWO", prn)]
2076-- input = S.fromList [(\"ONE",1),(\"TWO",2)]
2077-- in FL.foldl' (FL.demux_ table) input
2083{-# INLINE demux_ #-}
2084demux_ :: (Monad m, Ord k) => Map k (Fold m a ()) -> Fold m (k, a) ()
2085demux_ fs = demuxWith_ fst (Map.map (lmap snd) fs)
2087-- XXX instead of a Map we could yield the results as a pure stream as they
2088-- complete. We could then concatMap the fold results to implement the
2089-- windowing combinators.
2091-- | Split the input stream based on a key field and fold each split using the
2092-- given fold. Useful for map/reduce, bucketizing the input in different bins
2093-- or for generating histograms.
2096-- > let input = S.fromList [(\"ONE",1),(\"ONE",1.1),(\"TWO",2), (\"TWO",2.2)]
2097-- in FL.foldl' (FL.classify FL.toListRev) input
2098-- fromList [(\"ONE",[1.1,1.0]),(\"TWO",[2.2,2.0])]
2102{-# INLINE classifyWith #-}
2103classifyWith :: (Monad m, Ord k) => (a -> k) -> Fold m a b -> Fold m a (Map k b)
2104classifyWith f (Fold step initial extract) = Fold step' initial' extract'
2108 initial' = return Map.empty
2111 in case Map.lookup k kv of
2115 return $ Map.insert k r kv
2118 return $ Map.insert k r kv
2119 extract' = Prelude.mapM extract
2121-- | Given an input stream of key value pairs and a fold for values, fold all
2122-- the values belonging to each key. Useful for map/reduce, bucketizing the
2123-- input in different bins or for generating histograms.
2126-- > let input = S.fromList [(\"ONE",1),(\"ONE",1.1),(\"TWO",2), (\"TWO",2.2)]
2127-- in FL.foldl' (FL.classify FL.toListRev) input
2128-- fromList [(\"ONE",[1.1,1.0]),(\"TWO",[2.2,2.0])]
2133-- > classify fld = classifyWith fst (lmap snd fld)
2136{-# INLINE classify #-}
2137classify :: (Monad m, Ord k) => Fold m a b -> Fold m (k, a) (Map k b)
2138classify fld = classifyWith fst (lmap snd fld)
2140------------------------------------------------------------------------------
2142------------------------------------------------------------------------------
2144-- | Like 'unzipWith' but with a monadic splitter function.
2147{-# INLINE unzipWithM #-}
2148unzipWithM :: Monad m
2149 => (a -> m (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
2150unzipWithM f (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
2151 Fold step begin done
2155 step (Tuple' xL xR) a = do
2157 Tuple' <$> stepL xL b <*> stepR xR c
2158 begin = Tuple' <$> beginL <*> beginR
2159 done (Tuple' xL xR) = (,) <$> doneL xL <*> doneR xR
2161-- | Split elements in the input stream into two parts using a pure splitter
2162-- function, direct each part to a different fold and zip the results.
2165{-# INLINE unzipWith #-}
2167 => (a -> (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
2168unzipWith f = unzipWithM (return . f)
2170-- | Send the elements of tuples in a stream of tuples through two different
2175-- |-------Fold a x--------|
2176-- -----Stream m x----(a,b)--| |----m (x,y)
2177-- |-------Fold b y--------|
2181-- This is the consumer side dual of the producer side 'zip' operation.
2185unzip :: Monad m => Fold m a x -> Fold m b y -> Fold m (a,b) (x,y)
2188------------------------------------------------------------------------------
2190------------------------------------------------------------------------------
2192-- | Modify the fold such that when the fold is done, instead of returning the
2193-- accumulator, it returns a fold. The returned fold starts from where we left
2194-- i.e. it uses the last accumulator value as the initial value of the
2195-- accumulator. Thus we can resume the fold later and feed it more input.
2198-- > more <- FL.foldl (FL.duplicate FL.sum) (S.enumerateFromTo 1 10)
2199-- > evenMore <- FL.foldl (FL.duplicate more) (S.enumerateFromTo 11 20)
2200-- > FL.foldl evenMore (S.enumerateFromTo 21 30)
2204{-# INLINABLE duplicate #-}
2205duplicate :: Applicative m => Fold m a b -> Fold m a (Fold m a b)
2206duplicate (Fold step begin done) =
2207 Fold step begin (\x -> pure (Fold step (pure x) done))
2210-- All the stream flattening transformations can also be applied to a fold
2213-- | This can be used to apply all the stream generation operations on folds.
2214lconcatMap ::(IsStream t, Monad m) => (a -> t m b)
2217lconcatMap s f1 f2 = undefined
2220-- All the grouping transformation that we apply to a stream can also be
2221-- applied to a fold input stream.
2224-- | Group the input stream into groups of elements between @low@ and @high@.
2225-- Collection starts in chunks of @low@ and then keeps doubling until we reach
2226-- @high@. Each chunk is folded using the provided fold function.
2228-- This could be useful, for example, when we are folding a stream of unknown
2229-- size to a stream of arrays and we want to minimize the number of
2234-- XXX we should be able to implement it with parsers/terminating folds.
2236{-# INLINE lchunksInRange #-}
2237lchunksInRange :: Monad m
2238 => Int -> Int -> Fold m a b -> Fold m b c -> Fold m a c
2239lchunksInRange low high (Fold step1 initial1 extract1)
2240 (Fold step2 initial2 extract2) = undefined
2243-- | Group the input stream into groups of @n@ elements each and then fold each
2244-- group using the provided fold function.
2248-- -----Fold m a b----|-Fold n a c-|-Fold n a c-|-...-|----Fold m a c
2252{-# INLINE lchunksOf #-}
2253lchunksOf :: Monad m => Int -> Fold m a b -> Fold m b c -> Fold m a c
2254lchunksOf n (Fold step1 initial1 extract1) (Fold step2 initial2 extract2) =
2255 Fold step' initial' extract'
2259 initial' = (Tuple3' 0) <$> initial1 <*> initial2
2260 step' (Tuple3' i r1 r2) a = do
2264 return $ Tuple3' (i + 1) res r2
2268 res <- step2 r2 res1
2269 return $ Tuple3' 0 i1 res
2270 extract' (Tuple3' _ _ r) = extract2 r
2272-- | Group the input stream into windows of n second each and then fold each
2273-- group using the provided fold function.
2275-- For example, we can copy and distribute a stream to multiple folds where
2276-- each fold can group the input differently e.g. by one second, one minute and
2277-- one hour windows respectively and fold each resulting stream of folds.
2281-- -----Fold m a b----|-Fold n a c-|-Fold n a c-|-...-|----Fold m a c
2284{-# INLINE lsessionsOf #-}
2285lsessionsOf :: MonadAsync m => Double -> Fold m a b -> Fold m b c -> Fold m a c
2286lsessionsOf n (Fold step1 initial1 extract1) (Fold step2 initial2 extract2) =
2287 Fold step' initial' extract'
2291 -- XXX MVar may be expensive we need a cheaper synch mechanism here
2295 mv1 <- liftIO $ newMVar i1
2296 mv2 <- liftIO $ newMVar (Right i2)
2297 t <- control $ \run ->
2298 mask $ \restore -> do
2299 tid <- forkIO $ catch (restore $ void $ run (timerThread mv1 mv2))
2300 (handleChildException mv2)
2302 return $ Tuple3' t mv1 mv2
2303 step' acc@(Tuple3' _ mv1 _) a = do
2304 r1 <- liftIO $ takeMVar mv1
2306 liftIO $ putMVar mv1 res
2308 extract' (Tuple3' tid _ mv2) = do
2309 r2 <- liftIO $ takeMVar mv2
2310 liftIO $ killThread tid
2313 Right x -> extract2 x
2315 timerThread mv1 mv2 = do
2316 liftIO $ threadDelay (round $ n * 1000000)
2318 r1 <- liftIO $ takeMVar mv1
2320 liftIO $ putMVar mv1 i1
2323 r2 <- liftIO $ takeMVar mv2
2326 Right x -> fmap Right $ step2 x res1
2327 liftIO $ putMVar mv2 res
2330 handleChildException ::
2331 MVar (Either SomeException a) -> SomeException -> IO ()
2332 handleChildException mv2 e = do
2339------------------------------------------------------------------------------
2340-- Windowed classification
2341------------------------------------------------------------------------------
2343-- We divide the stream into windows or chunks in space or time and each window
2344-- can be associated with a key, all events associated with a particular key in
2345-- the window can be folded to a single result. The stream can be split into
2346-- windows by size or by using a split predicate on the elements in the stream.
2347-- For example, when we receive a closing flag, we can close the window.
2349-- A "chunk" is a space window and a "session" is a time window. Are there any
2350-- other better short words to describe them. An alternative is to use
2351-- "swindow" and "twindow". Another word for "session" could be "spell".
2353-- TODO: To mark the position in space or time we can have Indexed or
2354-- TimeStamped types. That can make it easy to deal with the position indices
2357------------------------------------------------------------------------------
2358-- Keyed Sliding Windows
2359------------------------------------------------------------------------------
2362{-# INLINABLE classifySlidingChunks #-}
2363classifySlidingChunks
2364 :: (IsStream t, MonadAsync m, Ord k)
2365 => Int -- ^ window size
2366 -> Int -- ^ window slide
2367 -> Fold m a b -- ^ Fold to be applied to window events
2368 -> t m (k, a, Bool) -- ^ window key, data, close event
2370classifySlidingChunks wsize wslide (Fold step initial extract) str
2373-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
2374-- slide the send window when an ack is received and we slide the receive
2375-- window when a sequence is complete. Sliding is stateful in case of TCP,
2376-- sliding releases the send buffer or makes data available to the user from
2377-- the receive buffer.
2378{-# INLINABLE classifySlidingSessions #-}
2379classifySlidingSessions
2380 :: (IsStream t, MonadAsync m, Ord k)
2381 => Double -- ^ timer tick in seconds
2382 -> Double -- ^ time window size
2383 -> Double -- ^ window slide
2384 -> Fold m a b -- ^ Fold to be applied to window events
2385 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
2387classifySlidingSessions tick interval slide (Fold step initial extract) str
2391------------------------------------------------------------------------------
2392-- Sliding Window Buffers
2393------------------------------------------------------------------------------
2395-- These buffered versions could be faster than concurrent incremental folds of
2396-- all overlapping windows as in many cases we may not need all the values to
2397-- compute the fold, we can just compute the result using the old value and new
2398-- value. However, we may need the buffer once in a while, for example for
2399-- string search we usually compute the hash incrementally but when the hash
2400-- matches the hash of the pattern we need to compare the whole string.
2402-- XXX we should be able to implement sequence based splitting combinators
2403-- using this combinator.
2406-- | Buffer n elements of the input in a ring buffer. When t new elements are
2407-- collected, slide the window to remove the same number of oldest elements,
2408-- insert the new elements, and apply an incremental fold on the sliding
2409-- window, supplying the outgoing elements, the new ring buffer as arguments.
2411 :: (IsStream t, Monad m, Ord a, Storable a)
2412 => Int -- window size
2413 -> Int -- window slide
2414 -> Fold m (Ring a, Array a) b
2417slidingChunkBuffer = undefined
2419-- Buffer n seconds worth of stream elements of the input in a radix tree.
2420-- Every t seconds, remove the items that are older than n seconds, and apply
2421-- an incremental fold on the sliding window, supplying the outgoing elements,
2422-- and the new radix tree buffer as arguments.
2424 :: (IsStream t, Monad m, Ord a, Storable a)
2425 => Int -- window size
2427 -> Fold m (RTree a, Array a) b
2430slidingSessionBuffer = undefined
2433------------------------------------------------------------------------------
2434-- Keyed Session Windows
2435------------------------------------------------------------------------------
2438-- | Keyed variable size space windows. Close the window if we do not receive a
2439-- window event in the next "spaceout" elements.
2440{-# INLINABLE classifyChunksBy #-}
2442 :: (IsStream t, MonadAsync m, Ord k)
2443 => Int -- ^ window spaceout (spread)
2444 -> Bool -- ^ reset the spaceout when a chunk window element is received
2445 -> Fold m a b -- ^ Fold to be applied to chunk window elements
2446 -> t m (k, a, Bool) -- ^ chunk key, data, last element
2448classifyChunksBy spanout reset (Fold step initial extract) str = undefined
2450-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
2451-- received within the chunk size window. The chunk gets closed only if no
2452-- element is received within the chunk window.
2454{-# INLINABLE classifyKeepAliveChunks #-}
2455classifyKeepAliveChunks
2456 :: (IsStream t, MonadAsync m, Ord k)
2457 => Int -- ^ window spaceout (spread)
2458 -> Fold m a b -- ^ Fold to be applied to chunk window elements
2459 -> t m (k, a, Bool) -- ^ chunk key, data, last element
2461classifyKeepAliveChunks spanout = classifyChunksBy spanout True
2464-- | @classifySessionsBy tick timeout reset f stream@ groups together all input
2465-- stream elements that belong to the same session. @timeout@ is the maximum
2466-- lifetime of a session in seconds. All elements belonging to a session are
2467-- purged after this duration. If "reset" is 'Ture' then the timeout is reset
2468-- after every event received in the session. Session duration is measured
2469-- using the timestamp of the first element seen for that session. To detect
2470-- session timeouts, a monotonic event time clock is maintained using the
2471-- timestamps seen in the inputs and a timer with a tick duration specified by
2474-- @session key@ is a key that uniquely identifies the session for the given
2475-- element, @timestamp@ characterizes the time when the input element was
2476-- generated, this is an absolute time measured from some @Epoch@. @session
2477-- close@ is a boolean indicating whether this element marks the closing of the
2478-- session. When an input element with @session close@ set to @True@ is seen
2479-- the session is purged immediately.
2481-- All the input elements belonging to a session are collected using the fold
2482-- @f@. The session key and the fold result are emitted in the output stream
2483-- when the session is purged either via the session close event or via the
2484-- session liftime timeout.
2487{-# INLINABLE classifySessionsBy #-}
2489 :: (IsStream t, MonadAsync m, Ord k)
2490 => Double -- ^ timer tick in seconds
2491 -> Double -- ^ session timeout
2492 -> Bool -- ^ reset the timeout when an event is received
2493 -> Fold m a b -- ^ Fold to be applied to session events
2494 -> t m (k, a, Bool, AbsTime) -- ^ session key, timestamp, close event, data
2496classifySessionsBy tick timeout reset (Fold step initial extract) str =
2497 S.concatMap (\(Tuple4' _ _ _ s) -> s) $ S.scanlM' sstep szero stream
2501 timeoutMs = toRelTime (round (timeout * 1000) :: MilliSecond64)
2502 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
2503 szero = Tuple4' (toAbsTime (0 :: MilliSecond64)) H.empty Map.empty S.nil
2505 -- Got a new stream input element
2506 sstep (Tuple4' evTime hp mp _) (Just (key, a, closing, ts)) =
2507 -- XXX we should use a heap in pinned memory to scale it to a large
2510 -- deleting a key from the heap is expensive, so we never delete a
2511 -- key, we just purge it from the Map and it gets purged from the
2512 -- heap on timeout. We just need an extra lookup in the Map when
2513 -- the key is purged from the heap, that should not be expensive.
2515 -- To detect session inactivity we keep a timestamp of the latest event
2516 -- in the Map along with the fold result. When we purge the session
2517 -- from the heap we match the timestamp in the heap with the timestamp
2518 -- in the Map, if the latest timestamp is newer and has not expired we
2519 -- reinsert the key in the heap.
2521 -- XXX if the key is an Int, we can also use an IntMap for slightly
2522 -- better performance.
2524 let accumulate v = do
2525 Tuple' _ old <- maybe (initial >>= return . Tuple' ts) return v
2527 return $ Tuple' ts new
2530 let (r, mp') = Map.updateLookupWithKey (\_ _ -> Nothing) key mp
2531 Tuple' _ acc <- accumulate r
2533 return $ Tuple4' evTime hp mp' (S.yield (key, res))
2535 let r = Map.lookup key mp
2537 let mp' = Map.insert key acc mp
2541 let expiry = addToAbsTime ts timeoutMs
2542 in H.insert (Entry expiry key) hp
2544 -- Event time is maintained as monotonically increasing
2545 -- time. If we have lagged behind any of the timestamps
2546 -- seen then we increase it to match the latest time seen
2547 -- in the timestamps. We also increase it on timer ticks.
2548 return $ Tuple4' (max evTime ts) hp' mp' S.nil
2550 -- Got a timer tick event
2551 -- XXX can we yield the entries without accumulating them?
2552 sstep (Tuple4' evTime heap sessions _) Nothing = do
2553 (hp', mp', out) <- go heap sessions S.nil
2554 return $ Tuple4' curTime hp' mp' out
2558 curTime = addToAbsTime evTime tickMs
2560 let hres = H.uncons hp
2562 Just (Entry ts key, hp') -> do
2563 let duration = diffAbsTime curTime ts
2564 if duration >= timeoutMs
2566 let (r, mp') = Map.updateLookupWithKey
2567 (\_ _ -> Nothing) key mp
2569 Nothing -> go hp' mp' out
2570 Just (Tuple' latestTS acc) -> do
2571 let dur = diffAbsTime curTime latestTS
2572 if dur >= timeoutMs || not reset
2575 go hp' mp' ((key, sess) `S.cons` out)
2577 -- reset the session timeout
2578 let expiry = addToAbsTime latestTS timeoutMs
2579 hp'' = H.insert (Entry expiry key) hp'
2580 mp'' = Map.insert key (Tuple' latestTS acc) mp'
2582 else return (hp, mp, out)
2583 Nothing -> return (hp, mp, out)
2585 -- merge timer events in the stream
2586 stream = S.map Just str `parallel` S.repeatM timer
2588 liftIO $ threadDelay (round $ tick * 1000000)
2591-- | Like 'classifySessionsOf' but the session is kept alive if an event is
2592-- received within the session window. The session times out and gets closed
2593-- only if no event is received within the specified session window size.
2596{-# INLINABLE classifyKeepAliveSessions #-}
2597classifyKeepAliveSessions
2598 :: (IsStream t, MonadAsync m, Ord k)
2599 => Double -- ^ session inactive timeout
2600 -> Fold m a b -- ^ Fold to be applied to session payload data
2601 -> t m (k, a, Bool, AbsTime) -- ^ session key, data, close flag, timestamp
2603classifyKeepAliveSessions timeout = classifySessionsBy 1 timeout True
2605------------------------------------------------------------------------------
2606-- Keyed tumbling windows
2607------------------------------------------------------------------------------
2609-- Tumbling windows is a special case of sliding windows where the window slide
2610-- is the same as the window size. Or it can be a special case of session
2611-- windows where the reset flag is set to False.
2613-- XXX instead of using the early termination flag in the stream, we can use an
2614-- early terminating fold instead.
2617-- | Split the stream into fixed size chunks of specified size. Within each
2618-- such chunk fold the elements in buckets identified by the keys. A particular
2619-- bucket fold can be terminated early if a closing flag is encountered in an
2620-- element for that key.
2623{-# INLINABLE classifyChunksOf #-}
2625 :: (IsStream t, MonadAsync m, Ord k)
2626 => Int -- ^ window size
2627 -> Fold m a b -- ^ Fold to be applied to window events
2628 -> t m (k, a, Bool) -- ^ window key, data, close event
2630classifyChunksOf wsize = classifyChunksBy wsize False
2633-- | Split the stream into fixed size time windows of specified interval in
2634-- seconds. Within each such window, fold the elements in buckets identified by
2635-- the keys. A particular bucket fold can be terminated early if a closing flag
2636-- is encountered in an element for that key. Once a fold is terminated the key
2637-- and value for that bucket are emitted in the output stream.
2639-- Session @timestamp@ in the input stream is an absolute time from some epoch,
2640-- characterizing the time when the input element was generated. To detect
2641-- session window end, a monotonic event time clock is maintained synced with
2642-- the timestamps with a clock resolution of 1 second.
2645{-# INLINABLE classifySessionsOf #-}
2647 :: (IsStream t, MonadAsync m, Ord k)
2648 => Double -- ^ time window size
2649 -> Fold m a b -- ^ Fold to be applied to window events
2650 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
2652classifySessionsOf interval = classifySessionsBy 1 interval False
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list