@@ -8,80 +8,968 @@
88--
99module Streamly.Internal.Data.Stream.Transform
1010 (
11− filter
12− , foldFilter
11+ -- * Piping
12+ -- | Pass through a 'Pipe'.
13+ transform
14+
15+ -- * Folding
16+ , foldrS
17+ , foldrSShared
18+ -- , foldrT
19+
20+ -- * Mapping
21+ -- | Stateless one-to-one maps.
1322 , map
23+ , sequence
1424 , mapM
25+ , smapM
26+
27+ -- * Mapping Side Effects (Observation)
28+ -- | See also the intersperse*_ combinators.
29+ , trace
30+ , trace_
31+ , tap
32+ --, tapOffsetEvery
33+ --, tapAsync
34+ --, tapAsyncK
35+ --, distributeAsync_
36+ --, pollCounts
37+
38+ -- * Scanning By 'Fold'
39+ , scan
40+ , scanMany
41+ , postscan
42+
43+ -- * Scanning
44+ -- | Left scans. Stateful, mostly one-to-one maps.
45+ , scanl'
46+ , scanlM'
47+ , scanlMAfter'
48+ , postscanlMAfter'
49+ , postscanl'
50+ , postscanlM'
51+ , prescanl'
52+ , prescanlM'
53+ , scanl1'
54+ , scanl1M'
55+
56+ -- XXX Once we have pipes the contravariant transformations can be
57+ -- represented by attaching pipes before a transformation.
58+ --
59+ -- , lscanl'
60+ -- , lscanlM'
61+ -- , lscanl1'
62+ -- , lscanl1M'
63+ --
64+ -- , lpostscanl'
65+ -- , lpostscanlM'
66+ -- , lprescanl'
67+ -- , lprescanlM'
68+
69+ -- * Filtering
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.
73+
74+ --, with
75+ , deleteBy
76+ , filter
77+ , filterM
78+ , foldFilter
79+ , uniq
80+ , uniqBy
81+ --, nubBy
82+ --, nubWindowBy
83+ --, prune
84+ --, repeated
85+
86+ -- * Trimming
87+ -- | Produce a subset of the stream trimmed at ends.
88+
89+ , take
90+ , takeInterval
91+ --, takeLast
92+ --, takeLastInterval
93+ , takeWhile
94+ , takeWhileM
95+ --, takeWhileLast
96+ --, takeWhileAround
97+ , drop
98+ , dropInterval
99+ --, dropLast
100+ --, dropLastInterval
101+ , dropWhile
102+ , dropWhileM
103+ --, dropWhileLast
104+ --, dropWhileAround
105+
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.
110+
111+ -- Element agnostic (Opposite of sampling)
112+ , intersperse
113+ , intersperseM -- XXX naming
114+ --, intersperseBySpan
115+
116+ , intersperseSuffix
117+ , intersperseSuffixBySpan
118+ --, interjectSuffix
119+
120+ -- , interspersePrefix
121+ -- , interspersePrefixBySpan
122+
123+ -- * Inserting Side Effects/Time
124+ , intersperseM_ -- XXX naming
125+ , delay
126+ , intersperseSuffix_
127+ , delayPost
128+ , interspersePrefix_
129+ , delayPre
130+
131+ -- * Element Aware Insertion
132+ -- | Opposite of filtering
133+ , insertBy
134+ -- , intersperseByBefore
135+ -- , intersperseByAfter
136+
137+ -- * Reordering
138+ , reverse
139+ , reverse'
140+ -- , reassembleBy
141+
142+ -- * Position Indexing
143+ , indexed
144+ , indexedR
145+
146+ -- * Searching
147+ , findIndices -- XXX indicesBy
148+ , elemIndices -- XXX indicesOf
149+
150+ -- * Rolling map
151+ -- | Map using the previous element.
152+ , rollingMapM
153+ , rollingMap
154+ , rollingMap2
155+
156+ -- * Maybe Streams
157+ -- Move these to Streamly.Data.Maybe.Stream?
158+ , catMaybes -- XXX justs (like lefts/rights)
159+ , mapMaybe
160+ , mapMaybeM
161+
162+ -- * Either Streams
163+ -- Move these to Streamly.Data.Either.Stream?
164+ , lefts
165+ , rights
166+ , both
15167 )
16168where
17169
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)
175+
18176import 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)
179+
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
20184
21−import qualified Streamly.Internal.Data.Stream.Type as Stream
22−import qualified Streamly.Internal.Data.Stream.StreamD.Transform as StreamD
185+import Streamly.Internal.Data.Stream.Bottom
186+import Streamly.Internal.Data.Stream.Type
23187
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)
191+
192+#include "Instances.hs"
193+#include "inline.hs"
25194
26195--
27196-- $setup
28197-- >>> :m
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
29203-- >>> 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
204+-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
205+-- >>> import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
206+--
207+-- >>> hSetBuffering stdout LineBuffering
32208
33209------------------------------------------------------------------------------
34210-- Transformation by Mapping
35211------------------------------------------------------------------------------
36212
37213-- |
38−-- >> mapM f = Stream.sequence . Stream.map f
214+-- @
215+-- mapM f = sequence . map f
216+-- @
39217--
40218-- Apply a monadic function to each element of the stream and replace it with
41219-- the output of the resulting action.
42220--
43−-- >>> s = Stream.unfold Unfold.fromList ["a", "b", "c"]
44−-- >>> Stream.fold Fold.drain $ Stream.mapM putStr s
221+-- @
222+-- > drain $ Stream.mapM putStr $ Stream.unfold Unfold.fromList ["a", "b", "c"]
45223-- abc
46224--
47−-- /Pre-release/
225+-- > :{
226+-- drain $ Stream.replicateM 10 (return 1)
227+-- & (fromSerial . Stream.mapM (\x -> threadDelay 1000000 >> print x))
228+-- :}
229+-- 1
230+-- ...
231+-- 1
48232--
49−{-# INLINE mapM #-}
50−mapM :: Monad m => (a -> m b) -> Stream m a -> Stream m b
51−mapM f = Stream.fromStreamD . StreamD.mapM f . Stream.toStreamD
52−
53−-- |
54−-- >>> map = fmap
233+-- > drain $ Stream.replicateM 10 (return 1)
234+-- & (fromAsync . Stream.mapM (\x -> threadDelay 1000000 >> print x))
235+-- @
55236--
56−-- Same as 'fmap'.
57−--
58−-- >>> s = Stream.unfold Unfold.fromList [1,2,3]
59−-- >>> Stream.fold Fold.toList $ Stream.map (+1) s
60−-- [2,3,4]
237+-- /Concurrent (do not use with 'fromParallel' on infinite streams)/
61238--
62239-- /Pre-release/
63240--
64−{-# INLINE map #-}
65−map :: Monad m => (a -> b) -> Stream m a -> Stream m b
66−map f = mapM (return . f)
241+{-# INLINE mapM #-}
242+mapM :: Monad m => (a -> m b) -> Stream m a -> Stream m b
243+mapM f m = fromStreamK $ D.toStreamK $ D.mapM f $ toStreamD m
67244
68245-- | Include only those elements that pass a predicate.
69246--
70247{-# INLINE filter #-}
71248filter :: 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
250+
251+-- | Same as 'filter' but with a monadic predicate.
252+--
253+-- @since 0.4.0
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
73257
74258-- | Use a filtering fold on a stream.
75259--
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
260+-- > Stream.sum $ Stream.foldFilter (Fold.satisfy (> 5)) $ Stream.unfold Unfold.fromList [1..10]
261+-- 40
82262--
83263-- /Pre-release/
84264--
85265{-# INLINE foldFilter #-}
86266foldFilter :: 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
268+
269+------------------------------------------------------------------------------
270+-- Piping
271+------------------------------------------------------------------------------
272+
273+-- | Use a 'Pipe' to transform a stream.
274+--
275+-- /Pre-release/
276+--
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)
280+
281+------------------------------------------------------------------------------
282+-- Transformation Folds
283+------------------------------------------------------------------------------
284+
285+-- | Right fold to a streaming monad.
286+--
287+-- > foldrS Stream.cons Stream.nil === id
288+--
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.
293+--
294+-- >>> Stream.fold Fold.toList $ Stream.foldrS Stream.cons Stream.nil $ Stream.unfold Unfold.fromList [1..5]
295+-- [1,2,3,4,5]
296+--
297+-- Find if any element in the stream is 'True':
298+--
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)
300+-- [True]
301+--
302+-- Map (+2) on odd elements and filter out the even elements:
303+--
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)
305+-- [3,5,7]
306+--
307+-- 'foldrM' can also be represented in terms of 'foldrS', however, the former
308+-- iStream much more efficient:
309+--
310+-- > foldrM f z s = runIdentityT $ foldrS (\x xs -> lift $ f x (runIdentityT xs)) (lift z) s
311+--
312+-- /Pre-release/
313+{-# INLINE foldrS #-}
314+foldrS ::
315+ (a -> Stream m b -> Stream m b)
316+ -> Stream m b
317+ -> Stream m a
318+ -> Stream m b
319+foldrS f z xs =
320+ fromStreamK
321+ $ K.foldrS
322+ (\y ys -> toStreamK $ f y (fromStreamK ys))
323+ (toStreamK z)
324+ (toStreamK xs)
325+
326+{-# INLINE foldrSShared #-}
327+foldrSShared ::
328+ (a -> Stream m b -> Stream m b)
329+ -> Stream m b
330+ -> Stream m a
331+ -> Stream m b
332+foldrSShared f z xs =
333+ fromStreamK
334+ $ K.foldrSShared
335+ (\y ys -> toStreamK $ f y (fromStreamK ys))
336+ (toStreamK z)
337+ (toStreamK xs)
338+
339+{-# INLINE sequence #-}
340+sequence :: (Monad m) => Stream m (m a) -> Stream m a
341+sequence = mapM id
342+
343+{-# INLINE tap #-}
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)
346+
347+-- | Apply a monadic function to each element flowing through the stream and
348+-- discard the results.
349+--
350+-- @
351+-- >>> Stream.fold Fold.drain $ Stream.trace print (Stream.unfold Unfold.fromList[1, 2])
352+-- 1
353+-- 2
354+--
355+-- @
356+--
357+-- Compare with 'tap'.
358+--
359+-- @since 0.7.0
360+{-# INLINE trace #-}
361+trace :: (Monad m) => (a -> m b) -> Stream m a -> Stream m a
362+trace f = mapM (\x -> void (f x) >> return x)
363+
364+-- | Perform a side effect before yielding each element of the stream and
365+-- discard the results.
366+--
367+-- @
368+-- >>> Stream.fold Fold.drain $ Stream.trace_ (print "got here") (Stream.unfold Unfold.fromList[1, 2])
369+-- "got here"
370+-- "got here"
371+--
372+-- @
373+--
374+-- Same as 'interspersePrefix_' but always serial.
375+--
376+-- See also: 'trace'
377+--
378+-- /Pre-release/
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
382+
383+------------------------------------------------------------------------------
384+-- Scanning with a Fold
385+------------------------------------------------------------------------------
386+
387+-- | Scan a stream using the given monadic fold.
388+--
389+-- >>> Stream.fold Fold.toList $ Stream.takeWhile (< 10) $ Stream.scan Fold.sum (Stream.unfold Unfold.fromList [1..10])
390+-- [0,1,3,6]
391+--
392+-- @since 0.7.0
393+{-# INLINE scan #-}
394+scan :: (Monad m) => Fold m a b -> Stream m a -> Stream m b
395+scan fld m = fromStreamD $ D.scanOnce fld $ toStreamD m
396+
397+-- | Like 'scan' but restarts scanning afresh when the scanning fold
398+-- terminates.
399+--
400+-- /Pre-release/
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
404+
405+-- | Postscan a stream using the given monadic fold.
406+--
407+-- The following example extracts the input stream up to a point where the
408+-- running average of elements is no more than 10:
409+--
410+-- >>> import Data.Maybe (fromJust)
411+-- >>> let avg = Fold.teeWith (/) Fold.sum (fmap fromIntegral Fold.length)
412+-- >>> :{
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)
417+-- :}
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]
419+--
420+-- @since 0.7.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
424+
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.
429+--
430+-- | Like 'scanl'' but with a monadic step function and a monadic seed.
431+--
432+-- /Since: 0.4.0/
433+--
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
438+
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.
442+
443+-- | Strict left scan. Like 'map', 'scanl'' too is a one to one transformation,
444+-- however it adds an extra element.
445+--
446+-- @
447+-- >>> Stream.fold Fold.toList $ Stream.scanl' (+) 0 $ Stream.unfold Unfold.fromList [1,2,3,4]
448+-- [0,1,3,6,10]
449+--
450+-- @
451+--
452+-- @
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]]
455+--
456+-- @
457+--
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''.
460+--
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.
466+--
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'@:
469+--
470+-- @
471+-- >>> Stream.foldl' (\(s, p) x -> (s + x, p * x)) (0,1) $ Stream.unfold Unfold.fromList [1,2,3,4]
472+-- (10,24)
473+--
474+-- @
475+--
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:
478+--
479+-- @
480+-- >>> :{
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]
484+-- :}
485+-- (10,24)
486+--
487+-- @
488+--
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.
492+--
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
496+--
497+-- See also: 'usingStateT'
498+--
499+-- @since 0.2.0
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
503+
504+-- | Like 'scanl'' but does not stream the initial value of the accumulator.
505+--
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
509+--
510+-- @since 0.7.0
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
514+
515+-- XXX prescanl does not sound very useful, enable only if there is a
516+-- compelling use case.
517+--
518+-- | Like scanl' but does not stream the final value of the accumulator.
519+--
520+-- /Pre-release/
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
524+
525+-- XXX this needs to be concurrent
526+-- | Like prescanl' but with a monadic step function and a monadic seed.
527+--
528+-- /Pre-release/
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
532+
533+-- XXX this needs to be concurrent
534+-- | Like 'scanl1'' but with a monadic step function.
535+--
536+-- @since 0.6.0
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
540+
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
543+-- is empty.
544+--
545+-- @
546+-- >>> Stream.fold Fold.toList $ Stream.scanl1' (+) $ Stream.unfold Unfold.fromList [1,2,3,4]
547+-- [1,3,6,10]
548+--
549+-- @
550+--
551+-- @since 0.6.0
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
555+
556+-- | Drop repeated elements that are adjacent to each other using the supplied
557+-- comparison function.
558+--
559+-- @uniq = uniqBy (==)
560+--
561+-- To strip duplicate path separators:
562+--
563+-- @
564+-- f x y = x == '/' && x == y
565+-- Stream.fold Fold.toList $ Stream.uniqBy f $ Stream.unfold Unfold.fromList "//a//b"
566+-- "/a/b"
567+-- @
568+--
569+-- Space: @O(1)@
570+--
571+-- See also: 'nubBy'.
572+--
573+-- /Pre-release/
574+--
575+{-# INLINE uniqBy #-}
576+uniqBy :: (Monad m) =>
577+ (a -> a -> Bool) -> Stream m a -> Stream m a
578+uniqBy eq = catMaybes . rollingMap f
579+
580+ where
581+
582+ f pre curr =
583+ case pre of
584+ Nothing -> Just curr
585+ Just x -> if x `eq` curr then Nothing else Just curr
586+
587+-- | Drop repeated elements that are adjacent to each other.
588+--
589+-- @since 0.6.0
590+{-# INLINE uniq #-}
591+uniq :: (Eq a, Monad m) => Stream m a -> Stream m a
592+uniq = fromStreamD . D.uniq . toStreamD
593+
594+-- | Deletes the first occurrence of the element in the stream that satisfies
595+-- the given equality predicate.
596+--
597+-- @
598+-- >>> Stream.fold Fold.toList $ Stream.deleteBy (==) 3 $ Stream.unfold Unfold.fromList [1,3,3,5]
599+-- [1,3,5]
600+--
601+-- @
602+--
603+-- @since 0.6.0
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)
607+
608+------------------------------------------------------------------------------
609+-- Trimming
610+------------------------------------------------------------------------------
611+
612+-- | Same as 'takeWhile' but with a monadic predicate.
613+--
614+-- @since 0.4.0
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
618+
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
623+-- stops.
624+--
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.
629+--
630+-- No element is yielded if the duration is zero. At least one element is
631+-- yielded if the duration is non-zero.
632+--
633+-- /Pre-release/
634+--
635+{-# INLINE takeInterval #-}
636+takeInterval ::(MonadIO m, TimeUnit64 d) => d -> Stream m a -> Stream m a
637+takeInterval d = fromStreamD . D.takeByTime d . toStreamD
638+
639+-- | Drop elements in the stream as long as the predicate succeeds and then
640+-- take the rest of the stream.
641+--
642+-- @since 0.1.0
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
646+
647+-- | Same as 'dropWhile' but with a monadic predicate.
648+--
649+-- @since 0.4.0
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
653+
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.
658+--
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.
663+--
664+-- All elements are yielded if the duration is zero.
665+--
666+-- /Pre-release/
667+--
668+{-# INLINE dropInterval #-}
669+dropInterval ::(MonadIO m, TimeUnit64 d) => d -> Stream m a -> Stream m a
670+dropInterval d = fromStreamD . D.dropByTime d . toStreamD
671+
672+------------------------------------------------------------------------------
673+-- Inserting Elements
674+------------------------------------------------------------------------------
675+
676+-- | @insertBy cmp elem stream@ inserts @elem@ before the first element in
677+-- @stream@ that is less than @elem@ when compared using @cmp@.
678+--
679+-- @
680+-- insertBy cmp x = 'mergeBy' cmp ('fromPure' x)
681+-- @
682+--
683+-- @
684+-- >>> Stream.fold Fold.toList $ Stream.insertBy compare 2 $ Stream.unfold Unfold.fromList [1,3,5]
685+-- [1,2,3,5]
686+--
687+-- @
688+--
689+-- @since 0.6.0
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)
693+
694+-- | Insert a pure value between successive elements of a stream.
695+--
696+-- >>> Stream.fold Fold.toList $ Stream.intersperse ',' $ Stream.unfold Unfold.fromList "hello"
697+-- "h,e,l,l,o"
698+--
699+-- @since 0.7.0
700+{-# INLINE intersperse #-}
701+intersperse :: (Monad m) => a -> Stream m a -> Stream m a
702+intersperse a = fromStreamD . D.intersperse a . toStreamD
703+
704+-- | Insert a side effect before consuming an element of a stream except the
705+-- first one.
706+--
707+-- >>> Stream.fold Fold.drain $ Stream.trace putChar $ Stream.intersperseM_ (putChar '.') $ Stream.unfold Unfold.fromList "hello"
708+-- h.e.l.l.o
709+--
710+-- /Pre-release/
711+{-# INLINE intersperseM_ #-}
712+intersperseM_ :: (Monad m) => m b -> Stream m a -> Stream m a
713+intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
714+
715+-- | Insert an effect and its output after consuming an element of a stream.
716+--
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,"
719+--
720+-- /Pre-release/
721+{-# INLINE intersperseSuffix #-}
722+intersperseSuffix :: (Monad m) => m a -> Stream m a -> Stream m a
723+intersperseSuffix m = fromStreamD . D.intersperseSuffix m . toStreamD
724+
725+-- | Insert a side effect after consuming an element of a stream.
726+--
727+-- @
728+-- >>> Stream.fold Fold.toList $ Stream.intersperseSuffix_ (threadDelay 1000000) $ Stream.unfold Unfold.fromList "hello"
729+-- "hello"
730+--
731+-- @
732+--
733+-- /Pre-release/
734+--
735+{-# INLINE intersperseSuffix_ #-}
736+intersperseSuffix_ :: (Monad m) => m b -> Stream m a -> Stream m a
737+intersperseSuffix_ m = fromStreamD . D.intersperseSuffix_ m . toStreamD
738+
739+-- XXX Use an offset argument, like tapOffsetEvery
740+--
741+-- | Like 'intersperseSuffix' but intersperses an effectful action into the
742+-- input stream after every @n@ elements and after the last element.
743+--
744+-- >>> Stream.fold Fold.toList $ Stream.intersperseSuffixBySpan 2 (return ',') $ Stream.unfold Unfold.fromList "hello"
745+-- "he,ll,o,"
746+--
747+-- /Pre-release/
748+--
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
754+
755+-- | Insert a side effect before consuming an element of a stream.
756+--
757+-- >>> Stream.fold Fold.toList $ Stream.trace putChar $ Stream.interspersePrefix_ (putChar '.' >> return ',') $ Stream.unfold Unfold.fromList "hello"
758+-- .h.e.l.l.o"hello"
759+--
760+-- Same as 'trace_' buStream may be concurrent.
761+--
762+-- /Concurrent/
763+--
764+-- /Pre-release/
765+--
766+{-# INLINE interspersePrefix_ #-}
767+interspersePrefix_ :: (Monad m) => m b -> Stream m a -> Stream m a
768+interspersePrefix_ m = mapM (\x -> void m >> return x)
769+
770+------------------------------------------------------------------------------
771+-- Inserting Time
772+------------------------------------------------------------------------------
773+
774+-- Note: delay must be serial.
775+--
776+-- | Introduce a delay of specified seconds before consuming an element of the
777+-- stream except the first one.
778+--
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)
783+--
784+-- @since 0.8.0
785+--
786+{-# INLINE delay #-}
787+delay :: (MonadIO m) => Double -> Stream m a -> Stream m a
788+delay n = intersperseM_ $ liftIO $ threadDelay $ round $ n * 1000000
789+
790+-- Note: delay must be serial.
791+--
792+-- | Introduce a delay of specified seconds after consuming an element of a
793+-- stream.
794+--
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)
799+--
800+-- /Pre-release/
801+--
802+{-# INLINE delayPost #-}
803+delayPost :: (MonadIO m) => Double -> Stream m a -> Stream m a
804+delayPost n = intersperseSuffix_ $ liftIO $ threadDelay $ round $ n * 1000000
805+
806+-- Note: delay must be serial, that's why 'trace_' is used.
807+--
808+-- | Introduce a delay of specified seconds before consuming an element of a
809+-- stream.
810+--
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)
815+--
816+-- /Pre-release/
817+--
818+{-# INLINE delayPre #-}
819+delayPre :: (MonadIO m) => Double -> Stream m a -> Stream m a
820+delayPre n = trace_ $ liftIO $ threadDelay $ round $ n * 1000000
821+
822+------------------------------------------------------------------------------
823+-- Position Indexing
824+------------------------------------------------------------------------------
825+
826+-- |
827+-- > indexed = Stream.postscanl' (\(i, _) x -> (i + 1, x)) (-1,undefined)
828+-- > indexed = Stream.zipWith (,) (Stream.enumerateFrom 0)
829+--
830+-- Pair each element in a stream with its index, starting from index 0.
831+--
832+-- >>> Stream.fold Fold.toList $ Stream.indexed $ Stream.unfold Unfold.fromList "hello"
833+-- [(0,'h'),(1,'e'),(2,'l'),(3,'l'),(4,'o')]
834+--
835+-- @since 0.6.0
836+{-# INLINE indexed #-}
837+indexed :: (Monad m) => Stream m a -> Stream m (Int, a)
838+indexed = fromStreamD . D.indexed . toStreamD
839+
840+-- |
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))
843+--
844+-- Pair each element in a stream with its index, starting from the
845+-- given index @n@ and counting down.
846+--
847+-- >>> Stream.fold Fold.toList $ Stream.indexedR 10 $ Stream.unfold Unfold.fromList "hello"
848+-- [(10,'h'),(9,'e'),(8,'l'),(7,'l'),(6,'o')]
849+--
850+-- @since 0.6.0
851+{-# INLINE indexedR #-}
852+indexedR :: (Monad m) => Int -> Stream m a -> Stream m (Int, a)
853+indexedR n = fromStreamD . D.indexedR n . toStreamD
854+
855+------------------------------------------------------------------------------
856+-- Searching
857+------------------------------------------------------------------------------
858+
859+-- | Find all the indices where the value of the element in the stream is equal
860+-- to the given value.
861+--
862+-- > elemIndices a = findIndices (== a)
863+--
864+-- @since 0.5.0
865+{-# INLINE elemIndices #-}
866+elemIndices :: (Eq a, Monad m) => a -> Stream m a -> Stream m Int
867+elemIndices a = findIndices (== a)
868+
869+------------------------------------------------------------------------------
870+-- Rolling map
871+------------------------------------------------------------------------------
872+
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.
875+--
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'.
880+--
881+-- /Pre-release/
882+--
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
886+
887+-- | Like 'rollingMap' but with an effectful map function.
888+--
889+-- /Pre-release/
890+--
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
894+
895+-- | Like 'rollingMap' but requires at least two elements in the stream,
896+-- returns an empty stream otherwise.
897+--
898+-- This is the stream equivalent of the list idiom @zipWith f xs (tail xs)@.
899+--
900+-- /Pre-release/
901+--
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
905+
906+------------------------------------------------------------------------------
907+-- Maybe Streams
908+------------------------------------------------------------------------------
909+
910+-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'
911+-- elements, and return a stream of values extracted from 'Just'.
912+--
913+-- Equivalent to:
914+--
915+-- @
916+-- mapMaybe f = Stream.map 'fromJust' . Stream.filter 'isJust' . Stream.map f
917+-- @
918+--
919+-- @since 0.3.0
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
923+
924+-- | Like 'mapMaybe' buStream maps a monadic function.
925+--
926+-- Equivalent to:
927+--
928+-- @
929+-- mapMaybeM f = Stream.map 'fromJust' . Stream.filter 'isJust' . Stream.mapM f
930+-- @
931+--
932+-- /Concurrent (do not use with 'fromParallel' on infinite streams)/
933+--
934+-- @since 0.3.0
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
939+
940+-- | In a stream of 'Maybe's, discard 'Nothing's and unwrap 'Just's.
941+--
942+-- /Pre-release/
943+--
944+{-# INLINE catMaybes #-}
945+catMaybes :: (Monad m) => Stream m (Maybe a) -> Stream m a
946+catMaybes = fmap fromJust . filter isJust
947+
948+------------------------------------------------------------------------------
949+-- Either streams
950+------------------------------------------------------------------------------
951+
952+-- | Discard 'Right's and unwrap 'Left's in an 'Either' stream.
953+--
954+-- /Pre-release/
955+--
956+{-# INLINE lefts #-}
957+lefts :: (Monad m) => Stream m (Either a b) -> Stream m a
958+lefts = fmap (fromLeft undefined) . filter isLeft
959+
960+-- | Discard 'Left's and unwrap 'Right's in an 'Either' stream.
961+--
962+-- /Pre-release/
963+--
964+{-# INLINE rights #-}
965+rights :: (Monad m) => Stream m (Either a b) -> Stream m b
966+rights = fmap (fromRight undefined) . filter isRight
967+
968+-- | Remove the either wrapper and flatten both lefts and as well as rights in
969+-- the output stream.
970+--
971+-- /Pre-release/
972+--
973+{-# INLINE both #-}
974+both :: Monad m => Stream m (Either a a) -> Stream m a
975+both = fmap (either id id)