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