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
9module Streamly.Internal.Data.Stream.Transform
12 -- | Pass through a 'Pipe'.
20 -- | Stateless one-to-one maps.
24 -- * Mapping Side Effects (Observation)
25 -- | See also the intersperse*_ combinators.
38 -- | Produce a subset of the stream using criteria based on the values of
39 -- the elements. We can use a concatMap and scan for filtering but these
40 -- combinators are more efficient and convenient.
42 -- mapMaybeM is a general filtering combinator as we can map the stream to
43 -- Just/Nothing using any stateful fold and then use this to filter out.
54 -- Stateful/scanning filters
61 -- | Produce a subset of the stream trimmed at ends.
75 -- * Position Indexing
86 , findIndices -- XXX indicesBy
87 , elemIndices -- XXX indicesOf
90 -- | Map using the previous element.
97 -- * Inserting Elements
98 -- | Produce a superset of the stream. This is the opposite of
99 -- filtering/sampling. We can always use concatMap and scan for inserting
100 -- but these combinators are more efficient and convenient.
102 -- Element agnostic (Opposite of sampling)
104 , intersperseM -- XXX naming
108 , intersperseSuffixBySpan
110 -- , interspersePrefix
111 -- , interspersePrefixBySpan
113 -- * Inserting Side Effects/Time
114 , intersperseM_ -- XXX naming
121 -- * Element Aware Insertion
122 -- | Opposite of filtering
124 -- , intersperseByBefore
125 -- , intersperseByAfter
127 -- Fold and Unfold, Buffering
135 -- Move these to Streamly.Data.Either.Stream?
144import Control.Concurrent (threadDelay)
145import Control.Monad (void)
146import Control.Monad.IO.Class (MonadIO (liftIO))
147import Control.Monad.Trans.Class (MonadTrans(..))
148import Data.Either (fromLeft, isLeft, isRight, fromRight)
149import Data.Maybe (isJust, fromJust)
151import Streamly.Internal.Data.Fold.Type (Fold)
152import Streamly.Internal.Data.Pipe (Pipe)
153import Streamly.Internal.Data.Time.Units (AbsTime, RelTime64)
155import qualified Streamly.Internal.Data.Fold as FL
156-- import qualified Streamly.Internal.Data.Fold.Window as Window
157import qualified Streamly.Internal.Data.Stream.StreamD.Transform as D
158import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
159import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
162import Streamly.Internal.Data.Stream.Bottom
163import Streamly.Internal.Data.Stream.Type
166 ( filter, drop, dropWhile, take, takeWhile, foldr, map, mapM, sequence
167 , reverse, foldr1 , repeat, scanl, scanl1, zipWith)
172-- >>> :set -package mtl
173-- >>> import Control.Concurrent (threadDelay)
174-- >>> import Control.Monad.IO.Class (MonadIO (liftIO))
175-- >>> import Control.Monad.Trans (lift)
176-- >>> import Control.Monad.Trans.Identity (runIdentityT)
177-- >>> import Data.Either (fromLeft, fromRight, isLeft, isRight, either)
178-- >>> import Data.Function ((&))
179-- >>> import Data.Maybe (fromJust, isJust)
180-- >>> import Prelude hiding (filter, drop, dropWhile, take, takeWhile, foldr, map, mapM, sequence, reverse, foldr1 , scanl, scanl1)
181-- >>> import Streamly.Internal.Data.Stream as Stream
182-- >>> import qualified Streamly.Data.Fold as Fold
183-- >>> import qualified Streamly.Internal.Data.Fold as Fold
184-- >>> import qualified Streamly.Internal.Data.Fold.Window as Window
185-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
186-- >>> import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
188-- >>> hSetBuffering stdout LineBuffering
190-- XXX because of the use of D.cons for appending, folds and scans have
191-- quadratic complexity when iterated over a stream. We should use StreamK for
192-- linear performance on iteration.
194------------------------------------------------------------------------------
196------------------------------------------------------------------------------
198-- | Use a 'Pipe' to transform a stream.
202{-# INLINE transform #-}
203transform :: Monad m => Pipe m a b -> Stream m a -> Stream m b
204transform pipe xs = fromStreamD $ D.transform pipe (toStreamD xs)
206------------------------------------------------------------------------------
207-- Transformation Folds
208------------------------------------------------------------------------------
210-- | Right fold to a streaming monad.
212-- > foldrS Stream.cons Stream.nil === id
214-- 'foldrS' can be used to perform stateless stream to stream transformations
215-- like map and filter in general. It can be coupled with a scan to perform
216-- stateful transformations. However, note that the custom map and filter
217-- routines can be much more efficient than this due to better stream fusion.
219-- >>> input = Stream.fromList [1..5]
220-- >>> Stream.fold Fold.toList $ Stream.foldrS Stream.cons Stream.nil input
223-- Find if any element in the stream is 'True':
225-- >>> step x xs = if odd x then return True else xs
226-- >>> input = Stream.fromList (2:4:5:undefined) :: Stream IO Int
227-- >>> Stream.fold Fold.toList $ Stream.foldrS step (return False) input
230-- Map (+2) on odd elements and filter out the even elements:
232-- >>> step x xs = if odd x then (x + 2) `Stream.cons` xs else xs
233-- >>> input = Stream.fromList [1..5] :: Stream IO Int
234-- >>> Stream.fold Fold.toList $ Stream.foldrS step Stream.nil input
240 (a -> Stream m b -> Stream m b)
247 (\y ys -> toStreamK $ f y (fromStreamK ys))
251-- | Right fold to a transformer monad. This is the most general right fold
252-- function. 'foldrS' is a special case of 'foldrT', however 'foldrS'
253-- implementation can be more efficient:
255-- >>> foldrS = foldrT
257-- >>> step f x xs = lift $ f x (runIdentityT xs)
258-- >>> foldrM f z s = runIdentityT $ foldrT (step f) (lift z) s
260-- 'foldrT' can be used to translate streamly streams to other transformer
261-- monads e.g. to a different streaming type.
265foldrT :: (Monad m, Monad (s m), MonadTrans s)
266 => (a -> s m b -> s m b) -> s m b -> Stream m a -> s m b
267foldrT f z s = D.foldrT f z (toStreamD s)
269------------------------------------------------------------------------------
270-- Transformation by Mapping
271------------------------------------------------------------------------------
274-- >>> mapM f = Stream.sequence . fmap f
276-- Apply a monadic function to each element of the stream and replace it with
277-- the output of the resulting action.
279-- >>> s = Stream.fromList ["a", "b", "c"]
280-- >>> Stream.fold Fold.drain $ Stream.mapM putStr s
284mapM :: Monad m => (a -> m b) -> Stream m a -> Stream m b
285mapM f m = fromStreamK $ D.toStreamK $ D.mapM f $ toStreamD m
288-- >>> sequence = Stream.mapM id
290-- Replace the elements of a stream of monadic actions with the outputs of
293-- >>> s = Stream.fromList [putStr "a", putStr "b", putStrLn "c"]
294-- >>> Stream.fold Fold.drain $ Stream.sequence s
297{-# INLINE sequence #-}
298sequence :: Monad m => Stream m (m a) -> Stream m a
301------------------------------------------------------------------------------
302-- Mapping side effects
303------------------------------------------------------------------------------
305-- | Tap the data flowing through a stream into a 'Fold'. For example, you may
306-- add a tap to log the contents flowing through the stream. The fold is used
307-- only for effects, its result is discarded.
312-- -----stream m a ---------------stream m a-----
316-- >>> s = Stream.enumerateFromTo 1 2
317-- >>> Stream.fold Fold.drain $ Stream.tap (Fold.drainMapM print) s
321-- Compare with 'trace'.
324tap :: Monad m => FL.Fold m a b -> Stream m a -> Stream m a
325tap f xs = fromStreamD $ D.tap f (toStreamD xs)
327-- | Apply a monadic function to each element flowing through the stream and
328-- discard the results.
330-- >>> s = Stream.enumerateFromTo 1 2
331-- >>> Stream.fold Fold.drain $ Stream.trace print s
335-- Compare with 'tap'.
338trace :: Monad m => (a -> m b) -> Stream m a -> Stream m a
339trace f = mapM (\x -> void (f x) >> return x)
341-- | Perform a side effect before yielding each element of the stream and
342-- discard the results.
344-- >>> s = Stream.enumerateFromTo 1 2
345-- >>> Stream.fold Fold.drain $ Stream.trace_ (print "got here") s
349-- Same as 'interspersePrefix_' but always serial.
355trace_ :: Monad m => m b -> Stream m a -> Stream m a
356trace_ eff = fromStreamD . D.mapM (\x -> eff >> return x) . toStreamD
358-------------------------------------------------------------------------------
360-------------------------------------------------------------------------------
362-- | @scanlMAfter' accumulate initial done stream@ is like 'scanlM'' except
363-- that it provides an additional @done@ function to be applied on the
364-- accumulator when the stream stops. The result of @done@ is also emitted in
367-- This function can be used to allocate a resource in the beginning of the
368-- scan and release it when the stream ends or to flush the internal state of
369-- the scan at the end.
373{-# INLINE scanlMAfter' #-}
381scanlMAfter' step initial done stream =
382 fromStreamD $ D.scanlMAfter' step initial done $ toStreamD stream
384------------------------------------------------------------------------------
385-- Scanning with a Fold
386------------------------------------------------------------------------------
388-- XXX It may be useful to have a version of scan where we can keep the
389-- accumulator independent of the value emitted. So that we do not necessarily
390-- have to keep a value in the accumulator which we are not using. We can pass
391-- an extraction function that will take the accumulator and the current value
392-- of the element and emit the next value in the stream. That will also make it
393-- possible to modify the accumulator after using it. In fact, the step function
394-- can return new accumulator and the value to be emitted. The signature would
395-- be more like mapAccumL.
397-- | Strict left scan. Scan a stream using the given monadic fold.
399-- >>> s = Stream.fromList [1..10]
400-- >>> Stream.fold Fold.toList $ Stream.takeWhile (< 10) $ Stream.scan Fold.sum s
403-- See also: 'usingStateT'
407-- >>> scanl' step z = Stream.scan (Fold.foldl' step z)
409-- Like 'map', 'scanl'' too is a one to one transformation,
410-- however it adds an extra element.
412-- >>> s = Stream.fromList [1,2,3,4]
413-- >>> Stream.fold Fold.toList $ scanl' (+) 0 s
416-- >>> Stream.fold Fold.toList $ scanl' (flip (:)) [] s
417-- [[],[1],[2,1],[3,2,1],[4,3,2,1]]
419-- The output of 'scanl'' is the initial value of the accumulator followed by
420-- all the intermediate steps and the final result of 'foldl''.
422-- By streaming the accumulated state after each fold step, we can share the
423-- state across multiple stages of stream composition. Each stage can modify or
424-- extend the state, do some processing with it and emit it for the next stage,
425-- thus modularizing the stream processing. This can be useful in
426-- stateful or event-driven programming.
428-- Consider the following monolithic example, computing the sum and the product
429-- of the elements in a stream in one go using a @foldl'@:
431-- >>> foldl' step z = Stream.fold (Fold.foldl' step z)
432-- >>> foldl' (\(s, p) x -> (s + x, p * x)) (0,1) s
435-- Using @scanl'@ we can make it modular by computing the sum in the first
436-- stage and passing it down to the next stage for computing the product:
439-- foldl' (\(_, p) (s, x) -> (s, p * x)) (0,1)
440-- $ scanl' (\(s, _) x -> (s + x, x)) (0,1)
441-- $ Stream.fromList [1,2,3,4]
445-- IMPORTANT: 'scanl'' evaluates the accumulator to WHNF. To avoid building
446-- lazy expressions inside the accumulator, it is recommended that a strict
447-- data structure is used for accumulator.
450scan :: Monad m => Fold m a b -> Stream m a -> Stream m b
451scan fld m = fromStreamD $ D.scanOnce fld $ toStreamD m
453-- | Like 'scan' but restarts scanning afresh when the scanning fold
456{-# INLINE scanMany #-}
457scanMany :: Monad m => Fold m a b -> Stream m a -> Stream m b
458scanMany fld m = fromStreamD $ D.scanMany fld $ toStreamD m
460------------------------------------------------------------------------------
462------------------------------------------------------------------------------
464-- | Modify a @Stream m a -> Stream m a@ stream transformation that accepts a
465-- predicate @(a -> b)@ to accept @((s, a) -> b)@ instead, provided a
466-- transformation @Stream m a -> Stream m (s, a)@. Convenient to filter with
469-- >>> filterWithIndex = Stream.with Stream.indexed Stream.filter
474 (Stream m a -> Stream m (s, a))
475 -> (((s, a) -> b) -> Stream m (s, a) -> Stream m (s, a))
476 -> (((s, a) -> b) -> Stream m a -> Stream m a)
477with f comb g = fmap snd . comb g . f
479-- | Include only those elements that pass a predicate.
481-- >>> filter p = Stream.filterM (return . p)
482-- >>> filter p = Stream.mapMaybe (\x -> if p x then Just x else Nothing)
483-- >>> filter p = Stream.scanMaybe (Fold.filtering p)
486filter :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
487-- filter p = scanMaybe (FL.filtering p)
488filter p m = fromStreamD $ D.filter p $ toStreamD m
490-- | Same as 'filter' but with a monadic predicate.
492-- >>> f p x = p x >>= \r -> return $ if r then Just x else Nothing
493-- >>> filterM p = Stream.mapMaybeM (f p)
495{-# INLINE filterM #-}
496filterM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
497filterM p m = fromStreamD $ D.filterM p $ toStreamD m
499-- | Drop repeated elements that are adjacent to each other using the supplied
500-- comparison function.
502-- >>> uniq = Stream.uniqBy (==)
504-- To strip duplicate path separators:
506-- >>> input = Stream.fromList "//a//b"
507-- >>> f x y = x == '/' && y == '/'
508-- >>> Stream.fold Fold.toList $ Stream.uniqBy f input
517 (a -> a -> Bool) -> Stream m a -> Stream m a
518-- uniqBy eq = scanMaybe (FL.uniqBy eq)
519uniqBy eq = catMaybes . rollingMap f
526 Just x -> if x `eq` curr then Nothing else Just curr
528-- | Drop repeated elements that are adjacent to each other.
530-- >>> uniq = Stream.uniqBy (==)
533uniq :: (Eq a, Monad m) => Stream m a -> Stream m a
534-- uniq = scanMaybe FL.uniq
535uniq = fromStreamD . D.uniq . toStreamD
537-- | Strip all leading and trailing occurrences of an element passing a
538-- predicate and make all other consecutive occurrences uniq.
540-- >> prune p = Stream.dropWhileAround p $ Stream.uniqBy (x y -> p x && p y)
543-- > Stream.prune isSpace (Stream.fromList " hello world! ")
553 -- (Monad m, Eq a) =>
554 (a -> Bool) -> Stream m a -> Stream m a
555prune = error "Not implemented yet!"
557-- Possible implementation:
559-- Stream.catMaybes . Stream.parseMany (Parser.groupBy (==) Fold.repeated)@
561-- 'Fold.repeated' should return 'Just' when repeated, and 'Nothing' for a
564-- | Emit only repeated elements, once.
567repeated :: -- (Monad m, Eq a) =>
568 Stream m a -> Stream m a
571-- | Deletes the first occurrence of the element in the stream that satisfies
572-- the given equality predicate.
574-- >>> input = Stream.fromList [1,3,3,5]
575-- >>> Stream.fold Fold.toList $ Stream.deleteBy (==) 3 input
578{-# INLINE deleteBy #-}
579deleteBy :: Monad m => (a -> a -> Bool) -> a -> Stream m a -> Stream m a
580-- deleteBy cmp x = scanMaybe (FL.deleteBy cmp x)
581deleteBy cmp x m = fromStreamD $ D.deleteBy cmp x (toStreamD m)
583------------------------------------------------------------------------------
585------------------------------------------------------------------------------
587-- | Same as 'takeWhile' but with a monadic predicate.
589{-# INLINE takeWhileM #-}
590takeWhileM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
591-- takeWhileM p = scanMaybe (FL.takingEndByM_ (\x -> not <$> p x))
592takeWhileM p m = fromStreamD $ D.takeWhileM p $ toStreamD m
594-- | Take all consecutive elements at the end of the stream for which the
597-- O(n) space, where n is the number elements taken.
600{-# INLINE takeWhileLast #-}
601takeWhileLast :: -- Monad m =>
602 (a -> Bool) -> Stream m a -> Stream m a
603takeWhileLast = undefined -- fromStreamD $ D.takeWhileLast n $ toStreamD m
605-- | Like 'takeWhile' and 'takeWhileLast' combined.
607-- O(n) space, where n is the number elements taken from the end.
610{-# INLINE takeWhileAround #-}
611takeWhileAround :: -- Monad m =>
612 (a -> Bool) -> Stream m a -> Stream m a
613takeWhileAround = undefined -- fromStreamD $ D.takeWhileAround n $ toStreamD m
615-- | Drop elements in the stream as long as the predicate succeeds and then
616-- take the rest of the stream.
618{-# INLINE dropWhile #-}
619dropWhile :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
620-- dropWhile p = scanMaybe (FL.droppingWhile p)
621dropWhile p m = fromStreamD $ D.dropWhile p $ toStreamD m
623-- | Same as 'dropWhile' but with a monadic predicate.
625{-# INLINE dropWhileM #-}
626dropWhileM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
627-- dropWhileM p = scanMaybe (FL.droppingWhileM p)
628dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m
630-- | Drop @n@ elements at the end of the stream.
632-- O(n) space, where n is the number elements dropped.
635{-# INLINE dropLast #-}
636dropLast :: -- Monad m =>
637 Int -> Stream m a -> Stream m a
638dropLast = undefined -- fromStreamD $ D.dropLast n $ toStreamD m
640-- | Drop all consecutive elements at the end of the stream for which the
643-- O(n) space, where n is the number elements dropped.
646{-# INLINE dropWhileLast #-}
647dropWhileLast :: -- Monad m =>
648 (a -> Bool) -> Stream m a -> Stream m a
649dropWhileLast = undefined -- fromStreamD $ D.dropWhileLast n $ toStreamD m
651-- | Like 'dropWhile' and 'dropWhileLast' combined.
653-- O(n) space, where n is the number elements dropped from the end.
656{-# INLINE dropWhileAround #-}
657dropWhileAround :: -- Monad m =>
658 (a -> Bool) -> Stream m a -> Stream m a
659dropWhileAround = undefined -- fromStreamD $ D.dropWhileAround n $ toStreamD m
661------------------------------------------------------------------------------
663------------------------------------------------------------------------------
665-- | @insertBy cmp elem stream@ inserts @elem@ before the first element in
666-- @stream@ that is less than @elem@ when compared using @cmp@.
668-- >>> insertBy cmp x = Stream.mergeBy cmp (Stream.fromPure x)
670-- >>> input = Stream.fromList [1,3,5]
671-- >>> Stream.fold Fold.toList $ Stream.insertBy compare 2 input
674{-# INLINE insertBy #-}
675insertBy ::Monad m => (a -> a -> Ordering) -> a -> Stream m a -> Stream m a
676insertBy cmp x m = fromStreamD $ D.insertBy cmp x (toStreamD m)
678-- | Insert a pure value between successive elements of a stream.
680-- >>> input = Stream.fromList "hello"
681-- >>> Stream.fold Fold.toList $ Stream.intersperse ',' input
684{-# INLINE intersperse #-}
685intersperse :: Monad m => a -> Stream m a -> Stream m a
686intersperse a = fromStreamD . D.intersperse a . toStreamD
688-- | Insert a side effect before consuming an element of a stream except the
691-- >>> input = Stream.fromList "hello"
692-- >>> Stream.fold Fold.drain $ Stream.trace putChar $ Stream.intersperseM_ (putChar '.') input
696{-# INLINE intersperseM_ #-}
697intersperseM_ :: Monad m => m b -> Stream m a -> Stream m a
698intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
700-- | Intersperse a monadic action into the input stream after every @n@
703-- >> input = Stream.fromList "hello"
704-- >> Stream.fold Fold.toList $ Stream.intersperseBySpan 2 (return ',') input
708{-# INLINE intersperseBySpan #-}
709intersperseBySpan :: -- Monad m =>
710 Int -> m a -> Stream m a -> Stream m a
711intersperseBySpan _n _f _xs = undefined
713-- | Insert an effect and its output after consuming an element of a stream.
715-- >>> input = Stream.fromList "hello"
716-- >>> Stream.fold Fold.toList $ Stream.trace putChar $ intersperseSuffix (putChar '.' >> return ',') input
717-- h.,e.,l.,l.,o.,"h,e,l,l,o,"
720{-# INLINE intersperseSuffix #-}
721intersperseSuffix :: Monad m => m a -> Stream m a -> Stream m a
722intersperseSuffix m = fromStreamD . D.intersperseSuffix m . toStreamD
724-- | Insert a side effect after consuming an element of a stream.
726-- >>> input = Stream.fromList "hello"
727-- >>> Stream.fold Fold.toList $ Stream.intersperseSuffix_ (threadDelay 1000000) input
732{-# INLINE intersperseSuffix_ #-}
733intersperseSuffix_ :: Monad m => m b -> Stream m a -> Stream m a
734intersperseSuffix_ m = fromStreamD . D.intersperseSuffix_ m . toStreamD
736-- XXX Use an offset argument, like tapOffsetEvery
738-- | Like 'intersperseSuffix' but intersperses an effectful action into the
739-- input stream after every @n@ elements and after the last element.
741-- >>> input = Stream.fromList "hello"
742-- >>> Stream.fold Fold.toList $ Stream.intersperseSuffixBySpan 2 (return ',') input
747{-# INLINE intersperseSuffixBySpan #-}
748intersperseSuffixBySpan :: Monad m
749 => Int -> m a -> Stream m a -> Stream m a
750intersperseSuffixBySpan n eff =
751 fromStreamD . D.intersperseSuffixBySpan n eff . toStreamD
753-- | Insert a side effect before consuming an element of a stream.
755-- >>> input = Stream.fromList "hello"
756-- >>> Stream.fold Fold.toList $ Stream.trace putChar $ Stream.interspersePrefix_ (putChar '.' >> return ',') input
763{-# INLINE interspersePrefix_ #-}
764interspersePrefix_ :: Monad m => m b -> Stream m a -> Stream m a
765interspersePrefix_ m = mapM (\x -> void m >> return x)
767------------------------------------------------------------------------------
769------------------------------------------------------------------------------
771-- | Introduce a delay of specified seconds between elements of the stream.
773-- >>> input = Stream.enumerateFromTo 1 3
774-- >>> Stream.fold (Fold.drainMapM print) $ Stream.delay 1 input
779-- >>> sleep n = liftIO $ threadDelay $ round $ n * 1000000
780-- >>> delay n = Stream.intersperseM_ $ sleep n
783delay :: MonadIO m => Double -> Stream m a -> Stream m a
784delay n = intersperseM_ $ liftIO $ threadDelay $ round $ n * 1000000
786-- | Introduce a delay of specified seconds after consuming an element of a
789-- >>> input = Stream.enumerateFromTo 1 3
790-- >>> Stream.fold (Fold.drainMapM print) $ Stream.delayPost 1 input
797{-# INLINE delayPost #-}
798delayPost :: MonadIO m => Double -> Stream m a -> Stream m a
799delayPost n = intersperseSuffix_ $ liftIO $ threadDelay $ round $ n * 1000000
801-- | Introduce a delay of specified seconds before consuming an element of a
804-- >>> input = Stream.enumerateFromTo 1 3
805-- >>> Stream.fold (Fold.drainMapM print) $ Stream.delayPre 1 input
812{-# INLINE delayPre #-}
813delayPre :: MonadIO m => Double -> Stream m a -> Stream m a
814delayPre n = trace_ $ liftIO $ threadDelay $ round $ n * 1000000
816------------------------------------------------------------------------------
817-- Reorder in sequence
818------------------------------------------------------------------------------
820-- | Buffer until the next element in sequence arrives. The function argument
821-- determines the difference in sequence numbers. This could be useful in
822-- implementing sequenced streams, for example, TCP reassembly.
826{-# INLINE reassembleBy #-}
833reassembleBy = undefined
835------------------------------------------------------------------------------
837------------------------------------------------------------------------------
840-- >>> f = Fold.foldl' (\(i, _) x -> (i + 1, x)) (-1,undefined)
841-- >>> indexed = Stream.postscan f
842-- >>> indexed = Stream.zipWith (,) (Stream.enumerateFrom 0)
843-- >>> indexedR n = fmap (\(i, a) -> (n - i, a)) . indexed
845-- Pair each element in a stream with its index, starting from index 0.
847-- >>> Stream.fold Fold.toList $ Stream.indexed $ Stream.fromList "hello"
848-- [(0,'h'),(1,'e'),(2,'l'),(3,'l'),(4,'o')]
850{-# INLINE indexed #-}
851indexed :: Monad m => Stream m a -> Stream m (Int, a)
852-- indexed = scanMaybe FL.indexing
853indexed = fromStreamD . D.indexed . toStreamD
856-- >>> f n = Fold.foldl' (\(i, _) x -> (i - 1, x)) (n + 1,undefined)
857-- >>> indexedR n = Stream.postscan (f n)
859-- >>> s n = Stream.enumerateFromThen n (n - 1)
860-- >>> indexedR n = Stream.zipWith (,) (s n)
862-- Pair each element in a stream with its index, starting from the
863-- given index @n@ and counting down.
865-- >>> Stream.fold Fold.toList $ Stream.indexedR 10 $ Stream.fromList "hello"
866-- [(10,'h'),(9,'e'),(8,'l'),(7,'l'),(6,'o')]
868{-# INLINE indexedR #-}
869indexedR :: Monad m => Int -> Stream m a -> Stream m (Int, a)
870-- indexedR n = scanMaybe (FL.indexingRev n)
871indexedR n = fromStreamD . D.indexedR n . toStreamD
873-------------------------------------------------------------------------------
875-------------------------------------------------------------------------------
877-- Note: The timestamp stream must be the second stream in the zip so that the
878-- timestamp is generated after generating the stream element and not before.
879-- If we do not do that then the following example will generate the same
880-- timestamp for first two elements:
882-- Stream.fold Fold.toList $ Stream.timestamped $ Stream.delay $ Stream.enumerateFromTo 1 3
884-- | Pair each element in a stream with an absolute timestamp, using a clock of
885-- specified granularity. The timestamp is generated just before the element
888-- >>> Stream.fold Fold.toList $ Stream.timestampWith 0.01 $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
889-- [(AbsTime (TimeSpec {sec = ..., nsec = ...}),1),(AbsTime (TimeSpec {sec = ..., nsec = ...}),2),(AbsTime (TimeSpec {sec = ..., nsec = ...}),3)]
893{-# INLINE timestampWith #-}
894timestampWith :: (MonadIO m)
895 => Double -> Stream m a -> Stream m (AbsTime, a)
896timestampWith g stream = zipWith (flip (,)) stream (absTimesWith g)
898-- TBD: check performance vs a custom implementation without using zipWith.
902{-# INLINE timestamped #-}
903timestamped :: (MonadIO m)
904 => Stream m a -> Stream m (AbsTime, a)
905timestamped = timestampWith 0.01
907-- | Pair each element in a stream with relative times starting from 0, using a
908-- clock with the specified granularity. The time is measured just before the
909-- element is consumed.
911-- >>> Stream.fold Fold.toList $ Stream.timeIndexWith 0.01 $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
912-- [(RelTime64 (NanoSecond64 ...),1),(RelTime64 (NanoSecond64 ...),2),(RelTime64 (NanoSecond64 ...),3)]
916{-# INLINE timeIndexWith #-}
917timeIndexWith :: (MonadIO m)
918 => Double -> Stream m a -> Stream m (RelTime64, a)
919timeIndexWith g stream = zipWith (flip (,)) stream (relTimesWith g)
921-- | Pair each element in a stream with relative times starting from 0, using a
922-- 10 ms granularity clock. The time is measured just before the element is
925-- >>> Stream.fold Fold.toList $ Stream.timeIndexed $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
926-- [(RelTime64 (NanoSecond64 ...),1),(RelTime64 (NanoSecond64 ...),2),(RelTime64 (NanoSecond64 ...),3)]
930{-# INLINE timeIndexed #-}
931timeIndexed :: (MonadIO m)
932 => Stream m a -> Stream m (RelTime64, a)
933timeIndexed = timeIndexWith 0.01
935------------------------------------------------------------------------------
937------------------------------------------------------------------------------
939-- | Find all the indices where the value of the element in the stream is equal
940-- to the given value.
942-- >>> elemIndices a = Stream.findIndices (== a)
944{-# INLINE elemIndices #-}
945elemIndices :: (Monad m, Eq a) => a -> Stream m a -> Stream m Int
946elemIndices a = findIndices (== a)
948------------------------------------------------------------------------------
950------------------------------------------------------------------------------
952-- XXX this is not a one-to-one map so calling it map may not be right.
953-- We can perhaps call it zipWithTail or rollWith.
955-- | Apply a function on every two successive elements of a stream. The first
956-- argument of the map function is the previous element and the second argument
957-- is the current element. When the current element is the first element, the
958-- previous element is 'Nothing'.
962{-# INLINE rollingMap #-}
963rollingMap :: Monad m => (Maybe a -> a -> b) -> Stream m a -> Stream m b
964-- rollingMap f = scanMaybe (FL.slide2 $ Window.rollingMap f)
965rollingMap f m = fromStreamD $ D.rollingMap f $ toStreamD m
967-- | Like 'rollingMap' but with an effectful map function.
971{-# INLINE rollingMapM #-}
972rollingMapM :: Monad m => (Maybe a -> a -> m b) -> Stream m a -> Stream m b
973-- rollingMapM f = scanMaybe (FL.slide2 $ Window.rollingMapM f)
974rollingMapM f m = fromStreamD $ D.rollingMapM f $ toStreamD m
976-- | Like 'rollingMap' but requires at least two elements in the stream,
977-- returns an empty stream otherwise.
979-- This is the stream equivalent of the list idiom @zipWith f xs (tail xs)@.
983{-# INLINE rollingMap2 #-}
984rollingMap2 :: Monad m => (a -> a -> b) -> Stream m a -> Stream m b
985rollingMap2 f m = fromStreamD $ D.rollingMap2 f $ toStreamD m
987------------------------------------------------------------------------------
989------------------------------------------------------------------------------
991-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'
992-- elements, and return a stream of values extracted from 'Just'.
996-- >>> mapMaybe f = Stream.catMaybes . fmap f
998{-# INLINE mapMaybe #-}
999mapMaybe :: Monad m => (a -> Maybe b) -> Stream m a -> Stream m b
1000mapMaybe f m = fromStreamD $ D.mapMaybe f $ toStreamD m
1002-- | Like 'mapMaybe' but maps a monadic function.
1006-- >>> mapMaybeM f = Stream.catMaybes . Stream.mapM f
1008-- >>> mapM f = Stream.mapMaybeM (\x -> Just <$> f x)
1010{-# INLINE_EARLY mapMaybeM #-}
1012 => (a -> m (Maybe b)) -> Stream m a -> Stream m b
1013mapMaybeM f = fmap fromJust . filter isJust . mapM f
1015------------------------------------------------------------------------------
1017------------------------------------------------------------------------------
1019-- | Discard 'Right's and unwrap 'Left's in an 'Either' stream.
1021-- >>> lefts = fmap (fromLeft undefined) . Stream.filter isLeft
1026lefts :: Monad m => Stream m (Either a b) -> Stream m a
1027lefts = fmap (fromLeft undefined) . filter isLeft
1029-- | Discard 'Left's and unwrap 'Right's in an 'Either' stream.
1031-- >>> rights = fmap (fromRight undefined) . Stream.filter isRight
1035{-# INLINE rights #-}
1036rights :: Monad m => Stream m (Either a b) -> Stream m b
1037rights = fmap (fromRight undefined) . filter isRight
1039-- | Remove the either wrapper and flatten both lefts and as well as rights in
1040-- the output stream.
1042-- >>> both = fmap (either id id)
1047both :: Monad m => Stream m (Either a a) -> Stream m a
1048both = fmap (either id id)
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list