@@ -1,1056 +0,0 @@
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
9−module Streamly.Internal.Data.Stream.Transform
12− -- | Pass through a 'Pipe'.
19− -- | Stateless one-to-one maps.
23− -- * Mapping Side Effects (Observation)
24− -- | See also the intersperse*_ combinators.
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.
53− -- Stateful/scanning filters
60− -- | Produce a subset of the stream trimmed at ends.
74− -- * Position Indexing
85− , findIndices -- XXX indicesBy
86− , elemIndices -- XXX indicesOf
89− -- | Map using the previous element.
96− -- * Inserting Elements
97− -- | Produce a superset of the stream. This is the opposite of
98− -- filtering/sampling. We can always use concatMap and scan for inserting
99− -- but these combinators are more efficient and convenient.
101− -- Element agnostic (Opposite of sampling)
103− , intersperseM -- XXX naming
106− , intersperseMSuffix
107− , intersperseMSuffixWith
109− -- , interspersePrefix
110− -- , interspersePrefixBySpan
112− -- * Inserting Side Effects/Time
113− , intersperseM_ -- XXX naming
115− , intersperseMSuffix_
117− , intersperseMPrefix_
120− -- * Element Aware Insertion
121− -- | Opposite of filtering
123− -- , intersperseByBefore
124− -- , intersperseByAfter
126− -- Fold and Unfold, Buffering
134− -- Move these to Streamly.Data.Either.Stream?
143−import Control.Concurrent (threadDelay)
144−import Control.Monad (void)
145−import Control.Monad.IO.Class (MonadIO (liftIO))
146−import Data.Either (fromLeft, isLeft, isRight, fromRight)
147−import Data.Maybe (isJust, fromJust)
149−import Streamly.Internal.Data.Fold.Type (Fold)
150−import Streamly.Internal.Data.Pipe (Pipe)
151−import Streamly.Internal.Data.Time.Units (AbsTime, RelTime64)
153−import qualified Streamly.Internal.Data.Fold as FL
154−-- import qualified Streamly.Internal.Data.Fold.Window as Window
155−import qualified Streamly.Internal.Data.Stream.StreamD.Transform as D
156−import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
157−import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
160−import Streamly.Internal.Data.Stream.Bottom
161−import Streamly.Internal.Data.Stream.Type
163−import Prelude hiding
164− ( filter, drop, dropWhile, take, takeWhile, foldr, map, mapM, sequence
165− , reverse, foldr1 , repeat, scanl, scanl1, zipWith)
170−-- >>> import Control.Concurrent (threadDelay)
171−-- >>> import Control.Monad (void)
172−-- >>> import Control.Monad.IO.Class (MonadIO (liftIO))
173−-- >>> import Data.Either (fromLeft, fromRight, isLeft, isRight, either)
174−-- >>> import Data.Function ((&))
175−-- >>> import Data.Maybe (fromJust, isJust)
176−-- >>> import Prelude hiding (filter, drop, dropWhile, take, takeWhile, foldr, map, mapM, sequence, reverse, foldr1 , scanl, scanl1)
177−-- >>> import Streamly.Internal.Data.Stream (Stream)
178−-- >>> import qualified Streamly.Data.Fold as Fold
179−-- >>> import qualified Streamly.Data.Unfold as Unfold
180−-- >>> import qualified Streamly.Internal.Data.Fold as Fold (filtering)
181−-- >>> import qualified Streamly.Internal.Data.Fold.Window as Window
182−-- >>> import qualified Streamly.Internal.Data.Stream as Stream
183−-- >>> import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
185−-- >>> hSetBuffering stdout LineBuffering
187−-- XXX because of the use of D.cons for appending, folds and scans have
188−-- quadratic complexity when iterated over a stream. We should use StreamK for
189−-- linear performance on iteration.
191−------------------------------------------------------------------------------
193−------------------------------------------------------------------------------
195−-- | Use a 'Pipe' to transform a stream.
199−{-# INLINE transform #-}
200−transform :: Monad m => Pipe m a b -> Stream m a -> Stream m b
201−transform pipe xs = fromStreamD $ D.transform pipe (toStreamD xs)
203−------------------------------------------------------------------------------
204−-- Transformation Folds
205−------------------------------------------------------------------------------
207−-- | Right fold to a streaming monad.
209−-- > foldrS Stream.cons Stream.nil === id
211−-- 'foldrS' can be used to perform stateless stream to stream transformations
212−-- like map and filter in general. It can be coupled with a scan to perform
213−-- stateful transformations. However, note that the custom map and filter
214−-- routines can be much more efficient than this due to better stream fusion.
216−-- >>> input = Stream.fromList [1..5]
217−-- >>> Stream.fold Fold.toList $ Stream.foldrS Stream.cons Stream.nil input
220−-- Find if any element in the stream is 'True':
222−-- >>> step x xs = if odd x then Stream.fromPure True else xs
223−-- >>> input = Stream.fromList (2:4:5:undefined) :: Stream IO Int
224−-- >>> Stream.fold Fold.toList $ Stream.foldrS step (Stream.fromPure False) input
227−-- Map (+2) on odd elements and filter out the even elements:
229−-- >>> step x xs = if odd x then (x + 2) `Stream.cons` xs else xs
230−-- >>> input = Stream.fromList [1..5] :: Stream IO Int
231−-- >>> Stream.fold Fold.toList $ Stream.foldrS step Stream.nil input
235−{-# INLINE foldrS #-}
237− (a -> Stream m b -> Stream m b)
244− (\y ys -> toStreamK $ f y (fromStreamK ys))
248−------------------------------------------------------------------------------
249−-- Transformation by Mapping
250−------------------------------------------------------------------------------
253−-- >>> mapM f = Stream.sequence . fmap f
255−-- Apply a monadic function to each element of the stream and replace it with
256−-- the output of the resulting action.
258−-- >>> s = Stream.fromList ["a", "b", "c"]
259−-- >>> Stream.fold Fold.drain $ Stream.mapM putStr s
263−mapM :: Monad m => (a -> m b) -> Stream m a -> Stream m b
264−mapM f m = fromStreamK $ D.toStreamK $ D.mapM f $ toStreamD m
267−-- >>> sequence = Stream.mapM id
269−-- Replace the elements of a stream of monadic actions with the outputs of
272−-- >>> s = Stream.fromList [putStr "a", putStr "b", putStrLn "c"]
273−-- >>> Stream.fold Fold.drain $ Stream.sequence s
276−{-# INLINE sequence #-}
277−sequence :: Monad m => Stream m (m a) -> Stream m a
280−------------------------------------------------------------------------------
281−-- Mapping side effects
282−------------------------------------------------------------------------------
284−-- | Tap the data flowing through a stream into a 'Fold'. For example, you may
285−-- add a tap to log the contents flowing through the stream. The fold is used
286−-- only for effects, its result is discarded.
291−-- -----stream m a ---------------stream m a-----
295−-- >>> s = Stream.enumerateFromTo 1 2
296−-- >>> Stream.fold Fold.drain $ Stream.tap (Fold.drainMapM print) s
300−-- Compare with 'trace'.
303−tap :: Monad m => FL.Fold m a b -> Stream m a -> Stream m a
304−tap f xs = fromStreamD $ D.tap f (toStreamD xs)
306−-- | Apply a monadic function to each element flowing through the stream and
307−-- discard the results.
309−-- >>> s = Stream.enumerateFromTo 1 2
310−-- >>> Stream.fold Fold.drain $ Stream.trace print s
314−-- Compare with 'tap'.
317−trace :: Monad m => (a -> m b) -> Stream m a -> Stream m a
318−trace f = mapM (\x -> void (f x) >> return x)
320−-- | Perform a side effect before yielding each element of the stream and
321−-- discard the results.
323−-- >>> s = Stream.enumerateFromTo 1 2
324−-- >>> Stream.fold Fold.drain $ Stream.trace_ (print "got here") s
328−-- Same as 'intersperseMPrefix_' but always serial.
333−{-# INLINE trace_ #-}
334−trace_ :: Monad m => m b -> Stream m a -> Stream m a
335−trace_ eff = fromStreamD . D.mapM (\x -> eff >> return x) . toStreamD
337−-------------------------------------------------------------------------------
339−-------------------------------------------------------------------------------
341−-- | @scanlMAfter' accumulate initial done stream@ is like 'scanlM'' except
342−-- that it provides an additional @done@ function to be applied on the
343−-- accumulator when the stream stops. The result of @done@ is also emitted in
346−-- This function can be used to allocate a resource in the beginning of the
347−-- scan and release it when the stream ends or to flush the internal state of
348−-- the scan at the end.
352−{-# INLINE scanlMAfter' #-}
360−scanlMAfter' step initial done stream =
361− fromStreamD $ D.scanlMAfter' step initial done $ toStreamD stream
363−------------------------------------------------------------------------------
364−-- Scanning with a Fold
365−------------------------------------------------------------------------------
367−-- XXX It may be useful to have a version of scan where we can keep the
368−-- accumulator independent of the value emitted. So that we do not necessarily
369−-- have to keep a value in the accumulator which we are not using. We can pass
370−-- an extraction function that will take the accumulator and the current value
371−-- of the element and emit the next value in the stream. That will also make it
372−-- possible to modify the accumulator after using it. In fact, the step function
373−-- can return new accumulator and the value to be emitted. The signature would
374−-- be more like mapAccumL.
376−-- | Strict left scan. Scan a stream using the given monadic fold.
378−-- >>> s = Stream.fromList [1..10]
379−-- >>> Stream.fold Fold.toList $ Stream.takeWhile (< 10) $ Stream.scan Fold.sum s
382−-- See also: 'usingStateT'
386−-- >>> scanl' step z = Stream.scan (Fold.foldl' step z)
388−-- Like 'map', 'scanl'' too is a one to one transformation,
389−-- however it adds an extra element.
391−-- >>> s = Stream.fromList [1,2,3,4]
392−-- >>> Stream.fold Fold.toList $ scanl' (+) 0 s
395−-- >>> Stream.fold Fold.toList $ scanl' (flip (:)) [] s
396−-- [[],[1],[2,1],[3,2,1],[4,3,2,1]]
398−-- The output of 'scanl'' is the initial value of the accumulator followed by
399−-- all the intermediate steps and the final result of 'foldl''.
401−-- By streaming the accumulated state after each fold step, we can share the
402−-- state across multiple stages of stream composition. Each stage can modify or
403−-- extend the state, do some processing with it and emit it for the next stage,
404−-- thus modularizing the stream processing. This can be useful in
405−-- stateful or event-driven programming.
407−-- Consider the following monolithic example, computing the sum and the product
408−-- of the elements in a stream in one go using a @foldl'@:
410−-- >>> foldl' step z = Stream.fold (Fold.foldl' step z)
411−-- >>> foldl' (\(s, p) x -> (s + x, p * x)) (0,1) s
414−-- Using @scanl'@ we can make it modular by computing the sum in the first
415−-- stage and passing it down to the next stage for computing the product:
418−-- foldl' (\(_, p) (s, x) -> (s, p * x)) (0,1)
419−-- $ scanl' (\(s, _) x -> (s + x, x)) (0,1)
420−-- $ Stream.fromList [1,2,3,4]
424−-- IMPORTANT: 'scanl'' evaluates the accumulator to WHNF. To avoid building
425−-- lazy expressions inside the accumulator, it is recommended that a strict
426−-- data structure is used for accumulator.
429−scan :: Monad m => Fold m a b -> Stream m a -> Stream m b
430−scan fld m = fromStreamD $ D.scan fld $ toStreamD m
432−-- | Like 'scan' but restarts scanning afresh when the scanning fold
435−{-# INLINE scanMany #-}
436−scanMany :: Monad m => Fold m a b -> Stream m a -> Stream m b
437−scanMany fld m = fromStreamD $ D.scanMany fld $ toStreamD m
439−------------------------------------------------------------------------------
441−------------------------------------------------------------------------------
443−-- | Modify a @Stream m a -> Stream m a@ stream transformation that accepts a
444−-- predicate @(a -> b)@ to accept @((s, a) -> b)@ instead, provided a
445−-- transformation @Stream m a -> Stream m (s, a)@. Convenient to filter with
448−-- >>> filterWithIndex = Stream.with Stream.indexed Stream.filter
453− (Stream m a -> Stream m (s, a))
454− -> (((s, a) -> b) -> Stream m (s, a) -> Stream m (s, a))
455− -> (((s, a) -> b) -> Stream m a -> Stream m a)
456−with f comb g = fmap snd . comb g . f
458−-- | Include only those elements that pass a predicate.
460−-- >>> filter p = Stream.filterM (return . p)
461−-- >>> filter p = Stream.mapMaybe (\x -> if p x then Just x else Nothing)
462−-- >>> filter p = Stream.scanMaybe (Fold.filtering p)
464−{-# INLINE filter #-}
465−filter :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
466−-- filter p = scanMaybe (FL.filtering p)
467−filter p m = fromStreamD $ D.filter p $ toStreamD m
469−-- | Same as 'filter' but with a monadic predicate.
471−-- >>> f p x = p x >>= \r -> return $ if r then Just x else Nothing
472−-- >>> filterM p = Stream.mapMaybeM (f p)
474−{-# INLINE filterM #-}
475−filterM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
476−filterM p m = fromStreamD $ D.filterM p $ toStreamD m
478−-- | Drop repeated elements that are adjacent to each other using the supplied
479−-- comparison function.
481−-- >>> uniq = Stream.uniqBy (==)
483−-- To strip duplicate path separators:
485−-- >>> input = Stream.fromList "//a//b"
486−-- >>> f x y = x == '/' && y == '/'
487−-- >>> Stream.fold Fold.toList $ Stream.uniqBy f input
494−{-# INLINE uniqBy #-}
496− (a -> a -> Bool) -> Stream m a -> Stream m a
497−-- uniqBy eq = scanMaybe (FL.uniqBy eq)
498−uniqBy eq = catMaybes . rollingMap f
504− Nothing -> Just curr
505− Just x -> if x `eq` curr then Nothing else Just curr
507−-- | Drop repeated elements that are adjacent to each other.
509−-- >>> uniq = Stream.uniqBy (==)
512−uniq :: (Eq a, Monad m) => Stream m a -> Stream m a
513−-- uniq = scanMaybe FL.uniq
514−uniq = fromStreamD . D.uniq . toStreamD
516−-- | Strip all leading and trailing occurrences of an element passing a
517−-- predicate and make all other consecutive occurrences uniq.
519−-- >> prune p = Stream.dropWhileAround p $ Stream.uniqBy (x y -> p x && p y)
522−-- > Stream.prune isSpace (Stream.fromList " hello world! ")
532− -- (Monad m, Eq a) =>
533− (a -> Bool) -> Stream m a -> Stream m a
534−prune = error "Not implemented yet!"
536−-- Possible implementation:
538−-- Stream.catMaybes . Stream.parseMany (Parser.groupBy (==) Fold.repeated)@
540−-- 'Fold.repeated' should return 'Just' when repeated, and 'Nothing' for a
543−-- | Emit only repeated elements, once.
546−repeated :: -- (Monad m, Eq a) =>
547− Stream m a -> Stream m a
550−-- | Deletes the first occurrence of the element in the stream that satisfies
551−-- the given equality predicate.
553−-- >>> input = Stream.fromList [1,3,3,5]
554−-- >>> Stream.fold Fold.toList $ Stream.deleteBy (==) 3 input
557−{-# INLINE deleteBy #-}
558−deleteBy :: Monad m => (a -> a -> Bool) -> a -> Stream m a -> Stream m a
559−-- deleteBy cmp x = scanMaybe (FL.deleteBy cmp x)
560−deleteBy cmp x m = fromStreamD $ D.deleteBy cmp x (toStreamD m)
562−------------------------------------------------------------------------------
564−------------------------------------------------------------------------------
566−-- | Same as 'takeWhile' but with a monadic predicate.
568−{-# INLINE takeWhileM #-}
569−takeWhileM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
570−-- takeWhileM p = scanMaybe (FL.takingEndByM_ (\x -> not <$> p x))
571−takeWhileM p m = fromStreamD $ D.takeWhileM p $ toStreamD m
573−-- | Take all consecutive elements at the end of the stream for which the
574−-- predicate is true.
576−-- O(n) space, where n is the number elements taken.
579−{-# INLINE takeWhileLast #-}
580−takeWhileLast :: -- Monad m =>
581− (a -> Bool) -> Stream m a -> Stream m a
582−takeWhileLast = undefined -- fromStreamD $ D.takeWhileLast n $ toStreamD m
584−-- | Like 'takeWhile' and 'takeWhileLast' combined.
586−-- O(n) space, where n is the number elements taken from the end.
589−{-# INLINE takeWhileAround #-}
590−takeWhileAround :: -- Monad m =>
591− (a -> Bool) -> Stream m a -> Stream m a
592−takeWhileAround = undefined -- fromStreamD $ D.takeWhileAround n $ toStreamD m
594−-- | Drop elements in the stream as long as the predicate succeeds and then
595−-- take the rest of the stream.
597−{-# INLINE dropWhile #-}
598−dropWhile :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
599−-- dropWhile p = scanMaybe (FL.droppingWhile p)
600−dropWhile p m = fromStreamD $ D.dropWhile p $ toStreamD m
602−-- | Same as 'dropWhile' but with a monadic predicate.
604−{-# INLINE dropWhileM #-}
605−dropWhileM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
606−-- dropWhileM p = scanMaybe (FL.droppingWhileM p)
607−dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m
609−-- | Drop @n@ elements at the end of the stream.
611−-- O(n) space, where n is the number elements dropped.
614−{-# INLINE dropLast #-}
615−dropLast :: -- Monad m =>
616− Int -> Stream m a -> Stream m a
617−dropLast = undefined -- fromStreamD $ D.dropLast n $ toStreamD m
619−-- | Drop all consecutive elements at the end of the stream for which the
620−-- predicate is true.
622−-- O(n) space, where n is the number elements dropped.
625−{-# INLINE dropWhileLast #-}
626−dropWhileLast :: -- Monad m =>
627− (a -> Bool) -> Stream m a -> Stream m a
628−dropWhileLast = undefined -- fromStreamD $ D.dropWhileLast n $ toStreamD m
630−-- | Like 'dropWhile' and 'dropWhileLast' combined.
632−-- O(n) space, where n is the number elements dropped from the end.
635−{-# INLINE dropWhileAround #-}
636−dropWhileAround :: -- Monad m =>
637− (a -> Bool) -> Stream m a -> Stream m a
638−dropWhileAround = undefined -- fromStreamD $ D.dropWhileAround n $ toStreamD m
640−------------------------------------------------------------------------------
641−-- Inserting Elements
642−------------------------------------------------------------------------------
644−-- | @insertBy cmp elem stream@ inserts @elem@ before the first element in
645−-- @stream@ that is less than @elem@ when compared using @cmp@.
647−-- >>> insertBy cmp x = Stream.mergeBy cmp (Stream.fromPure x)
649−-- >>> input = Stream.fromList [1,3,5]
650−-- >>> Stream.fold Fold.toList $ Stream.insertBy compare 2 input
653−{-# INLINE insertBy #-}
654−insertBy ::Monad m => (a -> a -> Ordering) -> a -> Stream m a -> Stream m a
655−insertBy cmp x m = fromStreamD $ D.insertBy cmp x (toStreamD m)
657−-- | Insert a pure value between successive elements of a stream.
659−-- >>> input = Stream.fromList "hello"
660−-- >>> Stream.fold Fold.toList $ Stream.intersperse ',' input
663−{-# INLINE intersperse #-}
664−intersperse :: Monad m => a -> Stream m a -> Stream m a
665−intersperse a = fromStreamD . D.intersperse a . toStreamD
667−-- | Insert a side effect before consuming an element of a stream except the
670−-- >>> input = Stream.fromList "hello"
671−-- >>> Stream.fold Fold.drain $ Stream.trace putChar $ Stream.intersperseM_ (putChar '.') input
675−{-# INLINE intersperseM_ #-}
676−intersperseM_ :: Monad m => m b -> Stream m a -> Stream m a
677−intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
679−-- | Intersperse a monadic action into the input stream after every @n@
682−-- >> input = Stream.fromList "hello"
683−-- >> Stream.fold Fold.toList $ Stream.intersperseMWith 2 (return ',') input
687−{-# INLINE intersperseMWith #-}
688−intersperseMWith :: -- Monad m =>
689− Int -> m a -> Stream m a -> Stream m a
690−intersperseMWith _n _f _xs = undefined
692−-- | Insert an effect and its output after consuming an element of a stream.
694−-- >>> input = Stream.fromList "hello"
695−-- >>> Stream.fold Fold.toList $ Stream.trace putChar $ Stream.intersperseMSuffix (putChar '.' >> return ',') input
696−-- h.,e.,l.,l.,o.,"h,e,l,l,o,"
699−{-# INLINE intersperseMSuffix #-}
700−intersperseMSuffix :: Monad m => m a -> Stream m a -> Stream m a
701−intersperseMSuffix m = fromStreamD . D.intersperseMSuffix m . toStreamD
703−-- | Insert a side effect after consuming an element of a stream.
705−-- >>> input = Stream.fromList "hello"
706−-- >>> Stream.fold Fold.toList $ Stream.intersperseMSuffix_ (threadDelay 1000000) input
711−{-# INLINE intersperseMSuffix_ #-}
712−intersperseMSuffix_ :: Monad m => m b -> Stream m a -> Stream m a
713−intersperseMSuffix_ m = fromStreamD . D.intersperseMSuffix_ m . toStreamD
715−-- XXX Use an offset argument, like tapOffsetEvery
717−-- | Like 'intersperseMSuffix' but intersperses an effectful action into the
718−-- input stream after every @n@ elements and after the last element.
720−-- >>> input = Stream.fromList "hello"
721−-- >>> Stream.fold Fold.toList $ Stream.intersperseMSuffixWith 2 (return ',') input
726−{-# INLINE intersperseMSuffixWith #-}
727−intersperseMSuffixWith :: Monad m
728− => Int -> m a -> Stream m a -> Stream m a
729−intersperseMSuffixWith n eff =
730− fromStreamD . D.intersperseMSuffixWith n eff . toStreamD
732−-- | Insert a side effect before consuming an element of a stream.
736−-- >>> intersperseMPrefix_ m = Stream.mapM (\x -> void m >> return x)
738−-- >>> input = Stream.fromList "hello"
739−-- >>> Stream.fold Fold.toList $ Stream.trace putChar $ Stream.intersperseMPrefix_ (putChar '.' >> return ',') input
746−{-# INLINE intersperseMPrefix_ #-}
747−intersperseMPrefix_ :: Monad m => m b -> Stream m a -> Stream m a
748−intersperseMPrefix_ m = mapM (\x -> void m >> return x)
750−------------------------------------------------------------------------------
752−------------------------------------------------------------------------------
754−-- XXX This should be in Prelude, should we export this as a helper function?
756−-- | Block the current thread for specified number of seconds.
758−sleep :: MonadIO m => Double -> m ()
759−sleep n = liftIO $ threadDelay $ round $ n * 1000000
761−-- | Introduce a delay of specified seconds between elements of the stream.
765−-- >>> sleep n = liftIO $ threadDelay $ round $ n * 1000000
766−-- >>> delay = Stream.intersperseM_ . sleep
770−-- >>> input = Stream.enumerateFromTo 1 3
771−-- >>> Stream.fold (Fold.drainMapM print) $ Stream.delay 1 input
777−delay :: MonadIO m => Double -> Stream m a -> Stream m a
778−delay = intersperseM_ . sleep
780−-- | Introduce a delay of specified seconds after consuming an element of a
785−-- >>> sleep n = liftIO $ threadDelay $ round $ n * 1000000
786−-- >>> delayPost = Stream.intersperseMSuffix_ . sleep
790−-- >>> input = Stream.enumerateFromTo 1 3
791−-- >>> Stream.fold (Fold.drainMapM print) $ Stream.delayPost 1 input
798−{-# INLINE delayPost #-}
799−delayPost :: MonadIO m => Double -> Stream m a -> Stream m a
800−delayPost n = intersperseMSuffix_ $ liftIO $ threadDelay $ round $ n * 1000000
802−-- | Introduce a delay of specified seconds before consuming an element of a
807−-- >>> sleep n = liftIO $ threadDelay $ round $ n * 1000000
808−-- >>> delayPre = Stream.intersperseMPrefix_. sleep
812−-- >>> input = Stream.enumerateFromTo 1 3
813−-- >>> Stream.fold (Fold.drainMapM print) $ Stream.delayPre 1 input
820−{-# INLINE delayPre #-}
821−delayPre :: MonadIO m => Double -> Stream m a -> Stream m a
822−delayPre = intersperseMPrefix_. sleep
824−------------------------------------------------------------------------------
825−-- Reorder in sequence
826−------------------------------------------------------------------------------
828−-- | Buffer until the next element in sequence arrives. The function argument
829−-- determines the difference in sequence numbers. This could be useful in
830−-- implementing sequenced streams, for example, TCP reassembly.
834−{-# INLINE reassembleBy #-}
841−reassembleBy = undefined
843−------------------------------------------------------------------------------
845−------------------------------------------------------------------------------
848−-- >>> f = Fold.foldl' (\(i, _) x -> (i + 1, x)) (-1,undefined)
849−-- >>> indexed = Stream.postscan f
850−-- >>> indexed = Stream.zipWith (,) (Stream.enumerateFrom 0)
851−-- >>> indexedR n = fmap (\(i, a) -> (n - i, a)) . indexed
853−-- Pair each element in a stream with its index, starting from index 0.
855−-- >>> Stream.fold Fold.toList $ Stream.indexed $ Stream.fromList "hello"
856−-- [(0,'h'),(1,'e'),(2,'l'),(3,'l'),(4,'o')]
858−{-# INLINE indexed #-}
859−indexed :: Monad m => Stream m a -> Stream m (Int, a)
860−-- indexed = scanMaybe FL.indexing
861−indexed = fromStreamD . D.indexed . toStreamD
864−-- >>> f n = Fold.foldl' (\(i, _) x -> (i - 1, x)) (n + 1,undefined)
865−-- >>> indexedR n = Stream.postscan (f n)
867−-- >>> s n = Stream.enumerateFromThen n (n - 1)
868−-- >>> indexedR n = Stream.zipWith (,) (s n)
870−-- Pair each element in a stream with its index, starting from the
871−-- given index @n@ and counting down.
873−-- >>> Stream.fold Fold.toList $ Stream.indexedR 10 $ Stream.fromList "hello"
874−-- [(10,'h'),(9,'e'),(8,'l'),(7,'l'),(6,'o')]
876−{-# INLINE indexedR #-}
877−indexedR :: Monad m => Int -> Stream m a -> Stream m (Int, a)
878−-- indexedR n = scanMaybe (FL.indexingRev n)
879−indexedR n = fromStreamD . D.indexedR n . toStreamD
881−-------------------------------------------------------------------------------
883−-------------------------------------------------------------------------------
885−-- Note: The timestamp stream must be the second stream in the zip so that the
886−-- timestamp is generated after generating the stream element and not before.
887−-- If we do not do that then the following example will generate the same
888−-- timestamp for first two elements:
890−-- Stream.fold Fold.toList $ Stream.timestamped $ Stream.delay $ Stream.enumerateFromTo 1 3
892−-- | Pair each element in a stream with an absolute timestamp, using a clock of
893−-- specified granularity. The timestamp is generated just before the element
896−-- >>> Stream.fold Fold.toList $ Stream.timestampWith 0.01 $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
897−-- [(AbsTime (TimeSpec {sec = ..., nsec = ...}),1),(AbsTime (TimeSpec {sec = ..., nsec = ...}),2),(AbsTime (TimeSpec {sec = ..., nsec = ...}),3)]
901−{-# INLINE timestampWith #-}
902−timestampWith :: (MonadIO m)
903− => Double -> Stream m a -> Stream m (AbsTime, a)
904−timestampWith g stream = zipWith (flip (,)) stream (absTimesWith g)
906−-- TBD: check performance vs a custom implementation without using zipWith.
910−{-# INLINE timestamped #-}
911−timestamped :: (MonadIO m)
912− => Stream m a -> Stream m (AbsTime, a)
913−timestamped = timestampWith 0.01
915−-- | Pair each element in a stream with relative times starting from 0, using a
916−-- clock with the specified granularity. The time is measured just before the
917−-- element is consumed.
919−-- >>> Stream.fold Fold.toList $ Stream.timeIndexWith 0.01 $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
920−-- [(RelTime64 (NanoSecond64 ...),1),(RelTime64 (NanoSecond64 ...),2),(RelTime64 (NanoSecond64 ...),3)]
922−-- /Pre-release/Monad
924−{-# INLINE timeIndexWith #-}
925−timeIndexWith :: (MonadIO m)
926− => Double -> Stream m a -> Stream m (RelTime64, a)
927−timeIndexWith g stream = zipWith (flip (,)) stream (relTimesWith g)
929−-- | Pair each element in a stream with relative times starting from 0, using a
930−-- 10 ms granularity clock. The time is measured just before the element is
933−-- >>> Stream.fold Fold.toList $ Stream.timeIndexed $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
934−-- [(RelTime64 (NanoSecond64 ...),1),(RelTime64 (NanoSecond64 ...),2),(RelTime64 (NanoSecond64 ...),3)]
938−{-# INLINE timeIndexed #-}
939−timeIndexed :: (MonadIO m)
940− => Stream m a -> Stream m (RelTime64, a)
941−timeIndexed = timeIndexWith 0.01
943−------------------------------------------------------------------------------
945−------------------------------------------------------------------------------
947−-- | Find all the indices where the value of the element in the stream is equal
948−-- to the given value.
950−-- >>> elemIndices a = Stream.findIndices (== a)
952−{-# INLINE elemIndices #-}
953−elemIndices :: (Monad m, Eq a) => a -> Stream m a -> Stream m Int
954−elemIndices a = findIndices (== a)
956−------------------------------------------------------------------------------
958−------------------------------------------------------------------------------
960−-- XXX this is not a one-to-one map so calling it map may not be right.
961−-- We can perhaps call it zipWithTail or rollWith.
963−-- | Apply a function on every two successive elements of a stream. The first
964−-- argument of the map function is the previous element and the second argument
965−-- is the current element. When the current element is the first element, the
966−-- previous element is 'Nothing'.
970−{-# INLINE rollingMap #-}
971−rollingMap :: Monad m => (Maybe a -> a -> b) -> Stream m a -> Stream m b
972−-- rollingMap f = scanMaybe (FL.slide2 $ Window.rollingMap f)
973−rollingMap f m = fromStreamD $ D.rollingMap f $ toStreamD m
975−-- | Like 'rollingMap' but with an effectful map function.
979−{-# INLINE rollingMapM #-}
980−rollingMapM :: Monad m => (Maybe a -> a -> m b) -> Stream m a -> Stream m b
981−-- rollingMapM f = scanMaybe (FL.slide2 $ Window.rollingMapM f)
982−rollingMapM f m = fromStreamD $ D.rollingMapM f $ toStreamD m
984−-- | Like 'rollingMap' but requires at least two elements in the stream,
985−-- returns an empty stream otherwise.
987−-- This is the stream equivalent of the list idiom @zipWith f xs (tail xs)@.
991−{-# INLINE rollingMap2 #-}
992−rollingMap2 :: Monad m => (a -> a -> b) -> Stream m a -> Stream m b
993−rollingMap2 f m = fromStreamD $ D.rollingMap2 f $ toStreamD m
995−------------------------------------------------------------------------------
997−------------------------------------------------------------------------------
999−-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'
1000−-- elements, and return a stream of values extracted from 'Just'.
1004−-- >>> mapMaybe f = Stream.catMaybes . fmap f
1006−{-# INLINE mapMaybe #-}
1007−mapMaybe :: Monad m => (a -> Maybe b) -> Stream m a -> Stream m b
1008−mapMaybe f m = fromStreamD $ D.mapMaybe f $ toStreamD m
1010−-- | Like 'mapMaybe' but maps a monadic function.
1014−-- >>> mapMaybeM f = Stream.catMaybes . Stream.mapM f
1016−-- >>> mapM f = Stream.mapMaybeM (\x -> Just <$> f x)
1018−{-# INLINE_EARLY mapMaybeM #-}
1019−mapMaybeM :: Monad m
1020− => (a -> m (Maybe b)) -> Stream m a -> Stream m b
1021−mapMaybeM f = fmap fromJust . filter isJust . mapM f
1023−------------------------------------------------------------------------------
1025−------------------------------------------------------------------------------
1027−-- | Discard 'Right's and unwrap 'Left's in an 'Either' stream.
1029−-- >>> catLefts = fmap (fromLeft undefined) . Stream.filter isLeft
1033−{-# INLINE catLefts #-}
1034−catLefts :: Monad m => Stream m (Either a b) -> Stream m a
1035−catLefts = fmap (fromLeft undefined) . filter isLeft
1037−-- | Discard 'Left's and unwrap 'Right's in an 'Either' stream.
1039−-- >>> catRights = fmap (fromRight undefined) . Stream.filter isRight
1043−{-# INLINE catRights #-}
1044−catRights :: Monad m => Stream m (Either a b) -> Stream m b
1045−catRights = fmap (fromRight undefined) . filter isRight
1047−-- | Remove the either wrapper and flatten both lefts and as well as rights in
1048−-- the output stream.
1050−-- >>> catEithers = fmap (either id id)
1054−{-# INLINE catEithers #-}
1055−catEithers :: Monad m => Stream m (Either a a) -> Stream m a
1056−catEithers = fmap (either id id)