2{-# LANGUAGE RankNTypes #-}
3{-# LANGUAGE FlexibleContexts #-}
5#if __GLASGOW_HASKELL__ >= 800
6{-# OPTIONS_GHC -Wno-orphans #-}
9#include "Streams/inline.hs"
12-- Module : Streamly.Prelude
13-- Copyright : (c) 2017 Harendra Kumar
16-- Maintainer : harendra.kumar@gmail.com
17-- Stability : experimental
20-- This module is designed to be imported qualified:
23-- import qualified Streamly.Prelude as S
26-- Functions with the suffix @M@ are general functions that work on monadic
27-- arguments. The corresponding functions without the suffix @M@ work on pure
28-- arguments and can in general be derived from their monadic versions but are
29-- provided for convenience and for consistency with other pure APIs in the
32-- In many cases, short definitions of the combinators are provided in the
33-- documentation for illustration. The actual implementation may differ for
34-- performance reasons.
36-- Functions having a 'MonadAsync' constraint work concurrently when used with
37-- appropriate stream type combinator. Please be careful to not use 'parallely'
38-- with infinite streams.
40-- Deconstruction and folds accept a 'SerialT' type instead of a polymorphic
41-- type to ensure that streams always have a concrete monomorphic type by
42-- default, reducing type errors. In case you want to use any other type of
43-- stream you can use one of the type combinators provided in the "Streamly"
44-- module to convert the stream type.
46module Streamly.Prelude
50 -- | Primitives to construct a stream from pure values or monadic actions.
51 -- All other stream construction and generation combinators described later
52 -- can be expressed in terms of these primitives. However, the special
53 -- versions provided in this module can be much more efficient in most
54 -- cases. Users can create custom combinators using these primitives.
64 -- | Generate a monadic stream from a seed value or values.
72 -- Note: Using enumeration functions e.g. 'Prelude.enumFromThen' turns out
73 -- to be slightly faster than the idioms like @[from, then..]@.
76 -- | We can use the 'Enum' type class to enumerate a type producing a list
77 -- and then convert it to a stream:
80 -- 'fromList' $ 'Prelude.enumFromThen' from then
83 -- However, this is not particularly efficient.
84 -- The 'Enumerable' type class provides corresponding functions that
85 -- generate a stream instead of a list, efficiently.
92 -- | Generate a monadic stream from a seed value and a generator function.
100 -- ** From Containers
101 -- | Convert an input structure, container or source into a stream. All of
102 -- these can be expressed in terms of primitives.
111 -- | It is easy to express all the folds in terms of the 'uncons' primitive,
112 -- however the specific implementations provided later are generally more
120-- | In imperative terms a fold can be considered as a loop over the stream
121-- that reduces the stream to a single value.
122-- Left and right folds use a fold function @f@ and an identity element @z@
123-- (@zero@) to recursively deconstruct a structure and then combine and reduce
124-- the values or transform and reconstruct a new container.
126-- In general, a right fold is suitable for transforming and reconstructing a
127-- right associated structure (e.g. cons lists and streamly streams) and a left
128-- fold is suitable for reducing a right associated structure. The behavior of
129-- right and left folds are described in detail in the individual fold's
130-- documentation. To illustrate the two folds for cons lists:
132-- > foldr :: (a -> b -> b) -> b -> [a] -> b
134-- > foldr f z (x:xs) = x `f` foldr f z xs
136-- > foldl :: (b -> a -> b) -> b -> [a] -> b
138-- > foldl f z (x:xs) = foldl f (z `f` x) xs
140-- @foldr@ is conceptually equivalent to:
143-- > foldr f z [x] = f x z
144-- > foldr f z xs = foldr f (foldr f z (tail xs)) [head xs]
146-- @foldl@ is conceptually equivalent to:
149-- > foldl f z [x] = f z x
150-- > foldl f z xs = foldl f (foldl f z (init xs)) [last xs]
152-- Left and right folds are duals of each other.
155-- foldr f z xs = foldl (flip f) z (reverse xs)
156-- foldl f z xs = foldr (flip f) z (reverse xs)
162-- foldr f z xs = foldl g id xs z where g k x = k . f x
163-- foldl f z xs = foldr g id xs z where g x k = k . flip f x
167-- As a general rule, foldr cannot have state and foldl cannot have control.
169-- NOTE: Folds are inherently serial as each step needs to use the result of
170-- the previous step. However, it is possible to fold parts of the stream in
171-- parallel and then combine the results using a monoid.
184 -- ** Composable Left Folds
190 -- | Folds that are guaranteed to evaluate the whole stream.
192 -- -- ** To Summary (Full Folds)
193 -- -- | Folds that summarize the stream to a single value.
201 -- -- ** To Summary (Maybe) (Full Folds)
202 -- -- | Folds that summarize a non-empty stream to a 'Just' value and return
203 -- 'Nothing' for an empty stream.
209 -- , toListRev -- experimental
213 -- | Folds that generate a lazy structure. Note that the generated
214 -- structure may not be lazy if the underlying monad is strict.
216 -- -- ** To Containers (Full Folds)
217 -- -- | Convert or divert a stream into an output structure, container or
222 -- | Folds that may terminate before evaluating the whole stream. These
223 -- folds strictly evaluate the stream until the result is determined.
225 -- -- ** To Elements (Partial Folds)
229 -- -- | Folds that extract selected elements of a stream or their properties.
238 -- -- ** To Boolean (Partial Folds)
239 -- -- | Folds that summarize the stream to a boolean value.
253 -- | In imperative terms a map operation can be considered as a loop over
254 -- the stream that transforms the stream into another stream by performing
255 -- an operation on each element of the stream.
257 -- 'map' is the least powerful transformation operation with strictest
258 -- guarantees. A map, (1) is a stateless loop which means that no state is
259 -- allowed to be carried from one iteration to another, therefore,
260 -- operations on different elements are guaranteed to not affect each
261 -- other, (2) is a strictly one-to-one transformation of stream elements
262 -- which means it guarantees that no elements can be added or removed from
263 -- the stream, it can merely transform them.
271 -- | A scan is more powerful than map. While a 'map' is a stateless loop, a
272 -- @scan@ is a stateful loop which means that a state can be shared across
273 -- all the loop iterations, therefore, future iterations can be impacted by
274 -- the state changes made by the past iterations. A scan yields the state
275 -- of the loop after each iteration. Like a map, a @postscan@ or @prescan@
276 -- does not add or remove elements in the stream, it just transforms them.
277 -- However, a @scan@ adds one extra element to the stream.
279 -- A left associative scan, also known as a prefix sum, can be thought of
280 -- as a stream transformation consisting of left folds of all prefixes of a
281 -- stream. Another way of thinking about it is that it streams all the
282 -- intermediate values of the accumulator while applying a left fold on the
283 -- input stream. A right associative scan, on the other hand, can be
284 -- thought of as a stream consisting of right folds of all the suffixes of
287 -- The following equations hold for lists:
289 -- > scanl f z xs == map (foldl f z) $ inits xs
290 -- > scanr f z xs == map (foldr f z) $ tails xs
293 -- > scanl (+) 0 [1,2,3,4]
298 -- 0 + 1 + 2 + 3 + 4 = 10
300 -- > scanr (+) 0 [1,2,3,4]
301 -- 1 + 2 + 3 + 4 + 0 = 10
308 -- Left and right scans are duals:
310 -- > scanr f z xs == reverse $ scanl (flip f) z (reverse xs)
311 -- > scanl f z xs == reverse $ scanr (flip f) z (reverse xs)
313 -- A scan is a stateful map i.e. a combination of map and fold:
315 -- > map f xs = tail $ scanl (\_ x -> f x) z xs
316 -- > map f xs = reverse $ head $ scanr (\_ x -> f x) z xs
318 -- > foldl f z xs = last $ scanl f z xs
319 -- > foldr f z xs = head $ scanr f z xs
331 -- ** Scan Using Fold
349 -- , timestampedR -- timer
352 -- | Remove some elements from the stream based on a predicate. In
353 -- imperative terms a filter over a stream corresponds to a loop with a
354 -- @continue@ clause for the cases when the predicate fails.
359 -- ** Stateful Filters
373 -- , uniqBy -- by predicate e.g. to remove duplicate "/" in a path
374 -- , uniqOn -- to remove duplicate sequences
375 -- , pruneBy -- dropAround + uniqBy - like words
377 -- ** Mapping Filters
378 -- | Mapping along with filtering
383 -- ** Scanning Filters
384 -- | Stateful transformation along with filtering
388 -- , seqIndices -- search a sequence in the stream
391 -- | Insertion adds more elements to the stream.
397 -- , intersperseBySpan
398 -- , intersperseByTime
404 -- * Multi-Stream Operations
405 -- | New streams can be constructed by appending, merging or zipping
409 -- | Streams form a 'Semigroup' and a 'Monoid' under the append
414 -- -------Stream m a------|-------Stream m a------|=>----Stream m a---
419 -- >> S.toList $ S.fromList [1,2] \<> S.fromList [3,4]
421 -- >> S.toList $ fold $ [S.fromList [1,2], S.fromList [3,4]]
426 -- | Streams form a commutative semigroup under the merge
431 -- -------Stream m a------|
432 -- |=>----Stream m a---
433 -- -------Stream m a------|
447 -- -------Stream m a------|
448 -- |=>----Stream m c---
449 -- -------Stream m b------|
458 -- | Stream operations represent loops in imperative programming, streams
459 -- of streams represent nested loops.
470 -- , splitAt -- spanN
471 -- , splitIn -- sessionN
474 -- , span -- spanWhile
475 -- , break -- breakBefore
486 -- | Streams can be split into segments in space or in time. We use the
487 -- term @chunk@ to refer to a spatial length of the stream (spatial window)
488 -- and the term @session@ to refer to a length in time (time window).
490 -- In imperative terms, grouped folding can be considered as a nested loop
491 -- where we loop over the stream to group elements and then loop over
492 -- individual groups to fold them to a single value that is yielded in the
504 -- -- *** Using Element Separators
505 -- On == Dropping the separator
510 -- By == Keeping the separator
514 , wordsBy -- stripAndCompactBy
516 -- -- *** Using Sequence Separators
518 -- , splitOnSuffixSeq
519 -- , splitOnPrefixSeq
521 -- Keeping the delimiters
523 -- , splitBySeqSuffix
524 -- , splitBySeqPrefix
527 -- Splitting using multiple sequence separators
529 -- , splitOnAnySuffixSeq
530 -- , splitOnAnyPrefixSeq
542 -- * Windowed Classification
543 -- | Split the stream into windows or chunks in space or time. Each window
544 -- can be associated with a key, all events associated with a particular
545 -- key in the window can be folded to a single result. The stream is split
546 -- into windows of specified size, the window can be terminated early if
547 -- the closing flag is specified in the input stream.
549 -- The term "chunk" is used for a space window and the term "session" is
550 -- used for a time window.
552 -- ** Tumbling Windows
553 -- | A new window starts after the previous window is finished.
554 -- , classifyChunksOf
555 -- , classifySessionsOf
557 -- ** Keep Alive Windows
558 -- | The window size is extended if an event arrives within the specified
559 -- window size. This can represent sessions with idle or inactive timeout.
560 -- , classifyKeepAliveChunks
561 -- , classifyKeepAliveSessions
564 -- ** Sliding Windows
565 -- | A new window starts after the specified slide from the previous
566 -- window. Therefore windows can overlap.
567 , classifySlidingChunks
568 , classifySlidingSessions
570 -- ** Sliding Window Buffers
571 -- , slidingChunkBuffer
572 -- , slidingSessionBuffer
575 -- ** Containers of Streams
576 -- | These are variants of standard 'Foldable' fold functions that use a
577 -- polymorphic stream sum operation (e.g. 'async' or 'wSerial') to fold a
578 -- finite container of streams.
618import Control.Concurrent (threadDelay)
619import Control.Exception (Exception)
620import Control.Monad (void)
621import Control.Monad.Catch (MonadCatch)
622import Control.Monad.IO.Class (MonadIO(..))
623import Control.Monad.Trans (MonadTrans(..))
624import Data.Functor.Identity (Identity)
625import Data.Heap (Entry(..))
626import Data.Maybe (isJust, fromJust, isNothing)
627import Foreign.Storable (Storable)
629 hiding (filter, drop, dropWhile, take, takeWhile, zipWith, foldr,
630 foldl, map, mapM, mapM_, sequence, all, any, sum, product, elem,
631 notElem, maximum, minimum, head, last, tail, length, null,
632 reverse, iterate, init, and, or, lookup, foldr1, (!!),
633 scanl, scanl1, replicate, concatMap, span)
635import qualified Data.Heap as H
636import qualified Data.Map.Strict as Map
637import qualified Prelude
638import qualified System.IO as IO
640import Streamly.Enumeration (Enumerable(..), enumerate, enumerateTo)
641import Streamly.Fold.Types (Fold (..))
642import Streamly.Memory.Array (Array)
643import Streamly.SVar (MonadAsync, defState)
644import Streamly.Streams.Async (mkAsync')
645import Streamly.Streams.Combinators (maxYields)
646import Streamly.Streams.Prelude
647 (fromStreamS, toStreamS, foldWith, foldMapWith, forEachWith)
648import Streamly.Streams.StreamD (fromStreamD, toStreamD)
649import Streamly.Streams.StreamK (IsStream(..))
650import Streamly.Streams.Serial (SerialT)
651import Streamly.Pipe.Types (Pipe (..))
652import Streamly.Time.Units
653 (AbsTime, MilliSecond64(..), addToAbsTime, diffAbsTime, toRelTime,
656import Streamly.Strict
658import qualified Streamly.Memory.Array as A
659import qualified Streamly.Fold as FL
660import qualified Streamly.Fold.Types as FL
661import qualified Streamly.Streams.Prelude as P
662import qualified Streamly.Streams.StreamK as K
663import qualified Streamly.Streams.StreamD as D
664import qualified Streamly.Streams.Zip as Z
666#ifdef USE_STREAMK_ONLY
667import qualified Streamly.Streams.StreamK as S
668import qualified Streamly.Streams.Zip as S
670import qualified Streamly.Streams.StreamD as S
673import qualified Streamly.Streams.Serial as Serial
674import qualified Streamly.Streams.Parallel as Par
676------------------------------------------------------------------------------
678------------------------------------------------------------------------------
680-- | Decompose a stream into its head and tail. If the stream is empty, returns
681-- 'Nothing'. If the stream is non-empty, returns @Just (a, ma)@, where @a@ is
682-- the head of the stream and @ma@ its tail.
684-- This is a brute force primitive. Avoid using it as long as possible, use it
685-- when no other combinator can do the job. This can be used to do pretty much
686-- anything in an imperative manner, as it just breaks down the stream into
687-- individual elements and we can loop over them as we deem fit. For example,
688-- this can be used to convert a streamly stream into other stream types.
692uncons :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (a, t m a))
693uncons m = K.uncons (K.adapt m)
695------------------------------------------------------------------------------
696-- Generation by Unfolding
697------------------------------------------------------------------------------
704-- Just (a, b) -> a \`cons` unfoldr step b
707-- Build a stream by unfolding a /pure/ step function @step@ starting from a
708-- seed @s@. The step function returns the next element in the stream and the
709-- next seed value. When it is done it returns 'Nothing' and the stream ends.
716-- else Just (b, b + 1)
717-- in toList $ unfoldr f 0
724{-# INLINE_EARLY unfoldr #-}
725unfoldr :: (Monad m, IsStream t) => (b -> Maybe (a, b)) -> b -> t m a
726unfoldr step seed = fromStreamS (S.unfoldr step seed)
727{-# RULES "unfoldr fallback to StreamK" [1]
728 forall a b. S.toStreamK (S.unfoldr a b) = K.unfoldr a b #-}
730-- | Build a stream by unfolding a /monadic/ step function starting from a
731-- seed. The step function returns the next element in the stream and the next
732-- seed value. When it is done it returns 'Nothing' and the stream ends. For
738-- then return Nothing
739-- else print b >> return (Just (b, b + 1))
740-- in drain $ unfoldrM f 0
748-- When run concurrently, the next unfold step can run concurrently with the
749-- processing of the output of the previous step. Note that more than one step
750-- cannot run concurrently as the next step depends on the output of the
754-- (asyncly $ S.unfoldrM (\\n -> liftIO (threadDelay 1000000) >> return (Just (n, n + 1))) 0)
755-- & S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
761{-# INLINE_EARLY unfoldrM #-}
762unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a
765{-# RULES "unfoldrM serial" unfoldrM = unfoldrMSerial #-}
766{-# INLINE_EARLY unfoldrMSerial #-}
767unfoldrMSerial :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> SerialT m a
768unfoldrMSerial step seed = fromStreamS (S.unfoldrM step seed)
770------------------------------------------------------------------------------
771-- Specialized Generation
772------------------------------------------------------------------------------
774-- Faster than yieldM because there is no bind.
778-- yield a = a \`cons` nil
781-- Create a singleton stream from a pure value.
783-- The following holds in monadic streams, but not in Zip streams:
787-- yield = yieldM . pure
790-- In Zip applicative streams 'yield' is not the same as 'pure' because in that
791-- case 'pure' is equivalent to 'repeat' instead. 'yield' and 'pure' are
792-- equally efficient, in other cases 'yield' may be slightly more efficient
793-- than the other equivalent definitions.
797yield :: IsStream t => a -> t m a
802-- yieldM m = m \`consM` nil
805-- Create a singleton stream from a monadic action.
808-- > toList $ yieldM getLine
815yieldM :: (Monad m, IsStream t) => m a -> t m a
820-- fromIndices f = let g i = f i \`cons` g (i + 1) in g 0
823-- Generate an infinite stream, whose values are the output of a function @f@
824-- applied on the corresponding index. Index starts at 0.
827-- > S.toList $ S.take 5 $ S.fromIndices id
832{-# INLINE fromIndices #-}
833fromIndices :: (IsStream t, Monad m) => (Int -> a) -> t m a
834fromIndices = fromStreamS . S.fromIndices
839-- fromIndicesM f = let g i = f i \`consM` g (i + 1) in g 0
842-- Generate an infinite stream, whose values are the output of a monadic
843-- function @f@ applied on the corresponding index. Index starts at 0.
848{-# INLINE_EARLY fromIndicesM #-}
849fromIndicesM :: (IsStream t, MonadAsync m) => (Int -> m a) -> t m a
850fromIndicesM = K.fromIndicesM
852{-# RULES "fromIndicesM serial" fromIndicesM = fromIndicesMSerial #-}
853{-# INLINE fromIndicesMSerial #-}
854fromIndicesMSerial :: MonadAsync m => (Int -> m a) -> SerialT m a
855fromIndicesMSerial = fromStreamS . S.fromIndicesM
859-- replicateM = take n . repeatM
862-- Generate a stream by performing a monadic action @n@ times. Same as:
865-- drain $ serially $ S.replicateM 10 $ (threadDelay 1000000 >> print 1)
866-- drain $ asyncly $ S.replicateM 10 $ (threadDelay 1000000 >> print 1)
872{-# INLINE_EARLY replicateM #-}
873replicateM :: (IsStream t, MonadAsync m) => Int -> m a -> t m a
874replicateM = K.replicateM
876{-# RULES "replicateM serial" replicateM = replicateMSerial #-}
877{-# INLINE replicateMSerial #-}
878replicateMSerial :: MonadAsync m => Int -> m a -> SerialT m a
879replicateMSerial n = fromStreamS . S.replicateM n
883-- replicate = take n . repeat
886-- Generate a stream of length @n@ by repeating a value @n@ times.
889{-# INLINE_NORMAL replicate #-}
890replicate :: (IsStream t, Monad m) => Int -> a -> t m a
891replicate n = fromStreamS . S.replicate n
895-- repeatM = fix . consM
896-- repeatM = cycle1 . yieldM
899-- Generate a stream by repeatedly executing a monadic action forever.
902-- drain $ serially $ S.take 10 $ S.repeatM $ (threadDelay 1000000 >> print 1)
903-- drain $ asyncly $ S.take 10 $ S.repeatM $ (threadDelay 1000000 >> print 1)
906-- /Concurrent, infinite (do not use with 'parallely')/
909repeatM :: (IsStream t, MonadAsync m) => m a -> t m a
911 where go m = m |: go m
915-- iterate f x = x \`cons` iterate f x
918-- Generate an infinite stream with @x@ as the first element and each
919-- successive element derived by applying the function @f@ on the previous
923-- > S.toList $ S.take 5 $ S.iterate (+1) 1
928iterate :: IsStream t => (a -> a) -> a -> t m a
929iterate step = fromStream . go
931 go s = K.cons s (go (step s))
935-- iterateM f m = m >>= \a -> return a \`consM` iterateM f (f a)
938-- Generate an infinite stream with the first element generated by the action
939-- @m@ and each successive element derived by applying the monadic function
940-- @f@ on the previous element.
942-- When run concurrently, the next iteration can run concurrently with the
943-- processing of the previous iteration. Note that more than one iteration
944-- cannot run concurrently as the next iteration depends on the output of the
945-- previous iteration.
948-- drain $ serially $ S.take 10 $ S.iterateM
949-- (\\x -> threadDelay 1000000 >> print x >> return (x + 1)) (return 0)
951-- drain $ asyncly $ S.take 10 $ S.iterateM
952-- (\\x -> threadDelay 1000000 >> print x >> return (x + 1)) (return 0)
957-- /Since: 0.7.0 (signature change)/
960iterateM :: (IsStream t, MonadAsync m) => (a -> m a) -> m a -> t m a
963 go s = K.mkStream $ \st stp sng yld -> do
965 K.foldStreamShared st stp sng yld (return next |: go (step next))
967------------------------------------------------------------------------------
969------------------------------------------------------------------------------
973-- fromListM = 'Prelude.foldr' 'K.consM' 'K.nil'
976-- Construct a stream from a list of monadic actions. This is more efficient
977-- than 'fromFoldableM' for serial streams.
980{-# INLINE_EARLY fromListM #-}
981fromListM :: (MonadAsync m, IsStream t) => [m a] -> t m a
982fromListM = fromStreamD . D.fromListM
983{-# RULES "fromListM fallback to StreamK" [1]
984 forall a. D.toStreamK (D.fromListM a) = fromFoldableM a #-}
988-- fromFoldableM = 'Prelude.foldr' 'consM' 'K.nil'
991-- Construct a stream from a 'Foldable' containing monadic actions.
994-- drain $ serially $ S.fromFoldableM $ replicateM 10 (threadDelay 1000000 >> print 1)
995-- drain $ asyncly $ S.fromFoldableM $ replicateM 10 (threadDelay 1000000 >> print 1)
998-- /Concurrent (do not use with 'parallely' on infinite containers)/
1001{-# INLINE fromFoldableM #-}
1002fromFoldableM :: (IsStream t, MonadAsync m, Foldable f) => f (m a) -> t m a
1003fromFoldableM = Prelude.foldr consM K.nil
1005-- | Same as 'fromFoldable'.
1008{-# DEPRECATED each "Please use fromFoldable instead." #-}
1010each :: (IsStream t, Foldable f) => f a -> t m a
1011each = K.fromFoldable
1013-- | Read lines from an IO Handle into a stream of Strings.
1016{-# DEPRECATED fromHandle
1017 "Please use Streamly.FileSystem.Handle module (see the changelog)" #-}
1018fromHandle :: (IsStream t, MonadIO m) => IO.Handle -> t m String
1021 go = K.mkStream $ \_ yld _ stp -> do
1022 eof <- liftIO $ IO.hIsEOF h
1026 str <- liftIO $ IO.hGetLine h
1029------------------------------------------------------------------------------
1030-- Elimination by Folding
1031------------------------------------------------------------------------------
1033-- | Right associative/pull fold.
1035-- Example, determine if any element is 'odd' in a stream:
1037-- >>> S.foldrM (\x xs -> if odd x then return True else xs) (return False) $ S.fromList (2:4:5:undefined)
1040-- Let's take a closer look at the @foldr@ definition for lists, as given
1044-- foldr f z (x:xs) = x \`f` foldr f z xs
1047-- @foldr@ invokes the fold step function @f@ as @f x (foldr f z xs)@. At each
1048-- invocation of @f@ @foldr@ gives us the next element in the input container
1049-- @x@ and a recursive expression @foldr f z xs@ representing the yet unbuilt
1050-- (lazy thunk) part of the output. Therefore, when @f x xs@ is lazy in @xs@
1051-- it can consume the input one element at a time in FIFO order to build a lazy
1052-- output expression. For example,
1054-- > f x remaining = show x : remaining
1056-- @take 2 $ foldr f [] (1:2:undefined)@ would consume the input lazily on
1057-- demand, consuming only first two elements and resulting in ["1", "2"]. @f@
1058-- can terminate recursion by not evaluating the @remaining@ part:
1060-- > f 2 remaining = show 2 : []
1061-- > f x remaining = show x : remaining
1063-- @f@ would terminate recursion whenever it sees element @2@ in the input.
1064-- Therefore, @foldr f [] (1:2:undefined)@ would work as before. If @f a b@ is
1065-- strict in @b@ it would end up consuming the whole input right away and
1066-- expanding the recursive expression @b@ (i.e. @foldr f z xs@) fully before
1067-- it yields an output expression, resulting in the following /right
1068-- associated expression/:
1071-- foldr f z xs == x1 \`f` (x2 \`f` ...(xn \`f` z))
1076-- > f x remaining = x + remaining
1078-- With this definition, @foldr f 0 [1..1000]@, would recurse completely until
1079-- it reaches the recursion terminating case @... `f` (1000 `f` 0)@, and then
1080-- start reducing the whole expression from right to left, therefore, consuming
1081-- the input elements in LIFO order. Thus, such an evaluation would require
1082-- memory proportional to the size of input. Try out @foldr (+) 0 (map (\\x ->
1083-- trace (show x) x) [1..10])@.
1085-- Notice, the order of the arguments to the step function @f a b@. It follows
1086-- the order of @a@ and @b@ in the right associative recursive expression
1087-- generated by expanding @a \`f` b@.
1089-- A right fold is a pull fold, the step function is the puller, it can pull
1090-- more data from the input container by using its second argument in the
1091-- output expression or terminate pulling by not using it. As a corollary:
1093-- 1. a step function which is lazy in its second argument (usually functions
1094-- or constructors that build a lazy structure e.g. @(:)@) can pull lazily on
1096-- 2. a step function strict in its second argument (usually reducers e.g.
1097-- (+)) would end up pulling all of its input and buffer it in memory before
1098-- potentially reducing it.
1100-- A right fold is suitable for lazy reconstructions e.g. transformation,
1101-- mapping, filtering of /right associated input structures/ (e.g. cons lists).
1102-- Whereas a left fold is suitable for reductions (e.g. summing a stream of
1103-- numbers) of right associated structures. Note that these roles will reverse
1104-- for left associated structures (e.g. snoc lists). Most of our observations
1105-- here assume right associated structures, lists being the canonical example.
1107-- 1. A lazy FIFO style pull using a right fold allows pulling a potentially
1108-- /infinite/ input stream lazily, perform transformations on it, and
1109-- reconstruct a new structure without having to buffer the whole structure. In
1110-- contrast, a left fold would buffer the entire structure before the
1111-- reconstructed structure can be consumed.
1112-- 2. Even if buffering the entire input structure is ok, we need to keep in
1113-- mind that a right fold reconstructs structures in a FIFO style, whereas a
1114-- left fold reconstructs in a LIFO style, thereby reversing the order of
1116-- 3. A right fold has termination control and therefore can terminate early
1117-- without going throught the entire input, a left fold cannot terminate
1118-- without consuming all of its input. For example, a right fold
1119-- implementation of 'or' can terminate as soon as it finds the first 'True'
1120-- element, whereas a left fold would necessarily go through the entire input
1121-- irrespective of that.
1122-- 4. Reduction (e.g. using (+) on a stream of numbers) using a right fold
1123-- occurs in a LIFO style, which means that the entire input gets buffered
1124-- before reduction starts. Whereas with a strict left fold reductions occur
1125-- incrementally in FIFO style. Therefore, a strict left fold is more suitable
1128-- /Since: 0.7.0 (signature changed)/
1130-- /Since: 0.2.0 (signature changed)/
1133{-# INLINE foldrM #-}
1134foldrM :: Monad m => (a -> m b -> m b) -> m b -> SerialT m a -> m b
1137-- | Right fold to a streaming monad.
1139-- > foldrS S.cons S.nil === id
1141-- 'foldrS' can be used to perform stateless stream to stream transformations
1142-- like map and filter in general. It can be coupled with a scan to perform
1143-- stateful transformations. However, note that the custom map and filter
1144-- routines can be much more efficient than this due to better stream fusion.
1146-- >>> S.toList $ S.foldrS S.cons S.nil $ S.fromList [1..5]
1149-- Find if any element in the stream is 'True':
1151-- >>> 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)
1154-- Map (+2) on odd elements and filter out the even elements:
1156-- >>> 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)
1159-- 'foldrM' can also be represented in terms of 'foldrS', however, the former
1160-- is much more efficient:
1162-- > foldrM f z s = runIdentityT $ foldrS (\x xs -> lift $ f x (runIdentityT xs)) (lift z) s
1165{-# INLINE foldrS #-}
1166foldrS :: IsStream t => (a -> t m b -> t m b) -> t m b -> t m a -> t m b
1169-- | Right fold to a transformer monad. This is the most general right fold
1170-- function. 'foldrS' is a special case of 'foldrT', however 'foldrS'
1171-- implementation can be more efficient:
1174-- > foldrM f z s = runIdentityT $ foldrT (\x xs -> lift $ f x (runIdentityT xs)) (lift z) s
1176-- 'foldrT' can be used to translate streamly streams to other transformer
1177-- monads e.g. to a different streaming type.
1180{-# INLINE foldrT #-}
1181foldrT :: (IsStream t, Monad m, Monad (s m), MonadTrans s)
1182 => (a -> s m b -> s m b) -> s m b -> t m a -> s m b
1183foldrT f z s = S.foldrT f z (toStreamS s)
1185-- | Right fold for lazy monads or pure streams.
1187-- Please avoid using this routine in strict monads like IO unless you need a
1188-- strict right fold. This is provided only for use in lazy monads (e.g.
1189-- Identity) or pure streams. Note that with this signature it is not possible
1190-- to implement a lazy foldr when the monad @m@ is strict. In that case it
1191-- would be strict in its accumulator and therefore would necessarily consume
1196foldr :: Monad m => (a -> b -> b) -> b -> SerialT m a -> m b
1199-- XXX This seems to be of limited use as it cannot be used to construct
1200-- recursive structures and for reduction foldl1' is better.
1202-- | Lazy right fold for non-empty streams, using first element as the starting
1203-- value. Returns 'Nothing' if the stream is empty.
1206{-# INLINE foldr1 #-}
1207{-# DEPRECATED foldr1 "Use foldrM instead." #-}
1208foldr1 :: Monad m => (a -> a -> a) -> SerialT m a -> m (Maybe a)
1209foldr1 f m = S.foldr1 f (toStreamS m)
1211-- | Strict left fold with an extraction function. Like the standard strict
1212-- left fold, but applies a user supplied extraction function (the third
1213-- argument) to the folded value at the end. This is designed to work with the
1214-- @foldl@ library. The suffix @x@ is a mnemonic for extraction.
1217{-# DEPRECATED foldx "Please use foldl' followed by fmap instead." #-}
1219foldx :: Monad m => (x -> a -> x) -> x -> (x -> b) -> SerialT m a -> m b
1222-- | Strict left associative/push fold. Note that the observations below about
1223-- the behavior of a left fold assume that we are working on a right associated
1224-- structure like cons lists and streamly streams. If we are working on a left
1225-- associated structure (e.g. snoc lists) the roles of right and left folds
1228-- Let's take a closer look at the @foldl@ definition for lists given above:
1231-- foldl f z (x:xs) = foldl f (z \`f` x) xs
1234-- @foldl@ calls itself recursively, in each call it invokes @f@ as @f z x@
1235-- providing it with the result accumulated till now @z@ (the state) and the
1236-- next element from the input container. First call to @f@ is supplied with
1237-- the initial value of the accumulator @z@ and each subsequent call uses the
1238-- output of the previous call to @f z x@.
1240-- >> foldl' (+) 0 [1,2,3]
1243-- The recursive call at the head of the output expression is bound to be
1244-- evaluated until recursion terminates, therefore, a left fold always
1245-- /consumes the whole input container/. The following would result in an
1246-- error, even though the fold is not using the values at all:
1248-- >> foldl' (\_ _ -> 0) 0 (1:undefined)
1249-- > *** Exception: Prelude.undefined
1251-- As @foldl@ recurses, it builds the left associated expression shown below.
1252-- Notice, the order of the arguments to the step function @f b a@. It follows
1253-- the left associative recursive expression generated by expanding @b \`f` a@.
1256-- foldl f z xs == (((z \`f` x1) \`f` x2) ...) \`f` xn
1260-- The strict left fold @foldl'@ forces the reduction of its argument @z \`f`
1261-- x@ before using it, therefore it never builds the whole expression in
1262-- memory. Thus, @z \`f` x1@ would get reduced to @z1@ and then @z1 \`f` x2@
1263-- would get reduced to @z2@ and so on, incrementally reducing the expression
1264-- from left to right as it recurses, consuming the input in FIFO order. Try
1265-- out @foldl' (+) 0 (map (\\x -> trace (show x) x) [1..10])@ to see how it
1266-- works. For example:
1269-- > S.foldl' (+) 0 $ S.fromList [1,2,3,4]
1280-- However, @foldl'@ evaluates the accumulator only to WHNF. It may further
1281-- help if the step function uses a strict data structure as accumulator to
1282-- improve performance and to keep the expression fully reduced at all times
1285-- A left fold can also build a new structure instead of reducing one if a
1286-- constructor is used as a fold step. However, it may not be very useful
1287-- because it will consume the whole input and construct the new structure in
1288-- memory before we can consume it. Thus the whole structure gets buffered in
1289-- memory. When the list constructor is used it would build a new list in
1290-- reverse (LIFO) order:
1293-- > S.foldl' (flip (:)) [] $ S.fromList [1,2,3,4]
1297-- A left fold is a push fold. The producer pushes its contents to the step
1298-- function of the fold. The step function therefore has no control to stop the
1299-- input, it can only discard it if it does not need it. We can also consider a
1300-- left fold as a state machine where the state is store in the accumulator,
1301-- the state can be modified based on new inputs that are pushed to the fold.
1303-- In general, a strict left fold is a reducing fold, whereas a right fold is a
1304-- constructing fold. A strict left fold reduces in a FIFO order whereas it
1305-- constructs in a LIFO order, and vice-versa for the right fold. See the
1306-- documentation of 'foldrM' for a discussion on where a left or right fold is
1309-- To perform a left fold lazily without having to consume all the input one
1310-- can use @scanl@ to stream the intermediate results of the fold and consume
1311-- the resulting stream lazily.
1314{-# INLINE foldl' #-}
1315foldl' :: Monad m => (b -> a -> b) -> b -> SerialT m a -> m b
1318-- | Strict left fold, for non-empty streams, using first element as the
1319-- starting value. Returns 'Nothing' if the stream is empty.
1322{-# INLINE foldl1' #-}
1323foldl1' :: Monad m => (a -> a -> a) -> SerialT m a -> m (Maybe a)
1327 Nothing -> return Nothing
1329 res <- foldl' step h t
1332-- | Like 'foldx', but with a monadic step function.
1335{-# DEPRECATED foldxM "Please use foldlM' followed by fmap instead." #-}
1336{-# INLINE foldxM #-}
1337foldxM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> SerialT m a -> m b
1340-- | Like 'foldl'' but with a monadic step function.
1343{-# INLINE foldlM' #-}
1344foldlM' :: Monad m => (b -> a -> m b) -> b -> SerialT m a -> m b
1345foldlM' step begin m = S.foldlM' step begin $ toStreamS m
1347------------------------------------------------------------------------------
1349------------------------------------------------------------------------------
1353-- "Streamly.Fold" module defines composable left folds which can be combined
1354-- together in many interesting ways. Those folds can be run using 'runFold'.
1355-- The following two ways of folding are equivalent in functionality and
1358-- >>> S.runFold FL.sum (S.enumerateFromTo 1 100)
1360-- >>> S.sum (S.enumerateFromTo 1 100)
1363-- However, left folds cannot terminate early even if it does not need to
1364-- consume more input to determine the result. Therefore, the performance is
1365-- equivalent only for full folds like 'sum' and 'length'. For partial folds
1366-- like 'head' or 'any' the the folds defined in this module may be much more
1367-- efficient because they are implemented as right folds that terminate as soon
1368-- as we get the result. Note that when a full fold is composed with a partial
1369-- fold in parallel the performance is not impacted as we anyway have to
1370-- consume the whole stream due to the full fold.
1372-- >>> S.head (1 `S.cons` undefined)
1374-- >>> S.runFold FL.head (1 `S.cons` undefined)
1375-- *** Exception: Prelude.undefined
1377-- However, we can wrap the fold in a scan to convert it into a lazy stream of
1378-- fold steps. We can then terminate the stream whenever we want. For example,
1380-- >>> S.toList $ S.take 1 $ S.runScan FL.head (1 `S.cons` undefined)
1383-- The following example extracts the input stream up to a point where the
1384-- running average of elements is no more than 10:
1388-- $ S.map (fromJust . fst)
1389-- $ S.takeWhile (\\(_,x) -> x <= 10)
1390-- $ S.runPostscan ((,) \<$> FL.last \<*> avg) (S.enumerateFromTo 1.0 100.0)
1393-- [1.0,2.0,3.0,4.0,5.0,6.0,7.0,8.0,9.0,10.0,11.0,12.0,13.0,14.0,15.0,16.0,17.0,18.0,19.0]
1396-- | Fold a stream using the supplied left fold.
1398-- >>> S.runFold FL.sum (S.enumerateFromTo 1 100)
1402{-# INLINE runFold #-}
1403runFold :: Monad m => Fold m a b -> SerialT m a -> m b
1406------------------------------------------------------------------------------
1408------------------------------------------------------------------------------
1411-- | Drain a stream to a 'Sink'.
1412{-# INLINE runSink #-}
1413runSink :: Monad m => Sink m a -> SerialT m a -> m ()
1414runSink = runFold . toFold
1417------------------------------------------------------------------------------
1419------------------------------------------------------------------------------
1422-- > drain = mapM_ (\_ -> return ())
1424-- Run a stream, discarding the results. By default it interprets the stream
1425-- as 'SerialT', to run other types of streams use the type adapting
1426-- combinators for example @drain . 'asyncly'@.
1430drain :: Monad m => SerialT m a -> m ()
1433-- | Run a stream, discarding the results. By default it interprets the stream
1434-- as 'SerialT', to run other types of streams use the type adapting
1435-- combinators for example @runStream . 'asyncly'@.
1438{-# DEPRECATED runStream "Please use \"drain\" instead" #-}
1439{-# INLINE runStream #-}
1440runStream :: Monad m => SerialT m a -> m ()
1444-- > drainN n = drain . take n
1446-- Run maximum up to @n@ iterations of a stream.
1449{-# INLINE drainN #-}
1450drainN :: Monad m => Int -> SerialT m a -> m ()
1451drainN n = drain . take n
1454-- > runN n = runStream . take n
1456-- Run maximum up to @n@ iterations of a stream.
1459{-# DEPRECATED runN "Please use \"drainN\" instead" #-}
1461runN :: Monad m => Int -> SerialT m a -> m ()
1465-- > drainWhile p = drain . takeWhile p
1467-- Run a stream as long as the predicate holds true.
1470{-# INLINE drainWhile #-}
1471drainWhile :: Monad m => (a -> Bool) -> SerialT m a -> m ()
1472drainWhile p = drain . takeWhile p
1475-- > runWhile p = runStream . takeWhile p
1477-- Run a stream as long as the predicate holds true.
1480{-# DEPRECATED runWhile "Please use \"drainWhile\" instead" #-}
1481{-# INLINE runWhile #-}
1482runWhile :: Monad m => (a -> Bool) -> SerialT m a -> m ()
1483runWhile = drainWhile
1485-- | Determine whether the stream is empty.
1489null :: Monad m => SerialT m a -> m Bool
1492-- | Extract the first element of the stream, if any.
1498head :: Monad m => SerialT m a -> m (Maybe a)
1502-- > tail = fmap (fmap snd) . uncons
1504-- Extract all but the first element of the stream, if any.
1508tail :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (t m a))
1509tail m = K.tail (K.adapt m)
1511-- | Extract all but the last element of the stream, if any.
1515init :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (t m a))
1516init m = K.init (K.adapt m)
1518-- | Extract the last element of the stream, if any.
1520-- > last xs = xs !! (length xs - 1)
1524last :: Monad m => SerialT m a -> m (Maybe a)
1525last m = S.last $ toStreamS m
1527-- | Determine whether an element is present in the stream.
1531elem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool
1532elem e m = S.elem e (toStreamS m)
1534-- | Determine whether an element is not present in the stream.
1537{-# INLINE notElem #-}
1538notElem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool
1539notElem e m = S.notElem e (toStreamS m)
1541-- | Determine the length of the stream.
1544{-# INLINE length #-}
1545length :: Monad m => SerialT m a -> m Int
1546length = foldl' (\n _ -> n + 1) 0
1548-- | Determine whether all elements of a stream satisfy a predicate.
1552all :: Monad m => (a -> Bool) -> SerialT m a -> m Bool
1553all p m = S.all p (toStreamS m)
1555-- | Determine whether any of the elements of a stream satisfy a predicate.
1559any :: Monad m => (a -> Bool) -> SerialT m a -> m Bool
1560any p m = S.any p (toStreamS m)
1562-- | Determines if all elements of a boolean stream are True.
1566and :: Monad m => SerialT m Bool -> m Bool
1569-- | Determines whether at least one element of a boolean stream is True.
1573or :: Monad m => SerialT m Bool -> m Bool
1576-- | Determine the sum of all elements of a stream of numbers. Returns @0@ when
1577-- the stream is empty. Note that this is not numerically stable for floating
1582sum :: (Monad m, Num a) => SerialT m a -> m a
1585-- | Determine the product of all elements of a stream of numbers. Returns @1@
1586-- when the stream is empty.
1589{-# INLINE product #-}
1590product :: (Monad m, Num a) => SerialT m a -> m a
1591product = foldl' (*) 1
1595-- minimum = 'minimumBy' compare
1598-- Determine the minimum element in a stream.
1601{-# INLINE minimum #-}
1602minimum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)
1603minimum m = S.minimum (toStreamS m)
1605-- | Determine the minimum element in a stream using the supplied comparison
1609{-# INLINE minimumBy #-}
1610minimumBy :: Monad m => (a -> a -> Ordering) -> SerialT m a -> m (Maybe a)
1611minimumBy cmp m = S.minimumBy cmp (toStreamS m)
1615-- maximum = 'maximumBy' compare
1618-- Determine the maximum element in a stream.
1621{-# INLINE maximum #-}
1622maximum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)
1623maximum m = S.maximum (toStreamS m)
1625-- | Determine the maximum element in a stream using the supplied comparison
1629{-# INLINE maximumBy #-}
1630maximumBy :: Monad m => (a -> a -> Ordering) -> SerialT m a -> m (Maybe a)
1631maximumBy cmp m = S.maximumBy cmp (toStreamS m)
1633-- | Lookup the element at the given index.
1637(!!) :: Monad m => SerialT m a -> Int -> m (Maybe a)
1638m !! i = toStreamS m S.!! i
1640-- | In a stream of (key-value) pairs @(a, b)@, return the value @b@ of the
1641-- first pair where the key equals the given value @a@.
1643-- > lookup = snd <$> find ((==) . fst)
1646{-# INLINE lookup #-}
1647lookup :: (Monad m, Eq a) => a -> SerialT m (a, b) -> m (Maybe b)
1648lookup a m = S.lookup a (toStreamS m)
1650-- | Like 'findM' but with a non-monadic predicate.
1652-- > find p = findM (return . p)
1656find :: Monad m => (a -> Bool) -> SerialT m a -> m (Maybe a)
1657find p m = S.find p (toStreamS m)
1659-- | Returns the first element that satisfies the given predicate.
1663findM :: Monad m => (a -> m Bool) -> SerialT m a -> m (Maybe a)
1664findM p m = S.findM p (toStreamS m)
1666-- | Find all the indices where the element in the stream satisfies the given
1670{-# INLINE findIndices #-}
1671findIndices :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m Int
1672findIndices p m = fromStreamS $ S.findIndices p (toStreamS m)
1674-- | Returns the first index that satisfies the given predicate.
1677{-# INLINE findIndex #-}
1678findIndex :: Monad m => (a -> Bool) -> SerialT m a -> m (Maybe Int)
1679findIndex p = head . findIndices p
1681-- | Find all the indices where the value of the element in the stream is equal
1682-- to the given value.
1685{-# INLINE elemIndices #-}
1686elemIndices :: (IsStream t, Eq a, Monad m) => a -> t m a -> t m Int
1687elemIndices a = findIndices (==a)
1689-- | Returns the first index where a given value is found in the stream.
1691-- > elemIndex a = findIndex (== a)
1694{-# INLINE elemIndex #-}
1695elemIndex :: (Monad m, Eq a) => a -> SerialT m a -> m (Maybe Int)
1696elemIndex a = findIndex (== a)
1698------------------------------------------------------------------------------
1700------------------------------------------------------------------------------
1702-- | Returns 'True' if the first stream is the same as or a prefix of the
1706-- > S.isPrefixOf (S.fromList "hello") (S.fromList "hello" :: SerialT IO Char)
1711{-# INLINE isPrefixOf #-}
1712isPrefixOf :: (Eq a, IsStream t, Monad m) => t m a -> t m a -> m Bool
1713isPrefixOf m1 m2 = D.isPrefixOf (toStreamD m1) (toStreamD m2)
1715-- | Returns 'True' if all the elements of the first stream occur, in order, in
1716-- the second stream. The elements do not have to occur consecutively. A stream
1717-- is treated as a subsequence of itself.
1720-- > S.isSubsequenceOf (S.fromList "hlo") (S.fromList "hello" :: SerialT IO Char)
1725{-# INLINE isSubsequenceOf #-}
1726isSubsequenceOf :: (Eq a, IsStream t, Monad m) => t m a -> t m a -> m Bool
1727isSubsequenceOf m1 m2 = D.isSubsequenceOf (toStreamD m1) (toStreamD m2)
1729-- | Drops the given prefix from a stream. Returns 'Nothing' if the stream does
1730-- not start with the given prefix. Returns @Just nil@ when the prefix is the
1731-- same as the stream.
1734{-# INLINE stripPrefix #-}
1736 :: (Eq a, IsStream t, Monad m)
1737 => t m a -> t m a -> m (Maybe (t m a))
1738stripPrefix m1 m2 = fmap fromStreamD <$>
1739 D.stripPrefix (toStreamD m1) (toStreamD m2)
1741------------------------------------------------------------------------------
1743------------------------------------------------------------------------------
1745-- XXX this can utilize parallel mapping if we implement it as drain . mapM
1747-- > mapM_ = drain . mapM
1749-- Apply a monadic action to each element of the stream and discard the output
1750-- of the action. This is not really a pure transformation operation but a
1751-- transformation followed by fold.
1755mapM_ :: Monad m => (a -> m b) -> SerialT m a -> m ()
1756mapM_ f m = S.mapM_ f $ toStreamS m
1758------------------------------------------------------------------------------
1760------------------------------------------------------------------------------
1764-- toList = S.foldr (:) []
1767-- Convert a stream into a list in the underlying monad. The list can be
1768-- consumed lazily in a lazy monad (e.g. 'Identity'). In a strict monad (e.g.
1769-- IO) the whole list is generated and buffered before it can be consumed.
1771-- /Warning!/ working on large lists accumulated as buffers in memory could be
1772-- very inefficient, consider using "Streamly.Array" instead.
1775{-# INLINE toList #-}
1776toList :: Monad m => SerialT m a -> m [a]
1781-- toListRev = S.foldl' (flip (:)) []
1784-- Convert a stream into a list in reverse order in the underlying monad.
1786-- /Warning!/ working on large lists accumulated as buffers in memory could be
1787-- very inefficient, consider using "Streamly.Array" instead.
1790{-# INLINE _toListRev #-}
1791_toListRev :: Monad m => SerialT m a -> m [a]
1792_toListRev = D.toListRev . toStreamD
1796-- toHandle h = S.mapM_ $ hPutStrLn h
1799-- Write a stream of Strings to an IO Handle.
1802{-# DEPRECATED toHandle
1803 "Please use Streamly.FileSystem.Handle module (see the changelog)" #-}
1804toHandle :: MonadIO m => IO.Handle -> SerialT m String -> m ()
1808 let stop = return ()
1809 single a = liftIO (IO.hPutStrLn h a)
1810 yieldk a r = liftIO (IO.hPutStrLn h a) >> go r
1811 in K.foldStream defState yieldk single stop m1
1813-- | A fold that buffers its input to a pure stream.
1815-- /Warning!/ working on large streams accumulated as buffers in memory could
1816-- be very inefficient, consider using "Streamly.Array" instead.
1819{-# INLINE _toStream #-}
1820_toStream :: Monad m => Fold m a (SerialT Identity a)
1821_toStream = Fold (\f x -> return $ f . (x `K.cons`))
1823 (return . ($ K.nil))
1825-- This is more efficient than 'toStream'. toStream is exactly the same as
1826-- reversing the stream after toStreamRev.
1828-- | Buffers the input stream to a pure stream in the reverse order of the
1831-- /Warning!/ working on large streams accumulated as buffers in memory could
1832-- be very inefficient, consider using "Streamly.Array" instead.
1836-- xn : ... : x2 : x1 : []
1837{-# INLINABLE _toStreamRev #-}
1838_toStreamRev :: Monad m => Fold m a (SerialT Identity a)
1839_toStreamRev = Fold (\xs x -> return $ x `K.cons` xs) (return K.nil) return
1841------------------------------------------------------------------------------
1842-- General Transformation
1843------------------------------------------------------------------------------
1845-- | Use a 'Pipe' to transform a stream.
1846{-# INLINE _transform #-}
1847_transform :: (IsStream t, Monad m) => Pipe m a b -> t m a -> t m b
1848_transform pipe xs = fromStreamD $ D.transform pipe (toStreamD xs)
1850------------------------------------------------------------------------------
1851-- Transformation by Folding (Scans)
1852------------------------------------------------------------------------------
1854-- XXX It may be useful to have a version of scan where we can keep the
1855-- accumulator independent of the value emitted. So that we do not necessarily
1856-- have to keep a value in the accumulator which we are not using. We can pass
1857-- an extraction function that will take the accumulator and the current value
1858-- of the element and emit the next value in the stream. That will also make it
1859-- possible to modify the accumulator after using it. In fact, the step function
1860-- can return new accumulator and the value to be emitted. The signature would
1861-- be more like mapAccumL. Or we can change the signature of scanx to
1864-- | Strict left scan with an extraction function. Like 'scanl'', but applies a
1865-- user supplied extraction function (the third argument) at each step. This is
1866-- designed to work with the @foldl@ library. The suffix @x@ is a mnemonic for
1869-- /Since: 0.7.0 (Monad m constraint)/
1872{-# DEPRECATED scanx "Please use scanl followed by map instead." #-}
1874scanx :: (IsStream t, Monad m) => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
1877-- XXX this needs to be concurrent
1878-- | Like 'scanl'' but with a monadic fold function.
1881{-# INLINE scanlM' #-}
1882scanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> t m b
1883scanlM' step begin m = fromStreamD $ D.scanlM' step begin $ toStreamD m
1885-- | Strict left scan. Like 'map', 'scanl'' too is a one to one transformation,
1886-- however it adds an extra element.
1889-- > S.toList $ S.scanl' (+) 0 $ fromList [1,2,3,4]
1894-- > S.toList $ S.scanl' (flip (:)) [] $ S.fromList [1,2,3,4]
1895-- [[],[1],[2,1],[3,2,1],[4,3,2,1]]
1898-- The output of 'scanl'' is the initial value of the accumulator followed by
1899-- all the intermediate steps and the final result of 'foldl''.
1901-- By streaming the accumulated state after each fold step, we can share the
1902-- state across multiple stages of stream composition. Each stage can modify or
1903-- extend the state, do some processing with it and emit it for the next stage,
1904-- thus modularizing the stream processing. This can be useful in
1905-- stateful or event-driven programming.
1907-- Consider the following monolothic example, computing the sum and the product
1908-- of the elements in a stream in one go using a @foldl'@:
1911-- > S.foldl' (\\(s, p) x -> (s + x, p * x)) (0,1) $ S.fromList \[1,2,3,4]
1915-- Using @scanl'@ we can make it modular by computing the sum in the first
1916-- stage and passing it down to the next stage for computing the product:
1919-- > S.foldl' (\\(_, p) (s, x) -> (s, p * x)) (0,1)
1920-- $ S.scanl' (\\(s, _) x -> (s + x, x)) (0,1)
1921-- $ S.fromList \[1,2,3,4]
1925-- IMPORTANT: 'scanl'' evaluates the accumulator to WHNF. To avoid building
1926-- lazy expressions inside the accumulator, it is recommended that a strict
1927-- data structure is used for accumulator.
1930{-# INLINE scanl' #-}
1931scanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
1932scanl' step z m = fromStreamS $ S.scanl' step z $ toStreamS m
1934-- | Like 'scanl'' but does not stream the initial value of the accumulator.
1936-- > postscanl' f z xs = S.drop 1 $ S.scanl' f z xs
1939{-# INLINE postscanl' #-}
1940postscanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
1941postscanl' step z m = fromStreamD $ D.postscanl' step z $ toStreamD m
1943-- XXX this needs to be concurrent
1944-- | Like 'postscanl'' but with a monadic step function.
1947{-# INLINE postscanlM' #-}
1948postscanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> t m b
1949postscanlM' step z m = fromStreamD $ D.postscanlM' step z $ toStreamD m
1951-- XXX prescanl does not sound very useful, enable only if there is a
1952-- compelling use case.
1954-- | Like scanl' but does not stream the final value of the accumulator.
1957{-# INLINE _prescanl' #-}
1958_prescanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
1959_prescanl' step z m = fromStreamD $ D.prescanl' step z $ toStreamD m
1961-- XXX this needs to be concurrent
1962-- | Like postscanl' but with a monadic step function.
1965{-# INLINE _prescanlM' #-}
1966_prescanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
1967_prescanlM' step z m = fromStreamD $ D.prescanlM' step z $ toStreamD m
1969-- XXX this needs to be concurrent
1970-- | Like 'scanl1'' but with a monadic step function.
1973{-# INLINE scanl1M' #-}
1974scanl1M' :: (IsStream t, Monad m) => (a -> a -> m a) -> t m a -> t m a
1975scanl1M' step m = fromStreamD $ D.scanl1M' step $ toStreamD m
1977-- | Like 'scanl'' but for a non-empty stream. The first element of the stream
1978-- is used as the initial value of the accumulator. Does nothing if the stream
1982-- > S.toList $ S.scanl1 (+) $ fromList [1,2,3,4]
1987{-# INLINE scanl1' #-}
1988scanl1' :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> t m a
1989scanl1' step m = fromStreamD $ D.scanl1' step $ toStreamD m
1991------------------------------------------------------------------------------
1992-- Scanning with a Fold
1993------------------------------------------------------------------------------
1995-- | Scan a stream using the given monadic fold.
1998{-# INLINE runScan #-}
1999runScan :: Monad m => Fold m a b -> SerialT m a -> SerialT m b
2000runScan (Fold step begin done) = P.scanlMx' step begin done
2002-- | Postscan a stream using the given monadic fold.
2005{-# INLINE runPostscan #-}
2006runPostscan :: Monad m => Fold m a b -> SerialT m a -> SerialT m b
2007runPostscan (Fold step begin done) = P.postscanlMx' step begin done
2011------------------------------------------------------------------------------
2012-- Transformation by Filtering
2013------------------------------------------------------------------------------
2015-- | Include only those elements that pass a predicate.
2018{-# INLINE filter #-}
2019#if __GLASGOW_HASKELL__ != 802
2020-- GHC 8.2.2 crashes with this code, when used with "stack"
2021filter :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
2022filter p m = fromStreamS $ S.filter p $ toStreamS m
2024filter :: IsStream t => (a -> Bool) -> t m a -> t m a
2028-- | Same as 'filter' but with a monadic predicate.
2031{-# INLINE filterM #-}
2032filterM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
2033filterM p m = fromStreamD $ D.filterM p $ toStreamD m
2035-- | Drop repeated elements that are adjacent to each other.
2039uniq :: (Eq a, IsStream t, Monad m) => t m a -> t m a
2040uniq = fromStreamD . D.uniq . toStreamD
2042-- | Ensures that all the elements of the stream are identical and then returns
2043-- that unique element.
2047the :: (Eq a, Monad m) => SerialT m a -> m (Maybe a)
2048the m = S.the (toStreamS m)
2050-- | Take first 'n' elements from the stream and discard the rest.
2054take :: (IsStream t, Monad m) => Int -> t m a -> t m a
2055take n m = fromStreamS $ S.take n $ toStreamS
2056 (maxYields (Just (fromIntegral n)) m)
2058-- | End the stream as soon as the predicate fails on an element.
2061{-# INLINE takeWhile #-}
2062takeWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
2063takeWhile p m = fromStreamS $ S.takeWhile p $ toStreamS m
2065-- | Same as 'takeWhile' but with a monadic predicate.
2068{-# INLINE takeWhileM #-}
2069takeWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
2070takeWhileM p m = fromStreamD $ D.takeWhileM p $ toStreamD m
2072-- | Discard first 'n' elements from the stream and take the rest.
2076drop :: (IsStream t, Monad m) => Int -> t m a -> t m a
2077drop n m = fromStreamS $ S.drop n $ toStreamS m
2079-- | Drop elements in the stream as long as the predicate succeeds and then
2080-- take the rest of the stream.
2083{-# INLINE dropWhile #-}
2084dropWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
2085dropWhile p m = fromStreamS $ S.dropWhile p $ toStreamS m
2087-- | Same as 'dropWhile' but with a monadic predicate.
2090{-# INLINE dropWhileM #-}
2091dropWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
2092dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m
2094------------------------------------------------------------------------------
2095-- Transformation by Mapping
2096------------------------------------------------------------------------------
2100-- mapM f = sequence . map f
2103-- Apply a monadic function to each element of the stream and replace it with
2104-- the output of the resulting action.
2107-- > drain $ S.mapM putStr $ S.fromList ["a", "b", "c"]
2110-- drain $ S.replicateM 10 (return 1)
2111-- & (serially . S.mapM (\\x -> threadDelay 1000000 >> print x))
2113-- drain $ S.replicateM 10 (return 1)
2114-- & (asyncly . S.mapM (\\x -> threadDelay 1000000 >> print x))
2117-- /Concurrent (do not use with 'parallely' on infinite streams)/
2120{-# INLINE_EARLY mapM #-}
2121mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
2124{-# RULES "mapM serial" mapM = mapMSerial #-}
2125{-# INLINE mapMSerial #-}
2126mapMSerial :: Monad m => (a -> m b) -> SerialT m a -> SerialT m b
2127mapMSerial = Serial.mapM
2131-- sequence = mapM id
2134-- Replace the elements of a stream of monadic actions with the outputs of
2138-- > drain $ S.sequence $ S.fromList [putStr "a", putStr "b", putStrLn "c"]
2141-- drain $ S.replicateM 10 (return $ threadDelay 1000000 >> print 1)
2142-- & (serially . S.sequence)
2144-- drain $ S.replicateM 10 (return $ threadDelay 1000000 >> print 1)
2145-- & (asyncly . S.sequence)
2148-- /Concurrent (do not use with 'parallely' on infinite streams)/
2151{-# INLINE sequence #-}
2152sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
2153sequence m = fromStreamS $ S.sequence (toStreamS m)
2155------------------------------------------------------------------------------
2156-- Transformation by Map and Filter
2157------------------------------------------------------------------------------
2159-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'
2160-- elements, and return a stream of values extracted from 'Just'.
2163{-# INLINE mapMaybe #-}
2164mapMaybe :: (IsStream t, Monad m) => (a -> Maybe b) -> t m a -> t m b
2165mapMaybe f m = fromStreamS $ S.mapMaybe f $ toStreamS m
2167-- | Like 'mapMaybe' but maps a monadic function.
2169-- /Concurrent (do not use with 'parallely' on infinite streams)/
2172{-# INLINE_EARLY mapMaybeM #-}
2173mapMaybeM :: (IsStream t, MonadAsync m, Functor (t m))
2174 => (a -> m (Maybe b)) -> t m a -> t m b
2175mapMaybeM f = fmap fromJust . filter isJust . K.mapM f
2177{-# RULES "mapMaybeM serial" mapMaybeM = mapMaybeMSerial #-}
2178{-# INLINE mapMaybeMSerial #-}
2179mapMaybeMSerial :: Monad m => (a -> m (Maybe b)) -> SerialT m a -> SerialT m b
2180mapMaybeMSerial f m = fromStreamD $ D.mapMaybeM f $ toStreamD m
2182------------------------------------------------------------------------------
2183-- Transformation by Reordering
2184------------------------------------------------------------------------------
2186-- XXX Use a compact region list to temporarily store the list, in both reverse
2187-- as well as in reverse'.
2189-- /Note:/ 'reverse'' is much faster than this, use that when performance
2193-- > reverse = S.foldlT (flip S.cons) S.nil
2195-- Returns the elements of the stream in reverse order. The stream must be
2196-- finite. Note that this necessarily buffers the entire stream in memory.
2198-- /Since 0.7.0 (Monad m constraint)/
2201{-# INLINE reverse #-}
2202reverse :: (IsStream t, Monad m) => t m a -> t m a
2203reverse s = fromStreamS $ S.reverse $ toStreamS s
2205-- | Like 'reverse' but several times faster, requires a 'Storable' instance.
2208{-# INLINE _reverse' #-}
2209_reverse' :: (IsStream t, MonadIO m, Storable a) => t m a -> t m a
2210_reverse' s = fromStreamD $ D.reverse' $ toStreamD s
2212------------------------------------------------------------------------------
2213-- Transformation by Inserting
2214------------------------------------------------------------------------------
2216-- intersperseM = intersperseBySpan 1
2218-- | Generate a stream by performing a monadic action between consecutive
2219-- elements of the given stream.
2221-- /Concurrent (do not use with 'parallely' on infinite streams)/
2224-- > S.toList $ S.intersperseM (return ',') $ S.fromList "hello"
2229{-# INLINE intersperseM #-}
2230intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
2231intersperseM m = fromStreamS . S.intersperseM m . toStreamS
2233-- | Generate a stream by inserting a given element between consecutive
2234-- elements of the given stream.
2237-- > S.toList $ S.intersperse ',' $ S.fromList "hello"
2242{-# INLINE intersperse #-}
2243intersperse :: (IsStream t, MonadAsync m) => a -> t m a -> t m a
2244intersperse a = fromStreamS . S.intersperse a . toStreamS
2246-- | Insert a monadic action after each element in the stream.
2249{-# INLINE _insertAfterEach #-}
2250_insertAfterEach :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
2251_insertAfterEach m = fromStreamD . D.insertAfterEach m . toStreamD
2254-- | Intersperse a monadic action into the input stream after every @n@
2258-- > S.toList $ S.intersperseBySpan 2 (return ',') $ S.fromList "hello"
2263{-# INLINE intersperseBySpan #-}
2264intersperseBySpan :: IsStream t => Int -> m a -> t m a -> t m a
2265intersperseBySpan _n _f _xs = undefined
2268-- | Intersperse a monadic action into the input stream after every @n@
2272-- > S.drain $ S.intersperseByTime 1 (putChar ',') $ S.mapM (\\x -> threadDelay 1000000 >> putChar x) $ S.fromList "hello"
2277{-# INLINE intersperseByTime #-}
2279 :: (IsStream t, MonadAsync m)
2280 => Double -> m a -> t m a -> t m a
2281intersperseByTime n f xs = xs `Par.parallelEndByFirst` repeatM timed
2282 where timed = liftIO (threadDelay (round $ n * 1000000)) >> f
2284-- | @insertBy cmp elem stream@ inserts @elem@ before the first element in
2285-- @stream@ that is less than @elem@ when compared using @cmp@.
2288-- insertBy cmp x = 'mergeBy' cmp ('yield' x)
2292-- > S.toList $ S.insertBy compare 2 $ S.fromList [1,3,5]
2297{-# INLINE insertBy #-}
2299 (IsStream t, Monad m) => (a -> a -> Ordering) -> a -> t m a -> t m a
2300insertBy cmp x m = fromStreamS $ S.insertBy cmp x (toStreamS m)
2302------------------------------------------------------------------------------
2304------------------------------------------------------------------------------
2306-- | Deletes the first occurence of the element in the stream that satisfies
2307-- the given equality predicate.
2310-- > S.toList $ S.deleteBy (==) 3 $ S.fromList [1,3,3,5]
2315{-# INLINE deleteBy #-}
2316deleteBy :: (IsStream t, Monad m) => (a -> a -> Bool) -> a -> t m a -> t m a
2317deleteBy cmp x m = fromStreamS $ S.deleteBy cmp x (toStreamS m)
2319------------------------------------------------------------------------------
2321------------------------------------------------------------------------------
2324-- > indexed = S.postscanl' (\(i, _) x -> (i + 1, x)) (-1,undefined)
2325-- > indexed = S.zipWith (,) (S.enumerateFrom 0)
2327-- Pair each element in a stream with its index, starting from index 0.
2330-- > S.toList $ S.indexed $ S.fromList "hello"
2331-- [(0,'h'),(1,'e'),(2,'l'),(3,'l'),(4,'o')]
2335{-# INLINE indexed #-}
2336indexed :: (IsStream t, Monad m) => t m a -> t m (Int, a)
2337indexed = fromStreamD . D.indexed . toStreamD
2340-- > indexedR n = S.postscanl' (\(i, _) x -> (i - 1, x)) (n + 1,undefined)
2341-- > indexedR n = S.zipWith (,) (S.enumerateFromThen n (n - 1))
2343-- Pair each element in a stream with its index, starting from the
2344-- given index @n@ and counting down.
2347-- > S.toList $ S.indexedR 10 $ S.fromList "hello"
2348-- [(10,'h'),(9,'e'),(8,'l'),(7,'l'),(6,'o')]
2352{-# INLINE indexedR #-}
2353indexedR :: (IsStream t, Monad m) => Int -> t m a -> t m (Int, a)
2354indexedR n = fromStreamD . D.indexedR n . toStreamD
2356-- | Like 'zipWith' but using a monadic zipping function.
2359{-# INLINABLE zipWithM #-}
2360zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c
2361zipWithM f m1 m2 = fromStreamS $ S.zipWithM f (toStreamS m1) (toStreamS m2)
2363-- | Zip two streams serially using a pure zipping function.
2366-- > S.toList $ S.zipWith (+) (S.fromList [1,2,3]) (S.fromList [4,5,6])
2371{-# INLINABLE zipWith #-}
2372zipWith :: (IsStream t, Monad m) => (a -> b -> c) -> t m a -> t m b -> t m c
2373zipWith f m1 m2 = fromStreamS $ S.zipWith f (toStreamS m1) (toStreamS m2)
2375------------------------------------------------------------------------------
2377------------------------------------------------------------------------------
2379-- | Compare two streams for equality using an equality function.
2382{-# INLINABLE eqBy #-}
2383eqBy :: (IsStream t, Monad m) => (a -> b -> Bool) -> t m a -> t m b -> m Bool
2386-- | Compare two streams lexicographically using a comparison function.
2389{-# INLINABLE cmpBy #-}
2391 :: (IsStream t, Monad m)
2392 => (a -> b -> Ordering) -> t m a -> t m b -> m Ordering
2395------------------------------------------------------------------------------
2397------------------------------------------------------------------------------
2399-- | Merge two streams using a comparison function. The head elements of both
2400-- the streams are compared and the smaller of the two elements is emitted, if
2401-- both elements are equal then the element from the first stream is used
2404-- If the streams are sorted in ascending order, the resulting stream would
2405-- also remain sorted in ascending order.
2408-- > S.toList $ S.mergeBy compare (S.fromList [1,3,5]) (S.fromList [2,4,6,8])
2413{-# INLINABLE mergeBy #-}
2415 (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> t m a -> t m a
2416mergeBy f m1 m2 = fromStreamS $ S.mergeBy f (toStreamS m1) (toStreamS m2)
2418-- | Like 'mergeBy' but with a monadic comparison function.
2420-- Merge two streams randomly:
2423-- > randomly _ _ = randomIO >>= \x -> return $ if x then LT else GT
2424-- > S.toList $ S.mergeByM randomly (S.fromList [1,1,1,1]) (S.fromList [2,2,2,2])
2428-- Merge two streams in a proportion of 2:1:
2431-- proportionately m n = do
2432-- ref <- newIORef $ cycle $ concat [replicate m LT, replicate n GT]
2433-- return $ \\_ _ -> do
2434-- r <- readIORef ref
2435-- writeIORef ref $ tail r
2439-- f <- proportionately 2 1
2440-- xs <- S.toList $ S.mergeByM f (S.fromList [1,1,1,1,1,1]) (S.fromList [2,2,2])
2444-- [1,1,2,1,1,2,1,1,2]
2448{-# INLINABLE mergeByM #-}
2450 :: (IsStream t, Monad m)
2451 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
2452mergeByM f m1 m2 = fromStreamS $ S.mergeByM f (toStreamS m1) (toStreamS m2)
2455-- | Like 'mergeByM' but stops merging as soon as any of the two streams stops.
2456{-# INLINABLE mergeUptoShortest #-}
2458 :: (IsStream t, Monad m)
2459 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
2460mergeEndByAny f m1 m2 = fromStreamD $
2461 D.mergeEndByAny f (toStreamD m1) (toStreamD m2)
2463-- Like 'mergeByM' but stops merging as soon as the first stream stops.
2464{-# INLINABLE mergeEndByFirst #-}
2466 :: (IsStream t, Monad m)
2467 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
2468mergeEndByFirst f m1 m2 = fromStreamS $
2469 D.mergeEndByFirst f (toStreamD m1) (toStreamD m2)
2472-- Holding this back for now, we may want to use the name "merge" differently
2474-- | Same as @'mergeBy' 'compare'@.
2477-- > S.toList $ S.merge (S.fromList [1,3,5]) (S.fromList [2,4,6,8])
2482{-# INLINABLE merge #-}
2484 (IsStream t, Monad m, Ord a) => t m a -> t m a -> t m a
2485merge = mergeBy compare
2488-- | Like 'mergeBy' but merges concurrently (i.e. both the elements being
2489-- merged are generated concurrently).
2492mergeAsyncBy :: (IsStream t, MonadAsync m)
2493 => (a -> a -> Ordering) -> t m a -> t m a -> t m a
2494mergeAsyncBy f m1 m2 = K.mkStream $ \st stp sng yld -> do
2495 ma <- mkAsync' st m1
2496 mb <- mkAsync' st m2
2497 K.foldStream st stp sng yld (K.mergeBy f ma mb)
2499-- | Like 'mergeByM' but merges concurrently (i.e. both the elements being
2500-- merged are generated concurrently).
2503mergeAsyncByM :: (IsStream t, MonadAsync m)
2504 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
2505mergeAsyncByM f m1 m2 = K.mkStream $ \st stp sng yld -> do
2506 ma <- mkAsync' st m1
2507 mb <- mkAsync' st m2
2508 K.foldStream st stp sng yld (K.mergeByM f ma mb)
2510------------------------------------------------------------------------------
2512------------------------------------------------------------------------------
2514-- | Map each element of the stream to produce a stream and then flatten the
2515-- results into a single stream.
2517-- > concatMap f = concatMapM (return . f)
2519-- In imperative terms, a 'concatMap' corresponds to nested loops. We loop over
2520-- the input stream and then for each element of the input stream another
2521-- stream is generated and then we loop over that inner stream as well to yield
2522-- each element of that stream to generate a single output stream.
2524-- 'concatMap' can perform filtering by mapping an element to a 'nil'
2527-- > filter p m = S.concatMap (\x -> if p x then S.yield x else S.nil) m
2529-- Its a map, therefore it can degenerate to a simple map operation as well:
2531-- > map f m = S.concatMap (\x -> S.yield (f x)) m
2533-- 'concatMap' is not inherently stateful, however, it can be combined with a
2534-- scan to perform stateful operations.
2537{-# INLINE concatMap #-}
2538concatMap ::(IsStream t, Monad m) => (a -> t m b) -> t m a -> t m b
2539concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)
2541-- | Map each element to a stream to produce a stream using a monadic function
2542-- and then flatten the results into a single stream.
2545{-# INLINE concatMapM #-}
2546concatMapM :: (IsStream t, Monad m) => (a -> m (t m b)) -> t m a -> t m b
2547concatMapM f m = fromStreamD $ D.concatMapM (fmap toStreamD . f) (toStreamD m)
2550-- | A generalization of insertBy to inserting sequences.
2551interposeBy :: (IsStream t, Monad m)
2552 => (a -> a -> Ordering) -> t m a -> t m a -> t m a
2554-- | A generalization of intersperseM to intersperse sequences.
2555intercalate :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
2558------------------------------------------------------------------------------
2559-- Grouping/Splitting
2560------------------------------------------------------------------------------
2562------------------------------------------------------------------------------
2563-- Grouping without looking at elements
2564------------------------------------------------------------------------------
2566------------------------------------------------------------------------------
2568------------------------------------------------------------------------------
2571-- | @splitAt n f1 f2@ composes folds @f1@ and @f2@ such that first @n@
2572-- elements of its input are consumed by fold @f1@ and the rest of the stream
2573-- is consumed by fold @f2@.
2575-- > let splitAt_ n xs = S.runFold (FL.splitAt n FL.toList FL.toList) $ S.fromList xs
2577-- >>> splitAt_ 6 "Hello World!"
2578-- > ("Hello ","World!")
2580-- >>> splitAt_ (-1) [1,2,3]
2583-- >>> splitAt_ 0 [1,2,3]
2586-- >>> splitAt_ 1 [1,2,3]
2589-- >>> splitAt_ 3 [1,2,3]
2592-- >>> splitAt_ 4 [1,2,3]
2595-- This can be considered as a two-fold version of 'ltake' where we take both
2596-- the segments instead of discarding the leftover.
2599{-# INLINE _splitAt #-}
2606_splitAt n (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
2607 Fold step initial extract
2609 initial = Tuple3' <$> return n <*> initialL <*> initialR
2611 step (Tuple3' i xL xR) input =
2613 then stepL xL input >>= (\a -> return (Tuple3' (i - 1) a xR))
2614 else stepR xR input >>= (\b -> return (Tuple3' i xL b))
2616 extract (Tuple3' _ a b) = (,) <$> extractL a <*> extractR b
2618------------------------------------------------------------------------------
2620------------------------------------------------------------------------------
2622------------------------------------------------------------------------------
2623-- Generalized grouping
2624------------------------------------------------------------------------------
2626-- This combinator is the most general grouping combinator and can be used to
2627-- implement all other grouping combinators.
2629-- XXX check if this can implement the splitOn combinator i.e. we can slide in
2630-- new elements, slide out old elements and incrementally compute the hash.
2631-- Also, can we implement the windowed classification combinators using this?
2633-- In fact this is a parse. Instead of using a special return value in the fold
2634-- we are using a mapping function.
2636-- Note that 'scanl'' (usually followed by a map to extract the desired value
2637-- from the accumulator) can be used to realize many implementations e.g. a
2638-- sliding window implementation. A scan followed by a mapMaybe is also a good
2639-- pattern to express many problems where we want to emit a filtered output and
2640-- not emit an output on every input.
2642-- Passing on of the initial accumulator value to the next fold is equivalent
2643-- to returning the leftover concept.
2646-- | @groupScan splitter fold stream@ folds the input stream using @fold@.
2647-- @splitter@ is applied on the accumulator of the fold every time an item is
2648-- consumed by the fold. The fold continues until @splitter@ returns a 'Just'
2649-- value. A 'Just' result from the @splitter@ specifies a result to be emitted
2650-- in the output stream and the initial value of the accumulator for the next
2651-- group's fold. This allows us to control whether to start fresh for the next
2652-- fold or to continue from the previous fold's output.
2654{-# INLINE groupScan #-}
2656 :: (IsStream t, Monad m)
2657 => (x -> m (Maybe (b, x))) -> Fold m a x -> t m a -> t m b
2658groupScan split fold m = undefined
2661-- | Group the input stream into groups of @n@ elements each and then fold each
2662-- group using the provided fold function.
2664-- >> S.toList $ S.chunksOf 2 FL.sum (S.enumerateFromTo 1 10)
2667-- This can be considered as an n-fold version of 'ltake' where we apply
2668-- 'ltake' repeatedly on the leftover stream until the stream exhausts.
2671{-# INLINE chunksOf #-}
2673 :: (IsStream t, Monad m)
2674 => Int -> Fold m a b -> t m a -> t m b
2675chunksOf n f m = D.fromStreamD $ D.groupsOf n f (D.toStreamD m)
2677-- | Transform a fold from a pure input to a 'Maybe' input, consuming only
2679{-# INLINE lcatMaybes #-}
2680lcatMaybes :: Monad m => Fold m a b -> Fold m (Maybe a) b
2681lcatMaybes = FL.lfilter isJust . FL.lmap fromJust
2683-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
2684-- XXX add this example after fixing the serial stream rate control
2685-- >>> S.toList $ S.take 5 $ sessionsOf 1 FL.sum $ constRate 2 $ S.enumerateFrom 1
2688-- | Group the input stream into windows of @n@ second each and then fold each
2689-- group using the provided fold function.
2692{-# INLINE sessionsOf #-}
2694 :: (IsStream t, MonadAsync m)
2695 => Double -> Fold m a b -> t m a -> t m b
2697 splitBySuffix isNothing (lcatMaybes f)
2698 (intersperseByTime n (return Nothing) (Serial.map Just xs))
2700------------------------------------------------------------------------------
2701-- Element Aware APIs
2702------------------------------------------------------------------------------
2704------------------------------------------------------------------------------
2706------------------------------------------------------------------------------
2708-- | Break the input stream into two groups, the first group takes the input as
2709-- long as the predicate applied to the first element of the stream and next
2710-- input element holds 'True', the second group takes the rest of the input.
2718_spanBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
2719 Fold step initial extract
2722 initial = Tuple3' <$> initialL <*> initialR <*> return (Tuple' Nothing True)
2724 step (Tuple3' a b (Tuple' (Just frst) isFirstG)) input =
2725 if cmp frst input && isFirstG
2727 >>= (\a' -> return (Tuple3' a' b (Tuple' (Just frst) isFirstG)))
2729 >>= (\a' -> return (Tuple3' a a' (Tuple' Nothing False)))
2731 step (Tuple3' a b (Tuple' Nothing isFirstG)) input =
2734 >>= (\a' -> return (Tuple3' a' b (Tuple' (Just input) isFirstG)))
2736 >>= (\a' -> return (Tuple3' a a' (Tuple' Nothing False)))
2738 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
2740-- | @span p f1 f2@ composes folds @f1@ and @f2@ such that @f1@ consumes the
2741-- input as long as the predicate @p@ is 'True'. @f2@ consumes the rest of the
2744-- > let span_ p xs = S.runFold (S.span p FL.toList FL.toList) $ S.fromList xs
2746-- >>> span_ (< 1) [1,2,3]
2749-- >>> span_ (< 2) [1,2,3]
2752-- >>> span_ (< 4) [1,2,3]
2757-- This can be considered as a two-fold version of 'ltakeWhile' where we take
2758-- both the segments instead of discarding the leftover.
2766span p (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
2767 Fold step initial extract
2771 initial = Tuple3' <$> initialL <*> initialR <*> return True
2773 step (Tuple3' a b isFirstG) input =
2774 if isFirstG && p input
2775 then stepL a input >>= (\a' -> return (Tuple3' a' b True))
2776 else stepR b input >>= (\a' -> return (Tuple3' a a' False))
2778 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
2781-- > break p = span (not . p)
2783-- Break as soon as the predicate becomes 'True'. @break p f1 f2@ composes
2784-- folds @f1@ and @f2@ such that @f1@ stops consuming input as soon as the
2785-- predicate @p@ becomes 'True'. The rest of the input is consumed @f2@.
2787-- This is the binary version of 'splitBy'.
2789-- > let break_ p xs = S.runFold (S.break p FL.toList FL.toList) $ S.fromList xs
2791-- >>> break_ (< 1) [3,2,1]
2794-- >>> break_ (< 2) [3,2,1]
2797-- >>> break_ (< 4) [3,2,1]
2801{-# INLINE _break #-}
2808_break p = span (not . p)
2810-- | Like 'spanBy' but applies the predicate in a rolling fashion i.e.
2811-- predicate is applied to the previous and the next input elements.
2812{-# INLINE _spanRollingBy #-}
2819_spanRollingBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
2820 Fold step initial extract
2823 initial = Tuple3' <$> initialL <*> initialR <*> return Nothing
2825 step (Tuple3' a b (Just frst)) input =
2827 then stepL a input >>= (\a' -> return (Tuple3' a' b (Just input)))
2828 else stepR b input >>= (\b' -> return (Tuple3' a b' (Just input)))
2830 step (Tuple3' a b Nothing) input =
2831 stepL a input >>= (\a' -> return (Tuple3' a' b (Just input)))
2833 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
2835------------------------------------------------------------------------------
2837------------------------------------------------------------------------------
2839-- | @groupsBy cmp f $ S.fromList [a,b,c,...]@ assigns the element @a@ to the
2840-- first group, if @a \`cmp` b@ is 'True' then @b@ is also assigned to the same
2841-- group. If @a \`cmp` c@ is 'True' then @c@ is also assigned to the same
2842-- group and so on. When the comparison fails a new group is started. Each
2843-- group is folded using the fold @f@ and the result of the fold is emitted in
2844-- the output stream.
2846-- >>> S.toList $ S.groupsBy (>) FL.toList $ S.fromList [1,3,7,0,2,5]
2847-- > [[1,3,7],[0,2,5]]
2850{-# INLINE groupsBy #-}
2852 :: (IsStream t, Monad m)
2857groupsBy cmp f m = D.fromStreamD $ D.groupsBy cmp f (D.toStreamD m)
2859-- | Unlike @groupsBy@ this function performs a rolling comparison of two
2860-- successive elements in the input stream. @groupsByRolling cmp f $ S.fromList
2861-- [a,b,c,...]@ assigns the element @a@ to the first group, if @a \`cmp` b@ is
2862-- 'True' then @b@ is also assigned to the same group. If @b \`cmp` c@ is
2863-- 'True' then @c@ is also assigned to the same group and so on. When the
2864-- comparison fails a new group is started. Each group is folded using the fold
2867-- >>> S.toList $ S.groupsByRolling (\a b -> a + 1 == b) FL.toList $ S.fromList [1,2,3,7,8,9]
2868-- > [[1,2,3],[7,8,9]]
2871{-# INLINE groupsByRolling #-}
2873 :: (IsStream t, Monad m)
2878groupsByRolling cmp f m = D.fromStreamD $ D.groupsRollingBy cmp f (D.toStreamD m)
2881-- > groups = groupsBy (==)
2882-- > groups = groupsByRolling (==)
2884-- Groups contiguous spans of equal elements together in individual groups.
2886-- >>> S.toList $ S.groups FL.toList $ S.fromList [1,1,2,2]
2890groups :: (IsStream t, Monad m, Eq a) => Fold m a b -> t m a -> t m b
2891groups = groupsBy (==)
2893------------------------------------------------------------------------------
2894-- Binary splitting on a separator
2895------------------------------------------------------------------------------
2898-- | Find the first occurrence of the specified sequence in the input stream
2899-- and break the input stream into two parts, the first part consisting of the
2900-- stream before the sequence and the second part consisting of the sequence
2901-- and the rest of the stream.
2903-- > let breakOn_ pat xs = S.runFold (S.breakOn pat FL.toList FL.toList) $ S.fromList xs
2905-- >>> breakOn_ "dear" "Hello dear world!"
2906-- > ("Hello ","dear world!")
2908{-# INLINE breakOn #-}
2909breakOn :: Monad m => Array a -> Fold m a b -> Fold m a c -> Fold m a (b,c)
2910breakOn pat f m = undefined
2913------------------------------------------------------------------------------
2914-- N-ary split on a predicate
2915------------------------------------------------------------------------------
2917-- TODO: Use a Splitter configuration similar to the "split" package to make it
2918-- possible to express all splitting combinations. In general, we can have
2919-- infix/suffix/prefix/condensing of separators, dropping both leading/trailing
2920-- separators. We can have a single split operation taking the splitter config
2923-- | Split a stream on separator elements determined by a predicate, dropping
2924-- the separator. Separator is considered as infixed between two segments, if
2925-- one side of the separator is missing then it is parsed as an empty stream.
2926-- With '-' representing elements and '.' as separator, 'splitOn' splits as
2930-- "--.--" => "--" "--"
2935-- @splitOn (== x)@ is an inverse of @intercalate (S.yield x)@
2937-- Let's use the following definition for illustration:
2939-- > splitOn' p xs = S.toList $ S.splitOn p (FL.toList) (S.fromList xs)
2941-- >>> splitOn' (== '.') ""
2944-- >>> splitOn' (== '.') "."
2947-- >>> splitOn' (== '.') ".a"
2950-- >>> splitOn' (== '.') "a."
2953-- >>> splitOn' (== '.') "a.b"
2956-- >>> splitOn' (== '.') "a..b"
2961-- This can be considered as an n-fold version of 'breakOn' where we apply
2962-- 'breakOn' successively on the input stream, dropping the first element
2963-- of the second segment after each break.
2965{-# INLINE splitOn #-}
2967 :: (IsStream t, Monad m)
2968 => (a -> Bool) -> Fold m a b -> t m a -> t m b
2969splitOn predicate f m =
2970 D.fromStreamD $ D.splitBy predicate f (D.toStreamD m)
2972-- | Like 'splitOn' but the separator is considered as suffixed to the segments
2973-- in the stream. A missing suffix at the end is allowed. A separator at the
2974-- beginning is parsed as empty segment. With '-' representing elements and
2975-- '.' as separator, 'splitOnSuffix' splits as follows:
2978-- "--.--." => "--" "--"
2979-- "--.--" => "--" "--"
2983-- > splitOnSuffix' p xs = S.toList $ S.splitSuffixBy p (FL.toList) (S.fromList xs)
2985-- >>> splitOnSuffix' (== '.') ""
2988-- >>> splitOnSuffix' (== '.') "."
2991-- >>> splitOnSuffix' (== '.') "a"
2994-- >>> splitOnSuffix' (== '.') ".a"
2997-- >>> splitOnSuffix' (== '.') "a."
3000-- >>> splitOnSuffix' (== '.') "a.b"
3003-- >>> splitOnSuffix' (== '.') "a.b."
3006-- >>> splitOnSuffix' (== '.') "a..b.."
3009-- > lines = splitOnSuffix (== '\n')
3013-- This can be considered as an n-fold version of 'breakPost' where we apply
3014-- 'breakPost' successively on the input stream, dropping the first element
3015-- of the second segment after each break.
3017{-# INLINE splitOnSuffix #-}
3019 :: (IsStream t, Monad m)
3020 => (a -> Bool) -> Fold m a b -> t m a -> t m b
3021splitOnSuffix predicate f m =
3022 D.fromStreamD $ D.splitSuffixBy predicate f (D.toStreamD m)
3024-- | Like 'splitBy' but ignores repeated separators or separators in leading
3025-- or trailing position. Therefore, @"..a..b.."@ would be parsed as
3026-- @["a","b"]@. In other words, it treats the input like words separated by
3027-- whitespace elements determined by the predicate.
3029-- > wordsBy' p xs = S.toList $ S.wordsBy p (FL.toList) (S.fromList xs)
3031-- >>> wordsBy' (== ',') ""
3034-- >>> wordsBy' (== ',') ","
3037-- >>> wordsBy' (== ',') ",a,,b,"
3040-- > words = wordsBy isSpace
3043{-# INLINE wordsBy #-}
3045 :: (IsStream t, Monad m)
3046 => (a -> Bool) -> Fold m a b -> t m a -> t m b
3047wordsBy predicate f m =
3048 D.fromStreamD $ D.wordsBy predicate f (D.toStreamD m)
3050-- | Like 'splitOnSuffix' but keeps the suffix attached to the resulting
3053-- > splitBySuffix' p xs = S.toList $ S.splitBySuffix p (FL.toList) (S.fromList xs)
3055-- >>> splitBySuffix' (== '.') ""
3058-- >>> splitBySuffix' (== '.') "."
3061-- >>> splitBySuffix' (== '.') "a"
3064-- >>> splitBySuffix' (== '.') ".a"
3067-- >>> splitBySuffix' (== '.') "a."
3070-- >>> splitBySuffix' (== '.') "a.b"
3073-- >>> splitBySuffix' (== '.') "a.b."
3076-- >>> splitBySuffix' (== '.') "a..b.."
3077-- > ["a.",".","b.","."]
3081-- This can be considered as an n-fold version of 'breakPost' where we apply
3082-- 'breakPost' successively on the input stream.
3084{-# INLINE splitBySuffix #-}
3086 :: (IsStream t, Monad m)
3087 => (a -> Bool) -> Fold m a b -> t m a -> t m b
3088splitBySuffix predicate f m =
3089 D.fromStreamD $ D.splitSuffixBy' predicate f (D.toStreamD m)
3091------------------------------------------------------------------------------
3092-- Split on a delimiter sequence
3093------------------------------------------------------------------------------
3095-- Int list examples for splitOn:
3097-- >>> splitList [] [1,2,3,3,4]
3098-- > [[1],[2],[3],[3],[4]]
3100-- >>> splitList [5] [1,2,3,3,4]
3103-- >>> splitList [1] [1,2,3,3,4]
3106-- >>> splitList [4] [1,2,3,3,4]
3109-- >>> splitList [2] [1,2,3,3,4]
3112-- >>> splitList [3] [1,2,3,3,4]
3115-- >>> splitList [3,3] [1,2,3,3,4]
3118-- >>> splitList [1,2,3,3,4] [1,2,3,3,4]
3121-- | Like 'splitOn' but the separator is a sequence of elements instead of a
3124-- For illustration, let's define a function that operates on pure lists:
3127-- splitOnSeq' pat xs = S.toList $ S.splitOnSeq (A.fromList pat) (FL.toList) (S.fromList xs)
3130-- >>> splitOnSeq' "" "hello"
3131-- > ["h","e","l","l","o"]
3133-- >>> splitOnSeq' "hello" ""
3136-- >>> splitOnSeq' "hello" "hello"
3139-- >>> splitOnSeq' "x" "hello"
3142-- >>> splitOnSeq' "h" "hello"
3145-- >>> splitOnSeq' "o" "hello"
3148-- >>> splitOnSeq' "e" "hello"
3151-- >>> splitOnSeq' "l" "hello"
3154-- >>> splitOnSeq' "ll" "hello"
3157-- 'splitOnSeq' is an inverse of 'intercalate'. The following law always holds:
3159-- > intercalate . splitOn == id
3161-- The following law holds when the separator is non-empty and contains none of
3162-- the elements present in the input lists:
3164-- > splitOn . intercalate == id
3168-- XXX We can use a polymorphic vector implemented by Array# to represent the
3169-- sequence, that way we can avoid the Storable constraint. If we still need
3170-- Storable Array for performance, we can use a separate splitOnArray API for
3171-- that. We can also have an API where the sequence itself is a lazy stream, so
3172-- that we can search files in files for example.
3173{-# INLINE splitOnSeq #-}
3175 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
3176 => Array a -> Fold m a b -> t m a -> t m b
3177splitOnSeq patt f m = D.fromStreamD $ D.splitOn patt f (D.toStreamD m)
3180-- This can be implemented easily using Rabin Karp
3181-- | Split on any one of the given patterns.
3182{-# INLINE splitOnAny #-}
3184 :: (IsStream t, Monad m, Storable a, Integral a)
3185 => [Array a] -> Fold m a b -> t m a -> t m b
3186splitOnAny subseq f m = undefined -- D.fromStreamD $ D.splitOnAny f subseq (D.toStreamD m)
3189-- | Like 'splitSuffixBy' but the separator is a sequence of elements, instead
3190-- of a predicate for a single element.
3192-- > splitSuffixOn_ pat xs = S.toList $ S.splitSuffixOn (A.fromList pat) (FL.toList) (S.fromList xs)
3194-- >>> splitSuffixOn_ "." ""
3197-- >>> splitSuffixOn_ "." "."
3200-- >>> splitSuffixOn_ "." "a"
3203-- >>> splitSuffixOn_ "." ".a"
3206-- >>> splitSuffixOn_ "." "a."
3209-- >>> splitSuffixOn_ "." "a.b"
3212-- >>> splitSuffixOn_ "." "a.b."
3215-- >>> splitSuffixOn_ "." "a..b.."
3218-- > lines = splitSuffixOn "\n"
3221{-# INLINE _splitOnSuffixSeq #-}
3223 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
3224 => Array a -> Fold m a b -> t m a -> t m b
3225_splitOnSuffixSeq patt f m =
3226 D.fromStreamD $ D.splitSuffixOn False patt f (D.toStreamD m)
3229-- | Like 'splitOn' but drops any empty splits.
3231{-# INLINE wordsOn #-}
3233 :: (IsStream t, Monad m, Storable a, Eq a)
3234 => Array a -> Fold m a b -> t m a -> t m b
3235wordsOn subseq f m = undefined -- D.fromStreamD $ D.wordsOn f subseq (D.toStreamD m)
3238-- XXX use a non-monadic intersperse to remove the MonadAsync constraint.
3240-- | Like 'splitOnSeq' but splits the separator as well, as an infix token.
3242-- > splitOn'_ pat xs = S.toList $ S.splitOn' (A.fromList pat) (FL.toList) (S.fromList xs)
3244-- >>> splitOn'_ "" "hello"
3245-- > ["h","","e","","l","","l","","o"]
3247-- >>> splitOn'_ "hello" ""
3250-- >>> splitOn'_ "hello" "hello"
3253-- >>> splitOn'_ "x" "hello"
3256-- >>> splitOn'_ "h" "hello"
3259-- >>> splitOn'_ "o" "hello"
3262-- >>> splitOn'_ "e" "hello"
3265-- >>> splitOn'_ "l" "hello"
3266-- > ["he","l","","l","o"]
3268-- >>> splitOn'_ "ll" "hello"
3272{-# INLINE _splitBySeq #-}
3274 :: (IsStream t, MonadAsync m, Storable a, Enum a, Eq a)
3275 => Array a -> Fold m a b -> t m a -> t m b
3276_splitBySeq patt f m = intersperseM (runFold f (A.read patt)) $ splitOnSeq patt f m
3278-- | Like 'splitSuffixOn' but keeps the suffix intact in the splits.
3280-- > splitSuffixOn'_ pat xs = S.toList $ FL.splitSuffixOn' (A.fromList pat) (FL.toList) (S.fromList xs)
3282-- >>> splitSuffixOn'_ "." ""
3285-- >>> splitSuffixOn'_ "." "."
3288-- >>> splitSuffixOn'_ "." "a"
3291-- >>> splitSuffixOn'_ "." ".a"
3294-- >>> splitSuffixOn'_ "." "a."
3297-- >>> splitSuffixOn'_ "." "a.b"
3300-- >>> splitSuffixOn'_ "." "a.b."
3303-- >>> splitSuffixOn'_ "." "a..b.."
3304-- > ["a.",".","b.","."]
3307{-# INLINE _splitBySuffixSeq #-}
3309 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
3310 => Array a -> Fold m a b -> t m a -> t m b
3311_splitBySuffixSeq patt f m =
3312 D.fromStreamD $ D.splitSuffixOn True patt f (D.toStreamD m)
3315-- This can be implemented easily using Rabin Karp
3316-- | Split post any one of the given patterns.
3317{-# INLINE splitSuffixOnAny #-}
3319 :: (IsStream t, Monad m, Storable a, Integral a)
3320 => [Array a] -> Fold m a b -> t m a -> t m b
3321splitSuffixOnAny subseq f m = undefined
3322 -- D.fromStreamD $ D.splitPostAny f subseq (D.toStreamD m)
3325------------------------------------------------------------------------------
3326-- Reorder in sequence
3327------------------------------------------------------------------------------
3330-- Buffer until the next element in sequence arrives. The function argument
3331-- determines the difference in sequence numbers. This could be useful in
3332-- implementing sequenced streams, for example, TCP reassembly.
3333{-# INLINE reassembleBy #-}
3335 :: (IsStream t, Monad m)
3340reassembleBy = undefined
3343------------------------------------------------------------------------------
3345------------------------------------------------------------------------------
3347-- | Tap the data flowing through a stream into a 'Fold'. For example, you may
3348-- add a tap to log the contents flowing through the stream. The fold is used
3349-- only for effects, its result is discarded.
3354-- -----stream m a ---------------stream m a-----
3359-- > S.drain $ S.tap (FL.drainBy print) (S.enumerateFromTo 1 2)
3364-- Compare with 'trace'.
3367tap :: (IsStream t, Monad m) => FL.Fold m a b -> t m a -> t m a
3368tap f xs = D.fromStreamD $ D.tap f (D.toStreamD xs)
3370-- | Apply a monadic function to each element flowing through the stream and
3371-- discard the results.
3374-- > S.drain $ S.trace print (S.enumerateFromTo 1 2)
3379-- Compare with 'tap'.
3382trace :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m a
3383trace f = mapM (\x -> void (f x) >> return x)
3385------------------------------------------------------------------------------
3386-- Windowed classification
3387------------------------------------------------------------------------------
3389-- We divide the stream into windows or chunks in space or time and each window
3390-- can be associated with a key, all events associated with a particular key in
3391-- the window can be folded to a single result. The stream can be split into
3392-- windows by size or by using a split predicate on the elements in the stream.
3393-- For example, when we receive a closing flag, we can close the window.
3395-- A "chunk" is a space window and a "session" is a time window. Are there any
3396-- other better short words to describe them. An alternative is to use
3397-- "swindow" and "twindow". Another word for "session" could be "spell".
3399-- TODO: To mark the position in space or time we can have Indexed or
3400-- TimeStamped types. That can make it easy to deal with the position indices
3403------------------------------------------------------------------------------
3404-- Keyed Sliding Windows
3405------------------------------------------------------------------------------
3408{-# INLINABLE classifySlidingChunks #-}
3409classifySlidingChunks
3410 :: (IsStream t, MonadAsync m, Ord k)
3411 => Int -- ^ window size
3412 -> Int -- ^ window slide
3413 -> Fold m a b -- ^ Fold to be applied to window events
3414 -> t m (k, a, Bool) -- ^ window key, data, close event
3416classifySlidingChunks wsize wslide (Fold step initial extract) str
3419-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
3420-- slide the send window when an ack is received and we slide the receive
3421-- window when a sequence is complete. Sliding is stateful in case of TCP,
3422-- sliding releases the send buffer or makes data available to the user from
3423-- the receive buffer.
3424{-# INLINABLE classifySlidingSessions #-}
3425classifySlidingSessions
3426 :: (IsStream t, MonadAsync m, Ord k)
3427 => Double -- ^ timer tick in seconds
3428 -> Double -- ^ time window size
3429 -> Double -- ^ window slide
3430 -> Fold m a b -- ^ Fold to be applied to window events
3431 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
3433classifySlidingSessions tick interval slide (Fold step initial extract) str
3437------------------------------------------------------------------------------
3438-- Sliding Window Buffers
3439------------------------------------------------------------------------------
3441-- These buffered versions could be faster than concurrent incremental folds of
3442-- all overlapping windows as in many cases we may not need all the values to
3443-- compute the fold, we can just compute the result using the old value and new
3444-- value. However, we may need the buffer once in a while, for example for
3445-- string search we usually compute the hash incrementally but when the hash
3446-- matches the hash of the pattern we need to compare the whole string.
3448-- XXX we should be able to implement sequence based splitting combinators
3449-- using this combinator.
3452-- | Buffer n elements of the input in a ring buffer. When t new elements are
3453-- collected, slide the window to remove the same number of oldest elements,
3454-- insert the new elements, and apply an incremental fold on the sliding
3455-- window, supplying the outgoing elements, the new ring buffer as arguments.
3457 :: (IsStream t, Monad m, Ord a, Storable a)
3458 => Int -- window size
3459 -> Int -- window slide
3460 -> Fold m (Ring a, Array a) b
3463slidingChunkBuffer = undefined
3465-- Buffer n seconds worth of stream elements of the input in a radix tree.
3466-- Every t seconds, remove the items that are older than n seconds, and apply
3467-- an incremental fold on the sliding window, supplying the outgoing elements,
3468-- and the new radix tree buffer as arguments.
3470 :: (IsStream t, Monad m, Ord a, Storable a)
3471 => Int -- window size
3473 -> Fold m (RTree a, Array a) b
3476slidingSessionBuffer = undefined
3479------------------------------------------------------------------------------
3480-- Keyed Session Windows
3481------------------------------------------------------------------------------
3484-- | Keyed variable size space windows. Close the window if we do not receive a
3485-- window event in the next "spaceout" elements.
3486{-# INLINABLE classifyChunksBy #-}
3488 :: (IsStream t, MonadAsync m, Ord k)
3489 => Int -- ^ window spaceout (spread)
3490 -> Bool -- ^ reset the spaceout when a chunk window element is received
3491 -> Fold m a b -- ^ Fold to be applied to chunk window elements
3492 -> t m (k, a, Bool) -- ^ chunk key, data, last element
3494classifyChunksBy spanout reset (Fold step initial extract) str = undefined
3496-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
3497-- received within the chunk size window. The chunk gets closed only if no
3498-- element is received within the chunk window.
3500{-# INLINABLE classifyKeepAliveChunks #-}
3501classifyKeepAliveChunks
3502 :: (IsStream t, MonadAsync m, Ord k)
3503 => Int -- ^ window spaceout (spread)
3504 -> Fold m a b -- ^ Fold to be applied to chunk window elements
3505 -> t m (k, a, Bool) -- ^ chunk key, data, last element
3507classifyKeepAliveChunks spanout = classifyChunksBy spanout True
3510-- | @classifySessionsBy tick timeout reset f stream@ groups together all input
3511-- stream elements that belong to the same session. @timeout@ is the maximum
3512-- lifetime of a session in seconds. All elements belonging to a session are
3513-- purged after this duration. If "reset" is 'Ture' then the timeout is reset
3514-- after every event received in the session. Session duration is measured
3515-- using the timestamp of the first element seen for that session. To detect
3516-- session timeouts, a monotonic event time clock is maintained using the
3517-- timestamps seen in the inputs and a timer with a tick duration specified by
3520-- @session key@ is a key that uniquely identifies the session for the given
3521-- element, @timestamp@ characterizes the time when the input element was
3522-- generated, this is an absolute time measured from some @Epoch@. @session
3523-- close@ is a boolean indicating whether this element marks the closing of the
3524-- session. When an input element with @session close@ set to @True@ is seen
3525-- the session is purged immediately.
3527-- All the input elements belonging to a session are collected using the fold
3528-- @f@. The session key and the fold result are emitted in the output stream
3529-- when the session is purged either via the session close event or via the
3530-- session liftime timeout.
3533{-# INLINABLE classifySessionsBy #-}
3535 :: (IsStream t, MonadAsync m, Ord k)
3536 => Double -- ^ timer tick in seconds
3537 -> Double -- ^ session timeout
3538 -> Bool -- ^ reset the timeout when an event is received
3539 -> Fold m a b -- ^ Fold to be applied to session events
3540 -> t m (k, a, Bool, AbsTime) -- ^ session key, timestamp, close event, data
3542classifySessionsBy tick timeout reset (Fold step initial extract) str =
3543 concatMap (\(Tuple4' _ _ _ s) -> s) $ scanlM' sstep szero stream
3547 timeoutMs = toRelTime (round (timeout * 1000) :: MilliSecond64)
3548 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
3549 szero = Tuple4' (toAbsTime (0 :: MilliSecond64)) H.empty Map.empty K.nil
3551 -- Got a new stream input element
3552 sstep (Tuple4' evTime hp mp _) (Just (key, a, closing, ts)) =
3553 -- XXX we should use a heap in pinned memory to scale it to a large
3556 -- deleting a key from the heap is expensive, so we never delete a
3557 -- key, we just purge it from the Map and it gets purged from the
3558 -- heap on timeout. We just need an extra lookup in the Map when
3559 -- the key is purged from the heap, that should not be expensive.
3561 -- To detect session inactivity we keep a timestamp of the latest event
3562 -- in the Map along with the fold result. When we purge the session
3563 -- from the heap we match the timestamp in the heap with the timestamp
3564 -- in the Map, if the latest timestamp is newer and has not expired we
3565 -- reinsert the key in the heap.
3567 -- XXX if the key is an Int, we can also use an IntMap for slightly
3568 -- better performance.
3570 let accumulate v = do
3571 Tuple' _ old <- maybe (initial >>= return . Tuple' ts) return v
3573 return $ Tuple' ts new
3576 let (r, mp') = Map.updateLookupWithKey (\_ _ -> Nothing) key mp
3577 Tuple' _ acc <- accumulate r
3579 return $ Tuple4' evTime hp mp' (yield (key, res))
3581 let r = Map.lookup key mp
3583 let mp' = Map.insert key acc mp
3587 let expiry = addToAbsTime ts timeoutMs
3588 in H.insert (Entry expiry key) hp
3590 -- Event time is maintained as monotonically increasing
3591 -- time. If we have lagged behind any of the timestamps
3592 -- seen then we increase it to match the latest time seen
3593 -- in the timestamps. We also increase it on timer ticks.
3594 return $ Tuple4' (max evTime ts) hp' mp' K.nil
3596 -- Got a timer tick event
3597 -- XXX can we yield the entries without accumulating them?
3598 sstep (Tuple4' evTime heap sessions _) Nothing = do
3599 (hp', mp', out) <- go heap sessions K.nil
3600 return $ Tuple4' curTime hp' mp' out
3604 curTime = addToAbsTime evTime tickMs
3606 let hres = H.uncons hp
3608 Just (Entry ts key, hp') -> do
3609 let duration = diffAbsTime curTime ts
3610 if duration >= timeoutMs
3612 let (r, mp') = Map.updateLookupWithKey
3613 (\_ _ -> Nothing) key mp
3615 Nothing -> go hp' mp' out
3616 Just (Tuple' latestTS acc) -> do
3617 let dur = diffAbsTime curTime latestTS
3618 if dur >= timeoutMs || not reset
3621 go hp' mp' ((key, sess) `K.cons` out)
3623 -- reset the session timeout
3624 let expiry = addToAbsTime latestTS timeoutMs
3625 hp'' = H.insert (Entry expiry key) hp'
3626 mp'' = Map.insert key (Tuple' latestTS acc) mp'
3628 else return (hp, mp, out)
3629 Nothing -> return (hp, mp, out)
3631 -- merge timer events in the stream
3632 stream = Serial.map Just str `Par.parallel` repeatM timer
3634 liftIO $ threadDelay (round $ tick * 1000000)
3637-- | Like 'classifySessionsOf' but the session is kept alive if an event is
3638-- received within the session window. The session times out and gets closed
3639-- only if no event is received within the specified session window size.
3642{-# INLINABLE _classifyKeepAliveSessions #-}
3643_classifyKeepAliveSessions
3644 :: (IsStream t, MonadAsync m, Ord k)
3645 => Double -- ^ session inactive timeout
3646 -> Fold m a b -- ^ Fold to be applied to session payload data
3647 -> t m (k, a, Bool, AbsTime) -- ^ session key, data, close flag, timestamp
3649_classifyKeepAliveSessions timeout = classifySessionsBy 1 timeout True
3651------------------------------------------------------------------------------
3652-- Keyed tumbling windows
3653------------------------------------------------------------------------------
3655-- Tumbling windows is a special case of sliding windows where the window slide
3656-- is the same as the window size. Or it can be a special case of session
3657-- windows where the reset flag is set to False.
3659-- XXX instead of using the early termination flag in the stream, we can use an
3660-- early terminating fold instead.
3663-- | Split the stream into fixed size chunks of specified size. Within each
3664-- such chunk fold the elements in buckets identified by the keys. A particular
3665-- bucket fold can be terminated early if a closing flag is encountered in an
3666-- element for that key.
3669{-# INLINABLE classifyChunksOf #-}
3671 :: (IsStream t, MonadAsync m, Ord k)
3672 => Int -- ^ window size
3673 -> Fold m a b -- ^ Fold to be applied to window events
3674 -> t m (k, a, Bool) -- ^ window key, data, close event
3676classifyChunksOf wsize = classifyChunksBy wsize False
3679-- | Split the stream into fixed size time windows of specified interval in
3680-- seconds. Within each such window, fold the elements in buckets identified by
3681-- the keys. A particular bucket fold can be terminated early if a closing flag
3682-- is encountered in an element for that key. Once a fold is terminated the key
3683-- and value for that bucket are emitted in the output stream.
3685-- Session @timestamp@ in the input stream is an absolute time from some epoch,
3686-- characterizing the time when the input element was generated. To detect
3687-- session window end, a monotonic event time clock is maintained synced with
3688-- the timestamps with a clock resolution of 1 second.
3691{-# INLINABLE _classifySessionsOf #-}
3693 :: (IsStream t, MonadAsync m, Ord k)
3694 => Double -- ^ time window size
3695 -> Fold m a b -- ^ Fold to be applied to window events
3696 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
3698_classifySessionsOf interval = classifySessionsBy 1 interval False
3699------------------------------------------------------------------------------
3701------------------------------------------------------------------------------
3703-- | Run a side effect before the stream yields its first element.
3706{-# INLINE before #-}
3707before :: (IsStream t, Monad m) => m b -> t m a -> t m a
3708before action xs = D.fromStreamD $ D.before action $ D.toStreamD xs
3710-- | Run a side effect whenever the stream stops normally.
3714after :: (IsStream t, Monad m) => m b -> t m a -> t m a
3715after action xs = D.fromStreamD $ D.after action $ D.toStreamD xs
3717-- | Run a side effect whenever the stream aborts due to an exception.
3720{-# INLINE onException #-}
3721onException :: (IsStream t, MonadCatch m) => m b -> t m a -> t m a
3722onException action xs = D.fromStreamD $ D.onException action $ D.toStreamD xs
3724-- | Run a side effect whenever the stream stops normally or aborts due to an
3728{-# INLINE finally #-}
3729finally :: (IsStream t, MonadCatch m) => m b -> t m a -> t m a
3730finally action xs = D.fromStreamD $ D.finally action $ D.toStreamD xs
3732-- | Run the first action before the stream starts and remember its output,
3733-- generate a stream using the output, run the second action using the
3734-- remembered value as an argument whenever the stream ends normally or due to
3738{-# INLINE bracket #-}
3739bracket :: (IsStream t, MonadCatch m)
3740 => m b -> (b -> m c) -> (b -> t m a) -> t m a
3741bracket bef aft bet = D.fromStreamD $
3742 D.bracket bef aft (\x -> toStreamD $ bet x)
3744-- | When evaluating a stream if an exception occurs, stream evaluation aborts
3745-- and the specified exception handler is run with the exception as argument.
3748{-# INLINE handle #-}
3749handle :: (IsStream t, MonadCatch m, Exception e)
3750 => (e -> m a) -> t m a -> t m a
3751handle 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