@@ -8,80 +8,968 @@
9module Streamly.Internal.Data.Stream.Transform
9module Streamly.Internal.Data.Stream.Transform
11− filter
11+ -- * Piping
12− , foldFilter
12+ -- | Pass through a 'Pipe'.
21+ -- | Stateless one-to-one maps.
27+ -- * Mapping Side Effects (Observation)
28+ -- | See also the intersperse*_ combinators.
38+ -- * Scanning By 'Fold'
44+ -- | Left scans. Stateful, mostly one-to-one maps.
56+ -- XXX Once we have pipes the contravariant transformations can be
57+ -- represented by attaching pipes before a transformation.
70+ -- | Produce a subset of the stream using criteria based on the values of
71+ -- the elements. We can use a concatMap and scan for filtering but these
72+ -- combinators are more efficient and convenient.
87+ -- | Produce a subset of the stream trimmed at ends.
100+ --, dropLastInterval
106+ -- * Inserting Elements
107+ -- | Produce a superset of the stream. This is the opposite of
108+ -- filtering/sampling. We can always use concatMap and scan for inserting
109+ -- but these combinators are more efficient and convenient.
111+ -- Element agnostic (Opposite of sampling)
113+ , intersperseM -- XXX naming
114+ --, intersperseBySpan
117+ , intersperseSuffixBySpan
120+ -- , interspersePrefix
121+ -- , interspersePrefixBySpan
123+ -- * Inserting Side Effects/Time
124+ , intersperseM_ -- XXX naming
126+ , intersperseSuffix_
128+ , interspersePrefix_
131+ -- * Element Aware Insertion
132+ -- | Opposite of filtering
134+ -- , intersperseByBefore
135+ -- , intersperseByAfter
142+ -- * Position Indexing
147+ , findIndices -- XXX indicesBy
148+ , elemIndices -- XXX indicesOf
151+ -- | Map using the previous element.
157+ -- Move these to Streamly.Data.Maybe.Stream?
158+ , catMaybes -- XXX justs (like lefts/rights)
163+ -- Move these to Streamly.Data.Either.Stream?
170+import Control.Concurrent (threadDelay)
171+import Control.Monad (void)
172+import Control.Monad.IO.Class (MonadIO (liftIO))
173+import Data.Either (fromLeft, isLeft, isRight, fromRight)
174+import Data.Maybe (isJust, fromJust)
18import Streamly.Internal.Data.Fold.Type (Fold)
176import Streamly.Internal.Data.Fold.Type (Fold)
19−import Streamly.Internal.Data.Stream.Type (Stream)
177+import Streamly.Internal.Data.Pipe (Pipe)
178+import Streamly.Internal.Data.Time.Units (TimeUnit64)
180+import qualified Streamly.Internal.Data.Fold as FL
181+import qualified Streamly.Internal.Data.Stream.StreamD.Transform as D
182+import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
183+import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
185+import Streamly.Internal.Data.Stream.Bottom
21−import qualified Streamly.Internal.Data.Stream.Type as Stream
186+import Streamly.Internal.Data.Stream.Type
22−import qualified Streamly.Internal.Data.Stream.StreamD.Transform as StreamD
24−import Prelude hiding (map, mapM, filter)
188+import Prelude hiding
189+ ( filter, drop, dropWhile, take, takeWhile, foldr, map, mapM, sequence
190+ , reverse, foldr1 , repeat, scanl, scanl1, zipWith)
192+#include "Instances.hs"
198+-- >>> import Control.Concurrent (threadDelay)
199+-- >>> import Data.Function ((&))
200+-- >>> import Prelude hiding ( filter, drop, dropWhile, take, takeWhile, foldr, map, mapM, sequence, reverse, foldr1 , scanl, scanl1)
201+-- >>> import Streamly.Internal.Data.Stream as Stream
202+-- >>> import qualified Streamly.Data.Fold as Fold
29-- >>> import qualified Streamly.Internal.Data.Fold as Fold
203-- >>> import qualified Streamly.Internal.Data.Fold as Fold
30−-- >>> import qualified Streamly.Data.Unfold as Unfold
204+-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
31−-- >>> import qualified Streamly.Internal.Data.Stream as Stream
205+-- >>> import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
207+-- >>> hSetBuffering stdout LineBuffering
33------------------------------------------------------------------------------
209------------------------------------------------------------------------------
34-- Transformation by Mapping
210-- Transformation by Mapping
35------------------------------------------------------------------------------
211------------------------------------------------------------------------------
38−-- >> mapM f = Stream.sequence . Stream.map f
215+-- mapM f = sequence . map f
40-- Apply a monadic function to each element of the stream and replace it with
218-- Apply a monadic function to each element of the stream and replace it with
41-- the output of the resulting action.
219-- the output of the resulting action.
43−-- >>> s = Stream.unfold Unfold.fromList ["a", "b", "c"]
222+-- > drain $ Stream.mapM putStr $ Stream.unfold Unfold.fromList ["a", "b", "c"]
44−-- >>> Stream.fold Fold.drain $ Stream.mapM putStr s
47−-- /Pre-release/
225+-- > :{
226+-- drain $ Stream.replicateM 10 (return 1)
227+-- & (fromSerial . Stream.mapM (\x -> threadDelay 1000000 >> print x))
49−{-# INLINE mapM #-}
233+-- > drain $ Stream.replicateM 10 (return 1)
50−mapM :: Monad m => (a -> m b) -> Stream m a -> Stream m b
234+-- & (fromAsync . Stream.mapM (\x -> threadDelay 1000000 >> print x))
51−mapM f = Stream.fromStreamD . StreamD.mapM f . Stream.toStreamD
235+-- @
56−-- Same as 'fmap'.
237+-- /Concurrent (do not use with 'fromParallel' on infinite streams)/
58−-- >>> s = Stream.unfold Unfold.fromList [1,2,3]
59−-- >>> Stream.fold Fold.toList $ Stream.map (+1) s
62-- /Pre-release/
239-- /Pre-release/
64−{-# INLINE map #-}
241+{-# INLINE mapM #-}
65−map :: Monad m => (a -> b) -> Stream m a -> Stream m b
242+mapM :: Monad m => (a -> m b) -> Stream m a -> Stream m b
66−map f = mapM (return . f)
243+mapM f m = fromStreamK $ D.toStreamK $ D.mapM f $ toStreamD m
68-- | Include only those elements that pass a predicate.
245-- | Include only those elements that pass a predicate.
70{-# INLINE filter #-}
247{-# INLINE filter #-}
71filter :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
248filter :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
72−filter p = Stream.fromStreamD . StreamD.filter p . Stream.toStreamD
249+filter p m = fromStreamD $ D.filter p $ toStreamD m
251+-- | Same as 'filter' but with a monadic predicate.
254+{-# INLINE filterM #-}
255+filterM :: (Monad m) => (a -> m Bool) -> Stream m a -> Stream m a
256+filterM p m = fromStreamD $ D.filterM p $ toStreamD m
74-- | Use a filtering fold on a stream.
258-- | Use a filtering fold on a stream.
77−-- Stream.fold Fold.sum
78−-- $ Stream.foldFilter (Fold.maybe (\x -> if x > 5 then Just x else Nothing))
79−-- $ Stream.unfold Unfold.fromList [1..10]
260+-- > Stream.sum $ Stream.foldFilter (Fold.satisfy (> 5)) $ Stream.unfold Unfold.fromList [1..10]
83-- /Pre-release/
263-- /Pre-release/
85{-# INLINE foldFilter #-}
265{-# INLINE foldFilter #-}
86foldFilter :: Monad m => Fold m a (Maybe b) -> Stream m a -> Stream m b
266foldFilter :: Monad m => Fold m a (Maybe b) -> Stream m a -> Stream m b
87−foldFilter p = Stream.fromStreamD . StreamD.foldFilter p . Stream.toStreamD
267+foldFilter p = fromStreamD . D.foldFilter p . toStreamD
269+------------------------------------------------------------------------------
271+------------------------------------------------------------------------------
273+-- | Use a 'Pipe' to transform a stream.
277+{-# INLINE transform #-}
278+transform :: Monad m => Pipe m a b -> Stream m a -> Stream m b
279+transform pipe xs = fromStreamD $ D.transform pipe (toStreamD xs)
281+------------------------------------------------------------------------------
282+-- Transformation Folds
283+------------------------------------------------------------------------------
285+-- | Right fold to a streaming monad.
287+-- > foldrS Stream.cons Stream.nil === id
289+-- 'foldrS' can be used to perform stateless stream to stream transformations
290+-- like map and filter in general. It can be coupled with a scan to perform
291+-- stateful transformations. However, note that the custom map and filter
292+-- routines can be much more efficient than this due to better stream fusion.
294+-- >>> Stream.fold Fold.toList $ Stream.foldrS Stream.cons Stream.nil $ Stream.unfold Unfold.fromList [1..5]
297+-- Find if any element in the stream is 'True':
299+-- >>> Stream.fold Fold.toList $ Stream.foldrS (\x xs -> if odd x then return True else xs) (return False) $ (Stream.unfold Unfold.fromList (2:4:5:undefined) :: Stream IO Int)
302+-- Map (+2) on odd elements and filter out the even elements:
304+-- >>> Stream.fold Fold.toList $ Stream.foldrS (\x xs -> if odd x then (x + 2) `Stream.cons` xs else xs) Stream.nil $ (Stream.unfold Unfold.fromList [1..5] :: Stream IO Int)
307+-- 'foldrM' can also be represented in terms of 'foldrS', however, the former
308+-- iStream much more efficient:
310+-- > foldrM f z s = runIdentityT $ foldrS (\x xs -> lift $ f x (runIdentityT xs)) (lift z) s
313+{-# INLINE foldrS #-}
315+ (a -> Stream m b -> Stream m b)
322+ (\y ys -> toStreamK $ f y (fromStreamK ys))
326+{-# INLINE foldrSShared #-}
328+ (a -> Stream m b -> Stream m b)
332+foldrSShared f z xs =
335+ (\y ys -> toStreamK $ f y (fromStreamK ys))
339+{-# INLINE sequence #-}
340+sequence :: (Monad m) => Stream m (m a) -> Stream m a
344+tap :: Monad m => FL.Fold m a b -> Stream m a -> Stream m a
345+tap f xs = fromStreamD $ D.tap f (toStreamD xs)
347+-- | Apply a monadic function to each element flowing through the stream and
348+-- discard the results.
351+-- >>> Stream.fold Fold.drain $ Stream.trace print (Stream.unfold Unfold.fromList[1, 2])
357+-- Compare with 'tap'.
361+trace :: (Monad m) => (a -> m b) -> Stream m a -> Stream m a
362+trace f = mapM (\x -> void (f x) >> return x)
364+-- | Perform a side effect before yielding each element of the stream and
365+-- discard the results.
368+-- >>> Stream.fold Fold.drain $ Stream.trace_ (print "got here") (Stream.unfold Unfold.fromList[1, 2])
374+-- Same as 'interspersePrefix_' but always serial.
379+{-# INLINE trace_ #-}
380+trace_ :: (Monad m) => m b -> Stream m a -> Stream m a
381+trace_ eff = fromStreamD . D.mapM (\x -> eff >> return x) . toStreamD
383+------------------------------------------------------------------------------
384+-- Scanning with a Fold
385+------------------------------------------------------------------------------
387+-- | Scan a stream using the given monadic fold.
389+-- >>> Stream.fold Fold.toList $ Stream.takeWhile (< 10) $ Stream.scan Fold.sum (Stream.unfold Unfold.fromList [1..10])
394+scan :: (Monad m) => Fold m a b -> Stream m a -> Stream m b
395+scan fld m = fromStreamD $ D.scanOnce fld $ toStreamD m
397+-- | Like 'scan' but restarts scanning afresh when the scanning fold
401+{-# INLINE scanMany #-}
402+scanMany :: (Monad m) => Fold m a b -> Stream m a -> Stream m b
403+scanMany fld m = fromStreamD $ D.scanMany fld $ toStreamD m
405+-- | Postscan a stream using the given monadic fold.
407+-- The following example extracts the input stream up to a point where the
408+-- running average of elements is no more than 10:
410+-- >>> import Data.Maybe (fromJust)
411+-- >>> let avg = Fold.teeWith (/) Fold.sum (fmap fromIntegral Fold.length)
413+-- Stream.fold Fold.toList
414+-- $ Stream.map (fromJust . fst)
415+-- $ Stream.takeWhile (\(_,x) -> x <= 10)
416+-- $ Stream.postscan (Fold.tee Fold.last avg) (Stream.unfold Unfold.fromList $ enumFromTo 1.0 100.0)
418+-- [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]
421+{-# INLINE postscan #-}
422+postscan :: (Monad m) => Fold m a b -> Stream m a -> Stream m b
423+postscan fld = fromStreamD . D.postscanOnce fld . toStreamD
425+-- XXX this needs to be concurrent
426+-- XXX because of the use of D.cons for appending, scanlM' has quadratic
427+-- complexity when iterated over a stream. We should use StreamK style scanlM'
428+-- for linear performance on iteration.
430+-- | Like 'scanl'' but with a monadic step function and a monadic seed.
434+-- /Since: 0.8.0 (signature change)/
435+{-# INLINE scanlM' #-}
436+scanlM' :: (Monad m) => (b -> a -> m b) -> m b -> Stream m a -> Stream m b
437+scanlM' step begin m = fromStreamD $ D.scanlM' step begin $ toStreamD m
439+-- XXX because of the use of D.cons for appending, scanl' has quadratic
440+-- complexity when iterated over a stream. We should use StreamK style scanlM'
441+-- for linear performance on iteration.
443+-- | Strict left scan. Like 'map', 'scanl'' too is a one to one transformation,
444+-- however it adds an extra element.
447+-- >>> Stream.fold Fold.toList $ Stream.scanl' (+) 0 $ Stream.unfold Unfold.fromList [1,2,3,4]
453+-- >>> Stream.fold Fold.toList $ Stream.scanl' (flip (:)) [] $ Stream.unfold Unfold.fromList [1,2,3,4]
454+-- [[],[1],[2,1],[3,2,1],[4,3,2,1]]
458+-- The output of 'scanl'' is the initial value of the accumulator followed by
459+-- all the intermediate steps and the final result of 'foldl''.
461+-- By streaming the accumulated state after each fold step, we can share the
462+-- state across multiple stages of stream composition. Each stage can modify or
463+-- extend the state, do some processing with it and emit it for the next stage,
464+-- thus modularizing the stream processing. This can be useful in
465+-- stateful or event-driven programming.
467+-- Consider the following monolithic example, computing the sum and the product
468+-- of the elements in a stream in one go using a @foldl'@:
471+-- >>> Stream.foldl' (\(s, p) x -> (s + x, p * x)) (0,1) $ Stream.unfold Unfold.fromList [1,2,3,4]
476+-- Using @scanl'@ we can make iStream modular by computing the sum in the first
477+-- stage and passing it down to the next stage for computing the product:
481+-- Stream.foldl' (\(_, p) (s, x) -> (s, p * x)) (0,1)
482+-- $ Stream.scanl' (\(s, _) x -> (s + x, x)) (0,1)
483+-- $ Stream.unfold Unfold.fromList [1,2,3,4]
489+-- IMPORTANT: 'scanl'' evaluates the accumulator to WHNF. To avoid building
490+-- lazy expressions inside the accumulator, it is recommended that a strict
491+-- data structure is used for accumulator.
493+-- >>> scanl' step z = scan (Fold.foldl' step z)
494+-- >>> scanl' f z xs = scanlM' (\a b -> return (f a b)) (return z) xs
495+-- >>> scanl' f z xs = z `Stream.cons` postscanl' f z xs
497+-- See also: 'usingStateT'
500+{-# INLINE scanl' #-}
501+scanl' :: (Monad m) => (b -> a -> b) -> b -> Stream m a -> Stream m b
502+scanl' step z m = fromStreamD $ D.scanl' step z $ toStreamD m
504+-- | Like 'scanl'' but does not stream the initial value of the accumulator.
506+-- >>> postscanl' step z = postscan (Fold.foldl' step z)
507+-- >>> postscanl' f z = postscanlM' (\a b -> return (f a b)) (return z)
508+-- >>> postscanl' f z xs = Stream.drop 1 $ Stream.scanl' f z xs
511+{-# INLINE postscanl' #-}
512+postscanl' :: (Monad m) => (b -> a -> b) -> b -> Stream m a -> Stream m b
513+postscanl' step z m = fromStreamD $ D.postscanl' step z $ toStreamD m
515+-- XXX prescanl does not sound very useful, enable only if there is a
516+-- compelling use case.
518+-- | Like scanl' but does not stream the final value of the accumulator.
521+{-# INLINE prescanl' #-}
522+prescanl' :: (Monad m) => (b -> a -> b) -> b -> Stream m a -> Stream m b
523+prescanl' step z m = fromStreamD $ D.prescanl' step z $ toStreamD m
525+-- XXX this needs to be concurrent
526+-- | Like prescanl' but with a monadic step function and a monadic seed.
529+{-# INLINE prescanlM' #-}
530+prescanlM' :: (Monad m) => (b -> a -> m b) -> m b -> Stream m a -> Stream m b
531+prescanlM' step z m = fromStreamD $ D.prescanlM' step z $ toStreamD m
533+-- XXX this needs to be concurrent
534+-- | Like 'scanl1'' but with a monadic step function.
537+{-# INLINE scanl1M' #-}
538+scanl1M' :: (Monad m) => (a -> a -> m a) -> Stream m a -> Stream m a
539+scanl1M' step m = fromStreamD $ D.scanl1M' step $ toStreamD m
541+-- | Like 'scanl'' but for a non-empty stream. The first element of the stream
542+-- is used as the initial value of the accumulator. Does nothing if the stream
546+-- >>> Stream.fold Fold.toList $ Stream.scanl1' (+) $ Stream.unfold Unfold.fromList [1,2,3,4]
552+{-# INLINE scanl1' #-}
553+scanl1' :: (Monad m) => (a -> a -> a) -> Stream m a -> Stream m a
554+scanl1' step m = fromStreamD $ D.scanl1' step $ toStreamD m
556+-- | Drop repeated elements that are adjacent to each other using the supplied
557+-- comparison function.
559+-- @uniq = uniqBy (==)
561+-- To strip duplicate path separators:
564+-- f x y = x == '/' && x == y
565+-- Stream.fold Fold.toList $ Stream.uniqBy f $ Stream.unfold Unfold.fromList "//a//b"
571+-- See also: 'nubBy'.
575+{-# INLINE uniqBy #-}
576+uniqBy :: (Monad m) =>
577+ (a -> a -> Bool) -> Stream m a -> Stream m a
578+uniqBy eq = catMaybes . rollingMap f
584+ Nothing -> Just curr
585+ Just x -> if x `eq` curr then Nothing else Just curr
587+-- | Drop repeated elements that are adjacent to each other.
591+uniq :: (Eq a, Monad m) => Stream m a -> Stream m a
592+uniq = fromStreamD . D.uniq . toStreamD
594+-- | Deletes the first occurrence of the element in the stream that satisfies
595+-- the given equality predicate.
598+-- >>> Stream.fold Fold.toList $ Stream.deleteBy (==) 3 $ Stream.unfold Unfold.fromList [1,3,3,5]
604+{-# INLINE deleteBy #-}
605+deleteBy :: (Monad m) => (a -> a -> Bool) -> a -> Stream m a -> Stream m a
606+deleteBy cmp x m = fromStreamD $ D.deleteBy cmp x (toStreamD m)
608+------------------------------------------------------------------------------
610+------------------------------------------------------------------------------
612+-- | Same as 'takeWhile' but with a monadic predicate.
615+{-# INLINE takeWhileM #-}
616+takeWhileM :: (Monad m) => (a -> m Bool) -> Stream m a -> Stream m a
617+takeWhileM p m = fromStreamD $ D.takeWhileM p $ toStreamD m
619+-- | @takeInterval duration@ yields stream elements upto specified time
620+-- @duration@. The duration starts when the stream is evaluated for the first
621+-- time, before the first element is yielded. The time duration is checked
622+-- before generating each element, if the duration has expired the stream
625+-- The total time taken in executing the stream is guaranteed to be /at least/
626+-- @duration@, however, because the duration is checked before generating an
627+-- element, the upper bound is indeterminate and depends on the time taken in
628+-- generating and processing the last element.
630+-- No element is yielded if the duration is zero. At least one element is
631+-- yielded if the duration is non-zero.
635+{-# INLINE takeInterval #-}
636+takeInterval ::(MonadIO m, TimeUnit64 d) => d -> Stream m a -> Stream m a
637+takeInterval d = fromStreamD . D.takeByTime d . toStreamD
639+-- | Drop elements in the stream as long as the predicate succeeds and then
640+-- take the rest of the stream.
643+{-# INLINE dropWhile #-}
644+dropWhile :: (Monad m) => (a -> Bool) -> Stream m a -> Stream m a
645+dropWhile p m = fromStreamD $ D.dropWhile p $ toStreamD m
647+-- | Same as 'dropWhile' but with a monadic predicate.
650+{-# INLINE dropWhileM #-}
651+dropWhileM :: (Monad m) => (a -> m Bool) -> Stream m a -> Stream m a
652+dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m
654+-- | @dropInterval duration@ drops stream elements until specified @duration@ has
655+-- passed. The duration begins when the stream is evaluated for the first
656+-- time. The time duration is checked /after/ generating a stream element, the
657+-- element is yielded if the duration has expired otherwise it is dropped.
659+-- The time elapsed before starting to generate the first element is /aStream most/
660+-- @duration@, however, because the duration expiry is checked after the
661+-- element is generated, the lower bound is indeterminate and depends on the
662+-- time taken in generating an element.
664+-- All elements are yielded if the duration is zero.
668+{-# INLINE dropInterval #-}
669+dropInterval ::(MonadIO m, TimeUnit64 d) => d -> Stream m a -> Stream m a
670+dropInterval d = fromStreamD . D.dropByTime d . toStreamD
672+------------------------------------------------------------------------------
673+-- Inserting Elements
674+------------------------------------------------------------------------------
676+-- | @insertBy cmp elem stream@ inserts @elem@ before the first element in
677+-- @stream@ that is less than @elem@ when compared using @cmp@.
680+-- insertBy cmp x = 'mergeBy' cmp ('fromPure' x)
684+-- >>> Stream.fold Fold.toList $ Stream.insertBy compare 2 $ Stream.unfold Unfold.fromList [1,3,5]
690+{-# INLINE insertBy #-}
691+insertBy ::(Monad m) => (a -> a -> Ordering) -> a -> Stream m a -> Stream m a
692+insertBy cmp x m = fromStreamD $ D.insertBy cmp x (toStreamD m)
694+-- | Insert a pure value between successive elements of a stream.
696+-- >>> Stream.fold Fold.toList $ Stream.intersperse ',' $ Stream.unfold Unfold.fromList "hello"
700+{-# INLINE intersperse #-}
701+intersperse :: (Monad m) => a -> Stream m a -> Stream m a
702+intersperse a = fromStreamD . D.intersperse a . toStreamD
704+-- | Insert a side effect before consuming an element of a stream except the
707+-- >>> Stream.fold Fold.drain $ Stream.trace putChar $ Stream.intersperseM_ (putChar '.') $ Stream.unfold Unfold.fromList "hello"
711+{-# INLINE intersperseM_ #-}
712+intersperseM_ :: (Monad m) => m b -> Stream m a -> Stream m a
713+intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
715+-- | Insert an effect and its output after consuming an element of a stream.
717+-- >>> Stream.fold Fold.toList $ Stream.trace putChar $ intersperseSuffix (putChar '.' >> return ',') $ Stream.unfold Unfold.fromList "hello"
718+-- h.,e.,l.,l.,o.,"h,e,l,l,o,"
721+{-# INLINE intersperseSuffix #-}
722+intersperseSuffix :: (Monad m) => m a -> Stream m a -> Stream m a
723+intersperseSuffix m = fromStreamD . D.intersperseSuffix m . toStreamD
725+-- | Insert a side effect after consuming an element of a stream.
728+-- >>> Stream.fold Fold.toList $ Stream.intersperseSuffix_ (threadDelay 1000000) $ Stream.unfold Unfold.fromList "hello"
735+{-# INLINE intersperseSuffix_ #-}
736+intersperseSuffix_ :: (Monad m) => m b -> Stream m a -> Stream m a
737+intersperseSuffix_ m = fromStreamD . D.intersperseSuffix_ m . toStreamD
739+-- XXX Use an offset argument, like tapOffsetEvery
741+-- | Like 'intersperseSuffix' but intersperses an effectful action into the
742+-- input stream after every @n@ elements and after the last element.
744+-- >>> Stream.fold Fold.toList $ Stream.intersperseSuffixBySpan 2 (return ',') $ Stream.unfold Unfold.fromList "hello"
749+{-# INLINE intersperseSuffixBySpan #-}
750+intersperseSuffixBySpan :: (Monad m)
751+ => Int -> m a -> Stream m a -> Stream m a
752+intersperseSuffixBySpan n eff =
753+ fromStreamD . D.intersperseSuffixBySpan n eff . toStreamD
755+-- | Insert a side effect before consuming an element of a stream.
757+-- >>> Stream.fold Fold.toList $ Stream.trace putChar $ Stream.interspersePrefix_ (putChar '.' >> return ',') $ Stream.unfold Unfold.fromList "hello"
760+-- Same as 'trace_' buStream may be concurrent.
766+{-# INLINE interspersePrefix_ #-}
767+interspersePrefix_ :: (Monad m) => m b -> Stream m a -> Stream m a
768+interspersePrefix_ m = mapM (\x -> void m >> return x)
770+------------------------------------------------------------------------------
772+------------------------------------------------------------------------------
774+-- Note: delay must be serial.
776+-- | Introduce a delay of specified seconds before consuming an element of the
777+-- stream except the first one.
779+-- >>> Stream.mapM_ print $ Stream.timestamped $ Stream.delay 1 $ Stream.unfold Unfold.fromList $ enumFromTo 1 3
780+-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),1)
781+-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),2)
782+-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),3)
787+delay :: (MonadIO m) => Double -> Stream m a -> Stream m a
788+delay n = intersperseM_ $ liftIO $ threadDelay $ round $ n * 1000000
790+-- Note: delay must be serial.
792+-- | Introduce a delay of specified seconds after consuming an element of a
795+-- >>> Stream.mapM_ print $ Stream.timestamped $ Stream.delayPost 1 $ Stream.unfold Unfold.fromList $ enumFromTo 1 3
796+-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),1)
797+-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),2)
798+-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),3)
802+{-# INLINE delayPost #-}
803+delayPost :: (MonadIO m) => Double -> Stream m a -> Stream m a
804+delayPost n = intersperseSuffix_ $ liftIO $ threadDelay $ round $ n * 1000000
806+-- Note: delay must be serial, that's why 'trace_' is used.
808+-- | Introduce a delay of specified seconds before consuming an element of a
811+-- >>> Stream.mapM_ print $ Stream.timestamped $ Stream.delayPre 1 $ Stream.unfold Unfold.fromList $ enumFromTo 1 3
812+-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),1)
813+-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),2)
814+-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),3)
818+{-# INLINE delayPre #-}
819+delayPre :: (MonadIO m) => Double -> Stream m a -> Stream m a
820+delayPre n = trace_ $ liftIO $ threadDelay $ round $ n * 1000000
822+------------------------------------------------------------------------------
824+------------------------------------------------------------------------------
827+-- > indexed = Stream.postscanl' (\(i, _) x -> (i + 1, x)) (-1,undefined)
828+-- > indexed = Stream.zipWith (,) (Stream.enumerateFrom 0)
830+-- Pair each element in a stream with its index, starting from index 0.
832+-- >>> Stream.fold Fold.toList $ Stream.indexed $ Stream.unfold Unfold.fromList "hello"
833+-- [(0,'h'),(1,'e'),(2,'l'),(3,'l'),(4,'o')]
836+{-# INLINE indexed #-}
837+indexed :: (Monad m) => Stream m a -> Stream m (Int, a)
838+indexed = fromStreamD . D.indexed . toStreamD
841+-- > indexedR n = Stream.postscanl' (\(i, _) x -> (i - 1, x)) (n + 1,undefined)
842+-- > indexedR n = Stream.zipWith (,) (Stream.unfold Unfold.fromList $ enumFromThen n (n - 1))
844+-- Pair each element in a stream with its index, starting from the
845+-- given index @n@ and counting down.
847+-- >>> Stream.fold Fold.toList $ Stream.indexedR 10 $ Stream.unfold Unfold.fromList "hello"
848+-- [(10,'h'),(9,'e'),(8,'l'),(7,'l'),(6,'o')]
851+{-# INLINE indexedR #-}
852+indexedR :: (Monad m) => Int -> Stream m a -> Stream m (Int, a)
853+indexedR n = fromStreamD . D.indexedR n . toStreamD
855+------------------------------------------------------------------------------
857+------------------------------------------------------------------------------
859+-- | Find all the indices where the value of the element in the stream is equal
860+-- to the given value.
862+-- > elemIndices a = findIndices (== a)
865+{-# INLINE elemIndices #-}
866+elemIndices :: (Eq a, Monad m) => a -> Stream m a -> Stream m Int
867+elemIndices a = findIndices (== a)
869+------------------------------------------------------------------------------
871+------------------------------------------------------------------------------
873+-- XXX this is not a one-to-one map so calling iStream map may not be right.
874+-- We can perhaps call it zipWithTail or rollWith.
876+-- | Apply a function on every two successive elements of a stream. The first
877+-- argument of the map function is the previous element and the second argument
878+-- is the current element. When the current element is the first element, the
879+-- previous element is 'Nothing'.
883+{-# INLINE rollingMap #-}
884+rollingMap :: (Monad m) => (Maybe a -> a -> b) -> Stream m a -> Stream m b
885+rollingMap f m = fromStreamD $ D.rollingMap f $ toStreamD m
887+-- | Like 'rollingMap' but with an effectful map function.
891+{-# INLINE rollingMapM #-}
892+rollingMapM :: (Monad m) => (Maybe a -> a -> m b) -> Stream m a -> Stream m b
893+rollingMapM f m = fromStreamD $ D.rollingMapM f $ toStreamD m
895+-- | Like 'rollingMap' but requires at least two elements in the stream,
896+-- returns an empty stream otherwise.
898+-- This is the stream equivalent of the list idiom @zipWith f xs (tail xs)@.
902+{-# INLINE rollingMap2 #-}
903+rollingMap2 :: (Monad m) => (a -> a -> b) -> Stream m a -> Stream m b
904+rollingMap2 f m = fromStreamD $ D.rollingMap2 f $ toStreamD m
906+------------------------------------------------------------------------------
908+------------------------------------------------------------------------------
910+-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'
911+-- elements, and return a stream of values extracted from 'Just'.
916+-- mapMaybe f = Stream.map 'fromJust' . Stream.filter 'isJust' . Stream.map f
920+{-# INLINE mapMaybe #-}
921+mapMaybe :: (Monad m) => (a -> Maybe b) -> Stream m a -> Stream m b
922+mapMaybe f m = fromStreamD $ D.mapMaybe f $ toStreamD m
924+-- | Like 'mapMaybe' buStream maps a monadic function.
929+-- mapMaybeM f = Stream.map 'fromJust' . Stream.filter 'isJust' . Stream.mapM f
932+-- /Concurrent (do not use with 'fromParallel' on infinite streams)/
935+{-# INLINE_EARLY mapMaybeM #-}
936+mapMaybeM :: (Monad m)
937+ => (a -> m (Maybe b)) -> Stream m a -> Stream m b
938+mapMaybeM f = fmap fromJust . filter isJust . mapM f
940+-- | In a stream of 'Maybe's, discard 'Nothing's and unwrap 'Just's.
944+{-# INLINE catMaybes #-}
945+catMaybes :: (Monad m) => Stream m (Maybe a) -> Stream m a
946+catMaybes = fmap fromJust . filter isJust
948+------------------------------------------------------------------------------
950+------------------------------------------------------------------------------
952+-- | Discard 'Right's and unwrap 'Left's in an 'Either' stream.
957+lefts :: (Monad m) => Stream m (Either a b) -> Stream m a
958+lefts = fmap (fromLeft undefined) . filter isLeft
960+-- | Discard 'Left's and unwrap 'Right's in an 'Either' stream.
964+{-# INLINE rights #-}
965+rights :: (Monad m) => Stream m (Either a b) -> Stream m b
966+rights = fmap (fromRight undefined) . filter isRight
968+-- | Remove the either wrapper and flatten both lefts and as well as rights in
969+-- the output stream.
974+both :: Monad m => Stream m (Either a a) -> Stream m a
975+both = fmap (either id id)