@@ -27,14 +27,16 @@ module Streamly.Internal.Data.Stream.StreamD.Transform
2727 -- * Mapping Effects
2828 , tap
2929 , tapOffsetEvery
30+ , trace
31+ , trace_
3032
3133 -- * Folding
3234 , foldrS
3335 , foldlS
3436
3537 -- * Scanning By 'Fold'
36− , postscanOnce -- XXX rename to postscan
37− , scanOnce -- XXX rename to scan
38+ , postscan
39+ , scan
3840 , scanMany
3941
4042 -- * Scanning
@@ -65,44 +67,67 @@ module Streamly.Internal.Data.Stream.StreamD.Transform
6567
6668 -- * Filtering
6769 -- | Produce a subset of the stream.
70+ , with
6871 , scanMaybe
6972 , filter
7073 , filterM
7174 , deleteBy
75+ , uniqBy
7276 , uniq
77+ , prune
78+ , repeated
7379
7480 -- * Trimming
7581 -- | Produce a subset of the stream trimmed at ends.
7682 , take
7783 , takeWhile
7884 , takeWhileM
85+ , takeWhileLast
86+ , takeWhileAround
7987 , drop
8088 , dropWhile
8189 , dropWhileM
90+ , dropLast
91+ , dropWhileLast
92+ , dropWhileAround
8293
8394 -- * Inserting Elements
8495 -- | Produce a superset of the stream.
8596 , insertBy
8697 , intersperse
8798 , intersperseM
99+ , intersperseMWith
88100 , intersperseMSuffix
89101 , intersperseMSuffixWith
90102
91103 -- * Inserting Side Effects
92104 , intersperseM_
93105 , intersperseMSuffix_
106+ , intersperseMPrefix_
107+
108+ , delay
109+ , delayPre
110+ , delayPost
94111
95112 -- * Reordering
96113 -- | Produce strictly the same set but reordered.
97114 , reverse
98− -- , reverse'
115+ , reverseUnbox
116+ , reassembleBy
99117
100118 -- * Position Indexing
101119 , indexed
102120 , indexedR
103121
122+ -- * Time Indexing
123+ , timestampWith
124+ , timestamped
125+ , timeIndexWith
126+ , timeIndexed
127+
104128 -- * Searching
105129 , findIndices
130+ , elemIndices
106131 , slicesBy
107132
108133 -- * Rolling map
@@ -115,32 +140,73 @@ module Streamly.Internal.Data.Stream.StreamD.Transform
115140 , mapMaybe
116141 , mapMaybeM
117142 , catMaybes
143+
144+ -- * Either Streams
145+ , catLefts
146+ , catRights
147+ , catEithers
118148 )
119149where
120150
121151#include "inline.hs"
122152
153+import Control.Concurrent (threadDelay)
123154import Control.Monad (void)
155+import Control.Monad.IO.Class (MonadIO (liftIO))
156+import Data.Either (fromLeft, isLeft, isRight, fromRight)
157+import Data.Functor ((<&>))
124158import Data.Maybe (fromJust, isJust)
125159import Fusion.Plugin.Types (Fuse(..))
126160
127161import Streamly.Internal.Data.Fold.Type (Fold(..))
128162import Streamly.Internal.Data.Pipe.Type (Pipe(..), PipeState(..))
129163import Streamly.Internal.Data.SVar.Type (adaptState)
164+import Streamly.Internal.Data.Time.Units (AbsTime, RelTime64)
165+import Streamly.Internal.Data.Unboxed (Unbox)
166+import Streamly.Internal.System.IO (defaultChunkSize)
130167
131−import qualified Streamly.Internal.Data.Fold.Type as FL
168+-- import qualified Data.List as List
169+import qualified Streamly.Internal.Data.Array.Type as A
170+import qualified Streamly.Internal.Data.Fold as FL
132171import qualified Streamly.Internal.Data.Pipe.Type as Pipe
172+import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
133173
134174import Prelude hiding
135175 ( drop, dropWhile, filter, map, mapM, reverse
136− , scanl, scanl1, sequence, take, takeWhile)
176+ , scanl, scanl1, sequence, take, takeWhile, zipWith)
137177
178+import Streamly.Internal.Data.Stream.StreamD.Generate
179+ (absTimesWith, relTimesWith)
138180import Streamly.Internal.Data.Stream.StreamD.Type
139181
182+--
183+-- $setup
184+-- >>> :m
185+-- >>> import Control.Concurrent (threadDelay)
186+-- >>> import Control.Monad (void)
187+-- >>> import Control.Monad.IO.Class (MonadIO (liftIO))
188+-- >>> import Data.Either (fromLeft, fromRight, isLeft, isRight, either)
189+-- >>> import Data.Function ((&))
190+-- >>> import Data.Maybe (fromJust, isJust)
191+-- >>> import Prelude hiding (filter, drop, dropWhile, take, takeWhile, foldr, map, mapM, sequence, reverse, foldr1 , scanl, scanl1)
192+-- >>> import Streamly.Internal.Data.Stream (Stream)
193+-- >>> import qualified Streamly.Data.Fold as Fold
194+-- >>> import qualified Streamly.Data.Unfold as Unfold
195+-- >>> import qualified Streamly.Internal.Data.Fold as Fold (filtering)
196+-- >>> import qualified Streamly.Internal.Data.Fold.Window as Window
197+-- >>> import qualified Streamly.Internal.Data.Stream as Stream
198+-- >>> import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
199+--
200+-- >>> hSetBuffering stdout LineBuffering
201+
140202------------------------------------------------------------------------------
141203-- Piping
142204------------------------------------------------------------------------------
143205
206+-- | Use a 'Pipe' to transform a stream.
207+--
208+-- /Pre-release/
209+--
144210{-# INLINE_NORMAL transform #-}
145211transform :: Monad m => Pipe m a b -> Stream m a -> Stream m b
146212transform (Pipe pstep1 pstep2 pstate) (Stream step state) =
@@ -197,6 +263,16 @@ foldlS fstep begin (Stream step state) = Stream step' (Left (state, begin))
197263-- Transformation by Mapping
198264------------------------------------------------------------------------------
199265
266+-- |
267+-- >>> sequence = Stream.mapM id
268+--
269+-- Replace the elements of a stream of monadic actions with the outputs of
270+-- those actions.
271+--
272+-- >>> s = Stream.fromList [putStr "a", putStr "b", putStrLn "c"]
273+-- >>> Stream.fold Fold.drain $ Stream.sequence s
274+-- abc
275+--
200276{-# INLINE_NORMAL sequence #-}
201277sequence :: Monad m => Stream m (m a) -> Stream m a
202278sequence (Stream step state) = Stream step' state
@@ -217,6 +293,25 @@ data TapState fs st a
217293 = TapInit | Tapping !fs st | TapDone st
218294
219295-- XXX Multiple yield points
296+
297+-- | Tap the data flowing through a stream into a 'Fold'. For example, you may
298+-- add a tap to log the contents flowing through the stream. The fold is used
299+-- only for effects, its result is discarded.
300+--
301+-- @
302+-- Fold m a b
303+-- |
304+-- -----stream m a ---------------stream m a-----
305+--
306+-- @
307+--
308+-- >>> s = Stream.enumerateFromTo 1 2
309+-- >>> Stream.fold Fold.drain $ Stream.tap (Fold.drainMapM print) s
310+-- 1
311+-- 2
312+--
313+-- Compare with 'trace'.
314+--
220315{-# INLINE tap #-}
221316tap :: Monad m => Fold m a b -> Stream m a -> Stream m a
222317tap (Fold fstep initial extract) (Stream step state) = Stream step' TapInit
@@ -301,15 +396,62 @@ tapOffsetEvery offset n (Fold fstep initial extract) (Stream step state) =
301396 Skip s -> Skip (TapOffDone s)
302397 Stop -> Stop
303398
399+-- | Apply a monadic function to each element flowing through the stream and
400+-- discard the results.
401+--
402+-- >>> s = Stream.enumerateFromTo 1 2
403+-- >>> Stream.fold Fold.drain $ Stream.trace print s
404+-- 1
405+-- 2
406+--
407+-- Compare with 'tap'.
408+--
409+{-# INLINE trace #-}
410+trace :: Monad m => (a -> m b) -> Stream m a -> Stream m a
411+trace f = mapM (\x -> void (f x) >> return x)
412+
413+-- | Perform a side effect before yielding each element of the stream and
414+-- discard the results.
415+--
416+-- >>> s = Stream.enumerateFromTo 1 2
417+-- >>> Stream.fold Fold.drain $ Stream.trace_ (print "got here") s
418+-- "got here"
419+-- "got here"
420+--
421+-- Same as 'intersperseMPrefix_' but always serial.
422+--
423+-- See also: 'trace'
424+--
425+-- /Pre-release/
426+{-# INLINE trace_ #-}
427+trace_ :: Monad m => m b -> Stream m a -> Stream m a
428+trace_ eff = mapM (\x -> eff >> return x)
429+
304430------------------------------------------------------------------------------
305431-- Scanning with a Fold
306432------------------------------------------------------------------------------
307433
308434data ScanState s f = ScanInit s | ScanDo s !f | ScanDone
309435
310−{-# INLINE_NORMAL postscanOnce #-}
311−postscanOnce :: Monad m => FL.Fold m a b -> Stream m a -> Stream m b
312−postscanOnce (FL.Fold fstep initial extract) (Stream sstep state) =
436+-- | Postscan a stream using the given monadic fold.
437+--
438+-- The following example extracts the input stream up to a point where the
439+-- running average of elements is no more than 10:
440+--
441+-- >>> import Data.Maybe (fromJust)
442+-- >>> let avg = Fold.teeWith (/) Fold.sum (fmap fromIntegral Fold.length)
443+-- >>> s = Stream.enumerateFromTo 1.0 100.0
444+-- >>> :{
445+-- Stream.fold Fold.toList
446+-- $ fmap (fromJust . fst)
447+-- $ Stream.takeWhile (\(_,x) -> x <= 10)
448+-- $ Stream.postscan (Fold.tee Fold.latest avg) s
449+-- :}
450+-- [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]
451+--
452+{-# INLINE_NORMAL postscan #-}
453+postscan :: Monad m => FL.Fold m a b -> Stream m a -> Stream m b
454+postscan (FL.Fold fstep initial extract) (Stream sstep state) =
313455 Stream step (ScanInit state)
314456
315457 where
@@ -364,11 +506,75 @@ scanWith restart (Fold fstep initial extract) (Stream sstep state) =
364506 Stop -> return Stop
365507 step _ ScanDone = return Stop
366508
367−{-# INLINE scanOnce #-}
368−scanOnce :: Monad m
509+-- XXX It may be useful to have a version of scan where we can keep the
510+-- accumulator independent of the value emitted. So that we do not necessarily
511+-- have to keep a value in the accumulator which we are not using. We can pass
512+-- an extraction function that will take the accumulator and the current value
513+-- of the element and emit the next value in the stream. That will also make it
514+-- possible to modify the accumulator after using it. In fact, the step function
515+-- can return new accumulator and the value to be emitted. The signature would
516+-- be more like mapAccumL.
517+
518+-- | Strict left scan. Scan a stream using the given monadic fold.
519+--
520+-- >>> s = Stream.fromList [1..10]
521+-- >>> Stream.fold Fold.toList $ Stream.takeWhile (< 10) $ Stream.scan Fold.sum s
522+-- [0,1,3,6]
523+--
524+-- See also: 'usingStateT'
525+--
526+
527+-- EXPLANATION:
528+-- >>> scanl' step z = Stream.scan (Fold.foldl' step z)
529+--
530+-- Like 'map', 'scanl'' too is a one to one transformation,
531+-- however it adds an extra element.
532+--
533+-- >>> s = Stream.fromList [1,2,3,4]
534+-- >>> Stream.fold Fold.toList $ scanl' (+) 0 s
535+-- [0,1,3,6,10]
536+--
537+-- >>> Stream.fold Fold.toList $ scanl' (flip (:)) [] s
538+-- [[],[1],[2,1],[3,2,1],[4,3,2,1]]
539+--
540+-- The output of 'scanl'' is the initial value of the accumulator followed by
541+-- all the intermediate steps and the final result of 'foldl''.
542+--
543+-- By streaming the accumulated state after each fold step, we can share the
544+-- state across multiple stages of stream composition. Each stage can modify or
545+-- extend the state, do some processing with it and emit it for the next stage,
546+-- thus modularizing the stream processing. This can be useful in
547+-- stateful or event-driven programming.
548+--
549+-- Consider the following monolithic example, computing the sum and the product
550+-- of the elements in a stream in one go using a @foldl'@:
551+--
552+-- >>> foldl' step z = Stream.fold (Fold.foldl' step z)
553+-- >>> foldl' (\(s, p) x -> (s + x, p * x)) (0,1) s
554+-- (10,24)
555+--
556+-- Using @scanl'@ we can make it modular by computing the sum in the first
557+-- stage and passing it down to the next stage for computing the product:
558+--
559+-- >>> :{
560+-- foldl' (\(_, p) (s, x) -> (s, p * x)) (0,1)
561+-- $ scanl' (\(s, _) x -> (s + x, x)) (0,1)
562+-- $ Stream.fromList [1,2,3,4]
563+-- :}
564+-- (10,24)
565+--
566+-- IMPORTANT: 'scanl'' evaluates the accumulator to WHNF. To avoid building
567+-- lazy expressions inside the accumulator, it is recommended that a strict
568+-- data structure is used for accumulator.
569+--
570+{-# INLINE scan #-}
571+scan :: Monad m
369572 => FL.Fold m a b -> Stream m a -> Stream m b
370−scanOnce = scanWith False
573+scan = scanWith False
371574
575+-- | Like 'scan' but restarts scanning afresh when the scanning fold
576+-- terminates.
577+--
372578{-# INLINE scanMany #-}
373579scanMany :: Monad m
374580 => FL.Fold m a b -> Stream m a -> Stream m b
@@ -557,6 +763,17 @@ scanlM' fstep begin (Stream step state) = Stream step' Nothing
557763 Skip s -> return $ Skip (Just (s, acc))
558764 Stop -> return Stop
559765
766+-- | @scanlMAfter' accumulate initial done stream@ is like 'scanlM'' except
767+-- that it provides an additional @done@ function to be applied on the
768+-- accumulator when the stream stops. The result of @done@ is also emitted in
769+-- the stream.
770+--
771+-- This function can be used to allocate a resource in the beginning of the
772+-- scan and release it when the stream ends or to flush the internal state of
773+-- the scan at the end.
774+--
775+-- /Pre-release/
776+--
560777{-# INLINE scanlMAfter' #-}
561778scanlMAfter' :: Monad m
562779 => (b -> a -> m b) -> m b -> (b -> m b) -> Stream m a -> Stream m b
@@ -644,7 +861,28 @@ scanl1' f = scanl1M' (\x y -> return (f x y))
644861-- Filtering
645862-------------------------------------------------------------------------------
646863
864+-- | Modify a @Stream m a -> Stream m a@ stream transformation that accepts a
865+-- predicate @(a -> b)@ to accept @((s, a) -> b)@ instead, provided a
866+-- transformation @Stream m a -> Stream m (s, a)@. Convenient to filter with
867+-- index or time.
868+--
869+-- >>> filterWithIndex = Stream.with Stream.indexed Stream.filter
870+--
871+-- /Pre-release/
872+{-# INLINE with #-}
873+with :: Monad m =>
874+ (Stream m a -> Stream m (s, a))
875+ -> (((s, a) -> b) -> Stream m (s, a) -> Stream m (s, a))
876+ -> (((s, a) -> b) -> Stream m a -> Stream m a)
877+with f comb g = fmap snd . comb g . f
878+
647879-- Adapted from the vector package
880+
881+-- | Same as 'filter' but with a monadic predicate.
882+--
883+-- >>> f p x = p x >>= \r -> return $ if r then Just x else Nothing
884+-- >>> filterM p = Stream.mapMaybeM (f p)
885+--
648886{-# INLINE_NORMAL filterM #-}
649887filterM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
650888filterM f (Stream step state) = Stream step' state
@@ -661,13 +899,55 @@ filterM f (Stream step state) = Stream step' state
661899 Skip s -> return $ Skip s
662900 Stop -> return Stop
663901
902+-- | Include only those elements that pass a predicate.
903+--
904+-- >>> filter p = Stream.filterM (return . p)
905+-- >>> filter p = Stream.mapMaybe (\x -> if p x then Just x else Nothing)
906+-- >>> filter p = Stream.scanMaybe (Fold.filtering p)
907+--
664908{-# INLINE filter #-}
665909filter :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
666910filter f = filterM (return . f)
911+-- filter p = scanMaybe (FL.filtering p)
912+
913+-- | Drop repeated elements that are adjacent to each other using the supplied
914+-- comparison function.
915+--
916+-- >>> uniq = Stream.uniqBy (==)
917+--
918+-- To strip duplicate path separators:
919+--
920+-- >>> input = Stream.fromList "//a//b"
921+-- >>> f x y = x == '/' && y == '/'
922+-- >>> Stream.fold Fold.toList $ Stream.uniqBy f input
923+-- "/a/b"
924+--
925+-- Space: @O(1)@
926+--
927+-- /Pre-release/
928+--
929+{-# INLINE uniqBy #-}
930+uniqBy :: Monad m =>
931+ (a -> a -> Bool) -> Stream m a -> Stream m a
932+-- uniqBy eq = scanMaybe (FL.uniqBy eq)
933+uniqBy eq = catMaybes . rollingMap f
934+
935+ where
936+
937+ f pre curr =
938+ case pre of
939+ Nothing -> Just curr
940+ Just x -> if x `eq` curr then Nothing else Just curr
667941
668942-- Adapted from the vector package
943+
944+-- | Drop repeated elements that are adjacent to each other.
945+--
946+-- >>> uniq = Stream.uniqBy (==)
947+--
669948{-# INLINE_NORMAL uniq #-}
670949uniq :: (Eq a, Monad m) => Stream m a -> Stream m a
950+-- uniq = scanMaybe FL.uniq
671951uniq (Stream step state) = Stream step' (Nothing, state)
672952 where
673953 {-# INLINE_LATE step' #-}
@@ -685,8 +965,16 @@ uniq (Stream step state) = Stream step' (Nothing, state)
685965 Skip s -> return $ Skip (Just x, s)
686966 Stop -> return Stop
687967
968+-- | Deletes the first occurrence of the element in the stream that satisfies
969+-- the given equality predicate.
970+--
971+-- >>> input = Stream.fromList [1,3,3,5]
972+-- >>> Stream.fold Fold.toList $ Stream.deleteBy (==) 3 input
973+-- [1,3,5]
974+--
688975{-# INLINE_NORMAL deleteBy #-}
689976deleteBy :: Monad m => (a -> a -> Bool) -> a -> Stream m a -> Stream m a
977+-- deleteBy cmp x = scanMaybe (FL.deleteBy cmp x)
690978deleteBy eq x (Stream step state) = Stream step' (state, False)
691979 where
692980 {-# INLINE_LATE step' #-}
@@ -705,11 +993,69 @@ deleteBy eq x (Stream step state) = Stream step' (state, False)
705993 Skip s -> return $ Skip (s, True)
706994 Stop -> return Stop
707995
996+-- | Strip all leading and trailing occurrences of an element passing a
997+-- predicate and make all other consecutive occurrences uniq.
998+--
999+-- >> prune p = Stream.dropWhileAround p $ Stream.uniqBy (x y -> p x && p y)
1000+--
1001+-- @
1002+-- > Stream.prune isSpace (Stream.fromList " hello world! ")
1003+-- "hello world!"
1004+--
1005+-- @
1006+--
1007+-- Space: @O(1)@
1008+--
1009+-- /Unimplemented/
1010+{-# INLINE prune #-}
1011+prune ::
1012+ -- (Monad m, Eq a) =>
1013+ (a -> Bool) -> Stream m a -> Stream m a
1014+prune = error "Not implemented yet!"
1015+
1016+-- Possible implementation:
1017+-- @repeated =
1018+-- Stream.catMaybes . Stream.parseMany (Parser.groupBy (==) Fold.repeated)@
1019+--
1020+-- 'Fold.repeated' should return 'Just' when repeated, and 'Nothing' for a
1021+-- single element.
1022+
1023+-- | Emit only repeated elements, once.
1024+--
1025+-- /Unimplemented/
1026+repeated :: -- (Monad m, Eq a) =>
1027+ Stream m a -> Stream m a
1028+repeated = undefined
1029+
7081030------------------------------------------------------------------------------
7091031-- Trimming
7101032------------------------------------------------------------------------------
7111033
1034+-- | Take all consecutive elements at the end of the stream for which the
1035+-- predicate is true.
1036+--
1037+-- O(n) space, where n is the number elements taken.
1038+--
1039+-- /Unimplemented/
1040+{-# INLINE takeWhileLast #-}
1041+takeWhileLast :: -- Monad m =>
1042+ (a -> Bool) -> Stream m a -> Stream m a
1043+takeWhileLast = undefined -- fromStreamD $ D.takeWhileLast n $ toStreamD m
1044+
1045+-- | Like 'takeWhile' and 'takeWhileLast' combined.
1046+--
1047+-- O(n) space, where n is the number elements taken from the end.
1048+--
1049+-- /Unimplemented/
1050+{-# INLINE takeWhileAround #-}
1051+takeWhileAround :: -- Monad m =>
1052+ (a -> Bool) -> Stream m a -> Stream m a
1053+takeWhileAround = undefined -- fromStreamD $ D.takeWhileAround n $ toStreamD m
1054+
7121055-- Adapted from the vector package
1056+
1057+-- | Discard first 'n' elements from the stream and take the rest.
1058+--
7131059{-# INLINE_NORMAL drop #-}
7141060drop :: Monad m => Int -> Stream m a -> Stream m a
7151061drop n (Stream step state) = Stream step' (state, Just n)
@@ -739,8 +1085,11 @@ data DropWhileState s a
7391085 | DropWhileYield a s
7401086 | DropWhileNext s
7411087
1088+-- | Same as 'dropWhile' but with a monadic predicate.
1089+--
7421090{-# INLINE_NORMAL dropWhileM #-}
7431091dropWhileM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
1092+-- dropWhileM p = scanMaybe (FL.droppingWhileM p)
7441093dropWhileM f (Stream step state) = Stream step' (DropWhileDrop state)
7451094 where
7461095 {-# INLINE_LATE step' #-}
@@ -764,14 +1113,58 @@ dropWhileM f (Stream step state) = Stream step' (DropWhileDrop state)
7641113
7651114 step' _ (DropWhileYield x st) = return $ Yield x (DropWhileNext st)
7661115
1116+-- | Drop elements in the stream as long as the predicate succeeds and then
1117+-- take the rest of the stream.
1118+--
7671119{-# INLINE dropWhile #-}
7681120dropWhile :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
1121+-- dropWhile p = scanMaybe (FL.droppingWhile p)
7691122dropWhile f = dropWhileM (return . f)
7701123
1124+-- | Drop @n@ elements at the end of the stream.
1125+--
1126+-- O(n) space, where n is the number elements dropped.
1127+--
1128+-- /Unimplemented/
1129+{-# INLINE dropLast #-}
1130+dropLast :: -- Monad m =>
1131+ Int -> Stream m a -> Stream m a
1132+dropLast = undefined -- fromStreamD $ D.dropLast n $ toStreamD m
1133+
1134+-- | Drop all consecutive elements at the end of the stream for which the
1135+-- predicate is true.
1136+--
1137+-- O(n) space, where n is the number elements dropped.
1138+--
1139+-- /Unimplemented/
1140+{-# INLINE dropWhileLast #-}
1141+dropWhileLast :: -- Monad m =>
1142+ (a -> Bool) -> Stream m a -> Stream m a
1143+dropWhileLast = undefined -- fromStreamD $ D.dropWhileLast n $ toStreamD m
1144+
1145+-- | Like 'dropWhile' and 'dropWhileLast' combined.
1146+--
1147+-- O(n) space, where n is the number elements dropped from the end.
1148+--
1149+-- /Unimplemented/
1150+{-# INLINE dropWhileAround #-}
1151+dropWhileAround :: -- Monad m =>
1152+ (a -> Bool) -> Stream m a -> Stream m a
1153+dropWhileAround = undefined -- fromStreamD $ D.dropWhileAround n $ toStreamD m
1154+
7711155------------------------------------------------------------------------------
7721156-- Inserting Elements
7731157------------------------------------------------------------------------------
7741158
1159+-- | @insertBy cmp elem stream@ inserts @elem@ before the first element in
1160+-- @stream@ that is less than @elem@ when compared using @cmp@.
1161+--
1162+-- >>> insertBy cmp x = Stream.mergeBy cmp (Stream.fromPure x)
1163+--
1164+-- >>> input = Stream.fromList [1,3,5]
1165+-- >>> Stream.fold Fold.toList $ Stream.insertBy compare 2 input
1166+-- [1,2,3,5]
1167+--
7751168{-# INLINE_NORMAL insertBy #-}
7761169insertBy :: Monad m => (a -> a -> Ordering) -> a -> Stream m a -> Stream m a
7771170insertBy cmp a (Stream step state) = Stream step' (state, False, Nothing)
@@ -799,6 +1192,22 @@ data LoopState x s = FirstYield s
7991192 | InterspersingYield s
8001193 | YieldAndCarry x s
8011194
1195+-- intersperseM = intersperseMWith 1
1196+
1197+-- | Insert an effect and its output before consuming an element of a stream
1198+-- except the first one.
1199+--
1200+-- >>> input = Stream.fromList "hello"
1201+-- >>> Stream.fold Fold.toList $ Stream.trace putChar $ Stream.intersperseM (putChar '.' >> return ',') input
1202+-- h.,e.,l.,l.,o"h,e,l,l,o"
1203+--
1204+-- Be careful about the order of effects. In the above example we used trace
1205+-- after the intersperse, if we use it before the intersperse the output would
1206+-- be he.l.l.o."h,e,l,l,o".
1207+--
1208+-- >>> Stream.fold Fold.toList $ Stream.intersperseM (putChar '.' >> return ',') $ Stream.trace putChar input
1209+-- he.l.l.o."h,e,l,l,o"
1210+--
8021211{-# INLINE_NORMAL intersperseM #-}
8031212intersperseM :: Monad m => m a -> Stream m a -> Stream m a
8041213intersperseM m (Stream step state) = Stream step' (FirstYield state)
@@ -823,10 +1232,24 @@ intersperseM m (Stream step state) = Stream step' (FirstYield state)
8231232
8241233 step' _ (YieldAndCarry x st) = return $ Yield x (InterspersingYield st)
8251234
1235+-- | Insert a pure value between successive elements of a stream.
1236+--
1237+-- >>> input = Stream.fromList "hello"
1238+-- >>> Stream.fold Fold.toList $ Stream.intersperse ',' input
1239+-- "h,e,l,l,o"
1240+--
8261241{-# INLINE intersperse #-}
8271242intersperse :: Monad m => a -> Stream m a -> Stream m a
8281243intersperse a = intersperseM (return a)
8291244
1245+-- | Insert a side effect before consuming an element of a stream except the
1246+-- first one.
1247+--
1248+-- >>> input = Stream.fromList "hello"
1249+-- >>> Stream.fold Fold.drain $ Stream.trace putChar $ Stream.intersperseM_ (putChar '.') input
1250+-- h.e.l.l.o
1251+--
1252+-- /Pre-release/
8301253{-# INLINE_NORMAL intersperseM_ #-}
8311254intersperseM_ :: Monad m => m b -> Stream m a -> Stream m a
8321255intersperseM_ m (Stream step1 state1) = Stream step (Left (pure (), state1))
@@ -841,11 +1264,31 @@ intersperseM_ m (Stream step1 state1) = Stream step (Left (pure (), state1))
8411264
8421265 step _ (Right st) = return $ Skip $ Left (void m, st)
8431266
1267+-- | Intersperse a monadic action into the input stream after every @n@
1268+-- elements.
1269+--
1270+-- >> input = Stream.fromList "hello"
1271+-- >> Stream.fold Fold.toList $ Stream.intersperseMWith 2 (return ',') input
1272+-- "he,ll,o"
1273+--
1274+-- /Unimplemented/
1275+{-# INLINE intersperseMWith #-}
1276+intersperseMWith :: -- Monad m =>
1277+ Int -> m a -> Stream m a -> Stream m a
1278+intersperseMWith _n _f _xs = undefined
1279+
8441280data SuffixState s a
8451281 = SuffixElem s
8461282 | SuffixSuffix s
8471283 | SuffixYield a (SuffixState s a)
8481284
1285+-- | Insert an effect and its output after consuming an element of a stream.
1286+--
1287+-- >>> input = Stream.fromList "hello"
1288+-- >>> Stream.fold Fold.toList $ Stream.trace putChar $ Stream.intersperseMSuffix (putChar '.' >> return ',') input
1289+-- h.,e.,l.,l.,o.,"h,e,l,l,o,"
1290+--
1291+-- /Pre-release/
8491292{-# INLINE_NORMAL intersperseMSuffix #-}
8501293intersperseMSuffix :: forall m a. Monad m => m a -> Stream m a -> Stream m a
8511294intersperseMSuffix action (Stream step state) = Stream step' (SuffixElem state)
@@ -863,6 +1306,14 @@ intersperseMSuffix action (Stream step state) = Stream step' (SuffixElem state)
8631306
8641307 step' _ (SuffixYield x next) = return $ Yield x next
8651308
1309+-- | Insert a side effect after consuming an element of a stream.
1310+--
1311+-- >>> input = Stream.fromList "hello"
1312+-- >>> Stream.fold Fold.toList $ Stream.intersperseMSuffix_ (threadDelay 1000000) input
1313+-- "hello"
1314+--
1315+-- /Pre-release/
1316+--
8661317{-# INLINE_NORMAL intersperseMSuffix_ #-}
8671318intersperseMSuffix_ :: Monad m => m b -> Stream m a -> Stream m a
8681319intersperseMSuffix_ m (Stream step1 state1) = Stream step (Left state1)
@@ -884,7 +1335,15 @@ data SuffixSpanState s a
8841335 | SuffixSpanLast
8851336 | SuffixSpanStop
8861337
887−-- | intersperse after every n items
1338+-- | Like 'intersperseMSuffix' but intersperses an effectful action into the
1339+-- input stream after every @n@ elements and after the last element.
1340+--
1341+-- >>> input = Stream.fromList "hello"
1342+-- >>> Stream.fold Fold.toList $ Stream.intersperseMSuffixWith 2 (return ',') input
1343+-- "he,ll,o,"
1344+--
1345+-- /Pre-release/
1346+--
8881347{-# INLINE_NORMAL intersperseMSuffixWith #-}
8891348intersperseMSuffixWith :: forall m a. Monad m
8901349 => Int -> m a -> Stream m a -> Stream m a
@@ -910,21 +1369,113 @@ intersperseMSuffixWith n action (Stream step state) =
9101369
9111370 step' _ SuffixSpanStop = return Stop
9121371
1372+-- | Insert a side effect before consuming an element of a stream.
1373+--
1374+-- Definition:
1375+--
1376+-- >>> intersperseMPrefix_ m = Stream.mapM (\x -> void m >> return x)
1377+--
1378+-- >>> input = Stream.fromList "hello"
1379+-- >>> Stream.fold Fold.toList $ Stream.trace putChar $ Stream.intersperseMPrefix_ (putChar '.' >> return ',') input
1380+-- .h.e.l.l.o"hello"
1381+--
1382+-- Same as 'trace_'.
1383+--
1384+-- /Pre-release/
1385+--
1386+{-# INLINE intersperseMPrefix_ #-}
1387+intersperseMPrefix_ :: Monad m => m b -> Stream m a -> Stream m a
1388+intersperseMPrefix_ m = mapM (\x -> void m >> return x)
1389+
1390+------------------------------------------------------------------------------
1391+-- Inserting Time
1392+------------------------------------------------------------------------------
1393+
1394+-- XXX This should be in Prelude, should we export this as a helper function?
1395+
1396+-- | Block the current thread for specified number of seconds.
1397+{-# INLINE sleep #-}
1398+sleep :: MonadIO m => Double -> m ()
1399+sleep n = liftIO $ threadDelay $ round $ n * 1000000
1400+
1401+-- | Introduce a delay of specified seconds between elements of the stream.
1402+--
1403+-- Definition:
1404+--
1405+-- >>> sleep n = liftIO $ threadDelay $ round $ n * 1000000
1406+-- >>> delay = Stream.intersperseM_ . sleep
1407+--
1408+-- Example:
1409+--
1410+-- >>> input = Stream.enumerateFromTo 1 3
1411+-- >>> Stream.fold (Fold.drainMapM print) $ Stream.delay 1 input
1412+-- 1
1413+-- 2
1414+-- 3
1415+--
1416+{-# INLINE delay #-}
1417+delay :: MonadIO m => Double -> Stream m a -> Stream m a
1418+delay = intersperseM_ . sleep
1419+
1420+-- | Introduce a delay of specified seconds after consuming an element of a
1421+-- stream.
1422+--
1423+-- Definition:
1424+--
1425+-- >>> sleep n = liftIO $ threadDelay $ round $ n * 1000000
1426+-- >>> delayPost = Stream.intersperseMSuffix_ . sleep
1427+--
1428+-- Example:
1429+--
1430+-- >>> input = Stream.enumerateFromTo 1 3
1431+-- >>> Stream.fold (Fold.drainMapM print) $ Stream.delayPost 1 input
1432+-- 1
1433+-- 2
1434+-- 3
1435+--
1436+-- /Pre-release/
1437+--
1438+{-# INLINE delayPost #-}
1439+delayPost :: MonadIO m => Double -> Stream m a -> Stream m a
1440+delayPost n = intersperseMSuffix_ $ liftIO $ threadDelay $ round $ n * 1000000
1441+
1442+-- | Introduce a delay of specified seconds before consuming an element of a
1443+-- stream.
1444+--
1445+-- Definition:
1446+--
1447+-- >>> sleep n = liftIO $ threadDelay $ round $ n * 1000000
1448+-- >>> delayPre = Stream.intersperseMPrefix_. sleep
1449+--
1450+-- Example:
1451+--
1452+-- >>> input = Stream.enumerateFromTo 1 3
1453+-- >>> Stream.fold (Fold.drainMapM print) $ Stream.delayPre 1 input
1454+-- 1
1455+-- 2
1456+-- 3
1457+--
1458+-- /Pre-release/
1459+--
1460+{-# INLINE delayPre #-}
1461+delayPre :: MonadIO m => Double -> Stream m a -> Stream m a
1462+delayPre = intersperseMPrefix_. sleep
1463+
9131464------------------------------------------------------------------------------
9141465-- Reordering
9151466------------------------------------------------------------------------------
9161467
917−-- We can implement reverse as:
1468+-- | Returns the elements of the stream in reverse order. The stream must be
1469+-- finite. Note that this necessarily buffers the entire stream in memory.
9181470--
919−-- > reverse = foldlS (flip cons) nil
1471+-- Definition:
9201472--
921−-- However, this implementation is unusable because of the horrible performance
922−-- of cons. So we just convert it to a list first and then stream from the
923−-- list.
1473+-- >>> reverse m = Stream.concatEffect $ Stream.fold Fold.toListRev m >>= return . Stream.fromList
9241474--
925−-- XXX Maybe we can use an Array instead of a list here?
9261475{-# INLINE_NORMAL reverse #-}
9271476reverse :: Monad m => Stream m a -> Stream m a
1477+reverse m = concatEffect $ fold FL.toListRev m <&> fromList
1478+{-
9281479reverse m = Stream step Nothing
9291480 where
9301481 {-# INLINE_LATE step #-}
@@ -933,47 +1484,57 @@ reverse m = Stream step Nothing
9331484 return $ Skip (Just xs)
9341485 step _ (Just (x:xs)) = return $ Yield x (Just xs)
9351486 step _ (Just []) = return Stop
936−
937−-- Much faster reverse for Storables
938−{-
939−{-# INLINE_NORMAL reverse' #-}
940−reverse' :: forall m a. (MonadIO m, Storable a) => Stream m a -> Stream m a
941−-- This commented implementation copies the whole stream into one single array
942−-- and then streams from that array, this has exactly the same performance as
943−-- the chunked code in IsStream.Common.reverse' . Though this could be problematic due to
944−-- unbounded large allocations. However, if we use an idiomatic implementation
945−-- of arraysOf instead of the custom implementation then the chunked code
946−-- becomes worse by 6 times. Need to investigate if that can be improved.
947−import Foreign.ForeignPtr (touchForeignPtr)
948−import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
949−import Foreign.Ptr (Ptr, plusPtr)
950−import Streamly.Internal.Data.Array.Mut.Type (sizeOfElem)
951−reverse' m = Stream step Nothing
952− where
953− {-# INLINE_LATE step #-}
954− step _ Nothing = do
955− arr <- A.fromStreamD m
956− let p = A.arrEnd arr `plusPtr` negate (sizeOfElem (undefined :: a))
957− return $ Skip $ Just (A.aStart arr, p)
958−
959− step _ (Just (start, p)) | p < unsafeForeignPtrToPtr start = return Stop
960−
961− step _ (Just (start, p)) = do
962− let !x = A.unsafeInlineIO $ do
963− r <- peek p
964− touchForeignPtr start
965− return r
966− next = p `plusPtr` negate (sizeOfElem (undefined :: a))
967− return $ Yield x (Just (start, next))
9681487-}
9691488
1489+-- | Like 'reverse' but several times faster, requires an 'Unbox' instance.
1490+--
1491+-- /O(n) space/
1492+--
1493+-- /Pre-release/
1494+{-# INLINE reverseUnbox #-}
1495+reverseUnbox :: (MonadIO m, Unbox a) => Stream m a -> Stream m a
1496+reverseUnbox =
1497+ A.flattenArraysRev -- unfoldMany A.readRev
1498+ . fromStreamK
1499+ . K.reverse
1500+ . toStreamK
1501+ . A.arraysOf defaultChunkSize
1502+
1503+-- | Buffer until the next element in sequence arrives. The function argument
1504+-- determines the difference in sequence numbers. This could be useful in
1505+-- implementing sequenced streams, for example, TCP reassembly.
1506+--
1507+-- /Unimplemented/
1508+--
1509+{-# INLINE reassembleBy #-}
1510+reassembleBy
1511+ :: -- Monad m =>
1512+ Fold m a b
1513+ -> (a -> a -> Int)
1514+ -> Stream m a
1515+ -> Stream m b
1516+reassembleBy = undefined
1517+
9701518------------------------------------------------------------------------------
9711519-- Position Indexing
9721520------------------------------------------------------------------------------
9731521
9741522-- Adapted from the vector package
1523+
1524+-- |
1525+-- >>> f = Fold.foldl' (\(i, _) x -> (i + 1, x)) (-1,undefined)
1526+-- >>> indexed = Stream.postscan f
1527+-- >>> indexed = Stream.zipWith (,) (Stream.enumerateFrom 0)
1528+-- >>> indexedR n = fmap (\(i, a) -> (n - i, a)) . indexed
1529+--
1530+-- Pair each element in a stream with its index, starting from index 0.
1531+--
1532+-- >>> Stream.fold Fold.toList $ Stream.indexed $ Stream.fromList "hello"
1533+-- [(0,'h'),(1,'e'),(2,'l'),(3,'l'),(4,'o')]
1534+--
9751535{-# INLINE_NORMAL indexed #-}
9761536indexed :: Monad m => Stream m a -> Stream m (Int, a)
1537+-- indexed = scanMaybe FL.indexing
9771538indexed (Stream step state) = Stream step' (state, 0)
9781539 where
9791540 {-# INLINE_LATE step' #-}
@@ -985,8 +1546,23 @@ indexed (Stream step state) = Stream step' (state, 0)
9851546 Stop -> return Stop
9861547
9871548-- Adapted from the vector package
1549+
1550+-- |
1551+-- >>> f n = Fold.foldl' (\(i, _) x -> (i - 1, x)) (n + 1,undefined)
1552+-- >>> indexedR n = Stream.postscan (f n)
1553+--
1554+-- >>> s n = Stream.enumerateFromThen n (n - 1)
1555+-- >>> indexedR n = Stream.zipWith (,) (s n)
1556+--
1557+-- Pair each element in a stream with its index, starting from the
1558+-- given index @n@ and counting down.
1559+--
1560+-- >>> Stream.fold Fold.toList $ Stream.indexedR 10 $ Stream.fromList "hello"
1561+-- [(10,'h'),(9,'e'),(8,'l'),(7,'l'),(6,'o')]
1562+--
9881563{-# INLINE_NORMAL indexedR #-}
9891564indexedR :: Monad m => Int -> Stream m a -> Stream m (Int, a)
1565+-- indexedR n = scanMaybe (FL.indexingRev n)
9901566indexedR m (Stream step state) = Stream step' (state, m)
9911567 where
9921568 {-# INLINE_LATE step' #-}
@@ -998,10 +1574,77 @@ indexedR m (Stream step state) = Stream step' (state, m)
9981574 Skip s -> return $ Skip (s, i)
9991575 Stop -> return Stop
10001576
1577+-------------------------------------------------------------------------------
1578+-- Time Indexing
1579+-------------------------------------------------------------------------------
1580+
1581+-- Note: The timestamp stream must be the second stream in the zip so that the
1582+-- timestamp is generated after generating the stream element and not before.
1583+-- If we do not do that then the following example will generate the same
1584+-- timestamp for first two elements:
1585+--
1586+-- Stream.fold Fold.toList $ Stream.timestamped $ Stream.delay $ Stream.enumerateFromTo 1 3
1587+
1588+-- | Pair each element in a stream with an absolute timestamp, using a clock of
1589+-- specified granularity. The timestamp is generated just before the element
1590+-- is consumed.
1591+--
1592+-- >>> Stream.fold Fold.toList $ Stream.timestampWith 0.01 $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
1593+-- [(AbsTime (TimeSpec {sec = ..., nsec = ...}),1),(AbsTime (TimeSpec {sec = ..., nsec = ...}),2),(AbsTime (TimeSpec {sec = ..., nsec = ...}),3)]
1594+--
1595+-- /Pre-release/
1596+--
1597+{-# INLINE timestampWith #-}
1598+timestampWith :: (MonadIO m)
1599+ => Double -> Stream m a -> Stream m (AbsTime, a)
1600+timestampWith g stream = zipWith (flip (,)) stream (absTimesWith g)
1601+
1602+-- TBD: check performance vs a custom implementation without using zipWith.
1603+--
1604+-- /Pre-release/
1605+--
1606+{-# INLINE timestamped #-}
1607+timestamped :: (MonadIO m)
1608+ => Stream m a -> Stream m (AbsTime, a)
1609+timestamped = timestampWith 0.01
1610+
1611+-- | Pair each element in a stream with relative times starting from 0, using a
1612+-- clock with the specified granularity. The time is measured just before the
1613+-- element is consumed.
1614+--
1615+-- >>> Stream.fold Fold.toList $ Stream.timeIndexWith 0.01 $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
1616+-- [(RelTime64 (NanoSecond64 ...),1),(RelTime64 (NanoSecond64 ...),2),(RelTime64 (NanoSecond64 ...),3)]
1617+--
1618+-- /Pre-release/
1619+--
1620+{-# INLINE timeIndexWith #-}
1621+timeIndexWith :: (MonadIO m)
1622+ => Double -> Stream m a -> Stream m (RelTime64, a)
1623+timeIndexWith g stream = zipWith (flip (,)) stream (relTimesWith g)
1624+
1625+-- | Pair each element in a stream with relative times starting from 0, using a
1626+-- 10 ms granularity clock. The time is measured just before the element is
1627+-- consumed.
1628+--
1629+-- >>> Stream.fold Fold.toList $ Stream.timeIndexed $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
1630+-- [(RelTime64 (NanoSecond64 ...),1),(RelTime64 (NanoSecond64 ...),2),(RelTime64 (NanoSecond64 ...),3)]
1631+--
1632+-- /Pre-release/
1633+--
1634+{-# INLINE timeIndexed #-}
1635+timeIndexed :: (MonadIO m)
1636+ => Stream m a -> Stream m (RelTime64, a)
1637+timeIndexed = timeIndexWith 0.01
1638+
10011639------------------------------------------------------------------------------
10021640-- Searching
10031641------------------------------------------------------------------------------
10041642
1643+-- | Find all the indices where the element in the stream satisfies the given
1644+-- predicate.
1645+--
1646+-- >>> findIndices p = Stream.scanMaybe (Fold.findIndices p)
1647+--
10051648{-# INLINE_NORMAL findIndices #-}
10061649findIndices :: Monad m => (a -> Bool) -> Stream m a -> Stream m Int
10071650findIndices p (Stream step state) = Stream step' (state, 0)
@@ -1014,6 +1657,15 @@ findIndices p (Stream step state) = Stream step' (state, 0)
10141657 Skip s -> Skip (s, i)
10151658 Stop -> Stop
10161659
1660+-- | Find all the indices where the value of the element in the stream is equal
1661+-- to the given value.
1662+--
1663+-- >>> elemIndices a = Stream.findIndices (== a)
1664+--
1665+{-# INLINE elemIndices #-}
1666+elemIndices :: (Monad m, Eq a) => a -> Stream m a -> Stream m Int
1667+elemIndices a = findIndices (== a)
1668+
10171669{-# INLINE_NORMAL slicesBy #-}
10181670slicesBy :: Monad m => (a -> Bool) -> Stream m a -> Stream m (Int, Int)
10191671slicesBy p (Stream step1 state1) = Stream step (Just (state1, 0, 0))
@@ -1039,8 +1691,13 @@ slicesBy p (Stream step1 state1) = Stream step (Just (state1, 0, 0))
10391691
10401692data RollingMapState s a = RollingMapGo s a
10411693
1694+-- | Like 'rollingMap' but with an effectful map function.
1695+--
1696+-- /Pre-release/
1697+--
10421698{-# INLINE rollingMapM #-}
10431699rollingMapM :: Monad m => (Maybe a -> a -> m b) -> Stream m a -> Stream m b
1700+-- rollingMapM f = scanMaybe (FL.slide2 $ Window.rollingMapM f)
10441701rollingMapM f (Stream step1 state1) = Stream step (RollingMapGo state1 Nothing)
10451702
10461703 where
@@ -1054,16 +1711,31 @@ rollingMapM f (Stream step1 state1) = Stream step (RollingMapGo state1 Nothing)
10541711 Skip s -> return $ Skip $ RollingMapGo s curr
10551712 Stop -> return Stop
10561713
1057−-- | rollingMap is a special case of an incremental sliding fold. It can be
1714+-- rollingMap is a special case of an incremental sliding fold. It can be
10581715-- written as:
10591716--
10601717-- > fld f = slidingWindow 1 (Fold.foldl' (\_ (x,y) -> f y x)
10611718-- > rollingMap f = Stream.postscan (fld f) undefined
1719+
1720+-- | Apply a function on every two successive elements of a stream. The first
1721+-- argument of the map function is the previous element and the second argument
1722+-- is the current element. When the current element is the first element, the
1723+-- previous element is 'Nothing'.
1724+--
1725+-- /Pre-release/
10621726--
10631727{-# INLINE rollingMap #-}
10641728rollingMap :: Monad m => (Maybe a -> a -> b) -> Stream m a -> Stream m b
1729+-- rollingMap f = scanMaybe (FL.slide2 $ Window.rollingMap f)
10651730rollingMap f = rollingMapM (\x y -> return $ f x y)
10661731
1732+-- | Like 'rollingMap' but requires at least two elements in the stream,
1733+-- returns an empty stream otherwise.
1734+--
1735+-- This is the stream equivalent of the list idiom @zipWith f xs (tail xs)@.
1736+--
1737+-- /Pre-release/
1738+--
10671739{-# INLINE rollingMap2 #-}
10681740rollingMap2 :: Monad m => (a -> a -> b) -> Stream m a -> Stream m b
10691741rollingMap2 f = catMaybes . rollingMap g
@@ -1078,14 +1750,37 @@ rollingMap2 f = catMaybes . rollingMap g
10781750------------------------------------------------------------------------------
10791751
10801752-- XXX Will this always fuse properly?
1753+
1754+-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'
1755+-- elements, and return a stream of values extracted from 'Just'.
1756+--
1757+-- Equivalent to:
1758+--
1759+-- >>> mapMaybe f = Stream.catMaybes . fmap f
1760+--
10811761{-# INLINE_NORMAL mapMaybe #-}
10821762mapMaybe :: Monad m => (a -> Maybe b) -> Stream m a -> Stream m b
10831763mapMaybe f = fmap fromJust . filter isJust . map f
10841764
1765+-- | Like 'mapMaybe' but maps a monadic function.
1766+--
1767+-- Equivalent to:
1768+--
1769+-- >>> mapMaybeM f = Stream.catMaybes . Stream.mapM f
1770+--
1771+-- >>> mapM f = Stream.mapMaybeM (\x -> Just <$> f x)
1772+--
10851773{-# INLINE_NORMAL mapMaybeM #-}
10861774mapMaybeM :: Monad m => (a -> m (Maybe b)) -> Stream m a -> Stream m b
10871775mapMaybeM f = fmap fromJust . filter isJust . mapM f
10881776
1777+-- | In a stream of 'Maybe's, discard 'Nothing's and unwrap 'Just's.
1778+--
1779+-- >>> catMaybes = Stream.mapMaybe id
1780+-- >>> catMaybes = fmap fromJust . Stream.filter isJust
1781+--
1782+-- /Pre-release/
1783+--
10891784{-# INLINE catMaybes #-}
10901785catMaybes :: Monad m => Stream m (Maybe a) -> Stream m a
10911786-- catMaybes = fmap fromJust . filter isJust
@@ -1107,7 +1802,43 @@ catMaybes (Stream step state) = Stream step1 state
11071802
11081803-- | Use a filtering fold on a stream.
11091804--
1110−-- /Pre-release/
1805+-- >>> scanMaybe f = Stream.catMaybes . Stream.postscan f
1806+--
11111807{-# INLINE scanMaybe #-}
11121808scanMaybe :: Monad m => Fold m a (Maybe b) -> Stream m a -> Stream m b
1113−scanMaybe f = catMaybes . postscanOnce f
1809+scanMaybe f = catMaybes . postscan f
1810+
1811+------------------------------------------------------------------------------
1812+-- Either streams
1813+------------------------------------------------------------------------------
1814+
1815+-- | Discard 'Right's and unwrap 'Left's in an 'Either' stream.
1816+--
1817+-- >>> catLefts = fmap (fromLeft undefined) . Stream.filter isLeft
1818+--
1819+-- /Pre-release/
1820+--
1821+{-# INLINE catLefts #-}
1822+catLefts :: Monad m => Stream m (Either a b) -> Stream m a
1823+catLefts = fmap (fromLeft undefined) . filter isLeft
1824+
1825+-- | Discard 'Left's and unwrap 'Right's in an 'Either' stream.
1826+--
1827+-- >>> catRights = fmap (fromRight undefined) . Stream.filter isRight
1828+--
1829+-- /Pre-release/
1830+--
1831+{-# INLINE catRights #-}
1832+catRights :: Monad m => Stream m (Either a b) -> Stream m b
1833+catRights = fmap (fromRight undefined) . filter isRight
1834+
1835+-- | Remove the either wrapper and flatten both lefts and as well as rights in
1836+-- the output stream.
1837+--
1838+-- >>> catEithers = fmap (either id id)
1839+--
1840+-- /Pre-release/
1841+--
1842+{-# INLINE catEithers #-}
1843+catEithers :: Monad m => Stream m (Either a a) -> Stream m a
1844+catEithers = fmap (either id id)