1{-# OPTIONS_GHC -Wno-deprecations #-}
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 {-# DEPRECATED "Please use \"Streamly.Data.Stream.*\" instead." #-}
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
89 -- * Combine Streams and Unfolds
91 -- Expand a stream by repeatedly using an unfold and merging the resulting
92 -- streams. Functions generally ending in the shape:
94 -- @Unfold m a b -> t m a -> t m b@
96 -- ** Append Many (Unfold)
97 -- | Unfold and flatten streams.
99 , unfoldManyInterleave
100 , unfoldManyRoundRobin
103 -- | Insert effects between streams. Like unfoldMany but intersperses an
104 -- effect between the streams. A special case of gintercalate.
110 -- | Insert Streams between Streams.
111 -- Like unfoldMany but intersperses streams from another source between
112 -- the streams from the first source.
118 -- * Append Many (concatMap)
119 -- | Map and serially append streams. 'concatMapM' is a generalization of
120 -- the binary append operation to append many streams.
126 -- * Flatten Containers
127 -- | Flatten 'Foldable' containers using the binary stream merging
129 , IsStream.concatFoldableWith
130 , IsStream.concatMapFoldableWith
131 , IsStream.concatForFoldableWith
134 -- | Map and flatten a stream like 'concatMap' but using a custom binary
135 -- stream merging combinator instead of just appending the streams. The
136 -- merging occurs sequentially, it works efficiently for 'serial', 'async',
137 -- 'ahead' like merge operations where we consume one stream before the
138 -- next or in case of 'wAsync' or 'parallel' where we consume all streams
139 -- simultaneously anyway.
141 -- However, in cases where the merging consumes streams in a round robin
142 -- fashion, a pair wise merging using 'concatPairsWith' would be more
143 -- efficient. These cases include operations like 'mergeBy' or 'zipWith'.
145 , IsStream.concatMapWith
150 -- | See the notes about suitable merge functions in the 'concatMapWith'
155 -- | Map and flatten Trees of Streams
158 , iterateMapLeftsWith
168import Streamly.Internal.Control.Concurrent (MonadAsync)
169import Streamly.Internal.Data.Stream.Ahead (aheadK)
170import Streamly.Internal.Data.Stream.Async (asyncK, wAsyncK)
171import Streamly.Internal.Data.Stream.IsStream.Common
172 ( concatM, concatMapM, concatMap, smapM, fromPure, fromEffect, parallelFst
174import Streamly.Internal.Data.Stream.IsStream.Type
175 (IsStream(..), fromStreamD, toStreamD)
176import Streamly.Data.Unfold (Unfold)
178import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
179import qualified Streamly.Internal.Data.Stream.Parallel as Par
180import qualified Streamly.Internal.Data.Stream.Serial as Serial
181import qualified Streamly.Internal.Data.Stream as D
182 (append, interleave, interleaveFstSuffix, interleaveFst, interleaveMin
183 , roundRobin, mergeByM, unfoldMany, unfoldInterleave, intersperse
184 , unfoldRoundRobin, interpose, interposeSuffix, gintercalate
185 , gintercalateSuffix, intersperseMSuffix)
186import qualified Streamly.Internal.Data.StreamK as K
187 (interleave, append, mergeMapWith, mergeBy, mergeByM)
188import qualified Streamly.Internal.Data.Stream.ZipAsync as ZipAsync
190import Prelude hiding (concat, concatMap, zipWith)
194-- >>> :set -fno-warn-deprecations
195-- >>> import Control.Concurrent (threadDelay)
196-- >>> import Data.IORef
197-- >>> import Prelude hiding (zipWith, concatMap, concat)
198-- >>> import qualified Streamly.Prelude as Stream
199-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream
200-- >>> import qualified Streamly.Data.Fold as Fold
201-- >>> import qualified Streamly.Internal.Data.Fold as Fold
202-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
203-- >>> import qualified Streamly.Internal.Data.Parser as Parser
204-- >>> import qualified Streamly.Data.Array as Array
207-- threadDelay (n * 1000000) -- sleep for n seconds
208-- putStrLn (show n ++ " sec") -- print "n sec"
213-------------------------------------------------------------------------------
215-------------------------------------------------------------------------------
217-- XXX Reconcile the names "serial" and "append".
219-- | Append the outputs of two streams, yielding all the elements from the
220-- first stream and then yielding all the elements from the second stream.
222-- IMPORTANT NOTE: This could be 100x faster than @serial/<>@ for appending a
223-- few (say 100) streams because it can fuse via stream fusion. However, it
224-- does not scale for a large number of streams (say 1000s) and becomes
225-- qudartically slow. Therefore use this for custom appending of a few streams
226-- but use 'concatMap' or 'concatMapWith serial' for appending @n@ streams or
227-- infinite containers of streams.
231append ::(IsStream t, Monad m) => t m b -> t m b -> t m b
232append m1 m2 = fromStreamD $ D.append (toStreamD m1) (toStreamD m2)
236-- | Appends two streams sequentially, yielding all elements from the first
237-- stream, and then all elements from the second stream.
239-- >>> import Streamly.Prelude (serial)
240-- >>> stream1 = Stream.fromList [1,2]
241-- >>> stream2 = Stream.fromList [3,4]
242-- >>> Stream.toList $ stream1 `serial` stream2
245-- This operation can be used to fold an infinite lazy container of streams.
247-- /Since: 0.2.0 ("Streamly")/
251serial :: IsStream t => t m a -> t m a -> t m a
252serial m1 m2 = fromStream $ K.append (toStream m1) (toStream m2)
254-------------------------------------------------------------------------------
256-------------------------------------------------------------------------------
260-- XXX doc duplicated from Stream.Serial module.
262-- | Interleaves two streams, yielding one element from each stream
263-- alternately. When one stream stops the rest of the other stream is used in
266-- >>> import Streamly.Prelude (wSerial)
267-- >>> stream1 = Stream.fromList [1,2]
268-- >>> stream2 = Stream.fromList [3,4]
269-- >>> Stream.toList $ Stream.fromWSerial $ stream1 `wSerial` stream2
272-- Note, for singleton streams 'wSerial' and 'serial' are identical.
274-- Note that this operation cannot be used to fold a container of infinite
275-- streams but it can be used for very large streams as the state that it needs
276-- to maintain is proportional to the logarithm of the number of streams.
280-- /Since: 0.2.0 ("Streamly")/
284-- Note that evaluation of @a \`wSerial` b \`wSerial` c@ does not interleave
285-- @a@, @b@ and @c@ with equal priority. This expression is equivalent to @a
286-- \`wSerial` (b \`wSerial` c)@, therefore, it fairly interleaves @a@ with the
287-- result of @b \`wSerial` c@. For example, @Stream.fromList [1,2] \`wSerial`
288-- Stream.fromList [3,4] \`wSerial` Stream.fromList [5,6]@ would result in
289-- [1,3,2,5,4,6]. In other words, the leftmost stream gets the same scheduling
290-- priority as the rest of the streams taken together. The same is true for
291-- each subexpression on the right.
293{-# INLINE wSerial #-}
294wSerial :: IsStream t => t m a -> t m a -> t m a
295wSerial m1 m2 = fromStream $ K.interleave (toStream m1) (toStream m2)
297-- XXX Same as 'wSerial'. We should perhaps rename wSerial to interleave.
298-- XXX Document the interleaving behavior of side effects in all the
299-- interleaving combinators.
300-- XXX Write time-domain equivalents of these. In the time domain we can
301-- interleave two streams such that the value of second stream is always taken
302-- from its last value even if no new value is being yielded, like
303-- zipWithLatest. It would be something like interleaveWithLatest.
305-- | Interleaves the outputs of two streams, yielding elements from each stream
306-- alternately, starting from the first stream. If any of the streams finishes
307-- early the other stream continues alone until it too finishes.
309-- >>> :set -XOverloadedStrings
310-- >>> import Data.Functor.Identity (Identity)
311-- >>> Stream.interleave "ab" ",,,," :: Stream.SerialT Identity Char
314-- >>> Stream.interleave "abcd" ",," :: Stream.SerialT Identity Char
317-- 'interleave' is dual to 'interleaveMin', it can be called @interleaveMax@.
319-- Do not use at scale in concatMapWith.
322{-# INLINE interleave #-}
323interleave ::(IsStream t, Monad m) => t m b -> t m b -> t m b
324interleave m1 m2 = fromStreamD $ D.interleave (toStreamD m1) (toStreamD m2)
326-- | Interleaves the outputs of two streams, yielding elements from each stream
327-- alternately, starting from the first stream. As soon as the first stream
328-- finishes, the output stops, discarding the remaining part of the second
329-- stream. In this case, the last element in the resulting stream would be from
330-- the second stream. If the second stream finishes early then the first stream
331-- still continues to yield elements until it finishes.
333-- >>> :set -XOverloadedStrings
334-- >>> import Data.Functor.Identity (Identity)
335-- >>> Stream.interleaveSuffix "abc" ",,,," :: Stream.SerialT Identity Char
337-- >>> Stream.interleaveSuffix "abc" "," :: Stream.SerialT Identity Char
340-- 'interleaveSuffix' is a dual of 'interleaveInfix'.
342-- Do not use at scale in concatMapWith.
345{-# INLINE interleaveSuffix #-}
346interleaveSuffix ::(IsStream t, Monad m) => t m b -> t m b -> t m b
347interleaveSuffix m1 m2 =
348 fromStreamD $ D.interleaveFstSuffix (toStreamD m1) (toStreamD m2)
350-- | Interleaves the outputs of two streams, yielding elements from each stream
351-- alternately, starting from the first stream and ending at the first stream.
352-- If the second stream is longer than the first, elements from the second
353-- stream are infixed with elements from the first stream. If the first stream
354-- is longer then it continues yielding elements even after the second stream
357-- >>> :set -XOverloadedStrings
358-- >>> import Data.Functor.Identity (Identity)
359-- >>> Stream.interleaveInfix "abc" ",,,," :: Stream.SerialT Identity Char
361-- >>> Stream.interleaveInfix "abc" "," :: Stream.SerialT Identity Char
364-- 'interleaveInfix' is a dual of 'interleaveSuffix'.
366-- Do not use at scale in concatMapWith.
369{-# INLINE interleaveInfix #-}
370interleaveInfix ::(IsStream t, Monad m) => t m b -> t m b -> t m b
371interleaveInfix m1 m2 =
372 fromStreamD $ D.interleaveFst (toStreamD m1) (toStreamD m2)
374-- | Interleaves the outputs of two streams, yielding elements from each stream
375-- alternately, starting from the first stream. The output stops as soon as any
376-- of the two streams finishes, discarding the remaining part of the other
377-- stream. The last element of the resulting stream would be from the longer
380-- >>> :set -XOverloadedStrings
381-- >>> import Data.Functor.Identity (Identity)
382-- >>> Stream.interleaveMin "ab" ",,,," :: Stream.SerialT Identity Char
384-- >>> Stream.interleaveMin "abcd" ",," :: Stream.SerialT Identity Char
387-- 'interleaveMin' is dual to 'interleave'.
389-- Do not use at scale in concatMapWith.
392{-# INLINE interleaveMin #-}
393interleaveMin ::(IsStream t, Monad m) => t m b -> t m b -> t m b
394interleaveMin m1 m2 = fromStreamD $ D.interleaveMin (toStreamD m1) (toStreamD m2)
396-------------------------------------------------------------------------------
398-------------------------------------------------------------------------------
400-- | Schedule the execution of two streams in a fair round-robin manner,
401-- executing each stream once, alternately. Execution of a stream may not
402-- necessarily result in an output, a stream may chose to @Skip@ producing an
403-- element until later giving the other stream a chance to run. Therefore, this
404-- combinator fairly interleaves the execution of two streams rather than
405-- fairly interleaving the output of the two streams. This can be useful in
406-- co-operative multitasking without using explicit threads. This can be used
407-- as an alternative to `async`.
409-- Do not use at scale in concatMapWith.
412{-# INLINE roundrobin #-}
413roundrobin ::(IsStream t, Monad m) => t m b -> t m b -> t m b
414roundrobin m1 m2 = fromStreamD $ D.roundRobin (toStreamD m1) (toStreamD m2)
418-- | Merges two streams, both the streams may be evaluated concurrently,
419-- outputs from both are used as they arrive:
421-- >>> import Streamly.Prelude (async)
422-- >>> stream1 = Stream.fromEffect (delay 4)
423-- >>> stream2 = Stream.fromEffect (delay 2)
424-- >>> Stream.toList $ stream1 `async` stream2
429-- Multiple streams can be combined. With enough threads, all of them can be
430-- scheduled simultaneously:
432-- >>> stream3 = Stream.fromEffect (delay 1)
433-- >>> Stream.toList $ stream1 `async` stream2 `async` stream3
437-- With 2 threads, only two can be scheduled at a time, when one of those
438-- finishes, the third one gets scheduled:
440-- >>> Stream.toList $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3
444-- With a single thread, it becomes serial:
446-- >>> Stream.toList $ Stream.maxThreads 1 $ stream1 `async` stream2 `async` stream3
450-- Only streams are scheduled for async evaluation, how actions within a
451-- stream are evaluated depends on the stream type. If it is a concurrent
452-- stream they will be evaluated concurrently.
454-- In the following example, both the streams are scheduled for concurrent
455-- evaluation but each individual stream is evaluated serially:
457-- >>> stream1 = Stream.fromListM $ Prelude.map delay [3,3] -- SerialT IO Int
458-- >>> stream2 = Stream.fromListM $ Prelude.map delay [1,1] -- SerialT IO Int
459-- >>> Stream.toList $ stream1 `async` stream2 -- IO [Int]
463-- If total threads are 2, the third stream is scheduled only after one of the
464-- first two has finished:
466-- > stream3 = Stream.fromListM $ Prelude.map delay [2,2] -- SerialT IO Int
467-- > Stream.toList $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3 -- IO [Int]
471-- Thus 'async' goes deep in first few streams rather than going wide in all
472-- streams. It prefers to evaluate the leftmost streams as much as possible.
473-- Because of this behavior, 'async' can be safely used to fold an infinite
474-- lazy container of streams.
476-- /Since: 0.2.0 ("Streamly")/
480async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
481async m1 m2 = fromStream $ asyncK (toStream m1) (toStream m2)
485-- | For singleton streams, 'wAsync' is the same as 'async'. See 'async' for
486-- singleton stream behavior. For multi-element streams, while 'async' is left
487-- biased i.e. it tries to evaluate the left side stream as much as possible,
488-- 'wAsync' tries to schedule them both fairly. In other words, 'async' goes
489-- deep while 'wAsync' goes wide. However, outputs are always used as they
492-- With a single thread, 'async' starts behaving like 'serial' while 'wAsync'
493-- starts behaving like 'wSerial'.
495-- >>> import Streamly.Prelude (async, wAsync)
496-- >>> stream1 = Stream.fromList [1,2,3]
497-- >>> stream2 = Stream.fromList [4,5,6]
498-- >>> Stream.toList $ Stream.fromAsync $ Stream.maxThreads 1 $ stream1 `async` stream2
501-- >>> Stream.toList $ Stream.fromWAsync $ Stream.maxThreads 1 $ stream1 `wAsync` stream2
504-- With two threads available, and combining three streams:
506-- >>> stream3 = Stream.fromList [7,8,9]
507-- >>> Stream.toList $ Stream.fromAsync $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3
508-- [1,2,3,4,5,6,7,8,9]
510-- >>> Stream.toList $ Stream.fromWAsync $ Stream.maxThreads 2 $ stream1 `wAsync` stream2 `wAsync` stream3
511-- [1,4,2,7,5,3,8,6,9]
513-- This operation cannot be used to fold an infinite lazy container of streams,
514-- because it schedules all the streams in a round robin manner.
516-- Note that 'WSerialT' and single threaded 'WAsyncT' both interleave streams
517-- but the exact scheduling is slightly different in both cases.
521-- /Since: 0.2.0 ("Streamly")/
523-- Scheduling details:
525-- This is how the execution of the above example proceeds:
527-- 1. The scheduler queue is initialized with @[S.fromList [1,2,3],
528-- (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])]@ assuming the head of the
529-- queue is represented by the rightmost item.
530-- 2. @S.fromList [1,2,3]@ is executed, yielding the element @1@ and putting
531-- @[2,3]@ at the back of the scheduler queue. The scheduler queue now looks
532-- like @[(S.fromList [4,5,6]) \<> (S.fromList [7,8,9]), S.fromList [2,3]]@.
533-- 3. Now @(S.fromList [4,5,6]) \<> (S.fromList [7,8,9])@ is picked up for
534-- execution, @S.fromList [7,8,9]@ is added at the back of the queue and
535-- @S.fromList [4,5,6]@ is executed, yielding the element @4@ and adding
536-- @S.fromList [5,6]@ at the back of the queue. The queue now looks like
537-- @[S.fromList [2,3], S.fromList [7,8,9], S.fromList [5,6]]@.
538-- 4. Note that the scheduler queue expands by one more stream component in
539-- every pass because one more @<>@ is broken down into two components. At this
540-- point there are no more @<>@ operations to be broken down further and the
541-- queue has reached its maximum size. Now these streams are scheduled in
542-- round-robin fashion yielding @[2,7,5,3,8,6,9]@.
544-- As we see above, in a right associated expression composed with @<>@, only
545-- one @<>@ operation is broken down into two components in one execution,
546-- therefore, if we have @n@ streams composed using @<>@ it will take @n@
547-- scheduler passes to expand the whole expression. By the time @n-th@
548-- component is added to the scheduler queue, the first component would have
549-- received @n@ scheduler passes.
552wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
553wAsync m1 m2 = fromStream $ wAsyncK (toStream m1) (toStream m2)
557-- | Appends two streams, both the streams may be evaluated concurrently but
558-- the outputs are used in the same order as the corresponding actions in the
559-- original streams, side effects will happen in the order in which the streams
562-- >>> import Streamly.Prelude (ahead, SerialT)
563-- >>> stream1 = Stream.fromEffect (delay 4) :: SerialT IO Int
564-- >>> stream2 = Stream.fromEffect (delay 2) :: SerialT IO Int
565-- >>> Stream.toList $ stream1 `ahead` stream2 :: IO [Int]
570-- Multiple streams can be combined. With enough threads, all of them can be
571-- scheduled simultaneously:
573-- >>> stream3 = Stream.fromEffect (delay 1)
574-- >>> Stream.toList $ stream1 `ahead` stream2 `ahead` stream3
580-- With 2 threads, only two can be scheduled at a time, when one of those
581-- finishes, the third one gets scheduled:
583-- >>> Stream.toList $ Stream.maxThreads 2 $ stream1 `ahead` stream2 `ahead` stream3
589-- Only streams are scheduled for ahead evaluation, how actions within a stream
590-- are evaluated depends on the stream type. If it is a concurrent stream they
591-- will be evaluated concurrently. It may not make much sense combining serial
592-- streams using 'ahead'.
594-- 'ahead' can be safely used to fold an infinite lazy container of streams.
596-- /Since: 0.3.0 ("Streamly")/
600ahead :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
601ahead m1 m2 = fromStream $ aheadK (toStream m1) (toStream m2)
605-- | Like 'Streamly.Prelude.async' except that the execution is much more
606-- strict. There is no limit on the number of threads. While
607-- 'Streamly.Prelude.async' may not schedule a stream if there is no demand
608-- from the consumer, 'parallel' always evaluates both the streams immediately.
609-- The only limit that applies to 'parallel' is 'Streamly.Prelude.maxBuffer'.
610-- Evaluation may block if the output buffer becomes full.
612-- >>> import Streamly.Prelude (parallel)
613-- >>> stream = Stream.fromEffect (delay 2) `parallel` Stream.fromEffect (delay 1)
614-- >>> Stream.toList stream -- IO [Int]
619-- 'parallel' guarantees that all the streams are scheduled for execution
620-- immediately, therefore, we could use things like starting timers inside the
621-- streams and relying on the fact that all timers were started at the same
624-- Unlike 'async' this operation cannot be used to fold an infinite lazy
625-- container of streams, because it schedules all the streams strictly
628-- /Since: 0.2.0 ("Streamly")/
631{-# INLINE parallel #-}
632parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
633parallel m1 m2 = fromStream $ Par.parallelK (toStream m1) (toStream m2)
635-- This is a race like combinator for streams.
637-- | Like `parallel` but stops the output as soon as any of the two streams
641{-# INLINE parallelMin #-}
642parallelMin :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
643parallelMin m1 m2 = fromStream $ Par.parallelMinK (toStream m1) (toStream m2)
645------------------------------------------------------------------------------
647------------------------------------------------------------------------------
649-- | Like 'zipAsyncWith' but with a monadic zipping function.
652{-# INLINE zipAsyncWithM #-}
653zipAsyncWithM :: (IsStream t, MonadAsync m)
654 => (a -> b -> m c) -> t m a -> t m b -> t m c
655zipAsyncWithM f m1 m2 =
656 fromStream $ ZipAsync.zipAsyncWithMK f (toStream m1) (toStream m2)
658-- XXX Should we rename this to zipParWith or zipParallelWith? This can happen
659-- along with the change of behvaior to end the stream concurrently.
661-- | Like 'zipWith' but zips concurrently i.e. both the streams being zipped
662-- are evaluated concurrently using the 'ParallelT' concurrent evaluation
663-- style. The maximum number of elements of each stream evaluated in advance
664-- can be controlled by 'maxBuffer'.
666-- The stream ends if stream @a@ or stream @b@ ends. However, if stream @b@
667-- ends while we are still evaluating stream @a@ and waiting for a result then
668-- stream will not end until after the evaluation of stream @a@ finishes. This
669-- behavior can potentially be changed in future to end the stream immediately
670-- as soon as any of the stream end is detected.
673{-# INLINE zipAsyncWith #-}
674zipAsyncWith :: (IsStream t, MonadAsync m)
675 => (a -> b -> c) -> t m a -> t m b -> t m c
676zipAsyncWith f = zipAsyncWithM (\a b -> return (f a b))
678------------------------------------------------------------------------------
679-- Merging (sorted streams)
680------------------------------------------------------------------------------
682-- | Merge two streams using a comparison function. The head elements of both
683-- the streams are compared and the smaller of the two elements is emitted, if
684-- both elements are equal then the element from the first stream is used
687-- If the streams are sorted in ascending order, the resulting stream would
688-- also remain sorted in ascending order.
691-- >>> Stream.toList $ Stream.mergeBy compare (Stream.fromList [1,3,5]) (Stream.fromList [2,4,6,8])
696-- See also: 'mergeByMFused'
699{-# INLINE mergeBy #-}
700mergeBy :: IsStream t => (a -> a -> Ordering) -> t m a -> t m a -> t m a
701mergeBy f m1 m2 = fromStream $ K.mergeBy f (toStream m1) (toStream m2)
703-- | Like 'mergeBy' but with a monadic comparison function.
705-- Merge two streams randomly:
708-- > randomly _ _ = randomIO >>= \x -> return $ if x then LT else GT
709-- > Stream.toList $ Stream.mergeByM randomly (Stream.fromList [1,1,1,1]) (Stream.fromList [2,2,2,2])
713-- Merge two streams in a proportion of 2:1:
718-- let proportionately m n = do
719-- ref <- newIORef $ cycle $ Prelude.concat [Prelude.replicate m LT, Prelude.replicate n GT]
720-- return $ \_ _ -> do
722-- writeIORef ref $ Prelude.tail r
723-- return $ Prelude.head r
724-- f <- proportionately 2 1
725-- xs <- Stream.toList $ Stream.mergeByM f (Stream.fromList [1,1,1,1,1,1]) (Stream.fromList [2,2,2])
728-- [1,1,2,1,1,2,1,1,2]
732-- See also: 'mergeByMFused'
735{-# INLINE mergeByM #-}
737 :: (IsStream t, Monad m)
738 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
739mergeByM f m1 m2 = fromStream $ K.mergeByM f (toStream m1) (toStream m2)
741-- XXX Fused versions should probably go to a separate module using the same
742-- names for the combinators.
744-- | Like 'mergeByM' but much faster, works best when merging statically known
745-- number of streams. When merging more than two streams try to merge pairs and
746-- pair pf pairs in a tree like structure.'mergeByM' works better with variable
747-- number of streams being merged using 'concatPairsWith'.
750{-# INLINE mergeByMFused #-}
752 :: (IsStream t, Monad m)
753 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
754mergeByMFused f m1 m2 =
755 fromStreamD $ D.mergeByM f (toStreamD m1) (toStreamD m2)
757-- | Like 'mergeByM' but stops merging as soon as any of the two streams stops.
760{-# INLINABLE mergeMinBy #-}
761mergeMinBy :: -- (IsStream t, Monad m) =>
762 (a -> a -> m Ordering) -> t m a -> t m a -> t m a
763mergeMinBy _f _m1 _m2 = undefined
764 -- fromStreamD $ D.mergeMinBy f (toStreamD m1) (toStreamD m2)
766-- | Like 'mergeByM' but stops merging as soon as the first stream stops.
769{-# INLINABLE mergeFstBy #-}
770mergeFstBy :: -- (IsStream t, Monad m) =>
771 (a -> a -> m Ordering) -> t m a -> t m a -> t m a
772mergeFstBy _f _m1 _m2 = undefined
773 -- fromStreamD $ D.mergeFstBy f (toStreamD m1) (toStreamD m2)
775-- XXX we may want to use the name "merge" differently
776-- | Same as @'mergeBy' 'compare'@.
778-- >>> Stream.toList $ Stream.merge (Stream.fromList [1,3,5]) (Stream.fromList [2,4,6,8])
783{-# INLINABLE merge #-}
785 (IsStream t, Ord a) => t m a -> t m a -> t m a
786merge = mergeBy compare
788-- | Like 'mergeBy' but merges concurrently (i.e. both the elements being
789-- merged are generated concurrently).
792{-# INLINE mergeAsyncBy #-}
793mergeAsyncBy :: (IsStream t, MonadAsync m)
794 => (a -> a -> Ordering) -> t m a -> t m a -> t m a
795mergeAsyncBy f = mergeAsyncByM (\a b -> return $ f a b)
797-- | Like 'mergeByM' but merges concurrently (i.e. both the elements being
798-- merged are generated concurrently).
801{-# INLINE mergeAsyncByM #-}
802mergeAsyncByM :: (IsStream t, MonadAsync m)
803 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
804mergeAsyncByM f m1 m2 =
806 let par = Par.mkParallelD . toStreamD
807 in D.mergeByM f (par m1) (par m2)
811{-# DEPRECATED concatUnfold "Please use unfoldMany instead." #-}
812{-# INLINE concatUnfold #-}
813concatUnfold ::(IsStream t, Monad m) => Unfold m a b -> t m a -> t m b
814concatUnfold u m = fromStreamD $ D.unfoldMany u (toStreamD m)
816------------------------------------------------------------------------------
817-- Combine N Streams - unfoldMany
818------------------------------------------------------------------------------
820-- | Like 'concatMap' but uses an 'Unfold' for stream generation. Unlike
821-- 'concatMap' this can fuse the 'Unfold' code with the inner loop and
822-- therefore provide many times better performance.
825{-# INLINE unfoldMany #-}
826unfoldMany ::(IsStream t, Monad m) => Unfold m a b -> t m a -> t m b
827unfoldMany u m = fromStreamD $ D.unfoldMany u (toStreamD m)
829-- | Like 'unfoldMany' but interleaves the streams in the same way as
830-- 'interleave' behaves instead of appending them.
833{-# INLINE unfoldManyInterleave #-}
834unfoldManyInterleave ::(IsStream t, Monad m)
835 => Unfold m a b -> t m a -> t m b
836unfoldManyInterleave u m =
837 fromStreamD $ D.unfoldInterleave u (toStreamD m)
839-- | Like 'unfoldMany' but executes the streams in the same way as
843{-# INLINE unfoldManyRoundRobin #-}
844unfoldManyRoundRobin ::(IsStream t, Monad m)
845 => Unfold m a b -> t m a -> t m b
846unfoldManyRoundRobin u m =
847 fromStreamD $ D.unfoldRoundRobin u (toStreamD m)
849------------------------------------------------------------------------------
850-- Combine N Streams - interpose
851------------------------------------------------------------------------------
853-- > interpose x unf str = gintercalate unf str UF.identity (repeat x)
855-- | Unfold the elements of a stream, intersperse the given element between the
856-- unfolded streams and then concat them into a single stream.
858-- > unwords = S.interpose ' '
861{-# INLINE interpose #-}
862interpose :: (IsStream t, Monad m)
863 => c -> Unfold m b c -> t m b -> t m c
865 fromStreamD $ D.interpose x unf (toStreamD str)
867-- interposeSuffix x unf str = gintercalateSuffix unf str UF.identity (repeat x)
869-- | Unfold the elements of a stream, append the given element after each
870-- unfolded stream and then concat them into a single stream.
872-- > unlines = S.interposeSuffix '\n'
875{-# INLINE interposeSuffix #-}
876interposeSuffix :: (IsStream t, Monad m)
877 => c -> Unfold m b c -> t m b -> t m c
878interposeSuffix x unf str =
879 fromStreamD $ D.interposeSuffix x unf (toStreamD str)
881------------------------------------------------------------------------------
882-- Combine N Streams - intercalate
883------------------------------------------------------------------------------
885-- XXX we can swap the order of arguments to gintercalate so that the
886-- definition of unfoldMany becomes simpler? The first stream should be
887-- infixed inside the second one. However, if we change the order in
888-- "interleave" as well similarly, then that will make it a bit unintuitive.
890-- > unfoldMany unf str =
891-- > gintercalate unf str (UF.nilM (\_ -> return ())) (repeat ())
893-- | 'interleaveInfix' followed by unfold and concat.
896{-# INLINE gintercalate #-}
898 :: (IsStream t, Monad m)
899 => Unfold m a c -> t m a -> Unfold m b c -> t m b -> t m c
900gintercalate unf1 str1 unf2 str2 =
901 fromStreamD $ D.gintercalate
902 unf1 (toStreamD str1)
903 unf2 (toStreamD str2)
905-- > intercalate unf seed str = gintercalate unf str unf (repeatM seed)
907-- | 'intersperse' followed by unfold and concat.
909-- > intercalate unf a str = unfoldMany unf $ intersperse a str
910-- > intersperse = intercalate (Unfold.function id)
911-- > unwords = intercalate Unfold.fromList " "
913-- >>> Stream.toList $ Stream.intercalate Unfold.fromList " " $ Stream.fromList ["abc", "def", "ghi"]
917{-# INLINE intercalate #-}
918intercalate :: (IsStream t, Monad m)
919 => Unfold m b c -> b -> t m b -> t m c
920intercalate unf seed str = fromStreamD $
921 D.unfoldMany unf $ D.intersperse seed (toStreamD str)
923-- | 'interleaveSuffix' followed by unfold and concat.
926{-# INLINE gintercalateSuffix #-}
928 :: (IsStream t, Monad m)
929 => Unfold m a c -> t m a -> Unfold m b c -> t m b -> t m c
930gintercalateSuffix unf1 str1 unf2 str2 =
931 fromStreamD $ D.gintercalateSuffix
932 unf1 (toStreamD str1)
933 unf2 (toStreamD str2)
935-- > intercalateSuffix unf seed str = gintercalateSuffix unf str unf (repeatM seed)
937-- | 'intersperseMSuffix' followed by unfold and concat.
939-- > intercalateSuffix unf a str = unfoldMany unf $ intersperseMSuffix a str
940-- > intersperseMSuffix = intercalateSuffix (Unfold.function id)
941-- > unlines = intercalateSuffix Unfold.fromList "\n"
943-- >>> Stream.toList $ Stream.intercalateSuffix Unfold.fromList "\n" $ Stream.fromList ["abc", "def", "ghi"]
947{-# INLINE intercalateSuffix #-}
948intercalateSuffix :: (IsStream t, Monad m)
949 => Unfold m b c -> b -> t m b -> t m c
950intercalateSuffix unf seed str = fromStreamD $ D.unfoldMany unf
951 $ D.intersperseMSuffix (return seed) (toStreamD str)
953------------------------------------------------------------------------------
954-- Combine N Streams - concatMap
955------------------------------------------------------------------------------
957-- | Flatten a stream of streams to a single stream.
960-- concat = concatMap id
965concat :: (IsStream t, Monad m) => t m (t m a) -> t m a
968------------------------------------------------------------------------------
969-- Combine N Streams - concatMap
970------------------------------------------------------------------------------
972-- | Like 'concatMapWith' but carries a state which can be used to share
973-- information across multiple steps of concat.
976-- concatSmapMWith combine f initial = concatMapWith combine id . smapM f initial
981{-# INLINE concatSmapMWith #-}
983 :: (IsStream t, Monad m)
984 => (t m b -> t m b -> t m b)
985 -> (s -> a -> m (s, t m b))
989concatSmapMWith combine f initial =
990 IsStream.concatMapWith combine id . smapM f initial
992-- XXX Implement a StreamD version for fusion.
994-- | Combine streams in pairs using a binary stream combinator, then combine
995-- the resulting streams in pairs recursively until we get to a single combined
998-- For example, you can sort a stream using merge sort like this:
1000-- >>> Stream.toList $ Stream.concatPairsWith (Stream.mergeBy compare) Stream.fromPure $ Stream.fromList [5,1,7,9,2]
1003-- /Caution: the stream of streams must be finite/
1007{-# INLINE concatPairsWith #-}
1008concatPairsWith :: IsStream t =>
1009 (t m b -> t m b -> t m b)
1013concatPairsWith par f m =
1016 (\s1 s2 -> toStream $ fromStream s1 `par` fromStream s2)
1020------------------------------------------------------------------------------
1021-- IterateMap - Map and flatten Trees of Streams
1022------------------------------------------------------------------------------
1024-- | Like 'iterateM' but iterates after mapping a stream generator on the
1027-- Yield an input element in the output stream, map a stream generator on it
1028-- and then do the same on the resulting stream. This can be used for a depth
1029-- first traversal of a tree like structure.
1031-- Note that 'iterateM' is a special case of 'iterateMapWith':
1034-- iterateM f = iterateMapWith serial (fromEffect . f) . fromEffect
1037-- It can be used to traverse a tree structure. For example, to list a
1041-- Stream.iterateMapWith Stream.serial
1042-- (either Dir.toEither (const nil))
1043-- (fromPure (Left "tmp"))
1048{-# INLINE iterateMapWith #-}
1051 => (t m a -> t m a -> t m a)
1055iterateMapWith combine f = IsStream.concatMapWith combine go
1057 go x = fromPure x `combine` IsStream.concatMapWith combine go (f x)
1059-- | Same as @iterateMapWith Stream.serial@ but more efficient due to stream
1063{-# INLINE iterateUnfold #-}
1064iterateUnfold :: -- (IsStream t, MonadAsync m) =>
1065 Unfold m a a -> t m a -> t m a
1066iterateUnfold = undefined
1068------------------------------------------------------------------------------
1070------------------------------------------------------------------------------
1072-- To traverse graphs we need a state to be carried around in the traversal.
1073-- For example, we can use a hashmap to store the visited status of nodes.
1075-- | Like 'iterateMap' but carries a state in the stream generation function.
1076-- This can be used to traverse graph like structures, we can remember the
1077-- visited nodes in the state to avoid cycles.
1079-- Note that a combination of 'iterateMap' and 'usingState' can also be used to
1080-- traverse graphs. However, this function provides a more localized state
1081-- instead of using a global state.
1087{-# INLINE iterateSmapMWith #-}
1089 :: (IsStream t, Monad m)
1090 => (t m a -> t m a -> t m a)
1091 -> (b -> a -> m (b, t m a))
1095iterateSmapMWith combine f initial stream =
1097 (\b -> IsStream.concatMapWith combine (go b) stream)
1098 (fromEffect initial)
1102 go b a = fromPure a `combine` feedback b a
1106 (\(b1, s) -> IsStream.concatMapWith combine (go b1) s)
1107 (fromEffect $ f b a)
1109------------------------------------------------------------------------------
1111------------------------------------------------------------------------------
1113-- Keep concating either streams as long as rights are generated, stop as soon
1114-- as a left is generated and concat the left stream.
1116-- See also: 'handle'
1122 :: -- (IsStream t, MonadAsync m) =>
1123 (forall x. t m x -> t m x -> t m x)
1124 -> (a -> t m (Either (t m b) b))
1127concatMapEitherWith = undefined
1130-- | In an 'Either' stream iterate on 'Left's. This is a special case of
1134-- iterateMapLeftsWith combine f = iterateMapWith combine (either f (const nil))
1137-- To traverse a directory tree:
1140-- iterateMapLeftsWith serial Dir.toEither (fromPure (Left "tmp"))
1145{-# INLINE iterateMapLeftsWith #-}
1147 :: (IsStream t, b ~ Either a c)
1148 => (t m b -> t m b -> t m b)
1152iterateMapLeftsWith combine f =
1153 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