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
165 -- ** Mapping Filters
169 -- ** Scanning Filters
174 -- | Insertion adds more elements to the stream.
214 -- | Streams can be split into segments in space or in time. We use the
215 -- term @chunk@ to refer to a spatial length of the stream (spatial window)
216 -- and the term @session@ to refer to a length in time (time window).
218 -- In imperative terms, grouped folding can be considered as a nested loop
219 -- where we loop over the stream to group elements and then loop over
220 -- individual groups to fold them to a single value that is yielded in the
245 -- Keeping the delimiters
250 -- Splitting by multiple sequences
252 -- , splitSuffixOnAny
253 -- , splitPrefixOnAny
262 -- The 'Applicative' instance of a distributing 'Fold' distributes one copy
263 -- of the stream to each fold and combines the results using a function.
267 -- |-------Fold m a b--------|
268 -- ---stream m a---| |---m (b,c,...)
269 -- |-------Fold m a c--------|
274 -- To compute the average of numbers in a stream without going throught he
277 -- >>> let avg = (/) <$> FL.sum <*> fmap fromIntegral FL.length
278 -- >>> FL.foldl' avg (S.enumerateFromTo 1.0 100.0)
281 -- The 'Semigroup' and 'Monoid' instances of a distributing fold distribute
282 -- the input to both the folds and combines the outputs using Monoid or
283 -- Semigroup instances of the output types:
285 -- >>> import Data.Monoid (Sum)
286 -- >>> FL.foldl' (FL.head <> FL.last) (fmap Sum $ S.enumerateFromTo 1.0 100.0)
287 -- Just (Sum {getSum = 101.0})
289 -- The 'Num', 'Floating', and 'Fractional' instances work in the same way.
296 -- Direct items in the input stream to different folds using a function to
297 -- select the fold. This is useful to demultiplex the input stream.
314 -- These can be expressed using lmap/lmapM and unzip
321 , duplicate -- experimental
323 -- * Windowed Classification
324 -- | Split the stream into windows or chunks in space or time. Each window
325 -- can be associated with a key, all events associated with a particular
326 -- key in the window can be folded to a single result. The stream is split
327 -- into windows of specified size, the window can be terminated early if
328 -- the closing flag is specified in the input stream.
330 -- The term "chunk" is used for a space window and the term "session" is
331 -- used for a time window.
333 -- ** Tumbling Windows
334 -- | A new window starts after the previous window is finished.
335 -- , classifyChunksOf
338 -- ** Keep Alive Windows
339 -- | The window size is extended if an event arrives within the specified
340 -- window size. This can represent sessions with idle or inactive timeout.
341 -- , classifyKeepAliveChunks
342 , classifyKeepAliveSessions
345 -- ** Sliding Windows
346 -- | A new window starts after the specified slide from the previous
347 -- window. Therefore windows can overlap.
348 , classifySlidingChunks
349 , classifySlidingSessions
351 -- ** Sliding Window Buffers
352 -- , slidingChunkBuffer
353 -- , slidingSessionBuffer
357import Control.Concurrent (threadDelay, forkIO, killThread)
358import Control.Concurrent.MVar (MVar, newMVar, takeMVar, putMVar)
359import Control.Exception (SomeException(..), catch, mask)
360import Control.Monad (void)
361import Control.Monad.Catch (throwM)
362import Control.Monad.IO.Class (MonadIO(..))
363import Control.Monad.Trans.Control (control)
364import Data.Functor.Identity (Identity)
365import Data.Heap (Entry(..))
366import Data.Map.Strict (Map)
367import Data.Maybe (fromJust, isJust, isNothing)
369import Foreign.Storable (Storable(..))
371 hiding (filter, drop, dropWhile, take, takeWhile, zipWith, foldr,
372 foldl, map, mapM_, sequence, all, any, sum, product, elem,
373 notElem, maximum, minimum, head, last, tail, length, null,
374 reverse, iterate, init, and, or, lookup, foldr1, (!!),
375 scanl, scanl1, replicate, concatMap, mconcat, foldMap, unzip,
376 span, splitAt, break, mapM)
378import qualified Data.Heap as H
379import qualified Data.Map.Strict as Map
380import qualified Prelude
382import Streamly (MonadAsync, parallel)
383import Streamly.Fold.Types (Fold(..))
384import Streamly.Pipe.Types (Pipe (..), PipeState(..))
385import Streamly.Mem.Array.Types (Array)
386-- import Streamly.Mem.Ring (Ring)
387import Streamly.Streams.Serial (SerialT)
388import Streamly.Streams.StreamK (IsStream())
389import Streamly.Time.Units
390 (AbsTime, MilliSecond64(..), addToAbsTime, diffAbsTime, toRelTime,
393import Streamly.Strict
395import qualified Streamly.Pipe.Types as Pipe
396import qualified Streamly.Mem.Array.Types as A
397import qualified Streamly.Prelude as S
398import qualified Streamly.Streams.StreamD as D
399import qualified Streamly.Streams.StreamK as K
400import qualified Streamly.Streams.Prelude as P
404-- We can use the left folds in this module instead of the folds in
405-- "Streamly.Prelude". For example the following two ways of folding are
406-- equivalent in functionality and performance,
408-- >>> FL.foldl' FL.sum (S.enumerateFromTo 1 100)
410-- >>> S.sum (S.enumerateFromTo 1 100)
413-- However, left folds are push type folds. That means we push the entire input
414-- to a fold before we can get the output. Therefore, the performance is
415-- equivalent only for full folds like 'sum' and 'length'. For partial folds
416-- like 'head' or 'any' the folds in "Streamly.Prelude" may be much more
417-- efficient because they are implemented as right folds that terminate as soon
418-- as we get the result. Note that when a full fold is composed with a partial
419-- fold in parallel the performance is not impacted as we anyway have to
420-- consume the whole stream due to the full fold.
422-- >>> S.head (1 `S.cons` undefined)
424-- >>> FL.foldl' FL.head (1 `S.cons` undefined)
425-- *** Exception: Prelude.undefined
427-- However, we can wrap the fold in a scan to convert it into a lazy stream of
428-- fold steps. We can then terminate the stream whenever we want. For example,
430-- >>> S.toList $ S.take 1 $ FL.scanl' FL.head (1 `S.cons` undefined)
433-- The following example extracts the input stream up to a point where the
434-- running average of elements is no more than 10:
438-- $ S.map (fromJust . fst)
439-- $ S.takeWhile (\\(_,x) -> x <= 10)
440-- $ FL.postscanl' ((,) \<$> FL.last \<*> avg) (S.enumerateFromTo 1.0 100.0)
443-- [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]
446------------------------------------------------------------------------------
448------------------------------------------------------------------------------
450-- XXX rename to foldStream to be consistent with foldArray for arrays? Should
451-- we move it in the stream module (Prelude as of now) or should we move
452-- foldArray as well here?
454-- | Fold a stream using the supplied monadic fold.
456-- >>> FL.foldl' FL.sum (S.enumerateFromTo 1 100)
461foldl' :: Monad m => Fold m a b -> SerialT m a -> m b
462foldl' (Fold step begin done) = P.foldlMx' step begin done
464------------------------------------------------------------------------------
465-- Transformations on fold inputs
466------------------------------------------------------------------------------
468-- | Convert a fold that generates a monadic action to a fold that generates
469-- the output of that monadic action.
472{-# INLINE sequence #-}
473sequence :: Monad m => Fold m a (m b) -> Fold m a b
474sequence (Fold step initial extract) = Fold step initial extract'
480-- | Convert a fold that generates a monadic action to a fold that generates
481-- the output of that monadic action.
485mapM :: Monad m => (b -> m c) -> Fold m a b -> Fold m a c
486mapM f = sequence . fmap f
488------------------------------------------------------------------------------
489-- Transformations on fold inputs
490------------------------------------------------------------------------------
492-- | Apply a transformation on a 'Fold' using a 'Pipe'.
495{-# INLINE transform #-}
496transform :: Monad m => Pipe m a b -> Fold m b c -> Fold m a c
497transform (Pipe pstep1 pstep2 pinitial) (Fold fstep finitial fextract) =
498 Fold step initial extract
502 initial = Tuple' <$> return pinitial <*> finitial
503 step (Tuple' ps fs) x = do
509 go acc (Pipe.Yield b (Consume ps')) = do
511 return (Tuple' ps' acc')
513 go acc (Pipe.Yield b (Produce ps')) = do
518 go acc (Pipe.Continue (Consume ps')) = return (Tuple' ps' acc)
520 go acc (Pipe.Continue (Produce ps')) = do
524 extract (Tuple' _ fs) = fextract fs
526-- | @(lmap f fold)@ maps the function @f@ on the input of the fold.
528-- >>> FL.foldl' (FL.lmap (\x -> x * x) FL.sum) (S.enumerateFromTo 1 100)
532{-# INLINABLE lmap #-}
533lmap :: (a -> b) -> Fold m b r -> Fold m a r
534lmap f (Fold step begin done) = Fold step' begin done
536 step' x a = step x (f a)
538-- | @(lmapM f fold)@ maps the monadic function @f@ on the input of the fold.
541{-# INLINABLE lmapM #-}
542lmapM :: Monad m => (a -> m b) -> Fold m b r -> Fold m a r
543lmapM f (Fold step begin done) = Fold step' begin done
545 step' x a = f a >>= step x
547------------------------------------------------------------------------------
549------------------------------------------------------------------------------
551-- | @_Fold1 step@ returns a new 'Fold' using just a step function that has the
552-- same type for the accumulator and the element. The result type is the
553-- accumulator type wrapped in 'Maybe'. The initial accumulator is retrieved
554-- from the 'Foldable', the result is 'None' for empty containers.
555{-# INLINABLE _Fold1 #-}
556_Fold1 :: Monad m => (a -> a -> a) -> Fold m a (Maybe a)
557_Fold1 step = Fold step_ (return Nothing') fromStrictMaybe
559 step_ mx a = return $ Just' $
564------------------------------------------------------------------------------
566------------------------------------------------------------------------------
568------------------------------------------------------------------------------
570------------------------------------------------------------------------------
572-- | A fold that drains all its input, running the effects and discarding the
576{-# INLINABLE drain #-}
577drain :: Monad m => Fold m a ()
578drain = Fold step begin done
585-- > drainBy f = lmapM f drain
587-- Drain all input after passing it through a monadic function. This is the
588-- dual of mapM_ on stream producers.
589{-# INLINABLE drainBy #-}
590drainBy :: Monad m => (a -> m b) -> Fold m a ()
591drainBy f = Fold (const (void . f)) (return ()) return
593-- | Extract the last element of the input stream, if any.
596{-# INLINABLE last #-}
597last :: Monad m => Fold m a (Maybe a)
598last = _Fold1 (flip const)
600------------------------------------------------------------------------------
602------------------------------------------------------------------------------
604-- | Like 'length', except with a more general 'Num' return value
607{-# INLINABLE genericLength #-}
608genericLength :: (Monad m, Num b) => Fold m a b
609genericLength = Fold (\n _ -> return $ n + 1) (return 0) return
611-- | Determine the length of the input stream.
614{-# INLINABLE length #-}
615length :: Monad m => Fold m a Int
616length = genericLength
618-- | Determine the sum of all elements of a stream of numbers. Returns additive
619-- identity (@0@) when the stream is empty. Note that this is not numerically
620-- stable for floating point numbers.
624sum :: (Monad m, Num a) => Fold m a a
625sum = Fold (\x a -> return $ x + a) (return 0) return
627-- | Determine the product of all elements of a stream of numbers. Returns
628-- multiplicative identity (@1@) when the stream is empty.
631{-# INLINABLE product #-}
632product :: (Monad m, Num a) => Fold m a a
633product = Fold (\x a -> return $ x * a) (return 1) return
635------------------------------------------------------------------------------
637------------------------------------------------------------------------------
639-- | Determine the maximum element in a stream using the supplied comparison
643{-# INLINABLE maximumBy #-}
644maximumBy :: Monad m => (a -> a -> Ordering) -> Fold m a (Maybe a)
645maximumBy cmp = _Fold1 max'
647 max' x y = case cmp x y of
653-- maximum = 'maximumBy' compare
656-- Determine the maximum element in a stream.
659{-# INLINABLE maximum #-}
660maximum :: (Monad m, Ord a) => Fold m a (Maybe a)
663-- | Computes the minimum element with respect to the given comparison function
664{-# INLINABLE minimumBy #-}
665minimumBy :: Monad m => (a -> a -> Ordering) -> Fold m a (Maybe a)
666minimumBy cmp = _Fold1 min'
668 min' x y = case cmp x y of
672-- | Determine the minimum element in a stream using the supplied comparison
676{-# INLINABLE minimum #-}
677minimum :: (Monad m, Ord a) => Fold m a (Maybe a)
680------------------------------------------------------------------------------
681-- To Summary (Statistical)
682------------------------------------------------------------------------------
684-- | Compute a numerically stable arithmetic mean of all elements in the input
688{-# INLINABLE mean #-}
689mean :: (Monad m, Fractional a) => Fold m a a
690mean = Fold step (return begin) (return . done)
693 step (Tuple' x n) y = return $
695 in Tuple' (x + (y - x) / n') n'
696 done (Tuple' x _) = x
698-- | Compute a numerically stable (population) variance over all elements in
702{-# INLINABLE variance #-}
703variance :: (Monad m, Fractional a) => Fold m a a
704variance = Fold step (return begin) (return . done)
706 begin = Tuple3' 0 0 0
708 step (Tuple3' n mean_ m2) x = return $ Tuple3' n' mean' m2'
711 mean' = (n * mean_ + x) / (n + 1)
713 m2' = m2 + delta * delta * n / (n + 1)
715 done (Tuple3' n _ m2) = m2 / n
717-- | Compute a numerically stable (population) standard deviation over all
718-- elements in the input stream.
721{-# INLINABLE stdDev #-}
722stdDev :: (Monad m, Floating a) => Fold m a a
723stdDev = sqrt variance
725------------------------------------------------------------------------------
726-- Monoidal left folds
727------------------------------------------------------------------------------
729-- | Fold an input stream consisting of monoidal elements using 'mappend'
732-- > FL.foldl FL.mconcat (S.map Sum $ S.enumerateFromTo 1 10)
735{-# INLINABLE mconcat #-}
736mconcat :: (Monad m, Monoid a) => Fold m a a
737mconcat = Fold (\x a -> return $ mappend x a) (return mempty) return
740-- > foldMap f = map f mconcat
742-- Make a fold from a pure function that folds the output of the function
743-- using 'mappend' and 'mempty'.
745-- > FL.foldl (FL.foldMap Sum) $ S.enumerateFromTo 1 10
748{-# INLINABLE foldMap #-}
749foldMap :: (Monad m, Monoid b) => (a -> b) -> Fold m a b
750foldMap f = lmap f mconcat
753-- > foldMapM f = mapM f mconcat
755-- Make a fold from a monadic function that folds the output of the function
756-- using 'mappend' and 'mempty'.
758-- > FL.foldM (FL.foldMapM (return . Sum)) $ S.enumerateFromTo 1 10
761{-# INLINABLE foldMapM #-}
762foldMapM :: (Monad m, Monoid b) => (a -> m b) -> Fold m a b
763foldMapM act = Fold step begin done
766 begin = return mempty
769 return $! mappend m m'
771------------------------------------------------------------------------------
773------------------------------------------------------------------------------
775-- | Folds the input stream to a list.
777-- /Warning!/ working on large lists accumulated as buffers in memory could be
778-- very inefficient, consider using "Streamly.Array" instead.
782-- id . (x1 :) . (x2 :) . (x3 :) . ... . (xn :) $ []
783{-# INLINABLE toList #-}
784toList :: Monad m => Fold m a [a]
785toList = Fold (\f x -> return $ f . (x :))
789-- This is more efficient than 'toList'. toList is exactly the same as
790-- reversing the list after toListRev.
792-- | Buffers the input stream to a list in the reverse order of the input.
794-- /Warning!/ working on large lists accumulated as buffers in memory could be
795-- very inefficient, consider using "Streamly.Array" instead.
799-- xn : ... : x2 : x1 : []
800{-# INLINABLE toListRev #-}
801toListRev :: Monad m => Fold m a [a]
802toListRev = Fold (\xs x -> return $ x:xs) (return []) return
804-- | A fold that buffers its input to a pure stream.
806-- /Warning!/ working on large streams accumulated as buffers in memory could
807-- be very inefficient, consider using "Streamly.Array" instead.
810{-# INLINE toStream #-}
811toStream :: Monad m => Fold m a (SerialT Identity a)
812toStream = Fold (\f x -> return $ f . (x `K.cons`))
816-- This is more efficient than 'toStream'. toStream is exactly the same as
817-- reversing the stream after toStreamRev.
819-- | Buffers the input stream to a pure stream in the reverse order of the
822-- /Warning!/ working on large streams accumulated as buffers in memory could
823-- be very inefficient, consider using "Streamly.Array" instead.
827-- xn : ... : x2 : x1 : []
828{-# INLINABLE toStreamRev #-}
829toStreamRev :: Monad m => Fold m a (SerialT Identity a)
830toStreamRev = Fold (\xs x -> return $ x `S.cons` xs) (return S.nil) return
832------------------------------------------------------------------------------
834------------------------------------------------------------------------------
836------------------------------------------------------------------------------
838------------------------------------------------------------------------------
840-- | Like 'index', except with a more general 'Integral' argument
843{-# INLINABLE genericIndex #-}
844genericIndex :: (Integral i, Monad m) => i -> Fold m a (Maybe a)
845genericIndex i = Fold step (return $ Left' 0) done
858-- | Lookup the element at the given index.
861{-# INLINABLE index #-}
862index :: Monad m => Int -> Fold m a (Maybe a)
865-- | Extract the first element of the stream, if any.
868{-# INLINABLE head #-}
869head :: Monad m => Fold m a (Maybe a)
872-- | Returns the first element that satisfies the given predicate.
875{-# INLINABLE find #-}
876find :: Monad m => (a -> Bool) -> Fold m a (Maybe a)
877find predicate = Fold step (return Nothing') fromStrictMaybe
881 Nothing' -> if predicate a
886-- | In a stream of (key-value) pairs @(a, b)@, return the value @b@ of the
887-- first pair where the key equals the given value @a@.
890{-# INLINABLE lookup #-}
891lookup :: (Eq a, Monad m) => a -> Fold m (a,b) (Maybe b)
892lookup a0 = Fold step (return Nothing') fromStrictMaybe
894 step x (a,b) = return $
896 Nothing' -> if a == a0
901-- | Convert strict 'Either'' to lazy 'Maybe'
902{-# INLINABLE hush #-}
903hush :: Either' a b -> Maybe b
904hush (Left' _) = Nothing
905hush (Right' b) = Just b
907-- | Returns the first index that satisfies the given predicate.
910{-# INLINABLE findIndex #-}
911findIndex :: Monad m => (a -> Bool) -> Fold m a (Maybe Int)
912findIndex predicate = Fold step (return $ Left' 0) (return . hush)
922-- | Returns the first index where a given value is found in the stream.
925{-# INLINABLE elemIndex #-}
926elemIndex :: (Eq a, Monad m) => a -> Fold m a (Maybe Int)
927elemIndex a = findIndex (a ==)
929------------------------------------------------------------------------------
931------------------------------------------------------------------------------
933-- | Return 'True' if the input stream is empty.
936{-# INLINABLE null #-}
937null :: Monad m => Fold m a Bool
938null = Fold (\_ _ -> return False) (return True) return
941-- > any p = lmap p or
943-- | Returns 'True' if any of the elements of a stream satisfies a predicate.
947any :: Monad m => (a -> Bool) -> Fold m a Bool
948any predicate = Fold (\x a -> return $ x || predicate a) (return False) return
950-- | Return 'True' if the given element is present in the stream.
953{-# INLINABLE elem #-}
954elem :: (Eq a, Monad m) => a -> Fold m a Bool
958-- > all p = lmap p and
960-- | Returns 'True' if all elements of a stream satisfy a predicate.
964all :: Monad m => (a -> Bool) -> Fold m a Bool
965all predicate = Fold (\x a -> return $ x && predicate a) (return True) return
967-- | Returns 'True' if the given element is not present in the stream.
970{-# INLINABLE notElem #-}
971notElem :: (Eq a, Monad m) => a -> Fold m a Bool
972notElem a = all (a /=)
974-- | Returns 'True' if all elements are 'True', 'False' otherwise
978and :: Monad m => Fold m Bool Bool
979and = Fold (\x a -> return $ x && a) (return True) return
981-- | Returns 'True' if any element is 'True', 'False' otherwise
985or :: Monad m => Fold m Bool Bool
986or = Fold (\x a -> return $ x || a) (return False) return
988------------------------------------------------------------------------------
989-- Scanning with a Fold
990------------------------------------------------------------------------------
992-- | Scan a stream using the given monadic fold.
996scanl' :: Monad m => Fold m a b -> SerialT m a -> SerialT m b
997scanl' (Fold step begin done) = P.scanlMx' step begin done
999-- | Postscan a stream using the given monadic fold.
1002{-# INLINE postscanl' #-}
1003postscanl' :: Monad m => Fold m a b -> SerialT m a -> SerialT m b
1004postscanl' (Fold step begin done) = P.postscanlMx' step begin done
1008------------------------------------------------------------------------------
1010------------------------------------------------------------------------------
1012-- | Include only those elements that pass a predicate.
1014-- >>> FL.foldl (lfilter (> 5) FL.sum) [1..10]
1018{-# INLINABLE lfilter #-}
1019lfilter :: Monad m => (a -> Bool) -> Fold m a r -> Fold m a r
1020lfilter f (Fold step begin done) = Fold step' begin done
1022 step' x a = if f a then step x a else return x
1024-- | Like 'lfilter' but with a monadic predicate.
1027{-# INLINABLE lfilterM #-}
1028lfilterM :: Monad m => (a -> m Bool) -> Fold m a r -> Fold m a r
1029lfilterM f (Fold step begin done) = Fold step' begin done
1033 if use then step x a else return x
1035-- | Take first 'n' elements from the stream and discard the rest.
1038{-# INLINABLE ltake #-}
1039ltake :: Monad m => Int -> Fold m a b -> Fold m a b
1040ltake n (Fold step initial done) = Fold step' initial' done'
1042 initial' = fmap (Tuple' 0) initial
1043 step' (Tuple' i r) a = do
1047 return $ Tuple' (i + 1) res
1048 else return $ Tuple' i r
1049 done' (Tuple' _ r) = done r
1051-- | Takes elements from the input as long as the predicate succeeds.
1054{-# INLINABLE ltakeWhile #-}
1055ltakeWhile :: Monad m => (a -> Bool) -> Fold m a b -> Fold m a b
1056ltakeWhile predicate (Fold step initial done) = Fold step' initial' done'
1058 initial' = fmap Left' initial
1059 step' (Left' r) a = do
1061 then fmap Left' $ step r a
1062 else return (Right' r)
1063 step' r _ = return r
1064 done' (Left' r) = done r
1065 done' (Right' r) = done r
1067------------------------------------------------------------------------------
1068-- Grouping/Splitting
1069------------------------------------------------------------------------------
1071------------------------------------------------------------------------------
1072-- Grouping without looking at elements
1073------------------------------------------------------------------------------
1075------------------------------------------------------------------------------
1077------------------------------------------------------------------------------
1080-- | @splitAt n f1 f2@ composes folds @f1@ and @f2@ such that first @n@
1081-- elements of its input are sent to fold @f1@ and the rest of the stream is
1082-- sent to fold @f2@.
1084-- > let splitAt_ n xs = FL.foldl' (FL.splitAt n FL.toList FL.toList) $ S.fromList xs
1086-- >>> splitAt_ 6 "Hello World!"
1087-- > ("Hello ","World!")
1089-- >>> splitAt_ (-1) [1,2,3]
1092-- >>> splitAt_ 0 [1,2,3]
1095-- >>> splitAt_ 1 [1,2,3]
1098-- >>> splitAt_ 3 [1,2,3]
1101-- >>> splitAt_ 4 [1,2,3]
1104-- This can be considered as a two-fold version of 'ltake' where we take both
1105-- the segments instead of discarding the leftover.
1108{-# INLINE splitAt #-}
1115splitAt n (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
1116 Fold step init extract
1118 init = Tuple3' <$> return n <*> initialL <*> initialR
1120 step (Tuple3' i xL xR) input =
1122 then stepL xL input >>= (\a -> return (Tuple3' (i - 1) a xR))
1123 else stepR xR input >>= (\b -> return (Tuple3' i xL b))
1125 extract (Tuple3' _ a b) = (,) <$> extractL a <*> extractR b
1127------------------------------------------------------------------------------
1129------------------------------------------------------------------------------
1131------------------------------------------------------------------------------
1132-- Generalized grouping
1133------------------------------------------------------------------------------
1135-- This combinator is the most general grouping combinator and can be used to
1136-- implement all other grouping combinators.
1138-- XXX check if this can implement the splitOn combinator i.e. we can slide in
1139-- new elements, slide out old elements and incrementally compute the hash.
1140-- Also, can we implement the windowed classification combinators using this?
1142-- In fact this is a parse. Instead of using a special return value in the fold
1143-- we are using a mapping function.
1145-- Note that 'scanl'' (usually followed by a map to extract the desired value
1146-- from the accumulator) can be used to realize many implementations e.g. a
1147-- sliding window implementation. A scan followed by a mapMaybe is also a good
1148-- pattern to express many problems where we want to emit a filtered output and
1149-- not emit an output on every input.
1151-- Passing on of the initial accumulator value to the next fold is equivalent
1152-- to returning the leftover concept.
1155-- | @groupScan splitter fold stream@ folds the input stream using @fold@.
1156-- @splitter@ is applied on the accumulator of the fold every time an item is
1157-- consumed by the fold. The fold continues until @splitter@ returns a 'Just'
1158-- value. A 'Just' result from the @splitter@ specifies a result to be emitted
1159-- in the output stream and the initial value of the accumulator for the next
1160-- group's fold. This allows us to control whether to start fresh for the next
1161-- fold or to continue from the previous fold's output.
1163{-# INLINE groupScan #-}
1165 :: (IsStream t, Monad m)
1166 => (x -> m (Maybe (b, x))) -> Fold m a x -> t m a -> t m b
1167groupScan split fold m = undefined
1170-- | Group the input stream into groups of @n@ elements each and then fold each
1171-- group using the provided fold function.
1173-- >> S.toList $ FL.chunksOf 2 FL.sum (S.enumerateFromTo 1 10)
1176-- This can be considered as an n-fold version of 'ltake' where we apply
1177-- 'ltake' repeatedly on the leftover stream until the stream exhausts.
1180{-# INLINE chunksOf #-}
1182 :: (IsStream t, Monad m)
1183 => Int -> Fold m a b -> t m a -> t m b
1184chunksOf n f m = D.fromStreamD $ D.groupsOf n f (D.toStreamD m)
1186-- | Transform a fold from a pure input to a 'Maybe' input, consuming only
1188{-# INLINE lcatMaybes #-}
1189lcatMaybes :: Monad m => Fold m a b -> Fold m (Maybe a) b
1190lcatMaybes = lfilter isJust . lmap fromJust
1192-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
1193-- XXX add this example after fixing the serial stream rate control
1194-- >>> S.toList $ S.take 5 $ sessionsOf 1 FL.sum $ constRate 2 $ S.enumerateFrom 1
1197-- | Group the input stream into windows of @n@ second each and then fold each
1198-- group using the provided fold function.
1201{-# INLINE sessionsOf #-}
1203 :: (IsStream t, MonadAsync m)
1204 => Double -> Fold m a b -> t m a -> t m b
1206 splitSuffixBy' isNothing (lcatMaybes f)
1207 (S.intersperseByTime n (return Nothing) (S.map Just xs))
1209------------------------------------------------------------------------------
1210-- Element Aware APIs
1211------------------------------------------------------------------------------
1213------------------------------------------------------------------------------
1215------------------------------------------------------------------------------
1217-- | Break the input stream into two groups, the first group takes the input as
1218-- long as the predicate applied to the first element of the stream and next
1219-- input element holds 'True', the second group takes the rest of the input.
1227_spanBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
1228 Fold step init extract
1231 init = Tuple3' <$> initialL <*> initialR <*> return (Tuple' Nothing True)
1233 step (Tuple3' a b (Tuple' (Just frst) isFirstG)) input =
1234 if cmp frst input && isFirstG
1236 >>= (\a' -> return (Tuple3' a' b (Tuple' (Just frst) isFirstG)))
1238 >>= (\a' -> return (Tuple3' a a' (Tuple' Nothing False)))
1240 step (Tuple3' a b (Tuple' Nothing isFirstG)) input =
1243 >>= (\a' -> return (Tuple3' a' b (Tuple' (Just input) isFirstG)))
1245 >>= (\a' -> return (Tuple3' a a' (Tuple' Nothing False)))
1247 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
1249-- | Span as long as the predicate is 'True'. @span p f1 f2@ composes folds
1250-- @f1@ and @f2@ such that the composed fold continues sending the input to
1251-- @f1@ as long as the predicate @p@ is 'True'. The rest of the input is sent
1254-- > let span_ p xs = FL.foldl' (FL.span p FL.toList FL.toList) $ S.fromList xs
1256-- >>> span_ (< 1) [1,2,3]
1259-- >>> span_ (< 2) [1,2,3]
1262-- >>> span_ (< 4) [1,2,3]
1265-- This can be considered as a two-fold version of 'ltakeWhile' where we take
1266-- both the segments instead of discarding the leftover.
1276span p (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
1277 Fold step init extract
1280 init = Tuple3' <$> initialL <*> initialR <*> return True
1282 step (Tuple3' a b isFirstG) input =
1283 if isFirstG && p input
1284 then stepL a input >>= (\a' -> return (Tuple3' a' b True))
1285 else stepR b input >>= (\a' -> return (Tuple3' a a' False))
1287 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
1290-- > break p = span (not . p)
1292-- Break as soon as the predicate becomes 'True'. @break p f1 f2@ composes
1293-- folds @f1@ and @f2@ such that @f1@ stops receiving input as soon as the
1294-- predicate @p@ becomes 'True'. The rest of the input is sent to @f2@.
1296-- This is the binary version of 'splitBy'.
1298-- > let break_ p xs = FL.foldl' (FL.break p FL.toList FL.toList) $ S.fromList xs
1300-- >>> break_ (< 1) [3,2,1]
1303-- >>> break_ (< 2) [3,2,1]
1306-- >>> break_ (< 4) [3,2,1]
1317break p = span (not . p)
1319-- | Like 'spanBy' but applies the predicate in a rolling fashion i.e.
1320-- predicate is applied to the previous and the next input elements.
1321{-# INLINE _spanRollingBy #-}
1328_spanRollingBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
1329 Fold step init extract
1332 init = Tuple3' <$> initialL <*> initialR <*> return Nothing
1334 step (Tuple3' a b (Just frst)) input =
1336 then stepL a input >>= (\a' -> return (Tuple3' a' b (Just input)))
1337 else stepR b input >>= (\b' -> return (Tuple3' a b' (Just input)))
1339 step (Tuple3' a b Nothing) input =
1340 stepL a input >>= (\a' -> return (Tuple3' a' b (Just input)))
1342 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
1344------------------------------------------------------------------------------
1346------------------------------------------------------------------------------
1348-- | @groupsBy cmp f $ S.fromList [a,b,c,...]@ assigns the element @a@ to the
1349-- first group, if @a \`cmp` b@ is 'True' then @b@ is also assigned to the same
1350-- group. If @a \`cmp` c@ is 'True' then @c@ is also assigned to the same
1351-- group and so on. When the comparison fails a new group is started. Each
1352-- group is folded using the fold @f@.
1354-- >>> S.toList $ FL.groupsBy (>) FL.toList $ S.fromList [1,3,7,0,2,5]
1355-- > [[1,3,7],[0,2,5]]
1358{-# INLINE groupsBy #-}
1360 :: (IsStream t, Monad m)
1365groupsBy cmp f m = D.fromStreamD $ D.groupsBy cmp f (D.toStreamD m)
1367-- | Unlike @groupsBy@ this function performs a rolling comparison of two
1368-- successive elements in the input stream. @groupsRollingBy cmp f $ S.fromList
1369-- [a,b,c,...]@ assigns the element @a@ to the first group, if @a \`cmp` b@ is
1370-- 'True' then @b@ is also assigned to the same group. If @b \`cmp` c@ is
1371-- 'True' then @c@ is also assigned to the same group and so on. When the
1372-- comparison fails a new group is started. Each group is folded using the fold
1375-- >>> S.toList $ FL.groupsRollingBy (\a b -> a + 1 == b) FL.toList $ S.fromList [1,2,3,7,8,9]
1376-- > [[1,2,3],[7,8,9]]
1379{-# INLINE groupsRollingBy #-}
1381 :: (IsStream t, Monad m)
1386groupsRollingBy cmp f m = D.fromStreamD $ D.groupsRollingBy cmp f (D.toStreamD m)
1389-- > groups = groupsBy (==)
1390-- > groups = groupsRollingBy (==)
1392-- Groups a contiguous span of equal elements together in one group.
1394-- >>> S.toList $ FL.groups FL.toList $ S.fromList [1,1,2,2]
1398groups :: (IsStream t, Monad m, Eq a) => Fold m a b -> t m a -> t m b
1399groups = groupsBy (==)
1401------------------------------------------------------------------------------
1402-- Binary splitting on a separator
1403------------------------------------------------------------------------------
1406-- | Find the first occurrence of the specified sequence in the input stream
1407-- and break the input stream into two parts, the first part consisting of the
1408-- stream before the sequence and the second part consisting of the sequence
1409-- and the rest of the stream.
1411-- > let breakOn_ pat xs = FL.foldl' (FL.breakOn pat FL.toList FL.toList) $ S.fromList xs
1413-- >>> breakOn_ "dear" "Hello dear world!"
1414-- > ("Hello ","dear world!")
1416{-# INLINE breakOn #-}
1417breakOn :: Monad m => Array a -> Fold m a b -> Fold m a c -> Fold m a (b,c)
1418breakOn pat f m = undefined
1421------------------------------------------------------------------------------
1422-- N-ary split on a predicate
1423------------------------------------------------------------------------------
1425-- TODO: Use a Splitter configuration similar to the "split" package to make it
1426-- possible to express all splitting combinations. In general, we can have
1427-- infix/suffix/prefix/condensing of separators, dropping both leading/trailing
1428-- separators. We can have a single split operation taking the splitter config
1431-- | Split a stream on separator elements determined by a predicate, dropping
1432-- the separator. Separators are not considered part of stream segments on
1433-- either side of it instead they are treated as infixed between two stream
1434-- segments. For example, with @.@ as separator, @"a.b.c"@ would be parsed as
1435-- @["a","b","c"]@. When @.@ is in leading or trailing position it is still
1436-- considered as infixed, treating the first or the last segment as empty. For
1437-- example, @".a."@ would be parsed as @["","a",""]@. This operation is
1438-- opposite of 'intercalate'.
1440-- Let's use the following definition for illustration:
1442-- > splitBy_ p xs = S.toList $ FL.splitBy p (FL.toList) (S.fromList xs)
1444-- >>> splitBy_ (== '.') ""
1447-- >>> splitBy_ (== '.') "."
1450-- >>> splitBy_ (== '.') ".a"
1453-- >>> splitBy_ (== '.') "a."
1456-- >>> splitBy_ (== '.') "a.b"
1459-- >>> splitBy_ (== '.') "a..b"
1462-- This can be considered as an n-fold version of 'break' where we apply
1463-- 'break' successively on the input stream, dropping the first element
1464-- of the second segment after each break.
1467{-# INLINE splitBy #-}
1469 :: (IsStream t, Monad m)
1470 => (a -> Bool) -> Fold m a b -> t m a -> t m b
1471splitBy predicate f m =
1472 D.fromStreamD $ D.splitBy predicate f (D.toStreamD m)
1474-- | Like 'splitBy' but the separator is treated as part of the previous
1475-- stream segment (suffix). Therefore, when the separator is in trailing
1476-- position, no empty segment is considered to follow it. For example, @"a.b."@
1477-- would be parsed as @["a","b"]@ instead of @["a","b",""]@ as in the case of
1480-- > splitSuffixBy_ p xs = S.toList $ FL.splitSuffixBy p (FL.toList) (S.fromList xs)
1482-- >>> splitSuffixBy_ (== '.') ""
1485-- >>> splitSuffixBy_ (== '.') "."
1488-- >>> splitSuffixBy_ (== '.') "a"
1491-- >>> splitSuffixBy_ (== '.') ".a"
1494-- >>> splitSuffixBy_ (== '.') "a."
1497-- >>> splitSuffixBy_ (== '.') "a.b"
1500-- >>> splitSuffixBy_ (== '.') "a.b."
1503-- >>> splitSuffixBy_ (== '.') "a..b.."
1506-- > lines = splitSuffixBy (== '\n')
1508-- This can be considered as an n-fold version of 'breakPost' where we apply
1509-- 'breakPost' successively on the input stream, dropping the first element
1510-- of the second segment after each break.
1513{-# INLINE splitSuffixBy #-}
1515 :: (IsStream t, Monad m)
1516 => (a -> Bool) -> Fold m a b -> t m a -> t m b
1517splitSuffixBy predicate f m =
1518 D.fromStreamD $ D.splitSuffixBy predicate f (D.toStreamD m)
1520-- | Like 'splitBy' but ignores repeated separators or separators in leading
1521-- or trailing position. Therefore, @"..a..b.."@ would be parsed as
1522-- @["a","b"]@. In other words, it treats the input like words separated by
1523-- whitespace elements determined by the predicate.
1525-- > wordsBy' p xs = S.toList $ FL.wordsBy p (FL.toList) (S.fromList xs)
1527-- >>> wordsBy' (== ',') ""
1530-- >>> wordsBy' (== ',') ","
1533-- >>> wordsBy' (== ',') ",a,,b,"
1536-- > words = wordsBy isSpace
1539{-# INLINE wordsBy #-}
1541 :: (IsStream t, Monad m)
1542 => (a -> Bool) -> Fold m a b -> t m a -> t m b
1543wordsBy predicate f m =
1544 D.fromStreamD $ D.wordsBy predicate f (D.toStreamD m)
1546-- XXX we should express this using the Splitter config.
1548-- We can get splitSuffixBy' by appending the suffix to the output segments
1549-- produced by splitSuffixBy. However, it may add an additional suffix if the last
1550-- fragment did not have a suffix in the first place.
1552-- | Like 'splitSuffixBy' but keeps the suffix in the splits.
1554-- > splitSuffixBy'_ p xs = S.toList $ FL.splitSuffixBy' p (FL.toList) (S.fromList xs)
1556-- >>> splitSuffixBy'_ (== '.') ""
1559-- >>> splitSuffixBy'_ (== '.') "."
1562-- >>> splitSuffixBy'_ (== '.') "a"
1565-- >>> splitSuffixBy'_ (== '.') ".a"
1568-- >>> splitSuffixBy'_ (== '.') "a."
1571-- >>> splitSuffixBy'_ (== '.') "a.b"
1574-- >>> splitSuffixBy'_ (== '.') "a.b."
1577-- >>> splitSuffixBy'_ (== '.') "a..b.."
1578-- > ["a.",".","b.","."]
1580-- This can be considered as an n-fold version of 'breakPost' where we apply
1581-- 'breakPost' successively on the input stream.
1584{-# INLINE splitSuffixBy' #-}
1586 :: (IsStream t, Monad m)
1587 => (a -> Bool) -> Fold m a b -> t m a -> t m b
1588splitSuffixBy' predicate f m =
1589 D.fromStreamD $ D.splitSuffixBy' predicate f (D.toStreamD m)
1591------------------------------------------------------------------------------
1592-- Split on a delimiter sequence
1593------------------------------------------------------------------------------
1595-- Int list examples for splitOn:
1597-- >>> splitList [] [1,2,3,3,4]
1598-- > [[1],[2],[3],[3],[4]]
1600-- >>> splitList [5] [1,2,3,3,4]
1603-- >>> splitList [1] [1,2,3,3,4]
1606-- >>> splitList [4] [1,2,3,3,4]
1609-- >>> splitList [2] [1,2,3,3,4]
1612-- >>> splitList [3] [1,2,3,3,4]
1615-- >>> splitList [3,3] [1,2,3,3,4]
1618-- >>> splitList [1,2,3,3,4] [1,2,3,3,4]
1621-- | Split the stream on both sides of a separator sequence, dropping the
1624-- For illustration, let's define a function that operates on pure lists:
1627-- splitOn_ pat xs = S.toList $ FL.splitOn (A.fromList pat) (FL.toList) (S.fromList xs)
1630-- >>> splitOn_ "" "hello"
1631-- > ["h","e","l","l","o"]
1633-- >>> splitOn_ "hello" ""
1636-- >>> splitOn_ "hello" "hello"
1639-- >>> splitOn_ "x" "hello"
1642-- >>> splitOn_ "h" "hello"
1645-- >>> splitOn_ "o" "hello"
1648-- >>> splitOn_ "e" "hello"
1651-- >>> splitOn_ "l" "hello"
1654-- >>> splitOn_ "ll" "hello"
1657-- 'splitOn' is an inverse of 'intercalate'. The following law always holds:
1659-- > intercalate . splitOn == id
1661-- The following law holds when the separator is non-empty and contains none of
1662-- the elements present in the input lists:
1664-- > splitOn . intercalate == id
1666-- The following law always holds:
1668-- > concat . splitOn . intercalate == concat
1671{-# INLINE splitOn #-}
1673 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
1674 => Array a -> Fold m a b -> t m a -> t m b
1675splitOn patt f m = D.fromStreamD $ D.splitOn patt f (D.toStreamD m)
1678-- This can be implemented easily using Rabin Karp
1679-- | Split on any one of the given patterns.
1680{-# INLINE splitOnAny #-}
1682 :: (IsStream t, Monad m, Storable a, Integral a)
1683 => [Array a] -> Fold m a b -> t m a -> t m b
1684splitOnAny subseq f m = undefined -- D.fromStreamD $ D.splitOnAny f subseq (D.toStreamD m)
1687-- | Like 'splitSuffixBy' but the separator is a sequence of elements, instead
1688-- of a predicate for a single element.
1690-- > splitSuffixOn_ pat xs = S.toList $ FL.splitSuffixOn (A.fromList pat) (FL.toList) (S.fromList xs)
1692-- >>> splitSuffixOn_ "." ""
1695-- >>> splitSuffixOn_ "." "."
1698-- >>> splitSuffixOn_ "." "a"
1701-- >>> splitSuffixOn_ "." ".a"
1704-- >>> splitSuffixOn_ "." "a."
1707-- >>> splitSuffixOn_ "." "a.b"
1710-- >>> splitSuffixOn_ "." "a.b."
1713-- >>> splitSuffixOn_ "." "a..b.."
1716-- > lines = splitSuffixOn "\n"
1719{-# INLINE splitSuffixOn #-}
1721 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
1722 => Array a -> Fold m a b -> t m a -> t m b
1723splitSuffixOn patt f m =
1724 D.fromStreamD $ D.splitSuffixOn False patt f (D.toStreamD m)
1727-- | Like 'splitOn' but drops any empty splits.
1729{-# INLINE wordsOn #-}
1731 :: (IsStream t, Monad m, Storable a, Eq a)
1732 => Array a -> Fold m a b -> t m a -> t m b
1733wordsOn subseq f m = undefined -- D.fromStreamD $ D.wordsOn f subseq (D.toStreamD m)
1736-- XXX use a non-monadic intersperse to remove the MonadAsync constraint.
1738-- | Like 'splitOn' but splits the separator as well, as an infix token.
1740-- > splitOn'_ pat xs = S.toList $ FL.splitOn' (A.fromList pat) (FL.toList) (S.fromList xs)
1742-- >>> splitOn'_ "" "hello"
1743-- > ["h","","e","","l","","l","","o"]
1745-- >>> splitOn'_ "hello" ""
1748-- >>> splitOn'_ "hello" "hello"
1751-- >>> splitOn'_ "x" "hello"
1754-- >>> splitOn'_ "h" "hello"
1757-- >>> splitOn'_ "o" "hello"
1760-- >>> splitOn'_ "e" "hello"
1763-- >>> splitOn'_ "l" "hello"
1764-- > ["he","l","","l","o"]
1766-- >>> splitOn'_ "ll" "hello"
1770{-# INLINE splitOn' #-}
1772 :: (IsStream t, MonadAsync m, Storable a, Enum a, Eq a)
1773 => Array a -> Fold m a b -> t m a -> t m b
1774splitOn' patt f m = S.intersperseM (foldl' f (A.read patt)) $ splitOn patt f m
1776-- | Like 'splitSuffixOn' but keeps the suffix intact in the splits.
1778-- > splitSuffixOn'_ pat xs = S.toList $ FL.splitSuffixOn' (A.fromList pat) (FL.toList) (S.fromList xs)
1780-- >>> splitSuffixOn'_ "." ""
1783-- >>> splitSuffixOn'_ "." "."
1786-- >>> splitSuffixOn'_ "." "a"
1789-- >>> splitSuffixOn'_ "." ".a"
1792-- >>> splitSuffixOn'_ "." "a."
1795-- >>> splitSuffixOn'_ "." "a.b"
1798-- >>> splitSuffixOn'_ "." "a.b."
1801-- >>> splitSuffixOn'_ "." "a..b.."
1802-- > ["a.",".","b.","."]
1805{-# INLINE splitSuffixOn' #-}
1807 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
1808 => Array a -> Fold m a b -> t m a -> t m b
1809splitSuffixOn' patt f m =
1810 D.fromStreamD $ D.splitSuffixOn True patt f (D.toStreamD m)
1813-- This can be implemented easily using Rabin Karp
1814-- | Split post any one of the given patterns.
1815{-# INLINE splitSuffixOnAny #-}
1817 :: (IsStream t, Monad m, Storable a, Integral a)
1818 => [Array a] -> Fold m a b -> t m a -> t m b
1819splitSuffixOnAny subseq f m = undefined
1820 -- D.fromStreamD $ D.splitPostAny f subseq (D.toStreamD m)
1823------------------------------------------------------------------------------
1824-- Reorder in sequence
1825------------------------------------------------------------------------------
1828-- Buffer until the next element in sequence arrives. The function argument
1829-- determines the difference in sequence numbers. This could be useful in
1830-- implementing sequenced streams, for example, TCP reassembly.
1831{-# INLINE reassembleBy #-}
1833 :: (IsStream t, Monad m)
1838reassembleBy = undefined
1841------------------------------------------------------------------------------
1843------------------------------------------------------------------------------
1845-- | Distribute one copy of the stream to each fold and zip the results.
1848-- |-------Fold m a b--------|
1849-- ---stream m a---| |---m (b,c)
1850-- |-------Fold m a c--------|
1852-- >>> FL.foldl' (FL.tee FL.sum FL.length) (S.enumerateFromTo 1.0 100.0)
1857tee :: Monad m => Fold m a b -> Fold m a c -> Fold m a (b,c)
1858tee f1 f2 = (,) <$> f1 <*> f2
1860{-# INLINE foldNil #-}
1861foldNil :: Monad m => Fold m a [b]
1862foldNil = Fold step begin done where
1864 step _ _ = return []
1867{-# INLINE foldCons #-}
1868foldCons :: Monad m => Fold m a b -> Fold m a [b] -> Fold m a [b]
1869foldCons (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1870 Fold step begin done
1874 begin = Tuple' <$> beginL <*> beginR
1875 step (Tuple' xL xR) a = Tuple' <$> stepL xL a <*> stepR xR a
1876 done (Tuple' xL xR) = (:) <$> (doneL xL) <*> (doneR xR)
1878-- XXX use "List" instead of "[]"?, use Array for output to scale it to a large
1879-- number of consumers?
1881-- | Distribute one copy of the stream to each fold and collect the results in
1886-- |-------Fold m a b--------|
1887-- ---stream m a---| |---m [b]
1888-- |-------Fold m a b--------|
1893-- >>> FL.foldl' (FL.distribute [FL.sum, FL.length]) (S.enumerateFromTo 1 5)
1896-- This is the consumer side dual of the producer side 'sequence' operation.
1899{-# INLINE distribute #-}
1900distribute :: Monad m => [Fold m a b] -> Fold m a [b]
1901distribute [] = foldNil
1902distribute (x:xs) = foldCons x (distribute xs)
1904------------------------------------------------------------------------------
1906------------------------------------------------------------------------------
1908-- | Partition the input over two folds using an 'Either' partitioning
1913-- |-------Fold b x--------|
1914-- -----stream m a --> (Either b c)----| |----(x,y)
1915-- |-------Fold c y--------|
1918-- Send input to either fold randomly:
1920-- >>> import System.Random (randomIO)
1921-- >>> randomly a = randomIO >>= \x -> return $ if x then Left a else Right a
1922-- >>> FL.foldl' (FL.partitionByM randomly FL.length FL.length) (S.enumerateFromTo 1 100)
1925-- Send input to the two folds in a proportion of 2:1:
1928-- import Data.IORef (newIORef, readIORef, writeIORef)
1929-- proportionately m n = do
1930-- ref <- newIORef $ cycle $ concat [replicate m Left, replicate n Right]
1931-- return $ \\a -> do
1932-- r <- readIORef ref
1933-- writeIORef ref $ tail r
1937-- f <- proportionately 2 1
1938-- r <- FL.foldl' (FL.partitionByM f FL.length FL.length) (S.enumerateFromTo (1 :: Int) 100)
1945-- This is the consumer side dual of the producer side 'mergeBy' operation.
1948{-# INLINE partitionByM #-}
1949partitionByM :: Monad m
1950 => (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1951partitionByM f (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1953 Fold step begin done
1957 begin = Tuple' <$> beginL <*> beginR
1958 step (Tuple' xL xR) a = do
1961 Left b -> Tuple' <$> stepL xL b <*> return xR
1962 Right c -> Tuple' <$> return xL <*> stepR xR c
1963 done (Tuple' xL xR) = (,) <$> doneL xL <*> doneR xR
1965-- Note: we could use (a -> Bool) instead of (a -> Either b c), but the latter
1966-- makes the signature clearer as to which case belongs to which fold.
1967-- XXX need to check the performance in both cases.
1969-- | Same as 'partitionByM' but with a pure partition function.
1971-- Count even and odd numbers in a stream:
1974-- >>> let f = FL.partitionBy (\\n -> if even n then Left n else Right n)
1975-- (fmap (("Even " ++) . show) FL.length)
1976-- (fmap (("Odd " ++) . show) FL.length)
1977-- in FL.foldl' f (S.enumerateFromTo 1 100)
1978-- ("Even 50","Odd 50")
1982{-# INLINE partitionBy #-}
1983partitionBy :: Monad m
1984 => (a -> Either b c) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1985partitionBy f = partitionByM (return . f)
1987-- | Compose two folds such that the combined fold accepts a stream of 'Either'
1988-- and routes the 'Left' values to the first fold and 'Right' values to the
1991-- > partition = partitionBy id
1994{-# INLINE partition #-}
1996 => Fold m b x -> Fold m c y -> Fold m (Either b c) (x, y)
1997partition = partitionBy id
2000-- | Send one item to each fold in a round-robin fashion. This is the consumer
2001-- side dual of producer side 'mergeN' operation.
2003-- partitionN :: Monad m => [Fold m a b] -> Fold m a [b]
2004-- partitionN fs = Fold step begin done
2007-- Demultiplex an input element into a number of typed variants. We want to
2008-- statically restrict the target values within a set of predefined types, an
2009-- enumeration of a GADT. We also want to make sure that the Map contains only
2010-- those types and the full set of those types. Instead of Map it should
2011-- probably be a lookup-table using an array and not in GC memory.
2013-- This is the consumer side dual of the producer side 'mux' operation (XXX to
2016-- | Split the input stream based on a key field and fold each split using a
2017-- specific fold collecting the results in a map from the keys to the results.
2018-- Useful for cases like protocol handlers to handle different type of packets
2019-- using different handlers.
2023-- |-------Fold m a b
2024-- -----stream m a-----Map-----|
2025-- |-------Fold m a b
2031{-# INLINE demuxWith #-}
2032demuxWith :: (Monad m, Ord k)
2033 => (a -> k) -> Map k (Fold m a b) -> Fold m a (Map k b)
2034demuxWith f kv = Fold step initial extract
2040 -- XXX should we raise an exception in Nothing case?
2041 -- Ideally we should enforce that it is a total map over k so that look
2043 -- XXX we could use a monadic update function for a single lookup and
2044 -- update in the map.
2046 in case Map.lookup k mp of
2047 Nothing -> return mp
2048 Just (Fold step' acc extract') -> do
2049 !r <- acc >>= \x -> step' x a
2050 return $ Map.insert k (Fold step' (return r) extract') mp
2051 extract = Prelude.mapM (\(Fold _ acc e) -> acc >>= e)
2053-- | Fold a stream of key value pairs using a map of specific folds for each
2054-- key into a map from keys to the results of fold outputs of the corresponding
2058-- > let table = Data.Map.fromList [(\"SUM", FL.sum), (\"PRODUCT", FL.product)]
2059-- input = S.fromList [(\"SUM",1),(\"PRODUCT",2),(\"SUM",3),(\"PRODUCT",4)]
2060-- in FL.foldl' (FL.demux table) input
2067demux :: (Monad m, Ord k)
2068 => Map k (Fold m a b) -> Fold m (k, a) (Map k b)
2069demux fs = demuxWith fst (Map.map (lmap snd) fs)
2071-- | Split the input stream based on a key field and fold each split using a
2072-- specific fold without collecting the results. Useful for cases like protocol
2073-- handlers to handle different type of packets.
2077-- |-------Fold m a ()
2078-- -----stream m a-----Map-----|
2079-- |-------Fold m a ()
2087-- demuxWith_ can be slightly faster than demuxWith because we do not need to
2088-- update the Map in this case. This may be significant only if the map is
2090{-# INLINE demuxWith_ #-}
2091demuxWith_ :: (Monad m, Ord k)
2092 => (a -> k) -> Map k (Fold m a b) -> Fold m a ()
2093demuxWith_ f kv = Fold step initial extract
2098 Prelude.mapM (\(Fold s i e) ->
2099 i >>= \r -> return (Fold s (return r) e)) kv
2101 -- XXX should we raise an exception in Nothing case?
2102 -- Ideally we should enforce that it is a total map over k so that look
2104 case Map.lookup (f a) mp of
2105 Nothing -> return mp
2106 Just (Fold step' acc _) -> do
2107 _ <- acc >>= \x -> step' x a
2109 extract mp = Prelude.mapM (\(Fold _ acc e) -> acc >>= e) mp >> return ()
2111-- | Given a stream of key value pairs and a map from keys to folds, fold the
2112-- values for each key using the corresponding folds, discarding the outputs.
2115-- > let prn = FL.drainBy print
2116-- > let table = Data.Map.fromList [(\"ONE", prn), (\"TWO", prn)]
2117-- input = S.fromList [(\"ONE",1),(\"TWO",2)]
2118-- in FL.foldl' (FL.demux_ table) input
2124{-# INLINE demux_ #-}
2125demux_ :: (Monad m, Ord k) => Map k (Fold m a ()) -> Fold m (k, a) ()
2126demux_ fs = demuxWith_ fst (Map.map (lmap snd) fs)
2128-- XXX instead of a Map we could yield the results as a pure stream as they
2129-- complete. We could then concatMap the fold results to implement the
2130-- windowing combinators.
2132-- | Split the input stream based on a key field and fold each split using the
2133-- given fold. Useful for map/reduce, bucketizing the input in different bins
2134-- or for generating histograms.
2137-- > let input = S.fromList [(\"ONE",1),(\"ONE",1.1),(\"TWO",2), (\"TWO",2.2)]
2138-- in FL.foldl' (FL.classify FL.toListRev) input
2139-- fromList [(\"ONE",[1.1,1.0]),(\"TWO",[2.2,2.0])]
2143{-# INLINE classifyWith #-}
2144classifyWith :: (Monad m, Ord k) => (a -> k) -> Fold m a b -> Fold m a (Map k b)
2145classifyWith f (Fold step initial extract) = Fold step' initial' extract'
2149 initial' = return Map.empty
2152 in case Map.lookup k kv of
2156 return $ Map.insert k r kv
2159 return $ Map.insert k r kv
2160 extract' = Prelude.mapM extract
2162-- | Given an input stream of key value pairs and a fold for values, fold all
2163-- the values belonging to each key. Useful for map/reduce, bucketizing the
2164-- input in different bins or for generating histograms.
2167-- > let input = S.fromList [(\"ONE",1),(\"ONE",1.1),(\"TWO",2), (\"TWO",2.2)]
2168-- in FL.foldl' (FL.classify FL.toListRev) input
2169-- fromList [(\"ONE",[1.1,1.0]),(\"TWO",[2.2,2.0])]
2174-- > classify fld = classifyWith fst (lmap snd fld)
2177{-# INLINE classify #-}
2178classify :: (Monad m, Ord k) => Fold m a b -> Fold m (k, a) (Map k b)
2179classify fld = classifyWith fst (lmap snd fld)
2181------------------------------------------------------------------------------
2183------------------------------------------------------------------------------
2185-- | Like 'unzipWith' but with a monadic splitter function.
2188{-# INLINE unzipWithM #-}
2189unzipWithM :: Monad m
2190 => (a -> m (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
2191unzipWithM f (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
2192 Fold step begin done
2196 step (Tuple' xL xR) a = do
2198 Tuple' <$> stepL xL b <*> stepR xR c
2199 begin = Tuple' <$> beginL <*> beginR
2200 done (Tuple' xL xR) = (,) <$> doneL xL <*> doneR xR
2202-- | Split elements in the input stream into two parts using a pure splitter
2203-- function, direct each part to a different fold and zip the results.
2206{-# INLINE unzipWith #-}
2208 => (a -> (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
2209unzipWith f = unzipWithM (return . f)
2211-- | Send the elements of tuples in a stream of tuples through two different
2216-- |-------Fold a x--------|
2217-- -----Stream m x----(a,b)--| |----m (x,y)
2218-- |-------Fold b y--------|
2222-- This is the consumer side dual of the producer side 'zip' operation.
2226unzip :: Monad m => Fold m a x -> Fold m b y -> Fold m (a,b) (x,y)
2229------------------------------------------------------------------------------
2231------------------------------------------------------------------------------
2233-- | Modify the fold such that when the fold is done, instead of returning the
2234-- accumulator, it returns a fold. The returned fold starts from where we left
2235-- i.e. it uses the last accumulator value as the initial value of the
2236-- accumulator. Thus we can resume the fold later and feed it more input.
2239-- > more <- FL.foldl (FL.duplicate FL.sum) (S.enumerateFromTo 1 10)
2240-- > evenMore <- FL.foldl (FL.duplicate more) (S.enumerateFromTo 11 20)
2241-- > FL.foldl evenMore (S.enumerateFromTo 21 30)
2245{-# INLINABLE duplicate #-}
2246duplicate :: Applicative m => Fold m a b -> Fold m a (Fold m a b)
2247duplicate (Fold step begin done) =
2248 Fold step begin (\x -> pure (Fold step (pure x) done))
2251-- All the stream flattening transformations can also be applied to a fold
2254-- | This can be used to apply all the stream generation operations on folds.
2255lconcatMap ::(IsStream t, Monad m) => (a -> t m b)
2258lconcatMap s f1 f2 = undefined
2261-- All the grouping transformation that we apply to a stream can also be
2262-- applied to a fold input stream.
2265-- | Group the input stream into groups of elements between @low@ and @high@.
2266-- Collection starts in chunks of @low@ and then keeps doubling until we reach
2267-- @high@. Each chunk is folded using the provided fold function.
2269-- This could be useful, for example, when we are folding a stream of unknown
2270-- size to a stream of arrays and we want to minimize the number of
2275-- XXX we should be able to implement it with parsers/terminating folds.
2277{-# INLINE lchunksInRange #-}
2278lchunksInRange :: Monad m
2279 => Int -> Int -> Fold m a b -> Fold m b c -> Fold m a c
2280lchunksInRange low high (Fold step1 initial1 extract1)
2281 (Fold step2 initial2 extract2) = undefined
2284-- | Group the input stream into groups of @n@ elements each and then fold each
2285-- group using the provided fold function.
2289-- -----Fold m a b----|-Fold n a c-|-Fold n a c-|-...-|----Fold m a c
2293{-# INLINE lchunksOf #-}
2294lchunksOf :: Monad m => Int -> Fold m a b -> Fold m b c -> Fold m a c
2295lchunksOf n (Fold step1 initial1 extract1) (Fold step2 initial2 extract2) =
2296 Fold step' initial' extract'
2300 initial' = (Tuple3' 0) <$> initial1 <*> initial2
2301 step' (Tuple3' i r1 r2) a = do
2305 return $ Tuple3' (i + 1) res r2
2308 acc2 <- step2 r2 res
2312 return $ Tuple3' 1 acc1 acc2
2313 extract' (Tuple3' _ _ r) = extract2 r
2315-- | Group the input stream into windows of n second each and then fold each
2316-- group using the provided fold function.
2318-- For example, we can copy and distribute a stream to multiple folds where
2319-- each fold can group the input differently e.g. by one second, one minute and
2320-- one hour windows respectively and fold each resulting stream of folds.
2324-- -----Fold m a b----|-Fold n a c-|-Fold n a c-|-...-|----Fold m a c
2327{-# INLINE lsessionsOf #-}
2328lsessionsOf :: MonadAsync m => Double -> Fold m a b -> Fold m b c -> Fold m a c
2329lsessionsOf n (Fold step1 initial1 extract1) (Fold step2 initial2 extract2) =
2330 Fold step' initial' extract'
2334 -- XXX MVar may be expensive we need a cheaper synch mechanism here
2338 mv1 <- liftIO $ newMVar i1
2339 mv2 <- liftIO $ newMVar (Right i2)
2340 t <- control $ \run ->
2341 mask $ \restore -> do
2342 tid <- forkIO $ catch (restore $ void $ run (timerThread mv1 mv2))
2343 (handleChildException mv2)
2345 return $ Tuple3' t mv1 mv2
2346 step' acc@(Tuple3' _ mv1 _) a = do
2347 r1 <- liftIO $ takeMVar mv1
2349 liftIO $ putMVar mv1 res
2351 extract' (Tuple3' tid _ mv2) = do
2352 r2 <- liftIO $ takeMVar mv2
2353 liftIO $ killThread tid
2356 Right x -> extract2 x
2358 timerThread mv1 mv2 = do
2359 liftIO $ threadDelay (round $ n * 1000000)
2361 r1 <- liftIO $ takeMVar mv1
2363 liftIO $ putMVar mv1 i1
2366 r2 <- liftIO $ takeMVar mv2
2369 Right x -> fmap Right $ step2 x res1
2370 liftIO $ putMVar mv2 res
2373 handleChildException ::
2374 MVar (Either SomeException a) -> SomeException -> IO ()
2375 handleChildException mv2 e = do
2382------------------------------------------------------------------------------
2383-- Windowed classification
2384------------------------------------------------------------------------------
2386-- We divide the stream into windows or chunks in space or time and each window
2387-- can be associated with a key, all events associated with a particular key in
2388-- the window can be folded to a single result. The stream can be split into
2389-- windows by size or by using a split predicate on the elements in the stream.
2390-- For example, when we receive a closing flag, we can close the window.
2392-- A "chunk" is a space window and a "session" is a time window. Are there any
2393-- other better short words to describe them. An alternative is to use
2394-- "swindow" and "twindow". Another word for "session" could be "spell".
2396-- TODO: To mark the position in space or time we can have Indexed or
2397-- TimeStamped types. That can make it easy to deal with the position indices
2400------------------------------------------------------------------------------
2401-- Keyed Sliding Windows
2402------------------------------------------------------------------------------
2405{-# INLINABLE classifySlidingChunks #-}
2406classifySlidingChunks
2407 :: (IsStream t, MonadAsync m, Ord k)
2408 => Int -- ^ window size
2409 -> Int -- ^ window slide
2410 -> Fold m a b -- ^ Fold to be applied to window events
2411 -> t m (k, a, Bool) -- ^ window key, data, close event
2413classifySlidingChunks wsize wslide (Fold step initial extract) str
2416-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
2417-- slide the send window when an ack is received and we slide the receive
2418-- window when a sequence is complete. Sliding is stateful in case of TCP,
2419-- sliding releases the send buffer or makes data available to the user from
2420-- the receive buffer.
2421{-# INLINABLE classifySlidingSessions #-}
2422classifySlidingSessions
2423 :: (IsStream t, MonadAsync m, Ord k)
2424 => Double -- ^ timer tick in seconds
2425 -> Double -- ^ time window size
2426 -> Double -- ^ window slide
2427 -> Fold m a b -- ^ Fold to be applied to window events
2428 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
2430classifySlidingSessions tick interval slide (Fold step initial extract) str
2434------------------------------------------------------------------------------
2435-- Sliding Window Buffers
2436------------------------------------------------------------------------------
2438-- These buffered versions could be faster than concurrent incremental folds of
2439-- all overlapping windows as in many cases we may not need all the values to
2440-- compute the fold, we can just compute the result using the old value and new
2441-- value. However, we may need the buffer once in a while, for example for
2442-- string search we usually compute the hash incrementally but when the hash
2443-- matches the hash of the pattern we need to compare the whole string.
2445-- XXX we should be able to implement sequence based splitting combinators
2446-- using this combinator.
2449-- | Buffer n elements of the input in a ring buffer. When t new elements are
2450-- collected, slide the window to remove the same number of oldest elements,
2451-- insert the new elements, and apply an incremental fold on the sliding
2452-- window, supplying the outgoing elements, the new ring buffer as arguments.
2454 :: (IsStream t, Monad m, Ord a, Storable a)
2455 => Int -- window size
2456 -> Int -- window slide
2457 -> Fold m (Ring a, Array a) b
2460slidingChunkBuffer = undefined
2462-- Buffer n seconds worth of stream elements of the input in a radix tree.
2463-- Every t seconds, remove the items that are older than n seconds, and apply
2464-- an incremental fold on the sliding window, supplying the outgoing elements,
2465-- and the new radix tree buffer as arguments.
2467 :: (IsStream t, Monad m, Ord a, Storable a)
2468 => Int -- window size
2470 -> Fold m (RTree a, Array a) b
2473slidingSessionBuffer = undefined
2476------------------------------------------------------------------------------
2477-- Keyed Session Windows
2478------------------------------------------------------------------------------
2481-- | Keyed variable size space windows. Close the window if we do not receive a
2482-- window event in the next "spaceout" elements.
2483{-# INLINABLE classifyChunksBy #-}
2485 :: (IsStream t, MonadAsync m, Ord k)
2486 => Int -- ^ window spaceout (spread)
2487 -> Bool -- ^ reset the spaceout when a chunk window element is received
2488 -> Fold m a b -- ^ Fold to be applied to chunk window elements
2489 -> t m (k, a, Bool) -- ^ chunk key, data, last element
2491classifyChunksBy spanout reset (Fold step initial extract) str = undefined
2493-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
2494-- received within the chunk size window. The chunk gets closed only if no
2495-- element is received within the chunk window.
2497{-# INLINABLE classifyKeepAliveChunks #-}
2498classifyKeepAliveChunks
2499 :: (IsStream t, MonadAsync m, Ord k)
2500 => Int -- ^ window spaceout (spread)
2501 -> Fold m a b -- ^ Fold to be applied to chunk window elements
2502 -> t m (k, a, Bool) -- ^ chunk key, data, last element
2504classifyKeepAliveChunks spanout = classifyChunksBy spanout True
2507-- | @classifySessionsBy tick timeout reset f stream@ groups together all input
2508-- stream elements that belong to the same session. @timeout@ is the maximum
2509-- lifetime of a session in seconds. All elements belonging to a session are
2510-- purged after this duration. If "reset" is 'Ture' then the timeout is reset
2511-- after every event received in the session. Session duration is measured
2512-- using the timestamp of the first element seen for that session. To detect
2513-- session timeouts, a monotonic event time clock is maintained using the
2514-- timestamps seen in the inputs and a timer with a tick duration specified by
2517-- @session key@ is a key that uniquely identifies the session for the given
2518-- element, @timestamp@ characterizes the time when the input element was
2519-- generated, this is an absolute time measured from some @Epoch@. @session
2520-- close@ is a boolean indicating whether this element marks the closing of the
2521-- session. When an input element with @session close@ set to @True@ is seen
2522-- the session is purged immediately.
2524-- All the input elements belonging to a session are collected using the fold
2525-- @f@. The session key and the fold result are emitted in the output stream
2526-- when the session is purged either via the session close event or via the
2527-- session liftime timeout.
2530{-# INLINABLE classifySessionsBy #-}
2532 :: (IsStream t, MonadAsync m, Ord k)
2533 => Double -- ^ timer tick in seconds
2534 -> Double -- ^ session timeout
2535 -> Bool -- ^ reset the timeout when an event is received
2536 -> Fold m a b -- ^ Fold to be applied to session events
2537 -> t m (k, a, Bool, AbsTime) -- ^ session key, timestamp, close event, data
2539classifySessionsBy tick timeout reset (Fold step initial extract) str =
2540 S.concatMap (\(Tuple4' _ _ _ s) -> s) $ S.scanlM' sstep szero stream
2544 timeoutMs = toRelTime (round (timeout * 1000) :: MilliSecond64)
2545 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
2546 szero = Tuple4' (toAbsTime (0 :: MilliSecond64)) H.empty Map.empty S.nil
2548 -- Got a new stream input element
2549 sstep (Tuple4' evTime hp mp _) (Just (key, a, closing, ts)) =
2550 -- XXX we should use a heap in pinned memory to scale it to a large
2553 -- deleting a key from the heap is expensive, so we never delete a
2554 -- key, we just purge it from the Map and it gets purged from the
2555 -- heap on timeout. We just need an extra lookup in the Map when
2556 -- the key is purged from the heap, that should not be expensive.
2558 -- To detect session inactivity we keep a timestamp of the latest event
2559 -- in the Map along with the fold result. When we purge the session
2560 -- from the heap we match the timestamp in the heap with the timestamp
2561 -- in the Map, if the latest timestamp is newer and has not expired we
2562 -- reinsert the key in the heap.
2564 -- XXX if the key is an Int, we can also use an IntMap for slightly
2565 -- better performance.
2567 let accumulate v = do
2568 Tuple' _ old <- maybe (initial >>= return . Tuple' ts) return v
2570 return $ Tuple' ts new
2573 let (r, mp') = Map.updateLookupWithKey (\_ _ -> Nothing) key mp
2574 Tuple' _ acc <- accumulate r
2576 return $ Tuple4' evTime hp mp' (S.yield (key, res))
2578 let r = Map.lookup key mp
2580 let mp' = Map.insert key acc mp
2584 let expiry = addToAbsTime ts timeoutMs
2585 in H.insert (Entry expiry key) hp
2587 -- Event time is maintained as monotonically increasing
2588 -- time. If we have lagged behind any of the timestamps
2589 -- seen then we increase it to match the latest time seen
2590 -- in the timestamps. We also increase it on timer ticks.
2591 return $ Tuple4' (max evTime ts) hp' mp' S.nil
2593 -- Got a timer tick event
2594 -- XXX can we yield the entries without accumulating them?
2595 sstep (Tuple4' evTime heap sessions _) Nothing = do
2596 (hp', mp', out) <- go heap sessions S.nil
2597 return $ Tuple4' curTime hp' mp' out
2601 curTime = addToAbsTime evTime tickMs
2603 let hres = H.uncons hp
2605 Just (Entry ts key, hp') -> do
2606 let duration = diffAbsTime curTime ts
2607 if duration >= timeoutMs
2609 let (r, mp') = Map.updateLookupWithKey
2610 (\_ _ -> Nothing) key mp
2612 Nothing -> go hp' mp' out
2613 Just (Tuple' latestTS acc) -> do
2614 let dur = diffAbsTime curTime latestTS
2615 if dur >= timeoutMs || not reset
2618 go hp' mp' ((key, sess) `S.cons` out)
2620 -- reset the session timeout
2621 let expiry = addToAbsTime latestTS timeoutMs
2622 hp'' = H.insert (Entry expiry key) hp'
2623 mp'' = Map.insert key (Tuple' latestTS acc) mp'
2625 else return (hp, mp, out)
2626 Nothing -> return (hp, mp, out)
2628 -- merge timer events in the stream
2629 stream = S.map Just str `parallel` S.repeatM timer
2631 liftIO $ threadDelay (round $ tick * 1000000)
2634-- | Like 'classifySessionsOf' but the session is kept alive if an event is
2635-- received within the session window. The session times out and gets closed
2636-- only if no event is received within the specified session window size.
2639{-# INLINABLE classifyKeepAliveSessions #-}
2640classifyKeepAliveSessions
2641 :: (IsStream t, MonadAsync m, Ord k)
2642 => Double -- ^ session inactive timeout
2643 -> Fold m a b -- ^ Fold to be applied to session payload data
2644 -> t m (k, a, Bool, AbsTime) -- ^ session key, data, close flag, timestamp
2646classifyKeepAliveSessions timeout = classifySessionsBy 1 timeout True
2648------------------------------------------------------------------------------
2649-- Keyed tumbling windows
2650------------------------------------------------------------------------------
2652-- Tumbling windows is a special case of sliding windows where the window slide
2653-- is the same as the window size. Or it can be a special case of session
2654-- windows where the reset flag is set to False.
2656-- XXX instead of using the early termination flag in the stream, we can use an
2657-- early terminating fold instead.
2660-- | Split the stream into fixed size chunks of specified size. Within each
2661-- such chunk fold the elements in buckets identified by the keys. A particular
2662-- bucket fold can be terminated early if a closing flag is encountered in an
2663-- element for that key.
2666{-# INLINABLE classifyChunksOf #-}
2668 :: (IsStream t, MonadAsync m, Ord k)
2669 => Int -- ^ window size
2670 -> Fold m a b -- ^ Fold to be applied to window events
2671 -> t m (k, a, Bool) -- ^ window key, data, close event
2673classifyChunksOf wsize = classifyChunksBy wsize False
2676-- | Split the stream into fixed size time windows of specified interval in
2677-- seconds. Within each such window, fold the elements in buckets identified by
2678-- the keys. A particular bucket fold can be terminated early if a closing flag
2679-- is encountered in an element for that key. Once a fold is terminated the key
2680-- and value for that bucket are emitted in the output stream.
2682-- Session @timestamp@ in the input stream is an absolute time from some epoch,
2683-- characterizing the time when the input element was generated. To detect
2684-- session window end, a monotonic event time clock is maintained synced with
2685-- the timestamps with a clock resolution of 1 second.
2688{-# INLINABLE classifySessionsOf #-}
2690 :: (IsStream t, MonadAsync m, Ord k)
2691 => Double -- ^ time window size
2692 -> Fold m a b -- ^ Fold to be applied to window events
2693 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
2695classifySessionsOf 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