2-- Module : Streamly.Internal.Data.Stream.IsStream.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.IsStream.Transform
12 -- | Pass through a 'Pipe'.
21 -- | Stateless one-to-one maps.
27 -- * Mapping Side Effects (Observation)
28 -- | See also the intersperse*_ combinators.
38 -- * Scanning By 'Fold'
44 -- | Left scans. Stateful, mostly one-to-one maps.
56 -- XXX Once we have pipes the contravariant transformations can be
57 -- represented by attaching pipes before a transformation.
70 -- | Produce a subset of the stream using criteria based on the values of
71 -- the elements. We can use a concatMap and scan for filtering but these
72 -- combinators are more efficient and convenient.
85 -- | Produce a subset of the stream trimmed at ends.
102 -- * Inserting Elements
103 -- | Produce a superset of the stream. This is the opposite of
104 -- filtering/sampling. We can always use concatMap and scan for inserting
105 -- but these combinators are more efficient and convenient.
107 -- Element agnostic (Opposite of sampling)
109 , intersperseM -- XXX naming
113 , intersperseSuffixBySpan
116 -- , interspersePrefix
117 -- , interspersePrefixBySpan
119 -- * Inserting Side Effects/Time
120 , intersperseM_ -- XXX naming
127 -- * Element Aware Insertion
128 -- | Opposite of filtering
130 -- , intersperseByBefore
131 -- , intersperseByAfter
138 -- * Position Indexing
145 -- , timestampedR -- timer
150 , findIndices -- XXX indicesBy
151 , elemIndices -- XXX indicesOf
154 -- | Map using the previous element.
160 -- Move these to Streamly.Data.Maybe.Stream?
161 , catMaybes -- XXX justs (like lefts/rights)
166 -- Move these to Streamly.Data.Either.Stream?
171 -- * Concurrent Evaluation
172 -- ** Concurrent Pipelines
173 -- | Run streaming stages concurrently.
181 -- ** Concurrency Control
184 -- ** Buffering and Sampling
185 -- | Evaluate strictly using a buffer of results. When the buffer becomes
186 -- full we can block, drop the new elements, drop the oldest element and
187 -- insert the new at the end or keep dropping elements uniformly to match
188 -- the rate of the consumer.
195 -- | Evaluate the stream at uniform intervals to maintain a specified
214import Control.Concurrent (threadDelay)
215import Control.Monad (void)
216import Control.Monad.IO.Class (MonadIO(..))
217import Control.Monad.Trans.Class (MonadTrans(..))
218import Data.Either (isLeft, isRight, fromLeft, fromRight)
219import Data.Kind (Type)
220import Data.Maybe (isJust, fromJust)
221import Streamly.Internal.Data.Fold.Type (Fold (..))
222import Streamly.Internal.Data.Pipe.Type (Pipe (..))
223import Streamly.Internal.Data.Stream.IsStream.Combinators
224 ( inspectMode, maxBuffer, maxThreads, rate, avgRate, minRate
225 , maxRate, constRate)
226import Streamly.Internal.Data.Stream.IsStream.Common
246import Streamly.Internal.Control.Concurrent (MonadAsync)
247import Streamly.Internal.Data.Stream.IsStream.Type
248 (IsStream(..), fromStreamD, toStreamD, toConsK)
249import Streamly.Internal.Data.Stream.Serial (SerialT)
250import Streamly.Internal.Data.SVar (Rate(..))
251import Streamly.Internal.Data.Time.Units (AbsTime, RelTime64)
253import qualified Streamly.Data.Fold as FL
254import qualified Streamly.Internal.Data.Stream.IsStream.Parallel as Par
255import qualified Streamly.Internal.Data.Stream.Serial as Serial
256import qualified Streamly.Internal.Data.Stream.StreamD as D
257 (transform, foldrT, tap, tapOffsetEvery, pollCounts, mapM, scanOnce
258 , scanMany, postscanOnce, scanlx', scanlM', scanl', postscanl', prescanl'
259 , prescanlM', scanl1M', scanl1', filter, filterM, uniq, deleteBy, takeWhileM
260 , dropWhile, dropWhileM, insertBy, intersperse
261 , intersperseM_, intersperseSuffix, intersperseSuffix_
262 , intersperseSuffixBySpan, indexed, indexedR, rollingMap, rollingMapM
263 , rollingMap2, mapMaybe, mapMaybeM)
264import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
265 (foldrS, foldrSShared, mapMWith)
266import qualified Prelude
269 ( filter, drop, dropWhile, take, takeWhile, foldr, map, mapM, sequence
270 , reverse, foldr1 , scanl, scanl1, zipWith)
275-- >>> import Control.Concurrent (threadDelay)
276-- >>> import Data.Function ((&))
277-- >>> import Streamly.Prelude ((|$))
278-- >>> import Prelude hiding ( filter, drop, dropWhile, take, takeWhile, foldr, map, mapM, sequence, reverse, foldr1 , scanl, scanl1)
279-- >>> import qualified Streamly.Prelude as Stream
280-- >>> import Streamly.Internal.Data.Stream.IsStream as Stream
281-- >>> import qualified Streamly.Data.Fold as Fold
282-- >>> import qualified Streamly.Internal.Data.Fold as Fold
283-- >>> import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
285-- >>> hSetBuffering stdout LineBuffering
287------------------------------------------------------------------------------
289------------------------------------------------------------------------------
291-- | Use a 'Pipe' to transform a stream.
295{-# INLINE transform #-}
296transform :: (IsStream t, Monad m) => Pipe m a b -> t m a -> t m b
297transform pipe xs = fromStreamD $ D.transform pipe (toStreamD xs)
299------------------------------------------------------------------------------
300-- Transformation Folds
301------------------------------------------------------------------------------
303-- | Right fold to a streaming monad.
305-- > foldrS Stream.cons Stream.nil === id
307-- 'foldrS' can be used to perform stateless stream to stream transformations
308-- like map and filter in general. It can be coupled with a scan to perform
309-- stateful transformations. However, note that the custom map and filter
310-- routines can be much more efficient than this due to better stream fusion.
312-- >>> Stream.toList $ Stream.foldrS Stream.cons Stream.nil $ Stream.fromList [1..5]
315-- Find if any element in the stream is 'True':
317-- >>> Stream.toList $ Stream.foldrS (\x xs -> if odd x then return True else xs) (return False) $ (Stream.fromList (2:4:5:undefined) :: Stream.SerialT IO Int)
320-- Map (+2) on odd elements and filter out the even elements:
322-- >>> Stream.toList $ Stream.foldrS (\x xs -> if odd x then (x + 2) `Stream.cons` xs else xs) Stream.nil $ (Stream.fromList [1..5] :: Stream.SerialT IO Int)
325-- 'foldrM' can also be represented in terms of 'foldrS', however, the former
326-- is much more efficient:
328-- > foldrM f z s = runIdentityT $ foldrS (\x xs -> lift $ f x (runIdentityT xs)) (lift z) s
332foldrS :: IsStream t => (a -> t m b -> t m b) -> t m b -> t m a -> t m b
336 (\y ys -> toStream $ f y (fromStream ys))
340{-# INLINE foldrSShared #-}
341foldrSShared :: IsStream t => (a -> t m b -> t m b) -> t m b -> t m a -> t m b
345 (\y ys -> toStream $ f y (fromStream ys))
349-- | Right fold to a transformer monad. This is the most general right fold
350-- function. 'foldrS' is a special case of 'foldrT', however 'foldrS'
351-- implementation can be more efficient:
354-- > foldrM f z s = runIdentityT $ foldrT (\x xs -> lift $ f x (runIdentityT xs)) (lift z) s
356-- 'foldrT' can be used to translate streamly streams to other transformer
357-- monads e.g. to a different streaming type.
361foldrT :: (IsStream t, Monad m, Monad (s m), MonadTrans s)
362 => (a -> s m b -> s m b) -> s m b -> t m a -> s m b
363foldrT f z s = D.foldrT f z (toStreamD s)
365------------------------------------------------------------------------------
366-- Transformation by Mapping
367------------------------------------------------------------------------------
371-- mapM f = sequence . map f
374-- Apply a monadic function to each element of the stream and replace it with
375-- the output of the resulting action.
378-- >>> drain $ Stream.mapM putStr $ Stream.fromList ["a", "b", "c"]
382-- drain $ Stream.replicateM 10 (return 1)
383-- & (fromSerial . Stream.mapM (\x -> threadDelay 1000000 >> print x))
389-- > drain $ Stream.replicateM 10 (return 1)
390-- & (fromAsync . Stream.mapM (\x -> threadDelay 1000000 >> print x))
393-- /Concurrent (do not use with 'fromParallel' on infinite streams)/
396{-# INLINE_EARLY mapM #-}
397mapM :: forall t m a b. (IsStream t, MonadAsync m) =>
398 (a -> m b) -> t m a -> t m b
399mapM f = fromStream . K.mapMWith (toConsK (consM @t)) f . toStream
401{-# RULES "mapM serial" mapM = mapMSerial #-}
402{-# INLINE mapMSerial #-}
403mapMSerial :: Monad m => (a -> m b) -> SerialT m a -> SerialT m b
404mapMSerial = Serial.mapM
411-- Replace the elements of a stream of monadic actions with the outputs of
415-- >>> drain $ Stream.sequence $ Stream.fromList [putStr "a", putStr "b", putStrLn "c"]
419-- drain $ Stream.replicateM 3 (return $ threadDelay 1000000 >> print 1)
420-- & (fromSerial . Stream.sequence)
427-- drain $ Stream.replicateM 3 (return $ threadDelay 1000000 >> print 1)
428-- & (fromAsync . Stream.sequence)
436-- /Concurrent (do not use with 'fromParallel' on infinite streams)/
439{-# INLINE sequence #-}
440sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
443------------------------------------------------------------------------------
444-- Mapping side effects
445------------------------------------------------------------------------------
447-- | Tap the data flowing through a stream into a 'Fold'. For example, you may
448-- add a tap to log the contents flowing through the stream. The fold is used
449-- only for effects, its result is discarded.
454-- -----stream m a ---------------stream m a-----
458-- >>> Stream.drain $ Stream.tap (Fold.drainBy print) (Stream.enumerateFromTo 1 2)
462-- Compare with 'trace'.
466tap :: (IsStream t, Monad m) => FL.Fold m a b -> t m a -> t m a
467tap f xs = fromStreamD $ D.tap f (toStreamD xs)
469-- XXX Remove this. It can be expressed in terms of Fold.sampleFromThen.
471-- | @tapOffsetEvery offset n@ taps every @n@th element in the stream
472-- starting at @offset@. @offset@ can be between @0@ and @n - 1@. Offset 0
473-- means start at the first element in the stream. If the offset is outside
474-- this range then @offset `mod` n@ is used as offset.
476-- >>> Stream.drain $ Stream.tapOffsetEvery 0 2 (Fold.rmapM print Fold.toList) $ Stream.enumerateFromTo 0 10
482{-# INLINE tapOffsetEvery #-}
483tapOffsetEvery :: (IsStream t, Monad m)
484 => Int -> Int -> FL.Fold m a b -> t m a -> t m a
485tapOffsetEvery offset n f xs =
486 fromStreamD $ D.tapOffsetEvery offset n f (toStreamD xs)
488-- | Redirect a copy of the stream to a supplied fold and run it concurrently
489-- in an independent thread. The fold may buffer some elements. The buffer size
490-- is determined by the prevailing 'maxBuffer' setting.
495-- -----stream m a ---------------stream m a-----
500-- >>> Stream.drain $ Stream.tapAsync (Fold.drainBy print) (Stream.enumerateFromTo 1 2)
506-- Exceptions from the concurrently running fold are propagated to the current
507-- computation. Note that, because of buffering in the fold, exceptions may be
508-- delayed and may not correspond to the current element being processed in the
509-- parent stream, but we guarantee that before the parent stream stops the tap
510-- finishes and all exceptions from it are drained.
512-- >>> tapAsync f = Stream.tapAsyncK (Stream.fold f . Stream.adapt)
514-- Compare with 'tap'.
517{-# INLINE tapAsync #-}
518tapAsync :: (IsStream t, MonadAsync m) => FL.Fold m a b -> t m a -> t m a
519tapAsync f xs = fromStreamD $ Par.tapAsyncF f (toStreamD xs)
521-- XXX We should keep only tapAsyncK as tapAsync can be implemented in terms
522-- of tapAsyncK and it has better performance. Rename this to tapAsync.
524-- | Like 'tapAsyncF' but uses a stream fold function instead of a 'Fold' type.
527{-# INLINE tapAsyncK #-}
528tapAsyncK :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> t m a
529tapAsyncK f m = fromStream $ Par.tapAsyncK (f . fromStream) (toStream m)
531-- XXX rename this to tapManyAsync_?
533-- | Concurrently distribute a stream to a collection of fold functions,
534-- discarding the outputs of the folds.
537-- > Stream.drain $ Stream.distributeAsync_ [Stream.mapM_ print, Stream.mapM_ print] (Stream.enumerateFromTo 1 2)
546-- distributeAsync_ = flip (foldr tapAsync)
551{-# INLINE distributeAsync_ #-}
552distributeAsync_ :: (Foldable f, IsStream t, MonadAsync m)
553 => f (t m a -> m b) -> t m a -> t m a
554distributeAsync_ = flip (Prelude.foldr tapAsyncK)
556-- | @pollCounts predicate transform fold stream@ counts those elements in the
557-- stream that pass the @predicate@. The resulting count stream is sent to
558-- another thread which transforms it using @transform@ and then folds it using
559-- @fold@. The thread is automatically cleaned up if the stream stops or
560-- aborts due to exception.
562-- For example, to print the count of elements processed every second:
565-- > Stream.drain $ Stream.pollCounts (const True) (Stream.rollingMap (-) . Stream.delayPost 1) (FLold.drainBy print)
566-- $ Stream.enumerateFrom 0
569-- Note: This may not work correctly on 32-bit machines.
573{-# INLINE pollCounts #-}
575 (IsStream t, MonadAsync m)
580pollCounts predicate f xs =
582 $ D.pollCounts predicate (f . fromStreamD)
585-- | Apply a monadic function to each element flowing through the stream and
586-- discard the results.
588-- >>> Stream.drain $ Stream.trace print (Stream.enumerateFromTo 1 2)
592-- Compare with 'tap'.
596trace :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m a
597trace f = mapM (\x -> void (f x) >> return x)
599-- | Perform a side effect before yielding each element of the stream and
600-- discard the results.
603-- >>> Stream.drain $ Stream.trace_ (print "got here") (Stream.enumerateFromTo 1 2)
609-- Same as 'interspersePrefix_' but always serial.
615trace_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
616trace_ eff = fromStreamD . D.mapM (\x -> eff >> return x) . toStreamD
618------------------------------------------------------------------------------
619-- Scanning with a Fold
620------------------------------------------------------------------------------
622-- | Scan a stream using the given monadic fold.
624-- >>> Stream.toList $ Stream.takeWhile (< 10) $ Stream.scan Fold.sum (Stream.fromList [1..10])
629scan :: (IsStream t, Monad m) => Fold m a b -> t m a -> t m b
630scan fld m = fromStreamD $ D.scanOnce fld $ toStreamD m
632-- | Like 'scan' but restarts scanning afresh when the scanning fold
636{-# INLINE scanMany #-}
637scanMany :: (IsStream t, Monad m) => Fold m a b -> t m a -> t m b
638scanMany fld m = fromStreamD $ D.scanMany fld $ toStreamD m
640-- | Postscan a stream using the given monadic fold.
642-- The following example extracts the input stream up to a point where the
643-- running average of elements is no more than 10:
645-- >>> import Data.Maybe (fromJust)
646-- >>> let avg = Fold.teeWith (/) Fold.sum (fmap fromIntegral Fold.length)
649-- $ Stream.map (fromJust . fst)
650-- $ Stream.takeWhile (\(_,x) -> x <= 10)
651-- $ Stream.postscan (Fold.tee Fold.last avg) (Stream.enumerateFromTo 1.0 100.0)
653-- [1.0,2.0,3.0,4.0,5.0,6.0,7.0,8.0,9.0,10.0,11.0,12.0,13.0,14.0,15.0,16.0,17.0,18.0,19.0]
656{-# INLINE postscan #-}
657postscan :: (IsStream t, Monad m) => Fold m a b -> t m a -> t m b
658postscan fld = fromStreamD . D.postscanOnce fld . toStreamD
660------------------------------------------------------------------------------
661-- Scanning - Transformation by Folding
662------------------------------------------------------------------------------
664-- XXX It may be useful to have a version of scan where we can keep the
665-- accumulator independent of the value emitted. So that we do not necessarily
666-- have to keep a value in the accumulator which we are not using. We can pass
667-- an extraction function that will take the accumulator and the current value
668-- of the element and emit the next value in the stream. That will also make it
669-- possible to modify the accumulator after using it. In fact, the step function
670-- can return new accumulator and the value to be emitted. The signature would
671-- be more like mapAccumL. Or we can change the signature of scanx to
674-- | Strict left scan with an extraction function. Like 'scanl'', but applies a
675-- user supplied extraction function (the third argument) at each step. This is
676-- designed to work with the @foldl@ library. The suffix @x@ is a mnemonic for
681-- /Since: 0.7.0 (Monad m constraint)/
682{-# DEPRECATED scanx "Please use scanl followed by map instead." #-}
684scanx :: (IsStream t, Monad m) => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
685scanx step begin done = fromStreamD . D.scanlx' step begin done . toStreamD
687-- XXX this needs to be concurrent
688-- XXX because of the use of D.cons for appending, scanlM' has quadratic
689-- complexity when iterated over a stream. We should use StreamK style scanlM'
690-- for linear performance on iteration.
692-- | Like 'scanl'' but with a monadic step function and a monadic seed.
696-- /Since: 0.8.0 (signature change)/
697{-# INLINE scanlM' #-}
698scanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
699scanlM' step begin m = fromStreamD $ D.scanlM' step begin $ toStreamD m
701-- XXX because of the use of D.cons for appending, scanl' has quadratic
702-- complexity when iterated over a stream. We should use StreamK style scanlM'
703-- for linear performance on iteration.
705-- | Strict left scan. Like 'map', 'scanl'' too is a one to one transformation,
706-- however it adds an extra element.
709-- >>> Stream.toList $ Stream.scanl' (+) 0 $ fromList [1,2,3,4]
715-- >>> Stream.toList $ Stream.scanl' (flip (:)) [] $ Stream.fromList [1,2,3,4]
716-- [[],[1],[2,1],[3,2,1],[4,3,2,1]]
720-- The output of 'scanl'' is the initial value of the accumulator followed by
721-- all the intermediate steps and the final result of 'foldl''.
723-- By streaming the accumulated state after each fold step, we can share the
724-- state across multiple stages of stream composition. Each stage can modify or
725-- extend the state, do some processing with it and emit it for the next stage,
726-- thus modularizing the stream processing. This can be useful in
727-- stateful or event-driven programming.
729-- Consider the following monolithic example, computing the sum and the product
730-- of the elements in a stream in one go using a @foldl'@:
733-- >>> Stream.foldl' (\(s, p) x -> (s + x, p * x)) (0,1) $ Stream.fromList [1,2,3,4]
738-- Using @scanl'@ we can make it modular by computing the sum in the first
739-- stage and passing it down to the next stage for computing the product:
743-- Stream.foldl' (\(_, p) (s, x) -> (s, p * x)) (0,1)
744-- $ Stream.scanl' (\(s, _) x -> (s + x, x)) (0,1)
745-- $ Stream.fromList [1,2,3,4]
751-- IMPORTANT: 'scanl'' evaluates the accumulator to WHNF. To avoid building
752-- lazy expressions inside the accumulator, it is recommended that a strict
753-- data structure is used for accumulator.
755-- >>> scanl' step z = scan (Fold.foldl' step z)
756-- >>> scanl' f z xs = scanlM' (\a b -> return (f a b)) (return z) xs
757-- >>> scanl' f z xs = z `Stream.cons` postscanl' f z xs
759-- See also: 'usingStateT'
763scanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
764scanl' step z m = fromStreamD $ D.scanl' step z $ toStreamD m
766-- | Like 'scanl'' but does not stream the initial value of the accumulator.
768-- >>> postscanl' step z = postscan (Fold.foldl' step z)
769-- >>> postscanl' f z = postscanlM' (\a b -> return (f a b)) (return z)
770-- >>> postscanl' f z xs = Stream.drop 1 $ Stream.scanl' f z xs
773{-# INLINE postscanl' #-}
774postscanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
775postscanl' step z m = fromStreamD $ D.postscanl' step z $ toStreamD m
777-- XXX prescanl does not sound very useful, enable only if there is a
778-- compelling use case.
780-- | Like scanl' but does not stream the final value of the accumulator.
783{-# INLINE prescanl' #-}
784prescanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
785prescanl' step z m = fromStreamD $ D.prescanl' step z $ toStreamD m
787-- XXX this needs to be concurrent
788-- | Like prescanl' but with a monadic step function and a monadic seed.
791{-# INLINE prescanlM' #-}
792prescanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
793prescanlM' step z m = fromStreamD $ D.prescanlM' step z $ toStreamD m
795-- XXX this needs to be concurrent
796-- | Like 'scanl1'' but with a monadic step function.
799{-# INLINE scanl1M' #-}
800scanl1M' :: (IsStream t, Monad m) => (a -> a -> m a) -> t m a -> t m a
801scanl1M' step m = fromStreamD $ D.scanl1M' step $ toStreamD m
803-- | Like 'scanl'' but for a non-empty stream. The first element of the stream
804-- is used as the initial value of the accumulator. Does nothing if the stream
808-- >>> Stream.toList $ Stream.scanl1' (+) $ fromList [1,2,3,4]
814{-# INLINE scanl1' #-}
815scanl1' :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> t m a
816scanl1' step m = fromStreamD $ D.scanl1' step $ toStreamD m
818------------------------------------------------------------------------------
820------------------------------------------------------------------------------
822-- | Modify a @t m a -> t m a@ stream transformation that accepts a predicate
823-- @(a -> b)@ to accept @((s, a) -> b)@ instead, provided a transformation @t m
824-- a -> t m (s, a)@. Convenient to filter with index or time.
827-- filterWithIndex = with indexed filter
828-- filterWithAbsTime = with timestamped filter
829-- filterWithRelTime = with timeIndexed filter
834with :: forall (t :: (Type -> Type) -> Type -> Type) m a b s. Functor (t m) =>
835 (t m a -> t m (s, a))
836 -> (((s, a) -> b) -> t m (s, a) -> t m (s, a))
837 -> (((s, a) -> b) -> t m a -> t m a)
838with f comb g = fmap snd . comb g . f
840-- | Include only those elements that pass a predicate.
844filter :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
845filter p m = fromStreamD $ D.filter p $ toStreamD m
847-- | Same as 'filter' but with a monadic predicate.
850{-# INLINE filterM #-}
851filterM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
852filterM p m = fromStreamD $ D.filterM p $ toStreamD m
854-- | Drop repeated elements that are adjacent to each other using the supplied
855-- comparison function.
857-- @uniq = uniqBy (==)
859-- To strip duplicate path separators:
862-- f x y = x == '/' && x == y
863-- Stream.toList $ Stream.uniqBy f $ Stream.fromList "//a//b"
874uniqBy :: (IsStream t, Monad m, Functor (t m)) =>
875 (a -> a -> Bool) -> t m a -> t m a
876uniqBy eq = catMaybes . rollingMap f
883 Just x -> if x `eq` curr then Nothing else Just curr
885-- | Drop repeated elements that are adjacent to each other.
889uniq :: (Eq a, IsStream t, Monad m) => t m a -> t m a
890uniq = fromStreamD . D.uniq . toStreamD
892-- | Strip all leading and trailing occurrences of an element passing a
893-- predicate and make all other consecutive occurrences uniq.
896-- prune p = dropWhileAround p $ uniqBy (x y -> p x && p y)
900-- > Stream.prune isSpace (Stream.fromList " hello world! ")
910 -- (IsStream t, Monad m, Eq a) =>
911 (a -> Bool) -> t m a -> t m a
912prune = error "Not implemented yet!"
914-- Possible implementation:
916-- Stream.catMaybes . Stream.parseMany (Parser.groupBy (==) Fold.repeated)@
918-- 'Fold.repeated' should return 'Just' when repeated, and 'Nothing' for a
921-- | Emit only repeated elements, once.
924repeated :: -- (IsStream t, Monad m, Eq a) =>
928-- We can have more efficient implementations for nubOrd and nubInt by using
929-- Set and IntSet to find/remove duplication. For Hashable we can use a
930-- hashmap. Use rewrite rules to specialize to more efficient impls.
932-- | Drop repeated elements anywhere in the stream.
934-- /Caution: not scalable for infinite streams/
936-- /See also: nubWindowBy/
941nubBy :: -- (IsStream t, Monad m) =>
942 (a -> a -> Bool) -> t m a -> t m a
943nubBy = undefined -- fromStreamD . D.nubBy . toStreamD
945-- | Deletes the first occurrence of the element in the stream that satisfies
946-- the given equality predicate.
949-- >>> Stream.toList $ Stream.deleteBy (==) 3 $ Stream.fromList [1,3,3,5]
955{-# INLINE deleteBy #-}
956deleteBy :: (IsStream t, Monad m) => (a -> a -> Bool) -> a -> t m a -> t m a
957deleteBy cmp x m = fromStreamD $ D.deleteBy cmp x (toStreamD m)
959------------------------------------------------------------------------------
961------------------------------------------------------------------------------
963-- XXX We could use 'maxBuffer Block/Drop/Rotate/Sample' instead. However we
964-- may want to have the evaluation rate independent of the sampling rate. To
965-- support that we can decouple evaluation and sampling in independent stages.
966-- The sampling stage would strictly evaluate and sample, the evaluation stage
967-- would control the evaluation rate.
969-- | Evaluate the input stream continuously and keep only the oldest @n@
970-- elements in the buffer, discard the new ones when the buffer is full. When
971-- the output stream is evaluated it consumes the values from the buffer in a
976{-# INLINE sampleOld #-}
977sampleOld :: -- (IsStream t, Monad m) =>
978 Int -> t m a -> t m a
981-- | Evaluate the input stream continuously and keep only the latest @n@
982-- elements in a ring buffer, keep discarding the older ones to make space for
983-- the new ones. When the output stream is evaluated it consumes the values
984-- from the buffer in a FIFO manner.
988{-# INLINE sampleNew #-}
989sampleNew :: -- (IsStream t, Monad m) =>
990 Int -> t m a -> t m a
993-- | Like 'sampleNew' but samples at uniform intervals to match the consumer
994-- rate. Note that 'sampleNew' leads to non-uniform sampling depending on the
999{-# INLINE sampleRate #-}
1000sampleRate :: -- (IsStream t, Monad m) =>
1001 Double -> t m a -> t m a
1002sampleRate = undefined
1004------------------------------------------------------------------------------
1006------------------------------------------------------------------------------
1008-- | Same as 'takeWhile' but with a monadic predicate.
1011{-# INLINE takeWhileM #-}
1012takeWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
1013takeWhileM p m = fromStreamD $ D.takeWhileM p $ toStreamD m
1015-- See the lastN fold for impl hints. Use a Data.Array based ring buffer.
1017-- takeLast n = Stream.concatM . fmap Array.toStream . fold Fold.lastN
1019-- | Take @n@ elements at the end of the stream.
1021-- O(n) space, where n is the number elements taken.
1024{-# INLINE takeLast #-}
1025takeLast :: -- (IsStream t, Monad m) =>
1026 Int -> t m a -> t m a
1027takeLast = undefined -- fromStreamD $ D.takeLast n $ toStreamD m
1029-- | Take time interval @i@ seconds at the end of the stream.
1031-- O(n) space, where n is the number elements taken.
1034{-# INLINE takeLastInterval #-}
1035takeLastInterval :: -- (IsStream t, Monad m) =>
1036 Double -> t m a -> t m a
1037takeLastInterval = undefined -- fromStreamD $ D.takeLast n $ toStreamD m
1039-- | Take all consecutive elements at the end of the stream for which the
1040-- predicate is true.
1042-- O(n) space, where n is the number elements taken.
1045{-# INLINE takeWhileLast #-}
1046takeWhileLast :: -- (IsStream t, Monad m) =>
1047 (a -> Bool) -> t m a -> t m a
1048takeWhileLast = undefined -- fromStreamD $ D.takeWhileLast n $ toStreamD m
1050-- | Like 'takeWhile' and 'takeWhileLast' combined.
1052-- O(n) space, where n is the number elements taken from the end.
1055{-# INLINE takeWhileAround #-}
1056takeWhileAround :: -- (IsStream t, Monad m) =>
1057 (a -> Bool) -> t m a -> t m a
1058takeWhileAround = undefined -- fromStreamD $ D.takeWhileAround n $ toStreamD m
1060-- | Drop elements in the stream as long as the predicate succeeds and then
1061-- take the rest of the stream.
1064{-# INLINE dropWhile #-}
1065dropWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
1066dropWhile p m = fromStreamD $ D.dropWhile p $ toStreamD m
1068-- | Same as 'dropWhile' but with a monadic predicate.
1071{-# INLINE dropWhileM #-}
1072dropWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
1073dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m
1075-- | Drop @n@ elements at the end of the stream.
1077-- O(n) space, where n is the number elements dropped.
1080{-# INLINE dropLast #-}
1081dropLast :: -- (IsStream t, Monad m) =>
1082 Int -> t m a -> t m a
1083dropLast = undefined -- fromStreamD $ D.dropLast n $ toStreamD m
1085-- | Drop time interval @i@ seconds at the end of the stream.
1087-- O(n) space, where n is the number elements dropped.
1090{-# INLINE dropLastInterval #-}
1091dropLastInterval :: -- (IsStream t, Monad m) =>
1092 Int -> t m a -> t m a
1093dropLastInterval = undefined
1095-- | Drop all consecutive elements at the end of the stream for which the
1096-- predicate is true.
1098-- O(n) space, where n is the number elements dropped.
1101{-# INLINE dropWhileLast #-}
1102dropWhileLast :: -- (IsStream t, Monad m) =>
1103 (a -> Bool) -> t m a -> t m a
1104dropWhileLast = undefined -- fromStreamD $ D.dropWhileLast n $ toStreamD m
1106-- | Like 'dropWhile' and 'dropWhileLast' combined.
1108-- O(n) space, where n is the number elements dropped from the end.
1111{-# INLINE dropWhileAround #-}
1112dropWhileAround :: -- (IsStream t, Monad m) =>
1113 (a -> Bool) -> t m a -> t m a
1114dropWhileAround = undefined -- fromStreamD $ D.dropWhileAround n $ toStreamD m
1116------------------------------------------------------------------------------
1117-- Inserting Elements
1118------------------------------------------------------------------------------
1120-- | @insertBy cmp elem stream@ inserts @elem@ before the first element in
1121-- @stream@ that is less than @elem@ when compared using @cmp@.
1124-- insertBy cmp x = 'mergeBy' cmp ('fromPure' x)
1128-- >>> Stream.toList $ Stream.insertBy compare 2 $ Stream.fromList [1,3,5]
1134{-# INLINE insertBy #-}
1136 (IsStream t, Monad m) => (a -> a -> Ordering) -> a -> t m a -> t m a
1137insertBy cmp x m = fromStreamD $ D.insertBy cmp x (toStreamD m)
1139-- | Insert a pure value between successive elements of a stream.
1141-- >>> Stream.toList $ Stream.intersperse ',' $ Stream.fromList "hello"
1145{-# INLINE intersperse #-}
1146intersperse :: (IsStream t, MonadAsync m) => a -> t m a -> t m a
1147intersperse a = fromStreamD . D.intersperse a . toStreamD
1149-- | Insert a side effect before consuming an element of a stream except the
1152-- >>> Stream.drain $ Stream.trace putChar $ Stream.intersperseM_ (putChar '.') $ Stream.fromList "hello"
1156{-# INLINE intersperseM_ #-}
1157intersperseM_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
1158intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
1160-- | Intersperse a monadic action into the input stream after every @n@
1164-- > Stream.toList $ Stream.intersperseBySpan 2 (return ',') $ Stream.fromList "hello"
1170{-# INLINE intersperseBySpan #-}
1171intersperseBySpan :: -- IsStream t =>
1172 Int -> m a -> t m a -> t m a
1173intersperseBySpan _n _f _xs = undefined
1175-- | Insert an effect and its output after consuming an element of a stream.
1177-- >>> Stream.toList $ Stream.trace putChar $ intersperseSuffix (putChar '.' >> return ',') $ Stream.fromList "hello"
1178-- h.,e.,l.,l.,o.,"h,e,l,l,o,"
1181{-# INLINE intersperseSuffix #-}
1182intersperseSuffix :: (IsStream t, Monad m) => m a -> t m a -> t m a
1183intersperseSuffix m = fromStreamD . D.intersperseSuffix m . toStreamD
1185-- | Insert a side effect after consuming an element of a stream.
1188-- >>> Stream.mapM_ putChar $ Stream.intersperseSuffix_ (threadDelay 1000000) $ Stream.fromList "hello"
1195{-# INLINE intersperseSuffix_ #-}
1196intersperseSuffix_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
1197intersperseSuffix_ m = fromStreamD . D.intersperseSuffix_ m . toStreamD
1199-- XXX Use an offset argument, like tapOffsetEvery
1201-- | Like 'intersperseSuffix' but intersperses an effectful action into the
1202-- input stream after every @n@ elements and after the last element.
1204-- >>> Stream.toList $ Stream.intersperseSuffixBySpan 2 (return ',') $ Stream.fromList "hello"
1209{-# INLINE intersperseSuffixBySpan #-}
1210intersperseSuffixBySpan :: (IsStream t, Monad m)
1211 => Int -> m a -> t m a -> t m a
1212intersperseSuffixBySpan n eff =
1213 fromStreamD . D.intersperseSuffixBySpan n eff . toStreamD
1215-- | Insert a side effect before consuming an element of a stream.
1217-- >>> Stream.toList $ Stream.trace putChar $ Stream.interspersePrefix_ (putChar '.' >> return ',') $ Stream.fromList "hello"
1220-- Same as 'trace_' but may be concurrent.
1226{-# INLINE interspersePrefix_ #-}
1227interspersePrefix_ :: (IsStream t, MonadAsync m) => m b -> t m a -> t m a
1228interspersePrefix_ m = mapM (\x -> void m >> return x)
1230------------------------------------------------------------------------------
1232------------------------------------------------------------------------------
1234-- Note: delay must be serial.
1236-- | Introduce a delay of specified seconds before consuming an element of the
1237-- stream except the first one.
1239-- >>> Stream.mapM_ print $ Stream.timestamped $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
1240-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),1)
1241-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),2)
1242-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),3)
1247delay :: (IsStream t, MonadIO m) => Double -> t m a -> t m a
1248delay n = intersperseM_ $ liftIO $ threadDelay $ round $ n * 1000000
1250-- Note: delay must be serial.
1252-- | Introduce a delay of specified seconds after consuming an element of a
1255-- >>> Stream.mapM_ print $ Stream.timestamped $ Stream.delayPost 1 $ Stream.enumerateFromTo 1 3
1256-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),1)
1257-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),2)
1258-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),3)
1262{-# INLINE delayPost #-}
1263delayPost :: (IsStream t, MonadIO m) => Double -> t m a -> t m a
1264delayPost n = intersperseSuffix_ $ liftIO $ threadDelay $ round $ n * 1000000
1266-- Note: delay must be serial, that's why 'trace_' is used.
1268-- | Introduce a delay of specified seconds before consuming an element of a
1271-- >>> Stream.mapM_ print $ Stream.timestamped $ Stream.delayPre 1 $ Stream.enumerateFromTo 1 3
1272-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),1)
1273-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),2)
1274-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),3)
1278{-# INLINE delayPre #-}
1279delayPre :: (IsStream t, MonadIO m) => Double -> t m a -> t m a
1280delayPre n = trace_ $ liftIO $ threadDelay $ round $ n * 1000000
1282------------------------------------------------------------------------------
1283-- Reorder in sequence
1284------------------------------------------------------------------------------
1286-- | Buffer until the next element in sequence arrives. The function argument
1287-- determines the difference in sequence numbers. This could be useful in
1288-- implementing sequenced streams, for example, TCP reassembly.
1292{-# INLINE reassembleBy #-}
1294 :: -- (IsStream t, Monad m) =>
1299reassembleBy = undefined
1301------------------------------------------------------------------------------
1303------------------------------------------------------------------------------
1306-- > indexed = Stream.postscanl' (\(i, _) x -> (i + 1, x)) (-1,undefined)
1307-- > indexed = Stream.zipWith (,) (Stream.enumerateFrom 0)
1309-- Pair each element in a stream with its index, starting from index 0.
1311-- >>> Stream.toList $ Stream.indexed $ Stream.fromList "hello"
1312-- [(0,'h'),(1,'e'),(2,'l'),(3,'l'),(4,'o')]
1315{-# INLINE indexed #-}
1316indexed :: (IsStream t, Monad m) => t m a -> t m (Int, a)
1317indexed = fromStreamD . D.indexed . toStreamD
1320-- > indexedR n = Stream.postscanl' (\(i, _) x -> (i - 1, x)) (n + 1,undefined)
1321-- > indexedR n = Stream.zipWith (,) (Stream.enumerateFromThen n (n - 1))
1323-- Pair each element in a stream with its index, starting from the
1324-- given index @n@ and counting down.
1326-- >>> Stream.toList $ Stream.indexedR 10 $ Stream.fromList "hello"
1327-- [(10,'h'),(9,'e'),(8,'l'),(7,'l'),(6,'o')]
1330{-# INLINE indexedR #-}
1331indexedR :: (IsStream t, Monad m) => Int -> t m a -> t m (Int, a)
1332indexedR n = fromStreamD . D.indexedR n . toStreamD
1334-------------------------------------------------------------------------------
1336-------------------------------------------------------------------------------
1338-- Note: The timestamp stream must be the second stream in the zip so that the
1339-- timestamp is generated after generating the stream element and not before.
1340-- If we do not do that then the following example will generate the same
1341-- timestamp for first two elements:
1343-- Stream.mapM_ print $ Stream.timestamped $ Stream.delay $ Stream.enumerateFromTo 1 3
1345-- | Pair each element in a stream with an absolute timestamp, using a clock of
1346-- specified granularity. The timestamp is generated just before the element
1349-- >>> Stream.mapM_ print $ Stream.timestampWith 0.01 $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
1350-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),1)
1351-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),2)
1352-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),3)
1356{-# INLINE timestampWith #-}
1357timestampWith :: (IsStream t, MonadAsync m, Functor (t m))
1358 => Double -> t m a -> t m (AbsTime, a)
1359timestampWith g stream = zipWith (flip (,)) stream (absTimesWith g)
1361-- TBD: check performance vs a custom implementation without using zipWith.
1365{-# INLINE timestamped #-}
1366timestamped :: (IsStream t, MonadAsync m, Functor (t m))
1367 => t m a -> t m (AbsTime, a)
1368timestamped = timestampWith 0.01
1370-- | Pair each element in a stream with relative times starting from 0, using a
1371-- clock with the specified granularity. The time is measured just before the
1372-- element is consumed.
1374-- >>> Stream.mapM_ print $ Stream.timeIndexWith 0.01 $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
1375-- (RelTime64 (NanoSecond64 ...),1)
1376-- (RelTime64 (NanoSecond64 ...),2)
1377-- (RelTime64 (NanoSecond64 ...),3)
1381{-# INLINE timeIndexWith #-}
1382timeIndexWith :: (IsStream t, MonadAsync m, Functor (t m))
1383 => Double -> t m a -> t m (RelTime64, a)
1384timeIndexWith g stream = zipWith (flip (,)) stream (relTimesWith g)
1386-- | Pair each element in a stream with relative times starting from 0, using a
1387-- 10 ms granularity clock. The time is measured just before the element is
1390-- >>> Stream.mapM_ print $ Stream.timeIndexed $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
1391-- (RelTime64 (NanoSecond64 ...),1)
1392-- (RelTime64 (NanoSecond64 ...),2)
1393-- (RelTime64 (NanoSecond64 ...),3)
1397{-# INLINE timeIndexed #-}
1398timeIndexed :: (IsStream t, MonadAsync m, Functor (t m))
1399 => t m a -> t m (RelTime64, a)
1400timeIndexed = timeIndexWith 0.01
1402------------------------------------------------------------------------------
1404------------------------------------------------------------------------------
1406-- | Find all the indices where the value of the element in the stream is equal
1407-- to the given value.
1409-- > elemIndices a = findIndices (== a)
1412{-# INLINE elemIndices #-}
1413elemIndices :: (IsStream t, Eq a, Monad m) => a -> t m a -> t m Int
1414elemIndices a = findIndices (== a)
1416------------------------------------------------------------------------------
1418------------------------------------------------------------------------------
1420-- XXX this is not a one-to-one map so calling it map may not be right.
1421-- We can perhaps call it zipWithTail or rollWith.
1423-- | Apply a function on every two successive elements of a stream. The first
1424-- argument of the map function is the previous element and the second argument
1425-- is the current element. When the current element is the first element, the
1426-- previous element is 'Nothing'.
1430{-# INLINE rollingMap #-}
1431rollingMap :: (IsStream t, Monad m) => (Maybe a -> a -> b) -> t m a -> t m b
1432rollingMap f m = fromStreamD $ D.rollingMap f $ toStreamD m
1434-- | Like 'rollingMap' but with an effectful map function.
1438{-# INLINE rollingMapM #-}
1439rollingMapM :: (IsStream t, Monad m) => (Maybe a -> a -> m b) -> t m a -> t m b
1440rollingMapM f m = fromStreamD $ D.rollingMapM f $ toStreamD m
1442-- | Like 'rollingMap' but requires at least two elements in the stream,
1443-- returns an empty stream otherwise.
1445-- This is the stream equivalent of the list idiom @zipWith f xs (tail xs)@.
1449{-# INLINE rollingMap2 #-}
1450rollingMap2 :: (IsStream t, Monad m) => (a -> a -> b) -> t m a -> t m b
1451rollingMap2 f m = fromStreamD $ D.rollingMap2 f $ toStreamD m
1453------------------------------------------------------------------------------
1455------------------------------------------------------------------------------
1457-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'
1458-- elements, and return a stream of values extracted from 'Just'.
1463-- mapMaybe f = Stream.map 'fromJust' . Stream.filter 'isJust' . Stream.map f
1467{-# INLINE mapMaybe #-}
1468mapMaybe :: (IsStream t, Monad m) => (a -> Maybe b) -> t m a -> t m b
1469mapMaybe f m = fromStreamD $ D.mapMaybe f $ toStreamD m
1471-- | Like 'mapMaybe' but maps a monadic function.
1476-- mapMaybeM f = Stream.map 'fromJust' . Stream.filter 'isJust' . Stream.mapM f
1479-- /Concurrent (do not use with 'fromParallel' on infinite streams)/
1482{-# INLINE_EARLY mapMaybeM #-}
1483mapMaybeM :: (IsStream t, MonadAsync m, Functor (t m))
1484 => (a -> m (Maybe b)) -> t m a -> t m b
1485mapMaybeM f = fmap fromJust . filter isJust . mapM f
1487{-# RULES "mapMaybeM serial" mapMaybeM = mapMaybeMSerial #-}
1488{-# INLINE mapMaybeMSerial #-}
1489mapMaybeMSerial :: Monad m => (a -> m (Maybe b)) -> SerialT m a -> SerialT m b
1490mapMaybeMSerial f m = fromStreamD $ D.mapMaybeM f $ toStreamD m
1492-- | In a stream of 'Maybe's, discard 'Nothing's and unwrap 'Just's.
1496{-# INLINE catMaybes #-}
1497catMaybes :: (IsStream t, Monad m, Functor (t m)) => t m (Maybe a) -> t m a
1498catMaybes = fmap fromJust . filter isJust
1500------------------------------------------------------------------------------
1502------------------------------------------------------------------------------
1504-- | Discard 'Right's and unwrap 'Left's in an 'Either' stream.
1509lefts :: (IsStream t, Monad m, Functor (t m)) => t m (Either a b) -> t m a
1510lefts = fmap (fromLeft undefined) . filter isLeft
1512-- | Discard 'Left's and unwrap 'Right's in an 'Either' stream.
1516{-# INLINE rights #-}
1517rights :: (IsStream t, Monad m, Functor (t m)) => t m (Either a b) -> t m b
1518rights = fmap (fromRight undefined) . filter isRight
1520-- | Remove the either wrapper and flatten both lefts and as well as rights in
1521-- the output stream.
1526both :: Functor (t m) => t m (Either a a) -> t m a
1527both = fmap (either id id)
1529------------------------------------------------------------------------------
1530-- Concurrent Application
1531------------------------------------------------------------------------------
1533-- | Parallel transform application operator; applies a stream transformation
1534-- function @t m a -> t m b@ to a stream @t m a@ concurrently; the input stream
1535-- is evaluated asynchronously in an independent thread yielding elements to a
1536-- buffer and the transformation function runs in another thread consuming the
1537-- input from the buffer. '|$' is just like regular function application
1538-- operator '$' except that it is concurrent.
1540-- If you read the signature as @(t m a -> t m b) -> (t m a -> t m b)@ you can
1541-- look at it as a transformation that converts a transform function to a
1542-- buffered concurrent transform function.
1544-- The following code prints a value every second even though each stage adds a
1550-- Stream.mapM (\x -> threadDelay 1000000 >> print x)
1551-- |$ Stream.replicateM 3 (threadDelay 1000000 >> return 1)
1559-- /Since: 0.3.0 ("Streamly")/
1563(|$) :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> (t m a -> t m b)
1564-- (|$) f = f . mkAsync
1565(|$) f = f . mkParallel
1573{-# INLINE applyAsync #-}
1574applyAsync :: (IsStream t, MonadAsync m)
1575 => (t m a -> t m b) -> (t m a -> t m b)
1578-- | Same as '|$' but with arguments reversed.
1584-- /Since: 0.3.0 ("Streamly")/
1588(|&) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> t m b) -> t m b
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