@@ -5,37 +5,249 @@
55-- Maintainer : streamly@composewell.com
66-- Stability : experimental
77-- Portability : GHC
8−--
8+
99module Streamly.Internal.Data.Stream.Transform
1010 (
11− filter
12− , foldFilter
13− , map
11+ -- * Piping
12+ -- | Pass through a 'Pipe'.
13+ transform
14+
15+ -- * Folding
16+ , foldrS
17+ , foldrT
18+
19+ -- * Mapping
20+ -- | Stateless one-to-one maps.
21+ , sequence
1422 , mapM
23+
24+ -- * Mapping Side Effects (Observation)
25+ -- | See also the intersperse*_ combinators.
26+ , trace
27+ , trace_
28+ , tap
29+
30+ -- * Scanning
31+ , scan
32+ , scanMany
33+ , postscan
34+ , smapM
35+
36+ -- * Filtering
37+ -- | Produce a subset of the stream using criteria based on the values of
38+ -- the elements. We can use a concatMap and scan for filtering but these
39+ -- combinators are more efficient and convenient.
40+
41+ , with
42+ , deleteBy
43+ , filter
44+ , filterM
45+ , foldFilter
46+ , uniq
47+ , uniqBy
48+ , nubBy
49+ , prune
50+ , repeated
51+
52+ -- * Trimming
53+ -- | Produce a subset of the stream trimmed at ends.
54+
55+ , take
56+ , takeWhile
57+ , takeWhileM
58+ , drop
59+ , dropWhile
60+ , dropWhileM
61+
62+ -- * Inserting Elements
63+ -- | Produce a superset of the stream. This is the opposite of
64+ -- filtering/sampling. We can always use concatMap and scan for inserting
65+ -- but these combinators are more efficient and convenient.
66+
67+ -- Element agnostic (Opposite of sampling)
68+ , intersperse
69+ , intersperseM -- XXX naming
70+ , intersperseBySpan
71+
72+ , intersperseSuffix
73+ , intersperseSuffixBySpan
74+
75+ -- , interspersePrefix
76+ -- , interspersePrefixBySpan
77+
78+ -- * Inserting Side Effects/Time
79+ , intersperseM_ -- XXX naming
80+ , delay
81+ , intersperseSuffix_
82+ , delayPost
83+ , interspersePrefix_
84+ , delayPre
85+
86+ -- * Element Aware Insertion
87+ -- | Opposite of filtering
88+ , insertBy
89+ -- , intersperseByBefore
90+ -- , intersperseByAfter
91+
92+ -- * Reordering
93+ , reverse
94+ , reverse'
95+ , reassembleBy
96+
97+ -- * Position Indexing
98+ , indexed
99+ , indexedR
100+
101+ -- * Searching
102+ , findIndices -- XXX indicesBy
103+ , elemIndices -- XXX indicesOf
104+
105+ -- * Rolling map
106+ -- | Map using the previous element.
107+ , rollingMapM
108+ , rollingMap
109+ , rollingMap2
110+
111+ -- * Maybe Streams
112+ -- Move these to Streamly.Data.Maybe.Stream?
113+ , catMaybes -- XXX justs (like lefts/rights)
114+ , mapMaybe
115+ , mapMaybeM
116+
117+ -- * Either Streams
118+ -- Move these to Streamly.Data.Either.Stream?
119+ , lefts
120+ , rights
121+ , both
15122 )
16123where
17124
125+#include "inline.hs"
126+
127+import Control.Concurrent (threadDelay)
128+import Control.Monad (void)
129+import Control.Monad.IO.Class (MonadIO (liftIO))
130+import Control.Monad.Trans.Class (MonadTrans(..))
131+import Data.Either (fromLeft, isLeft, isRight, fromRight)
132+import Data.Maybe (isJust, fromJust)
133+
18134import Streamly.Internal.Data.Fold.Type (Fold)
19−import Streamly.Internal.Data.Stream.Type (Stream)
135+import Streamly.Internal.Data.Pipe (Pipe)
136+
137+import qualified Streamly.Internal.Data.Fold as FL
138+import qualified Streamly.Internal.Data.Stream.StreamD.Transform as D
139+import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
140+import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
20141
21−import qualified Streamly.Internal.Data.Stream.Type as Stream
22−import qualified Streamly.Internal.Data.Stream.StreamD.Transform as StreamD
142+import Streamly.Internal.Data.Stream.Bottom
143+import Streamly.Internal.Data.Stream.Type
23144
24−import Prelude hiding (map, mapM, filter)
145+import Prelude hiding
146+ ( filter, drop, dropWhile, take, takeWhile, foldr, map, mapM, sequence
147+ , reverse, foldr1 , repeat, scanl, scanl1, zipWith)
25148
26149--
27150-- $setup
28151-- >>> :m
152+-- >>> import Control.Concurrent (threadDelay)
153+-- >>> import Control.Monad.Trans (lift)
154+-- >>> import Control.Monad.Trans.Identity (runIdentityT)
155+-- >>> import Data.Function ((&))
156+-- >>> import Data.Maybe (fromJust, isJust)
157+-- >>> import Prelude hiding (filter, drop, dropWhile, take, takeWhile, foldr, map, mapM, sequence, reverse, foldr1 , scanl, scanl1)
158+-- >>> import Streamly.Internal.Data.Stream as Stream
159+-- >>> import qualified Streamly.Data.Fold as Fold
29160-- >>> import qualified Streamly.Internal.Data.Fold as Fold
30−-- >>> import qualified Streamly.Data.Unfold as Unfold
31−-- >>> import qualified Streamly.Internal.Data.Stream as Stream
161+-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
162+-- >>> import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
163+--
164+-- >>> hSetBuffering stdout LineBuffering
165+
166+-- XXX because of the use of D.cons for appending, folds and scans have
167+-- quadratic complexity when iterated over a stream. We should use StreamK for
168+-- linear performance on iteration.
169+
170+------------------------------------------------------------------------------
171+-- Piping
172+------------------------------------------------------------------------------
173+
174+-- | Use a 'Pipe' to transform a stream.
175+--
176+-- /Pre-release/
177+--
178+{-# INLINE transform #-}
179+transform :: Monad m => Pipe m a b -> Stream m a -> Stream m b
180+transform pipe xs = fromStreamD $ D.transform pipe (toStreamD xs)
181+
182+------------------------------------------------------------------------------
183+-- Transformation Folds
184+------------------------------------------------------------------------------
185+
186+-- | Right fold to a streaming monad.
187+--
188+-- > foldrS Stream.cons Stream.nil === id
189+--
190+-- 'foldrS' can be used to perform stateless stream to stream transformations
191+-- like map and filter in general. It can be coupled with a scan to perform
192+-- stateful transformations. However, note that the custom map and filter
193+-- routines can be much more efficient than this due to better stream fusion.
194+--
195+-- >>> input = Stream.unfold Unfold.fromList [1..5]
196+-- >>> Stream.fold Fold.toList $ Stream.foldrS Stream.cons Stream.nil input
197+-- [1,2,3,4,5]
198+--
199+-- Find if any element in the stream is 'True':
200+--
201+-- >>> step x xs = if odd x then return True else xs
202+-- >>> input = Stream.unfold Unfold.fromList (2:4:5:undefined) :: Stream IO Int
203+-- >>> Stream.fold Fold.toList $ Stream.foldrS step (return False) input
204+-- [True]
205+--
206+-- Map (+2) on odd elements and filter out the even elements:
207+--
208+-- >>> step x xs = if odd x then (x + 2) `Stream.cons` xs else xs
209+-- >>> input = Stream.unfold Unfold.fromList [1..5] :: Stream IO Int
210+-- >>> Stream.fold Fold.toList $ Stream.foldrS step Stream.nil input
211+-- [3,5,7]
212+--
213+-- /Pre-release/
214+{-# INLINE foldrS #-}
215+foldrS ::
216+ (a -> Stream m b -> Stream m b)
217+ -> Stream m b
218+ -> Stream m a
219+ -> Stream m b
220+foldrS f z xs =
221+ fromStreamK
222+ $ K.foldrS
223+ (\y ys -> toStreamK $ f y (fromStreamK ys))
224+ (toStreamK z)
225+ (toStreamK xs)
226+
227+-- | Right fold to a transformer monad. This is the most general right fold
228+-- function. 'foldrS' is a special case of 'foldrT', however 'foldrS'
229+-- implementation can be more efficient:
230+--
231+-- >>> foldrS = foldrT
232+--
233+-- >>> step f x xs = lift $ f x (runIdentityT xs)
234+-- >>> foldrM f z s = runIdentityT $ foldrT (step f) (lift z) s
235+--
236+-- 'foldrT' can be used to translate streamly streams to other transformer
237+-- monads e.g. to a different streaming type.
238+--
239+-- /Pre-release/
240+{-# INLINE foldrT #-}
241+foldrT :: (Monad m, Monad (s m), MonadTrans s)
242+ => (a -> s m b -> s m b) -> s m b -> Stream m a -> s m b
243+foldrT f z s = D.foldrT f z (toStreamD s)
32244
33245------------------------------------------------------------------------------
34246-- Transformation by Mapping
35247------------------------------------------------------------------------------
36248
37249-- |
38−-- >> mapM f = Stream.sequence . Stream.map f
250+-- >>> mapM f = Stream.sequence . fmap f
39251--
40252-- Apply a monadic function to each element of the stream and replace it with
41253-- the output of the resulting action.
@@ -48,40 +260,631 @@ import Prelude hiding (map, mapM, filter)
48260--
49261{-# INLINE mapM #-}
50262mapM :: Monad m => (a -> m b) -> Stream m a -> Stream m b
51−mapM f = Stream.fromStreamD . StreamD.mapM f . Stream.toStreamD
263+mapM f m = fromStreamK $ D.toStreamK $ D.mapM f $ toStreamD m
52264
53265-- |
54−-- >>> map = fmap
266+-- sequence = Stream.mapM id
267+--
268+-- Replace the elements of a stream of monadic actions with the outputs of
269+-- those actions.
270+--
271+-- >>> s = Stream.unfold Unfold.fromList [putStr "a", putStr "b", putStrLn "c"]
272+-- >>> Stream.fold Fold.drain $ Stream.sequence s
273+-- abc
274+--
275+{-# INLINE sequence #-}
276+sequence :: Monad m => Stream m (m a) -> Stream m a
277+sequence = mapM id
278+
279+------------------------------------------------------------------------------
280+-- Mapping side effects
281+------------------------------------------------------------------------------
282+
283+-- | Tap the data flowing through a stream into a 'Fold'. For example, you may
284+-- add a tap to log the contents flowing through the stream. The fold is used
285+-- only for effects, its result is discarded.
286+--
287+-- @
288+-- Fold m a b
289+-- |
290+-- -----stream m a ---------------stream m a-----
291+--
292+-- @
293+--
294+-- >>> s = Stream.unfold Unfold.enumerateFromTo (1, 2)
295+-- >>> Stream.fold Fold.drain $ Stream.tap (Fold.drainBy print) s
296+-- 1
297+-- 2
298+--
299+-- Compare with 'trace'.
300+--
301+{-# INLINE tap #-}
302+tap :: Monad m => FL.Fold m a b -> Stream m a -> Stream m a
303+tap f xs = fromStreamD $ D.tap f (toStreamD xs)
304+
305+-- | Apply a monadic function to each element flowing through the stream and
306+-- discard the results.
55307--
56−-- Same as 'fmap'.
308+-- >>> s = Stream.unfold Unfold.enumerateFromTo (1, 2)
309+-- >>> Stream.fold Fold.drain $ Stream.trace print s
310+-- 1
311+-- 2
57312--
58−-- >>> s = Stream.unfold Unfold.fromList [1,2,3]
59−-- >>> Stream.fold Fold.toList $ Stream.map (+1) s
60−-- [2,3,4]
313+-- Compare with 'tap'.
314+--
315+{-# INLINE trace #-}
316+trace :: Monad m => (a -> m b) -> Stream m a -> Stream m a
317+trace f = mapM (\x -> void (f x) >> return x)
318+
319+-- | Perform a side effect before yielding each element of the stream and
320+-- discard the results.
321+--
322+-- >>> s = Stream.unfold Unfold.enumerateFromTo (1, 2)
323+-- >>> Stream.fold Fold.drain $ Stream.trace_ (print "got here") s
324+-- "got here"
325+-- "got here"
326+--
327+-- Same as 'interspersePrefix_' but always serial.
328+--
329+-- See also: 'trace'
330+--
331+-- /Pre-release/
332+{-# INLINE trace_ #-}
333+trace_ :: Monad m => m b -> Stream m a -> Stream m a
334+trace_ eff = fromStreamD . D.mapM (\x -> eff >> return x) . toStreamD
335+
336+------------------------------------------------------------------------------
337+-- Scanning with a Fold
338+------------------------------------------------------------------------------
339+
340+-- XXX It may be useful to have a version of scan where we can keep the
341+-- accumulator independent of the value emitted. So that we do not necessarily
342+-- have to keep a value in the accumulator which we are not using. We can pass
343+-- an extraction function that will take the accumulator and the current value
344+-- of the element and emit the next value in the stream. That will also make it
345+-- possible to modify the accumulator after using it. In fact, the step function
346+-- can return new accumulator and the value to be emitted. The signature would
347+-- be more like mapAccumL.
348+
349+-- | Strict left scan. Scan a stream using the given monadic fold.
350+--
351+-- >>> s = Stream.unfold Unfold.fromList [1..10]
352+-- >>> Stream.fold Fold.toList $ Stream.takeWhile (< 10) $ Stream.scan Fold.sum s
353+-- [0,1,3,6]
354+--
355+-- >>> scanl' step z = Stream.scan (Fold.foldl' step z)
356+--
357+-- Like 'map', 'scanl'' too is a one to one transformation,
358+-- however it adds an extra element.
359+--
360+-- >>> s = Stream.unfold Unfold.fromList [1,2,3,4]
361+-- >>> Stream.fold Fold.toList $ scanl' (+) 0 s
362+-- [0,1,3,6,10]
363+--
364+-- >>> Stream.fold Fold.toList $ scanl' (flip (:)) [] s
365+-- [[],[1],[2,1],[3,2,1],[4,3,2,1]]
366+--
367+-- The output of 'scanl'' is the initial value of the accumulator followed by
368+-- all the intermediate steps and the final result of 'foldl''.
369+--
370+-- By streaming the accumulated state after each fold step, we can share the
371+-- state across multiple stages of stream composition. Each stage can modify or
372+-- extend the state, do some processing with it and emit it for the next stage,
373+-- thus modularizing the stream processing. This can be useful in
374+-- stateful or event-driven programming.
375+--
376+-- Consider the following monolithic example, computing the sum and the product
377+-- of the elements in a stream in one go using a @foldl'@:
378+--
379+-- >>> foldl' step z = Stream.fold (Fold.foldl' step z)
380+-- >>> foldl' (\(s, p) x -> (s + x, p * x)) (0,1) s
381+-- (10,24)
382+--
383+-- Using @scanl'@ we can make it modular by computing the sum in the first
384+-- stage and passing it down to the next stage for computing the product:
385+--
386+-- >>> :{
387+-- foldl' (\(_, p) (s, x) -> (s, p * x)) (0,1)
388+-- $ scanl' (\(s, _) x -> (s + x, x)) (0,1)
389+-- $ Stream.unfold Unfold.fromList [1,2,3,4]
390+-- :}
391+-- (10,24)
392+--
393+-- IMPORTANT: 'scanl'' evaluates the accumulator to WHNF. To avoid building
394+-- lazy expressions inside the accumulator, it is recommended that a strict
395+-- data structure is used for accumulator.
396+--
397+-- See also: 'usingStateT'
398+--
399+{-# INLINE scan #-}
400+scan :: Monad m => Fold m a b -> Stream m a -> Stream m b
401+scan fld m = fromStreamD $ D.scanOnce fld $ toStreamD m
402+
403+-- | Like 'scan' but restarts scanning afresh when the scanning fold
404+-- terminates.
61405--
62406-- /Pre-release/
407+{-# INLINE scanMany #-}
408+scanMany :: Monad m => Fold m a b -> Stream m a -> Stream m b
409+scanMany fld m = fromStreamD $ D.scanMany fld $ toStreamD m
410+
411+------------------------------------------------------------------------------
412+-- Filtering
413+------------------------------------------------------------------------------
414+
415+-- | Modify a @Stream m a -> Stream m a@ stream transformation that accepts a
416+-- predicate @(a -> b)@ to accept @((s, a) -> b)@ instead, provided a
417+-- transformation @Stream m a -> Stream m (s, a)@. Convenient to filter with
418+-- index or time.
63419--
64−{-# INLINE map #-}
65−map :: Monad m => (a -> b) -> Stream m a -> Stream m b
66−map f = mapM (return . f)
420+-- >>> filterWithIndex = Stream.with Stream.indexed Stream.filter
421+--
422+-- /Pre-release/
423+{-# INLINE with #-}
424+with :: Monad m =>
425+ (Stream m a -> Stream m (s, a))
426+ -> (((s, a) -> b) -> Stream m (s, a) -> Stream m (s, a))
427+ -> (((s, a) -> b) -> Stream m a -> Stream m a)
428+with f comb g = fmap snd . comb g . f
67429
68430-- | Include only those elements that pass a predicate.
69431--
70432{-# INLINE filter #-}
71433filter :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
72−filter p = Stream.fromStreamD . StreamD.filter p . Stream.toStreamD
434+filter p m = fromStreamD $ D.filter p $ toStreamD m
435+
436+-- | Same as 'filter' but with a monadic predicate.
437+--
438+{-# INLINE filterM #-}
439+filterM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
440+filterM p m = fromStreamD $ D.filterM p $ toStreamD m
73441
74442-- | Use a filtering fold on a stream.
75443--
76−-- >>> :{
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]
80−-- :}
81−--40
444+-- >>> input = Stream.unfold Unfold.fromList [1..10]
445+-- >>> Stream.fold Fold.sum $ Stream.foldFilter (Fold.satisfy (> 5)) input
446+-- 40
82447--
83448-- /Pre-release/
84449--
85450{-# INLINE foldFilter #-}
86451foldFilter :: Monad m => Fold m a (Maybe b) -> Stream m a -> Stream m b
87−foldFilter p = Stream.fromStreamD . StreamD.foldFilter p . Stream.toStreamD
452+foldFilter p = fromStreamD . D.foldFilter p . toStreamD
453+
454+-- | Drop repeated elements that are adjacent to each other using the supplied
455+-- comparison function.
456+--
457+-- >>> uniq = Stream.uniqBy (==)
458+--
459+-- To strip duplicate path separators:
460+--
461+-- >>> input = Stream.unfold Unfold.fromList "//a//b"
462+-- >>> f x y = x == '/' && y == '/'
463+-- >>> Stream.fold Fold.toList $ Stream.uniqBy f input
464+-- "/a/b"
465+--
466+-- Space: @O(1)@
467+--
468+-- See also: 'nubBy'.
469+--
470+-- /Pre-release/
471+--
472+{-# INLINE uniqBy #-}
473+uniqBy :: Monad m =>
474+ (a -> a -> Bool) -> Stream m a -> Stream m a
475+uniqBy eq = catMaybes . rollingMap f
476+
477+ where
478+
479+ f pre curr =
480+ case pre of
481+ Nothing -> Just curr
482+ Just x -> if x `eq` curr then Nothing else Just curr
483+
484+-- | Drop repeated elements that are adjacent to each other.
485+--
486+{-# INLINE uniq #-}
487+uniq :: (Eq a, Monad m) => Stream m a -> Stream m a
488+uniq = fromStreamD . D.uniq . toStreamD
489+
490+-- | Strip all leading and trailing occurrences of an element passing a
491+-- predicate and make all other consecutive occurrences uniq.
492+--
493+-- >> prune p = Stream.dropWhileAround p $ Stream.uniqBy (x y -> p x && p y)
494+--
495+-- @
496+-- > Stream.prune isSpace (Stream.fromList " hello world! ")
497+-- "hello world!"
498+--
499+-- @
500+--
501+-- Space: @O(1)@
502+--
503+-- /Unimplemented/
504+{-# INLINE prune #-}
505+prune ::
506+ -- (Monad m, Eq a) =>
507+ (a -> Bool) -> Stream m a -> Stream m a
508+prune = error "Not implemented yet!"
509+
510+-- Possible implementation:
511+-- @repeated =
512+-- Stream.catMaybes . Stream.parseMany (Parser.groupBy (==) Fold.repeated)@
513+--
514+-- 'Fold.repeated' should return 'Just' when repeated, and 'Nothing' for a
515+-- single element.
516+
517+-- | Emit only repeated elements, once.
518+--
519+-- /Unimplemented/
520+repeated :: -- (Monad m, Eq a) =>
521+ Stream m a -> Stream m a
522+repeated = undefined
523+
524+-- We can have more efficient implementations for nubOrd and nubInt by using
525+-- Set and IntSet to find/remove duplication. For Hashable we can use a
526+-- hashmap. Use rewrite rules to specialize to more efficient impls.
527+
528+-- | Drop repeated elements anywhere in the stream.
529+--
530+-- /Caution: not scalable for infinite streams/
531+--
532+-- /Unimplemented/
533+--
534+{-# INLINE nubBy #-}
535+nubBy :: -- Monad m =>
536+ (a -> a -> Bool) -> Stream m a -> Stream m a
537+nubBy = undefined -- fromStreamD . D.nubBy . toStreamD
538+
539+-- | Deletes the first occurrence of the element in the stream that satisfies
540+-- the given equality predicate.
541+--
542+-- >>> input = Stream.unfold Unfold.fromList [1,3,3,5]
543+-- >>> Stream.fold Fold.toList $ Stream.deleteBy (==) 3 input
544+-- [1,3,5]
545+--
546+{-# INLINE deleteBy #-}
547+deleteBy :: Monad m => (a -> a -> Bool) -> a -> Stream m a -> Stream m a
548+deleteBy cmp x m = fromStreamD $ D.deleteBy cmp x (toStreamD m)
549+
550+------------------------------------------------------------------------------
551+-- Trimming
552+------------------------------------------------------------------------------
553+
554+-- | Same as 'takeWhile' but with a monadic predicate.
555+--
556+{-# INLINE takeWhileM #-}
557+takeWhileM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
558+takeWhileM p m = fromStreamD $ D.takeWhileM p $ toStreamD m
559+
560+-- | Drop elements in the stream as long as the predicate succeeds and then
561+-- take the rest of the stream.
562+--
563+{-# INLINE dropWhile #-}
564+dropWhile :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
565+dropWhile p m = fromStreamD $ D.dropWhile p $ toStreamD m
566+
567+-- | Same as 'dropWhile' but with a monadic predicate.
568+--
569+{-# INLINE dropWhileM #-}
570+dropWhileM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
571+dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m
572+
573+------------------------------------------------------------------------------
574+-- Inserting Elements
575+------------------------------------------------------------------------------
576+
577+-- | @insertBy cmp elem stream@ inserts @elem@ before the first element in
578+-- @stream@ that is less than @elem@ when compared using @cmp@.
579+--
580+-- >>> insertBy cmp x = Stream.mergeBy cmp (Stream.fromPure x)
581+--
582+-- >>> input = Stream.unfold Unfold.fromList [1,3,5]
583+-- >>> Stream.fold Fold.toList $ Stream.insertBy compare 2 input
584+-- [1,2,3,5]
585+--
586+{-# INLINE insertBy #-}
587+insertBy ::Monad m => (a -> a -> Ordering) -> a -> Stream m a -> Stream m a
588+insertBy cmp x m = fromStreamD $ D.insertBy cmp x (toStreamD m)
589+
590+-- | Insert a pure value between successive elements of a stream.
591+--
592+-- >>> input = Stream.unfold Unfold.fromList "hello"
593+-- >>> Stream.fold Fold.toList $ Stream.intersperse ',' input
594+-- "h,e,l,l,o"
595+--
596+{-# INLINE intersperse #-}
597+intersperse :: Monad m => a -> Stream m a -> Stream m a
598+intersperse a = fromStreamD . D.intersperse a . toStreamD
599+
600+-- | Insert a side effect before consuming an element of a stream except the
601+-- first one.
602+--
603+-- >>> input = Stream.unfold Unfold.fromList "hello"
604+-- >>> Stream.fold Fold.drain $ Stream.trace putChar $ Stream.intersperseM_ (putChar '.') input
605+-- h.e.l.l.o
606+--
607+-- /Pre-release/
608+{-# INLINE intersperseM_ #-}
609+intersperseM_ :: Monad m => m b -> Stream m a -> Stream m a
610+intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
611+
612+-- | Intersperse a monadic action into the input stream after every @n@
613+-- elements.
614+--
615+-- >> input = Stream.unfold Unfold.fromList "hello"
616+-- >> Stream.fold Fold.toList $ Stream.intersperseBySpan 2 (return ',') input
617+-- "he,ll,o"
618+--
619+-- /Unimplemented/
620+{-# INLINE intersperseBySpan #-}
621+intersperseBySpan :: -- Monad m =>
622+ Int -> m a -> t m a -> t m a
623+intersperseBySpan _n _f _xs = undefined
624+
625+-- | Insert an effect and its output after consuming an element of a stream.
626+--
627+-- >>> input = Stream.unfold Unfold.fromList "hello"
628+-- >>> Stream.fold Fold.toList $ Stream.trace putChar $ intersperseSuffix (putChar '.' >> return ',') input
629+-- h.,e.,l.,l.,o.,"h,e,l,l,o,"
630+--
631+-- /Pre-release/
632+{-# INLINE intersperseSuffix #-}
633+intersperseSuffix :: Monad m => m a -> Stream m a -> Stream m a
634+intersperseSuffix m = fromStreamD . D.intersperseSuffix m . toStreamD
635+
636+-- | Insert a side effect after consuming an element of a stream.
637+--
638+-- >>> input = Stream.unfold Unfold.fromList "hello"
639+-- >>> Stream.fold Fold.toList $ Stream.intersperseSuffix_ (threadDelay 1000000) input
640+-- "hello"
641+--
642+-- /Pre-release/
643+--
644+{-# INLINE intersperseSuffix_ #-}
645+intersperseSuffix_ :: Monad m => m b -> Stream m a -> Stream m a
646+intersperseSuffix_ m = fromStreamD . D.intersperseSuffix_ m . toStreamD
647+
648+-- XXX Use an offset argument, like tapOffsetEvery
649+
650+-- | Like 'intersperseSuffix' but intersperses an effectful action into the
651+-- input stream after every @n@ elements and after the last element.
652+--
653+-- >>> input = Stream.unfold Unfold.fromList "hello"
654+-- >>> Stream.fold Fold.toList $ Stream.intersperseSuffixBySpan 2 (return ',') input
655+-- "he,ll,o,"
656+--
657+-- /Pre-release/
658+--
659+{-# INLINE intersperseSuffixBySpan #-}
660+intersperseSuffixBySpan :: Monad m
661+ => Int -> m a -> Stream m a -> Stream m a
662+intersperseSuffixBySpan n eff =
663+ fromStreamD . D.intersperseSuffixBySpan n eff . toStreamD
664+
665+-- | Insert a side effect before consuming an element of a stream.
666+--
667+-- >>> input = Stream.unfold Unfold.fromList "hello"
668+-- >>> Stream.fold Fold.toList $ Stream.trace putChar $ Stream.interspersePrefix_ (putChar '.' >> return ',') input
669+-- .h.e.l.l.o"hello"
670+--
671+-- Same as 'trace_'.
672+--
673+-- /Pre-release/
674+--
675+{-# INLINE interspersePrefix_ #-}
676+interspersePrefix_ :: Monad m => m b -> Stream m a -> Stream m a
677+interspersePrefix_ m = mapM (\x -> void m >> return x)
678+
679+------------------------------------------------------------------------------
680+-- Inserting Time
681+------------------------------------------------------------------------------
682+
683+-- | Introduce a delay of specified seconds between elements of the stream.
684+--
685+-- >>> input = Stream.unfold Unfold.enumerateFromTo (1, 3)
686+-- >>> Stream.fold (Fold.drainBy print) $ Stream.delay 1 input
687+-- 1
688+-- 2
689+-- 3
690+--
691+{-# INLINE delay #-}
692+delay :: MonadIO m => Double -> Stream m a -> Stream m a
693+delay n = intersperseM_ $ liftIO $ threadDelay $ round $ n * 1000000
694+
695+-- | Introduce a delay of specified seconds after consuming an element of a
696+-- stream.
697+--
698+-- >>> input = Stream.unfold Unfold.enumerateFromTo (1, 3)
699+-- >>> Stream.fold (Fold.drainBy print) $ Stream.delayPost 1 input
700+-- 1
701+-- 2
702+-- 3
703+--
704+-- /Pre-release/
705+--
706+{-# INLINE delayPost #-}
707+delayPost :: MonadIO m => Double -> Stream m a -> Stream m a
708+delayPost n = intersperseSuffix_ $ liftIO $ threadDelay $ round $ n * 1000000
709+
710+-- | Introduce a delay of specified seconds before consuming an element of a
711+-- stream.
712+--
713+-- >>> input = Stream.unfold Unfold.enumerateFromTo (1, 3)
714+-- >>> Stream.fold (Fold.drainBy print) $ Stream.delayPre 1 input
715+-- 1
716+-- 2
717+-- 3
718+--
719+-- /Pre-release/
720+--
721+{-# INLINE delayPre #-}
722+delayPre :: MonadIO m => Double -> Stream m a -> Stream m a
723+delayPre n = trace_ $ liftIO $ threadDelay $ round $ n * 1000000
724+
725+------------------------------------------------------------------------------
726+-- Reorder in sequence
727+------------------------------------------------------------------------------
728+
729+-- | Buffer until the next element in sequence arrives. The function argument
730+-- determines the difference in sequence numbers. This could be useful in
731+-- implementing sequenced streams, for example, TCP reassembly.
732+--
733+-- /Unimplemented/
734+--
735+{-# INLINE reassembleBy #-}
736+reassembleBy
737+ :: -- Monad m =>
738+ Fold m a b
739+ -> (a -> a -> Int)
740+ -> t m a
741+ -> t m b
742+reassembleBy = undefined
743+
744+------------------------------------------------------------------------------
745+-- Position Indexing
746+------------------------------------------------------------------------------
747+
748+-- |
749+-- >>> f = Fold.foldl' (\(i, _) x -> (i + 1, x)) (-1,undefined)
750+-- >>> indexed = Stream.postscan f
751+-- >>> indexed = Stream.zipWith (,) (Stream.unfold Unfold.enumerateFrom 0)
752+--
753+-- Pair each element in a stream with its index, starting from index 0.
754+--
755+-- >>> Stream.fold Fold.toList $ Stream.indexed $ Stream.unfold Unfold.fromList "hello"
756+-- [(0,'h'),(1,'e'),(2,'l'),(3,'l'),(4,'o')]
757+--
758+{-# INLINE indexed #-}
759+indexed :: Monad m => Stream m a -> Stream m (Int, a)
760+indexed = fromStreamD . D.indexed . toStreamD
761+
762+-- |
763+-- >>> f n = Fold.foldl' (\(i, _) x -> (i - 1, x)) (n + 1,undefined)
764+-- >>> indexedR n = Stream.postscan (f n)
765+--
766+-- >>> s n = Stream.unfold Unfold.enumerateFromThen (n, (n - 1))
767+-- >>> indexedR n = Stream.zipWith (,) (s n)
768+--
769+-- Pair each element in a stream with its index, starting from the
770+-- given index @n@ and counting down.
771+--
772+-- >>> Stream.fold Fold.toList $ Stream.indexedR 10 $ Stream.unfold Unfold.fromList "hello"
773+-- [(10,'h'),(9,'e'),(8,'l'),(7,'l'),(6,'o')]
774+--
775+{-# INLINE indexedR #-}
776+indexedR :: Monad m => Int -> Stream m a -> Stream m (Int, a)
777+indexedR n = fromStreamD . D.indexedR n . toStreamD
778+
779+------------------------------------------------------------------------------
780+-- Searching
781+------------------------------------------------------------------------------
782+
783+-- | Find all the indices where the value of the element in the stream is equal
784+-- to the given value.
785+--
786+-- >>> elemIndices a = Stream.findIndices (== a)
787+--
788+{-# INLINE elemIndices #-}
789+elemIndices :: (Monad m, Eq a) => a -> Stream m a -> Stream m Int
790+elemIndices a = findIndices (== a)
791+
792+------------------------------------------------------------------------------
793+-- Rolling map
794+------------------------------------------------------------------------------
795+
796+-- XXX this is not a one-to-one map so calling it map may not be right.
797+-- We can perhaps call it zipWithTail or rollWith.
798+
799+-- | Apply a function on every two successive elements of a stream. The first
800+-- argument of the map function is the previous element and the second argument
801+-- is the current element. When the current element is the first element, the
802+-- previous element is 'Nothing'.
803+--
804+-- /Pre-release/
805+--
806+{-# INLINE rollingMap #-}
807+rollingMap :: Monad m => (Maybe a -> a -> b) -> Stream m a -> Stream m b
808+rollingMap f m = fromStreamD $ D.rollingMap f $ toStreamD m
809+
810+-- | Like 'rollingMap' but with an effectful map function.
811+--
812+-- /Pre-release/
813+--
814+{-# INLINE rollingMapM #-}
815+rollingMapM :: Monad m => (Maybe a -> a -> m b) -> Stream m a -> Stream m b
816+rollingMapM f m = fromStreamD $ D.rollingMapM f $ toStreamD m
817+
818+-- | Like 'rollingMap' but requires at least two elements in the stream,
819+-- returns an empty stream otherwise.
820+--
821+-- This is the stream equivalent of the list idiom @zipWith f xs (tail xs)@.
822+--
823+-- /Pre-release/
824+--
825+{-# INLINE rollingMap2 #-}
826+rollingMap2 :: Monad m => (a -> a -> b) -> Stream m a -> Stream m b
827+rollingMap2 f m = fromStreamD $ D.rollingMap2 f $ toStreamD m
828+
829+------------------------------------------------------------------------------
830+-- Maybe Streams
831+------------------------------------------------------------------------------
832+
833+-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'
834+-- elements, and return a stream of values extracted from 'Just'.
835+--
836+-- Equivalent to:
837+--
838+-- >>> mapMaybe f = fmap fromJust . Stream.filter isJust . fmap f
839+--
840+{-# INLINE mapMaybe #-}
841+mapMaybe :: Monad m => (a -> Maybe b) -> Stream m a -> Stream m b
842+mapMaybe f m = fromStreamD $ D.mapMaybe f $ toStreamD m
843+
844+-- | Like 'mapMaybe' but maps a monadic function.
845+--
846+-- Equivalent to:
847+--
848+-- >>> mapMaybeM f = fmap fromJust . Stream.filter isJust . Stream.mapM f
849+--
850+{-# INLINE_EARLY mapMaybeM #-}
851+mapMaybeM :: Monad m
852+ => (a -> m (Maybe b)) -> Stream m a -> Stream m b
853+mapMaybeM f = fmap fromJust . filter isJust . mapM f
854+
855+-- | In a stream of 'Maybe's, discard 'Nothing's and unwrap 'Just's.
856+--
857+-- /Pre-release/
858+--
859+{-# INLINE catMaybes #-}
860+catMaybes :: Monad m => Stream m (Maybe a) -> Stream m a
861+catMaybes = fmap fromJust . filter isJust
862+
863+------------------------------------------------------------------------------
864+-- Either streams
865+------------------------------------------------------------------------------
866+
867+-- | Discard 'Right's and unwrap 'Left's in an 'Either' stream.
868+--
869+-- /Pre-release/
870+--
871+{-# INLINE lefts #-}
872+lefts :: Monad m => Stream m (Either a b) -> Stream m a
873+lefts = fmap (fromLeft undefined) . filter isLeft
874+
875+-- | Discard 'Left's and unwrap 'Right's in an 'Either' stream.
876+--
877+-- /Pre-release/
878+--
879+{-# INLINE rights #-}
880+rights :: Monad m => Stream m (Either a b) -> Stream m b
881+rights = fmap (fromRight undefined) . filter isRight
882+
883+-- | Remove the either wrapper and flatten both lefts and as well as rights in
884+-- the output stream.
885+--
886+-- /Pre-release/
887+--
888+{-# INLINE both #-}
889+both :: Monad m => Stream m (Either a a) -> Stream m a
890+both = fmap (either id id)