on bc0da07 help sign in
890 lines 28 kB history raw
1-- |
2-- Module : Streamly.Internal.Data.Stream.Transform
3-- Copyright : (c) 2017 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
7-- Portability : GHC
9module Streamly.Internal.Data.Stream.Transform
10 (
11 -- * Piping
12 -- | Pass through a 'Pipe'.
13 transform
15 -- * Folding
16 , foldrS
17 , foldrT
19 -- * Mapping
20 -- | Stateless one-to-one maps.
21 , sequence
22 , mapM
24 -- * Mapping Side Effects (Observation)
25 -- | See also the intersperse*_ combinators.
26 , trace
27 , trace_
28 , tap
30 -- * Scanning
31 , scan
32 , scanMany
33 , postscan
34 , smapM
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.
41 , with
42 , deleteBy
43 , filter
44 , filterM
45 , foldFilter
46 , uniq
47 , uniqBy
48 , nubBy
49 , prune
50 , repeated
52 -- * Trimming
53 -- | Produce a subset of the stream trimmed at ends.
55 , take
56 , takeWhile
57 , takeWhileM
58 , drop
59 , dropWhile
60 , dropWhileM
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.
67 -- Element agnostic (Opposite of sampling)
68 , intersperse
69 , intersperseM -- XXX naming
70 , intersperseBySpan
72 , intersperseSuffix
73 , intersperseSuffixBySpan
75 -- , interspersePrefix
76 -- , interspersePrefixBySpan
78 -- * Inserting Side Effects/Time
79 , intersperseM_ -- XXX naming
80 , delay
81 , intersperseSuffix_
82 , delayPost
83 , interspersePrefix_
84 , delayPre
86 -- * Element Aware Insertion
87 -- | Opposite of filtering
88 , insertBy
89 -- , intersperseByBefore
90 -- , intersperseByAfter
92 -- * Reordering
93 , reverse
94 , reverse'
95 , reassembleBy
97 -- * Position Indexing
98 , indexed
99 , indexedR
101 -- * Searching
102 , findIndices -- XXX indicesBy
103 , elemIndices -- XXX indicesOf
105 -- * Rolling map
106 -- | Map using the previous element.
107 , rollingMapM
108 , rollingMap
109 , rollingMap2
111 -- * Maybe Streams
112 -- Move these to Streamly.Data.Maybe.Stream?
113 , catMaybes -- XXX justs (like lefts/rights)
114 , mapMaybe
115 , mapMaybeM
117 -- * Either Streams
118 -- Move these to Streamly.Data.Either.Stream?
119 , lefts
120 , rights
121 , both
122 )
123where
125#include "inline.hs"
127import Control.Concurrent (threadDelay)
128import Control.Monad (void)
129import Control.Monad.IO.Class (MonadIO (liftIO))
130import Control.Monad.Trans.Class (MonadTrans(..))
131import Data.Either (fromLeft, isLeft, isRight, fromRight)
132import Data.Maybe (isJust, fromJust)
134import Streamly.Internal.Data.Fold.Type (Fold)
135import Streamly.Internal.Data.Pipe (Pipe)
137import qualified Streamly.Internal.Data.Fold as FL
138import qualified Streamly.Internal.Data.Stream.StreamD.Transform as D
139import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
140import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
142import Streamly.Internal.Data.Stream.Bottom
143import Streamly.Internal.Data.Stream.Type
145import Prelude hiding
146 ( filter, drop, dropWhile, take, takeWhile, foldr, map, mapM, sequence
147 , reverse, foldr1 , repeat, scanl, scanl1, zipWith)
149--
150-- $setup
151-- >>> :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
160-- >>> import qualified Streamly.Internal.Data.Fold as Fold
161-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
162-- >>> import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
163--
164-- >>> hSetBuffering stdout LineBuffering
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.
170------------------------------------------------------------------------------
171-- Piping
172------------------------------------------------------------------------------
174-- | Use a 'Pipe' to transform a stream.
175--
176-- /Pre-release/
177--
178{-# INLINE transform #-}
179transform :: Monad m => Pipe m a b -> Stream m a -> Stream m b
180transform pipe xs = fromStreamD $ D.transform pipe (toStreamD xs)
182------------------------------------------------------------------------------
183-- Transformation Folds
184------------------------------------------------------------------------------
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 #-}
215foldrS ::
216 (a -> Stream m b -> Stream m b)
217 -> Stream m b
218 -> Stream m a
219 -> Stream m b
220foldrS f z xs =
221 fromStreamK
222 $ K.foldrS
223 (\y ys -> toStreamK $ f y (fromStreamK ys))
224 (toStreamK z)
225 (toStreamK xs)
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 #-}
241foldrT :: (Monad m, Monad (s m), MonadTrans s)
242 => (a -> s m b -> s m b) -> s m b -> Stream m a -> s m b
243foldrT f z s = D.foldrT f z (toStreamD s)
245------------------------------------------------------------------------------
246-- Transformation by Mapping
247------------------------------------------------------------------------------
249-- |
250-- >>> mapM f = Stream.sequence . fmap f
251--
252-- Apply a monadic function to each element of the stream and replace it with
253-- the output of the resulting action.
254--
255-- >>> s = Stream.unfold Unfold.fromList ["a", "b", "c"]
256-- >>> Stream.fold Fold.drain $ Stream.mapM putStr s
257-- abc
258--
259-- /Pre-release/
260--
261{-# INLINE mapM #-}
262mapM :: Monad m => (a -> m b) -> Stream m a -> Stream m b
263mapM f m = fromStreamK $ D.toStreamK $ D.mapM f $ toStreamD m
265-- |
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 #-}
276sequence :: Monad m => Stream m (m a) -> Stream m a
277sequence = mapM id
279------------------------------------------------------------------------------
280-- Mapping side effects
281------------------------------------------------------------------------------
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 #-}
302tap :: Monad m => FL.Fold m a b -> Stream m a -> Stream m a
303tap f xs = fromStreamD $ D.tap f (toStreamD xs)
305-- | Apply a monadic function to each element flowing through the stream and
306-- discard the results.
307--
308-- >>> s = Stream.unfold Unfold.enumerateFromTo (1, 2)
309-- >>> Stream.fold Fold.drain $ Stream.trace print s
310-- 1
311-- 2
312--
313-- Compare with 'tap'.
314--
315{-# INLINE trace #-}
316trace :: Monad m => (a -> m b) -> Stream m a -> Stream m a
317trace f = mapM (\x -> void (f x) >> return x)
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_ #-}
333trace_ :: Monad m => m b -> Stream m a -> Stream m a
334trace_ eff = fromStreamD . D.mapM (\x -> eff >> return x) . toStreamD
336------------------------------------------------------------------------------
337-- Scanning with a Fold
338------------------------------------------------------------------------------
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.
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 #-}
400scan :: Monad m => Fold m a b -> Stream m a -> Stream m b
401scan fld m = fromStreamD $ D.scanOnce fld $ toStreamD m
403-- | Like 'scan' but restarts scanning afresh when the scanning fold
404-- terminates.
405--
406-- /Pre-release/
407{-# INLINE scanMany #-}
408scanMany :: Monad m => Fold m a b -> Stream m a -> Stream m b
409scanMany fld m = fromStreamD $ D.scanMany fld $ toStreamD m
411------------------------------------------------------------------------------
412-- Filtering
413------------------------------------------------------------------------------
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.
419--
420-- >>> filterWithIndex = Stream.with Stream.indexed Stream.filter
421--
422-- /Pre-release/
423{-# INLINE with #-}
424with :: 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)
428with f comb g = fmap snd . comb g . f
430-- | Include only those elements that pass a predicate.
431--
432{-# INLINE filter #-}
433filter :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
434filter p m = fromStreamD $ D.filter p $ toStreamD m
436-- | Same as 'filter' but with a monadic predicate.
437--
438{-# INLINE filterM #-}
439filterM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
440filterM p m = fromStreamD $ D.filterM p $ toStreamD m
442-- | Use a filtering fold on a stream.
443--
444-- >>> input = Stream.unfold Unfold.fromList [1..10]
445-- >>> Stream.fold Fold.sum $ Stream.foldFilter (Fold.satisfy (> 5)) input
446-- 40
447--
448-- /Pre-release/
449--
450{-# INLINE foldFilter #-}
451foldFilter :: Monad m => Fold m a (Maybe b) -> Stream m a -> Stream m b
452foldFilter p = fromStreamD . D.foldFilter p . toStreamD
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 #-}
473uniqBy :: Monad m =>
474 (a -> a -> Bool) -> Stream m a -> Stream m a
475uniqBy eq = catMaybes . rollingMap f
477 where
479 f pre curr =
480 case pre of
481 Nothing -> Just curr
482 Just x -> if x `eq` curr then Nothing else Just curr
484-- | Drop repeated elements that are adjacent to each other.
485--
486{-# INLINE uniq #-}
487uniq :: (Eq a, Monad m) => Stream m a -> Stream m a
488uniq = fromStreamD . D.uniq . toStreamD
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 #-}
505prune ::
506 -- (Monad m, Eq a) =>
507 (a -> Bool) -> Stream m a -> Stream m a
508prune = error "Not implemented yet!"
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.
517-- | Emit only repeated elements, once.
518--
519-- /Unimplemented/
520repeated :: -- (Monad m, Eq a) =>
521 Stream m a -> Stream m a
522repeated = undefined
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.
528-- | Drop repeated elements anywhere in the stream.
529--
530-- /Caution: not scalable for infinite streams/
531--
532-- /Unimplemented/
533--
534{-# INLINE nubBy #-}
535nubBy :: -- Monad m =>
536 (a -> a -> Bool) -> Stream m a -> Stream m a
537nubBy = undefined -- fromStreamD . D.nubBy . toStreamD
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 #-}
547deleteBy :: Monad m => (a -> a -> Bool) -> a -> Stream m a -> Stream m a
548deleteBy cmp x m = fromStreamD $ D.deleteBy cmp x (toStreamD m)
550------------------------------------------------------------------------------
551-- Trimming
552------------------------------------------------------------------------------
554-- | Same as 'takeWhile' but with a monadic predicate.
555--
556{-# INLINE takeWhileM #-}
557takeWhileM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
558takeWhileM p m = fromStreamD $ D.takeWhileM p $ toStreamD m
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 #-}
564dropWhile :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
565dropWhile p m = fromStreamD $ D.dropWhile p $ toStreamD m
567-- | Same as 'dropWhile' but with a monadic predicate.
568--
569{-# INLINE dropWhileM #-}
570dropWhileM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
571dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m
573------------------------------------------------------------------------------
574-- Inserting Elements
575------------------------------------------------------------------------------
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 #-}
587insertBy ::Monad m => (a -> a -> Ordering) -> a -> Stream m a -> Stream m a
588insertBy cmp x m = fromStreamD $ D.insertBy cmp x (toStreamD m)
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 #-}
597intersperse :: Monad m => a -> Stream m a -> Stream m a
598intersperse a = fromStreamD . D.intersperse a . toStreamD
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_ #-}
609intersperseM_ :: Monad m => m b -> Stream m a -> Stream m a
610intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
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 #-}
621intersperseBySpan :: -- Monad m =>
622 Int -> m a -> t m a -> t m a
623intersperseBySpan _n _f _xs = undefined
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 #-}
633intersperseSuffix :: Monad m => m a -> Stream m a -> Stream m a
634intersperseSuffix m = fromStreamD . D.intersperseSuffix m . toStreamD
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_ #-}
645intersperseSuffix_ :: Monad m => m b -> Stream m a -> Stream m a
646intersperseSuffix_ m = fromStreamD . D.intersperseSuffix_ m . toStreamD
648-- XXX Use an offset argument, like tapOffsetEvery
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 #-}
660intersperseSuffixBySpan :: Monad m
661 => Int -> m a -> Stream m a -> Stream m a
662intersperseSuffixBySpan n eff =
663 fromStreamD . D.intersperseSuffixBySpan n eff . toStreamD
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_ #-}
676interspersePrefix_ :: Monad m => m b -> Stream m a -> Stream m a
677interspersePrefix_ m = mapM (\x -> void m >> return x)
679------------------------------------------------------------------------------
680-- Inserting Time
681------------------------------------------------------------------------------
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 #-}
692delay :: MonadIO m => Double -> Stream m a -> Stream m a
693delay n = intersperseM_ $ liftIO $ threadDelay $ round $ n * 1000000
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 #-}
707delayPost :: MonadIO m => Double -> Stream m a -> Stream m a
708delayPost n = intersperseSuffix_ $ liftIO $ threadDelay $ round $ n * 1000000
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 #-}
722delayPre :: MonadIO m => Double -> Stream m a -> Stream m a
723delayPre n = trace_ $ liftIO $ threadDelay $ round $ n * 1000000
725------------------------------------------------------------------------------
726-- Reorder in sequence
727------------------------------------------------------------------------------
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 #-}
736reassembleBy
737 :: -- Monad m =>
738 Fold m a b
739 -> (a -> a -> Int)
740 -> t m a
741 -> t m b
742reassembleBy = undefined
744------------------------------------------------------------------------------
745-- Position Indexing
746------------------------------------------------------------------------------
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 #-}
759indexed :: Monad m => Stream m a -> Stream m (Int, a)
760indexed = fromStreamD . D.indexed . toStreamD
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 #-}
776indexedR :: Monad m => Int -> Stream m a -> Stream m (Int, a)
777indexedR n = fromStreamD . D.indexedR n . toStreamD
779------------------------------------------------------------------------------
780-- Searching
781------------------------------------------------------------------------------
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 #-}
789elemIndices :: (Monad m, Eq a) => a -> Stream m a -> Stream m Int
790elemIndices a = findIndices (== a)
792------------------------------------------------------------------------------
793-- Rolling map
794------------------------------------------------------------------------------
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.
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 #-}
807rollingMap :: Monad m => (Maybe a -> a -> b) -> Stream m a -> Stream m b
808rollingMap f m = fromStreamD $ D.rollingMap f $ toStreamD m
810-- | Like 'rollingMap' but with an effectful map function.
811--
812-- /Pre-release/
813--
814{-# INLINE rollingMapM #-}
815rollingMapM :: Monad m => (Maybe a -> a -> m b) -> Stream m a -> Stream m b
816rollingMapM f m = fromStreamD $ D.rollingMapM f $ toStreamD m
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 #-}
826rollingMap2 :: Monad m => (a -> a -> b) -> Stream m a -> Stream m b
827rollingMap2 f m = fromStreamD $ D.rollingMap2 f $ toStreamD m
829------------------------------------------------------------------------------
830-- Maybe Streams
831------------------------------------------------------------------------------
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 #-}
841mapMaybe :: Monad m => (a -> Maybe b) -> Stream m a -> Stream m b
842mapMaybe f m = fromStreamD $ D.mapMaybe f $ toStreamD m
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 #-}
851mapMaybeM :: Monad m
852 => (a -> m (Maybe b)) -> Stream m a -> Stream m b
853mapMaybeM f = fmap fromJust . filter isJust . mapM f
855-- | In a stream of 'Maybe's, discard 'Nothing's and unwrap 'Just's.
856--
857-- /Pre-release/
858--
859{-# INLINE catMaybes #-}
860catMaybes :: Monad m => Stream m (Maybe a) -> Stream m a
861catMaybes = fmap fromJust . filter isJust
863------------------------------------------------------------------------------
864-- Either streams
865------------------------------------------------------------------------------
867-- | Discard 'Right's and unwrap 'Left's in an 'Either' stream.
868--
869-- /Pre-release/
870--
871{-# INLINE lefts #-}
872lefts :: Monad m => Stream m (Either a b) -> Stream m a
873lefts = fmap (fromLeft undefined) . filter isLeft
875-- | Discard 'Left's and unwrap 'Right's in an 'Either' stream.
876--
877-- /Pre-release/
878--
879{-# INLINE rights #-}
880rights :: Monad m => Stream m (Either a b) -> Stream m b
881rights = fmap (fromRight undefined) . filter isRight
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 #-}
889both :: Monad m => Stream m (Either a a) -> Stream m a
890both = fmap (either id id)