2{-# LANGUAGE RankNTypes #-}
3{-# LANGUAGE FlexibleContexts #-}
5#if __GLASGOW_HASKELL__ >= 800
6{-# OPTIONS_GHC -Wno-orphans #-}
9#include "../Streams/inline.hs"
12-- Module : Streamly.Internal.Prelude
13-- Copyright : (c) 2017 Harendra Kumar
16-- Maintainer : harendra.kumar@gmail.com
17-- Stability : experimental
21module Streamly.Internal.Prelude
79 -- ** Composable Left Folds
84 -- -- ** To Summary (Full Folds)
92 -- -- ** To Summary (Maybe) (Full Folds)
98 , toListRev -- experimental
104 -- -- ** To Containers (Full Folds)
109 -- -- ** To Elements (Partial Folds)
113 -- -- | Folds that extract selected elements of a stream or their properties.
122 -- -- ** To Boolean (Partial Folds)
151 -- ** Scan Using Fold
169 -- , timestampedR -- timer
176 -- ** Stateful Filters
190 -- , uniqBy -- by predicate e.g. to remove duplicate "/" in a path
191 -- , uniqOn -- to remove duplicate sequences
192 -- , pruneBy -- dropAround + uniqBy - like words
194 -- ** Mapping Filters
198 -- ** Scanning Filters
201 -- , seqIndices -- search a sequence in the stream
208 -- , intersperseBySpan
215 -- * Multi-Stream Operations
252 , concatUnfoldInterleave
253 , concatUnfoldRoundrobin
263 -- , splitIn -- sessionN
267 , break -- breakBefore
285 -- -- *** Using Element Separators
293 , wordsBy -- stripAndCompactBy
295 -- -- *** Using Sequence Separators
298 -- , splitOnPrefixSeq
300 -- Keeping the delimiters
303 -- , splitByPrefixSeq
306 -- Splitting using multiple sequence separators
308 -- , splitOnAnySuffixSeq
309 -- , splitOnAnyPrefixSeq
325 -- * Windowed Classification
327 -- ** Tumbling Windows
328 -- , classifyChunksOf
332 -- ** Keep Alive Windows
333 -- , classifyKeepAliveChunks
334 , classifyKeepAliveSessions
337 -- ** Sliding Windows
338 , classifySlidingChunks
339 , classifySlidingSessions
341 -- ** Sliding Window Buffers
342 -- , slidingChunkBuffer
343 -- , slidingSessionBuffer
345 -- ** Containers of Streams
387import Control.Concurrent (threadDelay)
388import Control.Exception (Exception)
389import Control.Monad (void)
390import Control.Monad.Catch (MonadCatch)
391import Control.Monad.IO.Class (MonadIO(..))
392import Control.Monad.Trans (MonadTrans(..))
393import Data.Functor.Identity (Identity)
394import Data.Heap (Entry(..))
395import Data.Maybe (isJust, fromJust, isNothing)
396import Foreign.Storable (Storable)
398 hiding (filter, drop, dropWhile, take, takeWhile, zipWith, foldr,
399 foldl, map, mapM, mapM_, sequence, all, any, sum, product, elem,
400 notElem, maximum, minimum, head, last, tail, length, null,
401 reverse, iterate, init, and, or, lookup, foldr1, (!!),
402 scanl, scanl1, replicate, concatMap, span, splitAt, break)
404import qualified Data.Heap as H
405import qualified Data.Map.Strict as Map
406import qualified Prelude
407import qualified System.IO as IO
409import Streamly.Streams.Enumeration (Enumerable(..), enumerate, enumerateTo)
410import Streamly.Internal.Data.Fold.Types (Fold (..))
411import Streamly.Internal.Data.Unfold.Types (Unfold)
412import Streamly.Internal.Memory.Array.Types (Array, writeNUnsafe)
413-- import Streamly.Memory.Ring (Ring)
414import Streamly.Internal.Data.SVar (MonadAsync, defState)
415import Streamly.Streams.Async (mkAsync')
416import Streamly.Streams.Combinators (inspectMode, maxYields)
417import Streamly.Streams.Prelude
418 (fromStreamS, toStreamS, foldWith, foldMapWith, forEachWith)
419import Streamly.Streams.StreamD (fromStreamD, toStreamD)
420import Streamly.Streams.StreamK (IsStream((|:), consM))
421import Streamly.Streams.Serial (SerialT)
422import Streamly.Internal.Data.Pipe.Types (Pipe (..))
423import Streamly.Internal.Data.Time.Units
424 (AbsTime, MilliSecond64(..), addToAbsTime, diffAbsTime, toRelTime,
427import Streamly.Internal.Data.Strict
429import qualified Streamly.Internal.Memory.Array as A
430import qualified Streamly.Data.Fold as FL
431import qualified Streamly.Internal.Data.Fold.Types as FL
432import qualified Streamly.Streams.Prelude as P
433import qualified Streamly.Streams.StreamK as K
434import qualified Streamly.Streams.StreamD as D
435import qualified Streamly.Streams.Zip as Z
437#ifdef USE_STREAMK_ONLY
438import qualified Streamly.Streams.StreamK as S
439import qualified Streamly.Streams.Zip as S
441import qualified Streamly.Streams.StreamD as S
444import qualified Streamly.Streams.Serial as Serial
445import qualified Streamly.Streams.Parallel as Par
447------------------------------------------------------------------------------
449------------------------------------------------------------------------------
451-- | Decompose a stream into its head and tail. If the stream is empty, returns
452-- 'Nothing'. If the stream is non-empty, returns @Just (a, ma)@, where @a@ is
453-- the head of the stream and @ma@ its tail.
455-- This is a brute force primitive. Avoid using it as long as possible, use it
456-- when no other combinator can do the job. This can be used to do pretty much
457-- anything in an imperative manner, as it just breaks down the stream into
458-- individual elements and we can loop over them as we deem fit. For example,
459-- this can be used to convert a streamly stream into other stream types.
463uncons :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (a, t m a))
464uncons m = K.uncons (K.adapt m)
466------------------------------------------------------------------------------
467-- Generation by Unfolding
468------------------------------------------------------------------------------
475-- Just (a, b) -> a \`cons` unfoldr step b
478-- Build a stream by unfolding a /pure/ step function @step@ starting from a
479-- seed @s@. The step function returns the next element in the stream and the
480-- next seed value. When it is done it returns 'Nothing' and the stream ends.
487-- else Just (b, b + 1)
488-- in toList $ unfoldr f 0
495{-# INLINE_EARLY unfoldr #-}
496unfoldr :: (Monad m, IsStream t) => (b -> Maybe (a, b)) -> b -> t m a
497unfoldr step seed = fromStreamS (S.unfoldr step seed)
498{-# RULES "unfoldr fallback to StreamK" [1]
499 forall a b. S.toStreamK (S.unfoldr a b) = K.unfoldr a b #-}
501-- | Build a stream by unfolding a /monadic/ step function starting from a
502-- seed. The step function returns the next element in the stream and the next
503-- seed value. When it is done it returns 'Nothing' and the stream ends. For
509-- then return Nothing
510-- else print b >> return (Just (b, b + 1))
511-- in drain $ unfoldrM f 0
519-- When run concurrently, the next unfold step can run concurrently with the
520-- processing of the output of the previous step. Note that more than one step
521-- cannot run concurrently as the next step depends on the output of the
525-- (asyncly $ S.unfoldrM (\\n -> liftIO (threadDelay 1000000) >> return (Just (n, n + 1))) 0)
526-- & S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
532{-# INLINE_EARLY unfoldrM #-}
533unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a
536{-# RULES "unfoldrM serial" unfoldrM = unfoldrMSerial #-}
537{-# INLINE_EARLY unfoldrMSerial #-}
538unfoldrMSerial :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> SerialT m a
539unfoldrMSerial step seed = fromStreamS (S.unfoldrM step seed)
541-- | Convert an 'Unfold' into a stream by supplying it an input seed.
543-- >>> unfold UF.replicateM 10 (putStrLn "hello")
547unfold :: (IsStream t, Monad m) => Unfold m a b -> a -> t m b
548unfold unf x = fromStreamD $ D.unfold unf x
550------------------------------------------------------------------------------
551-- Specialized Generation
552------------------------------------------------------------------------------
554-- Faster than yieldM because there is no bind.
558-- yield a = a \`cons` nil
561-- Create a singleton stream from a pure value.
563-- The following holds in monadic streams, but not in Zip streams:
567-- yield = yieldM . pure
570-- In Zip applicative streams 'yield' is not the same as 'pure' because in that
571-- case 'pure' is equivalent to 'repeat' instead. 'yield' and 'pure' are
572-- equally efficient, in other cases 'yield' may be slightly more efficient
573-- than the other equivalent definitions.
577yield :: IsStream t => a -> t m a
582-- yieldM m = m \`consM` nil
585-- Create a singleton stream from a monadic action.
588-- > toList $ yieldM getLine
595yieldM :: (Monad m, IsStream t) => m a -> t m a
600-- fromIndices f = let g i = f i \`cons` g (i + 1) in g 0
603-- Generate an infinite stream, whose values are the output of a function @f@
604-- applied on the corresponding index. Index starts at 0.
607-- > S.toList $ S.take 5 $ S.fromIndices id
612{-# INLINE fromIndices #-}
613fromIndices :: (IsStream t, Monad m) => (Int -> a) -> t m a
614fromIndices = fromStreamS . S.fromIndices
619-- fromIndicesM f = let g i = f i \`consM` g (i + 1) in g 0
622-- Generate an infinite stream, whose values are the output of a monadic
623-- function @f@ applied on the corresponding index. Index starts at 0.
628{-# INLINE_EARLY fromIndicesM #-}
629fromIndicesM :: (IsStream t, MonadAsync m) => (Int -> m a) -> t m a
630fromIndicesM = K.fromIndicesM
632{-# RULES "fromIndicesM serial" fromIndicesM = fromIndicesMSerial #-}
633{-# INLINE fromIndicesMSerial #-}
634fromIndicesMSerial :: MonadAsync m => (Int -> m a) -> SerialT m a
635fromIndicesMSerial = fromStreamS . S.fromIndicesM
639-- replicateM = take n . repeatM
642-- Generate a stream by performing a monadic action @n@ times. Same as:
645-- drain $ serially $ S.replicateM 10 $ (threadDelay 1000000 >> print 1)
646-- drain $ asyncly $ S.replicateM 10 $ (threadDelay 1000000 >> print 1)
652{-# INLINE_EARLY replicateM #-}
653replicateM :: (IsStream t, MonadAsync m) => Int -> m a -> t m a
654replicateM = K.replicateM
656{-# RULES "replicateM serial" replicateM = replicateMSerial #-}
657{-# INLINE replicateMSerial #-}
658replicateMSerial :: MonadAsync m => Int -> m a -> SerialT m a
659replicateMSerial n = fromStreamS . S.replicateM n
663-- replicate = take n . repeat
666-- Generate a stream of length @n@ by repeating a value @n@ times.
669{-# INLINE_NORMAL replicate #-}
670replicate :: (IsStream t, Monad m) => Int -> a -> t m a
671replicate n = fromStreamS . S.replicate n
675-- repeatM = fix . consM
676-- repeatM = cycle1 . yieldM
679-- Generate a stream by repeatedly executing a monadic action forever.
682-- drain $ serially $ S.take 10 $ S.repeatM $ (threadDelay 1000000 >> print 1)
683-- drain $ asyncly $ S.take 10 $ S.repeatM $ (threadDelay 1000000 >> print 1)
686-- /Concurrent, infinite (do not use with 'parallely')/
689repeatM :: (IsStream t, MonadAsync m) => m a -> t m a
691 where go m = m |: go m
695-- iterate f x = x \`cons` iterate f x
698-- Generate an infinite stream with @x@ as the first element and each
699-- successive element derived by applying the function @f@ on the previous
703-- > S.toList $ S.take 5 $ S.iterate (+1) 1
708iterate :: IsStream t => (a -> a) -> a -> t m a
709iterate step = K.fromStream . go
711 go s = K.cons s (go (step s))
715-- iterateM f m = m >>= \a -> return a \`consM` iterateM f (f a)
718-- Generate an infinite stream with the first element generated by the action
719-- @m@ and each successive element derived by applying the monadic function
720-- @f@ on the previous element.
722-- When run concurrently, the next iteration can run concurrently with the
723-- processing of the previous iteration. Note that more than one iteration
724-- cannot run concurrently as the next iteration depends on the output of the
725-- previous iteration.
728-- drain $ serially $ S.take 10 $ S.iterateM
729-- (\\x -> threadDelay 1000000 >> print x >> return (x + 1)) (return 0)
731-- drain $ asyncly $ S.take 10 $ S.iterateM
732-- (\\x -> threadDelay 1000000 >> print x >> return (x + 1)) (return 0)
737-- /Since: 0.7.0 (signature change)/
740iterateM :: (IsStream t, MonadAsync m) => (a -> m a) -> m a -> t m a
743 go s = K.mkStream $ \st stp sng yld -> do
745 K.foldStreamShared st stp sng yld (return next |: go (step next))
747------------------------------------------------------------------------------
749------------------------------------------------------------------------------
753-- fromListM = 'Prelude.foldr' 'K.consM' 'K.nil'
756-- Construct a stream from a list of monadic actions. This is more efficient
757-- than 'fromFoldableM' for serial streams.
760{-# INLINE_EARLY fromListM #-}
761fromListM :: (MonadAsync m, IsStream t) => [m a] -> t m a
762fromListM = fromStreamD . D.fromListM
763{-# RULES "fromListM fallback to StreamK" [1]
764 forall a. D.toStreamK (D.fromListM a) = fromFoldableM a #-}
768-- fromFoldableM = 'Prelude.foldr' 'consM' 'K.nil'
771-- Construct a stream from a 'Foldable' containing monadic actions.
774-- drain $ serially $ S.fromFoldableM $ replicateM 10 (threadDelay 1000000 >> print 1)
775-- drain $ asyncly $ S.fromFoldableM $ replicateM 10 (threadDelay 1000000 >> print 1)
778-- /Concurrent (do not use with 'parallely' on infinite containers)/
781{-# INLINE fromFoldableM #-}
782fromFoldableM :: (IsStream t, MonadAsync m, Foldable f) => f (m a) -> t m a
783fromFoldableM = Prelude.foldr consM K.nil
785-- | Same as 'fromFoldable'.
788{-# DEPRECATED each "Please use fromFoldable instead." #-}
790each :: (IsStream t, Foldable f) => f a -> t m a
793-- | Read lines from an IO Handle into a stream of Strings.
796{-# DEPRECATED fromHandle
797 "Please use Streamly.FileSystem.Handle module (see the changelog)" #-}
798fromHandle :: (IsStream t, MonadIO m) => IO.Handle -> t m String
801 go = K.mkStream $ \_ yld _ stp -> do
802 eof <- liftIO $ IO.hIsEOF h
806 str <- liftIO $ IO.hGetLine h
809------------------------------------------------------------------------------
810-- Elimination by Folding
811------------------------------------------------------------------------------
813-- | Right associative/lazy pull fold. @foldrM build final stream@ constructs
814-- an output structure using the step function @build@. @build@ is invoked with
815-- the next input element and the remaining (lazy) tail of the output
816-- structure. It builds a lazy output expression using the two. When the "tail
817-- structure" in the output expression is evaluated it calls @build@ again thus
818-- lazily consuming the input @stream@ until either the output expression built
819-- by @build@ is free of the "tail" or the input is exhausted in which case
820-- @final@ is used as the terminating case for the output structure. For more
821-- details see the description in the previous section.
823-- Example, determine if any element is 'odd' in a stream:
825-- >>> S.foldrM (\x xs -> if odd x then return True else xs) (return False) $ S.fromList (2:4:5:undefined)
828-- /Since: 0.7.0 (signature changed)/
830-- /Since: 0.2.0 (signature changed)/
834foldrM :: Monad m => (a -> m b -> m b) -> m b -> SerialT m a -> m b
837-- | Right fold to a streaming monad.
839-- > foldrS S.cons S.nil === id
841-- 'foldrS' can be used to perform stateless stream to stream transformations
842-- like map and filter in general. It can be coupled with a scan to perform
843-- stateful transformations. However, note that the custom map and filter
844-- routines can be much more efficient than this due to better stream fusion.
846-- >>> S.toList $ S.foldrS S.cons S.nil $ S.fromList [1..5]
849-- Find if any element in the stream is 'True':
851-- >>> S.toList $ S.foldrS (\x xs -> if odd x then return True else xs) (return False) $ (S.fromList (2:4:5:undefined) :: SerialT IO Int)
854-- Map (+2) on odd elements and filter out the even elements:
856-- >>> S.toList $ S.foldrS (\x xs -> if odd x then (x + 2) `S.cons` xs else xs) S.nil $ (S.fromList [1..5] :: SerialT IO Int)
859-- 'foldrM' can also be represented in terms of 'foldrS', however, the former
860-- is much more efficient:
862-- > foldrM f z s = runIdentityT $ foldrS (\x xs -> lift $ f x (runIdentityT xs)) (lift z) s
866foldrS :: IsStream t => (a -> t m b -> t m b) -> t m b -> t m a -> t m b
869-- | Right fold to a transformer monad. This is the most general right fold
870-- function. 'foldrS' is a special case of 'foldrT', however 'foldrS'
871-- implementation can be more efficient:
874-- > foldrM f z s = runIdentityT $ foldrT (\x xs -> lift $ f x (runIdentityT xs)) (lift z) s
876-- 'foldrT' can be used to translate streamly streams to other transformer
877-- monads e.g. to a different streaming type.
881foldrT :: (IsStream t, Monad m, Monad (s m), MonadTrans s)
882 => (a -> s m b -> s m b) -> s m b -> t m a -> s m b
883foldrT f z s = S.foldrT f z (toStreamS s)
885-- | Right fold, lazy for lazy monads and pure streams, and strict for strict
888-- Please avoid using this routine in strict monads like IO unless you need a
889-- strict right fold. This is provided only for use in lazy monads (e.g.
890-- Identity) or pure streams. Note that with this signature it is not possible
891-- to implement a lazy foldr when the monad @m@ is strict. In that case it
892-- would be strict in its accumulator and therefore would necessarily consume
897foldr :: Monad m => (a -> b -> b) -> b -> SerialT m a -> m b
900-- XXX This seems to be of limited use as it cannot be used to construct
901-- recursive structures and for reduction foldl1' is better.
903-- | Lazy right fold for non-empty streams, using first element as the starting
904-- value. Returns 'Nothing' if the stream is empty.
908{-# DEPRECATED foldr1 "Use foldrM instead." #-}
909foldr1 :: Monad m => (a -> a -> a) -> SerialT m a -> m (Maybe a)
910foldr1 f m = S.foldr1 f (toStreamS m)
912-- | Strict left fold with an extraction function. Like the standard strict
913-- left fold, but applies a user supplied extraction function (the third
914-- argument) to the folded value at the end. This is designed to work with the
915-- @foldl@ library. The suffix @x@ is a mnemonic for extraction.
918{-# DEPRECATED foldx "Please use foldl' followed by fmap instead." #-}
920foldx :: Monad m => (x -> a -> x) -> x -> (x -> b) -> SerialT m a -> m b
923-- | Left associative/strict push fold. @foldl' reduce initial stream@ invokes
924-- @reduce@ with the accumulator and the next input in the input stream, using
925-- @initial@ as the initial value of the current value of the accumulator. When
926-- the input is exhausted the current value of the accumulator is returned.
927-- Make sure to use a strict data structure for accumulator to not build
928-- unnecessary lazy expressions unless that's what you want. See the previous
929-- section for more details.
933foldl' :: Monad m => (b -> a -> b) -> b -> SerialT m a -> m b
936-- | Strict left fold, for non-empty streams, using first element as the
937-- starting value. Returns 'Nothing' if the stream is empty.
940{-# INLINE foldl1' #-}
941foldl1' :: Monad m => (a -> a -> a) -> SerialT m a -> m (Maybe a)
945 Nothing -> return Nothing
947 res <- foldl' step h t
950-- | Like 'foldx', but with a monadic step function.
953{-# DEPRECATED foldxM "Please use foldlM' followed by fmap instead." #-}
955foldxM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> SerialT m a -> m b
958-- | Like 'foldl'' but with a monadic step function.
961{-# INLINE foldlM' #-}
962foldlM' :: Monad m => (b -> a -> m b) -> b -> SerialT m a -> m b
963foldlM' step begin m = S.foldlM' step begin $ toStreamS m
965------------------------------------------------------------------------------
967------------------------------------------------------------------------------
969-- | Fold a stream using the supplied left fold.
971-- >>> S.fold FL.sum (S.enumerateFromTo 1 100)
976fold :: Monad m => Fold m a b -> SerialT m a -> m b
979------------------------------------------------------------------------------
981------------------------------------------------------------------------------
984-- | Drain a stream to a 'Sink'.
985{-# INLINE runSink #-}
986runSink :: Monad m => Sink m a -> SerialT m a -> m ()
987runSink = fold . toFold
990------------------------------------------------------------------------------
992------------------------------------------------------------------------------
995-- > drain = mapM_ (\_ -> return ())
997-- Run a stream, discarding the results. By default it interprets the stream
998-- as 'SerialT', to run other types of streams use the type adapting
999-- combinators for example @drain . 'asyncly'@.
1003drain :: Monad m => SerialT m a -> m ()
1006-- | Run a stream, discarding the results. By default it interprets the stream
1007-- as 'SerialT', to run other types of streams use the type adapting
1008-- combinators for example @runStream . 'asyncly'@.
1011{-# DEPRECATED runStream "Please use \"drain\" instead" #-}
1012{-# INLINE runStream #-}
1013runStream :: Monad m => SerialT m a -> m ()
1017-- > drainN n = drain . take n
1019-- Run maximum up to @n@ iterations of a stream.
1022{-# INLINE drainN #-}
1023drainN :: Monad m => Int -> SerialT m a -> m ()
1024drainN n = drain . take n
1027-- > runN n = runStream . take n
1029-- Run maximum up to @n@ iterations of a stream.
1032{-# DEPRECATED runN "Please use \"drainN\" instead" #-}
1034runN :: Monad m => Int -> SerialT m a -> m ()
1038-- > drainWhile p = drain . takeWhile p
1040-- Run a stream as long as the predicate holds true.
1043{-# INLINE drainWhile #-}
1044drainWhile :: Monad m => (a -> Bool) -> SerialT m a -> m ()
1045drainWhile p = drain . takeWhile p
1048-- > runWhile p = runStream . takeWhile p
1050-- Run a stream as long as the predicate holds true.
1053{-# DEPRECATED runWhile "Please use \"drainWhile\" instead" #-}
1054{-# INLINE runWhile #-}
1055runWhile :: Monad m => (a -> Bool) -> SerialT m a -> m ()
1056runWhile = drainWhile
1058-- | Determine whether the stream is empty.
1062null :: Monad m => SerialT m a -> m Bool
1063null = S.null . toStreamS
1065-- | Extract the first element of the stream, if any.
1071head :: Monad m => SerialT m a -> m (Maybe a)
1072head = S.head . toStreamS
1075-- > tail = fmap (fmap snd) . uncons
1077-- Extract all but the first element of the stream, if any.
1081tail :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (t m a))
1082tail m = K.tail (K.adapt m)
1084-- | Extract all but the last element of the stream, if any.
1088init :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (t m a))
1089init m = K.init (K.adapt m)
1091-- | Extract the last element of the stream, if any.
1093-- > last xs = xs !! (length xs - 1)
1097last :: Monad m => SerialT m a -> m (Maybe a)
1098last m = S.last $ toStreamS m
1100-- | Determine whether an element is present in the stream.
1104elem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool
1105elem e m = S.elem e (toStreamS m)
1107-- | Determine whether an element is not present in the stream.
1110{-# INLINE notElem #-}
1111notElem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool
1112notElem e m = S.notElem e (toStreamS m)
1114-- | Determine the length of the stream.
1117{-# INLINE length #-}
1118length :: Monad m => SerialT m a -> m Int
1119length = foldl' (\n _ -> n + 1) 0
1121-- | Determine whether all elements of a stream satisfy a predicate.
1125all :: Monad m => (a -> Bool) -> SerialT m a -> m Bool
1126all p m = S.all p (toStreamS m)
1128-- | Determine whether any of the elements of a stream satisfy a predicate.
1132any :: Monad m => (a -> Bool) -> SerialT m a -> m Bool
1133any p m = S.any p (toStreamS m)
1135-- | Determines if all elements of a boolean stream are True.
1139and :: Monad m => SerialT m Bool -> m Bool
1142-- | Determines whether at least one element of a boolean stream is True.
1146or :: Monad m => SerialT m Bool -> m Bool
1149-- | Determine the sum of all elements of a stream of numbers. Returns @0@ when
1150-- the stream is empty. Note that this is not numerically stable for floating
1155sum :: (Monad m, Num a) => SerialT m a -> m a
1158-- | Determine the product of all elements of a stream of numbers. Returns @1@
1159-- when the stream is empty.
1162{-# INLINE product #-}
1163product :: (Monad m, Num a) => SerialT m a -> m a
1164product = foldl' (*) 1
1168-- minimum = 'minimumBy' compare
1171-- Determine the minimum element in a stream.
1174{-# INLINE minimum #-}
1175minimum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)
1176minimum m = S.minimum (toStreamS m)
1178-- | Determine the minimum element in a stream using the supplied comparison
1182{-# INLINE minimumBy #-}
1183minimumBy :: Monad m => (a -> a -> Ordering) -> SerialT m a -> m (Maybe a)
1184minimumBy cmp m = S.minimumBy cmp (toStreamS m)
1188-- maximum = 'maximumBy' compare
1191-- Determine the maximum element in a stream.
1194{-# INLINE maximum #-}
1195maximum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)
1196maximum m = S.maximum (toStreamS m)
1198-- | Determine the maximum element in a stream using the supplied comparison
1202{-# INLINE maximumBy #-}
1203maximumBy :: Monad m => (a -> a -> Ordering) -> SerialT m a -> m (Maybe a)
1204maximumBy cmp m = S.maximumBy cmp (toStreamS m)
1206-- | Lookup the element at the given index.
1210(!!) :: Monad m => SerialT m a -> Int -> m (Maybe a)
1211m !! i = toStreamS m S.!! i
1213-- | In a stream of (key-value) pairs @(a, b)@, return the value @b@ of the
1214-- first pair where the key equals the given value @a@.
1216-- > lookup = snd <$> find ((==) . fst)
1219{-# INLINE lookup #-}
1220lookup :: (Monad m, Eq a) => a -> SerialT m (a, b) -> m (Maybe b)
1221lookup a m = S.lookup a (toStreamS m)
1223-- | Like 'findM' but with a non-monadic predicate.
1225-- > find p = findM (return . p)
1229find :: Monad m => (a -> Bool) -> SerialT m a -> m (Maybe a)
1230find p m = S.find p (toStreamS m)
1232-- | Returns the first element that satisfies the given predicate.
1236findM :: Monad m => (a -> m Bool) -> SerialT m a -> m (Maybe a)
1237findM p m = S.findM p (toStreamS m)
1239-- | Find all the indices where the element in the stream satisfies the given
1243{-# INLINE findIndices #-}
1244findIndices :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m Int
1245findIndices p m = fromStreamS $ S.findIndices p (toStreamS m)
1247-- | Returns the first index that satisfies the given predicate.
1250{-# INLINE findIndex #-}
1251findIndex :: Monad m => (a -> Bool) -> SerialT m a -> m (Maybe Int)
1252findIndex p = head . findIndices p
1254-- | Find all the indices where the value of the element in the stream is equal
1255-- to the given value.
1258{-# INLINE elemIndices #-}
1259elemIndices :: (IsStream t, Eq a, Monad m) => a -> t m a -> t m Int
1260elemIndices a = findIndices (==a)
1262-- | Returns the first index where a given value is found in the stream.
1264-- > elemIndex a = findIndex (== a)
1267{-# INLINE elemIndex #-}
1268elemIndex :: (Monad m, Eq a) => a -> SerialT m a -> m (Maybe Int)
1269elemIndex a = findIndex (== a)
1271------------------------------------------------------------------------------
1273------------------------------------------------------------------------------
1275-- | Returns 'True' if the first stream is the same as or a prefix of the
1276-- second. A stream is a prefix of itself.
1279-- > S.isPrefixOf (S.fromList "hello") (S.fromList "hello" :: SerialT IO Char)
1284{-# INLINE isPrefixOf #-}
1285isPrefixOf :: (Eq a, IsStream t, Monad m) => t m a -> t m a -> m Bool
1286isPrefixOf m1 m2 = D.isPrefixOf (toStreamD m1) (toStreamD m2)
1288-- | Returns 'True' if all the elements of the first stream occur, in order, in
1289-- the second stream. The elements do not have to occur consecutively. A stream
1290-- is a subsequence of itself.
1293-- > S.isSubsequenceOf (S.fromList "hlo") (S.fromList "hello" :: SerialT IO Char)
1298{-# INLINE isSubsequenceOf #-}
1299isSubsequenceOf :: (Eq a, IsStream t, Monad m) => t m a -> t m a -> m Bool
1300isSubsequenceOf m1 m2 = D.isSubsequenceOf (toStreamD m1) (toStreamD m2)
1302-- | Drops the given prefix from a stream. Returns 'Nothing' if the stream does
1303-- not start with the given prefix. Returns @Just nil@ when the prefix is the
1304-- same as the stream.
1307{-# INLINE stripPrefix #-}
1309 :: (Eq a, IsStream t, Monad m)
1310 => t m a -> t m a -> m (Maybe (t m a))
1311stripPrefix m1 m2 = fmap fromStreamD <$>
1312 D.stripPrefix (toStreamD m1) (toStreamD m2)
1314------------------------------------------------------------------------------
1316------------------------------------------------------------------------------
1318-- XXX this can utilize parallel mapping if we implement it as drain . mapM
1320-- > mapM_ = drain . mapM
1322-- Apply a monadic action to each element of the stream and discard the output
1323-- of the action. This is not really a pure transformation operation but a
1324-- transformation followed by fold.
1328mapM_ :: Monad m => (a -> m b) -> SerialT m a -> m ()
1329mapM_ f m = S.mapM_ f $ toStreamS m
1331------------------------------------------------------------------------------
1333------------------------------------------------------------------------------
1337-- toList = S.foldr (:) []
1340-- Convert a stream into a list in the underlying monad. The list can be
1341-- consumed lazily in a lazy monad (e.g. 'Identity'). In a strict monad (e.g.
1342-- IO) the whole list is generated and buffered before it can be consumed.
1344-- /Warning!/ working on large lists accumulated as buffers in memory could be
1345-- very inefficient, consider using "Streamly.Array" instead.
1348{-# INLINE toList #-}
1349toList :: Monad m => SerialT m a -> m [a]
1354-- toListRev = S.foldl' (flip (:)) []
1357-- Convert a stream into a list in reverse order in the underlying monad.
1359-- /Warning!/ working on large lists accumulated as buffers in memory could be
1360-- very inefficient, consider using "Streamly.Array" instead.
1363{-# INLINE toListRev #-}
1364toListRev :: Monad m => SerialT m a -> m [a]
1365toListRev = D.toListRev . toStreamD
1369-- toHandle h = S.mapM_ $ hPutStrLn h
1372-- Write a stream of Strings to an IO Handle.
1375{-# DEPRECATED toHandle
1376 "Please use Streamly.FileSystem.Handle module (see the changelog)" #-}
1377toHandle :: MonadIO m => IO.Handle -> SerialT m String -> m ()
1381 let stop = return ()
1382 single a = liftIO (IO.hPutStrLn h a)
1383 yieldk a r = liftIO (IO.hPutStrLn h a) >> go r
1384 in K.foldStream defState yieldk single stop m1
1386-- | A fold that buffers its input to a pure stream.
1388-- /Warning!/ working on large streams accumulated as buffers in memory could
1389-- be very inefficient, consider using "Streamly.Array" instead.
1392{-# INLINE toStream #-}
1393toStream :: Monad m => Fold m a (SerialT Identity a)
1394toStream = Fold (\f x -> return $ f . (x `K.cons`))
1396 (return . ($ K.nil))
1398-- This is more efficient than 'toStream'. toStream is exactly the same as
1399-- reversing the stream after toStreamRev.
1401-- | Buffers the input stream to a pure stream in the reverse order of the
1404-- /Warning!/ working on large streams accumulated as buffers in memory could
1405-- be very inefficient, consider using "Streamly.Array" instead.
1409-- xn : ... : x2 : x1 : []
1410{-# INLINABLE toStreamRev #-}
1411toStreamRev :: Monad m => Fold m a (SerialT Identity a)
1412toStreamRev = Fold (\xs x -> return $ x `K.cons` xs) (return K.nil) return
1414------------------------------------------------------------------------------
1415-- General Transformation
1416------------------------------------------------------------------------------
1418-- | Use a 'Pipe' to transform a stream.
1419{-# INLINE transform #-}
1420transform :: (IsStream t, Monad m) => Pipe m a b -> t m a -> t m b
1421transform pipe xs = fromStreamD $ D.transform pipe (toStreamD xs)
1423------------------------------------------------------------------------------
1424-- Transformation by Folding (Scans)
1425------------------------------------------------------------------------------
1427-- XXX It may be useful to have a version of scan where we can keep the
1428-- accumulator independent of the value emitted. So that we do not necessarily
1429-- have to keep a value in the accumulator which we are not using. We can pass
1430-- an extraction function that will take the accumulator and the current value
1431-- of the element and emit the next value in the stream. That will also make it
1432-- possible to modify the accumulator after using it. In fact, the step function
1433-- can return new accumulator and the value to be emitted. The signature would
1434-- be more like mapAccumL. Or we can change the signature of scanx to
1437-- | Strict left scan with an extraction function. Like 'scanl'', but applies a
1438-- user supplied extraction function (the third argument) at each step. This is
1439-- designed to work with the @foldl@ library. The suffix @x@ is a mnemonic for
1442-- /Since: 0.7.0 (Monad m constraint)/
1445{-# DEPRECATED scanx "Please use scanl followed by map instead." #-}
1447scanx :: (IsStream t, Monad m) => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
1450-- XXX this needs to be concurrent
1451-- | Like 'scanl'' but with a monadic fold function.
1454{-# INLINE scanlM' #-}
1455scanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> t m b
1456scanlM' step begin m = fromStreamD $ D.scanlM' step begin $ toStreamD m
1458-- | Strict left scan. Like 'map', 'scanl'' too is a one to one transformation,
1459-- however it adds an extra element.
1462-- > S.toList $ S.scanl' (+) 0 $ fromList [1,2,3,4]
1467-- > S.toList $ S.scanl' (flip (:)) [] $ S.fromList [1,2,3,4]
1468-- [[],[1],[2,1],[3,2,1],[4,3,2,1]]
1471-- The output of 'scanl'' is the initial value of the accumulator followed by
1472-- all the intermediate steps and the final result of 'foldl''.
1474-- By streaming the accumulated state after each fold step, we can share the
1475-- state across multiple stages of stream composition. Each stage can modify or
1476-- extend the state, do some processing with it and emit it for the next stage,
1477-- thus modularizing the stream processing. This can be useful in
1478-- stateful or event-driven programming.
1480-- Consider the following monolithic example, computing the sum and the product
1481-- of the elements in a stream in one go using a @foldl'@:
1484-- > S.foldl' (\\(s, p) x -> (s + x, p * x)) (0,1) $ S.fromList \[1,2,3,4]
1488-- Using @scanl'@ we can make it modular by computing the sum in the first
1489-- stage and passing it down to the next stage for computing the product:
1492-- > S.foldl' (\\(_, p) (s, x) -> (s, p * x)) (0,1)
1493-- $ S.scanl' (\\(s, _) x -> (s + x, x)) (0,1)
1494-- $ S.fromList \[1,2,3,4]
1498-- IMPORTANT: 'scanl'' evaluates the accumulator to WHNF. To avoid building
1499-- lazy expressions inside the accumulator, it is recommended that a strict
1500-- data structure is used for accumulator.
1503{-# INLINE scanl' #-}
1504scanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
1505scanl' step z m = fromStreamS $ S.scanl' step z $ toStreamS m
1507-- | Like 'scanl'' but does not stream the initial value of the accumulator.
1509-- > postscanl' f z xs = S.drop 1 $ S.scanl' f z xs
1512{-# INLINE postscanl' #-}
1513postscanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
1514postscanl' step z m = fromStreamD $ D.postscanl' step z $ toStreamD m
1516-- XXX this needs to be concurrent
1517-- | Like 'postscanl'' but with a monadic step function.
1520{-# INLINE postscanlM' #-}
1521postscanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> t m b
1522postscanlM' step z m = fromStreamD $ D.postscanlM' step z $ toStreamD m
1524-- XXX prescanl does not sound very useful, enable only if there is a
1525-- compelling use case.
1527-- | Like scanl' but does not stream the final value of the accumulator.
1530{-# INLINE prescanl' #-}
1531prescanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
1532prescanl' step z m = fromStreamD $ D.prescanl' step z $ toStreamD m
1534-- XXX this needs to be concurrent
1535-- | Like postscanl' but with a monadic step function.
1538{-# INLINE prescanlM' #-}
1539prescanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
1540prescanlM' step z m = fromStreamD $ D.prescanlM' step z $ toStreamD m
1542-- XXX this needs to be concurrent
1543-- | Like 'scanl1'' but with a monadic step function.
1546{-# INLINE scanl1M' #-}
1547scanl1M' :: (IsStream t, Monad m) => (a -> a -> m a) -> t m a -> t m a
1548scanl1M' step m = fromStreamD $ D.scanl1M' step $ toStreamD m
1550-- | Like 'scanl'' but for a non-empty stream. The first element of the stream
1551-- is used as the initial value of the accumulator. Does nothing if the stream
1555-- > S.toList $ S.scanl1 (+) $ fromList [1,2,3,4]
1560{-# INLINE scanl1' #-}
1561scanl1' :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> t m a
1562scanl1' step m = fromStreamD $ D.scanl1' step $ toStreamD m
1564------------------------------------------------------------------------------
1565-- Scanning with a Fold
1566------------------------------------------------------------------------------
1568-- | Scan a stream using the given monadic fold.
1572scan :: Monad m => Fold m a b -> SerialT m a -> SerialT m b
1573scan (Fold step begin done) = P.scanlMx' step begin done
1575-- | Postscan a stream using the given monadic fold.
1578{-# INLINE postscan #-}
1579postscan :: Monad m => Fold m a b -> SerialT m a -> SerialT m b
1580postscan (Fold step begin done) = P.postscanlMx' step begin done
1584------------------------------------------------------------------------------
1585-- Transformation by Filtering
1586------------------------------------------------------------------------------
1588-- | Include only those elements that pass a predicate.
1591{-# INLINE filter #-}
1592#if __GLASGOW_HASKELL__ != 802
1593-- GHC 8.2.2 crashes with this code, when used with "stack"
1594filter :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
1595filter p m = fromStreamS $ S.filter p $ toStreamS m
1597filter :: IsStream t => (a -> Bool) -> t m a -> t m a
1601-- | Same as 'filter' but with a monadic predicate.
1604{-# INLINE filterM #-}
1605filterM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
1606filterM p m = fromStreamD $ D.filterM p $ toStreamD m
1608-- | Drop repeated elements that are adjacent to each other.
1612uniq :: (Eq a, IsStream t, Monad m) => t m a -> t m a
1613uniq = fromStreamD . D.uniq . toStreamD
1615-- | Ensures that all the elements of the stream are identical and then returns
1616-- that unique element.
1620the :: (Eq a, Monad m) => SerialT m a -> m (Maybe a)
1621the m = S.the (toStreamS m)
1623-- | Take first 'n' elements from the stream and discard the rest.
1627take :: (IsStream t, Monad m) => Int -> t m a -> t m a
1628take n m = fromStreamS $ S.take n $ toStreamS
1629 (maxYields (Just (fromIntegral n)) m)
1631-- | End the stream as soon as the predicate fails on an element.
1634{-# INLINE takeWhile #-}
1635takeWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
1636takeWhile p m = fromStreamS $ S.takeWhile p $ toStreamS m
1638-- | Same as 'takeWhile' but with a monadic predicate.
1641{-# INLINE takeWhileM #-}
1642takeWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
1643takeWhileM p m = fromStreamD $ D.takeWhileM p $ toStreamD m
1645-- | Discard first 'n' elements from the stream and take the rest.
1649drop :: (IsStream t, Monad m) => Int -> t m a -> t m a
1650drop n m = fromStreamS $ S.drop n $ toStreamS m
1652-- | Drop elements in the stream as long as the predicate succeeds and then
1653-- take the rest of the stream.
1656{-# INLINE dropWhile #-}
1657dropWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
1658dropWhile p m = fromStreamS $ S.dropWhile p $ toStreamS m
1660-- | Same as 'dropWhile' but with a monadic predicate.
1663{-# INLINE dropWhileM #-}
1664dropWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
1665dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m
1667------------------------------------------------------------------------------
1668-- Transformation by Mapping
1669------------------------------------------------------------------------------
1673-- mapM f = sequence . map f
1676-- Apply a monadic function to each element of the stream and replace it with
1677-- the output of the resulting action.
1680-- > drain $ S.mapM putStr $ S.fromList ["a", "b", "c"]
1683-- drain $ S.replicateM 10 (return 1)
1684-- & (serially . S.mapM (\\x -> threadDelay 1000000 >> print x))
1686-- drain $ S.replicateM 10 (return 1)
1687-- & (asyncly . S.mapM (\\x -> threadDelay 1000000 >> print x))
1690-- /Concurrent (do not use with 'parallely' on infinite streams)/
1693{-# INLINE_EARLY mapM #-}
1694mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
1697{-# RULES "mapM serial" mapM = mapMSerial #-}
1698{-# INLINE mapMSerial #-}
1699mapMSerial :: Monad m => (a -> m b) -> SerialT m a -> SerialT m b
1700mapMSerial = Serial.mapM
1704-- sequence = mapM id
1707-- Replace the elements of a stream of monadic actions with the outputs of
1711-- > drain $ S.sequence $ S.fromList [putStr "a", putStr "b", putStrLn "c"]
1714-- drain $ S.replicateM 10 (return $ threadDelay 1000000 >> print 1)
1715-- & (serially . S.sequence)
1717-- drain $ S.replicateM 10 (return $ threadDelay 1000000 >> print 1)
1718-- & (asyncly . S.sequence)
1721-- /Concurrent (do not use with 'parallely' on infinite streams)/
1724{-# INLINE sequence #-}
1725sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
1726sequence m = fromStreamS $ S.sequence (toStreamS m)
1728------------------------------------------------------------------------------
1729-- Transformation by Map and Filter
1730------------------------------------------------------------------------------
1732-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'
1733-- elements, and return a stream of values extracted from 'Just'.
1738-- mapMaybe f = S.map 'fromJust' . S.filter 'isJust' . S.map f
1742{-# INLINE mapMaybe #-}
1743mapMaybe :: (IsStream t, Monad m) => (a -> Maybe b) -> t m a -> t m b
1744mapMaybe f m = fromStreamS $ S.mapMaybe f $ toStreamS m
1746-- | Like 'mapMaybe' but maps a monadic function.
1751-- mapMaybeM f = S.map 'fromJust' . S.filter 'isJust' . S.mapM f
1754-- /Concurrent (do not use with 'parallely' on infinite streams)/
1757{-# INLINE_EARLY mapMaybeM #-}
1758mapMaybeM :: (IsStream t, MonadAsync m, Functor (t m))
1759 => (a -> m (Maybe b)) -> t m a -> t m b
1760mapMaybeM f = fmap fromJust . filter isJust . K.mapM f
1762{-# RULES "mapMaybeM serial" mapMaybeM = mapMaybeMSerial #-}
1763{-# INLINE mapMaybeMSerial #-}
1764mapMaybeMSerial :: Monad m => (a -> m (Maybe b)) -> SerialT m a -> SerialT m b
1765mapMaybeMSerial f m = fromStreamD $ D.mapMaybeM f $ toStreamD m
1767------------------------------------------------------------------------------
1768-- Transformation by Reordering
1769------------------------------------------------------------------------------
1771-- XXX Use a compact region list to temporarily store the list, in both reverse
1772-- as well as in reverse'.
1774-- /Note:/ 'reverse'' is much faster than this, use that when performance
1777-- > reverse = S.foldlT (flip S.cons) S.nil
1779-- | Returns the elements of the stream in reverse order. The stream must be
1780-- finite. Note that this necessarily buffers the entire stream in memory.
1782-- /Since 0.7.0 (Monad m constraint)/
1785{-# INLINE reverse #-}
1786reverse :: (IsStream t, Monad m) => t m a -> t m a
1787reverse s = fromStreamS $ S.reverse $ toStreamS s
1789-- | Like 'reverse' but several times faster, requires a 'Storable' instance.
1792{-# INLINE reverse' #-}
1793reverse' :: (IsStream t, MonadIO m, Storable a) => t m a -> t m a
1794reverse' s = fromStreamD $ D.reverse' $ toStreamD s
1796------------------------------------------------------------------------------
1797-- Transformation by Inserting
1798------------------------------------------------------------------------------
1800-- intersperseM = intersperseBySpan 1
1802-- | Generate a stream by performing a monadic action between consecutive
1803-- elements of the given stream.
1805-- /Concurrent (do not use with 'parallely' on infinite streams)/
1808-- > S.toList $ S.intersperseM (return ',') $ S.fromList "hello"
1813{-# INLINE intersperseM #-}
1814intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
1815intersperseM m = fromStreamS . S.intersperseM m . toStreamS
1817-- | Generate a stream by inserting a given element between consecutive
1818-- elements of the given stream.
1821-- > S.toList $ S.intersperse ',' $ S.fromList "hello"
1826{-# INLINE intersperse #-}
1827intersperse :: (IsStream t, MonadAsync m) => a -> t m a -> t m a
1828intersperse a = fromStreamS . S.intersperse a . toStreamS
1830-- | Insert a monadic action after each element in the stream.
1833{-# INLINE insertAfterEach #-}
1834insertAfterEach :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
1835insertAfterEach m = fromStreamD . D.insertAfterEach m . toStreamD
1838-- | Intersperse a monadic action into the input stream after every @n@
1842-- > S.toList $ S.intersperseBySpan 2 (return ',') $ S.fromList "hello"
1847{-# INLINE intersperseBySpan #-}
1848intersperseBySpan :: IsStream t => Int -> m a -> t m a -> t m a
1849intersperseBySpan _n _f _xs = undefined
1852-- | Intersperse a monadic action into the input stream after every @n@
1856-- > S.drain $ S.interject 1 (putChar ',') $ S.mapM (\\x -> threadDelay 1000000 >> putChar x) $ S.fromList "hello"
1861{-# INLINE interject #-}
1863 :: (IsStream t, MonadAsync m)
1864 => Double -> m a -> t m a -> t m a
1865interject n f xs = xs `Par.parallelFst` repeatM timed
1866 where timed = liftIO (threadDelay (round $ n * 1000000)) >> f
1868-- | @insertBy cmp elem stream@ inserts @elem@ before the first element in
1869-- @stream@ that is less than @elem@ when compared using @cmp@.
1872-- insertBy cmp x = 'mergeBy' cmp ('yield' x)
1876-- > S.toList $ S.insertBy compare 2 $ S.fromList [1,3,5]
1881{-# INLINE insertBy #-}
1883 (IsStream t, Monad m) => (a -> a -> Ordering) -> a -> t m a -> t m a
1884insertBy cmp x m = fromStreamS $ S.insertBy cmp x (toStreamS m)
1886------------------------------------------------------------------------------
1888------------------------------------------------------------------------------
1890-- | Deletes the first occurence of the element in the stream that satisfies
1891-- the given equality predicate.
1894-- > S.toList $ S.deleteBy (==) 3 $ S.fromList [1,3,3,5]
1899{-# INLINE deleteBy #-}
1900deleteBy :: (IsStream t, Monad m) => (a -> a -> Bool) -> a -> t m a -> t m a
1901deleteBy cmp x m = fromStreamS $ S.deleteBy cmp x (toStreamS m)
1903------------------------------------------------------------------------------
1905------------------------------------------------------------------------------
1908-- > indexed = S.postscanl' (\(i, _) x -> (i + 1, x)) (-1,undefined)
1909-- > indexed = S.zipWith (,) (S.enumerateFrom 0)
1911-- Pair each element in a stream with its index, starting from index 0.
1914-- > S.toList $ S.indexed $ S.fromList "hello"
1915-- [(0,'h'),(1,'e'),(2,'l'),(3,'l'),(4,'o')]
1919{-# INLINE indexed #-}
1920indexed :: (IsStream t, Monad m) => t m a -> t m (Int, a)
1921indexed = fromStreamD . D.indexed . toStreamD
1924-- > indexedR n = S.postscanl' (\(i, _) x -> (i - 1, x)) (n + 1,undefined)
1925-- > indexedR n = S.zipWith (,) (S.enumerateFromThen n (n - 1))
1927-- Pair each element in a stream with its index, starting from the
1928-- given index @n@ and counting down.
1931-- > S.toList $ S.indexedR 10 $ S.fromList "hello"
1932-- [(10,'h'),(9,'e'),(8,'l'),(7,'l'),(6,'o')]
1936{-# INLINE indexedR #-}
1937indexedR :: (IsStream t, Monad m) => Int -> t m a -> t m (Int, a)
1938indexedR n = fromStreamD . D.indexedR n . toStreamD
1940-- | Like 'zipWith' but using a monadic zipping function.
1943{-# INLINABLE zipWithM #-}
1944zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c
1945zipWithM f m1 m2 = fromStreamS $ S.zipWithM f (toStreamS m1) (toStreamS m2)
1947-- | Zip two streams serially using a pure zipping function.
1950-- > S.toList $ S.zipWith (+) (S.fromList [1,2,3]) (S.fromList [4,5,6])
1955{-# INLINABLE zipWith #-}
1956zipWith :: (IsStream t, Monad m) => (a -> b -> c) -> t m a -> t m b -> t m c
1957zipWith f m1 m2 = fromStreamS $ S.zipWith f (toStreamS m1) (toStreamS m2)
1959------------------------------------------------------------------------------
1961------------------------------------------------------------------------------
1963-- | Compare two streams for equality using an equality function.
1966{-# INLINABLE eqBy #-}
1967eqBy :: (IsStream t, Monad m) => (a -> b -> Bool) -> t m a -> t m b -> m Bool
1970-- | Compare two streams lexicographically using a comparison function.
1973{-# INLINABLE cmpBy #-}
1975 :: (IsStream t, Monad m)
1976 => (a -> b -> Ordering) -> t m a -> t m b -> m Ordering
1979------------------------------------------------------------------------------
1981------------------------------------------------------------------------------
1983-- | Merge two streams using a comparison function. The head elements of both
1984-- the streams are compared and the smaller of the two elements is emitted, if
1985-- both elements are equal then the element from the first stream is used
1988-- If the streams are sorted in ascending order, the resulting stream would
1989-- also remain sorted in ascending order.
1992-- > S.toList $ S.mergeBy compare (S.fromList [1,3,5]) (S.fromList [2,4,6,8])
1997{-# INLINABLE mergeBy #-}
1999 (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> t m a -> t m a
2000mergeBy f m1 m2 = fromStreamS $ S.mergeBy f (toStreamS m1) (toStreamS m2)
2002-- | Like 'mergeBy' but with a monadic comparison function.
2004-- Merge two streams randomly:
2007-- > randomly _ _ = randomIO >>= \x -> return $ if x then LT else GT
2008-- > S.toList $ S.mergeByM randomly (S.fromList [1,1,1,1]) (S.fromList [2,2,2,2])
2012-- Merge two streams in a proportion of 2:1:
2015-- proportionately m n = do
2016-- ref <- newIORef $ cycle $ concat [replicate m LT, replicate n GT]
2017-- return $ \\_ _ -> do
2018-- r <- readIORef ref
2019-- writeIORef ref $ tail r
2023-- f <- proportionately 2 1
2024-- xs <- S.toList $ S.mergeByM f (S.fromList [1,1,1,1,1,1]) (S.fromList [2,2,2])
2028-- [1,1,2,1,1,2,1,1,2]
2032{-# INLINABLE mergeByM #-}
2034 :: (IsStream t, Monad m)
2035 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
2036mergeByM f m1 m2 = fromStreamS $ S.mergeByM f (toStreamS m1) (toStreamS m2)
2039-- | Like 'mergeByM' but stops merging as soon as any of the two streams stops.
2040{-# INLINABLE mergeEndByAny #-}
2042 :: (IsStream t, Monad m)
2043 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
2044mergeEndByAny f m1 m2 = fromStreamD $
2045 D.mergeEndByAny f (toStreamD m1) (toStreamD m2)
2047-- Like 'mergeByM' but stops merging as soon as the first stream stops.
2048{-# INLINABLE mergeEndByFirst #-}
2050 :: (IsStream t, Monad m)
2051 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
2052mergeEndByFirst f m1 m2 = fromStreamS $
2053 D.mergeEndByFirst f (toStreamD m1) (toStreamD m2)
2056-- Holding this back for now, we may want to use the name "merge" differently
2058-- | Same as @'mergeBy' 'compare'@.
2061-- > S.toList $ S.merge (S.fromList [1,3,5]) (S.fromList [2,4,6,8])
2066{-# INLINABLE merge #-}
2068 (IsStream t, Monad m, Ord a) => t m a -> t m a -> t m a
2069merge = mergeBy compare
2072-- | Like 'mergeBy' but merges concurrently (i.e. both the elements being
2073-- merged are generated concurrently).
2076mergeAsyncBy :: (IsStream t, MonadAsync m)
2077 => (a -> a -> Ordering) -> t m a -> t m a -> t m a
2078mergeAsyncBy f m1 m2 = K.mkStream $ \st stp sng yld -> do
2079 ma <- mkAsync' st m1
2080 mb <- mkAsync' st m2
2081 K.foldStream st stp sng yld (K.mergeBy f ma mb)
2083-- | Like 'mergeByM' but merges concurrently (i.e. both the elements being
2084-- merged are generated concurrently).
2087mergeAsyncByM :: (IsStream t, MonadAsync m)
2088 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
2089mergeAsyncByM f m1 m2 = K.mkStream $ \st stp sng yld -> do
2090 ma <- mkAsync' st m1
2091 mb <- mkAsync' st m2
2092 K.foldStream st stp sng yld (K.mergeByM f ma mb)
2094------------------------------------------------------------------------------
2096------------------------------------------------------------------------------
2098-- | @concatMapWith merge map stream@ is a two dimensional looping combinator.
2099-- The first argument specifies a merge or concat function that is used to
2100-- merge the streams generated by applying the second argument i.e. the @map@
2101-- function to each element of the input stream. The concat function could be
2102-- 'serial', 'parallel', 'async', 'ahead' or any other zip or merge function
2103-- and the second argument could be any stream generation function using a
2106-- /Compare 'foldMapWith'/
2109{-# INLINE concatMapWith #-}
2112 => (forall c. t m c -> t m c -> t m c)
2116concatMapWith = K.concatMapBy
2118-- | Map a stream producing function on each element of the stream and then
2119-- flatten the results into a single stream.
2122-- concatMap = 'concatMapWith' 'Serial.serial'
2123-- concatMap f = 'concatMapM' (return . f)
2127{-# INLINE concatMap #-}
2128concatMap ::(IsStream t, Monad m) => (a -> t m b) -> t m a -> t m b
2129concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)
2131-- | Append the outputs of two streams, yielding all the elements from the
2132-- first stream and then yielding all the elements from the second stream.
2134-- IMPORTANT NOTE: This could be 100x faster than @serial/<>@ for appending a
2135-- few (say 100) streams because it can fuse via stream fusion. However, it
2136-- does not scale for a large number of streams (say 1000s) and becomes
2137-- qudartically slow. Therefore use this for custom appending of a few streams
2138-- but use 'concatMap' or 'concatMapWith serial' for appending @n@ streams or
2139-- infinite containers of streams.
2142{-# INLINE append #-}
2143append ::(IsStream t, Monad m) => t m b -> t m b -> t m b
2144append m1 m2 = fromStreamD $ D.append (toStreamD m1) (toStreamD m2)
2146-- Same as 'wSerial'. We should perhaps rename wSerial. If named explicitly
2147-- this would be interleaveMax.
2149-- | Interleaves the outputs of two streams, yielding elements from each stream
2150-- alternately, starting from the first stream. If any of the streams finishes
2151-- early the other stream continues alone until it too finishes.
2153-- Do not use at scale in concatMapWith.
2156{-# INLINE interleave #-}
2157interleave ::(IsStream t, Monad m) => t m b -> t m b -> t m b
2158interleave m1 m2 = fromStreamD $ D.interleave (toStreamD m1) (toStreamD m2)
2160-- | Interleaves the outputs of two streams, yielding elements from each stream
2161-- alternately, starting from the first stream. As soon as the first stream
2162-- finishes the output stops discarding the second stream.
2164-- Do not use at scale in concatMapWith.
2167{-# INLINE interleaveFst #-}
2168interleaveFst ::(IsStream t, Monad m) => t m b -> t m b -> t m b
2169interleaveFst m1 m2 = fromStreamD $ D.interleaveFst (toStreamD m1) (toStreamD m2)
2171-- | Interleaves the outputs of two streams, yielding elements from each stream
2172-- alternately, starting from the first stream. The output stops as soon as any
2173-- of the two streams finishes discarding the other.
2175-- Do not use at scale in concatMapWith.
2178{-# INLINE interleaveMin #-}
2179interleaveMin ::(IsStream t, Monad m) => t m b -> t m b -> t m b
2180interleaveMin m1 m2 = fromStreamD $ D.interleaveMin (toStreamD m1) (toStreamD m2)
2182-- | Schedule the execution of two streams in a fair round-robin manner,
2183-- executing each stream once, alternately. Execution of a stream may not
2184-- necessarily result in an output, a stream may chose to @Skip@ producing an
2185-- element until later giving the other stream a chance to run. Therefore, this
2186-- combinator fairly interleaves the execution of two streams rather than
2187-- fairly interleaving the output of the two streams. This can be useful in
2188-- co-operative multitasking without using explicit threads. This can be used
2189-- as an alternative to `async`.
2191-- Do not use at scale in concatMapWith.
2194{-# INLINE roundrobin #-}
2195roundrobin ::(IsStream t, Monad m) => t m b -> t m b -> t m b
2196roundrobin m1 m2 = fromStreamD $ D.roundRobin (toStreamD m1) (toStreamD m2)
2198-- | Map a stream producing monadic function on each element of the stream
2199-- and then flatten the results into a single stream. Since the stream
2200-- generation function is monadic, unlike 'concatMap', it can produce an
2201-- effect at the beginning of each iteration of the inner loop.
2204{-# INLINE concatMapM #-}
2205concatMapM :: (IsStream t, Monad m) => (a -> m (t m b)) -> t m a -> t m b
2206concatMapM f m = fromStreamD $ D.concatMapM (fmap toStreamD . f) (toStreamD m)
2208-- | Like 'concatMap' but uses an 'Unfold' for stream generation. Unlike
2209-- 'concatMap' this can fuse the 'Unfold' code with the inner loop and
2210-- therefore provide many times better performance.
2213{-# INLINE concatUnfold #-}
2214concatUnfold ::(IsStream t, Monad m) => Unfold m a b -> t m a -> t m b
2215concatUnfold u m = fromStreamD $ D.concatMapU u (toStreamD m)
2217-- | Like 'concatUnfold' but interleaves the streams in the same way as
2218-- 'interleave' behaves instead of appending them.
2221{-# INLINE concatUnfoldInterleave #-}
2222concatUnfoldInterleave ::(IsStream t, Monad m)
2223 => Unfold m a b -> t m a -> t m b
2224concatUnfoldInterleave u m =
2225 fromStreamD $ D.concatUnfoldInterleave u (toStreamD m)
2227-- | Like 'concatUnfold' but executes the streams in the same way as
2231{-# INLINE concatUnfoldRoundrobin #-}
2232concatUnfoldRoundrobin ::(IsStream t, Monad m)
2233 => Unfold m a b -> t m a -> t m b
2234concatUnfoldRoundrobin u m =
2235 fromStreamD $ D.concatUnfoldRoundrobin u (toStreamD m)
2238-- | Insert a stream between segements of streams and flatten.
2239intercalate :: (IsStream t, MonadAsync m)
2240 => SerialT Identity b -> (a -> t m b) -> t m a -> t m b
2241intercalate = undefined
2244-- | @intercalateSuffix genSuffix seed gen stream@ generates streams from seeds
2245-- in the input @stream@ using the generator @gen@ and concatenates them
2246-- after suffixing the stream generated by @genSuffix@ using @seed@.
2248-- For example to insert CRLF in a stream of character strings:
2250-- > unlines = intercalateSuffix UF.fromList "\r\n" UF.fromList
2252{-# INLINE intercalateSuffix #-}
2254 :: (IsStream t, Monad m)
2255 => Unfold m a c -> a -> Unfold m b c -> t m b -> t m c
2256intercalateSuffix suffix seed unf str =
2257 D.fromStreamD $ D.interleaveFstThenConcat
2258 unf (D.toStreamD str)
2259 suffix (D.repeat seed)
2261------------------------------------------------------------------------------
2262-- Grouping/Splitting
2263------------------------------------------------------------------------------
2265------------------------------------------------------------------------------
2266-- Grouping without looking at elements
2267------------------------------------------------------------------------------
2269------------------------------------------------------------------------------
2271------------------------------------------------------------------------------
2274-- | @splitAt n f1 f2@ composes folds @f1@ and @f2@ such that first @n@
2275-- elements of its input are consumed by fold @f1@ and the rest of the stream
2276-- is consumed by fold @f2@.
2278-- > let splitAt_ n xs = S.fold (FL.splitAt n FL.toList FL.toList) $ S.fromList xs
2280-- >>> splitAt_ 6 "Hello World!"
2281-- > ("Hello ","World!")
2283-- >>> splitAt_ (-1) [1,2,3]
2286-- >>> splitAt_ 0 [1,2,3]
2289-- >>> splitAt_ 1 [1,2,3]
2292-- >>> splitAt_ 3 [1,2,3]
2295-- >>> splitAt_ 4 [1,2,3]
2300-- This can be considered as a two-fold version of 'ltake' where we take both
2301-- the segments instead of discarding the leftover.
2303{-# INLINE splitAt #-}
2310splitAt n (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
2311 Fold step initial extract
2313 initial = Tuple3' <$> return n <*> initialL <*> initialR
2315 step (Tuple3' i xL xR) input =
2317 then stepL xL input >>= (\a -> return (Tuple3' (i - 1) a xR))
2318 else stepR xR input >>= (\b -> return (Tuple3' i xL b))
2320 extract (Tuple3' _ a b) = (,) <$> extractL a <*> extractR b
2322------------------------------------------------------------------------------
2324------------------------------------------------------------------------------
2326------------------------------------------------------------------------------
2327-- Generalized grouping
2328------------------------------------------------------------------------------
2330-- This combinator is the most general grouping combinator and can be used to
2331-- implement all other grouping combinators.
2333-- XXX check if this can implement the splitOn combinator i.e. we can slide in
2334-- new elements, slide out old elements and incrementally compute the hash.
2335-- Also, can we implement the windowed classification combinators using this?
2337-- In fact this is a parse. Instead of using a special return value in the fold
2338-- we are using a mapping function.
2340-- Note that 'scanl'' (usually followed by a map to extract the desired value
2341-- from the accumulator) can be used to realize many implementations e.g. a
2342-- sliding window implementation. A scan followed by a mapMaybe is also a good
2343-- pattern to express many problems where we want to emit a filtered output and
2344-- not emit an output on every input.
2346-- Passing on of the initial accumulator value to the next fold is equivalent
2347-- to returning the leftover concept.
2350-- | @groupScan splitter fold stream@ folds the input stream using @fold@.
2351-- @splitter@ is applied on the accumulator of the fold every time an item is
2352-- consumed by the fold. The fold continues until @splitter@ returns a 'Just'
2353-- value. A 'Just' result from the @splitter@ specifies a result to be emitted
2354-- in the output stream and the initial value of the accumulator for the next
2355-- group's fold. This allows us to control whether to start fresh for the next
2356-- fold or to continue from the previous fold's output.
2358{-# INLINE groupScan #-}
2360 :: (IsStream t, Monad m)
2361 => (x -> m (Maybe (b, x))) -> Fold m a x -> t m a -> t m b
2362groupScan split fold m = undefined
2365-- | Group the input stream into groups of @n@ elements each and then fold each
2366-- group using the provided fold function.
2368-- >> S.toList $ S.chunksOf 2 FL.sum (S.enumerateFromTo 1 10)
2371-- This can be considered as an n-fold version of 'ltake' where we apply
2372-- 'ltake' repeatedly on the leftover stream until the stream exhausts.
2375{-# INLINE chunksOf #-}
2377 :: (IsStream t, Monad m)
2378 => Int -> Fold m a b -> t m a -> t m b
2379chunksOf n f m = D.fromStreamD $ D.groupsOf n f (D.toStreamD m)
2381-- | @arraysOf n stream@ groups the elements in the input stream into arrays of
2382-- @n@ elements each.
2384-- Same as the following but may be more efficient:
2386-- > arraysOf n = S.chunksOf n (A.writeN n)
2389{-# INLINE arraysOf #-}
2390arraysOf :: (IsStream t, MonadIO m, Storable a)
2391 => Int -> t m a -> t m (Array a)
2392arraysOf n = chunksOf n (writeNUnsafe n)
2394-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
2395-- XXX add this example after fixing the serial stream rate control
2396-- >>> S.toList $ S.take 5 $ intervalsOf 1 FL.sum $ constRate 2 $ S.enumerateFrom 1
2399-- | Group the input stream into windows of @n@ second each and then fold each
2400-- group using the provided fold function.
2403{-# INLINE intervalsOf #-}
2405 :: (IsStream t, MonadAsync m)
2406 => Double -> Fold m a b -> t m a -> t m b
2408 splitWithSuffix isNothing (FL.lcatMaybes f)
2409 (interject n (return Nothing) (Serial.map Just xs))
2411------------------------------------------------------------------------------
2412-- Element Aware APIs
2413------------------------------------------------------------------------------
2415------------------------------------------------------------------------------
2417------------------------------------------------------------------------------
2419-- | Break the input stream into two groups, the first group takes the input as
2420-- long as the predicate applied to the first element of the stream and next
2421-- input element holds 'True', the second group takes the rest of the input.
2429spanBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
2430 Fold step initial extract
2433 initial = Tuple3' <$> initialL <*> initialR <*> return (Tuple' Nothing True)
2435 step (Tuple3' a b (Tuple' (Just frst) isFirstG)) input =
2436 if cmp frst input && isFirstG
2438 >>= (\a' -> return (Tuple3' a' b (Tuple' (Just frst) isFirstG)))
2440 >>= (\a' -> return (Tuple3' a a' (Tuple' Nothing False)))
2442 step (Tuple3' a b (Tuple' Nothing isFirstG)) input =
2445 >>= (\a' -> return (Tuple3' a' b (Tuple' (Just input) isFirstG)))
2447 >>= (\a' -> return (Tuple3' a a' (Tuple' Nothing False)))
2449 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
2451-- | @span p f1 f2@ composes folds @f1@ and @f2@ such that @f1@ consumes the
2452-- input as long as the predicate @p@ is 'True'. @f2@ consumes the rest of the
2455-- > let span_ p xs = S.fold (S.span p FL.toList FL.toList) $ S.fromList xs
2457-- >>> span_ (< 1) [1,2,3]
2460-- >>> span_ (< 2) [1,2,3]
2463-- >>> span_ (< 4) [1,2,3]
2468-- This can be considered as a two-fold version of 'ltakeWhile' where we take
2469-- both the segments instead of discarding the leftover.
2477span p (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
2478 Fold step initial extract
2482 initial = Tuple3' <$> initialL <*> initialR <*> return True
2484 step (Tuple3' a b isFirstG) input =
2485 if isFirstG && p input
2486 then stepL a input >>= (\a' -> return (Tuple3' a' b True))
2487 else stepR b input >>= (\a' -> return (Tuple3' a a' False))
2489 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
2492-- > break p = span (not . p)
2494-- Break as soon as the predicate becomes 'True'. @break p f1 f2@ composes
2495-- folds @f1@ and @f2@ such that @f1@ stops consuming input as soon as the
2496-- predicate @p@ becomes 'True'. The rest of the input is consumed @f2@.
2498-- This is the binary version of 'splitBy'.
2500-- > let break_ p xs = S.fold (S.break p FL.toList FL.toList) $ S.fromList xs
2502-- >>> break_ (< 1) [3,2,1]
2505-- >>> break_ (< 2) [3,2,1]
2508-- >>> break_ (< 4) [3,2,1]
2519break p = span (not . p)
2521-- | Like 'spanBy' but applies the predicate in a rolling fashion i.e.
2522-- predicate is applied to the previous and the next input elements.
2523{-# INLINE spanByRolling #-}
2530spanByRolling cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
2531 Fold step initial extract
2534 initial = Tuple3' <$> initialL <*> initialR <*> return Nothing
2536 step (Tuple3' a b (Just frst)) input =
2538 then stepL a input >>= (\a' -> return (Tuple3' a' b (Just input)))
2539 else stepR b input >>= (\b' -> return (Tuple3' a b' (Just input)))
2541 step (Tuple3' a b Nothing) input =
2542 stepL a input >>= (\a' -> return (Tuple3' a' b (Just input)))
2544 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
2546------------------------------------------------------------------------------
2548------------------------------------------------------------------------------
2550-- | @groupsBy cmp f $ S.fromList [a,b,c,...]@ assigns the element @a@ to the
2551-- first group, if @a \`cmp` b@ is 'True' then @b@ is also assigned to the same
2552-- group. If @a \`cmp` c@ is 'True' then @c@ is also assigned to the same
2553-- group and so on. When the comparison fails a new group is started. Each
2554-- group is folded using the fold @f@ and the result of the fold is emitted in
2555-- the output stream.
2557-- >>> S.toList $ S.groupsBy (>) FL.toList $ S.fromList [1,3,7,0,2,5]
2558-- > [[1,3,7],[0,2,5]]
2561{-# INLINE groupsBy #-}
2563 :: (IsStream t, Monad m)
2568groupsBy cmp f m = D.fromStreamD $ D.groupsBy cmp f (D.toStreamD m)
2570-- | Unlike @groupsBy@ this function performs a rolling comparison of two
2571-- successive elements in the input stream. @groupsByRolling cmp f $ S.fromList
2572-- [a,b,c,...]@ assigns the element @a@ to the first group, if @a \`cmp` b@ is
2573-- 'True' then @b@ is also assigned to the same group. If @b \`cmp` c@ is
2574-- 'True' then @c@ is also assigned to the same group and so on. When the
2575-- comparison fails a new group is started. Each group is folded using the fold
2578-- >>> S.toList $ S.groupsByRolling (\a b -> a + 1 == b) FL.toList $ S.fromList [1,2,3,7,8,9]
2579-- > [[1,2,3],[7,8,9]]
2582{-# INLINE groupsByRolling #-}
2584 :: (IsStream t, Monad m)
2589groupsByRolling cmp f m = D.fromStreamD $ D.groupsRollingBy cmp f (D.toStreamD m)
2592-- > groups = groupsBy (==)
2593-- > groups = groupsByRolling (==)
2595-- Groups contiguous spans of equal elements together in individual groups.
2597-- >>> S.toList $ S.groups FL.toList $ S.fromList [1,1,2,2]
2601groups :: (IsStream t, Monad m, Eq a) => Fold m a b -> t m a -> t m b
2602groups = groupsBy (==)
2604------------------------------------------------------------------------------
2605-- Binary splitting on a separator
2606------------------------------------------------------------------------------
2609-- | Find the first occurrence of the specified sequence in the input stream
2610-- and break the input stream into two parts, the first part consisting of the
2611-- stream before the sequence and the second part consisting of the sequence
2612-- and the rest of the stream.
2614-- > let breakOn_ pat xs = S.fold (S.breakOn pat FL.toList FL.toList) $ S.fromList xs
2616-- >>> breakOn_ "dear" "Hello dear world!"
2617-- > ("Hello ","dear world!")
2619{-# INLINE breakOn #-}
2620breakOn :: Monad m => Array a -> Fold m a b -> Fold m a c -> Fold m a (b,c)
2621breakOn pat f m = undefined
2624------------------------------------------------------------------------------
2625-- N-ary split on a predicate
2626------------------------------------------------------------------------------
2628-- TODO: Use a Splitter configuration similar to the "split" package to make it
2629-- possible to express all splitting combinations. In general, we can have
2630-- infix/suffix/prefix/condensing of separators, dropping both leading/trailing
2631-- separators. We can have a single split operation taking the splitter config
2634-- | Split on an infixed separator element, dropping the separator. Splits the
2635-- stream on separator elements determined by the supplied predicate, separator
2636-- is considered as infixed between two segments, if one side of the separator
2637-- is missing then it is parsed as an empty stream. The supplied 'Fold' is
2638-- applied on the split segments. With '-' representing non-separator elements
2639-- and '.' as separator, 'splitOn' splits as follows:
2642-- "--.--" => "--" "--"
2647-- @splitOn (== x)@ is an inverse of @intercalate (S.yield x)@
2649-- Let's use the following definition for illustration:
2651-- > splitOn' p xs = S.toList $ S.splitOn p (FL.toList) (S.fromList xs)
2653-- >>> splitOn' (== '.') ""
2656-- >>> splitOn' (== '.') "."
2659-- >>> splitOn' (== '.') ".a"
2662-- >>> splitOn' (== '.') "a."
2665-- >>> splitOn' (== '.') "a.b"
2668-- >>> splitOn' (== '.') "a..b"
2673-- This can be considered as an n-fold version of 'breakOn' where we apply
2674-- 'breakOn' successively on the input stream, dropping the first element
2675-- of the second segment after each break.
2677{-# INLINE splitOn #-}
2679 :: (IsStream t, Monad m)
2680 => (a -> Bool) -> Fold m a b -> t m a -> t m b
2681splitOn predicate f m =
2682 D.fromStreamD $ D.splitBy predicate f (D.toStreamD m)
2684-- | Like 'splitOn' but the separator is considered as suffixed to the segments
2685-- in the stream. A missing suffix at the end is allowed. A separator at the
2686-- beginning is parsed as empty segment. With '-' representing elements and
2687-- '.' as separator, 'splitOnSuffix' splits as follows:
2690-- "--.--." => "--" "--"
2691-- "--.--" => "--" "--"
2695-- > splitOnSuffix' p xs = S.toList $ S.splitSuffixBy p (FL.toList) (S.fromList xs)
2697-- >>> splitOnSuffix' (== '.') ""
2700-- >>> splitOnSuffix' (== '.') "."
2703-- >>> splitOnSuffix' (== '.') "a"
2706-- >>> splitOnSuffix' (== '.') ".a"
2709-- >>> splitOnSuffix' (== '.') "a."
2712-- >>> splitOnSuffix' (== '.') "a.b"
2715-- >>> splitOnSuffix' (== '.') "a.b."
2718-- >>> splitOnSuffix' (== '.') "a..b.."
2721-- > lines = splitOnSuffix (== '\n')
2725-- This can be considered as an n-fold version of 'breakPost' where we apply
2726-- 'breakPost' successively on the input stream, dropping the first element
2727-- of the second segment after each break.
2729{-# INLINE splitOnSuffix #-}
2731 :: (IsStream t, Monad m)
2732 => (a -> Bool) -> Fold m a b -> t m a -> t m b
2733splitOnSuffix predicate f m =
2734 D.fromStreamD $ D.splitSuffixBy predicate f (D.toStreamD m)
2736-- | Like 'splitOn' after stripping leading, trailing, and repeated separators.
2737-- Therefore, @".a..b."@ with '.' as the separator would be parsed as
2738-- @["a","b"]@. In other words, its like parsing words from whitespace
2741-- > wordsBy' p xs = S.toList $ S.wordsBy p (FL.toList) (S.fromList xs)
2743-- >>> wordsBy' (== ',') ""
2746-- >>> wordsBy' (== ',') ","
2749-- >>> wordsBy' (== ',') ",a,,b,"
2752-- > words = wordsBy isSpace
2756-- It is equivalent to splitting in any of the infix/prefix/suffix styles
2757-- followed by removal of empty segments.
2758{-# INLINE wordsBy #-}
2760 :: (IsStream t, Monad m)
2761 => (a -> Bool) -> Fold m a b -> t m a -> t m b
2762wordsBy predicate f m =
2763 D.fromStreamD $ D.wordsBy predicate f (D.toStreamD m)
2765-- | Like 'splitOnSuffix' but keeps the suffix attached to the resulting
2768-- > splitWithSuffix' p xs = S.toList $ S.splitWithSuffix p (FL.toList) (S.fromList xs)
2770-- >>> splitWithSuffix' (== '.') ""
2773-- >>> splitWithSuffix' (== '.') "."
2776-- >>> splitWithSuffix' (== '.') "a"
2779-- >>> splitWithSuffix' (== '.') ".a"
2782-- >>> splitWithSuffix' (== '.') "a."
2785-- >>> splitWithSuffix' (== '.') "a.b"
2788-- >>> splitWithSuffix' (== '.') "a.b."
2791-- >>> splitWithSuffix' (== '.') "a..b.."
2792-- > ["a.",".","b.","."]
2796-- This can be considered as an n-fold version of 'breakPost' where we apply
2797-- 'breakPost' successively on the input stream.
2799{-# INLINE splitWithSuffix #-}
2801 :: (IsStream t, Monad m)
2802 => (a -> Bool) -> Fold m a b -> t m a -> t m b
2803splitWithSuffix predicate f m =
2804 D.fromStreamD $ D.splitSuffixBy' predicate f (D.toStreamD m)
2806------------------------------------------------------------------------------
2807-- Split on a delimiter sequence
2808------------------------------------------------------------------------------
2810-- Int list examples for splitOn:
2812-- >>> splitList [] [1,2,3,3,4]
2813-- > [[1],[2],[3],[3],[4]]
2815-- >>> splitList [5] [1,2,3,3,4]
2818-- >>> splitList [1] [1,2,3,3,4]
2821-- >>> splitList [4] [1,2,3,3,4]
2824-- >>> splitList [2] [1,2,3,3,4]
2827-- >>> splitList [3] [1,2,3,3,4]
2830-- >>> splitList [3,3] [1,2,3,3,4]
2833-- >>> splitList [1,2,3,3,4] [1,2,3,3,4]
2836-- | Like 'splitOn' but the separator is a sequence of elements instead of a
2839-- For illustration, let's define a function that operates on pure lists:
2842-- splitOnSeq' pat xs = S.toList $ S.splitOnSeq (A.fromList pat) (FL.toList) (S.fromList xs)
2845-- >>> splitOnSeq' "" "hello"
2846-- > ["h","e","l","l","o"]
2848-- >>> splitOnSeq' "hello" ""
2851-- >>> splitOnSeq' "hello" "hello"
2854-- >>> splitOnSeq' "x" "hello"
2857-- >>> splitOnSeq' "h" "hello"
2860-- >>> splitOnSeq' "o" "hello"
2863-- >>> splitOnSeq' "e" "hello"
2866-- >>> splitOnSeq' "l" "hello"
2869-- >>> splitOnSeq' "ll" "hello"
2872-- 'splitOnSeq' is an inverse of 'intercalate'. The following law always holds:
2874-- > intercalate . splitOn == id
2876-- The following law holds when the separator is non-empty and contains none of
2877-- the elements present in the input lists:
2879-- > splitOn . intercalate == id
2883-- XXX We can use a polymorphic vector implemented by Array# to represent the
2884-- sequence, that way we can avoid the Storable constraint. If we still need
2885-- Storable Array for performance, we can use a separate splitOnArray API for
2886-- that. We can also have an API where the sequence itself is a lazy stream, so
2887-- that we can search files in files for example.
2888{-# INLINE splitOnSeq #-}
2890 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
2891 => Array a -> Fold m a b -> t m a -> t m b
2892splitOnSeq patt f m = D.fromStreamD $ D.splitOn patt f (D.toStreamD m)
2895-- This can be implemented easily using Rabin Karp
2896-- | Split on any one of the given patterns.
2897{-# INLINE splitOnAny #-}
2899 :: (IsStream t, Monad m, Storable a, Integral a)
2900 => [Array a] -> Fold m a b -> t m a -> t m b
2901splitOnAny subseq f m = undefined -- D.fromStreamD $ D.splitOnAny f subseq (D.toStreamD m)
2904-- | Like 'splitSuffixBy' but the separator is a sequence of elements, instead
2905-- of a predicate for a single element.
2907-- > splitSuffixOn_ pat xs = S.toList $ S.splitSuffixOn (A.fromList pat) (FL.toList) (S.fromList xs)
2909-- >>> splitSuffixOn_ "." ""
2912-- >>> splitSuffixOn_ "." "."
2915-- >>> splitSuffixOn_ "." "a"
2918-- >>> splitSuffixOn_ "." ".a"
2921-- >>> splitSuffixOn_ "." "a."
2924-- >>> splitSuffixOn_ "." "a.b"
2927-- >>> splitSuffixOn_ "." "a.b."
2930-- >>> splitSuffixOn_ "." "a..b.."
2933-- > lines = splitSuffixOn "\n"
2936{-# INLINE splitOnSuffixSeq #-}
2938 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
2939 => Array a -> Fold m a b -> t m a -> t m b
2940splitOnSuffixSeq patt f m =
2941 D.fromStreamD $ D.splitSuffixOn False patt f (D.toStreamD m)
2944-- | Like 'splitOn' but drops any empty splits.
2946{-# INLINE wordsOn #-}
2948 :: (IsStream t, Monad m, Storable a, Eq a)
2949 => Array a -> Fold m a b -> t m a -> t m b
2950wordsOn subseq f m = undefined -- D.fromStreamD $ D.wordsOn f subseq (D.toStreamD m)
2953-- XXX use a non-monadic intersperse to remove the MonadAsync constraint.
2955-- | Like 'splitOnSeq' but splits the separator as well, as an infix token.
2957-- > splitOn'_ pat xs = S.toList $ S.splitOn' (A.fromList pat) (FL.toList) (S.fromList xs)
2959-- >>> splitOn'_ "" "hello"
2960-- > ["h","","e","","l","","l","","o"]
2962-- >>> splitOn'_ "hello" ""
2965-- >>> splitOn'_ "hello" "hello"
2968-- >>> splitOn'_ "x" "hello"
2971-- >>> splitOn'_ "h" "hello"
2974-- >>> splitOn'_ "o" "hello"
2977-- >>> splitOn'_ "e" "hello"
2980-- >>> splitOn'_ "l" "hello"
2981-- > ["he","l","","l","o"]
2983-- >>> splitOn'_ "ll" "hello"
2987{-# INLINE splitBySeq #-}
2989 :: (IsStream t, MonadAsync m, Storable a, Enum a, Eq a)
2990 => Array a -> Fold m a b -> t m a -> t m b
2991splitBySeq patt f m =
2992 intersperseM (fold f (A.toStream patt)) $ splitOnSeq patt f m
2994-- | Like 'splitSuffixOn' but keeps the suffix intact in the splits.
2996-- > splitSuffixOn'_ pat xs = S.toList $ FL.splitSuffixOn' (A.fromList pat) (FL.toList) (S.fromList xs)
2998-- >>> splitSuffixOn'_ "." ""
3001-- >>> splitSuffixOn'_ "." "."
3004-- >>> splitSuffixOn'_ "." "a"
3007-- >>> splitSuffixOn'_ "." ".a"
3010-- >>> splitSuffixOn'_ "." "a."
3013-- >>> splitSuffixOn'_ "." "a.b"
3016-- >>> splitSuffixOn'_ "." "a.b."
3019-- >>> splitSuffixOn'_ "." "a..b.."
3020-- > ["a.",".","b.","."]
3023{-# INLINE splitWithSuffixSeq #-}
3025 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
3026 => Array a -> Fold m a b -> t m a -> t m b
3027splitWithSuffixSeq patt f m =
3028 D.fromStreamD $ D.splitSuffixOn True patt f (D.toStreamD m)
3031-- This can be implemented easily using Rabin Karp
3032-- | Split post any one of the given patterns.
3033{-# INLINE splitSuffixOnAny #-}
3035 :: (IsStream t, Monad m, Storable a, Integral a)
3036 => [Array a] -> Fold m a b -> t m a -> t m b
3037splitSuffixOnAny subseq f m = undefined
3038 -- D.fromStreamD $ D.splitPostAny f subseq (D.toStreamD m)
3041------------------------------------------------------------------------------
3043------------------------------------------------------------------------------
3045-- | Consider a chunked stream of container elements e.g. a stream of @Word8@
3046-- chunked as a stream of arrays of @Word8@. @splitInnerBy splitter joiner
3047-- stream@ splits the inner containers @f a@ using the @splitter@ function and
3048-- joins back the resulting fragments from splitting across multiple containers
3049-- using the @joiner@ function such that the transformed output stream is
3050-- consolidated as one container per segment of the split.
3052-- CAUTION! This is not a true streaming function as the container size after
3053-- the split and merge may not be bounded.
3056{-# INLINE splitInnerBy #-}
3058 :: (IsStream t, Monad m)
3059 => (f a -> m (f a, Maybe (f a))) -- splitter
3060 -> (f a -> f a -> m (f a)) -- joiner
3063splitInnerBy splitter joiner xs =
3064 D.fromStreamD $ D.splitInnerBy splitter joiner $ D.toStreamD xs
3066-- | Like 'splitInnerBy' but splits assuming the separator joins the segment in
3070{-# INLINE splitInnerBySuffix #-}
3072 :: (IsStream t, Monad m, Eq (f a), Monoid (f a))
3073 => (f a -> m (f a, Maybe (f a))) -- splitter
3074 -> (f a -> f a -> m (f a)) -- joiner
3077splitInnerBySuffix splitter joiner xs =
3078 D.fromStreamD $ D.splitInnerBySuffix splitter joiner $ D.toStreamD xs
3080------------------------------------------------------------------------------
3081-- Reorder in sequence
3082------------------------------------------------------------------------------
3085-- Buffer until the next element in sequence arrives. The function argument
3086-- determines the difference in sequence numbers. This could be useful in
3087-- implementing sequenced streams, for example, TCP reassembly.
3088{-# INLINE reassembleBy #-}
3090 :: (IsStream t, Monad m)
3095reassembleBy = undefined
3098------------------------------------------------------------------------------
3100------------------------------------------------------------------------------
3102-- | Tap the data flowing through a stream into a 'Fold'. For example, you may
3103-- add a tap to log the contents flowing through the stream. The fold is used
3104-- only for effects, its result is discarded.
3109-- -----stream m a ---------------stream m a-----
3114-- > S.drain $ S.tap (FL.drainBy print) (S.enumerateFromTo 1 2)
3119-- Compare with 'trace'.
3122tap :: (IsStream t, Monad m) => FL.Fold m a b -> t m a -> t m a
3123tap f xs = D.fromStreamD $ D.tap f (D.toStreamD xs)
3125-- | Apply a monadic function to each element flowing through the stream and
3126-- discard the results.
3129-- > S.drain $ S.trace print (S.enumerateFromTo 1 2)
3134-- Compare with 'tap'.
3137trace :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m a
3138trace f = mapM (\x -> void (f x) >> return x)
3140------------------------------------------------------------------------------
3141-- Windowed classification
3142------------------------------------------------------------------------------
3144-- We divide the stream into windows or chunks in space or time and each window
3145-- can be associated with a key, all events associated with a particular key in
3146-- the window can be folded to a single result. The stream can be split into
3147-- windows by size or by using a split predicate on the elements in the stream.
3148-- For example, when we receive a closing flag, we can close the window.
3150-- A "chunk" is a space window and a "session" is a time window. Are there any
3151-- other better short words to describe them. An alternative is to use
3152-- "swindow" and "twindow". Another word for "session" could be "spell".
3154-- TODO: To mark the position in space or time we can have Indexed or
3155-- TimeStamped types. That can make it easy to deal with the position indices
3158------------------------------------------------------------------------------
3159-- Keyed Sliding Windows
3160------------------------------------------------------------------------------
3163{-# INLINABLE classifySlidingChunks #-}
3164classifySlidingChunks
3165 :: (IsStream t, MonadAsync m, Ord k)
3166 => Int -- ^ window size
3167 -> Int -- ^ window slide
3168 -> Fold m a b -- ^ Fold to be applied to window events
3169 -> t m (k, a, Bool) -- ^ window key, data, close event
3171classifySlidingChunks wsize wslide (Fold step initial extract) str
3174-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
3175-- slide the send window when an ack is received and we slide the receive
3176-- window when a sequence is complete. Sliding is stateful in case of TCP,
3177-- sliding releases the send buffer or makes data available to the user from
3178-- the receive buffer.
3179{-# INLINABLE classifySlidingSessions #-}
3180classifySlidingSessions
3181 :: (IsStream t, MonadAsync m, Ord k)
3182 => Double -- ^ timer tick in seconds
3183 -> Double -- ^ time window size
3184 -> Double -- ^ window slide
3185 -> Fold m a b -- ^ Fold to be applied to window events
3186 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
3188classifySlidingSessions tick interval slide (Fold step initial extract) str
3192------------------------------------------------------------------------------
3193-- Sliding Window Buffers
3194------------------------------------------------------------------------------
3196-- These buffered versions could be faster than concurrent incremental folds of
3197-- all overlapping windows as in many cases we may not need all the values to
3198-- compute the fold, we can just compute the result using the old value and new
3199-- value. However, we may need the buffer once in a while, for example for
3200-- string search we usually compute the hash incrementally but when the hash
3201-- matches the hash of the pattern we need to compare the whole string.
3203-- XXX we should be able to implement sequence based splitting combinators
3204-- using this combinator.
3207-- | Buffer n elements of the input in a ring buffer. When t new elements are
3208-- collected, slide the window to remove the same number of oldest elements,
3209-- insert the new elements, and apply an incremental fold on the sliding
3210-- window, supplying the outgoing elements, the new ring buffer as arguments.
3212 :: (IsStream t, Monad m, Ord a, Storable a)
3213 => Int -- window size
3214 -> Int -- window slide
3215 -> Fold m (Ring a, Array a) b
3218slidingChunkBuffer = undefined
3220-- Buffer n seconds worth of stream elements of the input in a radix tree.
3221-- Every t seconds, remove the items that are older than n seconds, and apply
3222-- an incremental fold on the sliding window, supplying the outgoing elements,
3223-- and the new radix tree buffer as arguments.
3225 :: (IsStream t, Monad m, Ord a, Storable a)
3226 => Int -- window size
3228 -> Fold m (RTree a, Array a) b
3231slidingSessionBuffer = undefined
3234------------------------------------------------------------------------------
3235-- Keyed Session Windows
3236------------------------------------------------------------------------------
3239-- | Keyed variable size space windows. Close the window if we do not receive a
3240-- window event in the next "spaceout" elements.
3241{-# INLINABLE classifyChunksBy #-}
3243 :: (IsStream t, MonadAsync m, Ord k)
3244 => Int -- ^ window spaceout (spread)
3245 -> Bool -- ^ reset the spaceout when a chunk window element is received
3246 -> Fold m a b -- ^ Fold to be applied to chunk window elements
3247 -> t m (k, a, Bool) -- ^ chunk key, data, last element
3249classifyChunksBy spanout reset (Fold step initial extract) str = undefined
3251-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
3252-- received within the chunk size window. The chunk gets closed only if no
3253-- element is received within the chunk window.
3255{-# INLINABLE classifyKeepAliveChunks #-}
3256classifyKeepAliveChunks
3257 :: (IsStream t, MonadAsync m, Ord k)
3258 => Int -- ^ window spaceout (spread)
3259 -> Fold m a b -- ^ Fold to be applied to chunk window elements
3260 -> t m (k, a, Bool) -- ^ chunk key, data, last element
3262classifyKeepAliveChunks spanout = classifyChunksBy spanout True
3265-- | @classifySessionsBy tick timeout reset f stream@ groups together all input
3266-- stream elements that belong to the same session. @timeout@ is the maximum
3267-- lifetime of a session in seconds. All elements belonging to a session are
3268-- purged after this duration. If "reset" is 'Ture' then the timeout is reset
3269-- after every event received in the session. Session duration is measured
3270-- using the timestamp of the first element seen for that session. To detect
3271-- session timeouts, a monotonic event time clock is maintained using the
3272-- timestamps seen in the inputs and a timer with a tick duration specified by
3275-- @session key@ is a key that uniquely identifies the session for the given
3276-- element, @timestamp@ characterizes the time when the input element was
3277-- generated, this is an absolute time measured from some @Epoch@. @session
3278-- close@ is a boolean indicating whether this element marks the closing of the
3279-- session. When an input element with @session close@ set to @True@ is seen
3280-- the session is purged immediately.
3282-- All the input elements belonging to a session are collected using the fold
3283-- @f@. The session key and the fold result are emitted in the output stream
3284-- when the session is purged either via the session close event or via the
3285-- session liftime timeout.
3288{-# INLINABLE classifySessionsBy #-}
3290 :: (IsStream t, MonadAsync m, Ord k)
3291 => Double -- ^ timer tick in seconds
3292 -> Double -- ^ session timeout
3293 -> Bool -- ^ reset the timeout when an event is received
3294 -> Fold m a b -- ^ Fold to be applied to session events
3295 -> t m (k, a, Bool, AbsTime) -- ^ session key, timestamp, close event, data
3297classifySessionsBy tick timeout reset (Fold step initial extract) str =
3298 concatMap (\(Tuple4' _ _ _ s) -> s) $ scanlM' sstep szero stream
3302 timeoutMs = toRelTime (round (timeout * 1000) :: MilliSecond64)
3303 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
3304 szero = Tuple4' (toAbsTime (0 :: MilliSecond64)) H.empty Map.empty K.nil
3306 -- Got a new stream input element
3307 sstep (Tuple4' evTime hp mp _) (Just (key, a, closing, ts)) =
3308 -- XXX we should use a heap in pinned memory to scale it to a large
3311 -- deleting a key from the heap is expensive, so we never delete a
3312 -- key, we just purge it from the Map and it gets purged from the
3313 -- heap on timeout. We just need an extra lookup in the Map when
3314 -- the key is purged from the heap, that should not be expensive.
3316 -- To detect session inactivity we keep a timestamp of the latest event
3317 -- in the Map along with the fold result. When we purge the session
3318 -- from the heap we match the timestamp in the heap with the timestamp
3319 -- in the Map, if the latest timestamp is newer and has not expired we
3320 -- reinsert the key in the heap.
3322 -- XXX if the key is an Int, we can also use an IntMap for slightly
3323 -- better performance.
3325 let accumulate v = do
3326 Tuple' _ old <- maybe (initial >>= return . Tuple' ts) return v
3328 return $ Tuple' ts new
3331 let (r, mp') = Map.updateLookupWithKey (\_ _ -> Nothing) key mp
3332 Tuple' _ acc <- accumulate r
3334 return $ Tuple4' evTime hp mp' (yield (key, res))
3336 let r = Map.lookup key mp
3338 let mp' = Map.insert key acc mp
3342 let expiry = addToAbsTime ts timeoutMs
3343 in H.insert (Entry expiry key) hp
3345 -- Event time is maintained as monotonically increasing
3346 -- time. If we have lagged behind any of the timestamps
3347 -- seen then we increase it to match the latest time seen
3348 -- in the timestamps. We also increase it on timer ticks.
3349 return $ Tuple4' (max evTime ts) hp' mp' K.nil
3351 -- Got a timer tick event
3352 -- XXX can we yield the entries without accumulating them?
3353 sstep (Tuple4' evTime heap sessions _) Nothing = do
3354 (hp', mp', out) <- go heap sessions K.nil
3355 return $ Tuple4' curTime hp' mp' out
3359 curTime = addToAbsTime evTime tickMs
3361 let hres = H.uncons hp
3363 Just (Entry ts key, hp') -> do
3364 let duration = diffAbsTime curTime ts
3365 if duration >= timeoutMs
3367 let (r, mp') = Map.updateLookupWithKey
3368 (\_ _ -> Nothing) key mp
3370 Nothing -> go hp' mp' out
3371 Just (Tuple' latestTS acc) -> do
3372 let dur = diffAbsTime curTime latestTS
3373 if dur >= timeoutMs || not reset
3376 go hp' mp' ((key, sess) `K.cons` out)
3378 -- reset the session timeout
3379 let expiry = addToAbsTime latestTS timeoutMs
3380 hp'' = H.insert (Entry expiry key) hp'
3381 mp'' = Map.insert key (Tuple' latestTS acc) mp'
3383 else return (hp, mp, out)
3384 Nothing -> return (hp, mp, out)
3386 -- merge timer events in the stream
3387 stream = Serial.map Just str `Par.parallel` repeatM timer
3389 liftIO $ threadDelay (round $ tick * 1000000)
3392-- | Like 'classifySessionsOf' but the session is kept alive if an event is
3393-- received within the session window. The session times out and gets closed
3394-- only if no event is received within the specified session window size.
3397{-# INLINABLE classifyKeepAliveSessions #-}
3398classifyKeepAliveSessions
3399 :: (IsStream t, MonadAsync m, Ord k)
3400 => Double -- ^ session inactive timeout
3401 -> Fold m a b -- ^ Fold to be applied to session payload data
3402 -> t m (k, a, Bool, AbsTime) -- ^ session key, data, close flag, timestamp
3404classifyKeepAliveSessions timeout = classifySessionsBy 1 timeout True
3406------------------------------------------------------------------------------
3407-- Keyed tumbling windows
3408------------------------------------------------------------------------------
3410-- Tumbling windows is a special case of sliding windows where the window slide
3411-- is the same as the window size. Or it can be a special case of session
3412-- windows where the reset flag is set to False.
3414-- XXX instead of using the early termination flag in the stream, we can use an
3415-- early terminating fold instead.
3418-- | Split the stream into fixed size chunks of specified size. Within each
3419-- such chunk fold the elements in buckets identified by the keys. A particular
3420-- bucket fold can be terminated early if a closing flag is encountered in an
3421-- element for that key.
3424{-# INLINABLE classifyChunksOf #-}
3426 :: (IsStream t, MonadAsync m, Ord k)
3427 => Int -- ^ window size
3428 -> Fold m a b -- ^ Fold to be applied to window events
3429 -> t m (k, a, Bool) -- ^ window key, data, close event
3431classifyChunksOf wsize = classifyChunksBy wsize False
3434-- | Split the stream into fixed size time windows of specified interval in
3435-- seconds. Within each such window, fold the elements in buckets identified by
3436-- the keys. A particular bucket fold can be terminated early if a closing flag
3437-- is encountered in an element for that key. Once a fold is terminated the key
3438-- and value for that bucket are emitted in the output stream.
3440-- Session @timestamp@ in the input stream is an absolute time from some epoch,
3441-- characterizing the time when the input element was generated. To detect
3442-- session window end, a monotonic event time clock is maintained synced with
3443-- the timestamps with a clock resolution of 1 second.
3446{-# INLINABLE classifySessionsOf #-}
3448 :: (IsStream t, MonadAsync m, Ord k)
3449 => Double -- ^ time window size
3450 -> Fold m a b -- ^ Fold to be applied to window events
3451 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
3453classifySessionsOf interval = classifySessionsBy 1 interval False
3455------------------------------------------------------------------------------
3457------------------------------------------------------------------------------
3459-- | Run a side effect before the stream yields its first element.
3462{-# INLINE before #-}
3463before :: (IsStream t, Monad m) => m b -> t m a -> t m a
3464before action xs = D.fromStreamD $ D.before action $ D.toStreamD xs
3466-- | Run a side effect whenever the stream stops normally.
3470after :: (IsStream t, Monad m) => m b -> t m a -> t m a
3471after action xs = D.fromStreamD $ D.after action $ D.toStreamD xs
3473-- | Run a side effect whenever the stream aborts due to an exception.
3476{-# INLINE onException #-}
3477onException :: (IsStream t, MonadCatch m) => m b -> t m a -> t m a
3478onException action xs = D.fromStreamD $ D.onException action $ D.toStreamD xs
3480-- | Run a side effect whenever the stream stops normally or aborts due to an
3484{-# INLINE finally #-}
3485finally :: (IsStream t, MonadCatch m) => m b -> t m a -> t m a
3486finally action xs = D.fromStreamD $ D.finally action $ D.toStreamD xs
3488-- | Run the first action before the stream starts and remember its output,
3489-- generate a stream using the output, run the second action using the
3490-- remembered value as an argument whenever the stream ends normally or due to
3494{-# INLINE bracket #-}
3495bracket :: (IsStream t, MonadCatch m)
3496 => m b -> (b -> m c) -> (b -> t m a) -> t m a
3497bracket bef aft bet = D.fromStreamD $
3498 D.bracket bef aft (\x -> toStreamD $ bet x)
3500-- | When evaluating a stream if an exception occurs, stream evaluation aborts
3501-- and the specified exception handler is run with the exception as argument.
3504{-# INLINE handle #-}
3505handle :: (IsStream t, MonadCatch m, Exception e)
3506 => (e -> m a) -> t m a -> t m a
3507handle handler xs = D.fromStreamD $ D.handle handler $ D.toStreamD xs
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