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.
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.
69 -- * Position Indexing
80 , findIndices -- XXX indicesBy
81 , elemIndices -- XXX indicesOf
84 -- | Map using the previous element.
91 -- * Inserting Elements
92 -- | Produce a superset of the stream. This is the opposite of
93 -- filtering/sampling. We can always use concatMap and scan for inserting
94 -- but these combinators are more efficient and convenient.
96 -- Element agnostic (Opposite of sampling)
98 , intersperseM -- XXX naming
102 , intersperseSuffixBySpan
104 -- , interspersePrefix
105 -- , interspersePrefixBySpan
107 -- * Inserting Side Effects/Time
108 , intersperseM_ -- XXX naming
115 -- * Element Aware Insertion
116 -- | Opposite of filtering
118 -- , intersperseByBefore
119 -- , intersperseByAfter
121 -- Fold and Unfold, Buffering
129 -- Move these to Streamly.Data.Either.Stream?
138import Control.Concurrent (threadDelay)
139import Control.Monad (void)
140import Control.Monad.IO.Class (MonadIO (liftIO))
141import Control.Monad.Trans.Class (MonadTrans(..))
142import Data.Either (fromLeft, isLeft, isRight, fromRight)
143import Data.Maybe (isJust, fromJust)
145import Streamly.Internal.Data.Fold.Type (Fold)
146import Streamly.Internal.Data.Pipe (Pipe)
147import Streamly.Internal.Data.Time.Units (AbsTime, RelTime64)
149import qualified Streamly.Internal.Data.Fold as FL
150-- import qualified Streamly.Internal.Data.Fold.Window as Window
151import qualified Streamly.Internal.Data.Stream.StreamD.Transform as D
152import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
153import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
156import Streamly.Internal.Data.Stream.Bottom
157import Streamly.Internal.Data.Stream.Type
160 ( filter, drop, dropWhile, take, takeWhile, foldr, map, mapM, sequence
161 , reverse, foldr1 , repeat, scanl, scanl1, zipWith)
166-- >>> import Control.Concurrent (threadDelay)
167-- >>> import Control.Monad.IO.Class (MonadIO (liftIO))
168-- >>> import Control.Monad.Trans (lift)
169-- >>> import Control.Monad.Trans.Identity (runIdentityT)
170-- >>> import Data.Either (fromLeft, fromRight, isLeft, isRight, either)
171-- >>> import Data.Function ((&))
172-- >>> import Data.Maybe (fromJust, isJust)
173-- >>> import Prelude hiding (filter, drop, dropWhile, take, takeWhile, foldr, map, mapM, sequence, reverse, foldr1 , scanl, scanl1)
174-- >>> import Streamly.Internal.Data.Stream as Stream
175-- >>> import qualified Streamly.Data.Fold as Fold
176-- >>> import qualified Streamly.Internal.Data.Fold as Fold
177-- >>> import qualified Streamly.Internal.Data.Fold.Window as Window
178-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
179-- >>> import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
181-- >>> hSetBuffering stdout LineBuffering
183-- XXX because of the use of D.cons for appending, folds and scans have
184-- quadratic complexity when iterated over a stream. We should use StreamK for
185-- linear performance on iteration.
187------------------------------------------------------------------------------
189------------------------------------------------------------------------------
191-- | Use a 'Pipe' to transform a stream.
195{-# INLINE transform #-}
196transform :: Monad m => Pipe m a b -> Stream m a -> Stream m b
197transform pipe xs = fromStreamD $ D.transform pipe (toStreamD xs)
199------------------------------------------------------------------------------
200-- Transformation Folds
201------------------------------------------------------------------------------
203-- | Right fold to a streaming monad.
205-- > foldrS Stream.cons Stream.nil === id
207-- 'foldrS' can be used to perform stateless stream to stream transformations
208-- like map and filter in general. It can be coupled with a scan to perform
209-- stateful transformations. However, note that the custom map and filter
210-- routines can be much more efficient than this due to better stream fusion.
212-- >>> input = Stream.fromList [1..5]
213-- >>> Stream.fold Fold.toList $ Stream.foldrS Stream.cons Stream.nil input
216-- Find if any element in the stream is 'True':
218-- >>> step x xs = if odd x then return True else xs
219-- >>> input = Stream.fromList (2:4:5:undefined) :: Stream IO Int
220-- >>> Stream.fold Fold.toList $ Stream.foldrS step (return False) input
223-- Map (+2) on odd elements and filter out the even elements:
225-- >>> step x xs = if odd x then (x + 2) `Stream.cons` xs else xs
226-- >>> input = Stream.fromList [1..5] :: Stream IO Int
227-- >>> Stream.fold Fold.toList $ Stream.foldrS step Stream.nil input
233 (a -> Stream m b -> Stream m b)
240 (\y ys -> toStreamK $ f y (fromStreamK ys))
244-- | Right fold to a transformer monad. This is the most general right fold
245-- function. 'foldrS' is a special case of 'foldrT', however 'foldrS'
246-- implementation can be more efficient:
248-- >>> foldrS = foldrT
250-- >>> step f x xs = lift $ f x (runIdentityT xs)
251-- >>> foldrM f z s = runIdentityT $ foldrT (step f) (lift z) s
253-- 'foldrT' can be used to translate streamly streams to other transformer
254-- monads e.g. to a different streaming type.
258foldrT :: (Monad m, Monad (s m), MonadTrans s)
259 => (a -> s m b -> s m b) -> s m b -> Stream m a -> s m b
260foldrT f z s = D.foldrT f z (toStreamD s)
262------------------------------------------------------------------------------
263-- Transformation by Mapping
264------------------------------------------------------------------------------
267-- >>> mapM f = Stream.sequence . fmap f
269-- Apply a monadic function to each element of the stream and replace it with
270-- the output of the resulting action.
272-- >>> s = Stream.fromList ["a", "b", "c"]
273-- >>> Stream.fold Fold.drain $ Stream.mapM putStr s
278mapM :: Monad m => (a -> m b) -> Stream m a -> Stream m b
279mapM f m = fromStreamK $ D.toStreamK $ D.mapM f $ toStreamD m
282-- >>> sequence = Stream.mapM id
284-- Replace the elements of a stream of monadic actions with the outputs of
287-- >>> s = Stream.fromList [putStr "a", putStr "b", putStrLn "c"]
288-- >>> Stream.fold Fold.drain $ Stream.sequence s
292{-# INLINE sequence #-}
293sequence :: Monad m => Stream m (m a) -> Stream m a
296------------------------------------------------------------------------------
297-- Mapping side effects
298------------------------------------------------------------------------------
300-- | Tap the data flowing through a stream into a 'Fold'. For example, you may
301-- add a tap to log the contents flowing through the stream. The fold is used
302-- only for effects, its result is discarded.
307-- -----stream m a ---------------stream m a-----
311-- >>> s = Stream.enumerateFromTo 1 2
312-- >>> Stream.fold Fold.drain $ Stream.tap (Fold.drainBy print) s
316-- Compare with 'trace'.
320tap :: Monad m => FL.Fold m a b -> Stream m a -> Stream m a
321tap f xs = fromStreamD $ D.tap f (toStreamD xs)
323-- | Apply a monadic function to each element flowing through the stream and
324-- discard the results.
326-- >>> s = Stream.enumerateFromTo 1 2
327-- >>> Stream.fold Fold.drain $ Stream.trace print s
331-- Compare with 'tap'.
334trace :: Monad m => (a -> m b) -> Stream m a -> Stream m a
335trace f = mapM (\x -> void (f x) >> return x)
337-- | Perform a side effect before yielding each element of the stream and
338-- discard the results.
340-- >>> s = Stream.enumerateFromTo 1 2
341-- >>> Stream.fold Fold.drain $ Stream.trace_ (print "got here") s
345-- Same as 'interspersePrefix_' but always serial.
351trace_ :: Monad m => m b -> Stream m a -> Stream m a
352trace_ eff = fromStreamD . D.mapM (\x -> eff >> return x) . toStreamD
354------------------------------------------------------------------------------
355-- Scanning with a Fold
356------------------------------------------------------------------------------
358-- XXX It may be useful to have a version of scan where we can keep the
359-- accumulator independent of the value emitted. So that we do not necessarily
360-- have to keep a value in the accumulator which we are not using. We can pass
361-- an extraction function that will take the accumulator and the current value
362-- of the element and emit the next value in the stream. That will also make it
363-- possible to modify the accumulator after using it. In fact, the step function
364-- can return new accumulator and the value to be emitted. The signature would
365-- be more like mapAccumL.
367-- | Strict left scan. Scan a stream using the given monadic fold.
369-- >>> s = Stream.fromList [1..10]
370-- >>> Stream.fold Fold.toList $ Stream.takeWhile (< 10) $ Stream.scan Fold.sum s
373-- See also: 'usingStateT'
378-- >>> scanl' step z = Stream.scan (Fold.foldl' step z)
380-- Like 'map', 'scanl'' too is a one to one transformation,
381-- however it adds an extra element.
383-- >>> s = Stream.fromList [1,2,3,4]
384-- >>> Stream.fold Fold.toList $ scanl' (+) 0 s
387-- >>> Stream.fold Fold.toList $ scanl' (flip (:)) [] s
388-- [[],[1],[2,1],[3,2,1],[4,3,2,1]]
390-- The output of 'scanl'' is the initial value of the accumulator followed by
391-- all the intermediate steps and the final result of 'foldl''.
393-- By streaming the accumulated state after each fold step, we can share the
394-- state across multiple stages of stream composition. Each stage can modify or
395-- extend the state, do some processing with it and emit it for the next stage,
396-- thus modularizing the stream processing. This can be useful in
397-- stateful or event-driven programming.
399-- Consider the following monolithic example, computing the sum and the product
400-- of the elements in a stream in one go using a @foldl'@:
402-- >>> foldl' step z = Stream.fold (Fold.foldl' step z)
403-- >>> foldl' (\(s, p) x -> (s + x, p * x)) (0,1) s
406-- Using @scanl'@ we can make it modular by computing the sum in the first
407-- stage and passing it down to the next stage for computing the product:
410-- foldl' (\(_, p) (s, x) -> (s, p * x)) (0,1)
411-- $ scanl' (\(s, _) x -> (s + x, x)) (0,1)
412-- $ Stream.fromList [1,2,3,4]
416-- IMPORTANT: 'scanl'' evaluates the accumulator to WHNF. To avoid building
417-- lazy expressions inside the accumulator, it is recommended that a strict
418-- data structure is used for accumulator.
422scan :: Monad m => Fold m a b -> Stream m a -> Stream m b
423scan fld m = fromStreamD $ D.scanOnce fld $ toStreamD m
425-- | Like 'scan' but restarts scanning afresh when the scanning fold
429{-# INLINE scanMany #-}
430scanMany :: Monad m => Fold m a b -> Stream m a -> Stream m b
431scanMany fld m = fromStreamD $ D.scanMany fld $ toStreamD m
433------------------------------------------------------------------------------
435------------------------------------------------------------------------------
437-- | Modify a @Stream m a -> Stream m a@ stream transformation that accepts a
438-- predicate @(a -> b)@ to accept @((s, a) -> b)@ instead, provided a
439-- transformation @Stream m a -> Stream m (s, a)@. Convenient to filter with
442-- >>> filterWithIndex = Stream.with Stream.indexed Stream.filter
447 (Stream m a -> Stream m (s, a))
448 -> (((s, a) -> b) -> Stream m (s, a) -> Stream m (s, a))
449 -> (((s, a) -> b) -> Stream m a -> Stream m a)
450with f comb g = fmap snd . comb g . f
452-- | Include only those elements that pass a predicate.
454-- >>> filter p = Stream.filterM (return . p)
455-- >>> filter p = Stream.mapMaybe (\x -> if p x then Just x else Nothing)
456-- >>> filter p = Stream.scanMaybe (Fold.filtering p)
460filter :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
461-- filter p = scanMaybe (FL.filtering p)
462filter p m = fromStreamD $ D.filter p $ toStreamD m
464-- | Same as 'filter' but with a monadic predicate.
466-- >>> f p x = p x >>= \r -> return $ if r then Just x else Nothing
467-- >>> filterM p = Stream.mapMaybeM (f p)
470{-# INLINE filterM #-}
471filterM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
472filterM p m = fromStreamD $ D.filterM p $ toStreamD m
474-- | Drop repeated elements that are adjacent to each other using the supplied
475-- comparison function.
477-- >>> uniq = Stream.uniqBy (==)
479-- To strip duplicate path separators:
481-- >>> input = Stream.fromList "//a//b"
482-- >>> f x y = x == '/' && y == '/'
483-- >>> Stream.fold Fold.toList $ Stream.uniqBy f input
494 (a -> a -> Bool) -> Stream m a -> Stream m a
495-- uniqBy eq = scanMaybe (FL.uniqBy eq)
496uniqBy eq = catMaybes . rollingMap f
503 Just x -> if x `eq` curr then Nothing else Just curr
505-- | Drop repeated elements that are adjacent to each other.
507-- >>> uniq = Stream.uniqBy (==)
511uniq :: (Eq a, Monad m) => Stream m a -> Stream m a
512-- uniq = scanMaybe FL.uniq
513uniq = fromStreamD . D.uniq . toStreamD
515-- | Strip all leading and trailing occurrences of an element passing a
516-- predicate and make all other consecutive occurrences uniq.
518-- >> prune p = Stream.dropWhileAround p $ Stream.uniqBy (x y -> p x && p y)
521-- > Stream.prune isSpace (Stream.fromList " hello world! ")
531 -- (Monad m, Eq a) =>
532 (a -> Bool) -> Stream m a -> Stream m a
533prune = error "Not implemented yet!"
535-- Possible implementation:
537-- Stream.catMaybes . Stream.parseMany (Parser.groupBy (==) Fold.repeated)@
539-- 'Fold.repeated' should return 'Just' when repeated, and 'Nothing' for a
542-- | Emit only repeated elements, once.
545repeated :: -- (Monad m, Eq a) =>
546 Stream m a -> Stream m a
549-- | Deletes the first occurrence of the element in the stream that satisfies
550-- the given equality predicate.
552-- >>> input = Stream.fromList [1,3,3,5]
553-- >>> Stream.fold Fold.toList $ Stream.deleteBy (==) 3 input
557{-# INLINE deleteBy #-}
558deleteBy :: Monad m => (a -> a -> Bool) -> a -> Stream m a -> Stream m a
559-- deleteBy cmp x = scanMaybe (FL.deleteBy cmp x)
560deleteBy cmp x m = fromStreamD $ D.deleteBy cmp x (toStreamD m)
562------------------------------------------------------------------------------
564------------------------------------------------------------------------------
566-- | Same as 'takeWhile' but with a monadic predicate.
569{-# INLINE takeWhileM #-}
570takeWhileM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
571-- takeWhileM p = scanMaybe (FL.takingEndByM_ (\x -> not <$> p x))
572takeWhileM p m = fromStreamD $ D.takeWhileM p $ toStreamD m
574-- | Drop elements in the stream as long as the predicate succeeds and then
575-- take the rest of the stream.
578{-# INLINE dropWhile #-}
579dropWhile :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
580-- dropWhile p = scanMaybe (FL.droppingWhile p)
581dropWhile p m = fromStreamD $ D.dropWhile p $ toStreamD m
583-- | Same as 'dropWhile' but with a monadic predicate.
586{-# INLINE dropWhileM #-}
587dropWhileM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
588-- dropWhileM p = scanMaybe (FL.droppingWhileM p)
589dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m
591------------------------------------------------------------------------------
593------------------------------------------------------------------------------
595-- | @insertBy cmp elem stream@ inserts @elem@ before the first element in
596-- @stream@ that is less than @elem@ when compared using @cmp@.
598-- >>> insertBy cmp x = Stream.mergeBy cmp (Stream.fromPure x)
600-- >>> input = Stream.fromList [1,3,5]
601-- >>> Stream.fold Fold.toList $ Stream.insertBy compare 2 input
605{-# INLINE insertBy #-}
606insertBy ::Monad m => (a -> a -> Ordering) -> a -> Stream m a -> Stream m a
607insertBy cmp x m = fromStreamD $ D.insertBy cmp x (toStreamD m)
609-- | Insert a pure value between successive elements of a stream.
611-- >>> input = Stream.fromList "hello"
612-- >>> Stream.fold Fold.toList $ Stream.intersperse ',' input
616{-# INLINE intersperse #-}
617intersperse :: Monad m => a -> Stream m a -> Stream m a
618intersperse a = fromStreamD . D.intersperse a . toStreamD
620-- | Insert a side effect before consuming an element of a stream except the
623-- >>> input = Stream.fromList "hello"
624-- >>> Stream.fold Fold.drain $ Stream.trace putChar $ Stream.intersperseM_ (putChar '.') input
628{-# INLINE intersperseM_ #-}
629intersperseM_ :: Monad m => m b -> Stream m a -> Stream m a
630intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
632-- | Intersperse a monadic action into the input stream after every @n@
635-- >> input = Stream.fromList "hello"
636-- >> Stream.fold Fold.toList $ Stream.intersperseBySpan 2 (return ',') input
640{-# INLINE intersperseBySpan #-}
641intersperseBySpan :: -- Monad m =>
642 Int -> m a -> Stream m a -> Stream m a
643intersperseBySpan _n _f _xs = undefined
645-- | Insert an effect and its output after consuming an element of a stream.
647-- >>> input = Stream.fromList "hello"
648-- >>> Stream.fold Fold.toList $ Stream.trace putChar $ intersperseSuffix (putChar '.' >> return ',') input
649-- h.,e.,l.,l.,o.,"h,e,l,l,o,"
652{-# INLINE intersperseSuffix #-}
653intersperseSuffix :: Monad m => m a -> Stream m a -> Stream m a
654intersperseSuffix m = fromStreamD . D.intersperseSuffix m . toStreamD
656-- | Insert a side effect after consuming an element of a stream.
658-- >>> input = Stream.fromList "hello"
659-- >>> Stream.fold Fold.toList $ Stream.intersperseSuffix_ (threadDelay 1000000) input
664{-# INLINE intersperseSuffix_ #-}
665intersperseSuffix_ :: Monad m => m b -> Stream m a -> Stream m a
666intersperseSuffix_ m = fromStreamD . D.intersperseSuffix_ m . toStreamD
668-- XXX Use an offset argument, like tapOffsetEvery
670-- | Like 'intersperseSuffix' but intersperses an effectful action into the
671-- input stream after every @n@ elements and after the last element.
673-- >>> input = Stream.fromList "hello"
674-- >>> Stream.fold Fold.toList $ Stream.intersperseSuffixBySpan 2 (return ',') input
679{-# INLINE intersperseSuffixBySpan #-}
680intersperseSuffixBySpan :: Monad m
681 => Int -> m a -> Stream m a -> Stream m a
682intersperseSuffixBySpan n eff =
683 fromStreamD . D.intersperseSuffixBySpan n eff . toStreamD
685-- | Insert a side effect before consuming an element of a stream.
687-- >>> input = Stream.fromList "hello"
688-- >>> Stream.fold Fold.toList $ Stream.trace putChar $ Stream.interspersePrefix_ (putChar '.' >> return ',') input
695{-# INLINE interspersePrefix_ #-}
696interspersePrefix_ :: Monad m => m b -> Stream m a -> Stream m a
697interspersePrefix_ m = mapM (\x -> void m >> return x)
699------------------------------------------------------------------------------
701------------------------------------------------------------------------------
703-- | Introduce a delay of specified seconds between elements of the stream.
705-- >>> input = Stream.enumerateFromTo 1 3
706-- >>> Stream.fold (Fold.drainBy print) $ Stream.delay 1 input
711-- >>> sleep n = liftIO $ threadDelay $ round $ n * 1000000
712-- >>> delay n = Stream.intersperseM_ $ sleep n
716delay :: MonadIO m => Double -> Stream m a -> Stream m a
717delay n = intersperseM_ $ liftIO $ threadDelay $ round $ n * 1000000
719-- | Introduce a delay of specified seconds after consuming an element of a
722-- >>> input = Stream.enumerateFromTo 1 3
723-- >>> Stream.fold (Fold.drainBy print) $ Stream.delayPost 1 input
730{-# INLINE delayPost #-}
731delayPost :: MonadIO m => Double -> Stream m a -> Stream m a
732delayPost n = intersperseSuffix_ $ liftIO $ threadDelay $ round $ n * 1000000
734-- | Introduce a delay of specified seconds before consuming an element of a
737-- >>> input = Stream.enumerateFromTo 1 3
738-- >>> Stream.fold (Fold.drainBy print) $ Stream.delayPre 1 input
745{-# INLINE delayPre #-}
746delayPre :: MonadIO m => Double -> Stream m a -> Stream m a
747delayPre n = trace_ $ liftIO $ threadDelay $ round $ n * 1000000
749------------------------------------------------------------------------------
750-- Reorder in sequence
751------------------------------------------------------------------------------
753-- | Buffer until the next element in sequence arrives. The function argument
754-- determines the difference in sequence numbers. This could be useful in
755-- implementing sequenced streams, for example, TCP reassembly.
759{-# INLINE reassembleBy #-}
766reassembleBy = undefined
768------------------------------------------------------------------------------
770------------------------------------------------------------------------------
773-- >>> f = Fold.foldl' (\(i, _) x -> (i + 1, x)) (-1,undefined)
774-- >>> indexed = Stream.postscan f
775-- >>> indexed = Stream.zipWith (,) (Stream.enumerateFrom 0)
777-- Pair each element in a stream with its index, starting from index 0.
779-- >>> Stream.fold Fold.toList $ Stream.indexed $ Stream.fromList "hello"
780-- [(0,'h'),(1,'e'),(2,'l'),(3,'l'),(4,'o')]
783{-# INLINE indexed #-}
784indexed :: Monad m => Stream m a -> Stream m (Int, a)
785-- indexed = scanMaybe FL.indexing
786indexed = fromStreamD . D.indexed . toStreamD
789-- >>> f n = Fold.foldl' (\(i, _) x -> (i - 1, x)) (n + 1,undefined)
790-- >>> indexedR n = Stream.postscan (f n)
792-- >>> s n = Stream.enumerateFromThen n (n - 1)
793-- >>> indexedR n = Stream.zipWith (,) (s n)
795-- Pair each element in a stream with its index, starting from the
796-- given index @n@ and counting down.
798-- >>> Stream.fold Fold.toList $ Stream.indexedR 10 $ Stream.fromList "hello"
799-- [(10,'h'),(9,'e'),(8,'l'),(7,'l'),(6,'o')]
802{-# INLINE indexedR #-}
803indexedR :: Monad m => Int -> Stream m a -> Stream m (Int, a)
804-- indexedR n = scanMaybe (FL.indexingRev n)
805indexedR n = fromStreamD . D.indexedR n . toStreamD
807-------------------------------------------------------------------------------
809-------------------------------------------------------------------------------
811-- Note: The timestamp stream must be the second stream in the zip so that the
812-- timestamp is generated after generating the stream element and not before.
813-- If we do not do that then the following example will generate the same
814-- timestamp for first two elements:
816-- Stream.fold Fold.toList $ Stream.timestamped $ Stream.delay $ Stream.enumerateFromTo 1 3
818-- | Pair each element in a stream with an absolute timestamp, using a clock of
819-- specified granularity. The timestamp is generated just before the element
822-- >>> Stream.fold Fold.toList $ Stream.timestampWith 0.01 $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
823-- [(AbsTime (TimeSpec {sec = ..., nsec = ...}),1),(AbsTime (TimeSpec {sec = ..., nsec = ...}),2),(AbsTime (TimeSpec {sec = ..., nsec = ...}),3)]
827{-# INLINE timestampWith #-}
828timestampWith :: (MonadIO m)
829 => Double -> Stream m a -> Stream m(AbsTime, a)
830timestampWith g stream = zipWith (flip (,)) stream (absTimesWith g)
832-- TBD: check performance vs a custom implementation without using zipWith.
836{-# INLINE timestamped #-}
837timestamped :: (MonadIO m)
838 => Stream m a -> Stream m(AbsTime, a)
839timestamped = timestampWith 0.01
841-- | Pair each element in a stream with relative times starting from 0, using a
842-- clock with the specified granularity. The time is measured just before the
843-- element is consumed.
845-- >>> Stream.fold Fold.toList $ Stream.timeIndexWith 0.01 $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
846-- [(RelTime64 (NanoSecond64 ...),1),(RelTime64 (NanoSecond64 ...),2),(RelTime64 (NanoSecond64 ...),3)]
850{-# INLINE timeIndexWith #-}
851timeIndexWith :: (MonadIO m)
852 => Double -> Stream m a -> Stream m(RelTime64, a)
853timeIndexWith g stream = zipWith (flip (,)) stream (relTimesWith g)
855-- | Pair each element in a stream with relative times starting from 0, using a
856-- 10 ms granularity clock. The time is measured just before the element is
859-- >>> Stream.fold Fold.toList $ Stream.timeIndexed $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
860-- [(RelTime64 (NanoSecond64 ...),1),(RelTime64 (NanoSecond64 ...),2),(RelTime64 (NanoSecond64 ...),3)]
864{-# INLINE timeIndexed #-}
865timeIndexed :: (MonadIO m)
866 => Stream m a -> Stream m(RelTime64, a)
867timeIndexed = timeIndexWith 0.01
869------------------------------------------------------------------------------
871------------------------------------------------------------------------------
873-- | Find all the indices where the value of the element in the stream is equal
874-- to the given value.
876-- >>> elemIndices a = Stream.findIndices (== a)
879{-# INLINE elemIndices #-}
880elemIndices :: (Monad m, Eq a) => a -> Stream m a -> Stream m Int
881elemIndices a = findIndices (== a)
883------------------------------------------------------------------------------
885------------------------------------------------------------------------------
887-- XXX this is not a one-to-one map so calling it map may not be right.
888-- We can perhaps call it zipWithTail or rollWith.
890-- | Apply a function on every two successive elements of a stream. The first
891-- argument of the map function is the previous element and the second argument
892-- is the current element. When the current element is the first element, the
893-- previous element is 'Nothing'.
897{-# INLINE rollingMap #-}
898rollingMap :: Monad m => (Maybe a -> a -> b) -> Stream m a -> Stream m b
899-- rollingMap f = scanMaybe (FL.slide2 $ Window.rollingMap f)
900rollingMap f m = fromStreamD $ D.rollingMap f $ toStreamD m
902-- | Like 'rollingMap' but with an effectful map function.
906{-# INLINE rollingMapM #-}
907rollingMapM :: Monad m => (Maybe a -> a -> m b) -> Stream m a -> Stream m b
908-- rollingMapM f = scanMaybe (FL.slide2 $ Window.rollingMapM f)
909rollingMapM f m = fromStreamD $ D.rollingMapM f $ toStreamD m
911-- | Like 'rollingMap' but requires at least two elements in the stream,
912-- returns an empty stream otherwise.
914-- This is the stream equivalent of the list idiom @zipWith f xs (tail xs)@.
918{-# INLINE rollingMap2 #-}
919rollingMap2 :: Monad m => (a -> a -> b) -> Stream m a -> Stream m b
920rollingMap2 f m = fromStreamD $ D.rollingMap2 f $ toStreamD m
922------------------------------------------------------------------------------
924------------------------------------------------------------------------------
926-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'
927-- elements, and return a stream of values extracted from 'Just'.
931-- >>> mapMaybe f = Stream.catMaybes . fmap f
934{-# INLINE mapMaybe #-}
935mapMaybe :: Monad m => (a -> Maybe b) -> Stream m a -> Stream m b
936mapMaybe f m = fromStreamD $ D.mapMaybe f $ toStreamD m
938-- | Like 'mapMaybe' but maps a monadic function.
942-- >>> mapMaybeM f = Stream.catMaybes . Stream.mapM f
944-- >>> mapM f = Stream.mapMaybeM (\x -> Just <$> f x)
947{-# INLINE_EARLY mapMaybeM #-}
949 => (a -> m (Maybe b)) -> Stream m a -> Stream m b
950mapMaybeM f = fmap fromJust . filter isJust . mapM f
952------------------------------------------------------------------------------
954------------------------------------------------------------------------------
956-- | Discard 'Right's and unwrap 'Left's in an 'Either' stream.
958-- >>> lefts = fmap (fromLeft undefined) . Stream.filter isLeft
963lefts :: Monad m => Stream m (Either a b) -> Stream m a
964lefts = fmap (fromLeft undefined) . filter isLeft
966-- | Discard 'Left's and unwrap 'Right's in an 'Either' stream.
968-- >>> rights = fmap (fromRight undefined) . Stream.filter isRight
973rights :: Monad m => Stream m (Either a b) -> Stream m b
974rights = fmap (fromRight undefined) . filter isRight
976-- | Remove the either wrapper and flatten both lefts and as well as rights in
979-- >>> both = fmap (either id id)
984both :: Monad m => Stream m (Either a a) -> Stream m a
985both = 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