2-- Module : Streamly.Internal.Data.Stream.IsStream.Expand
3-- Copyright : (c) 2017 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
9-- Expand a stream by combining two or more streams or by combining streams
12module Streamly.Internal.Data.Stream.IsStream.Expand
14 -- * Binary Combinators (Linear)
15 -- | Functions ending in the shape:
17 -- @t m a -> t m a -> t m a@.
19 -- The functions in this section have a linear or flat n-ary combining
20 -- characterstics. It means that when combined @n@ times (e.g. @a `serial`
21 -- b `serial` c ...@) the resulting expression will have an @O(n)@
22 -- complexity (instead O(n^2) for pair wise combinators described in the
23 -- next section. These functions can be used efficiently with
24 -- 'concatMapWith' et. al. combinators that combine streams in a linear
25 -- fashion (contrast with 'concatPairsWith' which combines streams as a
36 -- * Binary Combinators (Pair Wise)
37 -- | Like the functions in the section above these functions also combine
38 -- two streams into a single stream but when used @n@ times linearly they
39 -- exhibit O(n^2) complexity. They are best combined in a binary tree
40 -- fashion using 'concatPairsWith' giving a @n * log n@ complexity. Avoid
41 -- using these with 'concatMapWith' when combining a large or infinite
48 -- | 'wSerial' is a CPS based stream interleaving functions. Use
49 -- 'concatPairsWith wSerial' to interleave @n@ streams uniformly. It can be
50 -- used with 'concatMapWith' as well, however, the interleaving behavior of
51 -- @n@ streams would be asymmetric giving exponentially more weightage to
52 -- streams that come earlier in the composition.
59 -- | 'interleave' is like 'wSerial' but using a direct style
60 -- implementation instead of CPS. It is faster than 'wSerial' due to stream
61 -- fusion but has worse efficiency when used with 'concatMapWith' for large
87 -- * Combine Streams and Unfolds
89 -- Expand a stream by repeatedly using an unfold and merging the resulting
90 -- streams. Functions generally ending in the shape:
92 -- @Unfold m a b -> t m a -> t m b@
94 -- ** Append Many (Unfold)
95 -- | Unfold and flatten streams.
97 , unfoldManyInterleave
98 , unfoldManyRoundRobin
101 -- | Insert effects between streams. Like unfoldMany but intersperses an
102 -- effect between the streams. A special case of gintercalate.
108 -- | Insert Streams between Streams.
109 -- Like unfoldMany but intersperses streams from another source between
110 -- the streams from the first source.
116 -- * Append Many (concatMap)
117 -- | Map and serially append streams. 'concatMapM' is a generalization of
118 -- the binary append operation to append many streams.
124 -- * Flatten Containers
125 -- | Flatten 'Foldable' containers using the binary stream merging
127 , IsStream.concatFoldableWith
128 , IsStream.concatMapFoldableWith
129 , IsStream.concatForFoldableWith
132 -- | Map and flatten a stream like 'concatMap' but using a custom binary
133 -- stream merging combinator instead of just appending the streams. The
134 -- merging occurs sequentially, it works efficiently for 'serial', 'async',
135 -- 'ahead' like merge operations where we consume one stream before the
136 -- next or in case of 'wAsync' or 'parallel' where we consume all streams
137 -- simultaneously anyway.
139 -- However, in cases where the merging consumes streams in a round robin
140 -- fashion, a pair wise merging using 'concatPairsWith' would be more
141 -- efficient. These cases include operations like 'mergeBy' or 'zipWith'.
143 , IsStream.concatMapWith
148 -- | See the notes about suitable merge functions in the 'concatMapWith'
153 -- | Map and flatten Trees of Streams
156 , iterateMapLeftsWith
166import Streamly.Internal.Control.Concurrent (MonadAsync)
167import Streamly.Internal.Data.Stream.Ahead (aheadK)
168import Streamly.Internal.Data.Stream.Async (asyncK, wAsyncK)
169import Streamly.Internal.Data.Stream.IsStream.Common
170 ( concatM, concatMapM, concatMap, smapM, fromPure, fromEffect, parallelFst
172import Streamly.Internal.Data.Stream.IsStream.Type
173 (IsStream(..), fromStreamD, toStreamD)
174import Streamly.Data.Unfold (Unfold)
176import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
177import qualified Streamly.Internal.Data.Stream.Parallel as Par
178import qualified Streamly.Internal.Data.Stream.Serial as Serial
179import qualified Streamly.Internal.Data.Stream.StreamD as D
180 (append, interleave, interleaveSuffix, interleaveInfix, interleaveMin
181 , roundRobin, mergeByM, unfoldMany, unfoldManyInterleave, intersperse
182 , unfoldManyRoundRobin, interpose, interposeSuffix, gintercalate
183 , gintercalateSuffix, intersperseSuffix)
184import qualified Streamly.Internal.Data.Stream.StreamK as K (mergeBy, mergeByM)
185import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
186 (interleave, serial, concatPairsWith)
187import qualified Streamly.Internal.Data.Stream.ZipAsync as ZipAsync
189import Prelude hiding (concat, concatMap, zipWith)
193-- >>> import Control.Concurrent (threadDelay)
194-- >>> import Data.IORef
195-- >>> import Prelude hiding (zipWith, concatMap, concat)
196-- >>> import qualified Streamly.Prelude as Stream
197-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream
198-- >>> import qualified Streamly.Data.Fold as Fold
199-- >>> import qualified Streamly.Internal.Data.Fold as Fold
200-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
201-- >>> import qualified Streamly.Internal.Data.Parser as Parser
202-- >>> import qualified Streamly.Data.Array.Unboxed as Array
205-- threadDelay (n * 1000000) -- sleep for n seconds
206-- putStrLn (show n ++ " sec") -- print "n sec"
211-------------------------------------------------------------------------------
213-------------------------------------------------------------------------------
215-- XXX Reconcile the names "serial" and "append".
217-- | Append the outputs of two streams, yielding all the elements from the
218-- first stream and then yielding all the elements from the second stream.
220-- IMPORTANT NOTE: This could be 100x faster than @serial/<>@ for appending a
221-- few (say 100) streams because it can fuse via stream fusion. However, it
222-- does not scale for a large number of streams (say 1000s) and becomes
223-- qudartically slow. Therefore use this for custom appending of a few streams
224-- but use 'concatMap' or 'concatMapWith serial' for appending @n@ streams or
225-- infinite containers of streams.
229append ::(IsStream t, Monad m) => t m b -> t m b -> t m b
230append m1 m2 = fromStreamD $ D.append (toStreamD m1) (toStreamD m2)
234-- | Appends two streams sequentially, yielding all elements from the first
235-- stream, and then all elements from the second stream.
237-- >>> import Streamly.Prelude (serial)
238-- >>> stream1 = Stream.fromList [1,2]
239-- >>> stream2 = Stream.fromList [3,4]
240-- >>> Stream.toList $ stream1 `serial` stream2
243-- This operation can be used to fold an infinite lazy container of streams.
245-- /Since: 0.2.0 ("Streamly")/
249serial :: IsStream t => t m a -> t m a -> t m a
250serial m1 m2 = fromStream $ K.serial (toStream m1) (toStream m2)
252-------------------------------------------------------------------------------
254-------------------------------------------------------------------------------
258-- XXX doc duplicated from Stream.Serial module.
260-- | Interleaves two streams, yielding one element from each stream
261-- alternately. When one stream stops the rest of the other stream is used in
264-- >>> import Streamly.Prelude (wSerial)
265-- >>> stream1 = Stream.fromList [1,2]
266-- >>> stream2 = Stream.fromList [3,4]
267-- >>> Stream.toList $ Stream.fromWSerial $ stream1 `wSerial` stream2
270-- Note, for singleton streams 'wSerial' and 'serial' are identical.
272-- Note that this operation cannot be used to fold a container of infinite
273-- streams but it can be used for very large streams as the state that it needs
274-- to maintain is proportional to the logarithm of the number of streams.
278-- /Since: 0.2.0 ("Streamly")/
282-- Note that evaluation of @a \`wSerial` b \`wSerial` c@ does not interleave
283-- @a@, @b@ and @c@ with equal priority. This expression is equivalent to @a
284-- \`wSerial` (b \`wSerial` c)@, therefore, it fairly interleaves @a@ with the
285-- result of @b \`wSerial` c@. For example, @Stream.fromList [1,2] \`wSerial`
286-- Stream.fromList [3,4] \`wSerial` Stream.fromList [5,6]@ would result in
287-- [1,3,2,5,4,6]. In other words, the leftmost stream gets the same scheduling
288-- priority as the rest of the streams taken together. The same is true for
289-- each subexpression on the right.
291{-# INLINE wSerial #-}
292wSerial :: IsStream t => t m a -> t m a -> t m a
293wSerial m1 m2 = fromStream $ K.interleave (toStream m1) (toStream m2)
295-- XXX Same as 'wSerial'. We should perhaps rename wSerial to interleave.
296-- XXX Document the interleaving behavior of side effects in all the
297-- interleaving combinators.
298-- XXX Write time-domain equivalents of these. In the time domain we can
299-- interleave two streams such that the value of second stream is always taken
300-- from its last value even if no new value is being yielded, like
301-- zipWithLatest. It would be something like interleaveWithLatest.
303-- | Interleaves the outputs of two streams, yielding elements from each stream
304-- alternately, starting from the first stream. If any of the streams finishes
305-- early the other stream continues alone until it too finishes.
307-- >>> :set -XOverloadedStrings
308-- >>> import Data.Functor.Identity (Identity)
309-- >>> Stream.interleave "ab" ",,,," :: Stream.SerialT Identity Char
312-- >>> Stream.interleave "abcd" ",," :: Stream.SerialT Identity Char
315-- 'interleave' is dual to 'interleaveMin', it can be called @interleaveMax@.
317-- Do not use at scale in concatMapWith.
320{-# INLINE interleave #-}
321interleave ::(IsStream t, Monad m) => t m b -> t m b -> t m b
322interleave m1 m2 = fromStreamD $ D.interleave (toStreamD m1) (toStreamD m2)
324-- | Interleaves the outputs of two streams, yielding elements from each stream
325-- alternately, starting from the first stream. As soon as the first stream
326-- finishes, the output stops, discarding the remaining part of the second
327-- stream. In this case, the last element in the resulting stream would be from
328-- the second stream. If the second stream finishes early then the first stream
329-- still continues to yield elements until it finishes.
331-- >>> :set -XOverloadedStrings
332-- >>> import Data.Functor.Identity (Identity)
333-- >>> Stream.interleaveSuffix "abc" ",,,," :: Stream.SerialT Identity Char
335-- >>> Stream.interleaveSuffix "abc" "," :: Stream.SerialT Identity Char
338-- 'interleaveSuffix' is a dual of 'interleaveInfix'.
340-- Do not use at scale in concatMapWith.
343{-# INLINE interleaveSuffix #-}
344interleaveSuffix ::(IsStream t, Monad m) => t m b -> t m b -> t m b
345interleaveSuffix m1 m2 =
346 fromStreamD $ D.interleaveSuffix (toStreamD m1) (toStreamD m2)
348-- | Interleaves the outputs of two streams, yielding elements from each stream
349-- alternately, starting from the first stream and ending at the first stream.
350-- If the second stream is longer than the first, elements from the second
351-- stream are infixed with elements from the first stream. If the first stream
352-- is longer then it continues yielding elements even after the second stream
355-- >>> :set -XOverloadedStrings
356-- >>> import Data.Functor.Identity (Identity)
357-- >>> Stream.interleaveInfix "abc" ",,,," :: Stream.SerialT Identity Char
359-- >>> Stream.interleaveInfix "abc" "," :: Stream.SerialT Identity Char
362-- 'interleaveInfix' is a dual of 'interleaveSuffix'.
364-- Do not use at scale in concatMapWith.
367{-# INLINE interleaveInfix #-}
368interleaveInfix ::(IsStream t, Monad m) => t m b -> t m b -> t m b
369interleaveInfix m1 m2 =
370 fromStreamD $ D.interleaveInfix (toStreamD m1) (toStreamD m2)
372-- | Interleaves the outputs of two streams, yielding elements from each stream
373-- alternately, starting from the first stream. The output stops as soon as any
374-- of the two streams finishes, discarding the remaining part of the other
375-- stream. The last element of the resulting stream would be from the longer
378-- >>> :set -XOverloadedStrings
379-- >>> import Data.Functor.Identity (Identity)
380-- >>> Stream.interleaveMin "ab" ",,,," :: Stream.SerialT Identity Char
382-- >>> Stream.interleaveMin "abcd" ",," :: Stream.SerialT Identity Char
385-- 'interleaveMin' is dual to 'interleave'.
387-- Do not use at scale in concatMapWith.
390{-# INLINE interleaveMin #-}
391interleaveMin ::(IsStream t, Monad m) => t m b -> t m b -> t m b
392interleaveMin m1 m2 = fromStreamD $ D.interleaveMin (toStreamD m1) (toStreamD m2)
394-------------------------------------------------------------------------------
396-------------------------------------------------------------------------------
398-- | Schedule the execution of two streams in a fair round-robin manner,
399-- executing each stream once, alternately. Execution of a stream may not
400-- necessarily result in an output, a stream may chose to @Skip@ producing an
401-- element until later giving the other stream a chance to run. Therefore, this
402-- combinator fairly interleaves the execution of two streams rather than
403-- fairly interleaving the output of the two streams. This can be useful in
404-- co-operative multitasking without using explicit threads. This can be used
405-- as an alternative to `async`.
407-- Do not use at scale in concatMapWith.
410{-# INLINE roundrobin #-}
411roundrobin ::(IsStream t, Monad m) => t m b -> t m b -> t m b
412roundrobin m1 m2 = fromStreamD $ D.roundRobin (toStreamD m1) (toStreamD m2)
416-- | Merges two streams, both the streams may be evaluated concurrently,
417-- outputs from both are used as they arrive:
419-- >>> import Streamly.Prelude (async)
420-- >>> stream1 = Stream.fromEffect (delay 4)
421-- >>> stream2 = Stream.fromEffect (delay 2)
422-- >>> Stream.toList $ stream1 `async` stream2
427-- Multiple streams can be combined. With enough threads, all of them can be
428-- scheduled simultaneously:
430-- >>> stream3 = Stream.fromEffect (delay 1)
431-- >>> Stream.toList $ stream1 `async` stream2 `async` stream3
435-- With 2 threads, only two can be scheduled at a time, when one of those
436-- finishes, the third one gets scheduled:
438-- >>> Stream.toList $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3
442-- With a single thread, it becomes serial:
444-- >>> Stream.toList $ Stream.maxThreads 1 $ stream1 `async` stream2 `async` stream3
448-- Only streams are scheduled for async evaluation, how actions within a
449-- stream are evaluated depends on the stream type. If it is a concurrent
450-- stream they will be evaluated concurrently.
452-- In the following example, both the streams are scheduled for concurrent
453-- evaluation but each individual stream is evaluated serially:
455-- >>> stream1 = Stream.fromListM $ Prelude.map delay [3,3] -- SerialT IO Int
456-- >>> stream2 = Stream.fromListM $ Prelude.map delay [1,1] -- SerialT IO Int
457-- >>> Stream.toList $ stream1 `async` stream2 -- IO [Int]
461-- If total threads are 2, the third stream is scheduled only after one of the
462-- first two has finished:
464-- > stream3 = Stream.fromListM $ Prelude.map delay [2,2] -- SerialT IO Int
465-- > Stream.toList $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3 -- IO [Int]
469-- Thus 'async' goes deep in first few streams rather than going wide in all
470-- streams. It prefers to evaluate the leftmost streams as much as possible.
471-- Because of this behavior, 'async' can be safely used to fold an infinite
472-- lazy container of streams.
474-- /Since: 0.2.0 ("Streamly")/
478async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
479async m1 m2 = fromStream $ asyncK (toStream m1) (toStream m2)
483-- | For singleton streams, 'wAsync' is the same as 'async'. See 'async' for
484-- singleton stream behavior. For multi-element streams, while 'async' is left
485-- biased i.e. it tries to evaluate the left side stream as much as possible,
486-- 'wAsync' tries to schedule them both fairly. In other words, 'async' goes
487-- deep while 'wAsync' goes wide. However, outputs are always used as they
490-- With a single thread, 'async' starts behaving like 'serial' while 'wAsync'
491-- starts behaving like 'wSerial'.
493-- >>> import Streamly.Prelude (async, wAsync)
494-- >>> stream1 = Stream.fromList [1,2,3]
495-- >>> stream2 = Stream.fromList [4,5,6]
496-- >>> Stream.toList $ Stream.fromAsync $ Stream.maxThreads 1 $ stream1 `async` stream2
499-- >>> Stream.toList $ Stream.fromWAsync $ Stream.maxThreads 1 $ stream1 `wAsync` stream2
502-- With two threads available, and combining three streams:
504-- >>> stream3 = Stream.fromList [7,8,9]
505-- >>> Stream.toList $ Stream.fromAsync $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3
506-- [1,2,3,4,5,6,7,8,9]
508-- >>> Stream.toList $ Stream.fromWAsync $ Stream.maxThreads 2 $ stream1 `wAsync` stream2 `wAsync` stream3
509-- [1,4,2,7,5,3,8,6,9]
511-- This operation cannot be used to fold an infinite lazy container of streams,
512-- because it schedules all the streams in a round robin manner.
514-- Note that 'WSerialT' and single threaded 'WAsyncT' both interleave streams
515-- but the exact scheduling is slightly different in both cases.
519-- /Since: 0.2.0 ("Streamly")/
521-- Scheduling details:
523-- This is how the execution of the above example proceeds:
525-- 1. The scheduler queue is initialized with @[S.fromList [1,2,3],
526-- (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])]@ assuming the head of the
527-- queue is represented by the rightmost item.
528-- 2. @S.fromList [1,2,3]@ is executed, yielding the element @1@ and putting
529-- @[2,3]@ at the back of the scheduler queue. The scheduler queue now looks
530-- like @[(S.fromList [4,5,6]) \<> (S.fromList [7,8,9]), S.fromList [2,3]]@.
531-- 3. Now @(S.fromList [4,5,6]) \<> (S.fromList [7,8,9])@ is picked up for
532-- execution, @S.fromList [7,8,9]@ is added at the back of the queue and
533-- @S.fromList [4,5,6]@ is executed, yielding the element @4@ and adding
534-- @S.fromList [5,6]@ at the back of the queue. The queue now looks like
535-- @[S.fromList [2,3], S.fromList [7,8,9], S.fromList [5,6]]@.
536-- 4. Note that the scheduler queue expands by one more stream component in
537-- every pass because one more @<>@ is broken down into two components. At this
538-- point there are no more @<>@ operations to be broken down further and the
539-- queue has reached its maximum size. Now these streams are scheduled in
540-- round-robin fashion yielding @[2,7,5,3,8,6,9]@.
542-- As we see above, in a right associated expression composed with @<>@, only
543-- one @<>@ operation is broken down into two components in one execution,
544-- therefore, if we have @n@ streams composed using @<>@ it will take @n@
545-- scheduler passes to expand the whole expression. By the time @n-th@
546-- component is added to the scheduler queue, the first component would have
547-- received @n@ scheduler passes.
550wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
551wAsync m1 m2 = fromStream $ wAsyncK (toStream m1) (toStream m2)
555-- | Appends two streams, both the streams may be evaluated concurrently but
556-- the outputs are used in the same order as the corresponding actions in the
557-- original streams, side effects will happen in the order in which the streams
560-- >>> import Streamly.Prelude (ahead, SerialT)
561-- >>> stream1 = Stream.fromEffect (delay 4) :: SerialT IO Int
562-- >>> stream2 = Stream.fromEffect (delay 2) :: SerialT IO Int
563-- >>> Stream.toList $ stream1 `ahead` stream2 :: IO [Int]
568-- Multiple streams can be combined. With enough threads, all of them can be
569-- scheduled simultaneously:
571-- >>> stream3 = Stream.fromEffect (delay 1)
572-- >>> Stream.toList $ stream1 `ahead` stream2 `ahead` stream3
578-- With 2 threads, only two can be scheduled at a time, when one of those
579-- finishes, the third one gets scheduled:
581-- >>> Stream.toList $ Stream.maxThreads 2 $ stream1 `ahead` stream2 `ahead` stream3
587-- Only streams are scheduled for ahead evaluation, how actions within a stream
588-- are evaluated depends on the stream type. If it is a concurrent stream they
589-- will be evaluated concurrently. It may not make much sense combining serial
590-- streams using 'ahead'.
592-- 'ahead' can be safely used to fold an infinite lazy container of streams.
594-- /Since: 0.3.0 ("Streamly")/
598ahead :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
599ahead m1 m2 = fromStream $ aheadK (toStream m1) (toStream m2)
603-- | Like 'Streamly.Prelude.async' except that the execution is much more
604-- strict. There is no limit on the number of threads. While
605-- 'Streamly.Prelude.async' may not schedule a stream if there is no demand
606-- from the consumer, 'parallel' always evaluates both the streams immediately.
607-- The only limit that applies to 'parallel' is 'Streamly.Prelude.maxBuffer'.
608-- Evaluation may block if the output buffer becomes full.
610-- >>> import Streamly.Prelude (parallel)
611-- >>> stream = Stream.fromEffect (delay 2) `parallel` Stream.fromEffect (delay 1)
612-- >>> Stream.toList stream -- IO [Int]
617-- 'parallel' guarantees that all the streams are scheduled for execution
618-- immediately, therefore, we could use things like starting timers inside the
619-- streams and relying on the fact that all timers were started at the same
622-- Unlike 'async' this operation cannot be used to fold an infinite lazy
623-- container of streams, because it schedules all the streams strictly
626-- /Since: 0.2.0 ("Streamly")/
629{-# INLINE parallel #-}
630parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
631parallel m1 m2 = fromStream $ Par.parallelK (toStream m1) (toStream m2)
633-- This is a race like combinator for streams.
635-- | Like `parallel` but stops the output as soon as any of the two streams
639{-# INLINE parallelMin #-}
640parallelMin :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
641parallelMin m1 m2 = fromStream $ Par.parallelMinK (toStream m1) (toStream m2)
643------------------------------------------------------------------------------
645------------------------------------------------------------------------------
647-- | Like 'zipAsyncWith' but with a monadic zipping function.
650{-# INLINE zipAsyncWithM #-}
651zipAsyncWithM :: (IsStream t, MonadAsync m)
652 => (a -> b -> m c) -> t m a -> t m b -> t m c
653zipAsyncWithM f m1 m2 =
654 fromStream $ ZipAsync.zipAsyncWithMK f (toStream m1) (toStream m2)
656-- XXX Should we rename this to zipParWith or zipParallelWith? This can happen
657-- along with the change of behvaior to end the stream concurrently.
659-- | Like 'zipWith' but zips concurrently i.e. both the streams being zipped
660-- are evaluated concurrently using the 'ParallelT' concurrent evaluation
661-- style. The maximum number of elements of each stream evaluated in advance
662-- can be controlled by 'maxBuffer'.
664-- The stream ends if stream @a@ or stream @b@ ends. However, if stream @b@
665-- ends while we are still evaluating stream @a@ and waiting for a result then
666-- stream will not end until after the evaluation of stream @a@ finishes. This
667-- behavior can potentially be changed in future to end the stream immediately
668-- as soon as any of the stream end is detected.
671{-# INLINE zipAsyncWith #-}
672zipAsyncWith :: (IsStream t, MonadAsync m)
673 => (a -> b -> c) -> t m a -> t m b -> t m c
674zipAsyncWith f = zipAsyncWithM (\a b -> return (f a b))
676------------------------------------------------------------------------------
677-- Merging (sorted streams)
678------------------------------------------------------------------------------
680-- | Merge two streams using a comparison function. The head elements of both
681-- the streams are compared and the smaller of the two elements is emitted, if
682-- both elements are equal then the element from the first stream is used
685-- If the streams are sorted in ascending order, the resulting stream would
686-- also remain sorted in ascending order.
689-- >>> Stream.toList $ Stream.mergeBy compare (Stream.fromList [1,3,5]) (Stream.fromList [2,4,6,8])
694-- See also: 'mergeByMFused'
697{-# INLINE mergeBy #-}
698mergeBy :: IsStream t => (a -> a -> Ordering) -> t m a -> t m a -> t m a
699mergeBy f m1 m2 = fromStream $ K.mergeBy f (toStream m1) (toStream m2)
701-- | Like 'mergeBy' but with a monadic comparison function.
703-- Merge two streams randomly:
706-- > randomly _ _ = randomIO >>= \x -> return $ if x then LT else GT
707-- > Stream.toList $ Stream.mergeByM randomly (Stream.fromList [1,1,1,1]) (Stream.fromList [2,2,2,2])
711-- Merge two streams in a proportion of 2:1:
716-- let proportionately m n = do
717-- ref <- newIORef $ cycle $ Prelude.concat [Prelude.replicate m LT, Prelude.replicate n GT]
718-- return $ \_ _ -> do
720-- writeIORef ref $ Prelude.tail r
721-- return $ Prelude.head r
722-- f <- proportionately 2 1
723-- xs <- Stream.toList $ Stream.mergeByM f (Stream.fromList [1,1,1,1,1,1]) (Stream.fromList [2,2,2])
726-- [1,1,2,1,1,2,1,1,2]
730-- See also: 'mergeByMFused'
733{-# INLINE mergeByM #-}
735 :: (IsStream t, Monad m)
736 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
737mergeByM f m1 m2 = fromStream $ K.mergeByM f (toStream m1) (toStream m2)
739-- XXX Fused versions should probably go to a separate module using the same
740-- names for the combinators.
742-- | Like 'mergeByM' but much faster, works best when merging statically known
743-- number of streams. When merging more than two streams try to merge pairs and
744-- pair pf pairs in a tree like structure.'mergeByM' works better with variable
745-- number of streams being merged using 'concatPairsWith'.
748{-# INLINE mergeByMFused #-}
750 :: (IsStream t, Monad m)
751 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
752mergeByMFused f m1 m2 =
753 fromStreamD $ D.mergeByM f (toStreamD m1) (toStreamD m2)
755-- | Like 'mergeByM' but stops merging as soon as any of the two streams stops.
758{-# INLINABLE mergeMinBy #-}
759mergeMinBy :: -- (IsStream t, Monad m) =>
760 (a -> a -> m Ordering) -> t m a -> t m a -> t m a
761mergeMinBy _f _m1 _m2 = undefined
762 -- fromStreamD $ D.mergeMinBy f (toStreamD m1) (toStreamD m2)
764-- | Like 'mergeByM' but stops merging as soon as the first stream stops.
767{-# INLINABLE mergeFstBy #-}
768mergeFstBy :: -- (IsStream t, Monad m) =>
769 (a -> a -> m Ordering) -> t m a -> t m a -> t m a
770mergeFstBy _f _m1 _m2 = undefined
771 -- fromStreamD $ D.mergeFstBy f (toStreamD m1) (toStreamD m2)
773-- XXX we may want to use the name "merge" differently
774-- | Same as @'mergeBy' 'compare'@.
776-- >>> Stream.toList $ Stream.merge (Stream.fromList [1,3,5]) (Stream.fromList [2,4,6,8])
781{-# INLINABLE merge #-}
783 (IsStream t, Ord a) => t m a -> t m a -> t m a
784merge = mergeBy compare
786-- | Like 'mergeBy' but merges concurrently (i.e. both the elements being
787-- merged are generated concurrently).
790{-# INLINE mergeAsyncBy #-}
791mergeAsyncBy :: (IsStream t, MonadAsync m)
792 => (a -> a -> Ordering) -> t m a -> t m a -> t m a
793mergeAsyncBy f = mergeAsyncByM (\a b -> return $ f a b)
795-- | Like 'mergeByM' but merges concurrently (i.e. both the elements being
796-- merged are generated concurrently).
799{-# INLINE mergeAsyncByM #-}
800mergeAsyncByM :: (IsStream t, MonadAsync m)
801 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
802mergeAsyncByM f m1 m2 =
804 let par = Par.mkParallelD . toStreamD
805 in D.mergeByM f (par m1) (par m2)
809{-# DEPRECATED concatUnfold "Please use unfoldMany instead." #-}
810{-# INLINE concatUnfold #-}
811concatUnfold ::(IsStream t, Monad m) => Unfold m a b -> t m a -> t m b
812concatUnfold u m = fromStreamD $ D.unfoldMany u (toStreamD m)
814------------------------------------------------------------------------------
815-- Combine N Streams - unfoldMany
816------------------------------------------------------------------------------
818-- | Like 'concatMap' but uses an 'Unfold' for stream generation. Unlike
819-- 'concatMap' this can fuse the 'Unfold' code with the inner loop and
820-- therefore provide many times better performance.
823{-# INLINE unfoldMany #-}
824unfoldMany ::(IsStream t, Monad m) => Unfold m a b -> t m a -> t m b
825unfoldMany u m = fromStreamD $ D.unfoldMany u (toStreamD m)
827-- | Like 'unfoldMany' but interleaves the streams in the same way as
828-- 'interleave' behaves instead of appending them.
831{-# INLINE unfoldManyInterleave #-}
832unfoldManyInterleave ::(IsStream t, Monad m)
833 => Unfold m a b -> t m a -> t m b
834unfoldManyInterleave u m =
835 fromStreamD $ D.unfoldManyInterleave u (toStreamD m)
837-- | Like 'unfoldMany' but executes the streams in the same way as
841{-# INLINE unfoldManyRoundRobin #-}
842unfoldManyRoundRobin ::(IsStream t, Monad m)
843 => Unfold m a b -> t m a -> t m b
844unfoldManyRoundRobin u m =
845 fromStreamD $ D.unfoldManyRoundRobin u (toStreamD m)
847------------------------------------------------------------------------------
848-- Combine N Streams - interpose
849------------------------------------------------------------------------------
851-- > interpose x unf str = gintercalate unf str UF.identity (repeat x)
853-- | Unfold the elements of a stream, intersperse the given element between the
854-- unfolded streams and then concat them into a single stream.
856-- > unwords = S.interpose ' '
859{-# INLINE interpose #-}
860interpose :: (IsStream t, Monad m)
861 => c -> Unfold m b c -> t m b -> t m c
863 fromStreamD $ D.interpose (return x) unf (toStreamD str)
865-- interposeSuffix x unf str = gintercalateSuffix unf str UF.identity (repeat x)
867-- | Unfold the elements of a stream, append the given element after each
868-- unfolded stream and then concat them into a single stream.
870-- > unlines = S.interposeSuffix '\n'
873{-# INLINE interposeSuffix #-}
874interposeSuffix :: (IsStream t, Monad m)
875 => c -> Unfold m b c -> t m b -> t m c
876interposeSuffix x unf str =
877 fromStreamD $ D.interposeSuffix (return x) unf (toStreamD str)
879------------------------------------------------------------------------------
880-- Combine N Streams - intercalate
881------------------------------------------------------------------------------
883-- XXX we can swap the order of arguments to gintercalate so that the
884-- definition of unfoldMany becomes simpler? The first stream should be
885-- infixed inside the second one. However, if we change the order in
886-- "interleave" as well similarly, then that will make it a bit unintuitive.
888-- > unfoldMany unf str =
889-- > gintercalate unf str (UF.nilM (\_ -> return ())) (repeat ())
891-- | 'interleaveInfix' followed by unfold and concat.
894{-# INLINE gintercalate #-}
896 :: (IsStream t, Monad m)
897 => Unfold m a c -> t m a -> Unfold m b c -> t m b -> t m c
898gintercalate unf1 str1 unf2 str2 =
899 fromStreamD $ D.gintercalate
900 unf1 (toStreamD str1)
901 unf2 (toStreamD str2)
903-- > intercalate unf seed str = gintercalate unf str unf (repeatM seed)
905-- | 'intersperse' followed by unfold and concat.
907-- > intercalate unf a str = unfoldMany unf $ intersperse a str
908-- > intersperse = intercalate (Unfold.function id)
909-- > unwords = intercalate Unfold.fromList " "
911-- >>> Stream.toList $ Stream.intercalate Unfold.fromList " " $ Stream.fromList ["abc", "def", "ghi"]
915{-# INLINE intercalate #-}
916intercalate :: (IsStream t, Monad m)
917 => Unfold m b c -> b -> t m b -> t m c
918intercalate unf seed str = fromStreamD $
919 D.unfoldMany unf $ D.intersperse seed (toStreamD str)
921-- | 'interleaveSuffix' followed by unfold and concat.
924{-# INLINE gintercalateSuffix #-}
926 :: (IsStream t, Monad m)
927 => Unfold m a c -> t m a -> Unfold m b c -> t m b -> t m c
928gintercalateSuffix unf1 str1 unf2 str2 =
929 fromStreamD $ D.gintercalateSuffix
930 unf1 (toStreamD str1)
931 unf2 (toStreamD str2)
933-- > intercalateSuffix unf seed str = gintercalateSuffix unf str unf (repeatM seed)
935-- | 'intersperseSuffix' followed by unfold and concat.
937-- > intercalateSuffix unf a str = unfoldMany unf $ intersperseSuffix a str
938-- > intersperseSuffix = intercalateSuffix (Unfold.function id)
939-- > unlines = intercalateSuffix Unfold.fromList "\n"
941-- >>> Stream.toList $ Stream.intercalateSuffix Unfold.fromList "\n" $ Stream.fromList ["abc", "def", "ghi"]
945{-# INLINE intercalateSuffix #-}
946intercalateSuffix :: (IsStream t, Monad m)
947 => Unfold m b c -> b -> t m b -> t m c
948intercalateSuffix unf seed str = fromStreamD $ D.unfoldMany unf
949 $ D.intersperseSuffix (return seed) (toStreamD str)
951------------------------------------------------------------------------------
952-- Combine N Streams - concatMap
953------------------------------------------------------------------------------
955-- | Flatten a stream of streams to a single stream.
958-- concat = concatMap id
963concat :: (IsStream t, Monad m) => t m (t m a) -> t m a
966------------------------------------------------------------------------------
967-- Combine N Streams - concatMap
968------------------------------------------------------------------------------
970-- | Like 'concatMapWith' but carries a state which can be used to share
971-- information across multiple steps of concat.
974-- concatSmapMWith combine f initial = concatMapWith combine id . smapM f initial
979{-# INLINE concatSmapMWith #-}
981 :: (IsStream t, Monad m)
982 => (t m b -> t m b -> t m b)
983 -> (s -> a -> m (s, t m b))
987concatSmapMWith combine f initial =
988 IsStream.concatMapWith combine id . smapM f initial
990-- XXX Implement a StreamD version for fusion.
992-- | Combine streams in pairs using a binary stream combinator, then combine
993-- the resulting streams in pairs recursively until we get to a single combined
996-- For example, you can sort a stream using merge sort like this:
998-- >>> Stream.toList $ Stream.concatPairsWith (Stream.mergeBy compare) Stream.fromPure $ Stream.fromList [5,1,7,9,2]
1001-- /Caution: the stream of streams must be finite/
1005{-# INLINE concatPairsWith #-}
1006concatPairsWith :: IsStream t =>
1007 (t m b -> t m b -> t m b)
1011concatPairsWith par f m =
1014 (\s1 s2 -> toStream $ fromStream s1 `par` fromStream s2)
1018------------------------------------------------------------------------------
1019-- IterateMap - Map and flatten Trees of Streams
1020------------------------------------------------------------------------------
1022-- | Like 'iterateM' but iterates after mapping a stream generator on the
1025-- Yield an input element in the output stream, map a stream generator on it
1026-- and then do the same on the resulting stream. This can be used for a depth
1027-- first traversal of a tree like structure.
1029-- Note that 'iterateM' is a special case of 'iterateMapWith':
1032-- iterateM f = iterateMapWith serial (fromEffect . f) . fromEffect
1035-- It can be used to traverse a tree structure. For example, to list a
1039-- Stream.iterateMapWith Stream.serial
1040-- (either Dir.toEither (const nil))
1041-- (fromPure (Left "tmp"))
1046{-# INLINE iterateMapWith #-}
1049 => (t m a -> t m a -> t m a)
1053iterateMapWith combine f = IsStream.concatMapWith combine go
1055 go x = fromPure x `combine` IsStream.concatMapWith combine go (f x)
1057-- | Same as @iterateMapWith Stream.serial@ but more efficient due to stream
1061{-# INLINE iterateUnfold #-}
1062iterateUnfold :: -- (IsStream t, MonadAsync m) =>
1063 Unfold m a a -> t m a -> t m a
1064iterateUnfold = undefined
1066------------------------------------------------------------------------------
1068------------------------------------------------------------------------------
1070-- To traverse graphs we need a state to be carried around in the traversal.
1071-- For example, we can use a hashmap to store the visited status of nodes.
1073-- | Like 'iterateMap' but carries a state in the stream generation function.
1074-- This can be used to traverse graph like structures, we can remember the
1075-- visited nodes in the state to avoid cycles.
1077-- Note that a combination of 'iterateMap' and 'usingState' can also be used to
1078-- traverse graphs. However, this function provides a more localized state
1079-- instead of using a global state.
1085{-# INLINE iterateSmapMWith #-}
1087 :: (IsStream t, Monad m)
1088 => (t m a -> t m a -> t m a)
1089 -> (b -> a -> m (b, t m a))
1093iterateSmapMWith combine f initial stream =
1095 (\b -> IsStream.concatMapWith combine (go b) stream)
1096 (fromEffect initial)
1100 go b a = fromPure a `combine` feedback b a
1104 (\(b1, s) -> IsStream.concatMapWith combine (go b1) s)
1105 (fromEffect $ f b a)
1107------------------------------------------------------------------------------
1109------------------------------------------------------------------------------
1111-- Keep concating either streams as long as rights are generated, stop as soon
1112-- as a left is generated and concat the left stream.
1114-- See also: 'handle'
1120 :: -- (IsStream t, MonadAsync m) =>
1121 (forall x. t m x -> t m x -> t m x)
1122 -> (a -> t m (Either (t m b) b))
1125concatMapEitherWith = undefined
1128-- | In an 'Either' stream iterate on 'Left's. This is a special case of
1132-- iterateMapLeftsWith combine f = iterateMapWith combine (either f (const nil))
1135-- To traverse a directory tree:
1138-- iterateMapLeftsWith serial Dir.toEither (fromPure (Left "tmp"))
1143{-# INLINE iterateMapLeftsWith #-}
1145 :: (IsStream t, b ~ Either a c)
1146 => (t m b -> t m b -> t m b)
1150iterateMapLeftsWith combine f =
1151 iterateMapWith combine (either f (const IsStream.nil))
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