2-- Module : Streamly.Internal.Data.Pipe
3-- Copyright : (c) 2019 Composewell Technologies
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
9-- There are three fundamental types in streamly. They are streams
10-- ("Streamly.Prelude"), pipes ("Streamly.Internal.Data.Pipe") and folds ("Streamly.Data.Fold").
11-- Streams are sources or producers of values, multiple sources can be merged
12-- into a single source but a source cannot be split into multiple stream
13-- sources. Folds are sinks or consumers, a stream can be split and
14-- distributed to multiple folds but the results cannot be merged back into a
15-- stream source again. Pipes are transformations, a stream source can be split
16-- and distributed to multiple pipes each pipe can apply its own transform on
17-- the stream and the results can be merged back into a single pipe. Pipes can
18-- be attached to a source to produce a source or they can be attached to a
19-- fold to produce a fold, or multiple pipes can be merged or zipped into a
22-- > import qualified Streamly.Internal.Data.Pipe as Pipe
24module Streamly.Internal.Data.Pipe
46 -- ** Scanning Filters
51 -- | Insertion adds more elements to the stream.
73 -- | Streams can be split into segments in space or in time. We use the
74 -- term @chunk@ to refer to a spatial length of the stream (spatial window)
75 -- and the term @session@ to refer to a length in time (time window).
77 -- In imperative terms, grouped folding can be considered as a nested loop
78 -- where we loop over the stream to group elements and then loop over
79 -- individual groups to fold them to a single value that is yielded in the
99 -- Keeping the delimiters
104 -- Splitting by multiple sequences
106 -- , splitSuffixOnAny
107 -- , splitPrefixOnAny
123 -- The 'Applicative' instance of a distributing 'Fold' distributes one copy
124 -- of the stream to each fold and combines the results using a function.
128 -- |-------Fold m a b--------|
129 -- ---stream m a---| |---m (b,c,...)
130 -- |-------Fold m a c--------|
135 -- To compute the average of numbers in a stream without going through the
138 -- >>> let avg = (/) <$> FL.sum <*> fmap fromIntegral FL.length
139 -- >>> FL.foldl' avg (S.enumerateFromTo 1.0 100.0)
142 -- The 'Semigroup' and 'Monoid' instances of a distributing fold distribute
143 -- the input to both the folds and combines the outputs using Monoid or
144 -- Semigroup instances of the output types:
146 -- >>> import Data.Monoid (Sum)
147 -- >>> FL.foldl' (FL.head <> FL.last) (fmap Sum $ S.enumerateFromTo 1.0 100.0)
148 -- Just (Sum {getSum = 101.0})
150 -- The 'Num', 'Floating', and 'Fractional' instances work in the same way.
157 -- Direct items in the input stream to different folds using a function to
158 -- select the fold. This is useful to demultiplex the input stream.
175 -- These can be expressed using lmap/lmapM and unzip
182 , duplicate -- experimental
184 -- * Windowed Classification
185 -- | Split the stream into windows or chunks in space or time. Each window
186 -- can be associated with a key, all events associated with a particular
187 -- key in the window can be folded to a single result. The stream is split
188 -- into windows of specified size, the window can be terminated early if
189 -- the closing flag is specified in the input stream.
191 -- The term "chunk" is used for a space window and the term "session" is
192 -- used for a time window.
194 -- ** Tumbling Windows
195 -- | A new window starts after the previous window is finished.
196 -- , classifyChunksOf
199 -- ** Keep Alive Windows
200 -- | The window size is extended if an event arrives within the specified
201 -- window size. This can represent sessions with idle or inactive timeout.
202 -- , classifyKeepAliveChunks
203 , classifyKeepAliveSessions
206 -- ** Sliding Windows
207 -- | A new window starts after the specified slide from the previous
208 -- window. Therefore windows can overlap.
209 , classifySlidingChunks
210 , classifySlidingSessions
212 -- ** Sliding Window Buffers
213 -- , slidingChunkBuffer
214 -- , slidingSessionBuffer
219-- import Control.Concurrent (threadDelay, forkIO, killThread)
220-- import Control.Concurrent.MVar (MVar, newMVar, takeMVar, putMVar)
221-- import Control.Exception (SomeException(..), catch, mask)
222-- import Control.Monad (void)
223-- import Control.Monad.Catch (throwM)
224-- import Control.Monad.IO.Class (MonadIO(..))
225-- import Control.Monad.Trans (lift)
226-- import Control.Monad.Trans.Control (control)
227-- import Data.Functor.Identity (Identity)
228-- import Data.Heap (Entry(..))
229-- import Data.Map.Strict (Map)
230-- import Data.Maybe (fromJust, isJust, isNothing)
232-- import Foreign.Storable (Storable(..))
234 hiding (id, filter, drop, dropWhile, take, takeWhile, zipWith, foldr,
235 foldl, map, mapM_, sequence, all, any, sum, product, elem,
236 notElem, maximum, minimum, head, last, tail, length, null,
237 reverse, iterate, init, and, or, lookup, foldr1, (!!),
238 scanl, scanl1, replicate, concatMap, mconcat, foldMap, unzip,
239 span, splitAt, break, mapM)
241-- import qualified Data.Heap as H
242-- import qualified Data.Map.Strict as Map
243-- import qualified Prelude
245-- import Streamly.Prelude (MonadAsync, parallel)
246-- import Streamly.Data.Fold.Types (Fold(..))
247import Streamly.Internal.Data.Pipe.Type
248 (Pipe(..), PipeState(..), Step(..), zipWith, tee, map, compose)
249-- import Streamly.Internal.Data.Array.Foreign.Type (Array)
250-- import Streamly.Internal.Data.Ring.Foreign (Ring)
251-- import Streamly.Internal.Data.Stream.Serial (SerialT)
252-- import Streamly.Internal.Data.Stream.StreamK (IsStream())
253-- import Streamly.Internal.Data.Time.Units
254-- (AbsTime, MilliSecond64(..), addToAbsTime, diffAbsTime, toRelTime,
257-- import Streamly.Internal.Data.Strict
259-- import qualified Streamly.Internal.Data.Array.Foreign.Type as A
260-- import qualified Streamly.Prelude as S
261-- import qualified Streamly.Internal.Data.Stream.StreamD as D
262-- import qualified Streamly.Internal.Data.Stream.StreamK as K
263-- import qualified Streamly.Internal.Data.Stream.Common as P
265------------------------------------------------------------------------------
267------------------------------------------------------------------------------
269-- | Lift a monadic function to a 'Pipe'.
273mapM :: Monad m => (a -> m b) -> Pipe m a b
274mapM f = Pipe consume undefined ()
278 return $ Yield r (Consume ())
280------------------------------------------------------------------------------
282------------------------------------------------------------------------------
284-- | Include only those elements that pass a predicate.
286-- >>> FL.foldl (lfilter (> 5) FL.sum) [1..10]
290{-# INLINABLE lfilter #-}
291lfilter :: Monad m => (a -> Bool) -> Fold m a r -> Fold m a r
292lfilter f (Fold step begin done) = Fold step' begin done
294 step' x a = if f a then step x a else return x
296-- | Like 'lfilter' but with a monadic predicate.
299{-# INLINABLE lfilterM #-}
300lfilterM :: Monad m => (a -> m Bool) -> Fold m a r -> Fold m a r
301lfilterM f (Fold step begin done) = Fold step' begin done
305 if use then step x a else return x
307-- | Take first 'n' elements from the stream and discard the rest.
310{-# INLINABLE ltake #-}
311ltake :: Monad m => Int -> Fold m a b -> Fold m a b
312ltake n (Fold step initial done) = Fold step' initial' done'
314 initial' = fmap (Tuple' 0) initial
315 step' (Tuple' i r) a = do
319 return $ Tuple' (i + 1) res
320 else return $ Tuple' i r
321 done' (Tuple' _ r) = done r
323-- | Takes elements from the input as long as the predicate succeeds.
326{-# INLINABLE ltakeWhile #-}
327ltakeWhile :: Monad m => (a -> Bool) -> Fold m a b -> Fold m a b
328ltakeWhile predicate (Fold step initial done) = Fold step' initial' done'
330 initial' = fmap Left' initial
331 step' (Left' r) a = do
333 then fmap Left' $ step r a
334 else return (Right' r)
336 done' (Left' r) = done r
337 done' (Right' r) = done r
339------------------------------------------------------------------------------
341------------------------------------------------------------------------------
343------------------------------------------------------------------------------
344-- Grouping without looking at elements
345------------------------------------------------------------------------------
347------------------------------------------------------------------------------
349------------------------------------------------------------------------------
352-- | @splitAt n f1 f2@ composes folds @f1@ and @f2@ such that first @n@
353-- elements of its input are sent to fold @f1@ and the rest of the stream is
356-- > let splitAt_ n xs = FL.foldl' (FL.splitAt n FL.toList FL.toList) $ S.fromList xs
358-- >>> splitAt_ 6 "Hello World!"
359-- > ("Hello ","World!")
361-- >>> splitAt_ (-1) [1,2,3]
364-- >>> splitAt_ 0 [1,2,3]
367-- >>> splitAt_ 1 [1,2,3]
370-- >>> splitAt_ 3 [1,2,3]
373-- >>> splitAt_ 4 [1,2,3]
376-- This can be considered as a two-fold version of 'ltake' where we take both
377-- the segments instead of discarding the leftover.
380{-# INLINE splitAt #-}
387splitAt n (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
388 Fold step init extract
390 init = Tuple3' <$> return n <*> initialL <*> initialR
392 step (Tuple3' i xL xR) input =
394 then stepL xL input >>= (\a -> return (Tuple3' (i - 1) a xR))
395 else stepR xR input >>= (\b -> return (Tuple3' i xL b))
397 extract (Tuple3' _ a b) = (,) <$> extractL a <*> extractR b
399------------------------------------------------------------------------------
401------------------------------------------------------------------------------
403------------------------------------------------------------------------------
404-- Generalized grouping
405------------------------------------------------------------------------------
407-- This combinator is the most general grouping combinator and can be used to
408-- implement all other grouping combinators.
410-- XXX check if this can implement the splitOn combinator i.e. we can slide in
411-- new elements, slide out old elements and incrementally compute the hash.
412-- Also, can we implement the windowed classification combinators using this?
414-- In fact this is a parse. Instead of using a special return value in the fold
415-- we are using a mapping function.
417-- Note that 'scanl'' (usually followed by a map to extract the desired value
418-- from the accumulator) can be used to realize many implementations e.g. a
419-- sliding window implementation. A scan followed by a mapMaybe is also a good
420-- pattern to express many problems where we want to emit a filtered output and
421-- not emit an output on every input.
423-- Passing on of the initial accumulator value to the next fold is equivalent
424-- to returning the leftover concept.
427-- | @groupScan splitter fold stream@ folds the input stream using @fold@.
428-- @splitter@ is applied on the accumulator of the fold every time an item is
429-- consumed by the fold. The fold continues until @splitter@ returns a 'Just'
430-- value. A 'Just' result from the @splitter@ specifies a result to be emitted
431-- in the output stream and the initial value of the accumulator for the next
432-- group's fold. This allows us to control whether to start fresh for the next
433-- fold or to continue from the previous fold's output.
435{-# INLINE groupScan #-}
437 :: (IsStream t, Monad m)
438 => (x -> m (Maybe (b, x))) -> Fold m a x -> t m a -> t m b
439groupScan split fold m = undefined
442-- | Group the input stream into groups of @n@ elements each and then fold each
443-- group using the provided fold function.
445-- >> S.toList $ FL.chunksOf 2 FL.sum (S.enumerateFromTo 1 10)
448-- This can be considered as an n-fold version of 'ltake' where we apply
449-- 'ltake' repeatedly on the leftover stream until the stream exhausts.
452{-# INLINE chunksOf #-}
454 :: (IsStream t, Monad m)
455 => Int -> Fold m a b -> t m a -> t m b
456chunksOf n f m = D.fromStreamD $ D.groupsOf n f (D.toStreamD m)
458-- | Transform a fold from a pure input to a 'Maybe' input, consuming only
460{-# INLINE lcatMaybes #-}
461lcatMaybes :: Monad m => Fold m a b -> Fold m (Maybe a) b
462lcatMaybes = lfilter isJust . lmap fromJust
464-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
465-- XXX add this example after fixing the serial stream rate control
466-- >>> S.toList $ S.take 5 $ sessionsOf 1 FL.sum $ constRate 2 $ S.enumerateFrom 1
469-- | Group the input stream into windows of @n@ second each and then fold each
470-- group using the provided fold function.
473{-# INLINE sessionsOf #-}
475 :: (IsStream t, MonadAsync m)
476 => Double -> Fold m a b -> t m a -> t m b
478 splitSuffixBy' isNothing (lcatMaybes f)
479 (S.intersperseByTime n (return Nothing) (S.map Just xs))
481------------------------------------------------------------------------------
483------------------------------------------------------------------------------
485------------------------------------------------------------------------------
487------------------------------------------------------------------------------
489-- | Break the input stream into two groups, the first group takes the input as
490-- long as the predicate applied to the first element of the stream and next
491-- input element holds 'True', the second group takes the rest of the input.
499_spanBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
500 Fold step init extract
503 init = Tuple3' <$> initialL <*> initialR <*> return (Tuple' Nothing True)
505 step (Tuple3' a b (Tuple' (Just frst) isFirstG)) input =
506 if cmp frst input && isFirstG
508 >>= (\a' -> return (Tuple3' a' b (Tuple' (Just frst) isFirstG)))
510 >>= (\a' -> return (Tuple3' a a' (Tuple' Nothing False)))
512 step (Tuple3' a b (Tuple' Nothing isFirstG)) input =
515 >>= (\a' -> return (Tuple3' a' b (Tuple' (Just input) isFirstG)))
517 >>= (\a' -> return (Tuple3' a a' (Tuple' Nothing False)))
519 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
521-- | Span as long as the predicate is 'True'. @span p f1 f2@ composes folds
522-- @f1@ and @f2@ such that the composed fold continues sending the input to
523-- @f1@ as long as the predicate @p@ is 'True'. The rest of the input is sent
526-- > let span_ p xs = FL.foldl' (FL.span p FL.toList FL.toList) $ S.fromList xs
528-- >>> span_ (< 1) [1,2,3]
531-- >>> span_ (< 2) [1,2,3]
534-- >>> span_ (< 4) [1,2,3]
537-- This can be considered as a two-fold version of 'ltakeWhile' where we take
538-- both the segments instead of discarding the leftover.
548span p (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
549 Fold step init extract
552 init = Tuple3' <$> initialL <*> initialR <*> return True
554 step (Tuple3' a b isFirstG) input =
555 if isFirstG && p input
556 then stepL a input >>= (\a' -> return (Tuple3' a' b True))
557 else stepR b input >>= (\a' -> return (Tuple3' a a' False))
559 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
562-- > break p = span (not . p)
564-- Break as soon as the predicate becomes 'True'. @break p f1 f2@ composes
565-- folds @f1@ and @f2@ such that @f1@ stops receiving input as soon as the
566-- predicate @p@ becomes 'True'. The rest of the input is sent to @f2@.
568-- This is the binary version of 'splitBy'.
570-- > let break_ p xs = FL.foldl' (FL.break p FL.toList FL.toList) $ S.fromList xs
572-- >>> break_ (< 1) [3,2,1]
575-- >>> break_ (< 2) [3,2,1]
578-- >>> break_ (< 4) [3,2,1]
589break p = span (not . p)
591-- | Like 'spanBy' but applies the predicate in a rolling fashion i.e.
592-- predicate is applied to the previous and the next input elements.
593{-# INLINE _spanRollingBy #-}
600_spanRollingBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
601 Fold step init extract
604 init = Tuple3' <$> initialL <*> initialR <*> return Nothing
606 step (Tuple3' a b (Just frst)) input =
608 then stepL a input >>= (\a' -> return (Tuple3' a' b (Just input)))
609 else stepR b input >>= (\b' -> return (Tuple3' a b' (Just input)))
611 step (Tuple3' a b Nothing) input =
612 stepL a input >>= (\a' -> return (Tuple3' a' b (Just input)))
614 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
616------------------------------------------------------------------------------
618------------------------------------------------------------------------------
620-- | @groupsBy cmp f $ S.fromList [a,b,c,...]@ assigns the element @a@ to the
621-- first group, if @a \`cmp` b@ is 'True' then @b@ is also assigned to the same
622-- group. If @a \`cmp` c@ is 'True' then @c@ is also assigned to the same
623-- group and so on. When the comparison fails a new group is started. Each
624-- group is folded using the fold @f@.
626-- >>> S.toList $ FL.groupsBy (>) FL.toList $ S.fromList [1,3,7,0,2,5]
627-- > [[1,3,7],[0,2,5]]
630{-# INLINE groupsBy #-}
632 :: (IsStream t, Monad m)
637groupsBy cmp f m = D.fromStreamD $ D.groupsBy cmp f (D.toStreamD m)
639-- | Unlike @groupsBy@ this function performs a rolling comparison of two
640-- successive elements in the input stream. @groupsRollingBy cmp f $ S.fromList
641-- [a,b,c,...]@ assigns the element @a@ to the first group, if @a \`cmp` b@ is
642-- 'True' then @b@ is also assigned to the same group. If @b \`cmp` c@ is
643-- 'True' then @c@ is also assigned to the same group and so on. When the
644-- comparison fails a new group is started. Each group is folded using the fold
647-- >>> S.toList $ FL.groupsRollingBy (\a b -> a + 1 == b) FL.toList $ S.fromList [1,2,3,7,8,9]
648-- > [[1,2,3],[7,8,9]]
651{-# INLINE groupsRollingBy #-}
653 :: (IsStream t, Monad m)
658groupsRollingBy cmp f m = D.fromStreamD $ D.groupsRollingBy cmp f (D.toStreamD m)
661-- > groups = groupsBy (==)
662-- > groups = groupsRollingBy (==)
664-- Groups a contiguous span of equal elements together in one group.
666-- >>> S.toList $ FL.groups FL.toList $ S.fromList [1,1,2,2]
670groups :: (IsStream t, Monad m, Eq a) => Fold m a b -> t m a -> t m b
671groups = groupsBy (==)
673------------------------------------------------------------------------------
674-- Binary splitting on a separator
675------------------------------------------------------------------------------
678-- | Find the first occurrence of the specified sequence in the input stream
679-- and break the input stream into two parts, the first part consisting of the
680-- stream before the sequence and the second part consisting of the sequence
681-- and the rest of the stream.
683-- > let breakOn_ pat xs = FL.foldl' (FL.breakOn pat FL.toList FL.toList) $ S.fromList xs
685-- >>> breakOn_ "dear" "Hello dear world!"
686-- > ("Hello ","dear world!")
688{-# INLINE breakOn #-}
689breakOn :: Monad m => Array a -> Fold m a b -> Fold m a c -> Fold m a (b,c)
690breakOn pat f m = undefined
693------------------------------------------------------------------------------
694-- N-ary split on a predicate
695------------------------------------------------------------------------------
697-- TODO: Use a Splitter configuration similar to the "split" package to make it
698-- possible to express all splitting combinations. In general, we can have
699-- infix/suffix/prefix/condensing of separators, dropping both leading/trailing
700-- separators. We can have a single split operation taking the splitter config
703-- | Split a stream on separator elements determined by a predicate, dropping
704-- the separator. Separators are not considered part of stream segments on
705-- either side of it instead they are treated as infixed between two stream
706-- segments. For example, with @.@ as separator, @"a.b.c"@ would be parsed as
707-- @["a","b","c"]@. When @.@ is in leading or trailing position it is still
708-- considered as infixed, treating the first or the last segment as empty. For
709-- example, @".a."@ would be parsed as @["","a",""]@. This operation is
710-- opposite of 'intercalate'.
712-- Let's use the following definition for illustration:
714-- > splitBy_ p xs = S.toList $ FL.splitBy p (FL.toList) (S.fromList xs)
716-- >>> splitBy_ (== '.') ""
719-- >>> splitBy_ (== '.') "."
722-- >>> splitBy_ (== '.') ".a"
725-- >>> splitBy_ (== '.') "a."
728-- >>> splitBy_ (== '.') "a.b"
731-- >>> splitBy_ (== '.') "a..b"
734-- This can be considered as an n-fold version of 'break' where we apply
735-- 'break' successively on the input stream, dropping the first element
736-- of the second segment after each break.
739{-# INLINE splitBy #-}
741 :: (IsStream t, Monad m)
742 => (a -> Bool) -> Fold m a b -> t m a -> t m b
743splitBy predicate f m =
744 D.fromStreamD $ D.splitBy predicate f (D.toStreamD m)
746-- | Like 'splitBy' but the separator is treated as part of the previous
747-- stream segment (suffix). Therefore, when the separator is in trailing
748-- position, no empty segment is considered to follow it. For example, @"a.b."@
749-- would be parsed as @["a","b"]@ instead of @["a","b",""]@ as in the case of
752-- > splitSuffixBy_ p xs = S.toList $ FL.splitSuffixBy p (FL.toList) (S.fromList xs)
754-- >>> splitSuffixBy_ (== '.') ""
757-- >>> splitSuffixBy_ (== '.') "."
760-- >>> splitSuffixBy_ (== '.') "a"
763-- >>> splitSuffixBy_ (== '.') ".a"
766-- >>> splitSuffixBy_ (== '.') "a."
769-- >>> splitSuffixBy_ (== '.') "a.b"
772-- >>> splitSuffixBy_ (== '.') "a.b."
775-- >>> splitSuffixBy_ (== '.') "a..b.."
778-- > lines = splitSuffixBy (== '\n')
780-- This can be considered as an n-fold version of 'breakPost' where we apply
781-- 'breakPost' successively on the input stream, dropping the first element
782-- of the second segment after each break.
785{-# INLINE splitSuffixBy #-}
787 :: (IsStream t, Monad m)
788 => (a -> Bool) -> Fold m a b -> t m a -> t m b
789splitSuffixBy predicate f m =
790 D.fromStreamD $ D.splitSuffixBy predicate f (D.toStreamD m)
792-- | Like 'splitBy' but ignores repeated separators or separators in leading
793-- or trailing position. Therefore, @"..a..b.."@ would be parsed as
794-- @["a","b"]@. In other words, it treats the input like words separated by
795-- whitespace elements determined by the predicate.
797-- > wordsBy' p xs = S.toList $ FL.wordsBy p (FL.toList) (S.fromList xs)
799-- >>> wordsBy' (== ',') ""
802-- >>> wordsBy' (== ',') ","
805-- >>> wordsBy' (== ',') ",a,,b,"
808-- > words = wordsBy isSpace
811{-# INLINE wordsBy #-}
813 :: (IsStream t, Monad m)
814 => (a -> Bool) -> Fold m a b -> t m a -> t m b
815wordsBy predicate f m =
816 D.fromStreamD $ D.wordsBy predicate f (D.toStreamD m)
818-- XXX we should express this using the Splitter config.
820-- We can get splitSuffixBy' by appending the suffix to the output segments
821-- produced by splitSuffixBy. However, it may add an additional suffix if the last
822-- fragment did not have a suffix in the first place.
824-- | Like 'splitSuffixBy' but keeps the suffix in the splits.
826-- > splitSuffixBy'_ p xs = S.toList $ FL.splitSuffixBy' p (FL.toList) (S.fromList xs)
828-- >>> splitSuffixBy'_ (== '.') ""
831-- >>> splitSuffixBy'_ (== '.') "."
834-- >>> splitSuffixBy'_ (== '.') "a"
837-- >>> splitSuffixBy'_ (== '.') ".a"
840-- >>> splitSuffixBy'_ (== '.') "a."
843-- >>> splitSuffixBy'_ (== '.') "a.b"
846-- >>> splitSuffixBy'_ (== '.') "a.b."
849-- >>> splitSuffixBy'_ (== '.') "a..b.."
850-- > ["a.",".","b.","."]
852-- This can be considered as an n-fold version of 'breakPost' where we apply
853-- 'breakPost' successively on the input stream.
856{-# INLINE splitSuffixBy' #-}
858 :: (IsStream t, Monad m)
859 => (a -> Bool) -> Fold m a b -> t m a -> t m b
860splitSuffixBy' predicate f m =
861 D.fromStreamD $ D.splitSuffixBy' predicate f (D.toStreamD m)
863------------------------------------------------------------------------------
864-- Split on a delimiter sequence
865------------------------------------------------------------------------------
867-- Int list examples for splitOn:
869-- >>> splitList [] [1,2,3,3,4]
870-- > [[1],[2],[3],[3],[4]]
872-- >>> splitList [5] [1,2,3,3,4]
875-- >>> splitList [1] [1,2,3,3,4]
878-- >>> splitList [4] [1,2,3,3,4]
881-- >>> splitList [2] [1,2,3,3,4]
884-- >>> splitList [3] [1,2,3,3,4]
887-- >>> splitList [3,3] [1,2,3,3,4]
890-- >>> splitList [1,2,3,3,4] [1,2,3,3,4]
893-- | Split the stream on both sides of a separator sequence, dropping the
896-- For illustration, let's define a function that operates on pure lists:
899-- splitOn_ pat xs = S.toList $ FL.splitOn (A.fromList pat) (FL.toList) (S.fromList xs)
902-- >>> splitOn_ "" "hello"
903-- > ["h","e","l","l","o"]
905-- >>> splitOn_ "hello" ""
908-- >>> splitOn_ "hello" "hello"
911-- >>> splitOn_ "x" "hello"
914-- >>> splitOn_ "h" "hello"
917-- >>> splitOn_ "o" "hello"
920-- >>> splitOn_ "e" "hello"
923-- >>> splitOn_ "l" "hello"
926-- >>> splitOn_ "ll" "hello"
929-- 'splitOn' is an inverse of 'intercalate'. The following law always holds:
931-- > intercalate . splitOn == id
933-- The following law holds when the separator is non-empty and contains none of
934-- the elements present in the input lists:
936-- > splitOn . intercalate == id
938-- The following law always holds:
940-- > concat . splitOn . intercalate == concat
943{-# INLINE splitOn #-}
945 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
946 => Array a -> Fold m a b -> t m a -> t m b
947splitOn patt f m = D.fromStreamD $ D.splitOn patt f (D.toStreamD m)
950-- This can be implemented easily using Rabin Karp
951-- | Split on any one of the given patterns.
952{-# INLINE splitOnAny #-}
954 :: (IsStream t, Monad m, Storable a, Integral a)
955 => [Array a] -> Fold m a b -> t m a -> t m b
956splitOnAny subseq f m = undefined -- D.fromStreamD $ D.splitOnAny f subseq (D.toStreamD m)
959-- | Like 'splitSuffixBy' but the separator is a sequence of elements, instead
960-- of a predicate for a single element.
962-- > splitSuffixOn_ pat xs = S.toList $ FL.splitSuffixOn (A.fromList pat) (FL.toList) (S.fromList xs)
964-- >>> splitSuffixOn_ "." ""
967-- >>> splitSuffixOn_ "." "."
970-- >>> splitSuffixOn_ "." "a"
973-- >>> splitSuffixOn_ "." ".a"
976-- >>> splitSuffixOn_ "." "a."
979-- >>> splitSuffixOn_ "." "a.b"
982-- >>> splitSuffixOn_ "." "a.b."
985-- >>> splitSuffixOn_ "." "a..b.."
988-- > lines = splitSuffixOn "\n"
991{-# INLINE splitSuffixOn #-}
993 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
994 => Array a -> Fold m a b -> t m a -> t m b
995splitSuffixOn patt f m =
996 D.fromStreamD $ D.splitSuffixOn False patt f (D.toStreamD m)
999-- | Like 'splitOn' but drops any empty splits.
1001{-# INLINE wordsOn #-}
1003 :: (IsStream t, Monad m, Storable a, Eq a)
1004 => Array a -> Fold m a b -> t m a -> t m b
1005wordsOn subseq f m = undefined -- D.fromStreamD $ D.wordsOn f subseq (D.toStreamD m)
1008-- XXX use a non-monadic intersperse to remove the MonadAsync constraint.
1010-- | Like 'splitOn' but splits the separator as well, as an infix token.
1012-- > splitOn'_ pat xs = S.toList $ FL.splitOn' (A.fromList pat) (FL.toList) (S.fromList xs)
1014-- >>> splitOn'_ "" "hello"
1015-- > ["h","","e","","l","","l","","o"]
1017-- >>> splitOn'_ "hello" ""
1020-- >>> splitOn'_ "hello" "hello"
1023-- >>> splitOn'_ "x" "hello"
1026-- >>> splitOn'_ "h" "hello"
1029-- >>> splitOn'_ "o" "hello"
1032-- >>> splitOn'_ "e" "hello"
1035-- >>> splitOn'_ "l" "hello"
1036-- > ["he","l","","l","o"]
1038-- >>> splitOn'_ "ll" "hello"
1042{-# INLINE splitOn' #-}
1044 :: (IsStream t, MonadAsync m, Storable a, Enum a, Eq a)
1045 => Array a -> Fold m a b -> t m a -> t m b
1046splitOn' patt f m = S.intersperseM (foldl' f (A.read patt)) $ splitOn patt f m
1048-- | Like 'splitSuffixOn' but keeps the suffix intact in the splits.
1050-- > splitSuffixOn'_ pat xs = S.toList $ FL.splitSuffixOn' (A.fromList pat) (FL.toList) (S.fromList xs)
1052-- >>> splitSuffixOn'_ "." ""
1055-- >>> splitSuffixOn'_ "." "."
1058-- >>> splitSuffixOn'_ "." "a"
1061-- >>> splitSuffixOn'_ "." ".a"
1064-- >>> splitSuffixOn'_ "." "a."
1067-- >>> splitSuffixOn'_ "." "a.b"
1070-- >>> splitSuffixOn'_ "." "a.b."
1073-- >>> splitSuffixOn'_ "." "a..b.."
1074-- > ["a.",".","b.","."]
1077{-# INLINE splitSuffixOn' #-}
1079 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
1080 => Array a -> Fold m a b -> t m a -> t m b
1081splitSuffixOn' patt f m =
1082 D.fromStreamD $ D.splitSuffixOn True patt f (D.toStreamD m)
1085-- This can be implemented easily using Rabin Karp
1086-- | Split post any one of the given patterns.
1087{-# INLINE splitSuffixOnAny #-}
1089 :: (IsStream t, Monad m, Storable a, Integral a)
1090 => [Array a] -> Fold m a b -> t m a -> t m b
1091splitSuffixOnAny subseq f m = undefined
1092 -- D.fromStreamD $ D.splitPostAny f subseq (D.toStreamD m)
1095------------------------------------------------------------------------------
1096-- Reorder in sequence
1097------------------------------------------------------------------------------
1100-- Buffer until the next element in sequence arrives. The function argument
1101-- determines the difference in sequence numbers. This could be useful in
1102-- implementing sequenced streams, for example, TCP reassembly.
1103{-# INLINE reassembleBy #-}
1105 :: (IsStream t, Monad m)
1110reassembleBy = undefined
1113------------------------------------------------------------------------------
1115------------------------------------------------------------------------------
1117-- | Distribute one copy of the stream to each fold and zip the results.
1120-- |-------Fold m a b--------|
1121-- ---stream m a---| |---m (b,c)
1122-- |-------Fold m a c--------|
1124-- >>> FL.foldl' (FL.tee FL.sum FL.length) (S.enumerateFromTo 1.0 100.0)
1129tee :: Monad m => Fold m a b -> Fold m a c -> Fold m a (b,c)
1130tee f1 f2 = (,) <$> f1 <*> f2
1132{-# INLINE foldNil #-}
1133foldNil :: Monad m => Fold m a [b]
1134foldNil = Fold step begin done where
1136 step _ _ = return []
1139{-# INLINE foldCons #-}
1140foldCons :: Monad m => Fold m a b -> Fold m a [b] -> Fold m a [b]
1141foldCons (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1142 Fold step begin done
1146 begin = Tuple' <$> beginL <*> beginR
1147 step (Tuple' xL xR) a = Tuple' <$> stepL xL a <*> stepR xR a
1148 done (Tuple' xL xR) = (:) <$> (doneL xL) <*> (doneR xR)
1150-- XXX use "List" instead of "[]"?, use Array for output to scale it to a large
1151-- number of consumers?
1153-- | Distribute one copy of the stream to each fold and collect the results in
1158-- |-------Fold m a b--------|
1159-- ---stream m a---| |---m [b]
1160-- |-------Fold m a b--------|
1165-- >>> FL.foldl' (FL.distribute [FL.sum, FL.length]) (S.enumerateFromTo 1 5)
1168-- This is the consumer side dual of the producer side 'sequence' operation.
1171{-# INLINE distribute #-}
1172distribute :: Monad m => [Fold m a b] -> Fold m a [b]
1173distribute [] = foldNil
1174distribute (x:xs) = foldCons x (distribute xs)
1176------------------------------------------------------------------------------
1178------------------------------------------------------------------------------
1180-- | Partition the input over two folds using an 'Either' partitioning
1185-- |-------Fold b x--------|
1186-- -----stream m a --> (Either b c)----| |----(x,y)
1187-- |-------Fold c y--------|
1190-- Send input to either fold randomly:
1192-- >>> import System.Random (randomIO)
1193-- >>> randomly a = randomIO >>= \x -> return $ if x then Left a else Right a
1194-- >>> FL.foldl' (FL.partitionByM randomly FL.length FL.length) (S.enumerateFromTo 1 100)
1197-- Send input to the two folds in a proportion of 2:1:
1200-- import Data.IORef (newIORef, readIORef, writeIORef)
1201-- proportionately m n = do
1202-- ref <- newIORef $ cycle $ concat [replicate m Left, replicate n Right]
1203-- return $ \\a -> do
1204-- r <- readIORef ref
1205-- writeIORef ref $ tail r
1209-- f <- proportionately 2 1
1210-- r <- FL.foldl' (FL.partitionByM f FL.length FL.length) (S.enumerateFromTo (1 :: Int) 100)
1217-- This is the consumer side dual of the producer side 'mergeBy' operation.
1220{-# INLINE partitionByM #-}
1221partitionByM :: Monad m
1222 => (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1223partitionByM f (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1225 Fold step begin done
1229 begin = Tuple' <$> beginL <*> beginR
1230 step (Tuple' xL xR) a = do
1233 Left b -> Tuple' <$> stepL xL b <*> return xR
1234 Right c -> Tuple' <$> return xL <*> stepR xR c
1235 done (Tuple' xL xR) = (,) <$> doneL xL <*> doneR xR
1237-- Note: we could use (a -> Bool) instead of (a -> Either b c), but the latter
1238-- makes the signature clearer as to which case belongs to which fold.
1239-- XXX need to check the performance in both cases.
1241-- | Same as 'partitionByM' but with a pure partition function.
1243-- Count even and odd numbers in a stream:
1246-- >>> let f = FL.partitionBy (\\n -> if even n then Left n else Right n)
1247-- (fmap (("Even " ++) . show) FL.length)
1248-- (fmap (("Odd " ++) . show) FL.length)
1249-- in FL.foldl' f (S.enumerateFromTo 1 100)
1250-- ("Even 50","Odd 50")
1254{-# INLINE partitionBy #-}
1255partitionBy :: Monad m
1256 => (a -> Either b c) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1257partitionBy f = partitionByM (return . f)
1259-- | Compose two folds such that the combined fold accepts a stream of 'Either'
1260-- and routes the 'Left' values to the first fold and 'Right' values to the
1263-- > partition = partitionBy id
1266{-# INLINE partition #-}
1268 => Fold m b x -> Fold m c y -> Fold m (Either b c) (x, y)
1269partition = partitionBy id
1272-- | Send one item to each fold in a round-robin fashion. This is the consumer
1273-- side dual of producer side 'mergeN' operation.
1275-- partitionN :: Monad m => [Fold m a b] -> Fold m a [b]
1276-- partitionN fs = Fold step begin done
1279-- Demultiplex an input element into a number of typed variants. We want to
1280-- statically restrict the target values within a set of predefined types, an
1281-- enumeration of a GADT. We also want to make sure that the Map contains only
1282-- those types and the full set of those types. Instead of Map it should
1283-- probably be a lookup-table using an array and not in GC memory.
1285-- This is the consumer side dual of the producer side 'mux' operation (XXX to
1288-- | Split the input stream based on a key field and fold each split using a
1289-- specific fold collecting the results in a map from the keys to the results.
1290-- Useful for cases like protocol handlers to handle different type of packets
1291-- using different handlers.
1295-- |-------Fold m a b
1296-- -----stream m a-----Map-----|
1297-- |-------Fold m a b
1303{-# INLINE demuxWith #-}
1304demuxWith :: (Monad m, Ord k)
1305 => (a -> k) -> Map k (Fold m a b) -> Fold m a (Map k b)
1306demuxWith f kv = Fold step initial extract
1312 -- XXX should we raise an exception in Nothing case?
1313 -- Ideally we should enforce that it is a total map over k so that look
1315 -- XXX we could use a monadic update function for a single lookup and
1316 -- update in the map.
1318 in case Map.lookup k mp of
1319 Nothing -> return mp
1320 Just (Fold step' acc extract') -> do
1321 !r <- acc >>= \x -> step' x a
1322 return $ Map.insert k (Fold step' (return r) extract') mp
1323 extract = Prelude.mapM (\(Fold _ acc e) -> acc >>= e)
1325-- | Fold a stream of key value pairs using a map of specific folds for each
1326-- key into a map from keys to the results of fold outputs of the corresponding
1330-- > let table = Data.Map.fromList [(\"SUM", FL.sum), (\"PRODUCT", FL.product)]
1331-- input = S.fromList [(\"SUM",1),(\"PRODUCT",2),(\"SUM",3),(\"PRODUCT",4)]
1332-- in FL.foldl' (FL.demux table) input
1339demux :: (Monad m, Ord k)
1340 => Map k (Fold m a b) -> Fold m (k, a) (Map k b)
1341demux fs = demuxWith fst (Map.map (lmap snd) fs)
1343-- | Split the input stream based on a key field and fold each split using a
1344-- specific fold without collecting the results. Useful for cases like protocol
1345-- handlers to handle different type of packets.
1349-- |-------Fold m a ()
1350-- -----stream m a-----Map-----|
1351-- |-------Fold m a ()
1359-- demuxWith_ can be slightly faster than demuxWith because we do not need to
1360-- update the Map in this case. This may be significant only if the map is
1362{-# INLINE demuxWith_ #-}
1363demuxWith_ :: (Monad m, Ord k)
1364 => (a -> k) -> Map k (Fold m a b) -> Fold m a ()
1365demuxWith_ f kv = Fold step initial extract
1370 Prelude.mapM (\(Fold s i e) ->
1371 i >>= \r -> return (Fold s (return r) e)) kv
1373 -- XXX should we raise an exception in Nothing case?
1374 -- Ideally we should enforce that it is a total map over k so that look
1376 case Map.lookup (f a) mp of
1377 Nothing -> return mp
1378 Just (Fold step' acc _) -> do
1379 _ <- acc >>= \x -> step' x a
1381 extract mp = Prelude.mapM (\(Fold _ acc e) -> acc >>= e) mp >> return ()
1383-- | Given a stream of key value pairs and a map from keys to folds, fold the
1384-- values for each key using the corresponding folds, discarding the outputs.
1387-- > let prn = FL.drainBy print
1388-- > let table = Data.Map.fromList [(\"ONE", prn), (\"TWO", prn)]
1389-- input = S.fromList [(\"ONE",1),(\"TWO",2)]
1390-- in FL.foldl' (FL.demux_ table) input
1396{-# INLINE demux_ #-}
1397demux_ :: (Monad m, Ord k) => Map k (Fold m a ()) -> Fold m (k, a) ()
1398demux_ fs = demuxWith_ fst (Map.map (lmap snd) fs)
1400-- XXX instead of a Map we could yield the results as a pure stream as they
1401-- complete. We could then concatMap the fold results to implement the
1402-- windowing combinators.
1404-- | Split the input stream based on a key field and fold each split using the
1405-- given fold. Useful for map/reduce, bucketizing the input in different bins
1406-- or for generating histograms.
1409-- > let input = S.fromList [(\"ONE",1),(\"ONE",1.1),(\"TWO",2), (\"TWO",2.2)]
1410-- in FL.foldl' (FL.classify FL.toListRev) input
1411-- fromList [(\"ONE",[1.1,1.0]),(\"TWO",[2.2,2.0])]
1415{-# INLINE classifyWith #-}
1416classifyWith :: (Monad m, Ord k) => (a -> k) -> Fold m a b -> Fold m a (Map k b)
1417classifyWith f (Fold step initial extract) = Fold step' initial' extract'
1421 initial' = return Map.empty
1424 in case Map.lookup k kv of
1428 return $ Map.insert k r kv
1431 return $ Map.insert k r kv
1432 extract' = Prelude.mapM extract
1434-- | Given an input stream of key value pairs and a fold for values, fold all
1435-- the values belonging to each key. Useful for map/reduce, bucketizing the
1436-- input in different bins or for generating histograms.
1439-- > let input = S.fromList [(\"ONE",1),(\"ONE",1.1),(\"TWO",2), (\"TWO",2.2)]
1440-- in FL.foldl' (FL.classify FL.toListRev) input
1441-- fromList [(\"ONE",[1.1,1.0]),(\"TWO",[2.2,2.0])]
1446-- > classify fld = classifyWith fst (lmap snd fld)
1449{-# INLINE classify #-}
1450classify :: (Monad m, Ord k) => Fold m a b -> Fold m (k, a) (Map k b)
1451classify fld = classifyWith fst (lmap snd fld)
1453------------------------------------------------------------------------------
1455------------------------------------------------------------------------------
1457-- | Like 'unzipWith' but with a monadic splitter function.
1460{-# INLINE unzipWithM #-}
1461unzipWithM :: Monad m
1462 => (a -> m (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
1463unzipWithM f (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1464 Fold step begin done
1468 step (Tuple' xL xR) a = do
1470 Tuple' <$> stepL xL b <*> stepR xR c
1471 begin = Tuple' <$> beginL <*> beginR
1472 done (Tuple' xL xR) = (,) <$> doneL xL <*> doneR xR
1474-- | Split elements in the input stream into two parts using a pure splitter
1475-- function, direct each part to a different fold and zip the results.
1478{-# INLINE unzipWith #-}
1480 => (a -> (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
1481unzipWith f = unzipWithM (return . f)
1483-- | Send the elements of tuples in a stream of tuples through two different
1488-- |-------Fold a x--------|
1489-- -----Stream m x----(a,b)--| |----m (x,y)
1490-- |-------Fold b y--------|
1494-- This is the consumer side dual of the producer side 'zip' operation.
1498unzip :: Monad m => Fold m a x -> Fold m b y -> Fold m (a,b) (x,y)
1501------------------------------------------------------------------------------
1503------------------------------------------------------------------------------
1505-- | Modify the fold such that when the fold is done, instead of returning the
1506-- accumulator, it returns a fold. The returned fold starts from where we left
1507-- i.e. it uses the last accumulator value as the initial value of the
1508-- accumulator. Thus we can resume the fold later and feed it more input.
1511-- > more <- FL.foldl (FL.duplicate FL.sum) (S.enumerateFromTo 1 10)
1512-- > evenMore <- FL.foldl (FL.duplicate more) (S.enumerateFromTo 11 20)
1513-- > FL.foldl evenMore (S.enumerateFromTo 21 30)
1517{-# INLINABLE duplicate #-}
1518duplicate :: Applicative m => Fold m a b -> Fold m a (Fold m a b)
1519duplicate (Fold step begin done) =
1520 Fold step begin (\x -> pure (Fold step (pure x) done))
1523-- All the stream flattening transformations can also be applied to a fold
1526-- | This can be used to apply all the stream generation operations on folds.
1527lconcatMap ::(IsStream t, Monad m) => (a -> t m b)
1530lconcatMap s f1 f2 = undefined
1533-- All the grouping transformation that we apply to a stream can also be
1534-- applied to a fold input stream.
1537-- | Group the input stream into groups of elements between @low@ and @high@.
1538-- Collection starts in chunks of @low@ and then keeps doubling until we reach
1539-- @high@. Each chunk is folded using the provided fold function.
1541-- This could be useful, for example, when we are folding a stream of unknown
1542-- size to a stream of arrays and we want to minimize the number of
1547-- XXX we should be able to implement it with parsers/terminating folds.
1549{-# INLINE lchunksInRange #-}
1550lchunksInRange :: Monad m
1551 => Int -> Int -> Fold m a b -> Fold m b c -> Fold m a c
1552lchunksInRange low high (Fold step1 initial1 extract1)
1553 (Fold step2 initial2 extract2) = undefined
1556-- | Group the input stream into groups of @n@ elements each and then fold each
1557-- group using the provided fold function.
1561-- -----Fold m a b----|-Fold n a c-|-Fold n a c-|-...-|----Fold m a c
1565{-# INLINE lchunksOf #-}
1566lchunksOf :: Monad m => Int -> Fold m a b -> Fold m b c -> Fold m a c
1567lchunksOf n (Fold step1 initial1 extract1) (Fold step2 initial2 extract2) =
1568 Fold step' initial' extract'
1572 initial' = (Tuple3' 0) <$> initial1 <*> initial2
1573 step' (Tuple3' i r1 r2) a = do
1577 return $ Tuple3' (i + 1) res r2
1580 acc2 <- step2 r2 res
1584 return $ Tuple3' 1 acc1 acc2
1585 extract' (Tuple3' _ _ r) = extract2 r
1587-- | Group the input stream into windows of n second each and then fold each
1588-- group using the provided fold function.
1590-- For example, we can copy and distribute a stream to multiple folds where
1591-- each fold can group the input differently e.g. by one second, one minute and
1592-- one hour windows respectively and fold each resulting stream of folds.
1596-- -----Fold m a b----|-Fold n a c-|-Fold n a c-|-...-|----Fold m a c
1599{-# INLINE lsessionsOf #-}
1600lsessionsOf :: MonadAsync m => Double -> Fold m a b -> Fold m b c -> Fold m a c
1601lsessionsOf n (Fold step1 initial1 extract1) (Fold step2 initial2 extract2) =
1602 Fold step' initial' extract'
1606 -- XXX MVar may be expensive we need a cheaper synch mechanism here
1610 mv1 <- liftIO $ newMVar i1
1611 mv2 <- liftIO $ newMVar (Right i2)
1612 t <- control $ \run ->
1613 mask $ \restore -> do
1614 tid <- forkIO $ catch (restore $ void $ run (timerThread mv1 mv2))
1615 (handleChildException mv2)
1617 return $ Tuple3' t mv1 mv2
1618 step' acc@(Tuple3' _ mv1 _) a = do
1619 r1 <- liftIO $ takeMVar mv1
1621 liftIO $ putMVar mv1 res
1623 extract' (Tuple3' tid _ mv2) = do
1624 r2 <- liftIO $ takeMVar mv2
1625 liftIO $ killThread tid
1628 Right x -> extract2 x
1630 timerThread mv1 mv2 = do
1631 liftIO $ threadDelay (round $ n * 1000000)
1633 r1 <- liftIO $ takeMVar mv1
1635 liftIO $ putMVar mv1 i1
1638 r2 <- liftIO $ takeMVar mv2
1641 Right x -> fmap Right $ step2 x res1
1642 liftIO $ putMVar mv2 res
1645 handleChildException ::
1646 MVar (Either SomeException a) -> SomeException -> IO ()
1647 handleChildException mv2 e = do
1654------------------------------------------------------------------------------
1655-- Windowed classification
1656------------------------------------------------------------------------------
1658-- We divide the stream into windows or chunks in space or time and each window
1659-- can be associated with a key, all events associated with a particular key in
1660-- the window can be folded to a single result. The stream can be split into
1661-- windows by size or by using a split predicate on the elements in the stream.
1662-- For example, when we receive a closing flag, we can close the window.
1664-- A "chunk" is a space window and a "session" is a time window. Are there any
1665-- other better short words to describe them. An alternative is to use
1666-- "swindow" and "twindow". Another word for "session" could be "spell".
1668-- TODO: To mark the position in space or time we can have Indexed or
1669-- TimeStamped types. That can make it easy to deal with the position indices
1672------------------------------------------------------------------------------
1673-- Keyed Sliding Windows
1674------------------------------------------------------------------------------
1677{-# INLINABLE classifySlidingChunks #-}
1678classifySlidingChunks
1679 :: (IsStream t, MonadAsync m, Ord k)
1680 => Int -- ^ window size
1681 -> Int -- ^ window slide
1682 -> Fold m a b -- ^ Fold to be applied to window events
1683 -> t m (k, a, Bool) -- ^ window key, data, close event
1685classifySlidingChunks wsize wslide (Fold step initial extract) str
1688-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
1689-- slide the send window when an ack is received and we slide the receive
1690-- window when a sequence is complete. Sliding is stateful in case of TCP,
1691-- sliding releases the send buffer or makes data available to the user from
1692-- the receive buffer.
1693{-# INLINABLE classifySlidingSessions #-}
1694classifySlidingSessions
1695 :: (IsStream t, MonadAsync m, Ord k)
1696 => Double -- ^ timer tick in seconds
1697 -> Double -- ^ time window size
1698 -> Double -- ^ window slide
1699 -> Fold m a b -- ^ Fold to be applied to window events
1700 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
1702classifySlidingSessions tick interval slide (Fold step initial extract) str
1706------------------------------------------------------------------------------
1707-- Sliding Window Buffers
1708------------------------------------------------------------------------------
1710-- These buffered versions could be faster than concurrent incremental folds of
1711-- all overlapping windows as in many cases we may not need all the values to
1712-- compute the fold, we can just compute the result using the old value and new
1713-- value. However, we may need the buffer once in a while, for example for
1714-- string search we usually compute the hash incrementally but when the hash
1715-- matches the hash of the pattern we need to compare the whole string.
1717-- XXX we should be able to implement sequence based splitting combinators
1718-- using this combinator.
1721-- | Buffer n elements of the input in a ring buffer. When t new elements are
1722-- collected, slide the window to remove the same number of oldest elements,
1723-- insert the new elements, and apply an incremental fold on the sliding
1724-- window, supplying the outgoing elements, the new ring buffer as arguments.
1726 :: (IsStream t, Monad m, Ord a, Storable a)
1727 => Int -- window size
1728 -> Int -- window slide
1729 -> Fold m (Ring a, Array a) b
1732slidingChunkBuffer = undefined
1734-- Buffer n seconds worth of stream elements of the input in a radix tree.
1735-- Every t seconds, remove the items that are older than n seconds, and apply
1736-- an incremental fold on the sliding window, supplying the outgoing elements,
1737-- and the new radix tree buffer as arguments.
1739 :: (IsStream t, Monad m, Ord a, Storable a)
1740 => Int -- window size
1742 -> Fold m (RTree a, Array a) b
1745slidingSessionBuffer = undefined
1748------------------------------------------------------------------------------
1749-- Keyed Session Windows
1750------------------------------------------------------------------------------
1753-- | Keyed variable size space windows. Close the window if we do not receive a
1754-- window event in the next "spaceout" elements.
1755{-# INLINABLE classifyChunksBy #-}
1757 :: (IsStream t, MonadAsync m, Ord k)
1758 => Int -- ^ window spaceout (spread)
1759 -> Bool -- ^ reset the spaceout when a chunk window element is received
1760 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1761 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1763classifyChunksBy spanout reset (Fold step initial extract) str = undefined
1765-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
1766-- received within the chunk size window. The chunk gets closed only if no
1767-- element is received within the chunk window.
1769{-# INLINABLE classifyKeepAliveChunks #-}
1770classifyKeepAliveChunks
1771 :: (IsStream t, MonadAsync m, Ord k)
1772 => Int -- ^ window spaceout (spread)
1773 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1774 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1776classifyKeepAliveChunks spanout = classifyChunksBy spanout True
1779-- | @classifySessionsBy tick timeout reset f stream@ groups together all input
1780-- stream elements that belong to the same session. @timeout@ is the maximum
1781-- lifetime of a session in seconds. All elements belonging to a session are
1782-- purged after this duration. If "reset" is 'Ture' then the timeout is reset
1783-- after every event received in the session. Session duration is measured
1784-- using the timestamp of the first element seen for that session. To detect
1785-- session timeouts, a monotonic event time clock is maintained using the
1786-- timestamps seen in the inputs and a timer with a tick duration specified by
1789-- @session key@ is a key that uniquely identifies the session for the given
1790-- element, @timestamp@ characterizes the time when the input element was
1791-- generated, this is an absolute time measured from some @Epoch@. @session
1792-- close@ is a boolean indicating whether this element marks the closing of the
1793-- session. When an input element with @session close@ set to @True@ is seen
1794-- the session is purged immediately.
1796-- All the input elements belonging to a session are collected using the fold
1797-- @f@. The session key and the fold result are emitted in the output stream
1798-- when the session is purged either via the session close event or via the
1799-- session lifetime timeout.
1802{-# INLINABLE classifySessionsBy #-}
1804 :: (IsStream t, MonadAsync m, Ord k)
1805 => Double -- ^ timer tick in seconds
1806 -> Double -- ^ session timeout
1807 -> Bool -- ^ reset the timeout when an event is received
1808 -> Fold m a b -- ^ Fold to be applied to session events
1809 -> t m (k, a, Bool, AbsTime) -- ^ session key, timestamp, close event, data
1811classifySessionsBy tick timeout reset (Fold step initial extract) str =
1812 S.concatMap (\(Tuple4' _ _ _ s) -> s) $ S.scanlM' sstep szero stream
1816 timeoutMs = toRelTime (round (timeout * 1000) :: MilliSecond64)
1817 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
1818 szero = Tuple4' (toAbsTime (0 :: MilliSecond64)) H.empty Map.empty S.nil
1820 -- Got a new stream input element
1821 sstep (Tuple4' evTime hp mp _) (Just (key, a, closing, ts)) =
1822 -- XXX we should use a heap in pinned memory to scale it to a large
1825 -- deleting a key from the heap is expensive, so we never delete a
1826 -- key, we just purge it from the Map and it gets purged from the
1827 -- heap on timeout. We just need an extra lookup in the Map when
1828 -- the key is purged from the heap, that should not be expensive.
1830 -- To detect session inactivity we keep a timestamp of the latest event
1831 -- in the Map along with the fold result. When we purge the session
1832 -- from the heap we match the timestamp in the heap with the timestamp
1833 -- in the Map, if the latest timestamp is newer and has not expired we
1834 -- reinsert the key in the heap.
1836 -- XXX if the key is an Int, we can also use an IntMap for slightly
1837 -- better performance.
1839 let accumulate v = do
1840 Tuple' _ old <- maybe (initial >>= return . Tuple' ts) return v
1842 return $ Tuple' ts new
1845 let (r, mp') = Map.updateLookupWithKey (\_ _ -> Nothing) key mp
1846 Tuple' _ acc <- accumulate r
1848 return $ Tuple4' evTime hp mp' (S.fromPure (key, res))
1850 let r = Map.lookup key mp
1852 let mp' = Map.insert key acc mp
1856 let expiry = addToAbsTime ts timeoutMs
1857 in H.insert (Entry expiry key) hp
1859 -- Event time is maintained as monotonically increasing
1860 -- time. If we have lagged behind any of the timestamps
1861 -- seen then we increase it to match the latest time seen
1862 -- in the timestamps. We also increase it on timer ticks.
1863 return $ Tuple4' (max evTime ts) hp' mp' S.nil
1865 -- Got a timer tick event
1866 -- XXX can we yield the entries without accumulating them?
1867 sstep (Tuple4' evTime heap sessions _) Nothing = do
1868 (hp', mp', out) <- go heap sessions S.nil
1869 return $ Tuple4' curTime hp' mp' out
1873 curTime = addToAbsTime evTime tickMs
1875 let hres = H.uncons hp
1877 Just (Entry ts key, hp') -> do
1878 let duration = diffAbsTime curTime ts
1879 if duration >= timeoutMs
1881 let (r, mp') = Map.updateLookupWithKey
1882 (\_ _ -> Nothing) key mp
1884 Nothing -> go hp' mp' out
1885 Just (Tuple' latestTS acc) -> do
1886 let dur = diffAbsTime curTime latestTS
1887 if dur >= timeoutMs || not reset
1890 go hp' mp' ((key, sess) `S.cons` out)
1892 -- reset the session timeout
1893 let expiry = addToAbsTime latestTS timeoutMs
1894 hp'' = H.insert (Entry expiry key) hp'
1895 mp'' = Map.insert key (Tuple' latestTS acc) mp'
1897 else return (hp, mp, out)
1898 Nothing -> return (hp, mp, out)
1900 -- merge timer events in the stream
1901 stream = S.map Just str `parallel` S.repeatM timer
1903 liftIO $ threadDelay (round $ tick * 1000000)
1906-- | Like 'classifySessionsOf' but the session is kept alive if an event is
1907-- received within the session window. The session times out and gets closed
1908-- only if no event is received within the specified session window size.
1911{-# INLINABLE classifyKeepAliveSessions #-}
1912classifyKeepAliveSessions
1913 :: (IsStream t, MonadAsync m, Ord k)
1914 => Double -- ^ session inactive timeout
1915 -> Fold m a b -- ^ Fold to be applied to session payload data
1916 -> t m (k, a, Bool, AbsTime) -- ^ session key, data, close flag, timestamp
1918classifyKeepAliveSessions timeout = classifySessionsBy 1 timeout True
1920------------------------------------------------------------------------------
1921-- Keyed tumbling windows
1922------------------------------------------------------------------------------
1924-- Tumbling windows is a special case of sliding windows where the window slide
1925-- is the same as the window size. Or it can be a special case of session
1926-- windows where the reset flag is set to False.
1928-- XXX instead of using the early termination flag in the stream, we can use an
1929-- early terminating fold instead.
1932-- | Split the stream into fixed size chunks of specified size. Within each
1933-- such chunk fold the elements in buckets identified by the keys. A particular
1934-- bucket fold can be terminated early if a closing flag is encountered in an
1935-- element for that key.
1938{-# INLINABLE classifyChunksOf #-}
1940 :: (IsStream t, MonadAsync m, Ord k)
1941 => Int -- ^ window size
1942 -> Fold m a b -- ^ Fold to be applied to window events
1943 -> t m (k, a, Bool) -- ^ window key, data, close event
1945classifyChunksOf wsize = classifyChunksBy wsize False
1948-- | Split the stream into fixed size time windows of specified interval in
1949-- seconds. Within each such window, fold the elements in buckets identified by
1950-- the keys. A particular bucket fold can be terminated early if a closing flag
1951-- is encountered in an element for that key. Once a fold is terminated the key
1952-- and value for that bucket are emitted in the output stream.
1954-- Session @timestamp@ in the input stream is an absolute time from some epoch,
1955-- characterizing the time when the input element was generated. To detect
1956-- session window end, a monotonic event time clock is maintained synced with
1957-- the timestamps with a clock resolution of 1 second.
1960{-# INLINABLE classifySessionsOf #-}
1962 :: (IsStream t, MonadAsync m, Ord k)
1963 => Double -- ^ time window size
1964 -> Fold m a b -- ^ Fold to be applied to window events
1965 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
1967classifySessionsOf 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