1{-# OPTIONS_GHC -Wno-redundant-constraints #-}
4-- Module : Streamly.Internal.Data.Stream.IsStream.Expand
5-- Copyright : (c) 2017 Composewell Technologies
6-- License : BSD-3-Clause
7-- Maintainer : streamly@composewell.com
8-- Stability : experimental
11-- Expand a stream by combining two or more streams or by combining streams
14module Streamly.Internal.Data.Stream.IsStream.Expand
16 -- * Binary Combinators (Linear)
17 -- | Functions ending in the shape:
19 -- @t m a -> t m a -> t m a@.
21 -- The functions in this section have a linear or flat n-ary combining
22 -- characterstics. It means that when combined @n@ times (e.g. @a `serial`
23 -- b `serial` c ...@) the resulting expression will have an @O(n)@
24 -- complexity (instead O(n^2) for pair wise combinators described in the
25 -- next section. These functions can be used efficiently with
26 -- 'concatMapWith' et. al. combinators that combine streams in a linear
27 -- fashion (contrast with 'concatPairsWith' which combines streams as a
38 -- * Binary Combinators (Pair Wise)
39 -- | Like the functions in the section above these functions also combine
40 -- two streams into a single stream but when used @n@ times linearly they
41 -- exhibit O(n^2) complexity. They are best combined in a binary tree
42 -- fashion using 'concatPairsWith' giving a @n * log n@ complexity. Avoid
43 -- using these with 'concatMapWith' when combining a large or infinite
50 -- | 'wSerial' is a CPS based stream interleaving functions. Use
51 -- 'concatPairsWith wSerial' to interleave @n@ streams uniformly. It can be
52 -- used with 'concatMapWith' as well, however, the interleaving behavior of
53 -- @n@ streams would be asymmetric giving exponentially more weightage to
54 -- streams that come earlier in the composition.
61 -- | 'interleave' is like 'wSerial' but using a direct style
62 -- implementation instead of CPS. It is faster than 'wSerial' due to stream
63 -- 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
167import Streamly.Internal.Control.Concurrent (MonadAsync)
168import Streamly.Internal.Data.Stream.Ahead (aheadK)
169import Streamly.Internal.Data.Stream.Async (asyncK, wAsyncK)
170import Streamly.Internal.Data.Stream.IsStream.Common
171 ( concatM, concatMapM, concatMap, smapM, fromPure, fromEffect, parallelFst
173import Streamly.Internal.Data.Stream.IsStream.Type
174 (IsStream(..), fromStreamD, toStreamD)
175import Streamly.Internal.Data.Unfold.Type (Unfold)
177import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
178import qualified Streamly.Internal.Data.Stream.Parallel as Par
179import qualified Streamly.Internal.Data.Stream.Serial as Serial
180import qualified Streamly.Internal.Data.Stream.StreamD as D
181import qualified Streamly.Internal.Data.Stream.StreamK as K (mergeBy, mergeByM)
182import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
183import qualified Streamly.Internal.Data.Stream.Zip as Zip
185import Prelude hiding (concat, concatMap, zipWith)
189-- >>> import Control.Concurrent (threadDelay)
190-- >>> import Data.IORef
191-- >>> import Prelude hiding (zipWith, concatMap, concat)
192-- >>> import qualified Streamly.Prelude as Stream
193-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream
194-- >>> import qualified Streamly.Data.Fold as Fold
195-- >>> import qualified Streamly.Internal.Data.Fold as Fold
196-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
197-- >>> import qualified Streamly.Internal.Data.Parser as Parser
198-- >>> import qualified Streamly.Data.Array.Foreign as Array
201-- threadDelay (n * 1000000) -- sleep for n seconds
202-- putStrLn (show n ++ " sec") -- print "n sec"
207-------------------------------------------------------------------------------
209-------------------------------------------------------------------------------
211-- XXX Reconcile the names "serial" and "append".
213-- | Append the outputs of two streams, yielding all the elements from the
214-- first stream and then yielding all the elements from the second stream.
216-- IMPORTANT NOTE: This could be 100x faster than @serial/<>@ for appending a
217-- few (say 100) streams because it can fuse via stream fusion. However, it
218-- does not scale for a large number of streams (say 1000s) and becomes
219-- qudartically slow. Therefore use this for custom appending of a few streams
220-- but use 'concatMap' or 'concatMapWith serial' for appending @n@ streams or
221-- infinite containers of streams.
225append ::(IsStream t, Monad m) => t m b -> t m b -> t m b
226append m1 m2 = fromStreamD $ D.append (toStreamD m1) (toStreamD m2)
230-- | Appends two streams sequentially, yielding all elements from the first
231-- stream, and then all elements from the second stream.
233-- >>> import Streamly.Prelude (serial)
234-- >>> stream1 = Stream.fromList [1,2]
235-- >>> stream2 = Stream.fromList [3,4]
236-- >>> Stream.toList $ stream1 `serial` stream2
239-- This operation can be used to fold an infinite lazy container of streams.
241-- /Since: 0.2.0 ("Streamly")/
245serial :: IsStream t => t m a -> t m a -> t m a
246serial m1 m2 = fromStream $ K.serial (toStream m1) (toStream m2)
248-------------------------------------------------------------------------------
250-------------------------------------------------------------------------------
254-- XXX doc duplicated from Stream.Serial module.
256-- | Interleaves two streams, yielding one element from each stream
257-- alternately. When one stream stops the rest of the other stream is used in
260-- >>> import Streamly.Prelude (wSerial)
261-- >>> stream1 = Stream.fromList [1,2]
262-- >>> stream2 = Stream.fromList [3,4]
263-- >>> Stream.toList $ Stream.fromWSerial $ stream1 `wSerial` stream2
266-- Note, for singleton streams 'wSerial' and 'serial' are identical.
268-- Note that this operation cannot be used to fold a container of infinite
269-- streams but it can be used for very large streams as the state that it needs
270-- to maintain is proportional to the logarithm of the number of streams.
274-- /Since: 0.2.0 ("Streamly")/
278-- Note that evaluation of @a \`wSerial` b \`wSerial` c@ does not interleave
279-- @a@, @b@ and @c@ with equal priority. This expression is equivalent to @a
280-- \`wSerial` (b \`wSerial` c)@, therefore, it fairly interleaves @a@ with the
281-- result of @b \`wSerial` c@. For example, @Stream.fromList [1,2] \`wSerial`
282-- Stream.fromList [3,4] \`wSerial` Stream.fromList [5,6]@ would result in
283-- [1,3,2,5,4,6]. In other words, the leftmost stream gets the same scheduling
284-- priority as the rest of the streams taken together. The same is true for
285-- each subexpression on the right.
287{-# INLINE wSerial #-}
288wSerial :: IsStream t => t m a -> t m a -> t m a
289wSerial m1 m2 = fromStream $ Serial.wSerialK (toStream m1) (toStream m2)
293-- | Same as 'wSerial'.
296{-# DEPRECATED (<=>) "Please use 'wSerial' instead." #-}
298(<=>) :: IsStream t => t m a -> t m a -> t m a
301-- XXX Same as 'wSerial'. We should perhaps rename wSerial to interleave.
302-- XXX Document the interleaving behavior of side effects in all the
303-- interleaving combinators.
304-- XXX Write time-domain equivalents of these. In the time domain we can
305-- interleave two streams such that the value of second stream is always taken
306-- from its last value even if no new value is being yielded, like
307-- zipWithLatest. It would be something like interleaveWithLatest.
309-- | Interleaves the outputs of two streams, yielding elements from each stream
310-- alternately, starting from the first stream. If any of the streams finishes
311-- early the other stream continues alone until it too finishes.
313-- >>> :set -XOverloadedStrings
314-- >>> import Data.Functor.Identity (Identity)
315-- >>> Stream.interleave "ab" ",,,," :: Stream.SerialT Identity Char
318-- >>> Stream.interleave "abcd" ",," :: Stream.SerialT Identity Char
321-- 'interleave' is dual to 'interleaveMin', it can be called @interleaveMax@.
323-- Do not use at scale in concatMapWith.
326{-# INLINE interleave #-}
327interleave ::(IsStream t, Monad m) => t m b -> t m b -> t m b
328interleave m1 m2 = fromStreamD $ D.interleave (toStreamD m1) (toStreamD m2)
330-- | Interleaves the outputs of two streams, yielding elements from each stream
331-- alternately, starting from the first stream. As soon as the first stream
332-- finishes, the output stops, discarding the remaining part of the second
333-- stream. In this case, the last element in the resulting stream would be from
334-- the second stream. If the second stream finishes early then the first stream
335-- still continues to yield elements until it finishes.
337-- >>> :set -XOverloadedStrings
338-- >>> import Data.Functor.Identity (Identity)
339-- >>> Stream.interleaveSuffix "abc" ",,,," :: Stream.SerialT Identity Char
341-- >>> Stream.interleaveSuffix "abc" "," :: Stream.SerialT Identity Char
344-- 'interleaveSuffix' is a dual of 'interleaveInfix'.
346-- Do not use at scale in concatMapWith.
349{-# INLINE interleaveSuffix #-}
350interleaveSuffix ::(IsStream t, Monad m) => t m b -> t m b -> t m b
351interleaveSuffix m1 m2 =
352 fromStreamD $ D.interleaveSuffix (toStreamD m1) (toStreamD m2)
354-- | Interleaves the outputs of two streams, yielding elements from each stream
355-- alternately, starting from the first stream and ending at the first stream.
356-- If the second stream is longer than the first, elements from the second
357-- stream are infixed with elements from the first stream. If the first stream
358-- is longer then it continues yielding elements even after the second stream
361-- >>> :set -XOverloadedStrings
362-- >>> import Data.Functor.Identity (Identity)
363-- >>> Stream.interleaveInfix "abc" ",,,," :: Stream.SerialT Identity Char
365-- >>> Stream.interleaveInfix "abc" "," :: Stream.SerialT Identity Char
368-- 'interleaveInfix' is a dual of 'interleaveSuffix'.
370-- Do not use at scale in concatMapWith.
373{-# INLINE interleaveInfix #-}
374interleaveInfix ::(IsStream t, Monad m) => t m b -> t m b -> t m b
375interleaveInfix m1 m2 =
376 fromStreamD $ D.interleaveInfix (toStreamD m1) (toStreamD m2)
378-- | Interleaves the outputs of two streams, yielding elements from each stream
379-- alternately, starting from the first stream. The output stops as soon as any
380-- of the two streams finishes, discarding the remaining part of the other
381-- stream. The last element of the resulting stream would be from the longer
384-- >>> :set -XOverloadedStrings
385-- >>> import Data.Functor.Identity (Identity)
386-- >>> Stream.interleaveMin "ab" ",,,," :: Stream.SerialT Identity Char
388-- >>> Stream.interleaveMin "abcd" ",," :: Stream.SerialT Identity Char
391-- 'interleaveMin' is dual to 'interleave'.
393-- Do not use at scale in concatMapWith.
396{-# INLINE interleaveMin #-}
397interleaveMin ::(IsStream t, Monad m) => t m b -> t m b -> t m b
398interleaveMin m1 m2 = fromStreamD $ D.interleaveMin (toStreamD m1) (toStreamD m2)
400-------------------------------------------------------------------------------
402-------------------------------------------------------------------------------
404-- | Schedule the execution of two streams in a fair round-robin manner,
405-- executing each stream once, alternately. Execution of a stream may not
406-- necessarily result in an output, a stream may chose to @Skip@ producing an
407-- element until later giving the other stream a chance to run. Therefore, this
408-- combinator fairly interleaves the execution of two streams rather than
409-- fairly interleaving the output of the two streams. This can be useful in
410-- co-operative multitasking without using explicit threads. This can be used
411-- as an alternative to `async`.
413-- Do not use at scale in concatMapWith.
416{-# INLINE roundrobin #-}
417roundrobin ::(IsStream t, Monad m) => t m b -> t m b -> t m b
418roundrobin m1 m2 = fromStreamD $ D.roundRobin (toStreamD m1) (toStreamD m2)
422-- | Merges two streams, both the streams may be evaluated concurrently,
423-- outputs from both are used as they arrive:
425-- >>> import Streamly.Prelude (async)
426-- >>> stream1 = Stream.fromEffect (delay 4)
427-- >>> stream2 = Stream.fromEffect (delay 2)
428-- >>> Stream.toList $ stream1 `async` stream2
433-- Multiple streams can be combined. With enough threads, all of them can be
434-- scheduled simultaneously:
436-- >>> stream3 = Stream.fromEffect (delay 1)
437-- >>> Stream.toList $ stream1 `async` stream2 `async` stream3
441-- With 2 threads, only two can be scheduled at a time, when one of those
442-- finishes, the third one gets scheduled:
444-- >>> Stream.toList $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3
448-- With a single thread, it becomes serial:
450-- >>> Stream.toList $ Stream.maxThreads 1 $ stream1 `async` stream2 `async` stream3
454-- Only streams are scheduled for async evaluation, how actions within a
455-- stream are evaluated depends on the stream type. If it is a concurrent
456-- stream they will be evaluated concurrently.
458-- In the following example, both the streams are scheduled for concurrent
459-- evaluation but each individual stream is evaluated serially:
461-- >>> stream1 = Stream.fromListM $ Prelude.map delay [3,3] -- SerialT IO Int
462-- >>> stream2 = Stream.fromListM $ Prelude.map delay [1,1] -- SerialT IO Int
463-- >>> Stream.toList $ stream1 `async` stream2 -- IO [Int]
467-- If total threads are 2, the third stream is scheduled only after one of the
468-- first two has finished:
470-- > stream3 = Stream.fromListM $ Prelude.map delay [2,2] -- SerialT IO Int
471-- > Stream.toList $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3 -- IO [Int]
475-- Thus 'async' goes deep in first few streams rather than going wide in all
476-- streams. It prefers to evaluate the leftmost streams as much as possible.
477-- Because of this behavior, 'async' can be safely used to fold an infinite
478-- lazy container of streams.
480-- /Since: 0.2.0 ("Streamly")/
484async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
485async m1 m2 = fromStream $ asyncK (toStream m1) (toStream m2)
490{-# DEPRECATED (<|) "Please use 'async' instead." #-}
492(<|) :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
497-- | For singleton streams, 'wAsync' is the same as 'async'. See 'async' for
498-- singleton stream behavior. For multi-element streams, while 'async' is left
499-- biased i.e. it tries to evaluate the left side stream as much as possible,
500-- 'wAsync' tries to schedule them both fairly. In other words, 'async' goes
501-- deep while 'wAsync' goes wide. However, outputs are always used as they
504-- With a single thread, 'async' starts behaving like 'serial' while 'wAsync'
505-- starts behaving like 'wSerial'.
507-- >>> import Streamly.Prelude (async, wAsync)
508-- >>> stream1 = Stream.fromList [1,2,3]
509-- >>> stream2 = Stream.fromList [4,5,6]
510-- >>> Stream.toList $ Stream.fromAsync $ Stream.maxThreads 1 $ stream1 `async` stream2
513-- >>> Stream.toList $ Stream.fromWAsync $ Stream.maxThreads 1 $ stream1 `wAsync` stream2
516-- With two threads available, and combining three streams:
518-- >>> stream3 = Stream.fromList [7,8,9]
519-- >>> Stream.toList $ Stream.fromAsync $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3
520-- [1,2,3,4,5,6,7,8,9]
522-- >>> Stream.toList $ Stream.fromWAsync $ Stream.maxThreads 2 $ stream1 `wAsync` stream2 `wAsync` stream3
523-- [1,4,2,7,5,3,8,6,9]
525-- This operation cannot be used to fold an infinite lazy container of streams,
526-- because it schedules all the streams in a round robin manner.
528-- Note that 'WSerialT' and single threaded 'WAsyncT' both interleave streams
529-- but the exact scheduling is slightly different in both cases.
533-- /Since: 0.2.0 ("Streamly")/
535-- Scheduling details:
537-- This is how the execution of the above example proceeds:
539-- 1. The scheduler queue is initialized with @[S.fromList [1,2,3],
540-- (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])]@ assuming the head of the
541-- queue is represented by the rightmost item.
542-- 2. @S.fromList [1,2,3]@ is executed, yielding the element @1@ and putting
543-- @[2,3]@ at the back of the scheduler queue. The scheduler queue now looks
544-- like @[(S.fromList [4,5,6]) \<> (S.fromList [7,8,9]), S.fromList [2,3]]@.
545-- 3. Now @(S.fromList [4,5,6]) \<> (S.fromList [7,8,9])@ is picked up for
546-- execution, @S.fromList [7,8,9]@ is added at the back of the queue and
547-- @S.fromList [4,5,6]@ is executed, yielding the element @4@ and adding
548-- @S.fromList [5,6]@ at the back of the queue. The queue now looks like
549-- @[S.fromList [2,3], S.fromList [7,8,9], S.fromList [5,6]]@.
550-- 4. Note that the scheduler queue expands by one more stream component in
551-- every pass because one more @<>@ is broken down into two components. At this
552-- point there are no more @<>@ operations to be broken down further and the
553-- queue has reached its maximum size. Now these streams are scheduled in
554-- round-robin fashion yielding @[2,7,5,3,8,6,9]@.
556-- As we see above, in a right associated expression composed with @<>@, only
557-- one @<>@ operation is broken down into two components in one execution,
558-- therefore, if we have @n@ streams composed using @<>@ it will take @n@
559-- scheduler passes to expand the whole expression. By the time @n-th@
560-- component is added to the scheduler queue, the first component would have
561-- received @n@ scheduler passes.
564wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
565wAsync m1 m2 = fromStream $ wAsyncK (toStream m1) (toStream m2)
569-- | Appends two streams, both the streams may be evaluated concurrently but
570-- the outputs are used in the same order as the corresponding actions in the
571-- original streams, side effects will happen in the order in which the streams
574-- >>> import Streamly.Prelude (ahead, SerialT)
575-- >>> stream1 = Stream.fromEffect (delay 4) :: SerialT IO Int
576-- >>> stream2 = Stream.fromEffect (delay 2) :: SerialT IO Int
577-- >>> Stream.toList $ stream1 `ahead` stream2 :: IO [Int]
582-- Multiple streams can be combined. With enough threads, all of them can be
583-- scheduled simultaneously:
585-- >>> stream3 = Stream.fromEffect (delay 1)
586-- >>> Stream.toList $ stream1 `ahead` stream2 `ahead` stream3
592-- With 2 threads, only two can be scheduled at a time, when one of those
593-- finishes, the third one gets scheduled:
595-- >>> Stream.toList $ Stream.maxThreads 2 $ stream1 `ahead` stream2 `ahead` stream3
601-- Only streams are scheduled for ahead evaluation, how actions within a stream
602-- are evaluated depends on the stream type. If it is a concurrent stream they
603-- will be evaluated concurrently. It may not make much sense combining serial
604-- streams using 'ahead'.
606-- 'ahead' can be safely used to fold an infinite lazy container of streams.
608-- /Since: 0.3.0 ("Streamly")/
612ahead :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
613ahead m1 m2 = fromStream $ aheadK (toStream m1) (toStream m2)
617-- | Like 'Streamly.Prelude.async' except that the execution is much more
618-- strict. There is no limit on the number of threads. While
619-- 'Streamly.Prelude.async' may not schedule a stream if there is no demand
620-- from the consumer, 'parallel' always evaluates both the streams immediately.
621-- The only limit that applies to 'parallel' is 'Streamly.Prelude.maxBuffer'.
622-- Evaluation may block if the output buffer becomes full.
624-- >>> import Streamly.Prelude (parallel)
625-- >>> stream = Stream.fromEffect (delay 2) `parallel` Stream.fromEffect (delay 1)
626-- >>> Stream.toList stream -- IO [Int]
631-- 'parallel' guarantees that all the streams are scheduled for execution
632-- immediately, therefore, we could use things like starting timers inside the
633-- streams and relying on the fact that all timers were started at the same
636-- Unlike 'async' this operation cannot be used to fold an infinite lazy
637-- container of streams, because it schedules all the streams strictly
640-- /Since: 0.2.0 ("Streamly")/
643{-# INLINE parallel #-}
644parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
645parallel m1 m2 = fromStream $ Par.parallelK (toStream m1) (toStream m2)
647-- This is a race like combinator for streams.
649-- | Like `parallel` but stops the output as soon as any of the two streams
653{-# INLINE parallelMin #-}
654parallelMin :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
655parallelMin m1 m2 = fromStream $ Par.parallelMinK (toStream m1) (toStream m2)
657------------------------------------------------------------------------------
659------------------------------------------------------------------------------
661-- | Like 'zipAsyncWith' but with a monadic zipping function.
664{-# INLINE zipAsyncWithM #-}
665zipAsyncWithM :: (IsStream t, MonadAsync m)
666 => (a -> b -> m c) -> t m a -> t m b -> t m c
667zipAsyncWithM f m1 m2 =
668 fromStream $ Zip.zipAsyncWithMK f (toStream m1) (toStream m2)
670-- XXX Should we rename this to zipParWith or zipParallelWith? This can happen
671-- along with the change of behvaior to end the stream concurrently.
673-- | Like 'zipWith' but zips concurrently i.e. both the streams being zipped
674-- are evaluated concurrently using the 'ParallelT' concurrent evaluation
675-- style. The maximum number of elements of each stream evaluated in advance
676-- can be controlled by 'maxBuffer'.
678-- The stream ends if stream @a@ or stream @b@ ends. However, if stream @b@
679-- ends while we are still evaluating stream @a@ and waiting for a result then
680-- stream will not end until after the evaluation of stream @a@ finishes. This
681-- behavior can potentially be changed in future to end the stream immediately
682-- as soon as any of the stream end is detected.
685{-# INLINE zipAsyncWith #-}
686zipAsyncWith :: (IsStream t, MonadAsync m)
687 => (a -> b -> c) -> t m a -> t m b -> t m c
688zipAsyncWith f = zipAsyncWithM (\a b -> return (f a b))
690------------------------------------------------------------------------------
691-- Merging (sorted streams)
692------------------------------------------------------------------------------
694-- | Merge two streams using a comparison function. The head elements of both
695-- the streams are compared and the smaller of the two elements is emitted, if
696-- both elements are equal then the element from the first stream is used
699-- If the streams are sorted in ascending order, the resulting stream would
700-- also remain sorted in ascending order.
703-- >>> Stream.toList $ Stream.mergeBy compare (Stream.fromList [1,3,5]) (Stream.fromList [2,4,6,8])
708-- See also: 'mergeByMFused'
711{-# INLINE mergeBy #-}
713 (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> t m a -> t m a
714mergeBy f m1 m2 = fromStream $ K.mergeBy f (toStream m1) (toStream m2)
716-- | Like 'mergeBy' but with a monadic comparison function.
718-- Merge two streams randomly:
721-- > randomly _ _ = randomIO >>= \x -> return $ if x then LT else GT
722-- > Stream.toList $ Stream.mergeByM randomly (Stream.fromList [1,1,1,1]) (Stream.fromList [2,2,2,2])
726-- Merge two streams in a proportion of 2:1:
731-- let proportionately m n = do
732-- ref <- newIORef $ cycle $ Prelude.concat [Prelude.replicate m LT, Prelude.replicate n GT]
733-- return $ \_ _ -> do
735-- writeIORef ref $ Prelude.tail r
736-- return $ Prelude.head r
737-- f <- proportionately 2 1
738-- xs <- Stream.toList $ Stream.mergeByM f (Stream.fromList [1,1,1,1,1,1]) (Stream.fromList [2,2,2])
741-- [1,1,2,1,1,2,1,1,2]
745-- See also: 'mergeByMFused'
748{-# INLINABLE mergeByM #-}
750 :: (IsStream t, Monad m)
751 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
752mergeByM f m1 m2 = fromStream $ K.mergeByM f (toStream m1) (toStream m2)
754-- XXX Fused versions should probably go to a separate module using the same
755-- names for the combinators.
757-- | Like 'mergeByM' but much faster, works best when merging statically known
758-- number of streams. When merging more than two streams try to merge pairs and
759-- pair pf pairs in a tree like structure.'mergeByM' works better with variable
760-- number of streams being merged using 'concatPairsWith'.
763{-# INLINE mergeByMFused #-}
765 :: (IsStream t, Monad m)
766 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
767mergeByMFused f m1 m2 =
768 fromStreamD $ D.mergeByM f (toStreamD m1) (toStreamD m2)
771-- | Like 'mergeByM' but stops merging as soon as any of the two streams stops.
772{-# INLINABLE mergeEndByAny #-}
774 :: (IsStream t, Monad m)
775 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
776mergeEndByAny f m1 m2 = fromStreamD $
777 D.mergeEndByAny f (toStreamD m1) (toStreamD m2)
779-- Like 'mergeByM' but stops merging as soon as the first stream stops.
780{-# INLINABLE mergeEndByFirst #-}
782 :: (IsStream t, Monad m)
783 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
784mergeEndByFirst f m1 m2 = fromStreamS $
785 D.mergeEndByFirst f (toStreamD m1) (toStreamD m2)
787-- Holding this back for now, we may want to use the name "merge" differently
788-- | Same as @'mergeBy' 'compare'@.
791-- >>> Stream.toList $ Stream.merge (Stream.fromList [1,3,5]) (Stream.fromList [2,4,6,8])
796{-# INLINABLE merge #-}
798 (IsStream t, Monad m, Ord a) => t m a -> t m a -> t m a
799merge = mergeBy compare
802-- | Like 'mergeBy' but merges concurrently (i.e. both the elements being
803-- merged are generated concurrently).
806{-# INLINE mergeAsyncBy #-}
807mergeAsyncBy :: (IsStream t, MonadAsync m)
808 => (a -> a -> Ordering) -> t m a -> t m a -> t m a
809mergeAsyncBy f = mergeAsyncByM (\a b -> return $ f a b)
811-- | Like 'mergeByM' but merges concurrently (i.e. both the elements being
812-- merged are generated concurrently).
815{-# INLINE mergeAsyncByM #-}
816mergeAsyncByM :: (IsStream t, MonadAsync m)
817 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
818mergeAsyncByM f m1 m2 =
820 let par = Par.mkParallelD . toStreamD
821 in D.mergeByM f (par m1) (par m2)
825{-# DEPRECATED concatUnfold "Please use unfoldMany instead." #-}
826{-# INLINE concatUnfold #-}
827concatUnfold ::(IsStream t, Monad m) => Unfold m a b -> t m a -> t m b
828concatUnfold u m = fromStreamD $ D.unfoldMany u (toStreamD m)
830------------------------------------------------------------------------------
831-- Combine N Streams - unfoldMany
832------------------------------------------------------------------------------
834-- | Like 'concatMap' but uses an 'Unfold' for stream generation. Unlike
835-- 'concatMap' this can fuse the 'Unfold' code with the inner loop and
836-- therefore provide many times better performance.
839{-# INLINE unfoldMany #-}
840unfoldMany ::(IsStream t, Monad m) => Unfold m a b -> t m a -> t m b
841unfoldMany u m = fromStreamD $ D.unfoldMany u (toStreamD m)
843-- | Like 'unfoldMany' but interleaves the streams in the same way as
844-- 'interleave' behaves instead of appending them.
847{-# INLINE unfoldManyInterleave #-}
848unfoldManyInterleave ::(IsStream t, Monad m)
849 => Unfold m a b -> t m a -> t m b
850unfoldManyInterleave u m =
851 fromStreamD $ D.unfoldManyInterleave u (toStreamD m)
853-- | Like 'unfoldMany' but executes the streams in the same way as
857{-# INLINE unfoldManyRoundRobin #-}
858unfoldManyRoundRobin ::(IsStream t, Monad m)
859 => Unfold m a b -> t m a -> t m b
860unfoldManyRoundRobin u m =
861 fromStreamD $ D.unfoldManyRoundRobin u (toStreamD m)
863------------------------------------------------------------------------------
864-- Combine N Streams - interpose
865------------------------------------------------------------------------------
867-- > interpose x unf str = gintercalate unf str UF.identity (repeat x)
869-- | Unfold the elements of a stream, intersperse the given element between the
870-- unfolded streams and then concat them into a single stream.
872-- > unwords = S.interpose ' '
875{-# INLINE interpose #-}
876interpose :: (IsStream t, Monad m)
877 => c -> Unfold m b c -> t m b -> t m c
879 fromStreamD $ D.interpose (return x) unf (toStreamD str)
881-- interposeSuffix x unf str = gintercalateSuffix unf str UF.identity (repeat x)
883-- | Unfold the elements of a stream, append the given element after each
884-- unfolded stream and then concat them into a single stream.
886-- > unlines = S.interposeSuffix '\n'
889{-# INLINE interposeSuffix #-}
890interposeSuffix :: (IsStream t, Monad m)
891 => c -> Unfold m b c -> t m b -> t m c
892interposeSuffix x unf str =
893 fromStreamD $ D.interposeSuffix (return x) unf (toStreamD str)
895------------------------------------------------------------------------------
896-- Combine N Streams - intercalate
897------------------------------------------------------------------------------
899-- XXX we can swap the order of arguments to gintercalate so that the
900-- definition of unfoldMany becomes simpler? The first stream should be
901-- infixed inside the second one. However, if we change the order in
902-- "interleave" as well similarly, then that will make it a bit unintuitive.
904-- > unfoldMany unf str =
905-- > gintercalate unf str (UF.nilM (\_ -> return ())) (repeat ())
907-- | 'interleaveInfix' followed by unfold and concat.
910{-# INLINE gintercalate #-}
912 :: (IsStream t, Monad m)
913 => Unfold m a c -> t m a -> Unfold m b c -> t m b -> t m c
914gintercalate unf1 str1 unf2 str2 =
915 fromStreamD $ D.gintercalate
916 unf1 (toStreamD str1)
917 unf2 (toStreamD str2)
919-- > intercalate unf seed str = gintercalate unf str unf (repeatM seed)
921-- | 'intersperse' followed by unfold and concat.
923-- > intercalate unf a str = unfoldMany unf $ intersperse a str
924-- > intersperse = intercalate (Unfold.function id)
925-- > unwords = intercalate Unfold.fromList " "
927-- >>> Stream.toList $ Stream.intercalate Unfold.fromList " " $ Stream.fromList ["abc", "def", "ghi"]
931{-# INLINE intercalate #-}
932intercalate :: (IsStream t, Monad m)
933 => Unfold m b c -> b -> t m b -> t m c
934intercalate unf seed str = fromStreamD $
935 D.unfoldMany unf $ D.intersperse seed (toStreamD str)
937-- | 'interleaveSuffix' followed by unfold and concat.
940{-# INLINE gintercalateSuffix #-}
942 :: (IsStream t, Monad m)
943 => Unfold m a c -> t m a -> Unfold m b c -> t m b -> t m c
944gintercalateSuffix unf1 str1 unf2 str2 =
945 fromStreamD $ D.gintercalateSuffix
946 unf1 (toStreamD str1)
947 unf2 (toStreamD str2)
949-- > intercalateSuffix unf seed str = gintercalateSuffix unf str unf (repeatM seed)
951-- | 'intersperseSuffix' followed by unfold and concat.
953-- > intercalateSuffix unf a str = unfoldMany unf $ intersperseSuffix a str
954-- > intersperseSuffix = intercalateSuffix (Unfold.function id)
955-- > unlines = intercalateSuffix Unfold.fromList "\n"
957-- >>> Stream.toList $ Stream.intercalateSuffix Unfold.fromList "\n" $ Stream.fromList ["abc", "def", "ghi"]
961{-# INLINE intercalateSuffix #-}
962intercalateSuffix :: (IsStream t, Monad m)
963 => Unfold m b c -> b -> t m b -> t m c
964intercalateSuffix unf seed str = fromStreamD $ D.unfoldMany unf
965 $ D.intersperseSuffix (return seed) (toStreamD str)
968{-# INLINE iterateUnfold #-}
969iterateUnfold :: (IsStream t, MonadAsync m)
970 => Unfold m a a -> t m a -> t m a
971iterateUnfold unf xs = undefined
974------------------------------------------------------------------------------
975-- Combine N Streams - concatMap
976------------------------------------------------------------------------------
978-- | Flatten a stream of streams to a single stream.
981-- concat = concatMap id
986concat :: (IsStream t, Monad m) => t m (t m a) -> t m a
989------------------------------------------------------------------------------
990-- Combine N Streams - concatMap
991------------------------------------------------------------------------------
993-- | Like 'concatMapWith' but carries a state which can be used to share
994-- information across multiple steps of concat.
997-- concatSmapMWith combine f initial = concatMapWith combine id . smapM f initial
1002{-# INLINE concatSmapMWith #-}
1004 :: (IsStream t, Monad m)
1005 => (t m b -> t m b -> t m b)
1006 -> (s -> a -> m (s, t m b))
1010concatSmapMWith combine f initial =
1011 IsStream.concatMapWith combine id . smapM f initial
1013-- Keep concating either streams as long as rights are generated, stop as soon
1014-- as a left is generated and concat the left stream.
1016-- See also: 'handle'
1022 :: -- (IsStream t, MonadAsync m) =>
1023 (forall x. t m x -> t m x -> t m x)
1024 -> (a -> t m (Either (t m b) b))
1027concatMapEitherWith = undefined
1030-- XXX Implement a StreamD version for fusion.
1032-- | Combine streams in pairs using a binary stream combinator, then combine
1033-- the resulting streams in pairs recursively until we get to a single combined
1036-- For example, you can sort a stream using merge sort like this:
1038-- >>> Stream.toList $ Stream.concatPairsWith (Stream.mergeBy compare) Stream.fromPure $ Stream.fromList [5,1,7,9,2]
1041-- /Caution: the stream of streams must be finite/
1045{-# INLINE concatPairsWith #-}
1046concatPairsWith :: IsStream t =>
1047 (t m b -> t m b -> t m b)
1051concatPairsWith par f m =
1054 (\s1 s2 -> toStream $ fromStream s1 `par` fromStream s2)
1058------------------------------------------------------------------------------
1059-- IterateMap - Map and flatten Trees of Streams
1060------------------------------------------------------------------------------
1062-- | Like 'iterateM' but iterates after mapping a stream generator on the
1065-- Yield an input element in the output stream, map a stream generator on it
1066-- and then do the same on the resulting stream. This can be used for a depth
1067-- first traversal of a tree like structure.
1069-- Note that 'iterateM' is a special case of 'iterateMapWith':
1072-- iterateM f = iterateMapWith serial (fromEffect . f) . fromEffect
1075-- It can be used to traverse a tree structure. For example, to list a
1079-- Stream.iterateMapWith Stream.serial
1080-- (either Dir.toEither (const nil))
1081-- (fromPure (Left "tmp"))
1086{-# INLINE iterateMapWith #-}
1089 => (t m a -> t m a -> t m a)
1093iterateMapWith combine f = IsStream.concatMapWith combine go
1095 go x = fromPure x `combine` IsStream.concatMapWith combine go (f x)
1098{-# INLINE iterateUnfold #-}
1099iterateUnfold :: (IsStream t, MonadAsync m)
1100 => Unfold m a a -> t m a -> t m a
1101iterateUnfold unf xs = undefined
1104------------------------------------------------------------------------------
1106------------------------------------------------------------------------------
1108-- To traverse graphs we need a state to be carried around in the traversal.
1109-- For example, we can use a hashmap to store the visited status of nodes.
1111-- | Like 'iterateMap' but carries a state in the stream generation function.
1112-- This can be used to traverse graph like structures, we can remember the
1113-- visited nodes in the state to avoid cycles.
1115-- Note that a combination of 'iterateMap' and 'usingState' can also be used to
1116-- traverse graphs. However, this function provides a more localized state
1117-- instead of using a global state.
1123{-# INLINE iterateSmapMWith #-}
1125 :: (IsStream t, Monad m)
1126 => (t m a -> t m a -> t m a)
1127 -> (b -> a -> m (b, t m a))
1131iterateSmapMWith combine f initial stream =
1133 (\b -> IsStream.concatMapWith combine (go b) stream)
1134 (fromEffect initial)
1138 go b a = fromPure a `combine` feedback b a
1142 (\(b1, s) -> IsStream.concatMapWith combine (go b1) s)
1143 (fromEffect $ f b a)
1145------------------------------------------------------------------------------
1146-- iterateMap - Either streams
1147------------------------------------------------------------------------------
1149-- | In an 'Either' stream iterate on 'Left's. This is a special case of
1153-- iterateMapLeftsWith combine f = iterateMapWith combine (either f (const nil))
1156-- To traverse a directory tree:
1159-- iterateMapLeftsWith serial Dir.toEither (fromPure (Left "tmp"))
1164{-# INLINE iterateMapLeftsWith #-}
1167 => (t m (Either a b) -> t m (Either a b) -> t m (Either a b))
1168 -> (a -> t m (Either a b))
1171iterateMapLeftsWith combine f =
1172 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