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.
70 -- * Position Indexing
81 , findIndices -- XXX indicesBy
82 , elemIndices -- XXX indicesOf
85 -- | Map using the previous element.
92 -- * Inserting Elements
93 -- | Produce a superset of the stream. This is the opposite of
94 -- filtering/sampling. We can always use concatMap and scan for inserting
95 -- but these combinators are more efficient and convenient.
97 -- Element agnostic (Opposite of sampling)
99 , intersperseM -- XXX naming
103 , intersperseSuffixBySpan
105 -- , interspersePrefix
106 -- , interspersePrefixBySpan
108 -- * Inserting Side Effects/Time
109 , intersperseM_ -- XXX naming
116 -- * Element Aware Insertion
117 -- | Opposite of filtering
119 -- , intersperseByBefore
120 -- , intersperseByAfter
122 -- Fold and Unfold, Buffering
130 -- Move these to Streamly.Data.Either.Stream?
139import Control.Concurrent (threadDelay)
140import Control.Monad (void)
141import Control.Monad.IO.Class (MonadIO (liftIO))
142import Control.Monad.Trans.Class (MonadTrans(..))
143import Data.Either (fromLeft, isLeft, isRight, fromRight)
144import Data.Maybe (isJust, fromJust)
146import Streamly.Internal.Data.Fold.Type (Fold)
147import Streamly.Internal.Data.Pipe (Pipe)
148import Streamly.Internal.Data.Time.Units (AbsTime, RelTime64)
150import qualified Streamly.Internal.Data.Fold as FL
151-- import qualified Streamly.Internal.Data.Fold.Window as Window
152import qualified Streamly.Internal.Data.Stream.StreamD.Transform as D
153import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
154import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
157import Streamly.Internal.Data.Stream.Bottom
158import Streamly.Internal.Data.Stream.Type
161 ( filter, drop, dropWhile, take, takeWhile, foldr, map, mapM, sequence
162 , reverse, foldr1 , repeat, scanl, scanl1, zipWith)
167-- >>> import Control.Concurrent (threadDelay)
168-- >>> import Control.Monad.IO.Class (MonadIO (liftIO))
169-- >>> import Control.Monad.Trans (lift)
170-- >>> import Control.Monad.Trans.Identity (runIdentityT)
171-- >>> import Data.Either (fromLeft, fromRight, isLeft, isRight, either)
172-- >>> import Data.Function ((&))
173-- >>> import Data.Maybe (fromJust, isJust)
174-- >>> import Prelude hiding (filter, drop, dropWhile, take, takeWhile, foldr, map, mapM, sequence, reverse, foldr1 , scanl, scanl1)
175-- >>> import Streamly.Internal.Data.Stream as Stream
176-- >>> import qualified Streamly.Data.Fold as Fold
177-- >>> import qualified Streamly.Internal.Data.Fold as Fold
178-- >>> import qualified Streamly.Internal.Data.Fold.Window as Window
179-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
180-- >>> import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
182-- >>> hSetBuffering stdout LineBuffering
184-- XXX because of the use of D.cons for appending, folds and scans have
185-- quadratic complexity when iterated over a stream. We should use StreamK for
186-- linear performance on iteration.
188------------------------------------------------------------------------------
190------------------------------------------------------------------------------
192-- | Use a 'Pipe' to transform a stream.
196{-# INLINE transform #-}
197transform :: Monad m => Pipe m a b -> Stream m a -> Stream m b
198transform pipe xs = fromStreamD $ D.transform pipe (toStreamD xs)
200------------------------------------------------------------------------------
201-- Transformation Folds
202------------------------------------------------------------------------------
204-- | Right fold to a streaming monad.
206-- > foldrS Stream.cons Stream.nil === id
208-- 'foldrS' can be used to perform stateless stream to stream transformations
209-- like map and filter in general. It can be coupled with a scan to perform
210-- stateful transformations. However, note that the custom map and filter
211-- routines can be much more efficient than this due to better stream fusion.
213-- >>> input = Stream.fromList [1..5]
214-- >>> Stream.fold Fold.toList $ Stream.foldrS Stream.cons Stream.nil input
217-- Find if any element in the stream is 'True':
219-- >>> step x xs = if odd x then return True else xs
220-- >>> input = Stream.fromList (2:4:5:undefined) :: Stream IO Int
221-- >>> Stream.fold Fold.toList $ Stream.foldrS step (return False) input
224-- Map (+2) on odd elements and filter out the even elements:
226-- >>> step x xs = if odd x then (x + 2) `Stream.cons` xs else xs
227-- >>> input = Stream.fromList [1..5] :: Stream IO Int
228-- >>> Stream.fold Fold.toList $ Stream.foldrS step Stream.nil input
234 (a -> Stream m b -> Stream m b)
241 (\y ys -> toStreamK $ f y (fromStreamK ys))
245-- | Right fold to a transformer monad. This is the most general right fold
246-- function. 'foldrS' is a special case of 'foldrT', however 'foldrS'
247-- implementation can be more efficient:
249-- >>> foldrS = foldrT
251-- >>> step f x xs = lift $ f x (runIdentityT xs)
252-- >>> foldrM f z s = runIdentityT $ foldrT (step f) (lift z) s
254-- 'foldrT' can be used to translate streamly streams to other transformer
255-- monads e.g. to a different streaming type.
259foldrT :: (Monad m, Monad (s m), MonadTrans s)
260 => (a -> s m b -> s m b) -> s m b -> Stream m a -> s m b
261foldrT f z s = D.foldrT f z (toStreamD s)
263------------------------------------------------------------------------------
264-- Transformation by Mapping
265------------------------------------------------------------------------------
268-- >>> mapM f = Stream.sequence . fmap f
270-- Apply a monadic function to each element of the stream and replace it with
271-- the output of the resulting action.
273-- >>> s = Stream.fromList ["a", "b", "c"]
274-- >>> Stream.fold Fold.drain $ Stream.mapM putStr s
279mapM :: Monad m => (a -> m b) -> Stream m a -> Stream m b
280mapM f m = fromStreamK $ D.toStreamK $ D.mapM f $ toStreamD m
283-- >>> sequence = Stream.mapM id
285-- Replace the elements of a stream of monadic actions with the outputs of
288-- >>> s = Stream.fromList [putStr "a", putStr "b", putStrLn "c"]
289-- >>> Stream.fold Fold.drain $ Stream.sequence s
293{-# INLINE sequence #-}
294sequence :: Monad m => Stream m (m a) -> Stream m a
297------------------------------------------------------------------------------
298-- Mapping side effects
299------------------------------------------------------------------------------
301-- | Tap the data flowing through a stream into a 'Fold'. For example, you may
302-- add a tap to log the contents flowing through the stream. The fold is used
303-- only for effects, its result is discarded.
308-- -----stream m a ---------------stream m a-----
312-- >>> s = Stream.enumerateFromTo 1 2
313-- >>> Stream.fold Fold.drain $ Stream.tap (Fold.drainBy print) s
317-- Compare with 'trace'.
321tap :: Monad m => FL.Fold m a b -> Stream m a -> Stream m a
322tap f xs = fromStreamD $ D.tap f (toStreamD xs)
324-- | Apply a monadic function to each element flowing through the stream and
325-- discard the results.
327-- >>> s = Stream.enumerateFromTo 1 2
328-- >>> Stream.fold Fold.drain $ Stream.trace print s
332-- Compare with 'tap'.
335trace :: Monad m => (a -> m b) -> Stream m a -> Stream m a
336trace f = mapM (\x -> void (f x) >> return x)
338-- | Perform a side effect before yielding each element of the stream and
339-- discard the results.
341-- >>> s = Stream.enumerateFromTo 1 2
342-- >>> Stream.fold Fold.drain $ Stream.trace_ (print "got here") s
346-- Same as 'interspersePrefix_' but always serial.
352trace_ :: Monad m => m b -> Stream m a -> Stream m a
353trace_ eff = fromStreamD . D.mapM (\x -> eff >> return x) . toStreamD
355-------------------------------------------------------------------------------
357-------------------------------------------------------------------------------
359-- | @scanlMAfter' accumulate initial done stream@ is like 'scanlM'' except
360-- that it provides an additional @done@ function to be applied on the
361-- accumulator when the stream stops. The result of @done@ is also emitted in
364-- This function can be used to allocate a resource in the beginning of the
365-- scan and release it when the stream ends or to flush the internal state of
366-- the scan at the end.
370{-# INLINE scanlMAfter' #-}
378scanlMAfter' step initial done stream =
379 fromStreamD $ D.scanlMAfter' step initial done $ toStreamD stream
381------------------------------------------------------------------------------
382-- Scanning with a Fold
383------------------------------------------------------------------------------
385-- XXX It may be useful to have a version of scan where we can keep the
386-- accumulator independent of the value emitted. So that we do not necessarily
387-- have to keep a value in the accumulator which we are not using. We can pass
388-- an extraction function that will take the accumulator and the current value
389-- of the element and emit the next value in the stream. That will also make it
390-- possible to modify the accumulator after using it. In fact, the step function
391-- can return new accumulator and the value to be emitted. The signature would
392-- be more like mapAccumL.
394-- | Strict left scan. Scan a stream using the given monadic fold.
396-- >>> s = Stream.fromList [1..10]
397-- >>> Stream.fold Fold.toList $ Stream.takeWhile (< 10) $ Stream.scan Fold.sum s
400-- See also: 'usingStateT'
405-- >>> scanl' step z = Stream.scan (Fold.foldl' step z)
407-- Like 'map', 'scanl'' too is a one to one transformation,
408-- however it adds an extra element.
410-- >>> s = Stream.fromList [1,2,3,4]
411-- >>> Stream.fold Fold.toList $ scanl' (+) 0 s
414-- >>> Stream.fold Fold.toList $ scanl' (flip (:)) [] s
415-- [[],[1],[2,1],[3,2,1],[4,3,2,1]]
417-- The output of 'scanl'' is the initial value of the accumulator followed by
418-- all the intermediate steps and the final result of 'foldl''.
420-- By streaming the accumulated state after each fold step, we can share the
421-- state across multiple stages of stream composition. Each stage can modify or
422-- extend the state, do some processing with it and emit it for the next stage,
423-- thus modularizing the stream processing. This can be useful in
424-- stateful or event-driven programming.
426-- Consider the following monolithic example, computing the sum and the product
427-- of the elements in a stream in one go using a @foldl'@:
429-- >>> foldl' step z = Stream.fold (Fold.foldl' step z)
430-- >>> foldl' (\(s, p) x -> (s + x, p * x)) (0,1) s
433-- Using @scanl'@ we can make it modular by computing the sum in the first
434-- stage and passing it down to the next stage for computing the product:
437-- foldl' (\(_, p) (s, x) -> (s, p * x)) (0,1)
438-- $ scanl' (\(s, _) x -> (s + x, x)) (0,1)
439-- $ Stream.fromList [1,2,3,4]
443-- IMPORTANT: 'scanl'' evaluates the accumulator to WHNF. To avoid building
444-- lazy expressions inside the accumulator, it is recommended that a strict
445-- data structure is used for accumulator.
449scan :: Monad m => Fold m a b -> Stream m a -> Stream m b
450scan fld m = fromStreamD $ D.scanOnce fld $ toStreamD m
452-- | 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)
487filter :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
488-- filter p = scanMaybe (FL.filtering p)
489filter p m = fromStreamD $ D.filter p $ toStreamD m
491-- | Same as 'filter' but with a monadic predicate.
493-- >>> f p x = p x >>= \r -> return $ if r then Just x else Nothing
494-- >>> filterM p = Stream.mapMaybeM (f p)
497{-# INLINE filterM #-}
498filterM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
499filterM p m = fromStreamD $ D.filterM p $ toStreamD m
501-- | Drop repeated elements that are adjacent to each other using the supplied
502-- comparison function.
504-- >>> uniq = Stream.uniqBy (==)
506-- To strip duplicate path separators:
508-- >>> input = Stream.fromList "//a//b"
509-- >>> f x y = x == '/' && y == '/'
510-- >>> Stream.fold Fold.toList $ Stream.uniqBy f input
521 (a -> a -> Bool) -> Stream m a -> Stream m a
522-- uniqBy eq = scanMaybe (FL.uniqBy eq)
523uniqBy eq = catMaybes . rollingMap f
530 Just x -> if x `eq` curr then Nothing else Just curr
532-- | Drop repeated elements that are adjacent to each other.
534-- >>> uniq = Stream.uniqBy (==)
538uniq :: (Eq a, Monad m) => Stream m a -> Stream m a
539-- uniq = scanMaybe FL.uniq
540uniq = fromStreamD . D.uniq . toStreamD
542-- | Strip all leading and trailing occurrences of an element passing a
543-- predicate and make all other consecutive occurrences uniq.
545-- >> prune p = Stream.dropWhileAround p $ Stream.uniqBy (x y -> p x && p y)
548-- > Stream.prune isSpace (Stream.fromList " hello world! ")
558 -- (Monad m, Eq a) =>
559 (a -> Bool) -> Stream m a -> Stream m a
560prune = error "Not implemented yet!"
562-- Possible implementation:
564-- Stream.catMaybes . Stream.parseMany (Parser.groupBy (==) Fold.repeated)@
566-- 'Fold.repeated' should return 'Just' when repeated, and 'Nothing' for a
569-- | Emit only repeated elements, once.
572repeated :: -- (Monad m, Eq a) =>
573 Stream m a -> Stream m a
576-- | Deletes the first occurrence of the element in the stream that satisfies
577-- the given equality predicate.
579-- >>> input = Stream.fromList [1,3,3,5]
580-- >>> Stream.fold Fold.toList $ Stream.deleteBy (==) 3 input
584{-# INLINE deleteBy #-}
585deleteBy :: Monad m => (a -> a -> Bool) -> a -> Stream m a -> Stream m a
586-- deleteBy cmp x = scanMaybe (FL.deleteBy cmp x)
587deleteBy cmp x m = fromStreamD $ D.deleteBy cmp x (toStreamD m)
589------------------------------------------------------------------------------
591------------------------------------------------------------------------------
593-- | Same as 'takeWhile' but with a monadic predicate.
596{-# INLINE takeWhileM #-}
597takeWhileM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
598-- takeWhileM p = scanMaybe (FL.takingEndByM_ (\x -> not <$> p x))
599takeWhileM p m = fromStreamD $ D.takeWhileM p $ toStreamD m
601-- | Drop elements in the stream as long as the predicate succeeds and then
602-- take the rest of the stream.
605{-# INLINE dropWhile #-}
606dropWhile :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
607-- dropWhile p = scanMaybe (FL.droppingWhile p)
608dropWhile p m = fromStreamD $ D.dropWhile p $ toStreamD m
610-- | Same as 'dropWhile' but with a monadic predicate.
613{-# INLINE dropWhileM #-}
614dropWhileM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
615-- dropWhileM p = scanMaybe (FL.droppingWhileM p)
616dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m
618------------------------------------------------------------------------------
620------------------------------------------------------------------------------
622-- | @insertBy cmp elem stream@ inserts @elem@ before the first element in
623-- @stream@ that is less than @elem@ when compared using @cmp@.
625-- >>> insertBy cmp x = Stream.mergeBy cmp (Stream.fromPure x)
627-- >>> input = Stream.fromList [1,3,5]
628-- >>> Stream.fold Fold.toList $ Stream.insertBy compare 2 input
632{-# INLINE insertBy #-}
633insertBy ::Monad m => (a -> a -> Ordering) -> a -> Stream m a -> Stream m a
634insertBy cmp x m = fromStreamD $ D.insertBy cmp x (toStreamD m)
636-- | Insert a pure value between successive elements of a stream.
638-- >>> input = Stream.fromList "hello"
639-- >>> Stream.fold Fold.toList $ Stream.intersperse ',' input
643{-# INLINE intersperse #-}
644intersperse :: Monad m => a -> Stream m a -> Stream m a
645intersperse a = fromStreamD . D.intersperse a . toStreamD
647-- | Insert a side effect before consuming an element of a stream except the
650-- >>> input = Stream.fromList "hello"
651-- >>> Stream.fold Fold.drain $ Stream.trace putChar $ Stream.intersperseM_ (putChar '.') input
655{-# INLINE intersperseM_ #-}
656intersperseM_ :: Monad m => m b -> Stream m a -> Stream m a
657intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
659-- | Intersperse a monadic action into the input stream after every @n@
662-- >> input = Stream.fromList "hello"
663-- >> Stream.fold Fold.toList $ Stream.intersperseBySpan 2 (return ',') input
667{-# INLINE intersperseBySpan #-}
668intersperseBySpan :: -- Monad m =>
669 Int -> m a -> Stream m a -> Stream m a
670intersperseBySpan _n _f _xs = undefined
672-- | Insert an effect and its output after consuming an element of a stream.
674-- >>> input = Stream.fromList "hello"
675-- >>> Stream.fold Fold.toList $ Stream.trace putChar $ intersperseSuffix (putChar '.' >> return ',') input
676-- h.,e.,l.,l.,o.,"h,e,l,l,o,"
679{-# INLINE intersperseSuffix #-}
680intersperseSuffix :: Monad m => m a -> Stream m a -> Stream m a
681intersperseSuffix m = fromStreamD . D.intersperseSuffix m . toStreamD
683-- | Insert a side effect after consuming an element of a stream.
685-- >>> input = Stream.fromList "hello"
686-- >>> Stream.fold Fold.toList $ Stream.intersperseSuffix_ (threadDelay 1000000) input
691{-# INLINE intersperseSuffix_ #-}
692intersperseSuffix_ :: Monad m => m b -> Stream m a -> Stream m a
693intersperseSuffix_ m = fromStreamD . D.intersperseSuffix_ m . toStreamD
695-- XXX Use an offset argument, like tapOffsetEvery
697-- | Like 'intersperseSuffix' but intersperses an effectful action into the
698-- input stream after every @n@ elements and after the last element.
700-- >>> input = Stream.fromList "hello"
701-- >>> Stream.fold Fold.toList $ Stream.intersperseSuffixBySpan 2 (return ',') input
706{-# INLINE intersperseSuffixBySpan #-}
707intersperseSuffixBySpan :: Monad m
708 => Int -> m a -> Stream m a -> Stream m a
709intersperseSuffixBySpan n eff =
710 fromStreamD . D.intersperseSuffixBySpan n eff . toStreamD
712-- | Insert a side effect before consuming an element of a stream.
714-- >>> input = Stream.fromList "hello"
715-- >>> Stream.fold Fold.toList $ Stream.trace putChar $ Stream.interspersePrefix_ (putChar '.' >> return ',') input
722{-# INLINE interspersePrefix_ #-}
723interspersePrefix_ :: Monad m => m b -> Stream m a -> Stream m a
724interspersePrefix_ m = mapM (\x -> void m >> return x)
726------------------------------------------------------------------------------
728------------------------------------------------------------------------------
730-- | Introduce a delay of specified seconds between elements of the stream.
732-- >>> input = Stream.enumerateFromTo 1 3
733-- >>> Stream.fold (Fold.drainBy print) $ Stream.delay 1 input
738-- >>> sleep n = liftIO $ threadDelay $ round $ n * 1000000
739-- >>> delay n = Stream.intersperseM_ $ sleep n
743delay :: MonadIO m => Double -> Stream m a -> Stream m a
744delay n = intersperseM_ $ liftIO $ threadDelay $ round $ n * 1000000
746-- | Introduce a delay of specified seconds after consuming an element of a
749-- >>> input = Stream.enumerateFromTo 1 3
750-- >>> Stream.fold (Fold.drainBy print) $ Stream.delayPost 1 input
757{-# INLINE delayPost #-}
758delayPost :: MonadIO m => Double -> Stream m a -> Stream m a
759delayPost n = intersperseSuffix_ $ liftIO $ threadDelay $ round $ n * 1000000
761-- | Introduce a delay of specified seconds before consuming an element of a
764-- >>> input = Stream.enumerateFromTo 1 3
765-- >>> Stream.fold (Fold.drainBy print) $ Stream.delayPre 1 input
772{-# INLINE delayPre #-}
773delayPre :: MonadIO m => Double -> Stream m a -> Stream m a
774delayPre n = trace_ $ liftIO $ threadDelay $ round $ n * 1000000
776------------------------------------------------------------------------------
777-- Reorder in sequence
778------------------------------------------------------------------------------
780-- | Buffer until the next element in sequence arrives. The function argument
781-- determines the difference in sequence numbers. This could be useful in
782-- implementing sequenced streams, for example, TCP reassembly.
786{-# INLINE reassembleBy #-}
793reassembleBy = undefined
795------------------------------------------------------------------------------
797------------------------------------------------------------------------------
800-- >>> f = Fold.foldl' (\(i, _) x -> (i + 1, x)) (-1,undefined)
801-- >>> indexed = Stream.postscan f
802-- >>> indexed = Stream.zipWith (,) (Stream.enumerateFrom 0)
804-- Pair each element in a stream with its index, starting from index 0.
806-- >>> Stream.fold Fold.toList $ Stream.indexed $ Stream.fromList "hello"
807-- [(0,'h'),(1,'e'),(2,'l'),(3,'l'),(4,'o')]
810{-# INLINE indexed #-}
811indexed :: Monad m => Stream m a -> Stream m (Int, a)
812-- indexed = scanMaybe FL.indexing
813indexed = fromStreamD . D.indexed . toStreamD
816-- >>> f n = Fold.foldl' (\(i, _) x -> (i - 1, x)) (n + 1,undefined)
817-- >>> indexedR n = Stream.postscan (f n)
819-- >>> s n = Stream.enumerateFromThen n (n - 1)
820-- >>> indexedR n = Stream.zipWith (,) (s n)
822-- Pair each element in a stream with its index, starting from the
823-- given index @n@ and counting down.
825-- >>> Stream.fold Fold.toList $ Stream.indexedR 10 $ Stream.fromList "hello"
826-- [(10,'h'),(9,'e'),(8,'l'),(7,'l'),(6,'o')]
829{-# INLINE indexedR #-}
830indexedR :: Monad m => Int -> Stream m a -> Stream m (Int, a)
831-- indexedR n = scanMaybe (FL.indexingRev n)
832indexedR n = fromStreamD . D.indexedR n . toStreamD
834-------------------------------------------------------------------------------
836-------------------------------------------------------------------------------
838-- Note: The timestamp stream must be the second stream in the zip so that the
839-- timestamp is generated after generating the stream element and not before.
840-- If we do not do that then the following example will generate the same
841-- timestamp for first two elements:
843-- Stream.fold Fold.toList $ Stream.timestamped $ Stream.delay $ Stream.enumerateFromTo 1 3
845-- | Pair each element in a stream with an absolute timestamp, using a clock of
846-- specified granularity. The timestamp is generated just before the element
849-- >>> Stream.fold Fold.toList $ Stream.timestampWith 0.01 $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
850-- [(AbsTime (TimeSpec {sec = ..., nsec = ...}),1),(AbsTime (TimeSpec {sec = ..., nsec = ...}),2),(AbsTime (TimeSpec {sec = ..., nsec = ...}),3)]
854{-# INLINE timestampWith #-}
855timestampWith :: (MonadIO m)
856 => Double -> Stream m a -> Stream m(AbsTime, a)
857timestampWith g stream = zipWith (flip (,)) stream (absTimesWith g)
859-- TBD: check performance vs a custom implementation without using zipWith.
863{-# INLINE timestamped #-}
864timestamped :: (MonadIO m)
865 => Stream m a -> Stream m(AbsTime, a)
866timestamped = timestampWith 0.01
868-- | Pair each element in a stream with relative times starting from 0, using a
869-- clock with the specified granularity. The time is measured just before the
870-- element is consumed.
872-- >>> Stream.fold Fold.toList $ Stream.timeIndexWith 0.01 $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
873-- [(RelTime64 (NanoSecond64 ...),1),(RelTime64 (NanoSecond64 ...),2),(RelTime64 (NanoSecond64 ...),3)]
877{-# INLINE timeIndexWith #-}
878timeIndexWith :: (MonadIO m)
879 => Double -> Stream m a -> Stream m(RelTime64, a)
880timeIndexWith g stream = zipWith (flip (,)) stream (relTimesWith g)
882-- | Pair each element in a stream with relative times starting from 0, using a
883-- 10 ms granularity clock. The time is measured just before the element is
886-- >>> Stream.fold Fold.toList $ Stream.timeIndexed $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
887-- [(RelTime64 (NanoSecond64 ...),1),(RelTime64 (NanoSecond64 ...),2),(RelTime64 (NanoSecond64 ...),3)]
891{-# INLINE timeIndexed #-}
892timeIndexed :: (MonadIO m)
893 => Stream m a -> Stream m(RelTime64, a)
894timeIndexed = timeIndexWith 0.01
896------------------------------------------------------------------------------
898------------------------------------------------------------------------------
900-- | Find all the indices where the value of the element in the stream is equal
901-- to the given value.
903-- >>> elemIndices a = Stream.findIndices (== a)
906{-# INLINE elemIndices #-}
907elemIndices :: (Monad m, Eq a) => a -> Stream m a -> Stream m Int
908elemIndices a = findIndices (== a)
910------------------------------------------------------------------------------
912------------------------------------------------------------------------------
914-- XXX this is not a one-to-one map so calling it map may not be right.
915-- We can perhaps call it zipWithTail or rollWith.
917-- | Apply a function on every two successive elements of a stream. The first
918-- argument of the map function is the previous element and the second argument
919-- is the current element. When the current element is the first element, the
920-- previous element is 'Nothing'.
924{-# INLINE rollingMap #-}
925rollingMap :: Monad m => (Maybe a -> a -> b) -> Stream m a -> Stream m b
926-- rollingMap f = scanMaybe (FL.slide2 $ Window.rollingMap f)
927rollingMap f m = fromStreamD $ D.rollingMap f $ toStreamD m
929-- | Like 'rollingMap' but with an effectful map function.
933{-# INLINE rollingMapM #-}
934rollingMapM :: Monad m => (Maybe a -> a -> m b) -> Stream m a -> Stream m b
935-- rollingMapM f = scanMaybe (FL.slide2 $ Window.rollingMapM f)
936rollingMapM f m = fromStreamD $ D.rollingMapM f $ toStreamD m
938-- | Like 'rollingMap' but requires at least two elements in the stream,
939-- returns an empty stream otherwise.
941-- This is the stream equivalent of the list idiom @zipWith f xs (tail xs)@.
945{-# INLINE rollingMap2 #-}
946rollingMap2 :: Monad m => (a -> a -> b) -> Stream m a -> Stream m b
947rollingMap2 f m = fromStreamD $ D.rollingMap2 f $ toStreamD m
949------------------------------------------------------------------------------
951------------------------------------------------------------------------------
953-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'
954-- elements, and return a stream of values extracted from 'Just'.
958-- >>> mapMaybe f = Stream.catMaybes . fmap f
961{-# INLINE mapMaybe #-}
962mapMaybe :: Monad m => (a -> Maybe b) -> Stream m a -> Stream m b
963mapMaybe f m = fromStreamD $ D.mapMaybe f $ toStreamD m
965-- | Like 'mapMaybe' but maps a monadic function.
969-- >>> mapMaybeM f = Stream.catMaybes . Stream.mapM f
971-- >>> mapM f = Stream.mapMaybeM (\x -> Just <$> f x)
974{-# INLINE_EARLY mapMaybeM #-}
976 => (a -> m (Maybe b)) -> Stream m a -> Stream m b
977mapMaybeM f = fmap fromJust . filter isJust . mapM f
979------------------------------------------------------------------------------
981------------------------------------------------------------------------------
983-- | Discard 'Right's and unwrap 'Left's in an 'Either' stream.
985-- >>> lefts = fmap (fromLeft undefined) . Stream.filter isLeft
990lefts :: Monad m => Stream m (Either a b) -> Stream m a
991lefts = fmap (fromLeft undefined) . filter isLeft
993-- | Discard 'Left's and unwrap 'Right's in an 'Either' stream.
995-- >>> rights = fmap (fromRight undefined) . Stream.filter isRight
1000rights :: Monad m => Stream m (Either a b) -> Stream m b
1001rights = fmap (fromRight undefined) . filter isRight
1003-- | Remove the either wrapper and flatten both lefts and as well as rights in
1004-- the output stream.
1006-- >>> both = fmap (either id id)
1011both :: Monad m => Stream m (Either a a) -> Stream m a
1012both = 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