on 810ba5b help sign in
1589 lines 50 kB history raw
1-- |
2-- Module : Streamly.Internal.Data.Stream.IsStream.Transform
3-- Copyright : (c) 2017 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
7-- Portability : GHC
9module Streamly.Internal.Data.Stream.IsStream.Transform
10 (
11 -- * Piping
12 -- | Pass through a 'Pipe'.
13 transform
15 -- * Folding
16 , foldrS
17 , foldrT
19 -- * Mapping
20 -- | Stateless one-to-one maps.
21 , Serial.map
22 , sequence
23 , mapM
24 , smapM
26 -- * Mapping Side Effects (Observation)
27 -- | See also the intersperse*_ combinators.
28 , trace
29 , trace_
30 , tap
31 , tapOffsetEvery
32 , tapAsync
33 , tapRate
34 , pollCounts
36 -- * Scanning By 'Fold'
37 , scan
38 , postscan
40 -- * Scanning
41 -- | Left scans. Stateful, mostly one-to-one maps.
42 , scanl'
43 , scanlM'
44 , scanlMAfter'
45 , postscanl'
46 , postscanlM'
47 , prescanl'
48 , prescanlM'
49 , scanl1'
50 , scanl1M'
52 -- XXX Once we have pipes the contravariant transformations can be
53 -- represented by attaching pipes before a transformation.
54 --
55 -- , lscanl'
56 -- , lscanlM'
57 -- , lscanl1'
58 -- , lscanl1M'
59 --
60 -- , lpostscanl'
61 -- , lpostscanlM'
62 -- , lprescanl'
63 -- , lprescanlM'
65 -- * Filtering
66 -- | Produce a subset of the stream using criteria based on the values of
67 -- the elements. We can use a concatMap and scan for filtering but these
68 -- combinators are more efficient and convenient.
70 , with
71 , deleteBy
72 , filter
73 , filterM
74 , uniq
75 , uniqBy
76 , nubBy
77 , nubWindowBy
78 , prune
79 , repeated
81 -- * Trimming
82 -- | Produce a subset of the stream trimmed at ends.
84 , take
85 , takeInterval
86 , takeLast
87 , takeLastInterval
88 , takeWhile
89 , takeWhileM
90 , takeWhileLast
91 , takeWhileAround
92 , drop
93 , dropInterval
94 , dropLast
95 , dropLastInterval
96 , dropWhile
97 , dropWhileM
98 , dropWhileLast
99 , dropWhileAround
101 -- * Inserting Elements
102 -- | Produce a superset of the stream. This is the opposite of
103 -- filtering/sampling. We can always use concatMap and scan for inserting
104 -- but these combinators are more efficient and convenient.
106 -- Element agnostic (Opposite of sampling)
107 , intersperse
108 , intersperseM -- XXX naming
109 , intersperseBySpan
111 , intersperseSuffix
112 , intersperseSuffixBySpan
113 , interjectSuffix
115 -- , interspersePrefix
116 -- , interspersePrefixBySpan
118 -- * Inserting Side Effects/Time
119 , intersperseM_ -- XXX naming
120 , delay
121 , intersperseSuffix_
122 , delayPost
123 , interspersePrefix_
124 , delayPre
126 -- * Element Aware Insertion
127 -- | Opposite of filtering
128 , insertBy
129 -- , intersperseByBefore
130 -- , intersperseByAfter
132 -- * Reordering
133 , reverse
134 , reverse'
135 , reassembleBy
137 -- * Position Indexing
138 , indexed
139 , indexedR
141 -- * Time Indexing
142 , timestamped
143 , timestampWith
144 -- , timestampedR -- timer
145 , timeIndexed
146 , timeIndexWith
148 -- * Searching
149 , findIndices -- XXX indicesBy
150 , elemIndices -- XXX indicesOf
152 -- * Rolling map
153 -- | Map using the previous element.
154 , rollingMapM
155 , rollingMap
157 -- * Maybe Streams
158 -- Move these to Streamly.Data.Maybe.Stream?
159 , catMaybes -- XXX justs (like lefts/rights)
160 , mapMaybe
161 , mapMaybeM
163 -- * Either Streams
164 -- Move these to Streamly.Data.Either.Stream?
165 , lefts
166 , rights
167 , both
169 -- * Concurrent Evaluation
170 -- ** Concurrent Pipelines
171 -- | Run streaming stages concurrently.
173 , Par.mkParallel
174 , applyAsync
175 , (|$)
176 , (|&)
178 -- ** Concurrency Control
179 , maxThreads
181 -- ** Buffering and Sampling
182 -- | Evaluate strictly using a buffer of results. When the buffer becomes
183 -- full we can block, drop the new elements, drop the oldest element and
184 -- insert the new at the end or keep dropping elements uniformly to match
185 -- the rate of the consumer.
186 , maxBuffer
187 , sampleOld
188 , sampleNew
189 , sampleRate
191 -- ** Rate Limiting
192 -- | Evaluate the stream at uniform intervals to maintain a specified
193 -- evaluation rate.
194 , Rate (..)
195 , rate
196 , avgRate
197 , minRate
198 , maxRate
199 , constRate
201 -- * Diagnostics
202 , inspectMode
204 -- * Deprecated
205 , scanx
206 )
207where
209#include "inline.hs"
211import Control.Concurrent (threadDelay)
212import Control.Monad (void)
213import Control.Monad.Catch (MonadCatch)
214import Control.Monad.IO.Class (MonadIO(..))
215import Control.Monad.Trans.Class (MonadTrans(..))
216import Data.Either (isLeft, isRight)
217import Data.Kind (Type)
218import Data.Maybe (isJust, fromJust)
219import Streamly.Internal.BaseCompat (fromLeft, fromRight)
220import Streamly.Internal.Data.Fold.Type (Fold (..))
221import Streamly.Internal.Data.Pipe.Type (Pipe (..))
222import Streamly.Internal.Data.Stream.IsStream.Combinators
223 ( inspectMode, maxBuffer, maxThreads, rate, avgRate, minRate
224 , maxRate, constRate)
225import Streamly.Internal.Data.Stream.IsStream.Common
226 ( absTimesWith
227 , drop
228 , findIndices
229 , postscanlM'
230 , relTimesWith
231 , reverse
232 , reverse'
233 , scanlMAfter'
234 , smapM
235 , take
236 , takeWhile
237 , interjectSuffix
238 , intersperseM
239 )
240import Streamly.Internal.Control.Concurrent (MonadAsync)
241import Streamly.Internal.Data.Stream.Prelude (fromStreamS, toStreamS)
242import Streamly.Internal.Data.Stream.Serial (SerialT)
243import Streamly.Internal.Data.Stream.StreamD (fromStreamD, toStreamD)
244import Streamly.Internal.Data.Stream.StreamK (IsStream)
245import Streamly.Internal.Data.SVar (Rate(..))
246import Streamly.Internal.Data.Time.Units (TimeUnit64, AbsTime, RelTime64)
248import qualified Streamly.Internal.Data.Fold as FL
249import qualified Streamly.Internal.Data.Stream.Parallel as Par
250import qualified Streamly.Internal.Data.Stream.Prelude as P
251import qualified Streamly.Internal.Data.Stream.Serial as Serial
252import qualified Streamly.Internal.Data.Stream.StreamD as D
253import qualified Streamly.Internal.Data.Stream.StreamK as K
254import qualified Streamly.Internal.Data.Stream.Zip as Z
255#ifdef USE_STREAMK_ONLY
256import qualified Streamly.Internal.Data.Stream.StreamK as S
257#else
258import qualified Streamly.Internal.Data.Stream.StreamD as S
259#endif
261import Prelude hiding
262 ( filter, drop, dropWhile, take, takeWhile, foldr, map, mapM, sequence
263 , reverse, foldr1 , scanl, scanl1)
265--
266-- $setup
267-- >>> :m
268-- >>> import Control.Concurrent (threadDelay)
269-- >>> import Data.Function ((&))
270-- >>> import Streamly.Prelude ((|$))
271-- >>> import Prelude hiding ( filter, drop, dropWhile, take, takeWhile, foldr, map, mapM, sequence, reverse, foldr1 , scanl, scanl1)
272-- >>> import qualified Streamly.Prelude as Stream
273-- >>> import Streamly.Internal.Data.Stream.IsStream as Stream
274-- >>> import qualified Streamly.Data.Fold as Fold
275-- >>> import qualified Streamly.Internal.Data.Fold as Fold
277------------------------------------------------------------------------------
278-- Piping
279------------------------------------------------------------------------------
281-- | Use a 'Pipe' to transform a stream.
282--
283-- /Pre-release/
284--
285{-# INLINE transform #-}
286transform :: (IsStream t, Monad m) => Pipe m a b -> t m a -> t m b
287transform pipe xs = fromStreamD $ D.transform pipe (toStreamD xs)
289------------------------------------------------------------------------------
290-- Transformation Folds
291------------------------------------------------------------------------------
293-- | Right fold to a streaming monad.
294--
295-- > foldrS Stream.cons Stream.nil === id
296--
297-- 'foldrS' can be used to perform stateless stream to stream transformations
298-- like map and filter in general. It can be coupled with a scan to perform
299-- stateful transformations. However, note that the custom map and filter
300-- routines can be much more efficient than this due to better stream fusion.
301--
302-- >>> Stream.toList $ Stream.foldrS Stream.cons Stream.nil $ Stream.fromList [1..5]
303-- [1,2,3,4,5]
304--
305-- Find if any element in the stream is 'True':
306--
307-- >>> Stream.toList $ Stream.foldrS (\x xs -> if odd x then return True else xs) (return False) $ (Stream.fromList (2:4:5:undefined) :: Stream.SerialT IO Int)
308-- [True]
309--
310-- Map (+2) on odd elements and filter out the even elements:
311--
312-- >>> Stream.toList $ Stream.foldrS (\x xs -> if odd x then (x + 2) `Stream.cons` xs else xs) Stream.nil $ (Stream.fromList [1..5] :: Stream.SerialT IO Int)
313-- [3,5,7]
314--
315-- 'foldrM' can also be represented in terms of 'foldrS', however, the former
316-- is much more efficient:
317--
318-- > foldrM f z s = runIdentityT $ foldrS (\x xs -> lift $ f x (runIdentityT xs)) (lift z) s
319--
320-- /Pre-release/
321{-# INLINE foldrS #-}
322foldrS :: IsStream t => (a -> t m b -> t m b) -> t m b -> t m a -> t m b
323foldrS = K.foldrS
325-- | Right fold to a transformer monad. This is the most general right fold
326-- function. 'foldrS' is a special case of 'foldrT', however 'foldrS'
327-- implementation can be more efficient:
328--
329-- > foldrS = foldrT
330-- > foldrM f z s = runIdentityT $ foldrT (\x xs -> lift $ f x (runIdentityT xs)) (lift z) s
331--
332-- 'foldrT' can be used to translate streamly streams to other transformer
333-- monads e.g. to a different streaming type.
334--
335-- /Pre-release/
336{-# INLINE foldrT #-}
337foldrT :: (IsStream t, Monad m, Monad (s m), MonadTrans s)
338 => (a -> s m b -> s m b) -> s m b -> t m a -> s m b
339foldrT f z s = S.foldrT f z (toStreamS s)
341------------------------------------------------------------------------------
342-- Transformation by Mapping
343------------------------------------------------------------------------------
345-- |
346-- @
347-- mapM f = sequence . map f
348-- @
349--
350-- Apply a monadic function to each element of the stream and replace it with
351-- the output of the resulting action.
352--
353-- @
354-- >>> drain $ Stream.mapM putStr $ Stream.fromList ["a", "b", "c"]
355-- abc
356--
357-- >>> :{
358-- drain $ Stream.replicateM 10 (return 1)
359-- & (fromSerial . Stream.mapM (\x -> threadDelay 1000000 >> print x))
360-- :}
361-- 1
362-- ...
363-- 1
364--
365-- > drain $ Stream.replicateM 10 (return 1)
366-- & (fromAsync . Stream.mapM (\x -> threadDelay 1000000 >> print x))
367-- @
368--
369-- /Concurrent (do not use with 'fromParallel' on infinite streams)/
370--
371-- @since 0.1.0
372{-# INLINE_EARLY mapM #-}
373mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
374mapM = K.mapM
376{-# RULES "mapM serial" mapM = mapMSerial #-}
377{-# INLINE mapMSerial #-}
378mapMSerial :: Monad m => (a -> m b) -> SerialT m a -> SerialT m b
379mapMSerial = Serial.mapM
381-- |
382-- @
383-- sequence = mapM id
384-- @
385--
386-- Replace the elements of a stream of monadic actions with the outputs of
387-- those actions.
388--
389-- @
390-- >>> drain $ Stream.sequence $ Stream.fromList [putStr "a", putStr "b", putStrLn "c"]
391-- abc
392--
393-- >>> :{
394-- drain $ Stream.replicateM 10 (return $ threadDelay 1000000 >> print 1)
395-- & (fromSerial . Stream.sequence)
396-- :}
397-- 1
398-- ...
399-- 1
400--
401-- >>> :{
402-- drain $ Stream.replicateM 10 (return $ threadDelay 1000000 >> print 1)
403-- & (fromAsync . Stream.sequence)
404-- :}
405-- 1
406-- ...
407-- 1
408--
409-- @
410--
411-- /Concurrent (do not use with 'fromParallel' on infinite streams)/
412--
413-- @since 0.1.0
414{-# INLINE sequence #-}
415sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
416sequence m = fromStreamS $ S.sequence (toStreamS m)
418------------------------------------------------------------------------------
419-- Mapping side effects
420------------------------------------------------------------------------------
422-- | Tap the data flowing through a stream into a 'Fold'. For example, you may
423-- add a tap to log the contents flowing through the stream. The fold is used
424-- only for effects, its result is discarded.
425--
426-- @
427-- Fold m a b
428-- |
429-- -----stream m a ---------------stream m a-----
430--
431-- @
432--
433-- >>> Stream.drain $ Stream.tap (Fold.drainBy print) (Stream.enumerateFromTo 1 2)
434-- 1
435-- 2
436--
437-- Compare with 'trace'.
438--
439-- @since 0.7.0
440{-# INLINE tap #-}
441tap :: (IsStream t, Monad m) => FL.Fold m a b -> t m a -> t m a
442tap f xs = D.fromStreamD $ D.tap f (D.toStreamD xs)
444-- XXX Remove this. It can be expressed in terms of Fold.sampleFromThen.
445--
446-- | @tapOffsetEvery offset n@ taps every @n@th element in the stream
447-- starting at @offset@. @offset@ can be between @0@ and @n - 1@. Offset 0
448-- means start at the first element in the stream. If the offset is outside
449-- this range then @offset `mod` n@ is used as offset.
450--
451-- >>> Stream.drain $ Stream.tapOffsetEvery 0 2 (Fold.rmapM print Fold.toList) $ Stream.enumerateFromTo 0 10
452-- [0,2,4,6,8,10]
454--
455-- /Pre-release/
456--
457{-# INLINE tapOffsetEvery #-}
458tapOffsetEvery :: (IsStream t, Monad m)
459 => Int -> Int -> FL.Fold m a b -> t m a -> t m a
460tapOffsetEvery offset n f xs =
461 D.fromStreamD $ D.tapOffsetEvery offset n f (D.toStreamD xs)
463-- | Redirect a copy of the stream to a supplied fold and run it concurrently
464-- in an independent thread. The fold may buffer some elements. The buffer size
465-- is determined by the prevailing 'maxBuffer' setting.
466--
467-- @
468-- Stream m a -> m b
469-- |
470-- -----stream m a ---------------stream m a-----
471--
472-- @
473--
474-- @
475-- >>> Stream.drain $ Stream.tapAsync (Fold.drainBy print) (Stream.enumerateFromTo 1 2)
476-- 1
477-- 2
478--
479-- @
480--
481-- Exceptions from the concurrently running fold are propagated to the current
482-- computation. Note that, because of buffering in the fold, exceptions may be
483-- delayed and may not correspond to the current element being processed in the
484-- parent stream, but we guarantee that before the parent stream stops the tap
485-- finishes and all exceptions from it are drained.
486--
487--
488-- Compare with 'tap'.
489--
490-- /Pre-release/
491{-# INLINE tapAsync #-}
492tapAsync :: (IsStream t, MonadAsync m) => FL.Fold m a b -> t m a -> t m a
493tapAsync f xs = D.fromStreamD $ Par.tapAsyncF f (D.toStreamD xs)
495-- | @pollCounts predicate transform fold stream@ counts those elements in the
496-- stream that pass the @predicate@. The resulting count stream is sent to
497-- another thread which transforms it using @transform@ and then folds it using
498-- @fold@. The thread is automatically cleaned up if the stream stops or
499-- aborts due to exception.
500--
501-- For example, to print the count of elements processed every second:
502--
503-- @
504-- > Stream.drain $ Stream.pollCounts (const True) (Stream.rollingMap (-) . Stream.delayPost 1) (FLold.drainBy print)
505-- $ Stream.enumerateFrom 0
506-- @
507--
508-- Note: This may not work correctly on 32-bit machines.
509--
510-- /Pre-release/
511--
512{-# INLINE pollCounts #-}
513pollCounts ::
514 (IsStream t, MonadAsync m)
515 => (a -> Bool)
516 -> (t m Int -> t m Int)
517 -> Fold m Int b
518 -> t m a
519 -> t m a
520pollCounts predicate transf f xs =
521 D.fromStreamD
522 $ D.pollCounts predicate (D.toStreamD . transf . D.fromStreamD) f
523 $ D.toStreamD xs
525-- | Calls the supplied function with the number of elements consumed
526-- every @n@ seconds. The given function is run in a separate thread
527-- until the end of the stream. In case there is an exception in the
528-- stream the thread is killed during the next major GC.
529--
530-- Note: The action is not guaranteed to run if the main thread exits.
531--
532-- @
533-- > delay n = threadDelay (round $ n * 1000000) >> return n
534-- > Stream.toList $ Stream.tapRate 2 (\n -> print $ show n ++ " elements processed") (delay 1 Stream.|: delay 0.5 Stream.|: delay 0.5 Stream.|: Stream.nil)
535-- "2 elements processed"
536-- [1.0,0.5,0.5]
537-- "1 elements processed"
538-- @
539--
540-- Note: This may not work correctly on 32-bit machines.
541--
542-- /Pre-release/
543{-# INLINE tapRate #-}
544tapRate ::
545 (IsStream t, MonadAsync m, MonadCatch m)
546 => Double
547 -> (Int -> m b)
548 -> t m a
549 -> t m a
550tapRate n f xs = D.fromStreamD $ D.tapRate n f $ D.toStreamD xs
552-- | Apply a monadic function to each element flowing through the stream and
553-- discard the results.
554--
555-- @
556-- >>> Stream.drain $ Stream.trace print (Stream.enumerateFromTo 1 2)
557-- 1
558-- 2
559--
560-- @
561--
562-- Compare with 'tap'.
563--
564-- @since 0.7.0
565{-# INLINE trace #-}
566trace :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m a
567trace f = mapM (\x -> void (f x) >> return x)
569-- | Perform a side effect before yielding each element of the stream and
570-- discard the results.
571--
572-- @
573-- >>> Stream.drain $ Stream.trace_ (print "got here") (Stream.enumerateFromTo 1 2)
574-- "got here"
575-- "got here"
576--
577-- @
578--
579-- Same as 'interspersePrefix_' but always serial.
580--
581-- See also: 'trace'
582--
583-- /Pre-release/
584{-# INLINE trace_ #-}
585trace_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
586trace_ eff = Serial.mapM (\x -> eff >> return x)
588------------------------------------------------------------------------------
589-- Scanning with a Fold
590------------------------------------------------------------------------------
592-- | Scan a stream using the given monadic fold.
593--
594-- >>> Stream.toList $ Stream.takeWhile (< 10) $ Stream.scan Fold.sum (Stream.fromList [1..10])
595-- [0,1,3,6]
596--
597-- @since 0.7.0
598{-# INLINE scan #-}
599scan :: (IsStream t, Monad m) => Fold m a b -> t m a -> t m b
600scan = P.scanOnce
602-- | Postscan a stream using the given monadic fold.
603--
604-- The following example extracts the input stream up to a point where the
605-- running average of elements is no more than 10:
606--
607-- >>> import Data.Maybe (fromJust)
608-- >>> let avg = Fold.teeWith (/) Fold.sum (fmap fromIntegral Fold.length)
609-- >>> :{
610-- Stream.toList
611-- $ Stream.map (fromJust . fst)
612-- $ Stream.takeWhile (\(_,x) -> x <= 10)
613-- $ Stream.postscan (Fold.tee Fold.last avg) (Stream.enumerateFromTo 1.0 100.0)
614-- :}
615-- [1.0,2.0,3.0,4.0,5.0,6.0,7.0,8.0,9.0,10.0,11.0,12.0,13.0,14.0,15.0,16.0,17.0,18.0,19.0]
616--
617-- @since 0.7.0
618{-# INLINE postscan #-}
619postscan :: (IsStream t, Monad m) => Fold m a b -> t m a -> t m b
620postscan = P.postscanOnce
622------------------------------------------------------------------------------
623-- Scanning - Transformation by Folding
624------------------------------------------------------------------------------
626-- XXX It may be useful to have a version of scan where we can keep the
627-- accumulator independent of the value emitted. So that we do not necessarily
628-- have to keep a value in the accumulator which we are not using. We can pass
629-- an extraction function that will take the accumulator and the current value
630-- of the element and emit the next value in the stream. That will also make it
631-- possible to modify the accumulator after using it. In fact, the step function
632-- can return new accumulator and the value to be emitted. The signature would
633-- be more like mapAccumL. Or we can change the signature of scanx to
634-- accommodate this.
635--
636-- | Strict left scan with an extraction function. Like 'scanl'', but applies a
637-- user supplied extraction function (the third argument) at each step. This is
638-- designed to work with the @foldl@ library. The suffix @x@ is a mnemonic for
639-- extraction.
640--
641-- /Since 0.2.0/
642--
643-- /Since: 0.7.0 (Monad m constraint)/
644{-# DEPRECATED scanx "Please use scanl followed by map instead." #-}
645{-# INLINE scanx #-}
646scanx :: (IsStream t, Monad m) => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
647scanx = P.scanlx'
649-- XXX this needs to be concurrent
650-- XXX because of the use of D.cons for appending, scanlM' has quadratic
651-- complexity when iterated over a stream. We should use StreamK style scanlM'
652-- for linear performance on iteration.
653--
654-- | Like 'scanl'' but with a monadic step function and a monadic seed.
655--
656-- /Since: 0.4.0/
657--
658-- /Since: 0.8.0 (signature change)/
659{-# INLINE scanlM' #-}
660scanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
661scanlM' step begin m = fromStreamD $ D.scanlM' step begin $ toStreamD m
663-- XXX because of the use of D.cons for appending, scanl' has quadratic
664-- complexity when iterated over a stream. We should use StreamK style scanlM'
665-- for linear performance on iteration.
667-- | Strict left scan. Like 'map', 'scanl'' too is a one to one transformation,
668-- however it adds an extra element.
669--
670-- @
671-- >>> Stream.toList $ Stream.scanl' (+) 0 $ fromList [1,2,3,4]
672-- [0,1,3,6,10]
673--
674-- @
675--
676-- @
677-- >>> Stream.toList $ Stream.scanl' (flip (:)) [] $ Stream.fromList [1,2,3,4]
678-- [[],[1],[2,1],[3,2,1],[4,3,2,1]]
679--
680-- @
681--
682-- The output of 'scanl'' is the initial value of the accumulator followed by
683-- all the intermediate steps and the final result of 'foldl''.
684--
685-- By streaming the accumulated state after each fold step, we can share the
686-- state across multiple stages of stream composition. Each stage can modify or
687-- extend the state, do some processing with it and emit it for the next stage,
688-- thus modularizing the stream processing. This can be useful in
689-- stateful or event-driven programming.
690--
691-- Consider the following monolithic example, computing the sum and the product
692-- of the elements in a stream in one go using a @foldl'@:
693--
694-- @
695-- >>> Stream.foldl' (\(s, p) x -> (s + x, p * x)) (0,1) $ Stream.fromList [1,2,3,4]
696-- (10,24)
697--
698-- @
699--
700-- Using @scanl'@ we can make it modular by computing the sum in the first
701-- stage and passing it down to the next stage for computing the product:
702--
703-- @
704-- >>> :{
705-- Stream.foldl' (\(_, p) (s, x) -> (s, p * x)) (0,1)
706-- $ Stream.scanl' (\(s, _) x -> (s + x, x)) (0,1)
707-- $ Stream.fromList [1,2,3,4]
708-- :}
709-- (10,24)
710--
711-- @
712--
713-- IMPORTANT: 'scanl'' evaluates the accumulator to WHNF. To avoid building
714-- lazy expressions inside the accumulator, it is recommended that a strict
715-- data structure is used for accumulator.
716--
717-- >>> scanl' f z xs = scanlM' (\a b -> return (f a b)) (return z) xs
718-- >>> scanl' f z xs = z `Stream.cons` postscanl' f z xs
719--
720-- See also: 'usingStateT'
721--
722-- @since 0.2.0
723{-# INLINE scanl' #-}
724scanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
725scanl' step z m = fromStreamS $ S.scanl' step z $ toStreamS m
727-- | Like 'scanl'' but does not stream the initial value of the accumulator.
728--
729-- >>> postscanl' f z = postscanlM' (\a b -> return (f a b)) (return z)
730-- >>> postscanl' f z xs = Stream.drop 1 $ Stream.scanl' f z xs
731--
732-- @since 0.7.0
733{-# INLINE postscanl' #-}
734postscanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
735postscanl' step z m = fromStreamD $ D.postscanl' step z $ toStreamD m
737-- XXX prescanl does not sound very useful, enable only if there is a
738-- compelling use case.
739--
740-- | Like scanl' but does not stream the final value of the accumulator.
741--
742-- /Pre-release/
743{-# INLINE prescanl' #-}
744prescanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
745prescanl' step z m = fromStreamD $ D.prescanl' step z $ toStreamD m
747-- XXX this needs to be concurrent
748-- | Like prescanl' but with a monadic step function and a monadic seed.
749--
750-- /Pre-release/
751{-# INLINE prescanlM' #-}
752prescanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
753prescanlM' step z m = fromStreamD $ D.prescanlM' step z $ toStreamD m
755-- XXX this needs to be concurrent
756-- | Like 'scanl1'' but with a monadic step function.
757--
758-- @since 0.6.0
759{-# INLINE scanl1M' #-}
760scanl1M' :: (IsStream t, Monad m) => (a -> a -> m a) -> t m a -> t m a
761scanl1M' step m = fromStreamD $ D.scanl1M' step $ toStreamD m
763-- | Like 'scanl'' but for a non-empty stream. The first element of the stream
764-- is used as the initial value of the accumulator. Does nothing if the stream
765-- is empty.
766--
767-- @
768-- >>> Stream.toList $ Stream.scanl1' (+) $ fromList [1,2,3,4]
769-- [1,3,6,10]
770--
771-- @
772--
773-- @since 0.6.0
774{-# INLINE scanl1' #-}
775scanl1' :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> t m a
776scanl1' step m = fromStreamD $ D.scanl1' step $ toStreamD m
778------------------------------------------------------------------------------
779-- Filtering
780------------------------------------------------------------------------------
782-- | Modify a @t m a -> t m a@ stream transformation that accepts a predicate
783-- @(a -> b)@ to accept @((s, a) -> b)@ instead, provided a transformation @t m
784-- a -> t m (s, a)@. Convenient to filter with index or time.
785--
786-- @
787-- filterWithIndex = with indexed filter
788-- filterWithAbsTime = with timestamped filter
789-- filterWithRelTime = with timeIndexed filter
790-- @
791--
792-- /Pre-release/
793{-# INLINE with #-}
794with :: forall (t :: (Type -> Type) -> Type -> Type) m a b s. Functor (t m) =>
795 (t m a -> t m (s, a))
796 -> (((s, a) -> b) -> t m (s, a) -> t m (s, a))
797 -> (((s, a) -> b) -> t m a -> t m a)
798with f comb g = fmap snd . comb g . f
800-- | Include only those elements that pass a predicate.
801--
802-- @since 0.1.0
803{-# INLINE filter #-}
804#if __GLASGOW_HASKELL__ != 802
805-- GHC 8.2.2 crashes with this code, when used with "stack"
806filter :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
807filter p m = fromStreamS $ S.filter p $ toStreamS m
808#else
809filter :: IsStream t => (a -> Bool) -> t m a -> t m a
810filter = K.filter
811#endif
813-- | Same as 'filter' but with a monadic predicate.
814--
815-- @since 0.4.0
816{-# INLINE filterM #-}
817filterM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
818filterM p m = fromStreamD $ D.filterM p $ toStreamD m
820-- | Drop repeated elements that are adjacent to each other using the supplied
821-- comparison function.
822--
823-- @uniq = uniqBy (==)
824--
825-- To strip duplicate path separators:
826--
827-- @
828-- f x y = x == '/' && x == y
829-- Stream.toList $ Stream.uniqBy f $ Stream.fromList "//a//b"
830-- "/a/b"
831-- @
832--
833-- Space: @O(1)@
834--
835-- See also: 'nubBy'.
836--
837-- /Pre-release/
838--
839{-# INLINE uniqBy #-}
840uniqBy :: (IsStream t, Monad m, Functor (t m)) =>
841 (a -> a -> Bool) -> t m a -> t m a
842uniqBy eq =
843 catMaybes . rollingMap (\x y -> if x `eq` y then Nothing else Just y)
845-- | Drop repeated elements that are adjacent to each other.
846--
847-- @since 0.6.0
848{-# INLINE uniq #-}
849uniq :: (Eq a, IsStream t, Monad m) => t m a -> t m a
850uniq = fromStreamD . D.uniq . toStreamD
852-- | Strip all leading and trailing occurrences of an element passing a
853-- predicate and make all other consecutive occurrences uniq.
854--
855-- @
856-- prune p = dropWhileAround p $ uniqBy (x y -> p x && p y)
857-- @
858--
859-- @
860-- > Stream.prune isSpace (Stream.fromList " hello world! ")
861-- "hello world!"
862--
863-- @
864--
865-- Space: @O(1)@
866--
867-- /Unimplemented/
868{-# INLINE prune #-}
869prune ::
870 -- (IsStream t, Monad m, Eq a) =>
871 (a -> Bool) -> t m a -> t m a
872prune = error "Not implemented yet!"
874-- Possible implementation:
875-- @repeated =
876-- Stream.catMaybes . Stream.parseMany (Parser.groupBy (==) Fold.repeated)@
877--
878-- 'Fold.repeated' should return 'Just' when repeated, and 'Nothing' for a
879-- single element.
880--
881-- | Emit only repeated elements, once.
882--
883-- /Unimplemented/
884repeated :: -- (IsStream t, Monad m, Eq a) =>
885 t m a -> t m a
886repeated = undefined
888-- We can have more efficient implementations for nubOrd and nubInt by using
889-- Set and IntSet to find/remove duplication. For Hashable we can use a
890-- hashmap. Use rewrite rules to specialize to more efficient impls.
891--
892-- | Drop repeated elements anywhere in the stream.
893--
894-- /Caution: not scalable for infinite streams/
895--
896-- /See also: nubWindowBy/
897--
898-- /Unimplemented/
899--
900{-# INLINE nubBy #-}
901nubBy :: -- (IsStream t, Monad m) =>
902 (a -> a -> Bool) -> t m a -> t m a
903nubBy = undefined -- fromStreamD . D.nubBy . toStreamD
905-- | Drop repeated elements within the specified tumbling window in the stream.
906--
907-- @nubBy = nubWindowBy maxBound@
908--
909-- /Unimplemented/
910--
911{-# INLINE nubWindowBy #-}
912nubWindowBy :: -- (IsStream t, Monad m) =>
913 Int -> (a -> a -> Bool) -> t m a -> t m a
914nubWindowBy = undefined -- fromStreamD . D.nubWithinBy . toStreamD
916-- | Deletes the first occurrence of the element in the stream that satisfies
917-- the given equality predicate.
918--
919-- @
920-- >>> Stream.toList $ Stream.deleteBy (==) 3 $ Stream.fromList [1,3,3,5]
921-- [1,3,5]
922--
923-- @
924--
925-- @since 0.6.0
926{-# INLINE deleteBy #-}
927deleteBy :: (IsStream t, Monad m) => (a -> a -> Bool) -> a -> t m a -> t m a
928deleteBy cmp x m = fromStreamS $ S.deleteBy cmp x (toStreamS m)
930------------------------------------------------------------------------------
931-- Lossy Buffering
932------------------------------------------------------------------------------
934-- XXX We could use 'maxBuffer Block/Drop/Rotate/Sample' instead. However we
935-- may want to have the evaluation rate independent of the sampling rate. To
936-- support that we can decouple evaluation and sampling in independent stages.
937-- The sampling stage would strictly evaluate and sample, the evaluation stage
938-- would control the evaluation rate.
940-- | Evaluate the input stream continuously and keep only the oldest @n@
941-- elements in the buffer, discard the new ones when the buffer is full. When
942-- the output stream is evaluated it consumes the values from the buffer in a
943-- FIFO manner.
944--
945-- /Unimplemented/
946--
947{-# INLINE sampleOld #-}
948sampleOld :: -- (IsStream t, Monad m) =>
949 Int -> t m a -> t m a
950sampleOld = undefined
952-- | Evaluate the input stream continuously and keep only the latest @n@
953-- elements in a ring buffer, keep discarding the older ones to make space for
954-- the new ones. When the output stream is evaluated it consumes the values
955-- from the buffer in a FIFO manner.
956--
957-- /Unimplemented/
958--
959{-# INLINE sampleNew #-}
960sampleNew :: -- (IsStream t, Monad m) =>
961 Int -> t m a -> t m a
962sampleNew = undefined
964-- | Like 'sampleNew' but samples at uniform intervals to match the consumer
965-- rate. Note that 'sampleNew' leads to non-uniform sampling depending on the
966-- consumer pattern.
967--
968-- /Unimplemented/
969--
970{-# INLINE sampleRate #-}
971sampleRate :: -- (IsStream t, Monad m) =>
972 Double -> t m a -> t m a
973sampleRate = undefined
975------------------------------------------------------------------------------
976-- Trimming
977------------------------------------------------------------------------------
979-- | Same as 'takeWhile' but with a monadic predicate.
980--
981-- @since 0.4.0
982{-# INLINE takeWhileM #-}
983takeWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
984takeWhileM p m = fromStreamD $ D.takeWhileM p $ toStreamD m
986-- See the lastN fold for impl hints. Use a Data.Array based ring buffer.
987--
988-- takeLast n = Stream.concatM . fmap Array.toStream . fold Fold.lastN
989--
990-- | Take @n@ elements at the end of the stream.
991--
992-- O(n) space, where n is the number elements taken.
993--
994-- /Unimplemented/
995{-# INLINE takeLast #-}
996takeLast :: -- (IsStream t, Monad m) =>
997 Int -> t m a -> t m a
998takeLast = undefined -- fromStreamD $ D.takeLast n $ toStreamD m
1000-- | Take time interval @i@ seconds at the end of the stream.
1002-- O(n) space, where n is the number elements taken.
1004-- /Unimplemented/
1005{-# INLINE takeLastInterval #-}
1006takeLastInterval :: -- (IsStream t, Monad m) =>
1007 Double -> t m a -> t m a
1008takeLastInterval = undefined -- fromStreamD $ D.takeLast n $ toStreamD m
1010-- | Take all consecutive elements at the end of the stream for which the
1011-- predicate is true.
1013-- O(n) space, where n is the number elements taken.
1015-- /Unimplemented/
1016{-# INLINE takeWhileLast #-}
1017takeWhileLast :: -- (IsStream t, Monad m) =>
1018 (a -> Bool) -> t m a -> t m a
1019takeWhileLast = undefined -- fromStreamD $ D.takeWhileLast n $ toStreamD m
1021-- | Like 'takeWhile' and 'takeWhileLast' combined.
1023-- O(n) space, where n is the number elements taken from the end.
1025-- /Unimplemented/
1026{-# INLINE takeWhileAround #-}
1027takeWhileAround :: -- (IsStream t, Monad m) =>
1028 (a -> Bool) -> t m a -> t m a
1029takeWhileAround = undefined -- fromStreamD $ D.takeWhileAround n $ toStreamD m
1031-- | @takeInterval duration@ yields stream elements upto specified time
1032-- @duration@. The duration starts when the stream is evaluated for the first
1033-- time, before the first element is yielded. The time duration is checked
1034-- before generating each element, if the duration has expired the stream
1035-- stops.
1037-- The total time taken in executing the stream is guaranteed to be /at least/
1038-- @duration@, however, because the duration is checked before generating an
1039-- element, the upper bound is indeterminate and depends on the time taken in
1040-- generating and processing the last element.
1042-- No element is yielded if the duration is zero. At least one element is
1043-- yielded if the duration is non-zero.
1045-- /Pre-release/
1047{-# INLINE takeInterval #-}
1048takeInterval ::(MonadIO m, IsStream t, TimeUnit64 d) => d -> t m a -> t m a
1049takeInterval d = fromStreamD . D.takeByTime d . toStreamD
1051-- | Drop elements in the stream as long as the predicate succeeds and then
1052-- take the rest of the stream.
1054-- @since 0.1.0
1055{-# INLINE dropWhile #-}
1056dropWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
1057dropWhile p m = fromStreamS $ S.dropWhile p $ toStreamS m
1059-- | Same as 'dropWhile' but with a monadic predicate.
1061-- @since 0.4.0
1062{-# INLINE dropWhileM #-}
1063dropWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
1064dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m
1066-- | @dropInterval duration@ drops stream elements until specified @duration@ has
1067-- passed. The duration begins when the stream is evaluated for the first
1068-- time. The time duration is checked /after/ generating a stream element, the
1069-- element is yielded if the duration has expired otherwise it is dropped.
1071-- The time elapsed before starting to generate the first element is /at most/
1072-- @duration@, however, because the duration expiry is checked after the
1073-- element is generated, the lower bound is indeterminate and depends on the
1074-- time taken in generating an element.
1076-- All elements are yielded if the duration is zero.
1078-- /Pre-release/
1080{-# INLINE dropInterval #-}
1081dropInterval ::(MonadIO m, IsStream t, TimeUnit64 d) => d -> t m a -> t m a
1082dropInterval d = fromStreamD . D.dropByTime d . toStreamD
1084-- | Drop @n@ elements at the end of the stream.
1086-- O(n) space, where n is the number elements dropped.
1088-- /Unimplemented/
1089{-# INLINE dropLast #-}
1090dropLast :: -- (IsStream t, Monad m) =>
1091 Int -> t m a -> t m a
1092dropLast = undefined -- fromStreamD $ D.dropLast n $ toStreamD m
1094-- | Drop time interval @i@ seconds at the end of the stream.
1096-- O(n) space, where n is the number elements dropped.
1098-- /Unimplemented/
1099{-# INLINE dropLastInterval #-}
1100dropLastInterval :: -- (IsStream t, Monad m) =>
1101 Int -> t m a -> t m a
1102dropLastInterval = undefined
1104-- | Drop all consecutive elements at the end of the stream for which the
1105-- predicate is true.
1107-- O(n) space, where n is the number elements dropped.
1109-- /Unimplemented/
1110{-# INLINE dropWhileLast #-}
1111dropWhileLast :: -- (IsStream t, Monad m) =>
1112 (a -> Bool) -> t m a -> t m a
1113dropWhileLast = undefined -- fromStreamD $ D.dropWhileLast n $ toStreamD m
1115-- | Like 'dropWhile' and 'dropWhileLast' combined.
1117-- O(n) space, where n is the number elements dropped from the end.
1119-- /Unimplemented/
1120{-# INLINE dropWhileAround #-}
1121dropWhileAround :: -- (IsStream t, Monad m) =>
1122 (a -> Bool) -> t m a -> t m a
1123dropWhileAround = undefined -- fromStreamD $ D.dropWhileAround n $ toStreamD m
1125------------------------------------------------------------------------------
1126-- Inserting Elements
1127------------------------------------------------------------------------------
1129-- | @insertBy cmp elem stream@ inserts @elem@ before the first element in
1130-- @stream@ that is less than @elem@ when compared using @cmp@.
1132-- @
1133-- insertBy cmp x = 'mergeBy' cmp ('fromPure' x)
1134-- @
1136-- @
1137-- >>> Stream.toList $ Stream.insertBy compare 2 $ Stream.fromList [1,3,5]
1138-- [1,2,3,5]
1140-- @
1142-- @since 0.6.0
1143{-# INLINE insertBy #-}
1144insertBy ::
1145 (IsStream t, Monad m) => (a -> a -> Ordering) -> a -> t m a -> t m a
1146insertBy cmp x m = fromStreamS $ S.insertBy cmp x (toStreamS m)
1148-- | Insert a pure value between successive elements of a stream.
1150-- >>> Stream.toList $ Stream.intersperse ',' $ Stream.fromList "hello"
1151-- "h,e,l,l,o"
1153-- @since 0.7.0
1154{-# INLINE intersperse #-}
1155intersperse :: (IsStream t, MonadAsync m) => a -> t m a -> t m a
1156intersperse a = fromStreamS . S.intersperse a . toStreamS
1158-- | Insert a side effect before consuming an element of a stream except the
1159-- first one.
1161-- >>> Stream.drain $ Stream.trace putChar $ Stream.intersperseM_ (putChar '.') $ Stream.fromList "hello"
1162-- h.e.l.l.o
1164-- /Pre-release/
1165{-# INLINE intersperseM_ #-}
1166intersperseM_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
1167intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
1169-- | Intersperse a monadic action into the input stream after every @n@
1170-- elements.
1172-- @
1173-- > Stream.toList $ Stream.intersperseBySpan 2 (return ',') $ Stream.fromList "hello"
1174-- "he,ll,o"
1176-- @
1178-- /Unimplemented/
1179{-# INLINE intersperseBySpan #-}
1180intersperseBySpan :: -- IsStream t =>
1181 Int -> m a -> t m a -> t m a
1182intersperseBySpan _n _f _xs = undefined
1184-- | Insert an effect and its output after consuming an element of a stream.
1186-- >>> Stream.toList $ Stream.trace putChar $ intersperseSuffix (putChar '.' >> return ',') $ Stream.fromList "hello"
1187-- h.,e.,l.,l.,o.,"h,e,l,l,o,"
1189-- /Pre-release/
1190{-# INLINE intersperseSuffix #-}
1191intersperseSuffix :: (IsStream t, Monad m) => m a -> t m a -> t m a
1192intersperseSuffix m = fromStreamD . D.intersperseSuffix m . toStreamD
1194-- | Insert a side effect after consuming an element of a stream.
1196-- @
1197-- >>> Stream.mapM_ putChar $ Stream.intersperseSuffix_ (threadDelay 1000000) $ Stream.fromList "hello"
1198-- hello
1200-- @
1202-- /Pre-release/
1204{-# INLINE intersperseSuffix_ #-}
1205intersperseSuffix_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
1206intersperseSuffix_ m = fromStreamD . D.intersperseSuffix_ m . toStreamD
1208-- XXX Use an offset argument, like tapOffsetEvery
1210-- | Like 'intersperseSuffix' but intersperses an effectful action into the
1211-- input stream after every @n@ elements and after the last element.
1213-- >>> Stream.toList $ Stream.intersperseSuffixBySpan 2 (return ',') $ Stream.fromList "hello"
1214-- "he,ll,o,"
1216-- /Pre-release/
1218{-# INLINE intersperseSuffixBySpan #-}
1219intersperseSuffixBySpan :: (IsStream t, Monad m)
1220 => Int -> m a -> t m a -> t m a
1221intersperseSuffixBySpan n eff =
1222 fromStreamD . D.intersperseSuffixBySpan n eff . toStreamD
1224-- | Insert a side effect before consuming an element of a stream.
1226-- >>> Stream.toList $ Stream.trace putChar $ Stream.interspersePrefix_ (putChar '.' >> return ',') $ Stream.fromList "hello"
1227-- .h.e.l.l.o"hello"
1229-- Same as 'trace_' but may be concurrent.
1231-- /Concurrent/
1233-- /Pre-release/
1235{-# INLINE interspersePrefix_ #-}
1236interspersePrefix_ :: (IsStream t, MonadAsync m) => m b -> t m a -> t m a
1237interspersePrefix_ m = mapM (\x -> void m >> return x)
1239------------------------------------------------------------------------------
1240-- Inserting Time
1241------------------------------------------------------------------------------
1243-- Note: delay must be serial.
1245-- | Introduce a delay of specified seconds before consuming an element of the
1246-- stream except the first one.
1248-- >>> Stream.mapM_ print $ Stream.timestamped $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
1249-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),1)
1250-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),2)
1251-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),3)
1253-- @since 0.8.0
1255{-# INLINE delay #-}
1256delay :: (IsStream t, MonadIO m) => Double -> t m a -> t m a
1257delay n = intersperseM_ $ liftIO $ threadDelay $ round $ n * 1000000
1259-- Note: delay must be serial.
1261-- | Introduce a delay of specified seconds after consuming an element of a
1262-- stream.
1264-- >>> Stream.mapM_ print $ Stream.timestamped $ Stream.delayPost 1 $ Stream.enumerateFromTo 1 3
1265-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),1)
1266-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),2)
1267-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),3)
1269-- /Pre-release/
1271{-# INLINE delayPost #-}
1272delayPost :: (IsStream t, MonadIO m) => Double -> t m a -> t m a
1273delayPost n = intersperseSuffix_ $ liftIO $ threadDelay $ round $ n * 1000000
1275-- Note: delay must be serial, that's why 'trace_' is used.
1277-- | Introduce a delay of specified seconds before consuming an element of a
1278-- stream.
1280-- >>> Stream.mapM_ print $ Stream.timestamped $ Stream.delayPre 1 $ Stream.enumerateFromTo 1 3
1281-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),1)
1282-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),2)
1283-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),3)
1285-- /Pre-release/
1287{-# INLINE delayPre #-}
1288delayPre :: (IsStream t, MonadIO m) => Double -> t m a -> t m a
1289delayPre n = trace_ $ liftIO $ threadDelay $ round $ n * 1000000
1291------------------------------------------------------------------------------
1292-- Reorder in sequence
1293------------------------------------------------------------------------------
1295-- | Buffer until the next element in sequence arrives. The function argument
1296-- determines the difference in sequence numbers. This could be useful in
1297-- implementing sequenced streams, for example, TCP reassembly.
1299-- /Unimplemented/
1301{-# INLINE reassembleBy #-}
1302reassembleBy
1303 :: -- (IsStream t, Monad m) =>
1304 Fold m a b
1305 -> (a -> a -> Int)
1306 -> t m a
1307 -> t m b
1308reassembleBy = undefined
1310------------------------------------------------------------------------------
1311-- Position Indexing
1312------------------------------------------------------------------------------
1314-- |
1315-- > indexed = Stream.postscanl' (\(i, _) x -> (i + 1, x)) (-1,undefined)
1316-- > indexed = Stream.zipWith (,) (Stream.enumerateFrom 0)
1318-- Pair each element in a stream with its index, starting from index 0.
1320-- >>> Stream.toList $ Stream.indexed $ Stream.fromList "hello"
1321-- [(0,'h'),(1,'e'),(2,'l'),(3,'l'),(4,'o')]
1323-- @since 0.6.0
1324{-# INLINE indexed #-}
1325indexed :: (IsStream t, Monad m) => t m a -> t m (Int, a)
1326indexed = fromStreamD . D.indexed . toStreamD
1328-- |
1329-- > indexedR n = Stream.postscanl' (\(i, _) x -> (i - 1, x)) (n + 1,undefined)
1330-- > indexedR n = Stream.zipWith (,) (Stream.enumerateFromThen n (n - 1))
1332-- Pair each element in a stream with its index, starting from the
1333-- given index @n@ and counting down.
1335-- >>> Stream.toList $ Stream.indexedR 10 $ Stream.fromList "hello"
1336-- [(10,'h'),(9,'e'),(8,'l'),(7,'l'),(6,'o')]
1338-- @since 0.6.0
1339{-# INLINE indexedR #-}
1340indexedR :: (IsStream t, Monad m) => Int -> t m a -> t m (Int, a)
1341indexedR n = fromStreamD . D.indexedR n . toStreamD
1343-------------------------------------------------------------------------------
1344-- Time Indexing
1345-------------------------------------------------------------------------------
1347-- Note: The timestamp stream must be the second stream in the zip so that the
1348-- timestamp is generated after generating the stream element and not before.
1349-- If we do not do that then the following example will generate the same
1350-- timestamp for first two elements:
1352-- Stream.mapM_ print $ Stream.timestamped $ Stream.delay $ Stream.enumerateFromTo 1 3
1354-- | Pair each element in a stream with an absolute timestamp, using a clock of
1355-- specified granularity. The timestamp is generated just before the element
1356-- is consumed.
1358-- >>> Stream.mapM_ print $ Stream.timestampWith 0.01 $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
1359-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),1)
1360-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),2)
1361-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),3)
1363-- /Pre-release/
1365{-# INLINE timestampWith #-}
1366timestampWith :: (IsStream t, MonadAsync m, Functor (t m))
1367 => Double -> t m a -> t m (AbsTime, a)
1368timestampWith g stream = Z.zipWith (flip (,)) stream (absTimesWith g)
1370-- TBD: check performance vs a custom implementation without using zipWith.
1372-- /Pre-release/
1374{-# INLINE timestamped #-}
1375timestamped :: (IsStream t, MonadAsync m, Functor (t m))
1376 => t m a -> t m (AbsTime, a)
1377timestamped = timestampWith 0.01
1379-- | Pair each element in a stream with relative times starting from 0, using a
1380-- clock with the specified granularity. The time is measured just before the
1381-- element is consumed.
1383-- >>> Stream.mapM_ print $ Stream.timeIndexWith 0.01 $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
1384-- (RelTime64 (NanoSecond64 ...),1)
1385-- (RelTime64 (NanoSecond64 ...),2)
1386-- (RelTime64 (NanoSecond64 ...),3)
1388-- /Pre-release/
1390{-# INLINE timeIndexWith #-}
1391timeIndexWith :: (IsStream t, MonadAsync m, Functor (t m))
1392 => Double -> t m a -> t m (RelTime64, a)
1393timeIndexWith g stream = Z.zipWith (flip (,)) stream (relTimesWith g)
1395-- | Pair each element in a stream with relative times starting from 0, using a
1396-- 10 ms granularity clock. The time is measured just before the element is
1397-- consumed.
1399-- >>> Stream.mapM_ print $ Stream.timeIndexed $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
1400-- (RelTime64 (NanoSecond64 ...),1)
1401-- (RelTime64 (NanoSecond64 ...),2)
1402-- (RelTime64 (NanoSecond64 ...),3)
1404-- /Pre-release/
1406{-# INLINE timeIndexed #-}
1407timeIndexed :: (IsStream t, MonadAsync m, Functor (t m))
1408 => t m a -> t m (RelTime64, a)
1409timeIndexed = timeIndexWith 0.01
1411------------------------------------------------------------------------------
1412-- Searching
1413------------------------------------------------------------------------------
1415-- | Find all the indices where the value of the element in the stream is equal
1416-- to the given value.
1418-- > elemIndices a = findIndices (== a)
1420-- @since 0.5.0
1421{-# INLINE elemIndices #-}
1422elemIndices :: (IsStream t, Eq a, Monad m) => a -> t m a -> t m Int
1423elemIndices a = findIndices (== a)
1425------------------------------------------------------------------------------
1426-- Rolling map
1427------------------------------------------------------------------------------
1429-- XXX this is not a one-to-one map so calling it map may not be right.
1430-- We can perhaps call it zipWithTail or rollWith.
1432-- | Apply a function on every two successive elements of a stream. If the
1433-- stream consists of a single element the output is an empty stream.
1435-- This is the stream equivalent of the list idiom @zipWith f xs (tail xs)@.
1437-- /Pre-release/
1439{-# INLINE rollingMap #-}
1440rollingMap :: (IsStream t, Monad m) => (a -> a -> b) -> t m a -> t m b
1441rollingMap f m = fromStreamD $ D.rollingMap f $ toStreamD m
1443-- | Like 'rollingMap' but with an effectful map function.
1445-- /Pre-release/
1447{-# INLINE rollingMapM #-}
1448rollingMapM :: (IsStream t, Monad m) => (a -> a -> m b) -> t m a -> t m b
1449rollingMapM f m = fromStreamD $ D.rollingMapM f $ toStreamD m
1451------------------------------------------------------------------------------
1452-- Maybe Streams
1453------------------------------------------------------------------------------
1455-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'
1456-- elements, and return a stream of values extracted from 'Just'.
1458-- Equivalent to:
1460-- @
1461-- mapMaybe f = Stream.map 'fromJust' . Stream.filter 'isJust' . Stream.map f
1462-- @
1464-- @since 0.3.0
1465{-# INLINE mapMaybe #-}
1466mapMaybe :: (IsStream t, Monad m) => (a -> Maybe b) -> t m a -> t m b
1467mapMaybe f m = fromStreamS $ S.mapMaybe f $ toStreamS m
1469-- | Like 'mapMaybe' but maps a monadic function.
1471-- Equivalent to:
1473-- @
1474-- mapMaybeM f = Stream.map 'fromJust' . Stream.filter 'isJust' . Stream.mapM f
1475-- @
1477-- /Concurrent (do not use with 'fromParallel' on infinite streams)/
1479-- @since 0.3.0
1480{-# INLINE_EARLY mapMaybeM #-}
1481mapMaybeM :: (IsStream t, MonadAsync m, Functor (t m))
1482 => (a -> m (Maybe b)) -> t m a -> t m b
1483mapMaybeM f = fmap fromJust . filter isJust . K.mapM f
1485{-# RULES "mapMaybeM serial" mapMaybeM = mapMaybeMSerial #-}
1486{-# INLINE mapMaybeMSerial #-}
1487mapMaybeMSerial :: Monad m => (a -> m (Maybe b)) -> SerialT m a -> SerialT m b
1488mapMaybeMSerial f m = fromStreamD $ D.mapMaybeM f $ toStreamD m
1490-- | In a stream of 'Maybe's, discard 'Nothing's and unwrap 'Just's.
1492-- /Pre-release/
1494{-# INLINE catMaybes #-}
1495catMaybes :: (IsStream t, Monad m, Functor (t m)) => t m (Maybe a) -> t m a
1496catMaybes = fmap fromJust . filter isJust
1498------------------------------------------------------------------------------
1499-- Either streams
1500------------------------------------------------------------------------------
1502-- | Discard 'Right's and unwrap 'Left's in an 'Either' stream.
1504-- /Pre-release/
1506{-# INLINE lefts #-}
1507lefts :: (IsStream t, Monad m, Functor (t m)) => t m (Either a b) -> t m a
1508lefts = fmap (fromLeft undefined) . filter isLeft
1510-- | Discard 'Left's and unwrap 'Right's in an 'Either' stream.
1512-- /Pre-release/
1514{-# INLINE rights #-}
1515rights :: (IsStream t, Monad m, Functor (t m)) => t m (Either a b) -> t m b
1516rights = fmap (fromRight undefined) . filter isRight
1518-- | Remove the either wrapper and flatten both lefts and as well as rights in
1519-- the output stream.
1521-- /Pre-release/
1523{-# INLINE both #-}
1524both :: Functor (t m) => t m (Either a a) -> t m a
1525both = fmap (either id id)
1527------------------------------------------------------------------------------
1528-- Concurrent Application
1529------------------------------------------------------------------------------
1531-- | Parallel transform application operator; applies a stream transformation
1532-- function @t m a -> t m b@ to a stream @t m a@ concurrently; the input stream
1533-- is evaluated asynchronously in an independent thread yielding elements to a
1534-- buffer and the transformation function runs in another thread consuming the
1535-- input from the buffer. '|$' is just like regular function application
1536-- operator '$' except that it is concurrent.
1538-- If you read the signature as @(t m a -> t m b) -> (t m a -> t m b)@ you can
1539-- look at it as a transformation that converts a transform function to a
1540-- buffered concurrent transform function.
1542-- The following code prints a value every second even though each stage adds a
1543-- 1 second delay.
1546-- >>> :{
1547-- Stream.drain $
1548-- Stream.mapM (\x -> threadDelay 1000000 >> print x)
1549-- |$ Stream.replicateM 3 (threadDelay 1000000 >> return 1)
1550-- :}
1551-- 1
1552-- 1
1553-- 1
1555-- /Concurrent/
1557-- /Since: 0.3.0 ("Streamly")/
1559-- @since 0.8.0
1560{-# INLINE (|$) #-}
1561(|$) :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> (t m a -> t m b)
1562-- (|$) f = f . Async.mkAsync
1563(|$) f = f . Par.mkParallel
1565infixr 0 |$
1567-- | Same as '|$'.
1569-- /Internal/
1571{-# INLINE applyAsync #-}
1572applyAsync :: (IsStream t, MonadAsync m)
1573 => (t m a -> t m b) -> (t m a -> t m b)
1574applyAsync = (|$)
1576-- | Same as '|$' but with arguments reversed.
1578-- (|&) = flip (|$)
1580-- /Concurrent/
1582-- /Since: 0.3.0 ("Streamly")/
1584-- @since 0.8.0
1585{-# INLINE (|&) #-}
1586(|&) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> t m b) -> t m b
1587x |& f = f |$ x
1589infixl 1 |&