@@ -8,17 +8,22 @@
9module Streamly.Internal.Data.Stream.IsStream.Transform
9module Streamly.Internal.Data.Stream.IsStream.Transform
11− -- * Transformation
11+ -- * Piping
12+ -- | Pass through a 'Pipe'.
14− -- ** Mapping
19+ -- * Mapping
20+ -- | Stateless one-to-one maps.
15 , Serial.map
21 , Serial.map
16 , sequence
22 , sequence
21− -- ** Special Maps
26+ -- * Mapping Side Effects
@@ -27,13 +32,12 @@ module Streamly.Internal.Data.Stream.IsStream.Transform
28 , pollCounts
33 , pollCounts
30− -- ** Folding
35+ -- * Scanning By 'Fold'
31− -- ** Right folds
36+ , scan
32− , foldrS
37+ , postscan
35− -- ** Scanning
39+ -- * Scanning
36− -- ** Left scans
40+ -- | Left scans. Stateful, mostly one-to-one maps.
39 , scanlMAfter'
43 , scanlMAfter'
@@ -44,10 +48,6 @@ module Streamly.Internal.Data.Stream.IsStream.Transform
45 , scanl1M'
49 , scanl1M'
47− -- ** Scan Using Fold
51 -- XXX Once we have pipes the contravariant transformations can be
51 -- XXX Once we have pipes the contravariant transformations can be
52 -- represented by attaching pipes before a transformation.
52 -- represented by attaching pipes before a transformation.
@@ -61,69 +61,20 @@ module Streamly.Internal.Data.Stream.IsStream.Transform
61 -- , lprescanl'
61 -- , lprescanl'
62 -- , lprescanlM'
62 -- , lprescanlM'
64− -- ** Concurrent Transformation
64+ -- * Filtering
65− , D.mkParallel
65+ -- | Produce a subset of the stream.
76− -- ** Deleting Elements
77 , deleteBy
69 , deleteBy
79 -- , uniqBy -- by predicate e.g. to remove duplicate "/" in a path
71 -- , uniqBy -- by predicate e.g. to remove duplicate "/" in a path
80 -- , uniqOn -- to remove duplicate sequences
72 -- , uniqOn -- to remove duplicate sequences
81 -- , pruneBy -- dropAround + uniqBy - like words
73 -- , pruneBy -- dropAround + uniqBy - like words
83− -- ** Inserting Elements
75+ -- * Trimming
84−
76+ -- | Produce a subset of the stream trimmed at ends.
92− , intersperseSuffixBySpan
93− -- , intersperseBySpan
94− -- , intersperseByIndices -- using an index function/stream
96− -- time domain intersperse
97− -- , intersperseByTime
98− -- , intersperseByEvent
106− -- , timestampedR -- timer
112− -- ** Composable Left Folds
113− -- | See "Streamly.Internal.Data.Fold"
119− -- | See "Streamly.Internal.Data.Parser"
128 -- , takeGE
79 -- , takeGE
129 -- , takeBetween
80 -- , takeBetween
@@ -139,170 +90,74 @@ module Streamly.Internal.Data.Stream.IsStream.Transform
139 , dropWhileM
90 , dropWhileM
140 -- , dropWhileEnd
91 -- , dropWhileEnd
141 -- , dropAround
92 -- , dropAround
154− -- -- *** Searching Elements
158− -- -- *** Searching Sequences
159− -- , seqIndices -- search a sequence in the stream
161− -- -- *** Searching Multiple Sequences
162− -- , seqIndicesAny -- search any of the given sequence in the stream
164− -- -- -- ** Searching Streams
165− -- -- | Finding a stream within another stream.
168− -- | Streams can be sliced into segments in space or in time. We use the
169− -- term @chunk@ to refer to a spatial length of the stream (spatial window)
170− -- and the term @session@ to refer to a length in time (time window).
172− -- -- *** Using Element Separators
94+ -- * Inserting Elements
173− , splitOn
95+ -- | Produce a superset of the stream.
177− -- , splitBy
97+ , insertBy
178− , splitWithSuffix
98+ , intersperse
179− -- , splitByPrefix
99+ , intersperseM -- XXX naming
180− , wordsBy -- stripAndCompactBy
100+ , intersperseSuffix
181−
101+ , intersperseSuffixBySpan
182− -- -- *** Splitting By Sequences
102+ -- , intersperseBySpan
183− , splitOnSeq
103+ -- , intersperseByIndices -- using an index function/stream
185− -- , splitOnPrefixSeq
187− -- Keeping the delimiters
188− , splitWithSuffixSeq
189− -- , splitByPrefixSeq
192− -- Splitting using multiple sequence separators
105+ -- * Inserting Side Effects
193− -- , splitOnAnySeq
106+ , intersperseM_ -- XXX naming
194− -- , splitOnAnySuffixSeq
107+ , intersperseSuffix_
195− -- , splitOnAnyPrefixSeq
108+ , interspersePrefix_
197− -- -- *** Splitting By Streams
110+ -- * Inserting Time
198− -- -- | Splitting a stream using another stream as separator.
111+ -- , intersperseByTime
112+ -- , intersperseByEvent
200− -- Nested splitting
118+ -- * Reordering
201− , splitInnerBy
119+ , reverse
202− , splitInnerBySuffix
120+ , reverse'
204− -- ** Grouping
123+ -- * Position Indexing
205− -- In imperative terms, grouped folding can be considered as a nested loop
124+ , indexed
206− -- where we loop over the stream to group elements and then loop over
125+ , indexedR
207− -- individual groups to fold them to a single value that is yielded in the
210− -- , groupScan
127+ -- * Time Indexing
130+ -- , timestampedR -- timer
212− , groups
134+ -- * Searching
213− , groupsBy
135+ , findIndices -- XXX indicesBy
214− , groupsByRolling
136+ , elemIndices -- XXX indicesOf
216− -- ** Group map
138+ -- * Rolling map
139+ -- | Map using the previous element.
217 , rollingMapM
140 , rollingMapM
218 , rollingMap
141 , rollingMap
258− -- ** Flattening a Container of Streams
259− , concatFoldableWith
260− , concatMapFoldableWith
261− , concatForFoldableWith
263− -- ** Flattening a Stream of Streams
271− -- ** Flattening Using Unfolds
273− , concatUnfoldInterleave
274− , concatUnfoldRoundrobin
276− -- ** Inserting Streams in Streams
278− , gintercalateSuffix
285− -- * Generalize Inner Monad
289− -- * Transform Inner Monad
297− -- ** Maybe Streams
143+ -- * Maybe Streams
298− , catMaybes
144+ , catMaybes -- XXX justs (like lefts/rights)
299 , mapMaybe
145 , mapMaybe
300 , mapMaybeM
146 , mapMaybeM
302− -- ** Either Streams
148+ -- * Either Streams
153+ -- | Run streaming stages concurrently.
306 -- * Deprecated
161 -- * Deprecated
@@ -312,43 +167,42 @@ where
313import Control.Concurrent (threadDelay)
168import Control.Concurrent (threadDelay)
314import Control.Monad (void)
169import Control.Monad (void)
315−import Control.Monad.Catch (MonadCatch, MonadThrow)
170+import Control.Monad.Catch (MonadCatch)
316import Control.Monad.IO.Class (MonadIO(..))
171import Control.Monad.IO.Class (MonadIO(..))
317−import Control.Monad.Trans.Reader (ReaderT)
318−import Control.Monad.Trans.State.Strict (StateT)
319import Control.Monad.Trans.Class (MonadTrans(..))
172import Control.Monad.Trans.Class (MonadTrans(..))
320import Data.Either (isLeft, isRight)
173import Data.Either (isLeft, isRight)
321−import Data.Functor.Identity (Identity (..))
322import Data.Maybe (isJust, fromJust)
174import Data.Maybe (isJust, fromJust)
323−import Foreign.Storable (Storable)
324import Streamly.Internal.BaseCompat (fromLeft, fromRight)
175import Streamly.Internal.BaseCompat (fromLeft, fromRight)
325−import Streamly.Internal.Data.Fold.Types (Fold (..), Fold2 (..))
176+import Streamly.Internal.Data.Fold.Types (Fold (..))
326−import Streamly.Internal.Data.Parser (Parser (..))
327−import Streamly.Internal.Data.Unfold.Types (Unfold)
328−import Streamly.Internal.Data.Array.Storable.Foreign.Types (Array, writeNUnsafe)
329import Streamly.Internal.Data.SVar (MonadAsync)
177import Streamly.Internal.Data.SVar (MonadAsync)
330−import Streamly.Internal.Data.Stream.Ahead (ahead)
178+import Streamly.Internal.Data.Stream.IsStream.Common
331−import Streamly.Internal.Data.Stream.Async (async, wAsync)
179+ ( absTimesWith
332−import Streamly.Internal.Data.Stream.Combinators (maxYields)
180+ , drop
333−import Streamly.Internal.Data.Stream.Parallel (parallel)
181+ , findIndices
334−import Streamly.Internal.Data.Stream.Prelude
193+import Streamly.Internal.Data.Stream.Prelude (fromStreamS, toStreamS)
335− (fromStreamS, toStreamS, concatFoldableWith, concatMapFoldableWith
336− , concatForFoldableWith)
337import Streamly.Internal.Data.Stream.StreamD (fromStreamD, toStreamD)
194import Streamly.Internal.Data.Stream.StreamD (fromStreamD, toStreamD)
338import Streamly.Internal.Data.Stream.StreamK (IsStream)
195import Streamly.Internal.Data.Stream.StreamK (IsStream)
339−import Streamly.Internal.Data.Stream.Serial (SerialT, serial, wSerial)
196+import Streamly.Internal.Data.Stream.Serial (SerialT)
340import Streamly.Internal.Data.Pipe.Types (Pipe (..))
197import Streamly.Internal.Data.Pipe.Types (Pipe (..))
341−import Streamly.Internal.Data.Time.Units (TimeUnit64)
198+import Streamly.Internal.Data.Time.Units ( TimeUnit64, AbsTime, RelTime64)
343import qualified Streamly.Internal.Data.Fold as FL
200import qualified Streamly.Internal.Data.Fold as FL
344import qualified Streamly.Internal.Data.Stream.Prelude as P
201import qualified Streamly.Internal.Data.Stream.Prelude as P
345−import qualified Streamly.Internal.Data.Stream.StreamK as K
346−import qualified Streamly.Internal.Data.Stream.StreamD as D
347import qualified Streamly.Internal.Data.Stream.Serial as Serial
202import qualified Streamly.Internal.Data.Stream.Serial as Serial
348−import qualified Streamly.Internal.Data.Stream.Parallel as Par
203+import qualified Streamly.Internal.Data.Stream.StreamD as D
204+import qualified Streamly.Internal.Data.Stream.StreamK as K
349import qualified Streamly.Internal.Data.Stream.Zip as Z
205import qualified Streamly.Internal.Data.Stream.Zip as Z
350−import qualified Streamly.Internal.Data.Parser.ParserK.Types as PRK
351−import qualified Streamly.Internal.Data.Parser.ParserD as PRD
352#ifdef USE_STREAMK_ONLY
206#ifdef USE_STREAMK_ONLY
353import qualified Streamly.Internal.Data.Stream.StreamK as S
207import qualified Streamly.Internal.Data.Stream.StreamK as S
@@ -356,2440 +210,1038 @@ import qualified Streamly.Internal.Data.Stream.StreamD as S
358import Prelude hiding
212import Prelude hiding
359− ( filter, drop, dropWhile, take, takeWhile, zipWith, foldr
213+ ( filter, drop, dropWhile, take, takeWhile, foldr, map, mapM, sequence
360− , foldl, map, mapM, mapM_, sequence, all, any, sum, product, elem
361− , notElem, maximum, minimum, head, last, tail, length, null
362− , reverse, iterate, init, and, or, lookup, foldr1, (!!)
214+ , reverse, foldr1 , scanl, scanl1)
363− , scanl, scanl1, replicate, concatMap, span, splitAt, break
364− , repeat, concat, mconcat)
366------------------------------------------------------------------------------
216------------------------------------------------------------------------------
367−-- Searching
217+-- Piping
368------------------------------------------------------------------------------
218------------------------------------------------------------------------------
370−-- | Find all the indices where the element in the stream satisfies the given
220+-- | Use a 'Pipe' to transform a stream.
373−-- > findIndices = fold Fold.findIndices
376−{-# INLINE findIndices #-}
377−findIndices :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m Int
378−findIndices p m = fromStreamS $ S.findIndices p (toStreamS m)
380−-- | Find all the indices where the value of the element in the stream is equal
381−-- to the given value.
383−-- > elemIndices a = findIndices (== a)
222+-- /Internal/
385−-- @since 0.5.0
224+{-# INLINE transform #-}
386−{-# INLINE elemIndices #-}
387−elemIndices :: (IsStream t, Eq a, Monad m) => a -> t m a -> t m Int
225+transform :: (IsStream t, Monad m) => Pipe m a b -> t m a -> t m b
388−elemIndices a = findIndices (== a)
226+transform pipe xs = fromStreamD $ D.transform pipe (toStreamD xs)
390------------------------------------------------------------------------------
228------------------------------------------------------------------------------
391−-- Concurrent Application
229+-- Transformation Folds
392------------------------------------------------------------------------------
230------------------------------------------------------------------------------
394−infixr 0 |$
232+-- | Right fold to a streaming monad.
397−-- | Parallel transform application operator; applies a stream transformation
398−-- function @t m a -> t m b@ to a stream @t m a@ concurrently; the input stream
399−-- is evaluated asynchronously in an independent thread yielding elements to a
400−-- buffer and the transformation function runs in another thread consuming the
401−-- input from the buffer. '|$' is just like regular function application
402−-- operator '$' except that it is concurrent.
404−-- If you read the signature as @(t m a -> t m b) -> (t m a -> t m b)@ you can
405−-- look at it as a transformation that converts a transform function to a
406−-- buffered concurrent transform function.
408−-- The following code prints a value every second even though each stage adds a
234+-- > foldrS S.cons S.nil === id
236+-- 'foldrS' can be used to perform stateless stream to stream transformations
237+-- like map and filter in general. It can be coupled with a scan to perform
238+-- stateful transformations. However, note that the custom map and filter
239+-- routines can be much more efficient than this due to better stream fusion.
412−-- @
241+-- >>> S.toList $ S.foldrS S.cons S.nil $ S.fromList [1..5]
413−-- drain $
242+-- > [1,2,3,4,5]
414−-- S.mapM (\\x -> threadDelay 1000000 >> print x)
415−-- |$ S.repeatM (threadDelay 1000000 >> return 1)
418−-- /Concurrent/
244+-- Find if any element in the stream is 'True':
420−-- /Since: 0.3.0 ("Streamly")/
246+-- >>> 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)
422−-- @since 0.8.0
249+-- Map (+2) on odd elements and filter out the even elements:
424−(|$) :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> (t m a -> t m b)
425−-- (|$) f = f . Async.mkAsync
426−(|$) f = f . D.mkParallel
430−-- /Internal/
251+-- >>> 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)
432−{-# INLINE applyAsync #-}
254+-- 'foldrM' can also be represented in terms of 'foldrS', however, the former
433−applyAsync :: (IsStream t, MonadAsync m)
255+-- is much more efficient:
434− => (t m a -> t m b) -> (t m a -> t m b)
437−-- | Parallel reverse function application operator for streams; just like the
438−-- regular reverse function application operator '&' except that it is
441−-- @
257+-- > foldrM f z s = runIdentityT $ foldrS (\x xs -> lift $ f x (runIdentityT xs)) (lift z) s
443−-- S.repeatM (threadDelay 1000000 >> return 1)
444−-- |& S.mapM (\\x -> threadDelay 1000000 >> print x)
447−-- /Concurrent/
259+-- /Internal/
260+{-# INLINE foldrS #-}
261+foldrS :: IsStream t => (a -> t m b -> t m b) -> t m b -> t m a -> t m b
264+-- | Right fold to a transformer monad. This is the most general right fold
265+-- function. 'foldrS' is a special case of 'foldrT', however 'foldrS'
266+-- implementation can be more efficient:
449−-- /Since: 0.3.0 ("Streamly")/
268+-- > foldrS = foldrT
269+-- > foldrM f z s = runIdentityT $ foldrT (\x xs -> lift $ f x (runIdentityT xs)) (lift z) s
451−-- @since 0.8.0
271+-- 'foldrT' can be used to translate streamly streams to other transformer
452−{-# INLINE (|&) #-}
272+-- monads e.g. to a different streaming type.
453−(|&) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> t m b) -> t m b
456−------------------------------------------------------------------------------
457−-- General Transformation
458−------------------------------------------------------------------------------
460−-- | Use a 'Pipe' to transform a stream.
462-- /Internal/
274-- /Internal/
464−{-# INLINE transform #-}
275+{-# INLINE foldrT #-}
465−transform :: (IsStream t, Monad m) => Pipe m a b -> t m a -> t m b
276+foldrT :: (IsStream t, Monad m, Monad (s m), MonadTrans s)
466−transform pipe xs = fromStreamD $ D.transform pipe (toStreamD xs)
277+ => (a -> s m b -> s m b) -> s m b -> t m a -> s m b
278+foldrT f z s = S.foldrT f z (toStreamS s)
468------------------------------------------------------------------------------
280------------------------------------------------------------------------------
469−-- Transformation by Folding (Scans)
281+-- Transformation by Mapping
470------------------------------------------------------------------------------
282------------------------------------------------------------------------------
472−-- XXX It may be useful to have a version of scan where we can keep the
284+-- |
473−-- accumulator independent of the value emitted. So that we do not necessarily
285+-- @
474−-- have to keep a value in the accumulator which we are not using. We can pass
286+-- mapM f = sequence . map f
475−-- an extraction function that will take the accumulator and the current value
287+-- @
476−-- of the element and emit the next value in the stream. That will also make it
477−-- possible to modify the accumulator after using it. In fact, the step function
478−-- can return new accumulator and the value to be emitted. The signature would
479−-- be more like mapAccumL. Or we can change the signature of scanx to
482−-- | Strict left scan with an extraction function. Like 'scanl'', but applies a
483−-- user supplied extraction function (the third argument) at each step. This is
484−-- designed to work with the @foldl@ library. The suffix @x@ is a mnemonic for
489−-- /Since: 0.7.0 (Monad m constraint)/
490−{-# DEPRECATED scanx "Please use scanl followed by map instead." #-}
492−scanx :: (IsStream t, Monad m) => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
495−-- XXX this needs to be concurrent
496−-- XXX because of the use of D.cons for appending, scanlM' has quadratic
497−-- complexity when iterated over a stream. We should use StreamK style scanlM'
498−-- for linear performance on iteration.
500−-- | Like 'scanl'' but with a monadic step function and a monadic seed.
289+-- Apply a monadic function to each element of the stream and replace it with
290+-- the output of the resulting action.
502−-- /Since: 0.4.0/
292+-- @
293+-- > drain $ S.mapM putStr $ S.fromList ["a", "b", "c"]
504−-- /Since: 0.8.0 (signature change)/
296+-- drain $ S.replicateM 10 (return 1)
505−{-# INLINE scanlM' #-}
297+-- & (serially . S.mapM (\\x -> threadDelay 1000000 >> print x))
506−scanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
507−scanlM' step begin m = fromStreamD $ D.scanlM' step begin $ toStreamD m
509−-- | @scanlMAfter' accumulate initial done stream@ is like 'scanlM'' except
510−-- that it provides an additional @done@ function to be applied on the
511−-- accumulator when the stream stops. The result of @done@ is also emitted in
514−-- This function can be used to allocate a resource in the beginning of the
299+-- drain $ S.replicateM 10 (return 1)
515−-- scan and release it when the stream ends or to flush the internal state of
300+-- & (asyncly . S.mapM (\\x -> threadDelay 1000000 >> print x))
516−-- the scan at the end.
301+-- @
518−-- /Internal/
303+-- /Concurrent (do not use with 'parallely' on infinite streams)/
520−{-# INLINE scanlMAfter' #-}
305+-- @since 0.1.0
521−scanlMAfter' :: (IsStream t, Monad m)
306+{-# INLINE_EARLY mapM #-}
522− => (b -> a -> m b) -> m b -> (b -> m b) -> t m a -> t m b
307+mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
523−scanlMAfter' step initial done stream =
308+mapM = K.mapM
524− fromStreamD $ D.scanlMAfter' step initial done $ toStreamD stream
526−-- XXX because of the use of D.cons for appending, scanl' has quadratic
310+{-# RULES "mapM serial" mapM = mapMSerial #-}
527−-- complexity when iterated over a stream. We should use StreamK style scanlM'
311+{-# INLINE mapMSerial #-}
528−-- for linear performance on iteration.
312+mapMSerial :: Monad m => (a -> m b) -> SerialT m a -> SerialT m b
529−
313+mapMSerial = Serial.mapM
530−-- | Strict left scan. Like 'map', 'scanl'' too is a one to one transformation,
531−-- however it adds an extra element.
534−-- > S.toList $ S.scanl' (+) 0 $ fromList [1,2,3,4]
539−-- > S.toList $ S.scanl' (flip (:)) [] $ S.fromList [1,2,3,4]
540−-- [[],[1],[2,1],[3,2,1],[4,3,2,1]]
543−-- The output of 'scanl'' is the initial value of the accumulator followed by
544−-- all the intermediate steps and the final result of 'foldl''.
546−-- By streaming the accumulated state after each fold step, we can share the
547−-- state across multiple stages of stream composition. Each stage can modify or
548−-- extend the state, do some processing with it and emit it for the next stage,
549−-- thus modularizing the stream processing. This can be useful in
550−-- stateful or event-driven programming.
552−-- Consider the following monolithic example, computing the sum and the product
553−-- of the elements in a stream in one go using a @foldl'@:
556−-- > S.foldl' (\\(s, p) x -> (s + x, p * x)) (0,1) $ S.fromList \[1,2,3,4]
560−-- Using @scanl'@ we can make it modular by computing the sum in the first
561−-- stage and passing it down to the next stage for computing the product:
564−-- > S.foldl' (\\(_, p) (s, x) -> (s, p * x)) (0,1)
565−-- $ S.scanl' (\\(s, _) x -> (s + x, x)) (0,1)
566−-- $ S.fromList \[1,2,3,4]
570−-- IMPORTANT: 'scanl'' evaluates the accumulator to WHNF. To avoid building
571−-- lazy expressions inside the accumulator, it is recommended that a strict
572−-- data structure is used for accumulator.
574−-- See also: 'usingStateT'
577−{-# INLINE scanl' #-}
578−scanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
579−scanl' step z m = fromStreamS $ S.scanl' step z $ toStreamS m
581−-- | Like 'scanl'' but does not stream the initial value of the accumulator.
583−-- > postscanl' f z xs = S.drop 1 $ S.scanl' f z xs
586−{-# INLINE postscanl' #-}
587−postscanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
588−postscanl' step z m = fromStreamD $ D.postscanl' step z $ toStreamD m
590−-- XXX this needs to be concurrent
591−-- | Like 'postscanl'' but with a monadic step function and a monadic seed.
595−-- /Since: 0.8.0 (signature change)/
596−{-# INLINE postscanlM' #-}
597−postscanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
598−postscanlM' step z m = fromStreamD $ D.postscanlM' step z $ toStreamD m
600−-- XXX prescanl does not sound very useful, enable only if there is a
601−-- compelling use case.
603−-- | Like scanl' but does not stream the final value of the accumulator.
606−{-# INLINE prescanl' #-}
607−prescanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
608−prescanl' step z m = fromStreamD $ D.prescanl' step z $ toStreamD m
610−-- XXX this needs to be concurrent
611−-- | Like prescanl' but with a monadic step function and a monadic seed.
614−{-# INLINE prescanlM' #-}
615−prescanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
616−prescanlM' step z m = fromStreamD $ D.prescanlM' step z $ toStreamD m
618−-- XXX this needs to be concurrent
619−-- | Like 'scanl1'' but with a monadic step function.
622−{-# INLINE scanl1M' #-}
623−scanl1M' :: (IsStream t, Monad m) => (a -> a -> m a) -> t m a -> t m a
624−scanl1M' step m = fromStreamD $ D.scanl1M' step $ toStreamD m
626−-- | Like 'scanl'' but for a non-empty stream. The first element of the stream
627−-- is used as the initial value of the accumulator. Does nothing if the stream
631−-- > S.toList $ S.scanl1 (+) $ fromList [1,2,3,4]
636−{-# INLINE scanl1' #-}
637−scanl1' :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> t m a
638−scanl1' step m = fromStreamD $ D.scanl1' step $ toStreamD m
640−------------------------------------------------------------------------------
641−-- Scanning with a Fold
642−------------------------------------------------------------------------------
644−-- | Scan a stream using the given monadic fold.
648−scan :: (IsStream t, Monad m) => Fold m a b -> t m a -> t m b
651−-- | Postscan a stream using the given monadic fold.
654−{-# INLINE postscan #-}
655−postscan :: (IsStream t, Monad m) => Fold m a b -> t m a -> t m b
656−postscan = P.postscanOnce
658−------------------------------------------------------------------------------
659−-- Stateful Transformations
660−------------------------------------------------------------------------------
662−-- XXX this is not a one-to-one map so calling it map may not be right.
663−-- We can perhaps call it zipWithTail or rollWith.
665−-- | Apply a function on every two successive elements of a stream. If the
666−-- stream consists of a single element the output is an empty stream.
668−-- This is the stream equivalent of the list idiom @zipWith f xs (tail xs)@.
672−{-# INLINE rollingMap #-}
673−rollingMap :: (IsStream t, Monad m) => (a -> a -> b) -> t m a -> t m b
674−rollingMap f m = fromStreamD $ D.rollingMap f $ toStreamD m
676−-- | Like 'rollingMap' but with an effectful map function.
680−{-# INLINE rollingMapM #-}
681−rollingMapM :: (IsStream t, Monad m) => (a -> a -> m b) -> t m a -> t m b
682−rollingMapM f m = fromStreamD $ D.rollingMapM f $ toStreamD m
684−------------------------------------------------------------------------------
685−-- Transformation by Filtering
686−------------------------------------------------------------------------------
688−-- | Include only those elements that pass a predicate.
691−{-# INLINE filter #-}
692−#if __GLASGOW_HASKELL__ != 802
693−-- GHC 8.2.2 crashes with this code, when used with "stack"
694−filter :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
695−filter p m = fromStreamS $ S.filter p $ toStreamS m
697−filter :: IsStream t => (a -> Bool) -> t m a -> t m a
701−-- | Same as 'filter' but with a monadic predicate.
704−{-# INLINE filterM #-}
705−filterM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
706−filterM p m = fromStreamD $ D.filterM p $ toStreamD m
708−-- | Drop repeated elements that are adjacent to each other.
712−uniq :: (Eq a, IsStream t, Monad m) => t m a -> t m a
713−uniq = fromStreamD . D.uniq . toStreamD
715−-- | Take first 'n' elements from the stream and discard the rest.
719−take :: (IsStream t, Monad m) => Int -> t m a -> t m a
720−take n m = fromStreamS $ S.take n $ toStreamS
721− (maxYields (Just (fromIntegral n)) m)
723−-- | End the stream as soon as the predicate fails on an element.
726−{-# INLINE takeWhile #-}
727−takeWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
728−takeWhile p m = fromStreamS $ S.takeWhile p $ toStreamS m
730−-- | Same as 'takeWhile' but with a monadic predicate.
733−{-# INLINE takeWhileM #-}
734−takeWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
735−takeWhileM p m = fromStreamD $ D.takeWhileM p $ toStreamD m
737−-- | @takeByTime duration@ yields stream elements upto specified time
738−-- @duration@. The duration starts when the stream is evaluated for the first
739−-- time, before the first element is yielded. The time duration is checked
740−-- before generating each element, if the duration has expired the stream
743−-- The total time taken in executing the stream is guaranteed to be /at least/
744−-- @duration@, however, because the duration is checked before generating an
745−-- element, the upper bound is indeterminate and depends on the time taken in
746−-- generating and processing the last element.
748−-- No element is yielded if the duration is zero. At least one element is
749−-- yielded if the duration is non-zero.
753−{-# INLINE takeByTime #-}
754−takeByTime ::(MonadIO m, IsStream t, TimeUnit64 d) => d -> t m a -> t m a
755−takeByTime d = fromStreamD . D.takeByTime d . toStreamD
757−-- | Discard first 'n' elements from the stream and take the rest.
761−drop :: (IsStream t, Monad m) => Int -> t m a -> t m a
762−drop n m = fromStreamS $ S.drop n $ toStreamS m
764−-- | Drop elements in the stream as long as the predicate succeeds and then
765−-- take the rest of the stream.
768−{-# INLINE dropWhile #-}
769−dropWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
770−dropWhile p m = fromStreamS $ S.dropWhile p $ toStreamS m
772−-- | Same as 'dropWhile' but with a monadic predicate.
775−{-# INLINE dropWhileM #-}
776−dropWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
777−dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m
779−-- | @dropByTime duration@ drops stream elements until specified @duration@ has
780−-- passed. The duration begins when the stream is evaluated for the first
781−-- time. The time duration is checked /after/ generating a stream element, the
782−-- element is yielded if the duration has expired otherwise it is dropped.
784−-- The time elapsed before starting to generate the first element is /at most/
785−-- @duration@, however, because the duration expiry is checked after the
786−-- element is generated, the lower bound is indeterminate and depends on the
787−-- time taken in generating an element.
789−-- All elements are yielded if the duration is zero.
793−{-# INLINE dropByTime #-}
794−dropByTime ::(MonadIO m, IsStream t, TimeUnit64 d) => d -> t m a -> t m a
795−dropByTime d = fromStreamD . D.dropByTime d . toStreamD
797−-- | Drop prefix from the input stream if present.
801−-- /Unimplemented/ - Help wanted.
802−{-# INLINE dropPrefix #-}
804− -- (Eq a, IsStream t, Monad m) =>
805− t m a -> t m a -> t m a
806−dropPrefix = error "Not implemented yet!"
808−-- | Drop all matching infix from the input stream if present. Infix stream
809−-- may be consumed multiple times.
811−-- Space: @O(n)@ where n is the length of the infix.
813−-- /Unimplemented/ - Help wanted.
814−{-# INLINE dropInfix #-}
816− -- (Eq a, IsStream t, Monad m) =>
817− t m a -> t m a -> t m a
818−dropInfix = error "Not implemented yet!"
820−-- | Drop suffix from the input stream if present. Suffix stream may be
821−-- consumed multiple times.
823−-- Space: @O(n)@ where n is the length of the suffix.
825−-- /Unimplemented/ - Help wanted.
826−{-# INLINE dropSuffix #-}
828− -- (Eq a, IsStream t, Monad m) =>
829− t m a -> t m a -> t m a
830−dropSuffix = error "Not implemented yet!"
832−------------------------------------------------------------------------------
833−-- Transformation by Mapping
834−------------------------------------------------------------------------------
838−-- mapM f = sequence . map f
841−-- Apply a monadic function to each element of the stream and replace it with
842−-- the output of the resulting action.
845−-- > drain $ S.mapM putStr $ S.fromList ["a", "b", "c"]
848−-- drain $ S.replicateM 10 (return 1)
849−-- & (serially . S.mapM (\\x -> threadDelay 1000000 >> print x))
851−-- drain $ S.replicateM 10 (return 1)
852−-- & (asyncly . S.mapM (\\x -> threadDelay 1000000 >> print x))
855−-- /Concurrent (do not use with 'parallely' on infinite streams)/
858−{-# INLINE_EARLY mapM #-}
859−mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
862−{-# RULES "mapM serial" mapM = mapMSerial #-}
863−{-# INLINE mapMSerial #-}
864−mapMSerial :: Monad m => (a -> m b) -> SerialT m a -> SerialT m b
865−mapMSerial = Serial.mapM
869−-- sequence = mapM id
872−-- Replace the elements of a stream of monadic actions with the outputs of
876−-- > drain $ S.sequence $ S.fromList [putStr "a", putStr "b", putStrLn "c"]
879−-- drain $ S.replicateM 10 (return $ threadDelay 1000000 >> print 1)
880−-- & (serially . S.sequence)
882−-- drain $ S.replicateM 10 (return $ threadDelay 1000000 >> print 1)
883−-- & (asyncly . S.sequence)
886−-- /Concurrent (do not use with 'parallely' on infinite streams)/
889−{-# INLINE sequence #-}
890−sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
891−sequence m = fromStreamS $ S.sequence (toStreamS m)
895−-- The stateful step function can be simplified to @(s -> a -> m b)@ to provide
896−-- a read-only environment. However, that would just be 'mapM'.
898−-- The initial action could be @m (s, Maybe b)@, and we can also add a final
899−-- action @s -> m (Maybe b)@. This can be used to get pre/post scan like
900−-- functionality and also to flush the state in the end like scanlMAfter'.
901−-- We can also use it along with a fusible version of bracket to get
902−-- scanlMAfter' like functionality. See issue #677.
904−-- This can be further generalized to a type similar to Fold/Parser, giving it
905−-- filtering and parsing capability as well (this is in fact equivalent to
908−-- smapM :: (s -> a -> m (Step s b)) -> m s -> t m a -> t m b
911−-- | A stateful 'mapM', equivalent to a left scan, more like mapAccumL.
912−-- Hopefully, this is a better alternative to @scan@. Separation of state from
913−-- the output makes it easier to think in terms of a shared state, and also
914−-- makes it easier to keep the state fully strict and the output lazy.
916−-- See also: 'scanlM''
921−smapM :: (IsStream t, Monad m) =>
922− (s -> a -> m (s, b))
926−smapM step initial stream =
927− -- XXX implement this directly instead of using scanlM'
928− -- Once we have postscanlM' with monadic initial we can use this code
929− -- let r = postscanlM'
930− -- (\(s, _) a -> step s a)
931− -- (fmap (,undefined) initial)
934− (\(s, _) a -> step s a)
935− (fmap (,undefined) initial)
939−------------------------------------------------------------------------------
941−------------------------------------------------------------------------------
943−-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'
944−-- elements, and return a stream of values extracted from 'Just'.
949−-- mapMaybe f = S.map 'fromJust' . S.filter 'isJust' . S.map f
953−{-# INLINE mapMaybe #-}
954−mapMaybe :: (IsStream t, Monad m) => (a -> Maybe b) -> t m a -> t m b
955−mapMaybe f m = fromStreamS $ S.mapMaybe f $ toStreamS m
957−-- | Like 'mapMaybe' but maps a monadic function.
962−-- mapMaybeM f = S.map 'fromJust' . S.filter 'isJust' . S.mapM f
965−-- /Concurrent (do not use with 'parallely' on infinite streams)/
968−{-# INLINE_EARLY mapMaybeM #-}
969−mapMaybeM :: (IsStream t, MonadAsync m, Functor (t m))
970− => (a -> m (Maybe b)) -> t m a -> t m b
971−mapMaybeM f = fmap fromJust . filter isJust . K.mapM f
973−{-# RULES "mapMaybeM serial" mapMaybeM = mapMaybeMSerial #-}
974−{-# INLINE mapMaybeMSerial #-}
975−mapMaybeMSerial :: Monad m => (a -> m (Maybe b)) -> SerialT m a -> SerialT m b
976−mapMaybeMSerial f m = fromStreamD $ D.mapMaybeM f $ toStreamD m
978−-- | In a stream of 'Maybe's, discard 'Nothing's and unwrap 'Just's.
982−catMaybes :: (IsStream t, Monad m, Functor (t m)) => t m (Maybe a) -> t m a
983−catMaybes = fmap fromJust . filter isJust
985−------------------------------------------------------------------------------
986−-- Transformation by Reordering
987−------------------------------------------------------------------------------
989−-- XXX Use a compact region list to temporarily store the list, in both reverse
990−-- as well as in reverse'.
992−-- /Note:/ 'reverse'' is much faster than this, use that when performance
995−-- > reverse = S.foldlT (flip S.cons) S.nil
997−-- | Returns the elements of the stream in reverse order. The stream must be
998−-- finite. Note that this necessarily buffers the entire stream in memory.
1000−-- /Since 0.7.0 (Monad m constraint)/
1003−{-# INLINE reverse #-}
1004−reverse :: (IsStream t, Monad m) => t m a -> t m a
1005−reverse s = fromStreamS $ S.reverse $ toStreamS s
1007−-- | Like 'reverse' but several times faster, requires a 'Storable' instance.
1010−{-# INLINE reverse' #-}
1011−reverse' :: (IsStream t, MonadIO m, Storable a) => t m a -> t m a
1012−reverse' s = fromStreamD $ D.reverse' $ toStreamD s
1014−------------------------------------------------------------------------------
1015−-- Transformation by Inserting
1016−------------------------------------------------------------------------------
1018−-- intersperseM = intersperseBySpan 1
1020−-- | Insert an effect and its output before consuming an element of a stream
1021−-- except the first one.
1024−-- >>> S.toList $ S.trace putChar $ S.intersperseM (putChar '.' >> return ',') $ S.fromList "hello"
1025−-- > h.,e.,l.,l.,o"h,e,l,l,o"
1029−{-# INLINE intersperseM #-}
1030−intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
1031−intersperseM m = fromStreamS . S.intersperseM m . toStreamS
1033−-- | Insert a side effect before consuming an element of a stream except the
1037−-- >>> S.drain $ S.trace putChar $ S.intersperseM_ (putChar '.') $ S.fromList "hello"
1042−{-# INLINE intersperseM_ #-}
1043−intersperseM_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
1044−intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
1046−-- | Insert a pure value between successive elements of a stream.
1049−-- > S.toList $ S.intersperse ',' $ S.fromList "hello"
1054−{-# INLINE intersperse #-}
1055−intersperse :: (IsStream t, MonadAsync m) => a -> t m a -> t m a
1056−intersperse a = fromStreamS . S.intersperse a . toStreamS
1058−-- | Insert an effect and its output after consuming an element of a stream.
1061−-- >>> S.toList $ S.trace putChar $ S.intersperseSuffix (putChar '.' >> return ',') $ S.fromList "hello"
1062−-- > h.,e.,l.,l.,o.,"h,e,l,l,o,"
1066−{-# INLINE intersperseSuffix #-}
1067−intersperseSuffix :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
1068−intersperseSuffix m = fromStreamD . D.intersperseSuffix m . toStreamD
1070−-- | Insert a side effect after consuming an element of a stream.
1073−-- > S.mapM_ putChar $ S.intersperseSuffix_ (threadDelay 1000000) $ S.fromList "hello"
1078−{-# INLINE intersperseSuffix_ #-}
1079−intersperseSuffix_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
1080−intersperseSuffix_ m = fromStreamD . D.intersperseSuffix_ m . toStreamD
1082−-- | Insert a side effect before consuming an element of a stream.
1085−-- >>> S.toList $ S.trace putChar $ S.interspersePrefix_ (putChar '.' >> return ',') $ S.fromList "hello"
1086−-- > .h.e.l.l.o"hello"
1089−-- See also: 'trace_'
1095−{-# INLINE interspersePrefix_ #-}
1096−interspersePrefix_ :: (IsStream t, MonadAsync m) => m b -> t m a -> t m a
1097−interspersePrefix_ m = mapM (\x -> void m >> return x)
1099−-- Note: delay must be serial.
1101−-- | Introduce a delay of specified seconds before consuming an element of the
1102−-- stream except the first one.
1105−-- >>> S.mapM_ print $ S.timestamped $ S.delay 1 $ S.enumerateFromTo 1 3
1106−-- > (AbsTime (TimeSpec {sec = 2502706, nsec = 751137000}),1)
1107−-- > (AbsTime (TimeSpec {sec = 2502707, nsec = 743535000}),2)
1108−-- > (AbsTime (TimeSpec {sec = 2502708, nsec = 749758000}),3)
1113−{-# INLINE delay #-}
1114−delay :: (IsStream t, MonadIO m) => Double -> t m a -> t m a
1115−delay n = intersperseM_ $ liftIO $ threadDelay $ round $ n * 1000000
1117−-- Note: delay must be serial.
1119−-- | Introduce a delay of specified seconds after consuming an element of a
1123−-- >>> S.mapM_ print $ S.timestamped $ S.delayPost 1 $ S.enumerateFromTo 1 3
1124−-- > (AbsTime (TimeSpec {sec = 2502826, nsec = 119030000}),1)
1125−-- > (AbsTime (TimeSpec {sec = 2502827, nsec = 111393000}),2)
1126−-- > (AbsTime (TimeSpec {sec = 2502828, nsec = 112221000}),3)
1131−{-# INLINE delayPost #-}
1132−delayPost :: (IsStream t, MonadIO m) => Double -> t m a -> t m a
1133−delayPost n = intersperseSuffix_ $ liftIO $ threadDelay $ round $ n * 1000000
1135−-- Note: delay must be serial, that's why 'trace_' is used.
1137−-- | Introduce a delay of specified seconds before consuming an element of a
1141−-- >>> S.mapM_ print $ S.timestamped $ S.delayPre 1 $ S.enumerateFromTo 1 3
1142−-- > (AbsTime (TimeSpec {sec = 2502207, nsec = 533177000}),1)
1143−-- > (AbsTime (TimeSpec {sec = 2502208, nsec = 530859000}),2)
1144−-- > (AbsTime (TimeSpec {sec = 2502209, nsec = 531619000}),3)
1149−{-# INLINE delayPre #-}
1150−delayPre :: (IsStream t, MonadIO m) => Double -> t m a -> t m a
1151−delayPre n = trace_ $ liftIO $ threadDelay $ round $ n * 1000000
1153−-- | Like 'intersperseSuffix' but intersperses an effectful action into the
1154−-- input stream after every @n@ elements and after the last element.
1157−-- > S.toList $ S.intersperseSuffixBySpan 2 (return ',') $ S.fromList "hello"
1163−{-# INLINE intersperseSuffixBySpan #-}
1164−intersperseSuffixBySpan :: (IsStream t, MonadAsync m)
1165− => Int -> m a -> t m a -> t m a
1166−intersperseSuffixBySpan n eff =
1167− fromStreamD . D.intersperseSuffixBySpan n eff . toStreamD
1170−-- | Intersperse a monadic action into the input stream after every @n@
1174−-- > S.toList $ S.intersperseBySpan 2 (return ',') $ S.fromList "hello"
1179−{-# INLINE intersperseBySpan #-}
1180−intersperseBySpan :: IsStream t => Int -> m a -> t m a -> t m a
1181−intersperseBySpan _n _f _xs = undefined
1184−-- | @insertBy cmp elem stream@ inserts @elem@ before the first element in
1185−-- @stream@ that is less than @elem@ when compared using @cmp@.
1188−-- insertBy cmp x = 'mergeBy' cmp ('yield' x)
1192−-- > S.toList $ S.insertBy compare 2 $ S.fromList [1,3,5]
1197−{-# INLINE insertBy #-}
1199− (IsStream t, Monad m) => (a -> a -> Ordering) -> a -> t m a -> t m a
1200−insertBy cmp x m = fromStreamS $ S.insertBy cmp x (toStreamS m)
1202−------------------------------------------------------------------------------
1204−------------------------------------------------------------------------------
1206−-- | Deletes the first occurrence of the element in the stream that satisfies
1207−-- the given equality predicate.
1210−-- > S.toList $ S.deleteBy (==) 3 $ S.fromList [1,3,3,5]
1215−{-# INLINE deleteBy #-}
1216−deleteBy :: (IsStream t, Monad m) => (a -> a -> Bool) -> a -> t m a -> t m a
1217−deleteBy cmp x m = fromStreamS $ S.deleteBy cmp x (toStreamS m)
1219−------------------------------------------------------------------------------
1221−------------------------------------------------------------------------------
1224−-- > indexed = S.postscanl' (\(i, _) x -> (i + 1, x)) (-1,undefined)
1225−-- > indexed = S.zipWith (,) (S.enumerateFrom 0)
1227−-- Pair each element in a stream with its index, starting from index 0.
1230−-- > S.toList $ S.indexed $ S.fromList "hello"
1231−-- [(0,'h'),(1,'e'),(2,'l'),(3,'l'),(4,'o')]
1235−{-# INLINE indexed #-}
1236−indexed :: (IsStream t, Monad m) => t m a -> t m (Int, a)
1237−indexed = fromStreamD . D.indexed . toStreamD
1240−-- > indexedR n = S.postscanl' (\(i, _) x -> (i - 1, x)) (n + 1,undefined)
1241−-- > indexedR n = S.zipWith (,) (S.enumerateFromThen n (n - 1))
1243−-- Pair each element in a stream with its index, starting from the
1244−-- given index @n@ and counting down.
1247−-- > S.toList $ S.indexedR 10 $ S.fromList "hello"
1248−-- [(10,'h'),(9,'e'),(8,'l'),(7,'l'),(6,'o')]
1252−{-# INLINE indexedR #-}
1253−indexedR :: (IsStream t, Monad m) => Int -> t m a -> t m (Int, a)
1254−indexedR n = fromStreamD . D.indexedR n . toStreamD
1256−------------------------------------------------------------------------------
1258−------------------------------------------------------------------------------
1260−-- | Merge two streams using a comparison function. The head elements of both
1261−-- the streams are compared and the smaller of the two elements is emitted, if
1262−-- both elements are equal then the element from the first stream is used
1265−-- If the streams are sorted in ascending order, the resulting stream would
1266−-- also remain sorted in ascending order.
1269−-- > S.toList $ S.mergeBy compare (S.fromList [1,3,5]) (S.fromList [2,4,6,8])
1274−{-# INLINABLE mergeBy #-}
1276− (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> t m a -> t m a
1277−mergeBy f m1 m2 = fromStreamS $ S.mergeBy f (toStreamS m1) (toStreamS m2)
1279−-- | Like 'mergeBy' but with a monadic comparison function.
1281−-- Merge two streams randomly:
1284−-- > randomly _ _ = randomIO >>= \x -> return $ if x then LT else GT
1285−-- > S.toList $ S.mergeByM randomly (S.fromList [1,1,1,1]) (S.fromList [2,2,2,2])
1286−-- [2,1,2,2,2,1,1,1]
1289−-- Merge two streams in a proportion of 2:1:
1292−-- proportionately m n = do
1293−-- ref <- newIORef $ cycle $ concat [replicate m LT, replicate n GT]
1294−-- return $ \\_ _ -> do
1295−-- r <- readIORef ref
1296−-- writeIORef ref $ tail r
1300−-- f <- proportionately 2 1
1301−-- xs <- S.toList $ S.mergeByM f (S.fromList [1,1,1,1,1,1]) (S.fromList [2,2,2])
1305−-- [1,1,2,1,1,2,1,1,2]
1309−{-# INLINABLE mergeByM #-}
1311− :: (IsStream t, Monad m)
1312− => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
1313−mergeByM f m1 m2 = fromStreamS $ S.mergeByM f (toStreamS m1) (toStreamS m2)
1316−-- | Like 'mergeByM' but stops merging as soon as any of the two streams stops.
1317−{-# INLINABLE mergeEndByAny #-}
1319− :: (IsStream t, Monad m)
1320− => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
1321−mergeEndByAny f m1 m2 = fromStreamD $
1322− D.mergeEndByAny f (toStreamD m1) (toStreamD m2)
1324−-- Like 'mergeByM' but stops merging as soon as the first stream stops.
1325−{-# INLINABLE mergeEndByFirst #-}
1327− :: (IsStream t, Monad m)
1328− => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
1329−mergeEndByFirst f m1 m2 = fromStreamS $
1330− D.mergeEndByFirst f (toStreamD m1) (toStreamD m2)
1333−-- Holding this back for now, we may want to use the name "merge" differently
1335−-- | Same as @'mergeBy' 'compare'@.
1338−-- > S.toList $ S.merge (S.fromList [1,3,5]) (S.fromList [2,4,6,8])
1343−{-# INLINABLE merge #-}
1345− (IsStream t, Monad m, Ord a) => t m a -> t m a -> t m a
1346−merge = mergeBy compare
1349−-- | Like 'mergeBy' but merges concurrently (i.e. both the elements being
1350−-- merged are generated concurrently).
1353−{-# INLINE mergeAsyncBy #-}
1354−mergeAsyncBy :: (IsStream t, MonadAsync m)
1355− => (a -> a -> Ordering) -> t m a -> t m a -> t m a
1356−mergeAsyncBy f = mergeAsyncByM (\a b -> return $ f a b)
1358−-- | Like 'mergeByM' but merges concurrently (i.e. both the elements being
1359−-- merged are generated concurrently).
1362−{-# INLINE mergeAsyncByM #-}
1363−mergeAsyncByM :: (IsStream t, MonadAsync m)
1364− => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
1365−mergeAsyncByM f m1 m2 = fromStreamD $
1366− D.mergeByM f (D.mkParallelD $ toStreamD m1) (D.mkParallelD $ toStreamD m2)
1368−------------------------------------------------------------------------------
1370−------------------------------------------------------------------------------
1372−-- | @concatMapWith merge map stream@ is a two dimensional looping combinator.
1373−-- The first argument specifies a merge or concat function that is used to
1374−-- merge the streams generated by applying the second argument i.e. the @map@
1375−-- function to each element of the input stream. The concat function could be
1376−-- 'serial', 'parallel', 'async', 'ahead' or any other zip or merge function
1377−-- and the second argument could be any stream generation function using a
1380−-- /Compare 'foldMapWith'/
1384−-- /Since: 0.8.0 (signature change)/
1385−{-# INLINE concatMapWith #-}
1388− => (t m b -> t m b -> t m b)
1392−concatMapWith = K.concatMapBy
1394−-- | Like 'concatMapWith' but carries a state which can be used to share
1395−-- information across multiple steps of concat.
1398−-- concatSmapMWith combine f initial = concatMapWith combine id . smapM f initial
1403−{-# INLINE concatSmapMWith #-}
1405− :: (IsStream t, Monad m)
1406− => (t m b -> t m b -> t m b)
1407− -> (s -> a -> m (s, t m b))
1411−concatSmapMWith combine f initial = concatMapWith combine id . smapM f initial
1413−-- | Map a stream producing function on each element of the stream and then
315+-- |
1414−-- flatten the results into a single stream.
1417−-- concatMap f = 'concat . map f'
317+-- sequence = mapM id
1418−-- concatMap = 'concatMapWith' 'Serial.serial'
1419−-- concatMap f = 'concatMapM' (return . f)
1420−-- concatMap f = 'concatUnfold' (UF.lmap f UF.fromStream)
1423−-- @since 0.6.0
320+-- Replace the elements of a stream of monadic actions with the outputs of
1424−{-# INLINE concatMap #-}
321+-- those actions.
1425−concatMap ::(IsStream t, Monad m) => (a -> t m b) -> t m a -> t m b
1426−concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)
1428−-- | Flatten a stream of streams to a single stream.
1431−-- concat = concatMap id
324+-- > drain $ S.sequence $ S.fromList [putStr "a", putStr "b", putStrLn "c"]
1435−{-# INLINE concat #-}
1436−concat :: (IsStream t, Monad m) => t m (t m a) -> t m a
1437−concat = concatMap id
1439−-- | Append the outputs of two streams, yielding all the elements from the
1440−-- first stream and then yielding all the elements from the second stream.
1442−-- IMPORTANT NOTE: This could be 100x faster than @serial/<>@ for appending a
1443−-- few (say 100) streams because it can fuse via stream fusion. However, it
1444−-- does not scale for a large number of streams (say 1000s) and becomes
1445−-- qudartically slow. Therefore use this for custom appending of a few streams
1446−-- but use 'concatMap' or 'concatMapWith serial' for appending @n@ streams or
1447−-- infinite containers of streams.
1450−{-# INLINE append #-}
1451−append ::(IsStream t, Monad m) => t m b -> t m b -> t m b
1452−append m1 m2 = fromStreamD $ D.append (toStreamD m1) (toStreamD m2)
1454−-- XXX Same as 'wSerial'. We should perhaps rename wSerial to interleave.
1455−-- XXX Document the interleaving behavior of side effects in all the
1456−-- interleaving combinators.
1457−-- XXX Write time-domain equivalents of these. In the time domain we can
1458−-- interleave two streams such that the value of second stream is always taken
1459−-- from its last value even if no new value is being yielded, like
1460−-- zipWithLatest. It would be something like interleaveWithLatest.
1462−-- | Interleaves the outputs of two streams, yielding elements from each stream
1463−-- alternately, starting from the first stream. If any of the streams finishes
1464−-- early the other stream continues alone until it too finishes.
1466−-- >>> :set -XOverloadedStrings
327+-- drain $ S.replicateM 10 (return $ threadDelay 1000000 >> print 1)
1467−-- >>> interleave "ab" ",,,," :: SerialT Identity Char
328+-- & (serially . S.sequence)
1468−-- fromList "a,b,,,"
1469−-- >>> interleave "abcd" ",," :: SerialT Identity Char
1470−-- fromList "a,b,cd"
1472−-- 'interleave' is dual to 'interleaveMin', it can be called @interleaveMax@.
330+-- drain $ S.replicateM 10 (return $ threadDelay 1000000 >> print 1)
331+-- & (asyncly . S.sequence)
1474−-- Do not use at scale in concatMapWith.
334+-- /Concurrent (do not use with 'parallely' on infinite streams)/
1476−-- /Internal/
336+-- @since 0.1.0
1477−{-# INLINE interleave #-}
337+{-# INLINE sequence #-}
1478−interleave ::(IsStream t, Monad m) => t m b -> t m b -> t m b
338+sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
1479−interleave m1 m2 = fromStreamD $ D.interleave (toStreamD m1) (toStreamD m2)
339+sequence m = fromStreamS $ S.sequence (toStreamS m)
1481−-- | Interleaves the outputs of two streams, yielding elements from each stream
341+------------------------------------------------------------------------------
1482−-- alternately, starting from the first stream. As soon as the first stream
342+-- Mapping side effects
1483−-- finishes, the output stops, discarding the remaining part of the second
343+------------------------------------------------------------------------------
1484−-- stream. In this case, the last element in the resulting stream would be from
1485−-- the second stream. If the second stream finishes early then the first stream
1486−-- still continues to yield elements until it finishes.
1488−-- >>> :set -XOverloadedStrings
1489−-- >>> interleaveSuffix "abc" ",,,," :: SerialT Identity Char
1490−-- fromList "a,b,c,"
1491−-- >>> interleaveSuffix "abc" "," :: SerialT Identity Char
1494−-- 'interleaveSuffix' is a dual of 'interleaveInfix'.
1496−-- Do not use at scale in concatMapWith.
1499−{-# INLINE interleaveSuffix #-}
1500−interleaveSuffix ::(IsStream t, Monad m) => t m b -> t m b -> t m b
1501−interleaveSuffix m1 m2 =
1502− fromStreamD $ D.interleaveSuffix (toStreamD m1) (toStreamD m2)
1504−-- | Interleaves the outputs of two streams, yielding elements from each stream
345+-- | Tap the data flowing through a stream into a 'Fold'. For example, you may
1505−-- alternately, starting from the first stream and ending at the first stream.
346+-- add a tap to log the contents flowing through the stream. The fold is used
1506−-- If the second stream is longer than the first, elements from the second
347+-- only for effects, its result is discarded.
1507−-- stream are infixed with elements from the first stream. If the first stream
1508−-- is longer then it continues yielding elements even after the second stream
1511−-- >>> :set -XOverloadedStrings
1512−-- >>> interleaveInfix "abc" ",,,," :: SerialT Identity Char
1514−-- >>> interleaveInfix "abc" "," :: SerialT Identity Char
1517−-- 'interleaveInfix' is a dual of 'interleaveSuffix'.
1519−-- Do not use at scale in concatMapWith.
1522−{-# INLINE interleaveInfix #-}
1523−interleaveInfix ::(IsStream t, Monad m) => t m b -> t m b -> t m b
1524−interleaveInfix m1 m2 =
1525− fromStreamD $ D.interleaveInfix (toStreamD m1) (toStreamD m2)
1527−-- | Interleaves the outputs of two streams, yielding elements from each stream
1528−-- alternately, starting from the first stream. The output stops as soon as any
1529−-- of the two streams finishes, discarding the remaining part of the other
1530−-- stream. The last element of the resulting stream would be from the longer
1533−-- >>> :set -XOverloadedStrings
1534−-- >>> interleaveMin "ab" ",,,," :: SerialT Identity Char
1536−-- >>> interleaveMin "abcd" ",," :: SerialT Identity Char
1539−-- 'interleaveMin' is dual to 'interleave'.
1541−-- Do not use at scale in concatMapWith.
1543−-- /Internal/
349+-- @
1544−{-# INLINE interleaveMin #-}
350+-- Fold m a b
1545−interleaveMin ::(IsStream t, Monad m) => t m b -> t m b -> t m b
351+-- |
1546−interleaveMin m1 m2 = fromStreamD $ D.interleaveMin (toStreamD m1) (toStreamD m2)
352+-- -----stream m a ---------------stream m a-----
1548−-- | Schedule the execution of two streams in a fair round-robin manner,
1549−-- executing each stream once, alternately. Execution of a stream may not
1550−-- necessarily result in an output, a stream may chose to @Skip@ producing an
1551−-- element until later giving the other stream a chance to run. Therefore, this
1552−-- combinator fairly interleaves the execution of two streams rather than
1553−-- fairly interleaving the output of the two streams. This can be useful in
1554−-- co-operative multitasking without using explicit threads. This can be used
1555−-- as an alternative to `async`.
1557−-- Do not use at scale in concatMapWith.
354+-- @
1559−-- /Internal/
356+-- @
1560−{-# INLINE roundrobin #-}
357+-- > S.drain $ S.tap (FL.drainBy print) (S.enumerateFromTo 1 2)
1561−roundrobin ::(IsStream t, Monad m) => t m b -> t m b -> t m b
358+-- 1
1562−roundrobin m1 m2 = fromStreamD $ D.roundRobin (toStreamD m1) (toStreamD m2)
359+-- 2
1564−-- | Map a stream producing monadic function on each element of the stream
1565−-- and then flatten the results into a single stream. Since the stream
1566−-- generation function is monadic, unlike 'concatMap', it can produce an
1567−-- effect at the beginning of each iteration of the inner loop.
1569−-- @since 0.6.0
362+-- Compare with 'trace'.
1570−{-# INLINE concatMapM #-}
1571−concatMapM :: (IsStream t, Monad m) => (a -> m (t m b)) -> t m a -> t m b
1572−concatMapM f m = fromStreamD $ D.concatMapM (fmap toStreamD . f) (toStreamD m)
1574−-- | Like 'concatMap' but uses an 'Unfold' for stream generation. Unlike
1575−-- 'concatMap' this can fuse the 'Unfold' code with the inner loop and
1576−-- therefore provide many times better performance.
1578-- @since 0.7.0
364-- @since 0.7.0
1579−{-# INLINE concatUnfold #-}
365+{-# INLINE tap #-}
1580−concatUnfold ::(IsStream t, Monad m) => Unfold m a b -> t m a -> t m b
366+tap :: (IsStream t, Monad m) => FL.Fold m a b -> t m a -> t m a
1581−concatUnfold u m = fromStreamD $ D.concatMapU u (toStreamD m)
367+tap f xs = D.fromStreamD $ D.tap f (D.toStreamD xs)
1583−-- | Like 'concatUnfold' but interleaves the streams in the same way as
1584−-- 'interleave' behaves instead of appending them.
1587−{-# INLINE concatUnfoldInterleave #-}
1588−concatUnfoldInterleave ::(IsStream t, Monad m)
1589− => Unfold m a b -> t m a -> t m b
1590−concatUnfoldInterleave u m =
1591− fromStreamD $ D.concatUnfoldInterleave u (toStreamD m)
1593−-- | Like 'concatUnfold' but executes the streams in the same way as
1597−{-# INLINE concatUnfoldRoundrobin #-}
1598−concatUnfoldRoundrobin ::(IsStream t, Monad m)
1599− => Unfold m a b -> t m a -> t m b
1600−concatUnfoldRoundrobin u m =
1601− fromStreamD $ D.concatUnfoldRoundrobin u (toStreamD m)
1603−-- XXX we can swap the order of arguments to gintercalate so that the
1604−-- definition of concatUnfold becomes simpler? The first stream should be
1605−-- infixed inside the second one. However, if we change the order in
1606−-- "interleave" as well similarly, then that will make it a bit unintuitive.
1608−-- > concatUnfold unf str =
1609−-- > gintercalate unf str (UF.nilM (\_ -> return ())) (repeat ())
1611−-- | 'interleaveInfix' followed by unfold and concat.
1614−{-# INLINE gintercalate #-}
1616− :: (IsStream t, Monad m)
1617− => Unfold m a c -> t m a -> Unfold m b c -> t m b -> t m c
1618−gintercalate unf1 str1 unf2 str2 =
1619− D.fromStreamD $ D.gintercalate
1620− unf1 (D.toStreamD str1)
1621− unf2 (D.toStreamD str2)
1623−-- XXX The order of arguments in "intercalate" is consistent with the list
1624−-- intercalate but inconsistent with gintercalate and other stream interleaving
1625−-- combinators. We can change the order of the arguments in other combinators
1626−-- but then 'interleave' combinator may become a bit unintuitive because we
1627−-- will be starting with the second stream.
1629−-- > intercalate seed unf str = gintercalate unf str unf (repeatM seed)
1630−-- > intercalate a unf str = concatUnfold unf $ intersperse a str
1632−-- | 'intersperse' followed by unfold and concat.
1634−-- > unwords = intercalate " " UF.fromList
1636−-- >>> intercalate " " UF.fromList ["abc", "def", "ghi"]
1640−{-# INLINE intercalate #-}
1641−intercalate :: (IsStream t, Monad m)
1642− => b -> Unfold m b c -> t m b -> t m c
1643−intercalate seed unf str = D.fromStreamD $
1644− D.concatMapU unf $ D.intersperse seed (toStreamD str)
1646−-- > interpose x unf str = gintercalate unf str UF.identity (repeat x)
1648−-- | Unfold the elements of a stream, intersperse the given element between the
1649−-- unfolded streams and then concat them into a single stream.
1651−-- > unwords = S.interpose ' '
1654−{-# INLINE interpose #-}
1655−interpose :: (IsStream t, Monad m)
1656− => c -> Unfold m b c -> t m b -> t m c
1657−interpose x unf str =
1658− D.fromStreamD $ D.interpose (return x) unf (D.toStreamD str)
1660−-- | 'interleaveSuffix' followed by unfold and concat.
1663−{-# INLINE gintercalateSuffix #-}
1665− :: (IsStream t, Monad m)
1666− => Unfold m a c -> t m a -> Unfold m b c -> t m b -> t m c
1667−gintercalateSuffix unf1 str1 unf2 str2 =
1668− D.fromStreamD $ D.gintercalateSuffix
1669− unf1 (D.toStreamD str1)
1670− unf2 (D.toStreamD str2)
1672−-- > intercalateSuffix seed unf str = gintercalateSuffix unf str unf (repeatM seed)
1673−-- > intercalateSuffix a unf str = concatUnfold unf $ intersperseSuffix a str
1675−-- | 'intersperseSuffix' followed by unfold and concat.
1677−-- > unlines = intercalateSuffix "\n" UF.fromList
1679−-- >>> intercalate "\n" UF.fromList ["abc", "def", "ghi"]
1680−-- > "abc\ndef\nghi\n"
1683−{-# INLINE intercalateSuffix #-}
1684−intercalateSuffix :: (IsStream t, Monad m)
1685− => b -> Unfold m b c -> t m b -> t m c
1686−intercalateSuffix seed unf str = fromStreamD $ D.concatMapU unf
1687− $ D.intersperseSuffix (return seed) (D.toStreamD str)
1689−-- interposeSuffix x unf str = gintercalateSuffix unf str UF.identity (repeat x)
1691−-- | Unfold the elements of a stream, append the given element after each
1692−-- unfolded stream and then concat them into a single stream.
1694−-- > unlines = S.interposeSuffix '\n'
1697−{-# INLINE interposeSuffix #-}
1698−interposeSuffix :: (IsStream t, Monad m)
1699− => c -> Unfold m b c -> t m b -> t m c
1700−interposeSuffix x unf str =
1701− D.fromStreamD $ D.interposeSuffix (return x) unf (D.toStreamD str)
1703−------------------------------------------------------------------------------
1705−------------------------------------------------------------------------------
1707−-- | Discard 'Right's and unwrap 'Left's in an 'Either' stream.
369+-- | @tapOffsetEvery offset n@ taps every @n@th element in the stream
1708−--
370+-- starting at @offset@. @offset@ can be between @0@ and @n - 1@. Offset 0
1709−-- /Internal/
371+-- means start at the first element in the stream. If the offset is outside
372+-- this range then @offset `mod` n@ is used as offset.
1711−lefts :: (IsStream t, Monad m, Functor (t m)) => t m (Either a b) -> t m a
374+-- @
1712−lefts = fmap (fromLeft undefined) . filter isLeft
375+-- >>> S.drain $ S.tapOffsetEvery 0 2 (FL.mapM print FL.toList) $ S.enumerateFromTo 0 10
1713−
376+-- > [0,2,4,6,8,10]
1714−-- | Discard 'Left's and unwrap 'Right's in an 'Either' stream.
377+-- @
1716-- /Internal/
379-- /Internal/
381+{-# INLINE tapOffsetEvery #-}
1718−rights :: (IsStream t, Monad m, Functor (t m)) => t m (Either a b) -> t m b
382+tapOffsetEvery :: (IsStream t, Monad m)
1719−rights = fmap (fromRight undefined) . filter isRight
383+ => Int -> Int -> FL.Fold m a b -> t m a -> t m a
1720−
384+tapOffsetEvery offset n f xs =
1721−------------------------------------------------------------------------------
385+ D.fromStreamD $ D.tapOffsetEvery offset n f (D.toStreamD xs)
1723−------------------------------------------------------------------------------
1725−-- Splitting operations that take a predicate and a Fold can be
1726−-- expressed using parseMany. Operations like chunksOf, intervalsOf, split*,
1727−-- can be expressed using parseMany when used with an appropriate Parser.
1729−-- XXX We need takeGE/takeBetween to implement "some" using "many".
1731−-- | Apply a 'Fold' repeatedly on a stream and emit the parsed values in the
387+-- | Redirect a copy of the stream to a supplied fold and run it concurrently
1732−-- output stream.
388+-- in an independent thread. The fold may buffer some elements. The buffer size
1733−--
389+-- is determined by the prevailing 'maxBuffer' setting.
1734−-- This is the streaming dual of the 'Streamly.Internal.Data.Fold.many'
1735−-- parse combinator.
1737−-- >>> f = Fold.ltake 2 Fold.sum
1738−-- >>> Stream.toList $ Stream.foldMany f $ Stream.fromList [1..10]
1741−-- >>> f = Fold.sliceEndWith Fold.toList
1742−-- >>> Stream.toList $ Stream.foldMany f $ Stream.fromList "hello\nworld"
1743−-- > ["hello\n","world"]
1745−-- /Internal/
391+-- @
394+-- -----stream m a ---------------stream m a-----
1747−{-# INLINE foldMany #-}
396+-- @
1749− :: (IsStream t, Monad m)
1753−foldMany f m = D.fromStreamD $ D.foldMany f (D.toStreamD m)
1755−-- | Apply a stream of folds to an input stream and emit the results in the
1758−-- /Internal/
398+-- @
399+-- > S.drain $ S.tapAsync (S.mapM_ print) (S.enumerateFromTo 1 2)
1760−{-# INLINE foldSequence #-}
404+-- Exceptions from the concurrently running fold are propagated to the current
1761−foldSequence
405+-- computation. Note that, because of buffering in the fold, exceptions may be
1762− :: -- (IsStream t, Monad m) =>
406+-- delayed and may not correspond to the current element being processed in the
1763− t m (Fold m a b)
407+-- parent stream, but we guarantee that before the parent stream stops the tap
1764− -> t m a
408+-- finishes and all exceptions from it are drained.
1766−foldSequence _f _m = undefined
1768−-- | Iterate a fold generator on a stream. The initial value @b@ is used to
1769−-- generate the first fold, the fold is applied on the stream and the result of
1770−-- the fold is used to generate the next fold and so on.
1772−-- >>> f x = Fold.ltake 2 (Fold.mconcatTo x)
1773−-- >>> s = Stream.map Sum $ Stream.fromList [1..10]
1774−-- >>> Stream.toList $ Stream.map getSum $ Stream.foldIterate f 0 s
1775−-- > [3,10,21,36,55,55]
1777−-- This is the streaming equivalent of monad like sequenced application of
411+-- Compare with 'tap'.
1778−-- folds where next fold is dependent on the previous fold.
1780-- /Internal/
413-- /Internal/
1781−--
414+{-# INLINE tapAsync #-}
1782−{-# INLINE foldIterate #-}
415+tapAsync :: (IsStream t, MonadAsync m) => FL.Fold m a b -> t m a -> t m a
1784− :: -- (IsStream t, Monad m) =>
1789−foldIterate _f _i _m = undefined
1790−-- D.fromStreamD $ D.foldIterate f i (D.toStreamD m)
416+tapAsync f xs = D.fromStreamD $ D.tapAsync f (D.toStreamD xs)
1792−-- | Apply a 'Parser' repeatedly on a stream and emit the parsed values in the
418+-- | @pollCounts predicate transform fold stream@ counts those elements in the
1793−-- output stream.
419+-- stream that pass the @predicate@. The resulting count stream is sent to
420+-- another thread which transforms it using @transform@ and then folds it using
421+-- @fold@. The thread is automatically cleaned up if the stream stops or
422+-- aborts due to exception.
1795−-- This is the streaming equivalent of the 'Streamly.Internal.Data.Parser.many'
424+-- For example, to print the count of elements processed every second:
1796−-- parse combinator.
1798−-- >>> S.toList $ S.parseMany (PR.take 2 $ PR.fromFold FL.sum) $ S.fromList [1..10]
426+-- @
1799−-- > [3,7,11,15,19]
427+-- > S.drain $ S.pollCounts (const True) (S.rollingMap (-) . S.delayPost 1) (FL.drainBy print)
428+-- $ S.enumerateFrom 0
1801−-- >>> S.toList $ S.parseMany (PR.line FL.toList) $ S.fromList "hello\nworld"
431+-- Note: This may not work correctly on 32-bit machines.
1802−-- > ["hello\n","world"]
1804-- /Internal/
433-- /Internal/
1806−{-# INLINE parseMany #-}
435+{-# INLINE pollCounts #-}
1807−parseMany
436+pollCounts ::
1808− :: (IsStream t, MonadThrow m)
437+ (IsStream t, MonadAsync m)
1809− => Parser m a b
438+ => (a -> Bool)
1810− -> t m a
439+ -> (t m Int -> t m Int)
1811− -> t m b
440+ -> Fold m Int b
1813− D.fromStreamD $ D.parseMany (PRK.fromParserK p) (D.toStreamD m)
1815−{-# INLINE parseManyD #-}
1817− :: (IsStream t, MonadThrow m)
1818− => PRD.Parser m a b
1819 -> t m a
441 -> t m a
1822− D.fromStreamD $ D.parseMany p (D.toStreamD m)
1824−-- | Apply a stream of parsers to an input stream and emit the results in the
1829−{-# INLINE parseSequence #-}
1831− :: -- (IsStream t, Monad m) =>
1833 -> t m a
442 -> t m a
1834− -> t m b
443+pollCounts predicate transf f xs =
1835−parseSequence _f _m = undefined
444+ D.fromStreamD
445+ $ D.pollCounts predicate (D.toStreamD . transf . D.fromStreamD) f
1837−-- | @parseManyTill collect test stream@ tries the parser @test@ on the input,
448+-- | Calls the supplied function with the number of elements consumed
1838−-- if @test@ fails it backtracks and tries @collect@, after @collect@ succeeds
449+-- every @n@ seconds. The given function is run in a separate thread
1839−-- @test@ is tried again and so on. The parser stops when @test@ succeeds. The
450+-- until the end of the stream. In case there is an exception in the
1840−-- output of @test@ is discarded and the output of @collect@ is emitted in the
451+-- stream the thread is killed during the next major GC.
1841−-- output stream. The parser fails if @collect@ fails.
1845−{-# INLINE parseManyTill #-}
453+-- Note: The action is not guaranteed to run if the main thread exits.
1847− -- (IsStream t, MonadThrow m) =>
1852−parseManyTill = undefined
1854−-- | Iterate a parser generating function on a stream. The initial value @b@ is
1855−-- used to generate the first parser, the parser is applied on the stream and
1856−-- the result is used to generate the next parser and so on.
1858−-- >>> S.toList $ S.map getSum $ S.parseIterate (\b -> PR.take 2 (FL.mconcatTo b)) 0 $ S.map Sum $ S.fromList [1..10]
455+-- @
1859−-- > [3,10,21,36,55,55]
456+-- > delay n = threadDelay (round $ n * 1000000) >> return n
457+-- > S.drain $ S.tapRate 2 (\\n -> print $ show n ++ " elements processed") (delay 1 S.|: delay 0.5 S.|: delay 0.5 S.|: S.nil)
458+-- 2 elements processed
459+-- 1 elements processed
1861−-- This is the streaming equivalent of monad like sequenced application of
462+-- Note: This may not work correctly on 32-bit machines.
1862−-- parsers where next parser is dependent on the previous parser.
1864-- /Internal/
464-- /Internal/
1865−--
465+{-# INLINE tapRate #-}
1866−{-# INLINE parseIterate #-}
466+tapRate ::
1867−parseIterate
467+ (IsStream t, MonadAsync m, MonadCatch m)
1868− :: (IsStream t, MonadThrow m)
468+ => Double
1869− => (b -> Parser m a b)
469+ -> (Int -> m b)
1871− -> t m a
470+ -> t m a
1872− -> t m b
471+ -> t m a
1873−parseIterate f i m = D.fromStreamD $
472+tapRate n f xs = D.fromStreamD $ D.tapRate n f $ D.toStreamD xs
1874− D.parseIterate (PRK.fromParserK . f) i (D.toStreamD m)
1876−------------------------------------------------------------------------------
1877−-- Grouping/Splitting
1878−------------------------------------------------------------------------------
1880−------------------------------------------------------------------------------
1881−-- Grouping without looking at elements
1882−------------------------------------------------------------------------------
1884−------------------------------------------------------------------------------
1886−------------------------------------------------------------------------------
1889−------------------------------------------------------------------------------
1891−------------------------------------------------------------------------------
1893−------------------------------------------------------------------------------
1894−-- Generalized grouping
1895−------------------------------------------------------------------------------
1897−-- This combinator is the most general grouping combinator and can be used to
1898−-- implement all other grouping combinators.
1900−-- XXX check if this can implement the splitOn combinator i.e. we can slide in
1901−-- new elements, slide out old elements and incrementally compute the hash.
1902−-- Also, can we implement the windowed classification combinators using this?
1904−-- In fact this is a parse. Instead of using a special return value in the fold
1905−-- we are using a mapping function.
1907−-- Note that 'scanl'' (usually followed by a map to extract the desired value
1908−-- from the accumulator) can be used to realize many implementations e.g. a
1909−-- sliding window implementation. A scan followed by a mapMaybe is also a good
1910−-- pattern to express many problems where we want to emit a filtered output and
1911−-- not emit an output on every input.
1913−-- Passing on of the initial accumulator value to the next fold is equivalent
1914−-- to returning the leftover concept.
1917−-- | @groupScan splitter fold stream@ folds the input stream using @fold@.
1918−-- @splitter@ is applied on the accumulator of the fold every time an item is
1919−-- consumed by the fold. The fold continues until @splitter@ returns a 'Just'
1920−-- value. A 'Just' result from the @splitter@ specifies a result to be emitted
1921−-- in the output stream and the initial value of the accumulator for the next
1922−-- group's fold. This allows us to control whether to start fresh for the next
1923−-- fold or to continue from the previous fold's output.
1925−{-# INLINE groupScan #-}
1927− :: (IsStream t, Monad m)
1928− => (x -> m (Maybe (b, x))) -> Fold m a x -> t m a -> t m b
1929−groupScan split fold m = undefined
1932−-- | Group the input stream into groups of @n@ elements each and then fold each
474+-- | Apply a monadic function to each element flowing through the stream and
1933−-- group using the provided fold function.
475+-- discard the results.
1935−-- >> S.toList $ S.chunksOf 2 FL.sum (S.enumerateFromTo 1 10)
478+-- > S.drain $ S.trace print (S.enumerateFromTo 1 2)
1936−-- > [3,7,11,15,19]
479+-- 1
1938−-- This can be considered as an n-fold version of 'ltake' where we apply
483+-- Compare with 'tap'.
1939−-- 'ltake' repeatedly on the leftover stream until the stream exhausts.
1941-- @since 0.7.0
485-- @since 0.7.0
1942−{-# INLINE chunksOf #-}
486+{-# INLINE trace #-}
1944− :: (IsStream t, Monad m)
487+trace :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m a
1945− => Int -> Fold m a b -> t m a -> t m b
488+trace f = mapM (\x -> void (f x) >> return x)
1946−chunksOf n f = foldMany (FL.ltake n f)
1951−{-# INLINE chunksOf2 #-}
1953− :: (IsStream t, Monad m)
1954− => Int -> m c -> Fold2 m c a b -> t m a -> t m b
1955−chunksOf2 n action f m = D.fromStreamD $ D.groupsOf2 n action f (D.toStreamD m)
1957−-- | @arraysOf n stream@ groups the elements in the input stream into arrays of
490+-- | Perform a side effect before yielding each element of the stream and
1958−-- @n@ elements each.
491+-- discard the results.
1960−-- Same as the following but may be more efficient:
493+-- @
494+-- > S.drain $ S.trace_ (print "got here") (S.enumerateFromTo 1 2)
1962−-- > arraysOf n = S.chunksOf n (A.writeN n)
499+-- See also: 'trace', 'interspersePrefix_'
1964-- /Internal/
501-- /Internal/
1965−{-# INLINE arraysOf #-}
502+{-# INLINE trace_ #-}
1966−arraysOf :: (IsStream t, MonadIO m, Storable a)
503+trace_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
1967− => Int -> t m a -> t m (Array a)
504+trace_ eff = Serial.mapM (\x -> eff >> return x)
1968−arraysOf n = chunksOf n (writeNUnsafe n)
1970------------------------------------------------------------------------------
506------------------------------------------------------------------------------
1971−-- N-ary APIs
507+-- Scanning with a Fold
1972------------------------------------------------------------------------------
508------------------------------------------------------------------------------
1974−-- | @groupsBy cmp f $ S.fromList [a,b,c,...]@ assigns the element @a@ to the
510+-- | Scan a stream using the given monadic fold.
1975−-- first group, if @b \`cmp` a@ is 'True' then @b@ is also assigned to the same
1976−-- group. If @c \`cmp` a@ is 'True' then @c@ is also assigned to the same
1977−-- group and so on. When the comparison fails a new group is started. Each
1978−-- group is folded using the fold @f@ and the result of the fold is emitted in
1979−-- the output stream.
1981−-- >>> S.toList $ S.groupsBy (>) FL.toList $ S.fromList [1,3,7,0,2,5]
1982−-- > [[1,3,7],[0,2,5]]
1985−{-# INLINE groupsBy #-}
1987− :: (IsStream t, Monad m)
1988− => (a -> a -> Bool)
1992−groupsBy cmp f m = D.fromStreamD $ D.groupsBy cmp f (D.toStreamD m)
1994−-- | Unlike @groupsBy@ this function performs a rolling comparison of two
1995−-- successive elements in the input stream. @groupsByRolling cmp f $ S.fromList
1996−-- [a,b,c,...]@ assigns the element @a@ to the first group, if @a \`cmp` b@ is
1997−-- 'True' then @b@ is also assigned to the same group. If @b \`cmp` c@ is
1998−-- 'True' then @c@ is also assigned to the same group and so on. When the
1999−-- comparison fails a new group is started. Each group is folded using the fold
2002−-- >>> S.toList $ S.groupsByRolling (\a b -> a + 1 == b) FL.toList $ S.fromList [1,2,3,7,8,9]
2003−-- > [[1,2,3],[7,8,9]]
2005-- @since 0.7.0
512-- @since 0.7.0
2006−{-# INLINE groupsByRolling #-}
513+{-# INLINE scan #-}
2008− :: (IsStream t, Monad m)
514+scan :: (IsStream t, Monad m) => Fold m a b -> t m a -> t m b
2009− => (a -> a -> Bool)
515+scan = P.scanOnce
2013−groupsByRolling cmp f m = D.fromStreamD $ D.groupsRollingBy cmp f (D.toStreamD m)
2015−-- |
517+-- | Postscan a stream using the given monadic fold.
2016−-- > groups = groupsBy (==)
2017−-- > groups = groupsByRolling (==)
2019−-- Groups contiguous spans of equal elements together in individual groups.
2021−-- >>> S.toList $ S.groups FL.toList $ S.fromList [1,1,2,2]
2024-- @since 0.7.0
519-- @since 0.7.0
2025−{-# INLINE groups #-}
520+{-# INLINE postscan #-}
2026−groups :: (IsStream t, Monad m, Eq a) => Fold m a b -> t m a -> t m b
521+postscan :: (IsStream t, Monad m) => Fold m a b -> t m a -> t m b
2027−groups = groupsBy (==)
522+postscan = P.postscanOnce
2029------------------------------------------------------------------------------
524------------------------------------------------------------------------------
2030−-- N-ary split on a predicate
525+-- Scanning - Transformation by Folding
2031------------------------------------------------------------------------------
526------------------------------------------------------------------------------
2033−-- TODO: Use a Splitter configuration similar to the "split" package to make it
528+-- XXX It may be useful to have a version of scan where we can keep the
2034−-- possible to express all splitting combinations. In general, we can have
529+-- accumulator independent of the value emitted. So that we do not necessarily
2035−-- infix/suffix/prefix/condensing of separators, dropping both leading/trailing
530+-- have to keep a value in the accumulator which we are not using. We can pass
2036−-- separators. We can have a single split operation taking the splitter config
531+-- an extraction function that will take the accumulator and the current value
2037−-- as argument.
532+-- of the element and emit the next value in the stream. That will also make it
2038−
533+-- possible to modify the accumulator after using it. In fact, the step function
2039−-- | Split on an infixed separator element, dropping the separator. Splits the
534+-- can return new accumulator and the value to be emitted. The signature would
2040−-- stream on separator elements determined by the supplied predicate, separator
535+-- be more like mapAccumL. Or we can change the signature of scanx to
2041−-- is considered as infixed between two segments, if one side of the separator
536+-- accommodate this.
2042−-- is missing then it is parsed as an empty stream. The supplied 'Fold' is
2043−-- applied on the split segments. With '-' representing non-separator elements
2044−-- and '.' as separator, 'splitOn' splits as follows:
2047−-- "--.--" => "--" "--"
2052−-- @splitOn (== x)@ is an inverse of @intercalate (S.yield x)@
538+-- | Strict left scan with an extraction function. Like 'scanl'', but applies a
539+-- user supplied extraction function (the third argument) at each step. This is
540+-- designed to work with the @foldl@ library. The suffix @x@ is a mnemonic for
2054−-- Let's use the following definition for illustration:
543+-- /Since 0.2.0/
2056−-- > splitOn' p xs = S.toList $ S.splitOn p (FL.toList) (S.fromList xs)
545+-- /Since: 0.7.0 (Monad m constraint)/
546+{-# DEPRECATED scanx "Please use scanl followed by map instead." #-}
548+scanx :: (IsStream t, Monad m) => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
551+-- XXX this needs to be concurrent
552+-- XXX because of the use of D.cons for appending, scanlM' has quadratic
553+-- complexity when iterated over a stream. We should use StreamK style scanlM'
554+-- for linear performance on iteration.
2058−-- >>> splitOn' (== '.') ""
556+-- | Like 'scanl'' but with a monadic step function and a monadic seed.
2061−-- >>> splitOn' (== '.') "."
558+-- /Since: 0.4.0/
2064−-- >>> splitOn' (== '.') ".a"
560+-- /Since: 0.8.0 (signature change)/
2065−-- > ["","a"]
561+{-# INLINE scanlM' #-}
562+scanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
563+scanlM' step begin m = fromStreamD $ D.scanlM' step begin $ toStreamD m
565+-- XXX because of the use of D.cons for appending, scanl' has quadratic
566+-- complexity when iterated over a stream. We should use StreamK style scanlM'
567+-- for linear performance on iteration.
569+-- | Strict left scan. Like 'map', 'scanl'' too is a one to one transformation,
570+-- however it adds an extra element.
2067−-- >>> splitOn' (== '.') "a."
572+-- @
2068−-- > ["a",""]
573+-- > S.toList $ S.scanl' (+) 0 $ fromList [1,2,3,4]
2070−-- >>> splitOn' (== '.') "a.b"
577+-- @
2071−-- > ["a","b"]
578+-- > S.toList $ S.scanl' (flip (:)) [] $ S.fromList [1,2,3,4]
579+-- [[],[1],[2,1],[3,2,1],[4,3,2,1]]
2073−-- >>> splitOn' (== '.') "a..b"
582+-- The output of 'scanl'' is the initial value of the accumulator followed by
2074−-- > ["a","","b"]
583+-- all the intermediate steps and the final result of 'foldl''.
2076−-- @since 0.7.0
585+-- By streaming the accumulated state after each fold step, we can share the
2077−
586+-- state across multiple stages of stream composition. Each stage can modify or
2078−-- This can be considered as an n-fold version of 'breakOn' where we apply
587+-- extend the state, do some processing with it and emit it for the next stage,
2079−-- 'breakOn' successively on the input stream, dropping the first element
588+-- thus modularizing the stream processing. This can be useful in
2080−-- of the second segment after each break.
589+-- stateful or event-driven programming.
2082−{-# INLINE splitOn #-}
591+-- Consider the following monolithic example, computing the sum and the product
2083−splitOn
592+-- of the elements in a stream in one go using a @foldl'@:
2084− :: (IsStream t, Monad m)
2085− => (a -> Bool) -> Fold m a b -> t m a -> t m b
2086−splitOn predicate f = foldMany (FL.sliceSepBy predicate f)
2088−-- | Like 'splitOn' but the separator is considered as suffixed to the segments
2089−-- in the stream. A missing suffix at the end is allowed. A separator at the
2090−-- beginning is parsed as empty segment. With '-' representing elements and
2091−-- '.' as separator, 'splitOnSuffix' splits as follows:
2094−-- "--.--." => "--" "--"
595+-- > S.foldl' (\\(s, p) x -> (s + x, p * x)) (0,1) $ S.fromList \[1,2,3,4]
2095−-- "--.--" => "--" "--"
596+-- (10,24)
2096−-- ".--." => "" "--"
2099−-- > splitOnSuffix' p xs = S.toList $ S.splitSuffixBy p (FL.toList) (S.fromList xs)
599+-- Using @scanl'@ we can make it modular by computing the sum in the first
2100−--
600+-- stage and passing it down to the next stage for computing the product:
2101−-- >>> splitOnSuffix' (== '.') ""
2104−-- >>> splitOnSuffix' (== '.') "."
602+-- @
2105−-- [""]
603+-- > S.foldl' (\\(_, p) (s, x) -> (s, p * x)) (0,1)
604+-- $ S.scanl' (\\(s, _) x -> (s + x, x)) (0,1)
605+-- $ S.fromList \[1,2,3,4]
2107−-- >>> splitOnSuffix' (== '.') "a"
609+-- IMPORTANT: 'scanl'' evaluates the accumulator to WHNF. To avoid building
2108−-- ["a"]
610+-- lazy expressions inside the accumulator, it is recommended that a strict
611+-- data structure is used for accumulator.
2110−-- >>> splitOnSuffix' (== '.') ".a"
613+-- See also: 'usingStateT'
2113−-- >>> splitOnSuffix' (== '.') "a."
615+-- @since 0.2.0
2114−-- > ["a"]
616+{-# INLINE scanl' #-}
617+scanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
618+scanl' step z m = fromStreamS $ S.scanl' step z $ toStreamS m
620+-- | Like 'scanl'' but does not stream the initial value of the accumulator.
2116−-- >>> splitOnSuffix' (== '.') "a.b"
622+-- > postscanl' f z xs = S.drop 1 $ S.scanl' f z xs
2119−-- >>> splitOnSuffix' (== '.') "a.b."
624+-- @since 0.7.0
2120−-- > ["a","b"]
625+{-# INLINE postscanl' #-}
626+postscanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
627+postscanl' step z m = fromStreamD $ D.postscanl' step z $ toStreamD m
629+-- XXX prescanl does not sound very useful, enable only if there is a
630+-- compelling use case.
2122−-- >>> splitOnSuffix' (== '.') "a..b.."
632+-- | Like scanl' but does not stream the final value of the accumulator.
2123−-- > ["a","","b",""]
2125−-- > lines = splitOnSuffix (== '\n')
634+-- /Internal/
635+{-# INLINE prescanl' #-}
636+prescanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
637+prescanl' step z m = fromStreamD $ D.prescanl' step z $ toStreamD m
639+-- XXX this needs to be concurrent
640+-- | Like prescanl' but with a monadic step function and a monadic seed.
2127−-- @since 0.7.0
642+-- /Internal/
643+{-# INLINE prescanlM' #-}
644+prescanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
645+prescanlM' step z m = fromStreamD $ D.prescanlM' step z $ toStreamD m
2129−-- This can be considered as an n-fold version of 'breakPost' where we apply
647+-- XXX this needs to be concurrent
2130−-- 'breakPost' successively on the input stream, dropping the first element
648+-- | Like 'scanl1'' but with a monadic step function.
2131−-- of the second segment after each break.
2133−{-# INLINE splitOnSuffix #-}
650+-- @since 0.6.0
2134−splitOnSuffix
651+{-# INLINE scanl1M' #-}
2135− :: (IsStream t, Monad m)
652+scanl1M' :: (IsStream t, Monad m) => (a -> a -> m a) -> t m a -> t m a
2136− => (a -> Bool) -> Fold m a b -> t m a -> t m b
2137−splitOnSuffix predicate f m =
2138− D.fromStreamD $ D.foldMany1 (FL.sliceSepBy predicate f) (D.toStreamD m)
653+scanl1M' step m = fromStreamD $ D.scanl1M' step $ toStreamD m
2140−-- | Like 'splitOn' after stripping leading, trailing, and repeated separators.
655+-- | Like 'scanl'' but for a non-empty stream. The first element of the stream
2141−-- Therefore, @".a..b."@ with '.' as the separator would be parsed as
656+-- is used as the initial value of the accumulator. Does nothing if the stream
2142−-- @["a","b"]@. In other words, its like parsing words from whitespace
657+-- is empty.
2143−-- separated text.
658+--
660+-- > S.toList $ S.scanl1 (+) $ fromList [1,2,3,4]
2145−-- > wordsBy' p xs = S.toList $ S.wordsBy p (FL.toList) (S.fromList xs)
664+-- @since 0.6.0
665+{-# INLINE scanl1' #-}
666+scanl1' :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> t m a
667+scanl1' step m = fromStreamD $ D.scanl1' step $ toStreamD m
669+------------------------------------------------------------------------------
671+------------------------------------------------------------------------------
673+-- | Include only those elements that pass a predicate.
2147−-- >>> wordsBy' (== ',') ""
675+-- @since 0.1.0
2148−-- > []
676+{-# INLINE filter #-}
677+#if __GLASGOW_HASKELL__ != 802
678+-- GHC 8.2.2 crashes with this code, when used with "stack"
679+filter :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
680+filter p m = fromStreamS $ S.filter p $ toStreamS m
682+filter :: IsStream t => (a -> Bool) -> t m a -> t m a
686+-- | Same as 'filter' but with a monadic predicate.
2150−-- >>> wordsBy' (== ',') ","
688+-- @since 0.4.0
2151−-- > []
689+{-# INLINE filterM #-}
690+filterM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
691+filterM p m = fromStreamD $ D.filterM p $ toStreamD m
693+-- | Drop repeated elements that are adjacent to each other.
2153−-- >>> wordsBy' (== ',') ",a,,b,"
695+-- @since 0.6.0
2154−-- > ["a","b"]
696+{-# INLINE uniq #-}
697+uniq :: (Eq a, IsStream t, Monad m) => t m a -> t m a
698+uniq = fromStreamD . D.uniq . toStreamD
700+-- | Deletes the first occurrence of the element in the stream that satisfies
701+-- the given equality predicate.
2156−-- > words = wordsBy isSpace
703+-- @
704+-- > S.toList $ S.deleteBy (==) 3 $ S.fromList [1,3,3,5]
2158−-- @since 0.7.0
708+-- @since 0.6.0
709+{-# INLINE deleteBy #-}
710+deleteBy :: (IsStream t, Monad m) => (a -> a -> Bool) -> a -> t m a -> t m a
711+deleteBy cmp x m = fromStreamS $ S.deleteBy cmp x (toStreamS m)
2160−-- It is equivalent to splitting in any of the infix/prefix/suffix styles
713+------------------------------------------------------------------------------
2161−-- followed by removal of empty segments.
714+-- Trimming
2162−{-# INLINE wordsBy #-}
715+------------------------------------------------------------------------------
2164− :: (IsStream t, Monad m)
2165− => (a -> Bool) -> Fold m a b -> t m a -> t m b
2166−wordsBy predicate f m =
2167− D.fromStreamD $ D.wordsBy predicate f (D.toStreamD m)
2169−-- | Like 'splitOnSuffix' but keeps the suffix attached to the resulting
717+-- | Same as 'takeWhile' but with a monadic predicate.
2172−-- > splitWithSuffix' p xs = S.toList $ S.splitWithSuffix p (FL.toList) (S.fromList xs)
719+-- @since 0.4.0
720+{-# INLINE takeWhileM #-}
721+takeWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
722+takeWhileM p m = fromStreamD $ D.takeWhileM p $ toStreamD m
724+-- | @takeByTime duration@ yields stream elements upto specified time
725+-- @duration@. The duration starts when the stream is evaluated for the first
726+-- time, before the first element is yielded. The time duration is checked
727+-- before generating each element, if the duration has expired the stream
2174−-- >>> splitWithSuffix' (== '.') ""
730+-- The total time taken in executing the stream is guaranteed to be /at least/
2175−-- []
731+-- @duration@, however, because the duration is checked before generating an
732+-- element, the upper bound is indeterminate and depends on the time taken in
733+-- generating and processing the last element.
2177−-- >>> splitWithSuffix' (== '.') "."
735+-- No element is yielded if the duration is zero. At least one element is
2178−-- ["."]
736+-- yielded if the duration is non-zero.
2180−-- >>> splitWithSuffix' (== '.') "a"
738+-- /Internal/
2183−-- >>> splitWithSuffix' (== '.') ".a"
740+{-# INLINE takeByTime #-}
2184−-- > [".","a"]
741+takeByTime ::(MonadIO m, IsStream t, TimeUnit64 d) => d -> t m a -> t m a
742+takeByTime d = fromStreamD . D.takeByTime d . toStreamD
744+-- | Drop elements in the stream as long as the predicate succeeds and then
745+-- take the rest of the stream.
2186−-- >>> splitWithSuffix' (== '.') "a."
747+-- @since 0.1.0
2187−-- > ["a."]
748+{-# INLINE dropWhile #-}
749+dropWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
750+dropWhile p m = fromStreamS $ S.dropWhile p $ toStreamS m
752+-- | Same as 'dropWhile' but with a monadic predicate.
2189−-- >>> splitWithSuffix' (== '.') "a.b"
754+-- @since 0.4.0
2190−-- > ["a.","b"]
755+{-# INLINE dropWhileM #-}
756+dropWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
757+dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m
759+-- | @dropByTime duration@ drops stream elements until specified @duration@ has
760+-- passed. The duration begins when the stream is evaluated for the first
761+-- time. The time duration is checked /after/ generating a stream element, the
762+-- element is yielded if the duration has expired otherwise it is dropped.
2192−-- >>> splitWithSuffix' (== '.') "a.b."
764+-- The time elapsed before starting to generate the first element is /at most/
2193−-- > ["a.","b."]
765+-- @duration@, however, because the duration expiry is checked after the
766+-- element is generated, the lower bound is indeterminate and depends on the
767+-- time taken in generating an element.
2195−-- >>> splitWithSuffix' (== '.') "a..b.."
769+-- All elements are yielded if the duration is zero.
2196−-- > ["a.",".","b.","."]
2198−-- @since 0.7.0
771+-- /Internal/
2200−-- This can be considered as an n-fold version of 'breakPost' where we apply
2201−-- 'breakPost' successively on the input stream.
2203−{-# INLINE splitWithSuffix #-}
773+{-# INLINE dropByTime #-}
2204−splitWithSuffix
774+dropByTime ::(MonadIO m, IsStream t, TimeUnit64 d) => d -> t m a -> t m a
2205− :: (IsStream t, Monad m)
2206− => (a -> Bool) -> Fold m a b -> t m a -> t m b
2207−splitWithSuffix predicate f m =
2208− D.fromStreamD $ D.foldMany1 (FL.sliceEndWith predicate f) (D.toStreamD m)
775+dropByTime d = fromStreamD . D.dropByTime d . toStreamD
2210------------------------------------------------------------------------------
777------------------------------------------------------------------------------
2211−-- Split on a delimiter sequence
778+-- Inserting Elements
2212------------------------------------------------------------------------------
779------------------------------------------------------------------------------
2214−-- Int list examples for splitOn:
781+-- | @insertBy cmp elem stream@ inserts @elem@ before the first element in
2215−--
782+-- @stream@ that is less than @elem@ when compared using @cmp@.
2216−-- >>> splitList [] [1,2,3,3,4]
2217−-- > [[1],[2],[3],[3],[4]]
2219−-- >>> splitList [5] [1,2,3,3,4]
2222−-- >>> splitList [1] [1,2,3,3,4]
2225−-- >>> splitList [4] [1,2,3,3,4]
2228−-- >>> splitList [2] [1,2,3,3,4]
2231−-- >>> splitList [3] [1,2,3,3,4]
784+-- @
2232−-- > [[1,2],[],[4]]
785+-- insertBy cmp x = 'mergeBy' cmp ('yield' x)
2234−-- >>> splitList [3,3] [1,2,3,3,4]
788+-- @
789+-- > S.toList $ S.insertBy compare 2 $ S.fromList [1,3,5]
2235−-- > [[1,2],[4]]
790+-- [1,2,3,5]
2237−-- >>> splitList [1,2,3,3,4] [1,2,3,3,4]
793+-- @since 0.6.0
2238−-- > [[],[]]
794+{-# INLINE insertBy #-}
796+ (IsStream t, Monad m) => (a -> a -> Ordering) -> a -> t m a -> t m a
797+insertBy cmp x m = fromStreamS $ S.insertBy cmp x (toStreamS m)
2240−-- | Like 'splitOn' but the separator is a sequence of elements instead of a
799+-- | Insert a pure value between successive elements of a stream.
2243−-- For illustration, let's define a function that operates on pure lists:
2246−-- splitOnSeq' pat xs = S.toList $ S.splitOnSeq (A.fromList pat) (FL.toList) (S.fromList xs)
802+-- > S.toList $ S.intersperse ',' $ S.fromList "hello"
2249−-- >>> splitOnSeq' "" "hello"
806+-- @since 0.7.0
2250−-- > ["h","e","l","l","o"]
807+{-# INLINE intersperse #-}
2251−--
808+intersperse :: (IsStream t, MonadAsync m) => a -> t m a -> t m a
2252−-- >>> splitOnSeq' "hello" ""
809+intersperse a = fromStreamS . S.intersperse a . toStreamS
2255−-- >>> splitOnSeq' "hello" "hello"
2258−-- >>> splitOnSeq' "x" "hello"
2261−-- >>> splitOnSeq' "h" "hello"
2264−-- >>> splitOnSeq' "o" "hello"
2267−-- >>> splitOnSeq' "e" "hello"
2270−-- >>> splitOnSeq' "l" "hello"
2273−-- >>> splitOnSeq' "ll" "hello"
2276−-- 'splitOnSeq' is an inverse of 'intercalate'. The following law always holds:
2278−-- > intercalate . splitOn == id
2280−-- The following law holds when the separator is non-empty and contains none of
2281−-- the elements present in the input lists:
2283−-- > splitOn . intercalate == id
2287−-- XXX We can use a polymorphic vector implemented by Array# to represent the
2288−-- sequence, that way we can avoid the Storable constraint. If we still need
2289−-- Storable Array for performance, we can use a separate splitOnArray API for
2290−-- that. We can also have an API where the sequence itself is a lazy stream, so
2291−-- that we can search files in files for example.
2292−{-# INLINE splitOnSeq #-}
2294− :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
2295− => Array a -> Fold m a b -> t m a -> t m b
2296−splitOnSeq patt f m = D.fromStreamD $ D.splitOnSeq patt f (D.toStreamD m)
2299−-- This can be implemented easily using Rabin Karp
2300−-- | Split on any one of the given patterns.
2301−{-# INLINE splitOnAny #-}
2303− :: (IsStream t, Monad m, Storable a, Integral a)
2304− => [Array a] -> Fold m a b -> t m a -> t m b
2305−splitOnAny subseq f m = undefined -- D.fromStreamD $ D.splitOnAny f subseq (D.toStreamD m)
2308−-- | Like 'splitSuffixBy' but the separator is a sequence of elements, instead
811+-- | Insert a side effect before consuming an element of a stream except the
2309−-- of a predicate for a single element.
812+-- first one.
2311−-- > splitOnSuffixSeq_ pat xs = S.toList $ S.splitOnSuffixSeq (A.fromList pat) (FL.toList) (S.fromList xs)
2313−-- >>> splitOnSuffixSeq_ "." ""
2316−-- >>> splitOnSuffixSeq_ "." "."
2319−-- >>> splitOnSuffixSeq_ "." "a"
2322−-- >>> splitOnSuffixSeq_ "." ".a"
2325−-- >>> splitOnSuffixSeq_ "." "a."
2328−-- >>> splitOnSuffixSeq_ "." "a.b"
2331−-- >>> splitOnSuffixSeq_ "." "a.b."
2334−-- >>> splitOnSuffixSeq_ "." "a..b.."
2335−-- > ["a","","b",""]
2337−-- > lines = splitOnSuffixSeq "\n"
2339−-- 'splitOnSuffixSeq' is an inverse of 'intercalateSuffix'. The following law
2342−-- > intercalateSuffix . splitOnSuffixSeq == id
2344−-- The following law holds when the separator is non-empty and contains none of
2345−-- the elements present in the input lists:
2347−-- > splitSuffixOn . intercalateSuffix == id
814+-- @
815+-- >>> S.drain $ S.trace putChar $ S.intersperseM_ (putChar '.') $ S.fromList "hello"
2349-- /Internal/
819-- /Internal/
2350−{-# INLINE splitOnSuffixSeq #-}
820+{-# INLINE intersperseM_ #-}
2351−splitOnSuffixSeq
821+intersperseM_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
2352− :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
2353− => Array a -> Fold m a b -> t m a -> t m b
2354−splitOnSuffixSeq patt f m =
2355− D.fromStreamD $ D.splitOnSuffixSeq False patt f (D.toStreamD m)
822+intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
2358−-- | Like 'splitOn' but drops any empty splits.
825+-- | Intersperse a monadic action into the input stream after every @n@
829+-- > S.toList $ S.intersperseBySpan 2 (return ',') $ S.fromList "hello"
2360−{-# INLINE wordsOn #-}
833+-- @since 0.7.0
2361−wordsOn
834+{-# INLINE intersperseBySpan #-}
2362− :: (IsStream t, Monad m, Storable a, Eq a)
835+intersperseBySpan :: IsStream t => Int -> m a -> t m a -> t m a
2363− => Array a -> Fold m a b -> t m a -> t m b
836+intersperseBySpan _n _f _xs = undefined
2364−wordsOn subseq f m = undefined -- D.fromStreamD $ D.wordsOn f subseq (D.toStreamD m)
2367−-- | Like 'splitOnSuffixSeq' but keeps the suffix intact in the splits.
839+-- | Insert an effect and its output after consuming an element of a stream.
2369−-- > splitWithSuffixSeq'_ pat xs = S.toList $ S.splitWithSuffixSeq (A.fromList pat) (FL.toList) (S.fromList xs)
2371−-- >>> splitWithSuffixSeq' "." ""
2374−-- >>> splitWithSuffixSeq' "." "."
2377−-- >>> splitWithSuffixSeq' "." "a"
841+-- @
2378−-- ["a"]
842+-- >>> S.toList $ S.trace putChar $ S.intersperseSuffix (putChar '.' >> return ',') $ S.fromList "hello"
843+-- > h.,e.,l.,l.,o.,"h,e,l,l,o,"
2380−-- >>> splitWithSuffixSeq' "." ".a"
846+-- /Internal/
2381−-- > [".","a"]
847+{-# INLINE intersperseSuffix #-}
848+intersperseSuffix :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
849+intersperseSuffix m = fromStreamD . D.intersperseSuffix m . toStreamD
851+-- | Insert a side effect after consuming an element of a stream.
2383−-- >>> splitWithSuffixSeq' "." "a."
853+-- @
2384−-- > ["a."]
854+-- > S.mapM_ putChar $ S.intersperseSuffix_ (threadDelay 1000000) $ S.fromList "hello"
2386−-- >>> splitWithSuffixSeq' "." "a.b"
857+-- /Internal/
2389−-- >>> splitWithSuffixSeq' "." "a.b."
859+{-# INLINE intersperseSuffix_ #-}
2390−-- > ["a.","b."]
860+intersperseSuffix_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
861+intersperseSuffix_ m = fromStreamD . D.intersperseSuffix_ m . toStreamD
863+-- | Like 'intersperseSuffix' but intersperses an effectful action into the
864+-- input stream after every @n@ elements and after the last element.
2392−-- >>> splitWithSuffixSeq' "." "a..b.."
866+-- @
2393−-- > ["a.",".","b.","."]
867+-- > S.toList $ S.intersperseSuffixBySpan 2 (return ',') $ S.fromList "hello"
2395-- /Internal/
871-- /Internal/
2396−{-# INLINE splitWithSuffixSeq #-}
872+--
2397−splitWithSuffixSeq
873+{-# INLINE intersperseSuffixBySpan #-}
2398− :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
874+intersperseSuffixBySpan :: (IsStream t, MonadAsync m)
2399− => Array a -> Fold m a b -> t m a -> t m b
2400−splitWithSuffixSeq patt f m =
2401− D.fromStreamD $ D.splitOnSuffixSeq True patt f (D.toStreamD m)
2404−-- This can be implemented easily using Rabin Karp
2405−-- | Split post any one of the given patterns.
2406−{-# INLINE splitOnSuffixSeqAny #-}
2408− :: (IsStream t, Monad m, Storable a, Integral a)
2409− => [Array a] -> Fold m a b -> t m a -> t m b
875+ => Int -> m a -> t m a -> t m a
2410−splitOnSuffixSeqAny subseq f m = undefined
876+intersperseSuffixBySpan n eff =
2411− -- D.fromStreamD $ D.splitPostAny f subseq (D.toStreamD m)
877+ fromStreamD . D.intersperseSuffixBySpan n eff . toStreamD
2414−------------------------------------------------------------------------------
2416−------------------------------------------------------------------------------
2418−-- | @splitInnerBy splitter joiner stream@ splits the inner containers @f a@ of
879+-- | Insert a side effect before consuming an element of a stream.
2419−-- an input stream @t m (f a)@ using the @splitter@ function. Container
880+--
2420−-- elements @f a@ are collected until a split occurs, then all the elements
881+-- @
2421−-- before the split are joined using the @joiner@ function.
882+-- >>> S.toList $ S.trace putChar $ S.interspersePrefix_ (putChar '.' >> return ',') $ S.fromList "hello"
883+-- > .h.e.l.l.o"hello"
2423−-- For example, if we have a stream of @Array Word8@, we may want to split the
886+-- See also: 'trace_'
2424−-- stream into arrays representing lines separated by '\n' byte such that the
2425−-- resulting stream after a split would be one array for each line.
2427−-- CAUTION! This is not a true streaming function as the container size after
888+-- /Concurrent/
2428−-- the split and merge may not be bounded.
2430-- /Internal/
890-- /Internal/
2431−{-# INLINE splitInnerBy #-}
2433− :: (IsStream t, Monad m)
2434− => (f a -> m (f a, Maybe (f a))) -- splitter
2435− -> (f a -> f a -> m (f a)) -- joiner
2438−splitInnerBy splitter joiner xs =
2439− D.fromStreamD $ D.splitInnerBy splitter joiner $ D.toStreamD xs
2441−-- | Like 'splitInnerBy' but splits assuming the separator joins the segment in
2444−-- /Internal/
892+{-# INLINE interspersePrefix_ #-}
2445−{-# INLINE splitInnerBySuffix #-}
893+interspersePrefix_ :: (IsStream t, MonadAsync m) => m b -> t m a -> t m a
2446−splitInnerBySuffix
894+interspersePrefix_ m = mapM (\x -> void m >> return x)
2447− :: (IsStream t, Monad m, Eq (f a), Monoid (f a))
2448− => (f a -> m (f a, Maybe (f a))) -- splitter
2449− -> (f a -> f a -> m (f a)) -- joiner
2452−splitInnerBySuffix splitter joiner xs =
2453− D.fromStreamD $ D.splitInnerBySuffix splitter joiner $ D.toStreamD xs
2455−------------------------------------------------------------------------------
2456−-- Reorder in sequence
2457−------------------------------------------------------------------------------
2460−-- Buffer until the next element in sequence arrives. The function argument
2461−-- determines the difference in sequence numbers. This could be useful in
2462−-- implementing sequenced streams, for example, TCP reassembly.
2463−{-# INLINE reassembleBy #-}
2465− :: (IsStream t, Monad m)
2470−reassembleBy = undefined
2473------------------------------------------------------------------------------
896------------------------------------------------------------------------------
2474−-- Distributing
897+-- Inserting Time
2475------------------------------------------------------------------------------
898------------------------------------------------------------------------------
2477−-- | Tap the data flowing through a stream into a 'Fold'. For example, you may
900+-- Note: delay must be serial.
2478−-- add a tap to log the contents flowing through the stream. The fold is used
2479−-- only for effects, its result is discarded.
2484−-- -----stream m a ---------------stream m a-----
2486−-- @
902+-- | Introduce a delay of specified seconds before consuming an element of the
903+-- stream except the first one.
2489−-- > S.drain $ S.tap (FL.drainBy print) (S.enumerateFromTo 1 2)
906+-- >>> S.mapM_ print $ S.timestamped $ S.delay 1 $ S.enumerateFromTo 1 3
2490−-- 1
907+-- > (AbsTime (TimeSpec {sec = 2502706, nsec = 751137000}),1)
2491−-- 2
908+-- > (AbsTime (TimeSpec {sec = 2502707, nsec = 743535000}),2)
909+-- > (AbsTime (TimeSpec {sec = 2502708, nsec = 749758000}),3)
2494−-- Compare with 'trace'.
912+-- /Internal/
2496−-- @since 0.7.0
914+{-# INLINE delay #-}
2497−{-# INLINE tap #-}
915+delay :: (IsStream t, MonadIO m) => Double -> t m a -> t m a
2498−tap :: (IsStream t, Monad m) => FL.Fold m a b -> t m a -> t m a
916+delay n = intersperseM_ $ liftIO $ threadDelay $ round $ n * 1000000
2499−tap f xs = D.fromStreamD $ D.tap f (D.toStreamD xs)
2501−-- | @tapOffsetEvery offset n@ taps every @n@th element in the stream
918+-- Note: delay must be serial.
2502−-- starting at @offset@. @offset@ can be between @0@ and @n - 1@. Offset 0
919+--
2503−-- means start at the first element in the stream. If the offset is outside
920+-- | Introduce a delay of specified seconds after consuming an element of a
2504−-- this range then @offset `mod` n@ is used as offset.
921+-- stream.
2507−-- >>> S.drain $ S.tapOffsetEvery 0 2 (FL.mapM print FL.toList) $ S.enumerateFromTo 0 10
924+-- >>> S.mapM_ print $ S.timestamped $ S.delayPost 1 $ S.enumerateFromTo 1 3
2508−-- > [0,2,4,6,8,10]
925+-- > (AbsTime (TimeSpec {sec = 2502826, nsec = 119030000}),1)
926+-- > (AbsTime (TimeSpec {sec = 2502827, nsec = 111393000}),2)
927+-- > (AbsTime (TimeSpec {sec = 2502828, nsec = 112221000}),3)
2511-- /Internal/
930-- /Internal/
2513−{-# INLINE tapOffsetEvery #-}
932+{-# INLINE delayPost #-}
2514−tapOffsetEvery :: (IsStream t, Monad m)
933+delayPost :: (IsStream t, MonadIO m) => Double -> t m a -> t m a
2515− => Int -> Int -> FL.Fold m a b -> t m a -> t m a
934+delayPost n = intersperseSuffix_ $ liftIO $ threadDelay $ round $ n * 1000000
2516−tapOffsetEvery offset n f xs =
2517− D.fromStreamD $ D.tapOffsetEvery offset n f (D.toStreamD xs)
2519−-- | Redirect a copy of the stream to a supplied fold and run it concurrently
936+-- Note: delay must be serial, that's why 'trace_' is used.
2520−-- in an independent thread. The fold may buffer some elements. The buffer size
2521−-- is determined by the prevailing 'maxBuffer' setting.
2524−-- Stream m a -> m b
2526−-- -----stream m a ---------------stream m a-----
2528−-- @
938+-- | Introduce a delay of specified seconds before consuming an element of a
2531−-- > S.drain $ S.tapAsync (S.mapM_ print) (S.enumerateFromTo 1 2)
942+-- >>> S.mapM_ print $ S.timestamped $ S.delayPre 1 $ S.enumerateFromTo 1 3
2532−-- 1
943+-- > (AbsTime (TimeSpec {sec = 2502207, nsec = 533177000}),1)
2533−-- 2
944+-- > (AbsTime (TimeSpec {sec = 2502208, nsec = 530859000}),2)
945+-- > (AbsTime (TimeSpec {sec = 2502209, nsec = 531619000}),3)
2536−-- Exceptions from the concurrently running fold are propagated to the current
948+-- /Internal/
2537−-- computation. Note that, because of buffering in the fold, exceptions may be
2538−-- delayed and may not correspond to the current element being processed in the
2539−-- parent stream, but we guarantee that before the parent stream stops the tap
2540−-- finishes and all exceptions from it are drained.
950+{-# INLINE delayPre #-}
951+delayPre :: (IsStream t, MonadIO m) => Double -> t m a -> t m a
952+delayPre n = trace_ $ liftIO $ threadDelay $ round $ n * 1000000
954+------------------------------------------------------------------------------
955+-- Reorder in sequence
956+------------------------------------------------------------------------------
958+-- | Buffer until the next element in sequence arrives. The function argument
959+-- determines the difference in sequence numbers. This could be useful in
960+-- implementing sequenced streams, for example, TCP reassembly.
2543−-- Compare with 'tap'.
962+-- /Unimplemented/
2545−-- /Internal/
964+{-# INLINE reassembleBy #-}
2546−{-# INLINE tapAsync #-}
965+reassembleBy
2547−tapAsync :: (IsStream t, MonadAsync m) => FL.Fold m a b -> t m a -> t m a
966+ :: -- (IsStream t, Monad m) =>
2548−tapAsync f xs = D.fromStreamD $ D.tapAsync f (D.toStreamD xs)
967+ Fold m a b
2549−
968+ -> (a -> a -> Int)
2550−-- | @pollCounts predicate transform fold stream@ counts those elements in the
969+ -> t m a
2551−-- stream that pass the @predicate@. The resulting count stream is sent to
970+ -> t m b
2552−-- another thread which transforms it using @transform@ and then folds it using
971+reassembleBy = undefined
2553−-- @fold@. The thread is automatically cleaned up if the stream stops or
972+
2554−-- aborts due to exception.
973+------------------------------------------------------------------------------
975+------------------------------------------------------------------------------
978+-- > indexed = S.postscanl' (\(i, _) x -> (i + 1, x)) (-1,undefined)
979+-- > indexed = S.zipWith (,) (S.enumerateFrom 0)
2556−-- For example, to print the count of elements processed every second:
981+-- Pair each element in a stream with its index, starting from index 0.
2559−-- > S.drain $ S.pollCounts (const True) (S.rollingMap (-) . S.delayPost 1) (FL.drainBy print)
984+-- > S.toList $ S.indexed $ S.fromList "hello"
2560−-- $ S.enumerateFrom 0
985+-- [(0,'h'),(1,'e'),(2,'l'),(3,'l'),(4,'o')]
2563−-- Note: This may not work correctly on 32-bit machines.
988+-- @since 0.6.0
989+{-# INLINE indexed #-}
990+indexed :: (IsStream t, Monad m) => t m a -> t m (Int, a)
991+indexed = fromStreamD . D.indexed . toStreamD
994+-- > indexedR n = S.postscanl' (\(i, _) x -> (i - 1, x)) (n + 1,undefined)
995+-- > indexedR n = S.zipWith (,) (S.enumerateFromThen n (n - 1))
2565−-- /Internal/
997+-- Pair each element in a stream with its index, starting from the
998+-- given index @n@ and counting down.
2567−{-# INLINE pollCounts #-}
1000+-- @
2568−pollCounts ::
1001+-- > S.toList $ S.indexedR 10 $ S.fromList "hello"
2569− (IsStream t, MonadAsync m)
1002+-- [(10,'h'),(9,'e'),(8,'l'),(7,'l'),(6,'o')]
2570− => (a -> Bool)
1003+-- @
2571− -> (t m Int -> t m Int)
1004+--
2572− -> Fold m Int b
1005+-- @since 0.6.0
2573− -> t m a
1006+{-# INLINE indexedR #-}
2574− -> t m a
1007+indexedR :: (IsStream t, Monad m) => Int -> t m a -> t m (Int, a)
2575−pollCounts predicate transf f xs =
1008+indexedR n = fromStreamD . D.indexedR n . toStreamD
2577− $ D.pollCounts predicate (D.toStreamD . transf . D.fromStreamD) f
2580−-- | Calls the supplied function with the number of elements consumed
1010+-------------------------------------------------------------------------------
2581−-- every @n@ seconds. The given function is run in a separate thread
1011+-- Time Indexing
2582−-- until the end of the stream. In case there is an exception in the
1012+-------------------------------------------------------------------------------
2583−-- stream the thread is killed during the next major GC.
1013+
1014+-- Note: The timestamp stream must be the second stream in the zip so that the
1015+-- timestamp is generated after generating the stream element and not before.
1016+-- If we do not do that then the following example will generate the same
1017+-- timestamp for first two elements:
2585−-- Note: The action is not guaranteed to run if the main thread exits.
1019+-- S.mapM_ print $ S.timestamped $ S.delay $ S.enumerateFromTo 1 3
1021+-- | Pair each element in a stream with an absolute timestamp, using a clock of
1022+-- specified granularity. The timestamp is generated just before the element
2588−-- > delay n = threadDelay (round $ n * 1000000) >> return n
1026+-- >>> S.mapM_ print $ S.timestampWith 0.01 $ S.delay 1 $ S.enumerateFromTo 1 3
2589−-- > S.drain $ S.tapRate 2 (\\n -> print $ show n ++ " elements processed") (delay 1 S.|: delay 0.5 S.|: delay 0.5 S.|: S.nil)
2590−-- 2 elements processed
2591−-- 1 elements processed
2594−-- Note: This may not work correctly on 32-bit machines.
1029+-- /Internal/
1031+{-# INLINE timestampWith #-}
1032+timestampWith :: (IsStream t, MonadAsync m, Functor (t m))
1033+ => Double -> t m a -> t m (AbsTime, a)
1034+timestampWith g stream = Z.zipWith (flip (,)) stream (absTimesWith g)
1036+-- TBD: check performance vs a custom implementation without using zipWith.
2596-- /Internal/
1038-- /Internal/
2597−{-# INLINE tapRate #-}
1039+--
2598−tapRate ::
1040+{-# INLINE timestamped #-}
2599− (IsStream t, MonadAsync m, MonadCatch m)
1041+timestamped :: (IsStream t, MonadAsync m, Functor (t m))
2600− => Double
1042+ => t m a -> t m (AbsTime, a)
2601− -> (Int -> m b)
1043+timestamped = timestampWith 0.01
2604−tapRate n f xs = D.fromStreamD $ D.tapRate n f $ D.toStreamD xs
2606−-- | Apply a monadic function to each element flowing through the stream and
1045+-- | Pair each element in a stream with relative times starting from 0, using a
2607−-- discard the results.
1046+-- clock with the specified granularity. The time is measured just before the
1047+-- element is consumed.
2610−-- > S.drain $ S.trace print (S.enumerateFromTo 1 2)
1050+-- >>> S.mapM_ print $ S.timeIndexWith 0.01 $ S.delay 1 $ S.enumerateFromTo 1 3
2615−-- Compare with 'tap'.
1053+-- /Internal/
2617−-- @since 0.7.0
1055+{-# INLINE timeIndexWith #-}
2618−{-# INLINE trace #-}
1056+timeIndexWith :: (IsStream t, MonadAsync m, Functor (t m))
2619−trace :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m a
1057+ => Double -> t m a -> t m (RelTime64, a)
2620−trace f = mapM (\x -> void (f x) >> return x)
1058+timeIndexWith g stream = Z.zipWith (flip (,)) stream (relTimesWith g)
2622−-- | Perform a side effect before yielding each element of the stream and
1060+-- | Pair each element in a stream with relative times starting from 0, using a
2623−-- discard the results.
1061+-- 10 ms granularity clock. The time is measured just before the element is
2626−-- > S.drain $ S.trace_ (print "got here") (S.enumerateFromTo 1 2)
1065+-- >>> S.mapM_ print $ S.timeIndexed $ S.delay 1 $ S.enumerateFromTo 1 3
2627−-- "got here"
1066+-- (RelTime64 (NanoSecond64 0),1)
2628−-- "got here"
1067+-- (RelTime64 (NanoSecond64 996239000),2)
1068+-- (RelTime64 (NanoSecond64 1996712000),3)
2631−-- See also: 'trace', 'interspersePrefix_'
2633-- /Internal/
1071-- /Internal/
2634−{-# INLINE trace_ #-}
1072+--
2635−trace_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
1073+{-# INLINE timeIndexed #-}
2636−trace_ eff = Serial.mapM (\x -> eff >> return x)
1074+timeIndexed :: (IsStream t, MonadAsync m, Functor (t m))
1075+ => t m a -> t m (RelTime64, a)
1076+timeIndexed = timeIndexWith 0.01
2638------------------------------------------------------------------------------
1078------------------------------------------------------------------------------
2639−-- Folding transformations
1079+-- Searching
2640------------------------------------------------------------------------------
1080------------------------------------------------------------------------------
2642−-- | Right fold to a streaming monad.
1082+-- | Find all the indices where the value of the element in the stream is equal
2643−--
1083+-- to the given value.
2644−-- > foldrS S.cons S.nil === id
2646−-- 'foldrS' can be used to perform stateless stream to stream transformations
2647−-- like map and filter in general. It can be coupled with a scan to perform
2648−-- stateful transformations. However, note that the custom map and filter
2649−-- routines can be much more efficient than this due to better stream fusion.
2651−-- >>> S.toList $ S.foldrS S.cons S.nil $ S.fromList [1..5]
2654−-- Find if any element in the stream is 'True':
2656−-- >>> 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)
2659−-- Map (+2) on odd elements and filter out the even elements:
1085+-- > elemIndices a = findIndices (== a)
2661−-- >>> 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)
1087+-- @since 0.5.0
2662−-- > [3,5,7]
1088+{-# INLINE elemIndices #-}
1089+elemIndices :: (IsStream t, Eq a, Monad m) => a -> t m a -> t m Int
1090+elemIndices a = findIndices (== a)
1092+------------------------------------------------------------------------------
1094+------------------------------------------------------------------------------
1096+-- XXX this is not a one-to-one map so calling it map may not be right.
1097+-- We can perhaps call it zipWithTail or rollWith.
2664−-- 'foldrM' can also be represented in terms of 'foldrS', however, the former
1099+-- | Apply a function on every two successive elements of a stream. If the
2665−-- is much more efficient:
1100+-- stream consists of a single element the output is an empty stream.
2667−-- > foldrM f z s = runIdentityT $ foldrS (\x xs -> lift $ f x (runIdentityT xs)) (lift z) s
1102+-- This is the stream equivalent of the list idiom @zipWith f xs (tail xs)@.
2669-- /Internal/
1104-- /Internal/
2670−{-# INLINE foldrS #-}
2671−foldrS :: IsStream t => (a -> t m b -> t m b) -> t m b -> t m a -> t m b
2674−-- | Right fold to a transformer monad. This is the most general right fold
2675−-- function. 'foldrS' is a special case of 'foldrT', however 'foldrS'
2676−-- implementation can be more efficient:
2678−-- > foldrS = foldrT
2679−-- > foldrM f z s = runIdentityT $ foldrT (\x xs -> lift $ f x (runIdentityT xs)) (lift z) s
2681−-- 'foldrT' can be used to translate streamly streams to other transformer
1106+{-# INLINE rollingMap #-}
2682−-- monads e.g. to a different streaming type.
1107+rollingMap :: (IsStream t, Monad m) => (a -> a -> b) -> t m a -> t m b
1108+rollingMap f m = fromStreamD $ D.rollingMap f $ toStreamD m
1110+-- | Like 'rollingMap' but with an effectful map function.
2684-- /Internal/
1112-- /Internal/
2685−{-# INLINE foldrT #-}
1113+--
2686−foldrT :: (IsStream t, Monad m, Monad (s m), MonadTrans s)
1114+{-# INLINE rollingMapM #-}
2687− => (a -> s m b -> s m b) -> s m b -> t m a -> s m b
1115+rollingMapM :: (IsStream t, Monad m) => (a -> a -> m b) -> t m a -> t m b
2688−foldrT f z s = S.foldrT f z (toStreamS s)
1116+rollingMapM f m = fromStreamD $ D.rollingMapM f $ toStreamD m
2690------------------------------------------------------------------------------
1118------------------------------------------------------------------------------
2691−-- Generalize the underlying monad
1119+-- Maybe Streams
2692------------------------------------------------------------------------------
1120------------------------------------------------------------------------------
2694−-- | Transform the inner monad of a stream using a natural transformation.
1122+-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'
1123+-- elements, and return a stream of values extracted from 'Just'.
2696−-- / Internal/
1127+-- @
1128+-- mapMaybe f = S.map 'fromJust' . S.filter 'isJust' . S.map f
2698−{-# INLINE hoist #-}
1131+-- @since 0.3.0
2699−hoist :: (Monad m, Monad n)
1132+{-# INLINE mapMaybe #-}
2700− => (forall x. m x -> n x) -> SerialT m a -> SerialT n a
1133+mapMaybe :: (IsStream t, Monad m) => (a -> Maybe b) -> t m a -> t m b
2701−hoist f xs = fromStreamS $ S.hoist f (toStreamS xs)
1134+mapMaybe f m = fromStreamS $ S.mapMaybe f $ toStreamS m
2703−-- | Generalize the inner monad of the stream from 'Identity' to any monad.
1136+-- | Like 'mapMaybe' but maps a monadic function.
1141+-- mapMaybeM f = S.map 'fromJust' . S.filter 'isJust' . S.mapM f
2705−-- / Internal/
1144+-- /Concurrent (do not use with 'parallely' on infinite streams)/
2707−{-# INLINE generally #-}
1146+-- @since 0.3.0
2708−generally :: (IsStream t, Monad m) => t Identity a -> t m a
1147+{-# INLINE_EARLY mapMaybeM #-}
2709−generally xs = fromStreamS $ S.hoist (return . runIdentity) (toStreamS xs)
1148+mapMaybeM :: (IsStream t, MonadAsync m, Functor (t m))
1149+ => (a -> m (Maybe b)) -> t m a -> t m b
1150+mapMaybeM f = fmap fromJust . filter isJust . K.mapM f
2711−------------------------------------------------------------------------------
1152+{-# RULES "mapMaybeM serial" mapMaybeM = mapMaybeMSerial #-}
2712−-- Add and remove a monad transformer
1153+{-# INLINE mapMaybeMSerial #-}
2713−------------------------------------------------------------------------------
1154+mapMaybeMSerial :: Monad m => (a -> m (Maybe b)) -> SerialT m a -> SerialT m b
1155+mapMaybeMSerial f m = fromStreamD $ D.mapMaybeM f $ toStreamD m
2715−-- | Lift the inner monad @m@ of a stream @t m a@ to @tr m@ using the monad
1157+-- | In a stream of 'Maybe's, discard 'Nothing's and unwrap 'Just's.
2716−-- transformer @tr@.
2718−-- / Internal/
1159+-- /Internal/
2720−{-# INLINE liftInner #-}
1161+catMaybes :: (IsStream t, Monad m, Functor (t m)) => t m (Maybe a) -> t m a
2721−liftInner :: (Monad m, IsStream t, MonadTrans tr, Monad (tr m))
1162+catMaybes = fmap fromJust . filter isJust
2722− => t m a -> t (tr m) a
2723−liftInner xs = fromStreamD $ D.liftInner (toStreamD xs)
2725------------------------------------------------------------------------------
1164------------------------------------------------------------------------------
2726−-- Sharing read only state in a stream
1165+-- Either streams
2727------------------------------------------------------------------------------
1166------------------------------------------------------------------------------
2729−-- | Evaluate the inner monad of a stream as 'ReaderT'.
1168+-- | Discard 'Right's and unwrap 'Left's in an 'Either' stream.
2731−-- / Internal/
1170+-- /Internal/
2733−{-# INLINE runReaderT #-}
1172+lefts :: (IsStream t, Monad m, Functor (t m)) => t m (Either a b) -> t m a
2734−runReaderT :: (IsStream t, Monad m) => m s -> t (ReaderT s m) a -> t m a
1173+lefts = fmap (fromLeft undefined) . filter isLeft
2735−runReaderT s xs = fromStreamD $ D.runReaderT s (toStreamD xs)
2737−-- | Run a stream transformation using a given environment.
1175+-- | Discard 'Left's and unwrap 'Right's in an 'Either' stream.
2739−-- See also: 'Serial.map'
2741−-- / Internal/
1177+-- /Internal/
2743−{-# INLINE usingReaderT #-}
1179+rights :: (IsStream t, Monad m, Functor (t m)) => t m (Either a b) -> t m b
2744−usingReaderT
1180+rights = fmap (fromRight undefined) . filter isRight
2745− :: (Monad m, IsStream t)
2747− -> (t (ReaderT r m) a -> t (ReaderT r m) a)
2750−usingReaderT r f xs = runReaderT r $ f $ liftInner xs
2752------------------------------------------------------------------------------
1182------------------------------------------------------------------------------
2753−-- Sharing read write state in a stream
1183+-- Concurrent Application
2754------------------------------------------------------------------------------
1184------------------------------------------------------------------------------
2756−-- | Evaluate the inner monad of a stream as 'StateT'.
1186+-- | Parallel transform application operator; applies a stream transformation
1187+-- function @t m a -> t m b@ to a stream @t m a@ concurrently; the input stream
1188+-- is evaluated asynchronously in an independent thread yielding elements to a
1189+-- buffer and the transformation function runs in another thread consuming the
1190+-- input from the buffer. '|$' is just like regular function application
1191+-- operator '$' except that it is concurrent.
2758−-- This is supported only for 'SerialT' as concurrent state updation may not be
1193+-- If you read the signature as @(t m a -> t m b) -> (t m a -> t m b)@ you can
2759−-- safe.
1194+-- look at it as a transformation that converts a transform function to a
1195+-- buffered concurrent transform function.
2761−-- / Internal/
1197+-- The following code prints a value every second even though each stage adds a
2763−{-# INLINE evalStateT #-}
2764−evalStateT :: Monad m => m s -> SerialT (StateT s m) a -> SerialT m a
2765−evalStateT s xs = fromStreamD $ D.evalStateT s (toStreamD xs)
2767−-- | Run a stateful (StateT) stream transformation using a given state.
2769−-- This is supported only for 'SerialT' as concurrent state updation may not be
1201+-- @
2770−-- safe.
1202+-- drain $
1203+-- S.mapM (\\x -> threadDelay 1000000 >> print x)
1204+-- |$ S.repeatM (threadDelay 1000000 >> return 1)
1209+-- /Since: 0.3.0 ("Streamly")/
2772−-- See also: 'scanl''
1211+-- @since 0.8.0
1213+(|$) :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> (t m a -> t m b)
1214+-- (|$) f = f . Async.mkAsync
1215+(|$) f = f . D.mkParallel
2774−-- / Internal/
1221+-- /Internal/
2776−{-# INLINE usingStateT #-}
1223+{-# INLINE applyAsync #-}
2777−usingStateT
1224+applyAsync :: (IsStream t, MonadAsync m)
2778− :: Monad m
1225+ => (t m a -> t m b) -> (t m a -> t m b)
2779− => m s
1226+applyAsync = (|$)
2780− -> (SerialT (StateT s m) a -> SerialT (StateT s m) a)
2783−usingStateT s f xs = evalStateT s $ f $ liftInner xs
2785−-- | Evaluate the inner monad of a stream as 'StateT' and emit the resulting
1228+-- | Parallel reverse function application operator for streams; just like the
2786−-- state and value pair after each step.
1229+-- regular reverse function application operator '&' except that it is
1234+-- S.repeatM (threadDelay 1000000 >> return 1)
1235+-- |& S.mapM (\\x -> threadDelay 1000000 >> print x)
2788−-- This is supported only for 'SerialT' as concurrent state updation may not be
1238+-- /Concurrent/
2791−-- / Internal/
1240+-- /Since: 0.3.0 ("Streamly")/
2793−{-# INLINE runStateT #-}
1242+-- @since 0.8.0
2794−runStateT :: Monad m => m s -> SerialT (StateT s m) a -> SerialT m (s, a)
1243+{-# INLINE (|&) #-}
2795−runStateT s xs = fromStreamD $ D.runStateT s (toStreamD xs)
1244+(|&) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> t m b) -> t m b