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'.
20 -- | Stateless one-to-one maps.
26 -- * Mapping Side Effects (Observation)
35 -- * Scanning By 'Fold'
40 -- | Left scans. Stateful, mostly one-to-one maps.
51 -- XXX Once we have pipes the contravariant transformations can be
52 -- represented by attaching pipes before a transformation.
65 -- | Produce a subset of the stream.
71 -- , uniqBy -- by predicate e.g. to remove duplicate "/" in a path
72 -- , uniqOn -- to remove duplicate sequences
73 -- , pruneBy -- dropAround + uniqBy - like words
76 -- | Produce a subset of the stream trimmed at ends.
94 -- * Inserting Elements
95 -- | Produce a superset of the stream.
99 , intersperseM -- XXX naming
101 , intersperseSuffixBySpan
102 -- , intersperseBySpan
103 -- , intersperseByIndices -- using an index function/stream
105 -- * Inserting Side Effects
106 , intersperseM_ -- XXX naming
111 -- , intersperseByTime
112 -- , intersperseByEvent
123 -- * Position Indexing
130 -- , timestampedR -- timer
135 , findIndices -- XXX indicesBy
136 , elemIndices -- XXX indicesOf
139 -- | Map using the previous element.
144 -- Move these to Streamly.Data.Maybe.Stream?
145 , catMaybes -- XXX justs (like lefts/rights)
150 -- Move these to Streamly.Data.Either.Stream?
155 -- ** Concurrent Pipelines
156 -- | Run streaming stages concurrently.
163 -- ** Concurrency Control
185import Control.Concurrent (threadDelay)
186import Control.Monad (void)
187import Control.Monad.Catch (MonadCatch)
188import Control.Monad.IO.Class (MonadIO(..))
189import Control.Monad.Trans.Class (MonadTrans(..))
190import Data.Either (isLeft, isRight)
191import Data.Maybe (isJust, fromJust)
192import Streamly.Internal.BaseCompat (fromLeft, fromRight)
193import Streamly.Internal.Data.Fold.Types (Fold (..))
194import Streamly.Internal.Data.Pipe.Types (Pipe (..))
195import Streamly.Internal.Data.Stream.IsStream.Combinators
196 ( inspectMode, maxBuffer, maxThreads, rate, avgRate, minRate
197 , maxRate, constRate)
198import Streamly.Internal.Data.Stream.IsStream.Common
213import Streamly.Internal.Data.Stream.Prelude (fromStreamS, toStreamS)
214import Streamly.Internal.Data.Stream.Serial (SerialT)
215import Streamly.Internal.Data.Stream.StreamD (fromStreamD, toStreamD)
216import Streamly.Internal.Data.Stream.StreamK (IsStream)
217import Streamly.Internal.Data.SVar (MonadAsync, Rate(..))
218import Streamly.Internal.Data.Time.Units ( TimeUnit64, AbsTime, RelTime64)
220import qualified Streamly.Internal.Data.Fold as FL
221import qualified Streamly.Internal.Data.Stream.Parallel as Par
222import qualified Streamly.Internal.Data.Stream.Prelude as P
223import qualified Streamly.Internal.Data.Stream.Serial as Serial
224import qualified Streamly.Internal.Data.Stream.StreamD as D
225import qualified Streamly.Internal.Data.Stream.StreamK as K
226import qualified Streamly.Internal.Data.Stream.Zip as Z
227#ifdef USE_STREAMK_ONLY
228import qualified Streamly.Internal.Data.Stream.StreamK as S
230import qualified Streamly.Internal.Data.Stream.StreamD as S
234 ( filter, drop, dropWhile, take, takeWhile, foldr, map, mapM, sequence
235 , reverse, foldr1 , scanl, scanl1)
237------------------------------------------------------------------------------
239------------------------------------------------------------------------------
241-- | Use a 'Pipe' to transform a stream.
245{-# INLINE transform #-}
246transform :: (IsStream t, Monad m) => Pipe m a b -> t m a -> t m b
247transform pipe xs = fromStreamD $ D.transform pipe (toStreamD xs)
249------------------------------------------------------------------------------
250-- Transformation Folds
251------------------------------------------------------------------------------
253-- | Right fold to a streaming monad.
255-- > foldrS S.cons S.nil === id
257-- 'foldrS' can be used to perform stateless stream to stream transformations
258-- like map and filter in general. It can be coupled with a scan to perform
259-- stateful transformations. However, note that the custom map and filter
260-- routines can be much more efficient than this due to better stream fusion.
262-- >>> S.toList $ S.foldrS S.cons S.nil $ S.fromList [1..5]
265-- Find if any element in the stream is 'True':
267-- >>> S.toList $ S.foldrS (\x xs -> if odd x then return True else xs) (return False) $ (S.fromList (2:4:5:undefined) :: SerialT IO Int)
270-- Map (+2) on odd elements and filter out the even elements:
272-- >>> S.toList $ S.foldrS (\x xs -> if odd x then (x + 2) `S.cons` xs else xs) S.nil $ (S.fromList [1..5] :: SerialT IO Int)
275-- 'foldrM' can also be represented in terms of 'foldrS', however, the former
276-- is much more efficient:
278-- > foldrM f z s = runIdentityT $ foldrS (\x xs -> lift $ f x (runIdentityT xs)) (lift z) s
282foldrS :: IsStream t => (a -> t m b -> t m b) -> t m b -> t m a -> t m b
285-- | Right fold to a transformer monad. This is the most general right fold
286-- function. 'foldrS' is a special case of 'foldrT', however 'foldrS'
287-- implementation can be more efficient:
290-- > foldrM f z s = runIdentityT $ foldrT (\x xs -> lift $ f x (runIdentityT xs)) (lift z) s
292-- 'foldrT' can be used to translate streamly streams to other transformer
293-- monads e.g. to a different streaming type.
297foldrT :: (IsStream t, Monad m, Monad (s m), MonadTrans s)
298 => (a -> s m b -> s m b) -> s m b -> t m a -> s m b
299foldrT f z s = S.foldrT f z (toStreamS s)
301------------------------------------------------------------------------------
302-- Transformation by Mapping
303------------------------------------------------------------------------------
307-- mapM f = sequence . map f
310-- Apply a monadic function to each element of the stream and replace it with
311-- the output of the resulting action.
314-- > drain $ S.mapM putStr $ S.fromList ["a", "b", "c"]
317-- drain $ S.replicateM 10 (return 1)
318-- & (serially . S.mapM (\\x -> threadDelay 1000000 >> print x))
320-- drain $ S.replicateM 10 (return 1)
321-- & (asyncly . S.mapM (\\x -> threadDelay 1000000 >> print x))
324-- /Concurrent (do not use with 'parallely' on infinite streams)/
327{-# INLINE_EARLY mapM #-}
328mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
331{-# RULES "mapM serial" mapM = mapMSerial #-}
332{-# INLINE mapMSerial #-}
333mapMSerial :: Monad m => (a -> m b) -> SerialT m a -> SerialT m b
334mapMSerial = Serial.mapM
341-- Replace the elements of a stream of monadic actions with the outputs of
345-- > drain $ S.sequence $ S.fromList [putStr "a", putStr "b", putStrLn "c"]
348-- drain $ S.replicateM 10 (return $ threadDelay 1000000 >> print 1)
349-- & (serially . S.sequence)
351-- drain $ S.replicateM 10 (return $ threadDelay 1000000 >> print 1)
352-- & (asyncly . S.sequence)
355-- /Concurrent (do not use with 'parallely' on infinite streams)/
358{-# INLINE sequence #-}
359sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
360sequence m = fromStreamS $ S.sequence (toStreamS m)
362------------------------------------------------------------------------------
363-- Mapping side effects
364------------------------------------------------------------------------------
366-- | Tap the data flowing through a stream into a 'Fold'. For example, you may
367-- add a tap to log the contents flowing through the stream. The fold is used
368-- only for effects, its result is discarded.
373-- -----stream m a ---------------stream m a-----
378-- > S.drain $ S.tap (FL.drainBy print) (S.enumerateFromTo 1 2)
383-- Compare with 'trace'.
387tap :: (IsStream t, Monad m) => FL.Fold m a b -> t m a -> t m a
388tap f xs = D.fromStreamD $ D.tap f (D.toStreamD xs)
390-- XXX Remove this. It can be expressed in terms of Fold.sampleFromThen.
392-- | @tapOffsetEvery offset n@ taps every @n@th element in the stream
393-- starting at @offset@. @offset@ can be between @0@ and @n - 1@. Offset 0
394-- means start at the first element in the stream. If the offset is outside
395-- this range then @offset `mod` n@ is used as offset.
398-- >>> S.drain $ S.tapOffsetEvery 0 2 (FL.rmapM print FL.toList) $ S.enumerateFromTo 0 10
404{-# INLINE tapOffsetEvery #-}
405tapOffsetEvery :: (IsStream t, Monad m)
406 => Int -> Int -> FL.Fold m a b -> t m a -> t m a
407tapOffsetEvery offset n f xs =
408 D.fromStreamD $ D.tapOffsetEvery offset n f (D.toStreamD xs)
410-- | Redirect a copy of the stream to a supplied fold and run it concurrently
411-- in an independent thread. The fold may buffer some elements. The buffer size
412-- is determined by the prevailing 'maxBuffer' setting.
417-- -----stream m a ---------------stream m a-----
422-- > S.drain $ S.tapAsync (S.mapM_ print) (S.enumerateFromTo 1 2)
427-- Exceptions from the concurrently running fold are propagated to the current
428-- computation. Note that, because of buffering in the fold, exceptions may be
429-- delayed and may not correspond to the current element being processed in the
430-- parent stream, but we guarantee that before the parent stream stops the tap
431-- finishes and all exceptions from it are drained.
434-- Compare with 'tap'.
437{-# INLINE tapAsync #-}
438tapAsync :: (IsStream t, MonadAsync m) => FL.Fold m a b -> t m a -> t m a
439tapAsync f xs = D.fromStreamD $ Par.tapAsyncF f (D.toStreamD xs)
441-- | @pollCounts predicate transform fold stream@ counts those elements in the
442-- stream that pass the @predicate@. The resulting count stream is sent to
443-- another thread which transforms it using @transform@ and then folds it using
444-- @fold@. The thread is automatically cleaned up if the stream stops or
445-- aborts due to exception.
447-- For example, to print the count of elements processed every second:
450-- > S.drain $ S.pollCounts (const True) (S.rollingMap (-) . S.delayPost 1) (FL.drainBy print)
451-- $ S.enumerateFrom 0
454-- Note: This may not work correctly on 32-bit machines.
458{-# INLINE pollCounts #-}
460 (IsStream t, MonadAsync m)
462 -> (t m Int -> t m Int)
466pollCounts predicate transf f xs =
468 $ D.pollCounts predicate (D.toStreamD . transf . D.fromStreamD) f
471-- | Calls the supplied function with the number of elements consumed
472-- every @n@ seconds. The given function is run in a separate thread
473-- until the end of the stream. In case there is an exception in the
474-- stream the thread is killed during the next major GC.
476-- Note: The action is not guaranteed to run if the main thread exits.
479-- > delay n = threadDelay (round $ n * 1000000) >> return n
480-- > S.drain $ S.tapRate 2 (\\n -> print $ show n ++ " elements processed") (delay 1 S.|: delay 0.5 S.|: delay 0.5 S.|: S.nil)
481-- 2 elements processed
482-- 1 elements processed
485-- Note: This may not work correctly on 32-bit machines.
488{-# INLINE tapRate #-}
490 (IsStream t, MonadAsync m, MonadCatch m)
495tapRate n f xs = D.fromStreamD $ D.tapRate n f $ D.toStreamD xs
497-- | Apply a monadic function to each element flowing through the stream and
498-- discard the results.
501-- > S.drain $ S.trace print (S.enumerateFromTo 1 2)
506-- Compare with 'tap'.
510trace :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m a
511trace f = mapM (\x -> void (f x) >> return x)
513-- | Perform a side effect before yielding each element of the stream and
514-- discard the results.
517-- > S.drain $ S.trace_ (print "got here") (S.enumerateFromTo 1 2)
522-- See also: 'trace', 'interspersePrefix_'
526trace_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
527trace_ eff = Serial.mapM (\x -> eff >> return x)
529------------------------------------------------------------------------------
530-- Scanning with a Fold
531------------------------------------------------------------------------------
533-- | Scan a stream using the given monadic fold.
537scan :: (IsStream t, Monad m) => Fold m a b -> t m a -> t m b
540-- | Postscan a stream using the given monadic fold.
543{-# INLINE postscan #-}
544postscan :: (IsStream t, Monad m) => Fold m a b -> t m a -> t m b
545postscan = P.postscanOnce
547------------------------------------------------------------------------------
548-- Scanning - Transformation by Folding
549------------------------------------------------------------------------------
551-- XXX It may be useful to have a version of scan where we can keep the
552-- accumulator independent of the value emitted. So that we do not necessarily
553-- have to keep a value in the accumulator which we are not using. We can pass
554-- an extraction function that will take the accumulator and the current value
555-- of the element and emit the next value in the stream. That will also make it
556-- possible to modify the accumulator after using it. In fact, the step function
557-- can return new accumulator and the value to be emitted. The signature would
558-- be more like mapAccumL. Or we can change the signature of scanx to
561-- | Strict left scan with an extraction function. Like 'scanl'', but applies a
562-- user supplied extraction function (the third argument) at each step. This is
563-- designed to work with the @foldl@ library. The suffix @x@ is a mnemonic for
568-- /Since: 0.7.0 (Monad m constraint)/
569{-# DEPRECATED scanx "Please use scanl followed by map instead." #-}
571scanx :: (IsStream t, Monad m) => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
574-- XXX this needs to be concurrent
575-- XXX because of the use of D.cons for appending, scanlM' has quadratic
576-- complexity when iterated over a stream. We should use StreamK style scanlM'
577-- for linear performance on iteration.
579-- | Like 'scanl'' but with a monadic step function and a monadic seed.
583-- /Since: 0.8.0 (signature change)/
584{-# INLINE scanlM' #-}
585scanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
586scanlM' step begin m = fromStreamD $ D.scanlM' step begin $ toStreamD m
588-- XXX because of the use of D.cons for appending, scanl' has quadratic
589-- complexity when iterated over a stream. We should use StreamK style scanlM'
590-- for linear performance on iteration.
592-- | Strict left scan. Like 'map', 'scanl'' too is a one to one transformation,
593-- however it adds an extra element.
596-- > S.toList $ S.scanl' (+) 0 $ fromList [1,2,3,4]
601-- > S.toList $ S.scanl' (flip (:)) [] $ S.fromList [1,2,3,4]
602-- [[],[1],[2,1],[3,2,1],[4,3,2,1]]
605-- The output of 'scanl'' is the initial value of the accumulator followed by
606-- all the intermediate steps and the final result of 'foldl''.
608-- By streaming the accumulated state after each fold step, we can share the
609-- state across multiple stages of stream composition. Each stage can modify or
610-- extend the state, do some processing with it and emit it for the next stage,
611-- thus modularizing the stream processing. This can be useful in
612-- stateful or event-driven programming.
614-- Consider the following monolithic example, computing the sum and the product
615-- of the elements in a stream in one go using a @foldl'@:
618-- > S.foldl' (\\(s, p) x -> (s + x, p * x)) (0,1) $ S.fromList \[1,2,3,4]
622-- Using @scanl'@ we can make it modular by computing the sum in the first
623-- stage and passing it down to the next stage for computing the product:
626-- > S.foldl' (\\(_, p) (s, x) -> (s, p * x)) (0,1)
627-- $ S.scanl' (\\(s, _) x -> (s + x, x)) (0,1)
628-- $ S.fromList \[1,2,3,4]
632-- IMPORTANT: 'scanl'' evaluates the accumulator to WHNF. To avoid building
633-- lazy expressions inside the accumulator, it is recommended that a strict
634-- data structure is used for accumulator.
636-- See also: 'usingStateT'
640scanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
641scanl' step z m = fromStreamS $ S.scanl' step z $ toStreamS m
643-- | Like 'scanl'' but does not stream the initial value of the accumulator.
645-- > postscanl' f z xs = S.drop 1 $ S.scanl' f z xs
648{-# INLINE postscanl' #-}
649postscanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
650postscanl' step z m = fromStreamD $ D.postscanl' step z $ toStreamD m
652-- XXX prescanl does not sound very useful, enable only if there is a
653-- compelling use case.
655-- | Like scanl' but does not stream the final value of the accumulator.
658{-# INLINE prescanl' #-}
659prescanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
660prescanl' step z m = fromStreamD $ D.prescanl' step z $ toStreamD m
662-- XXX this needs to be concurrent
663-- | Like prescanl' but with a monadic step function and a monadic seed.
666{-# INLINE prescanlM' #-}
667prescanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
668prescanlM' step z m = fromStreamD $ D.prescanlM' step z $ toStreamD m
670-- XXX this needs to be concurrent
671-- | Like 'scanl1'' but with a monadic step function.
674{-# INLINE scanl1M' #-}
675scanl1M' :: (IsStream t, Monad m) => (a -> a -> m a) -> t m a -> t m a
676scanl1M' step m = fromStreamD $ D.scanl1M' step $ toStreamD m
678-- | Like 'scanl'' but for a non-empty stream. The first element of the stream
679-- is used as the initial value of the accumulator. Does nothing if the stream
683-- > S.toList $ S.scanl1 (+) $ fromList [1,2,3,4]
688{-# INLINE scanl1' #-}
689scanl1' :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> t m a
690scanl1' step m = fromStreamD $ D.scanl1' step $ toStreamD m
692------------------------------------------------------------------------------
694------------------------------------------------------------------------------
696-- | Include only those elements that pass a predicate.
700#if __GLASGOW_HASKELL__ != 802
701-- GHC 8.2.2 crashes with this code, when used with "stack"
702filter :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
703filter p m = fromStreamS $ S.filter p $ toStreamS m
705filter :: IsStream t => (a -> Bool) -> t m a -> t m a
709-- | Same as 'filter' but with a monadic predicate.
712{-# INLINE filterM #-}
713filterM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
714filterM p m = fromStreamD $ D.filterM p $ toStreamD m
716-- | Drop repeated elements that are adjacent to each other.
720uniq :: (Eq a, IsStream t, Monad m) => t m a -> t m a
721uniq = fromStreamD . D.uniq . toStreamD
723-- | Deletes the first occurrence of the element in the stream that satisfies
724-- the given equality predicate.
727-- > S.toList $ S.deleteBy (==) 3 $ S.fromList [1,3,3,5]
732{-# INLINE deleteBy #-}
733deleteBy :: (IsStream t, Monad m) => (a -> a -> Bool) -> a -> t m a -> t m a
734deleteBy cmp x m = fromStreamS $ S.deleteBy cmp x (toStreamS m)
736------------------------------------------------------------------------------
738------------------------------------------------------------------------------
740-- | Same as 'takeWhile' but with a monadic predicate.
743{-# INLINE takeWhileM #-}
744takeWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
745takeWhileM p m = fromStreamD $ D.takeWhileM p $ toStreamD m
747-- | @takeByTime duration@ yields stream elements upto specified time
748-- @duration@. The duration starts when the stream is evaluated for the first
749-- time, before the first element is yielded. The time duration is checked
750-- before generating each element, if the duration has expired the stream
753-- The total time taken in executing the stream is guaranteed to be /at least/
754-- @duration@, however, because the duration is checked before generating an
755-- element, the upper bound is indeterminate and depends on the time taken in
756-- generating and processing the last element.
758-- No element is yielded if the duration is zero. At least one element is
759-- yielded if the duration is non-zero.
763{-# INLINE takeByTime #-}
764takeByTime ::(MonadIO m, IsStream t, TimeUnit64 d) => d -> t m a -> t m a
765takeByTime d = fromStreamD . D.takeByTime d . toStreamD
767-- | Drop elements in the stream as long as the predicate succeeds and then
768-- take the rest of the stream.
771{-# INLINE dropWhile #-}
772dropWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
773dropWhile p m = fromStreamS $ S.dropWhile p $ toStreamS m
775-- | Same as 'dropWhile' but with a monadic predicate.
778{-# INLINE dropWhileM #-}
779dropWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
780dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m
782-- | @dropByTime duration@ drops stream elements until specified @duration@ has
783-- passed. The duration begins when the stream is evaluated for the first
784-- time. The time duration is checked /after/ generating a stream element, the
785-- element is yielded if the duration has expired otherwise it is dropped.
787-- The time elapsed before starting to generate the first element is /at most/
788-- @duration@, however, because the duration expiry is checked after the
789-- element is generated, the lower bound is indeterminate and depends on the
790-- time taken in generating an element.
792-- All elements are yielded if the duration is zero.
796{-# INLINE dropByTime #-}
797dropByTime ::(MonadIO m, IsStream t, TimeUnit64 d) => d -> t m a -> t m a
798dropByTime d = fromStreamD . D.dropByTime d . toStreamD
800------------------------------------------------------------------------------
802------------------------------------------------------------------------------
804-- | @insertBy cmp elem stream@ inserts @elem@ before the first element in
805-- @stream@ that is less than @elem@ when compared using @cmp@.
808-- insertBy cmp x = 'mergeBy' cmp ('yield' x)
812-- > S.toList $ S.insertBy compare 2 $ S.fromList [1,3,5]
817{-# INLINE insertBy #-}
819 (IsStream t, Monad m) => (a -> a -> Ordering) -> a -> t m a -> t m a
820insertBy cmp x m = fromStreamS $ S.insertBy cmp x (toStreamS m)
822-- | Insert a pure value between successive elements of a stream.
825-- > S.toList $ S.intersperse ',' $ S.fromList "hello"
830{-# INLINE intersperse #-}
831intersperse :: (IsStream t, MonadAsync m) => a -> t m a -> t m a
832intersperse a = fromStreamS . S.intersperse a . toStreamS
834-- | Insert a side effect before consuming an element of a stream except the
838-- >>> S.drain $ S.trace putChar $ S.intersperseM_ (putChar '.') $ S.fromList "hello"
843{-# INLINE intersperseM_ #-}
844intersperseM_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
845intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
848-- | Intersperse a monadic action into the input stream after every @n@
852-- > S.toList $ S.intersperseBySpan 2 (return ',') $ S.fromList "hello"
857{-# INLINE intersperseBySpan #-}
858intersperseBySpan :: IsStream t => Int -> m a -> t m a -> t m a
859intersperseBySpan _n _f _xs = undefined
862-- | Insert an effect and its output after consuming an element of a stream.
865-- >>> S.toList $ S.trace putChar $ S.intersperseSuffix (putChar '.' >> return ',') $ S.fromList "hello"
866-- > h.,e.,l.,l.,o.,"h,e,l,l,o,"
870{-# INLINE intersperseSuffix #-}
871intersperseSuffix :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
872intersperseSuffix m = fromStreamD . D.intersperseSuffix m . toStreamD
874-- | Insert a side effect after consuming an element of a stream.
877-- > S.mapM_ putChar $ S.intersperseSuffix_ (threadDelay 1000000) $ S.fromList "hello"
882{-# INLINE intersperseSuffix_ #-}
883intersperseSuffix_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
884intersperseSuffix_ m = fromStreamD . D.intersperseSuffix_ m . toStreamD
886-- | Like 'intersperseSuffix' but intersperses an effectful action into the
887-- input stream after every @n@ elements and after the last element.
890-- > S.toList $ S.intersperseSuffixBySpan 2 (return ',') $ S.fromList "hello"
896{-# INLINE intersperseSuffixBySpan #-}
897intersperseSuffixBySpan :: (IsStream t, MonadAsync m)
898 => Int -> m a -> t m a -> t m a
899intersperseSuffixBySpan n eff =
900 fromStreamD . D.intersperseSuffixBySpan n eff . toStreamD
902-- | Insert a side effect before consuming an element of a stream.
905-- >>> S.toList $ S.trace putChar $ S.interspersePrefix_ (putChar '.' >> return ',') $ S.fromList "hello"
906-- > .h.e.l.l.o"hello"
915{-# INLINE interspersePrefix_ #-}
916interspersePrefix_ :: (IsStream t, MonadAsync m) => m b -> t m a -> t m a
917interspersePrefix_ m = mapM (\x -> void m >> return x)
919------------------------------------------------------------------------------
921------------------------------------------------------------------------------
923-- Note: delay must be serial.
925-- | Introduce a delay of specified seconds before consuming an element of the
926-- stream except the first one.
929-- >>> S.mapM_ print $ S.timestamped $ S.delay 1 $ S.enumerateFromTo 1 3
930-- > (AbsTime (TimeSpec {sec = 2502706, nsec = 751137000}),1)
931-- > (AbsTime (TimeSpec {sec = 2502707, nsec = 743535000}),2)
932-- > (AbsTime (TimeSpec {sec = 2502708, nsec = 749758000}),3)
938delay :: (IsStream t, MonadIO m) => Double -> t m a -> t m a
939delay n = intersperseM_ $ liftIO $ threadDelay $ round $ n * 1000000
941-- Note: delay must be serial.
943-- | Introduce a delay of specified seconds after consuming an element of a
947-- >>> S.mapM_ print $ S.timestamped $ S.delayPost 1 $ S.enumerateFromTo 1 3
948-- > (AbsTime (TimeSpec {sec = 2502826, nsec = 119030000}),1)
949-- > (AbsTime (TimeSpec {sec = 2502827, nsec = 111393000}),2)
950-- > (AbsTime (TimeSpec {sec = 2502828, nsec = 112221000}),3)
955{-# INLINE delayPost #-}
956delayPost :: (IsStream t, MonadIO m) => Double -> t m a -> t m a
957delayPost n = intersperseSuffix_ $ liftIO $ threadDelay $ round $ n * 1000000
959-- Note: delay must be serial, that's why 'trace_' is used.
961-- | Introduce a delay of specified seconds before consuming an element of a
965-- >>> S.mapM_ print $ S.timestamped $ S.delayPre 1 $ S.enumerateFromTo 1 3
966-- > (AbsTime (TimeSpec {sec = 2502207, nsec = 533177000}),1)
967-- > (AbsTime (TimeSpec {sec = 2502208, nsec = 530859000}),2)
968-- > (AbsTime (TimeSpec {sec = 2502209, nsec = 531619000}),3)
973{-# INLINE delayPre #-}
974delayPre :: (IsStream t, MonadIO m) => Double -> t m a -> t m a
975delayPre n = trace_ $ liftIO $ threadDelay $ round $ n * 1000000
977------------------------------------------------------------------------------
978-- Reorder in sequence
979------------------------------------------------------------------------------
981-- | Buffer until the next element in sequence arrives. The function argument
982-- determines the difference in sequence numbers. This could be useful in
983-- implementing sequenced streams, for example, TCP reassembly.
987{-# INLINE reassembleBy #-}
989 :: -- (IsStream t, Monad m) =>
994reassembleBy = undefined
996------------------------------------------------------------------------------
998------------------------------------------------------------------------------
1001-- > indexed = S.postscanl' (\(i, _) x -> (i + 1, x)) (-1,undefined)
1002-- > indexed = S.zipWith (,) (S.enumerateFrom 0)
1004-- Pair each element in a stream with its index, starting from index 0.
1007-- > S.toList $ S.indexed $ S.fromList "hello"
1008-- [(0,'h'),(1,'e'),(2,'l'),(3,'l'),(4,'o')]
1012{-# INLINE indexed #-}
1013indexed :: (IsStream t, Monad m) => t m a -> t m (Int, a)
1014indexed = fromStreamD . D.indexed . toStreamD
1017-- > indexedR n = S.postscanl' (\(i, _) x -> (i - 1, x)) (n + 1,undefined)
1018-- > indexedR n = S.zipWith (,) (S.enumerateFromThen n (n - 1))
1020-- Pair each element in a stream with its index, starting from the
1021-- given index @n@ and counting down.
1024-- > S.toList $ S.indexedR 10 $ S.fromList "hello"
1025-- [(10,'h'),(9,'e'),(8,'l'),(7,'l'),(6,'o')]
1029{-# INLINE indexedR #-}
1030indexedR :: (IsStream t, Monad m) => Int -> t m a -> t m (Int, a)
1031indexedR n = fromStreamD . D.indexedR n . toStreamD
1033-------------------------------------------------------------------------------
1035-------------------------------------------------------------------------------
1037-- Note: The timestamp stream must be the second stream in the zip so that the
1038-- timestamp is generated after generating the stream element and not before.
1039-- If we do not do that then the following example will generate the same
1040-- timestamp for first two elements:
1042-- S.mapM_ print $ S.timestamped $ S.delay $ S.enumerateFromTo 1 3
1044-- | Pair each element in a stream with an absolute timestamp, using a clock of
1045-- specified granularity. The timestamp is generated just before the element
1049-- >>> S.mapM_ print $ S.timestampWith 0.01 $ S.delay 1 $ S.enumerateFromTo 1 3
1054{-# INLINE timestampWith #-}
1055timestampWith :: (IsStream t, MonadAsync m, Functor (t m))
1056 => Double -> t m a -> t m (AbsTime, a)
1057timestampWith g stream = Z.zipWith (flip (,)) stream (absTimesWith g)
1059-- TBD: check performance vs a custom implementation without using zipWith.
1063{-# INLINE timestamped #-}
1064timestamped :: (IsStream t, MonadAsync m, Functor (t m))
1065 => t m a -> t m (AbsTime, a)
1066timestamped = timestampWith 0.01
1068-- | Pair each element in a stream with relative times starting from 0, using a
1069-- clock with the specified granularity. The time is measured just before the
1070-- element is consumed.
1073-- >>> S.mapM_ print $ S.timeIndexWith 0.01 $ S.delay 1 $ S.enumerateFromTo 1 3
1078{-# INLINE timeIndexWith #-}
1079timeIndexWith :: (IsStream t, MonadAsync m, Functor (t m))
1080 => Double -> t m a -> t m (RelTime64, a)
1081timeIndexWith g stream = Z.zipWith (flip (,)) stream (relTimesWith g)
1083-- | Pair each element in a stream with relative times starting from 0, using a
1084-- 10 ms granularity clock. The time is measured just before the element is
1088-- >>> S.mapM_ print $ S.timeIndexed $ S.delay 1 $ S.enumerateFromTo 1 3
1089-- (RelTime64 (NanoSecond64 0),1)
1090-- (RelTime64 (NanoSecond64 996239000),2)
1091-- (RelTime64 (NanoSecond64 1996712000),3)
1096{-# INLINE timeIndexed #-}
1097timeIndexed :: (IsStream t, MonadAsync m, Functor (t m))
1098 => t m a -> t m (RelTime64, a)
1099timeIndexed = timeIndexWith 0.01
1101------------------------------------------------------------------------------
1103------------------------------------------------------------------------------
1105-- | Find all the indices where the value of the element in the stream is equal
1106-- to the given value.
1108-- > elemIndices a = findIndices (== a)
1111{-# INLINE elemIndices #-}
1112elemIndices :: (IsStream t, Eq a, Monad m) => a -> t m a -> t m Int
1113elemIndices a = findIndices (== a)
1115------------------------------------------------------------------------------
1117------------------------------------------------------------------------------
1119-- XXX this is not a one-to-one map so calling it map may not be right.
1120-- We can perhaps call it zipWithTail or rollWith.
1122-- | Apply a function on every two successive elements of a stream. If the
1123-- stream consists of a single element the output is an empty stream.
1125-- This is the stream equivalent of the list idiom @zipWith f xs (tail xs)@.
1129{-# INLINE rollingMap #-}
1130rollingMap :: (IsStream t, Monad m) => (a -> a -> b) -> t m a -> t m b
1131rollingMap f m = fromStreamD $ D.rollingMap f $ toStreamD m
1133-- | Like 'rollingMap' but with an effectful map function.
1137{-# INLINE rollingMapM #-}
1138rollingMapM :: (IsStream t, Monad m) => (a -> a -> m b) -> t m a -> t m b
1139rollingMapM f m = fromStreamD $ D.rollingMapM f $ toStreamD m
1141------------------------------------------------------------------------------
1143------------------------------------------------------------------------------
1145-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'
1146-- elements, and return a stream of values extracted from 'Just'.
1151-- mapMaybe f = S.map 'fromJust' . S.filter 'isJust' . S.map f
1155{-# INLINE mapMaybe #-}
1156mapMaybe :: (IsStream t, Monad m) => (a -> Maybe b) -> t m a -> t m b
1157mapMaybe f m = fromStreamS $ S.mapMaybe f $ toStreamS m
1159-- | Like 'mapMaybe' but maps a monadic function.
1164-- mapMaybeM f = S.map 'fromJust' . S.filter 'isJust' . S.mapM f
1167-- /Concurrent (do not use with 'parallely' on infinite streams)/
1170{-# INLINE_EARLY mapMaybeM #-}
1171mapMaybeM :: (IsStream t, MonadAsync m, Functor (t m))
1172 => (a -> m (Maybe b)) -> t m a -> t m b
1173mapMaybeM f = fmap fromJust . filter isJust . K.mapM f
1175{-# RULES "mapMaybeM serial" mapMaybeM = mapMaybeMSerial #-}
1176{-# INLINE mapMaybeMSerial #-}
1177mapMaybeMSerial :: Monad m => (a -> m (Maybe b)) -> SerialT m a -> SerialT m b
1178mapMaybeMSerial f m = fromStreamD $ D.mapMaybeM f $ toStreamD m
1180-- | In a stream of 'Maybe's, discard 'Nothing's and unwrap 'Just's.
1184catMaybes :: (IsStream t, Monad m, Functor (t m)) => t m (Maybe a) -> t m a
1185catMaybes = fmap fromJust . filter isJust
1187------------------------------------------------------------------------------
1189------------------------------------------------------------------------------
1191-- | Discard 'Right's and unwrap 'Left's in an 'Either' stream.
1195lefts :: (IsStream t, Monad m, Functor (t m)) => t m (Either a b) -> t m a
1196lefts = fmap (fromLeft undefined) . filter isLeft
1198-- | Discard 'Left's and unwrap 'Right's in an 'Either' stream.
1202rights :: (IsStream t, Monad m, Functor (t m)) => t m (Either a b) -> t m b
1203rights = fmap (fromRight undefined) . filter isRight
1205------------------------------------------------------------------------------
1206-- Concurrent Application
1207------------------------------------------------------------------------------
1209-- | Parallel transform application operator; applies a stream transformation
1210-- function @t m a -> t m b@ to a stream @t m a@ concurrently; the input stream
1211-- is evaluated asynchronously in an independent thread yielding elements to a
1212-- buffer and the transformation function runs in another thread consuming the
1213-- input from the buffer. '|$' is just like regular function application
1214-- operator '$' except that it is concurrent.
1216-- If you read the signature as @(t m a -> t m b) -> (t m a -> t m b)@ you can
1217-- look at it as a transformation that converts a transform function to a
1218-- buffered concurrent transform function.
1220-- The following code prints a value every second even though each stage adds a
1226-- S.mapM (\\x -> threadDelay 1000000 >> print x)
1227-- |$ S.repeatM (threadDelay 1000000 >> return 1)
1232-- /Since: 0.3.0 ("Streamly")/
1236(|$) :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> (t m a -> t m b)
1237-- (|$) f = f . Async.mkAsync
1238(|$) f = f . Par.mkParallel
1246{-# INLINE applyAsync #-}
1247applyAsync :: (IsStream t, MonadAsync m)
1248 => (t m a -> t m b) -> (t m a -> t m b)
1251-- | Parallel reverse function application operator for streams; just like the
1252-- regular reverse function application operator '&' except that it is
1257-- S.repeatM (threadDelay 1000000 >> return 1)
1258-- |& S.mapM (\\x -> threadDelay 1000000 >> print x)
1263-- /Since: 0.3.0 ("Streamly")/
1267(|&) :: (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