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.Internal.Data.Unfold.Type (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.WSerial as WSerial
179import qualified Streamly.Internal.Data.Stream.StreamD as D
180import qualified Streamly.Internal.Data.Stream.StreamK as K (mergeBy, mergeByM)
181import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
182import qualified Streamly.Internal.Data.Stream.ZipAsync as ZipAsync
184import Prelude hiding (concat, concatMap, zipWith)
188-- >>> import Control.Concurrent (threadDelay)
189-- >>> import Data.IORef
190-- >>> import Prelude hiding (zipWith, concatMap, concat)
191-- >>> import qualified Streamly.Prelude as Stream
192-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream
193-- >>> import qualified Streamly.Data.Fold as Fold
194-- >>> import qualified Streamly.Internal.Data.Fold as Fold
195-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
196-- >>> import qualified Streamly.Internal.Data.Parser as Parser
197-- >>> import qualified Streamly.Data.Array.Foreign as Array
200-- threadDelay (n * 1000000) -- sleep for n seconds
201-- putStrLn (show n ++ " sec") -- print "n sec"
206-------------------------------------------------------------------------------
208-------------------------------------------------------------------------------
210-- XXX Reconcile the names "serial" and "append".
212-- | Append the outputs of two streams, yielding all the elements from the
213-- first stream and then yielding all the elements from the second stream.
215-- IMPORTANT NOTE: This could be 100x faster than @serial/<>@ for appending a
216-- few (say 100) streams because it can fuse via stream fusion. However, it
217-- does not scale for a large number of streams (say 1000s) and becomes
218-- qudartically slow. Therefore use this for custom appending of a few streams
219-- but use 'concatMap' or 'concatMapWith serial' for appending @n@ streams or
220-- infinite containers of streams.
224append ::(IsStream t, Monad m) => t m b -> t m b -> t m b
225append m1 m2 = fromStreamD $ D.append (toStreamD m1) (toStreamD m2)
229-- | Appends two streams sequentially, yielding all elements from the first
230-- stream, and then all elements from the second stream.
232-- >>> import Streamly.Prelude (serial)
233-- >>> stream1 = Stream.fromList [1,2]
234-- >>> stream2 = Stream.fromList [3,4]
235-- >>> Stream.toList $ stream1 `serial` stream2
238-- This operation can be used to fold an infinite lazy container of streams.
240-- /Since: 0.2.0 ("Streamly")/
244serial :: IsStream t => t m a -> t m a -> t m a
245serial m1 m2 = fromStream $ K.serial (toStream m1) (toStream m2)
247-------------------------------------------------------------------------------
249-------------------------------------------------------------------------------
253-- XXX doc duplicated from Stream.Serial module.
255-- | Interleaves two streams, yielding one element from each stream
256-- alternately. When one stream stops the rest of the other stream is used in
259-- >>> import Streamly.Prelude (wSerial)
260-- >>> stream1 = Stream.fromList [1,2]
261-- >>> stream2 = Stream.fromList [3,4]
262-- >>> Stream.toList $ Stream.fromWSerial $ stream1 `wSerial` stream2
265-- Note, for singleton streams 'wSerial' and 'serial' are identical.
267-- Note that this operation cannot be used to fold a container of infinite
268-- streams but it can be used for very large streams as the state that it needs
269-- to maintain is proportional to the logarithm of the number of streams.
273-- /Since: 0.2.0 ("Streamly")/
277-- Note that evaluation of @a \`wSerial` b \`wSerial` c@ does not interleave
278-- @a@, @b@ and @c@ with equal priority. This expression is equivalent to @a
279-- \`wSerial` (b \`wSerial` c)@, therefore, it fairly interleaves @a@ with the
280-- result of @b \`wSerial` c@. For example, @Stream.fromList [1,2] \`wSerial`
281-- Stream.fromList [3,4] \`wSerial` Stream.fromList [5,6]@ would result in
282-- [1,3,2,5,4,6]. In other words, the leftmost stream gets the same scheduling
283-- priority as the rest of the streams taken together. The same is true for
284-- each subexpression on the right.
286{-# INLINE wSerial #-}
287wSerial :: IsStream t => t m a -> t m a -> t m a
288wSerial m1 m2 = fromStream $ WSerial.wSerialK (toStream m1) (toStream m2)
290-- XXX Same as 'wSerial'. We should perhaps rename wSerial to interleave.
291-- XXX Document the interleaving behavior of side effects in all the
292-- interleaving combinators.
293-- XXX Write time-domain equivalents of these. In the time domain we can
294-- interleave two streams such that the value of second stream is always taken
295-- from its last value even if no new value is being yielded, like
296-- zipWithLatest. It would be something like interleaveWithLatest.
298-- | Interleaves the outputs of two streams, yielding elements from each stream
299-- alternately, starting from the first stream. If any of the streams finishes
300-- early the other stream continues alone until it too finishes.
302-- >>> :set -XOverloadedStrings
303-- >>> import Data.Functor.Identity (Identity)
304-- >>> Stream.interleave "ab" ",,,," :: Stream.SerialT Identity Char
307-- >>> Stream.interleave "abcd" ",," :: Stream.SerialT Identity Char
310-- 'interleave' is dual to 'interleaveMin', it can be called @interleaveMax@.
312-- Do not use at scale in concatMapWith.
315{-# INLINE interleave #-}
316interleave ::(IsStream t, Monad m) => t m b -> t m b -> t m b
317interleave m1 m2 = fromStreamD $ D.interleave (toStreamD m1) (toStreamD m2)
319-- | Interleaves the outputs of two streams, yielding elements from each stream
320-- alternately, starting from the first stream. As soon as the first stream
321-- finishes, the output stops, discarding the remaining part of the second
322-- stream. In this case, the last element in the resulting stream would be from
323-- the second stream. If the second stream finishes early then the first stream
324-- still continues to yield elements until it finishes.
326-- >>> :set -XOverloadedStrings
327-- >>> import Data.Functor.Identity (Identity)
328-- >>> Stream.interleaveSuffix "abc" ",,,," :: Stream.SerialT Identity Char
330-- >>> Stream.interleaveSuffix "abc" "," :: Stream.SerialT Identity Char
333-- 'interleaveSuffix' is a dual of 'interleaveInfix'.
335-- Do not use at scale in concatMapWith.
338{-# INLINE interleaveSuffix #-}
339interleaveSuffix ::(IsStream t, Monad m) => t m b -> t m b -> t m b
340interleaveSuffix m1 m2 =
341 fromStreamD $ D.interleaveSuffix (toStreamD m1) (toStreamD m2)
343-- | Interleaves the outputs of two streams, yielding elements from each stream
344-- alternately, starting from the first stream and ending at the first stream.
345-- If the second stream is longer than the first, elements from the second
346-- stream are infixed with elements from the first stream. If the first stream
347-- is longer then it continues yielding elements even after the second stream
350-- >>> :set -XOverloadedStrings
351-- >>> import Data.Functor.Identity (Identity)
352-- >>> Stream.interleaveInfix "abc" ",,,," :: Stream.SerialT Identity Char
354-- >>> Stream.interleaveInfix "abc" "," :: Stream.SerialT Identity Char
357-- 'interleaveInfix' is a dual of 'interleaveSuffix'.
359-- Do not use at scale in concatMapWith.
362{-# INLINE interleaveInfix #-}
363interleaveInfix ::(IsStream t, Monad m) => t m b -> t m b -> t m b
364interleaveInfix m1 m2 =
365 fromStreamD $ D.interleaveInfix (toStreamD m1) (toStreamD m2)
367-- | Interleaves the outputs of two streams, yielding elements from each stream
368-- alternately, starting from the first stream. The output stops as soon as any
369-- of the two streams finishes, discarding the remaining part of the other
370-- stream. The last element of the resulting stream would be from the longer
373-- >>> :set -XOverloadedStrings
374-- >>> import Data.Functor.Identity (Identity)
375-- >>> Stream.interleaveMin "ab" ",,,," :: Stream.SerialT Identity Char
377-- >>> Stream.interleaveMin "abcd" ",," :: Stream.SerialT Identity Char
380-- 'interleaveMin' is dual to 'interleave'.
382-- Do not use at scale in concatMapWith.
385{-# INLINE interleaveMin #-}
386interleaveMin ::(IsStream t, Monad m) => t m b -> t m b -> t m b
387interleaveMin m1 m2 = fromStreamD $ D.interleaveMin (toStreamD m1) (toStreamD m2)
389-------------------------------------------------------------------------------
391-------------------------------------------------------------------------------
393-- | Schedule the execution of two streams in a fair round-robin manner,
394-- executing each stream once, alternately. Execution of a stream may not
395-- necessarily result in an output, a stream may chose to @Skip@ producing an
396-- element until later giving the other stream a chance to run. Therefore, this
397-- combinator fairly interleaves the execution of two streams rather than
398-- fairly interleaving the output of the two streams. This can be useful in
399-- co-operative multitasking without using explicit threads. This can be used
400-- as an alternative to `async`.
402-- Do not use at scale in concatMapWith.
405{-# INLINE roundrobin #-}
406roundrobin ::(IsStream t, Monad m) => t m b -> t m b -> t m b
407roundrobin m1 m2 = fromStreamD $ D.roundRobin (toStreamD m1) (toStreamD m2)
411-- | Merges two streams, both the streams may be evaluated concurrently,
412-- outputs from both are used as they arrive:
414-- >>> import Streamly.Prelude (async)
415-- >>> stream1 = Stream.fromEffect (delay 4)
416-- >>> stream2 = Stream.fromEffect (delay 2)
417-- >>> Stream.toList $ stream1 `async` stream2
422-- Multiple streams can be combined. With enough threads, all of them can be
423-- scheduled simultaneously:
425-- >>> stream3 = Stream.fromEffect (delay 1)
426-- >>> Stream.toList $ stream1 `async` stream2 `async` stream3
430-- With 2 threads, only two can be scheduled at a time, when one of those
431-- finishes, the third one gets scheduled:
433-- >>> Stream.toList $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3
437-- With a single thread, it becomes serial:
439-- >>> Stream.toList $ Stream.maxThreads 1 $ stream1 `async` stream2 `async` stream3
443-- Only streams are scheduled for async evaluation, how actions within a
444-- stream are evaluated depends on the stream type. If it is a concurrent
445-- stream they will be evaluated concurrently.
447-- In the following example, both the streams are scheduled for concurrent
448-- evaluation but each individual stream is evaluated serially:
450-- >>> stream1 = Stream.fromListM $ Prelude.map delay [3,3] -- SerialT IO Int
451-- >>> stream2 = Stream.fromListM $ Prelude.map delay [1,1] -- SerialT IO Int
452-- >>> Stream.toList $ stream1 `async` stream2 -- IO [Int]
456-- If total threads are 2, the third stream is scheduled only after one of the
457-- first two has finished:
459-- > stream3 = Stream.fromListM $ Prelude.map delay [2,2] -- SerialT IO Int
460-- > Stream.toList $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3 -- IO [Int]
464-- Thus 'async' goes deep in first few streams rather than going wide in all
465-- streams. It prefers to evaluate the leftmost streams as much as possible.
466-- Because of this behavior, 'async' can be safely used to fold an infinite
467-- lazy container of streams.
469-- /Since: 0.2.0 ("Streamly")/
473async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
474async m1 m2 = fromStream $ asyncK (toStream m1) (toStream m2)
478-- | For singleton streams, 'wAsync' is the same as 'async'. See 'async' for
479-- singleton stream behavior. For multi-element streams, while 'async' is left
480-- biased i.e. it tries to evaluate the left side stream as much as possible,
481-- 'wAsync' tries to schedule them both fairly. In other words, 'async' goes
482-- deep while 'wAsync' goes wide. However, outputs are always used as they
485-- With a single thread, 'async' starts behaving like 'serial' while 'wAsync'
486-- starts behaving like 'wSerial'.
488-- >>> import Streamly.Prelude (async, wAsync)
489-- >>> stream1 = Stream.fromList [1,2,3]
490-- >>> stream2 = Stream.fromList [4,5,6]
491-- >>> Stream.toList $ Stream.fromAsync $ Stream.maxThreads 1 $ stream1 `async` stream2
494-- >>> Stream.toList $ Stream.fromWAsync $ Stream.maxThreads 1 $ stream1 `wAsync` stream2
497-- With two threads available, and combining three streams:
499-- >>> stream3 = Stream.fromList [7,8,9]
500-- >>> Stream.toList $ Stream.fromAsync $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3
501-- [1,2,3,4,5,6,7,8,9]
503-- >>> Stream.toList $ Stream.fromWAsync $ Stream.maxThreads 2 $ stream1 `wAsync` stream2 `wAsync` stream3
504-- [1,4,2,7,5,3,8,6,9]
506-- This operation cannot be used to fold an infinite lazy container of streams,
507-- because it schedules all the streams in a round robin manner.
509-- Note that 'WSerialT' and single threaded 'WAsyncT' both interleave streams
510-- but the exact scheduling is slightly different in both cases.
514-- /Since: 0.2.0 ("Streamly")/
516-- Scheduling details:
518-- This is how the execution of the above example proceeds:
520-- 1. The scheduler queue is initialized with @[S.fromList [1,2,3],
521-- (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])]@ assuming the head of the
522-- queue is represented by the rightmost item.
523-- 2. @S.fromList [1,2,3]@ is executed, yielding the element @1@ and putting
524-- @[2,3]@ at the back of the scheduler queue. The scheduler queue now looks
525-- like @[(S.fromList [4,5,6]) \<> (S.fromList [7,8,9]), S.fromList [2,3]]@.
526-- 3. Now @(S.fromList [4,5,6]) \<> (S.fromList [7,8,9])@ is picked up for
527-- execution, @S.fromList [7,8,9]@ is added at the back of the queue and
528-- @S.fromList [4,5,6]@ is executed, yielding the element @4@ and adding
529-- @S.fromList [5,6]@ at the back of the queue. The queue now looks like
530-- @[S.fromList [2,3], S.fromList [7,8,9], S.fromList [5,6]]@.
531-- 4. Note that the scheduler queue expands by one more stream component in
532-- every pass because one more @<>@ is broken down into two components. At this
533-- point there are no more @<>@ operations to be broken down further and the
534-- queue has reached its maximum size. Now these streams are scheduled in
535-- round-robin fashion yielding @[2,7,5,3,8,6,9]@.
537-- As we see above, in a right associated expression composed with @<>@, only
538-- one @<>@ operation is broken down into two components in one execution,
539-- therefore, if we have @n@ streams composed using @<>@ it will take @n@
540-- scheduler passes to expand the whole expression. By the time @n-th@
541-- component is added to the scheduler queue, the first component would have
542-- received @n@ scheduler passes.
545wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
546wAsync m1 m2 = fromStream $ wAsyncK (toStream m1) (toStream m2)
550-- | Appends two streams, both the streams may be evaluated concurrently but
551-- the outputs are used in the same order as the corresponding actions in the
552-- original streams, side effects will happen in the order in which the streams
555-- >>> import Streamly.Prelude (ahead, SerialT)
556-- >>> stream1 = Stream.fromEffect (delay 4) :: SerialT IO Int
557-- >>> stream2 = Stream.fromEffect (delay 2) :: SerialT IO Int
558-- >>> Stream.toList $ stream1 `ahead` stream2 :: IO [Int]
563-- Multiple streams can be combined. With enough threads, all of them can be
564-- scheduled simultaneously:
566-- >>> stream3 = Stream.fromEffect (delay 1)
567-- >>> Stream.toList $ stream1 `ahead` stream2 `ahead` stream3
573-- With 2 threads, only two can be scheduled at a time, when one of those
574-- finishes, the third one gets scheduled:
576-- >>> Stream.toList $ Stream.maxThreads 2 $ stream1 `ahead` stream2 `ahead` stream3
582-- Only streams are scheduled for ahead evaluation, how actions within a stream
583-- are evaluated depends on the stream type. If it is a concurrent stream they
584-- will be evaluated concurrently. It may not make much sense combining serial
585-- streams using 'ahead'.
587-- 'ahead' can be safely used to fold an infinite lazy container of streams.
589-- /Since: 0.3.0 ("Streamly")/
593ahead :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
594ahead m1 m2 = fromStream $ aheadK (toStream m1) (toStream m2)
598-- | Like 'Streamly.Prelude.async' except that the execution is much more
599-- strict. There is no limit on the number of threads. While
600-- 'Streamly.Prelude.async' may not schedule a stream if there is no demand
601-- from the consumer, 'parallel' always evaluates both the streams immediately.
602-- The only limit that applies to 'parallel' is 'Streamly.Prelude.maxBuffer'.
603-- Evaluation may block if the output buffer becomes full.
605-- >>> import Streamly.Prelude (parallel)
606-- >>> stream = Stream.fromEffect (delay 2) `parallel` Stream.fromEffect (delay 1)
607-- >>> Stream.toList stream -- IO [Int]
612-- 'parallel' guarantees that all the streams are scheduled for execution
613-- immediately, therefore, we could use things like starting timers inside the
614-- streams and relying on the fact that all timers were started at the same
617-- Unlike 'async' this operation cannot be used to fold an infinite lazy
618-- container of streams, because it schedules all the streams strictly
621-- /Since: 0.2.0 ("Streamly")/
624{-# INLINE parallel #-}
625parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
626parallel m1 m2 = fromStream $ Par.parallelK (toStream m1) (toStream m2)
628-- This is a race like combinator for streams.
630-- | Like `parallel` but stops the output as soon as any of the two streams
634{-# INLINE parallelMin #-}
635parallelMin :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
636parallelMin m1 m2 = fromStream $ Par.parallelMinK (toStream m1) (toStream m2)
638------------------------------------------------------------------------------
640------------------------------------------------------------------------------
642-- | Like 'zipAsyncWith' but with a monadic zipping function.
645{-# INLINE zipAsyncWithM #-}
646zipAsyncWithM :: (IsStream t, MonadAsync m)
647 => (a -> b -> m c) -> t m a -> t m b -> t m c
648zipAsyncWithM f m1 m2 =
649 fromStream $ ZipAsync.zipAsyncWithMK f (toStream m1) (toStream m2)
651-- XXX Should we rename this to zipParWith or zipParallelWith? This can happen
652-- along with the change of behvaior to end the stream concurrently.
654-- | Like 'zipWith' but zips concurrently i.e. both the streams being zipped
655-- are evaluated concurrently using the 'ParallelT' concurrent evaluation
656-- style. The maximum number of elements of each stream evaluated in advance
657-- can be controlled by 'maxBuffer'.
659-- The stream ends if stream @a@ or stream @b@ ends. However, if stream @b@
660-- ends while we are still evaluating stream @a@ and waiting for a result then
661-- stream will not end until after the evaluation of stream @a@ finishes. This
662-- behavior can potentially be changed in future to end the stream immediately
663-- as soon as any of the stream end is detected.
666{-# INLINE zipAsyncWith #-}
667zipAsyncWith :: (IsStream t, MonadAsync m)
668 => (a -> b -> c) -> t m a -> t m b -> t m c
669zipAsyncWith f = zipAsyncWithM (\a b -> return (f a b))
671------------------------------------------------------------------------------
672-- Merging (sorted streams)
673------------------------------------------------------------------------------
675-- | Merge two streams using a comparison function. The head elements of both
676-- the streams are compared and the smaller of the two elements is emitted, if
677-- both elements are equal then the element from the first stream is used
680-- If the streams are sorted in ascending order, the resulting stream would
681-- also remain sorted in ascending order.
684-- >>> Stream.toList $ Stream.mergeBy compare (Stream.fromList [1,3,5]) (Stream.fromList [2,4,6,8])
689-- See also: 'mergeByMFused'
692{-# INLINE mergeBy #-}
693mergeBy :: IsStream t => (a -> a -> Ordering) -> t m a -> t m a -> t m a
694mergeBy f m1 m2 = fromStream $ K.mergeBy f (toStream m1) (toStream m2)
696-- | Like 'mergeBy' but with a monadic comparison function.
698-- Merge two streams randomly:
701-- > randomly _ _ = randomIO >>= \x -> return $ if x then LT else GT
702-- > Stream.toList $ Stream.mergeByM randomly (Stream.fromList [1,1,1,1]) (Stream.fromList [2,2,2,2])
706-- Merge two streams in a proportion of 2:1:
711-- let proportionately m n = do
712-- ref <- newIORef $ cycle $ Prelude.concat [Prelude.replicate m LT, Prelude.replicate n GT]
713-- return $ \_ _ -> do
715-- writeIORef ref $ Prelude.tail r
716-- return $ Prelude.head r
717-- f <- proportionately 2 1
718-- xs <- Stream.toList $ Stream.mergeByM f (Stream.fromList [1,1,1,1,1,1]) (Stream.fromList [2,2,2])
721-- [1,1,2,1,1,2,1,1,2]
725-- See also: 'mergeByMFused'
728{-# INLINE mergeByM #-}
730 :: (IsStream t, Monad m)
731 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
732mergeByM f m1 m2 = fromStream $ K.mergeByM f (toStream m1) (toStream m2)
734-- XXX Fused versions should probably go to a separate module using the same
735-- names for the combinators.
737-- | Like 'mergeByM' but much faster, works best when merging statically known
738-- number of streams. When merging more than two streams try to merge pairs and
739-- pair pf pairs in a tree like structure.'mergeByM' works better with variable
740-- number of streams being merged using 'concatPairsWith'.
743{-# INLINE mergeByMFused #-}
745 :: (IsStream t, Monad m)
746 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
747mergeByMFused f m1 m2 =
748 fromStreamD $ D.mergeByM f (toStreamD m1) (toStreamD m2)
750-- | Like 'mergeByM' but stops merging as soon as any of the two streams stops.
753{-# INLINABLE mergeMinBy #-}
754mergeMinBy :: -- (IsStream t, Monad m) =>
755 (a -> a -> m Ordering) -> t m a -> t m a -> t m a
756mergeMinBy _f _m1 _m2 = undefined
757 -- fromStreamD $ D.mergeMinBy f (toStreamD m1) (toStreamD m2)
759-- | Like 'mergeByM' but stops merging as soon as the first stream stops.
762{-# INLINABLE mergeFstBy #-}
763mergeFstBy :: -- (IsStream t, Monad m) =>
764 (a -> a -> m Ordering) -> t m a -> t m a -> t m a
765mergeFstBy _f _m1 _m2 = undefined
766 -- fromStreamS $ D.mergeFstBy f (toStreamD m1) (toStreamD m2)
768-- XXX we may want to use the name "merge" differently
769-- | Same as @'mergeBy' 'compare'@.
771-- >>> Stream.toList $ Stream.merge (Stream.fromList [1,3,5]) (Stream.fromList [2,4,6,8])
776{-# INLINABLE merge #-}
778 (IsStream t, Ord a) => t m a -> t m a -> t m a
779merge = mergeBy compare
781-- | Like 'mergeBy' but merges concurrently (i.e. both the elements being
782-- merged are generated concurrently).
785{-# INLINE mergeAsyncBy #-}
786mergeAsyncBy :: (IsStream t, MonadAsync m)
787 => (a -> a -> Ordering) -> t m a -> t m a -> t m a
788mergeAsyncBy f = mergeAsyncByM (\a b -> return $ f a b)
790-- | Like 'mergeByM' but merges concurrently (i.e. both the elements being
791-- merged are generated concurrently).
794{-# INLINE mergeAsyncByM #-}
795mergeAsyncByM :: (IsStream t, MonadAsync m)
796 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
797mergeAsyncByM f m1 m2 =
799 let par = Par.mkParallelD . toStreamD
800 in D.mergeByM f (par m1) (par m2)
804{-# DEPRECATED concatUnfold "Please use unfoldMany instead." #-}
805{-# INLINE concatUnfold #-}
806concatUnfold ::(IsStream t, Monad m) => Unfold m a b -> t m a -> t m b
807concatUnfold u m = fromStreamD $ D.unfoldMany u (toStreamD m)
809------------------------------------------------------------------------------
810-- Combine N Streams - unfoldMany
811------------------------------------------------------------------------------
813-- | Like 'concatMap' but uses an 'Unfold' for stream generation. Unlike
814-- 'concatMap' this can fuse the 'Unfold' code with the inner loop and
815-- therefore provide many times better performance.
818{-# INLINE unfoldMany #-}
819unfoldMany ::(IsStream t, Monad m) => Unfold m a b -> t m a -> t m b
820unfoldMany u m = fromStreamD $ D.unfoldMany u (toStreamD m)
822-- | Like 'unfoldMany' but interleaves the streams in the same way as
823-- 'interleave' behaves instead of appending them.
826{-# INLINE unfoldManyInterleave #-}
827unfoldManyInterleave ::(IsStream t, Monad m)
828 => Unfold m a b -> t m a -> t m b
829unfoldManyInterleave u m =
830 fromStreamD $ D.unfoldManyInterleave u (toStreamD m)
832-- | Like 'unfoldMany' but executes the streams in the same way as
836{-# INLINE unfoldManyRoundRobin #-}
837unfoldManyRoundRobin ::(IsStream t, Monad m)
838 => Unfold m a b -> t m a -> t m b
839unfoldManyRoundRobin u m =
840 fromStreamD $ D.unfoldManyRoundRobin u (toStreamD m)
842------------------------------------------------------------------------------
843-- Combine N Streams - interpose
844------------------------------------------------------------------------------
846-- > interpose x unf str = gintercalate unf str UF.identity (repeat x)
848-- | Unfold the elements of a stream, intersperse the given element between the
849-- unfolded streams and then concat them into a single stream.
851-- > unwords = S.interpose ' '
854{-# INLINE interpose #-}
855interpose :: (IsStream t, Monad m)
856 => c -> Unfold m b c -> t m b -> t m c
858 fromStreamD $ D.interpose (return x) unf (toStreamD str)
860-- interposeSuffix x unf str = gintercalateSuffix unf str UF.identity (repeat x)
862-- | Unfold the elements of a stream, append the given element after each
863-- unfolded stream and then concat them into a single stream.
865-- > unlines = S.interposeSuffix '\n'
868{-# INLINE interposeSuffix #-}
869interposeSuffix :: (IsStream t, Monad m)
870 => c -> Unfold m b c -> t m b -> t m c
871interposeSuffix x unf str =
872 fromStreamD $ D.interposeSuffix (return x) unf (toStreamD str)
874------------------------------------------------------------------------------
875-- Combine N Streams - intercalate
876------------------------------------------------------------------------------
878-- XXX we can swap the order of arguments to gintercalate so that the
879-- definition of unfoldMany becomes simpler? The first stream should be
880-- infixed inside the second one. However, if we change the order in
881-- "interleave" as well similarly, then that will make it a bit unintuitive.
883-- > unfoldMany unf str =
884-- > gintercalate unf str (UF.nilM (\_ -> return ())) (repeat ())
886-- | 'interleaveInfix' followed by unfold and concat.
889{-# INLINE gintercalate #-}
891 :: (IsStream t, Monad m)
892 => Unfold m a c -> t m a -> Unfold m b c -> t m b -> t m c
893gintercalate unf1 str1 unf2 str2 =
894 fromStreamD $ D.gintercalate
895 unf1 (toStreamD str1)
896 unf2 (toStreamD str2)
898-- > intercalate unf seed str = gintercalate unf str unf (repeatM seed)
900-- | 'intersperse' followed by unfold and concat.
902-- > intercalate unf a str = unfoldMany unf $ intersperse a str
903-- > intersperse = intercalate (Unfold.function id)
904-- > unwords = intercalate Unfold.fromList " "
906-- >>> Stream.toList $ Stream.intercalate Unfold.fromList " " $ Stream.fromList ["abc", "def", "ghi"]
910{-# INLINE intercalate #-}
911intercalate :: (IsStream t, Monad m)
912 => Unfold m b c -> b -> t m b -> t m c
913intercalate unf seed str = fromStreamD $
914 D.unfoldMany unf $ D.intersperse seed (toStreamD str)
916-- | 'interleaveSuffix' followed by unfold and concat.
919{-# INLINE gintercalateSuffix #-}
921 :: (IsStream t, Monad m)
922 => Unfold m a c -> t m a -> Unfold m b c -> t m b -> t m c
923gintercalateSuffix unf1 str1 unf2 str2 =
924 fromStreamD $ D.gintercalateSuffix
925 unf1 (toStreamD str1)
926 unf2 (toStreamD str2)
928-- > intercalateSuffix unf seed str = gintercalateSuffix unf str unf (repeatM seed)
930-- | 'intersperseSuffix' followed by unfold and concat.
932-- > intercalateSuffix unf a str = unfoldMany unf $ intersperseSuffix a str
933-- > intersperseSuffix = intercalateSuffix (Unfold.function id)
934-- > unlines = intercalateSuffix Unfold.fromList "\n"
936-- >>> Stream.toList $ Stream.intercalateSuffix Unfold.fromList "\n" $ Stream.fromList ["abc", "def", "ghi"]
940{-# INLINE intercalateSuffix #-}
941intercalateSuffix :: (IsStream t, Monad m)
942 => Unfold m b c -> b -> t m b -> t m c
943intercalateSuffix unf seed str = fromStreamD $ D.unfoldMany unf
944 $ D.intersperseSuffix (return seed) (toStreamD str)
946------------------------------------------------------------------------------
947-- Combine N Streams - concatMap
948------------------------------------------------------------------------------
950-- | Flatten a stream of streams to a single stream.
953-- concat = concatMap id
958concat :: (IsStream t, Monad m) => t m (t m a) -> t m a
961------------------------------------------------------------------------------
962-- Combine N Streams - concatMap
963------------------------------------------------------------------------------
965-- | Like 'concatMapWith' but carries a state which can be used to share
966-- information across multiple steps of concat.
969-- concatSmapMWith combine f initial = concatMapWith combine id . smapM f initial
974{-# INLINE concatSmapMWith #-}
976 :: (IsStream t, Monad m)
977 => (t m b -> t m b -> t m b)
978 -> (s -> a -> m (s, t m b))
982concatSmapMWith combine f initial =
983 IsStream.concatMapWith combine id . smapM f initial
985-- XXX Implement a StreamD version for fusion.
987-- | Combine streams in pairs using a binary stream combinator, then combine
988-- the resulting streams in pairs recursively until we get to a single combined
991-- For example, you can sort a stream using merge sort like this:
993-- >>> Stream.toList $ Stream.concatPairsWith (Stream.mergeBy compare) Stream.fromPure $ Stream.fromList [5,1,7,9,2]
996-- /Caution: the stream of streams must be finite/
1000{-# INLINE concatPairsWith #-}
1001concatPairsWith :: IsStream t =>
1002 (t m b -> t m b -> t m b)
1006concatPairsWith par f m =
1009 (\s1 s2 -> toStream $ fromStream s1 `par` fromStream s2)
1013------------------------------------------------------------------------------
1014-- IterateMap - Map and flatten Trees of Streams
1015------------------------------------------------------------------------------
1017-- | Like 'iterateM' but iterates after mapping a stream generator on the
1020-- Yield an input element in the output stream, map a stream generator on it
1021-- and then do the same on the resulting stream. This can be used for a depth
1022-- first traversal of a tree like structure.
1024-- Note that 'iterateM' is a special case of 'iterateMapWith':
1027-- iterateM f = iterateMapWith serial (fromEffect . f) . fromEffect
1030-- It can be used to traverse a tree structure. For example, to list a
1034-- Stream.iterateMapWith Stream.serial
1035-- (either Dir.toEither (const nil))
1036-- (fromPure (Left "tmp"))
1041{-# INLINE iterateMapWith #-}
1044 => (t m a -> t m a -> t m a)
1048iterateMapWith combine f = IsStream.concatMapWith combine go
1050 go x = fromPure x `combine` IsStream.concatMapWith combine go (f x)
1052-- | Same as @iterateMapWith Stream.serial@ but more efficient due to stream
1056{-# INLINE iterateUnfold #-}
1057iterateUnfold :: -- (IsStream t, MonadAsync m) =>
1058 Unfold m a a -> t m a -> t m a
1059iterateUnfold = undefined
1061------------------------------------------------------------------------------
1063------------------------------------------------------------------------------
1065-- To traverse graphs we need a state to be carried around in the traversal.
1066-- For example, we can use a hashmap to store the visited status of nodes.
1068-- | Like 'iterateMap' but carries a state in the stream generation function.
1069-- This can be used to traverse graph like structures, we can remember the
1070-- visited nodes in the state to avoid cycles.
1072-- Note that a combination of 'iterateMap' and 'usingState' can also be used to
1073-- traverse graphs. However, this function provides a more localized state
1074-- instead of using a global state.
1080{-# INLINE iterateSmapMWith #-}
1082 :: (IsStream t, Monad m)
1083 => (t m a -> t m a -> t m a)
1084 -> (b -> a -> m (b, t m a))
1088iterateSmapMWith combine f initial stream =
1090 (\b -> IsStream.concatMapWith combine (go b) stream)
1091 (fromEffect initial)
1095 go b a = fromPure a `combine` feedback b a
1099 (\(b1, s) -> IsStream.concatMapWith combine (go b1) s)
1100 (fromEffect $ f b a)
1102------------------------------------------------------------------------------
1104------------------------------------------------------------------------------
1106-- Keep concating either streams as long as rights are generated, stop as soon
1107-- as a left is generated and concat the left stream.
1109-- See also: 'handle'
1115 :: -- (IsStream t, MonadAsync m) =>
1116 (forall x. t m x -> t m x -> t m x)
1117 -> (a -> t m (Either (t m b) b))
1120concatMapEitherWith = undefined
1123-- | In an 'Either' stream iterate on 'Left's. This is a special case of
1127-- iterateMapLeftsWith combine f = iterateMapWith combine (either f (const nil))
1130-- To traverse a directory tree:
1133-- iterateMapLeftsWith serial Dir.toEither (fromPure (Left "tmp"))
1138{-# INLINE iterateMapLeftsWith #-}
1140 :: (IsStream t, b ~ Either a c)
1141 => (t m b -> t m b -> t m b)
1145iterateMapLeftsWith combine f =
1146 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