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.StreamD 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.Stream.StreamK as K (mergeBy, mergeByM)
187import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
188 (interleave, append, mergeMapWith)
189import qualified Streamly.Internal.Data.Stream.ZipAsync as ZipAsync
191import Prelude hiding (concat, concatMap, zipWith)
195-- >>> :set -fno-warn-deprecations
196-- >>> import Control.Concurrent (threadDelay)
197-- >>> import Data.IORef
198-- >>> import Prelude hiding (zipWith, concatMap, concat)
199-- >>> import qualified Streamly.Prelude as Stream
200-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream
201-- >>> import qualified Streamly.Data.Fold as Fold
202-- >>> import qualified Streamly.Internal.Data.Fold as Fold
203-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
204-- >>> import qualified Streamly.Internal.Data.Parser as Parser
205-- >>> import qualified Streamly.Data.Array as Array
208-- threadDelay (n * 1000000) -- sleep for n seconds
209-- putStrLn (show n ++ " sec") -- print "n sec"
214-------------------------------------------------------------------------------
216-------------------------------------------------------------------------------
218-- XXX Reconcile the names "serial" and "append".
220-- | Append the outputs of two streams, yielding all the elements from the
221-- first stream and then yielding all the elements from the second stream.
223-- IMPORTANT NOTE: This could be 100x faster than @serial/<>@ for appending a
224-- few (say 100) streams because it can fuse via stream fusion. However, it
225-- does not scale for a large number of streams (say 1000s) and becomes
226-- qudartically slow. Therefore use this for custom appending of a few streams
227-- but use 'concatMap' or 'concatMapWith serial' for appending @n@ streams or
228-- infinite containers of streams.
232append ::(IsStream t, Monad m) => t m b -> t m b -> t m b
233append m1 m2 = fromStreamD $ D.append (toStreamD m1) (toStreamD m2)
237-- | Appends two streams sequentially, yielding all elements from the first
238-- stream, and then all elements from the second stream.
240-- >>> import Streamly.Prelude (serial)
241-- >>> stream1 = Stream.fromList [1,2]
242-- >>> stream2 = Stream.fromList [3,4]
243-- >>> Stream.toList $ stream1 `serial` stream2
246-- This operation can be used to fold an infinite lazy container of streams.
248-- /Since: 0.2.0 ("Streamly")/
252serial :: IsStream t => t m a -> t m a -> t m a
253serial m1 m2 = fromStream $ K.append (toStream m1) (toStream m2)
255-------------------------------------------------------------------------------
257-------------------------------------------------------------------------------
261-- XXX doc duplicated from Stream.Serial module.
263-- | Interleaves two streams, yielding one element from each stream
264-- alternately. When one stream stops the rest of the other stream is used in
267-- >>> import Streamly.Prelude (wSerial)
268-- >>> stream1 = Stream.fromList [1,2]
269-- >>> stream2 = Stream.fromList [3,4]
270-- >>> Stream.toList $ Stream.fromWSerial $ stream1 `wSerial` stream2
273-- Note, for singleton streams 'wSerial' and 'serial' are identical.
275-- Note that this operation cannot be used to fold a container of infinite
276-- streams but it can be used for very large streams as the state that it needs
277-- to maintain is proportional to the logarithm of the number of streams.
281-- /Since: 0.2.0 ("Streamly")/
285-- Note that evaluation of @a \`wSerial` b \`wSerial` c@ does not interleave
286-- @a@, @b@ and @c@ with equal priority. This expression is equivalent to @a
287-- \`wSerial` (b \`wSerial` c)@, therefore, it fairly interleaves @a@ with the
288-- result of @b \`wSerial` c@. For example, @Stream.fromList [1,2] \`wSerial`
289-- Stream.fromList [3,4] \`wSerial` Stream.fromList [5,6]@ would result in
290-- [1,3,2,5,4,6]. In other words, the leftmost stream gets the same scheduling
291-- priority as the rest of the streams taken together. The same is true for
292-- each subexpression on the right.
294{-# INLINE wSerial #-}
295wSerial :: IsStream t => t m a -> t m a -> t m a
296wSerial m1 m2 = fromStream $ K.interleave (toStream m1) (toStream m2)
298-- XXX Same as 'wSerial'. We should perhaps rename wSerial to interleave.
299-- XXX Document the interleaving behavior of side effects in all the
300-- interleaving combinators.
301-- XXX Write time-domain equivalents of these. In the time domain we can
302-- interleave two streams such that the value of second stream is always taken
303-- from its last value even if no new value is being yielded, like
304-- zipWithLatest. It would be something like interleaveWithLatest.
306-- | Interleaves the outputs of two streams, yielding elements from each stream
307-- alternately, starting from the first stream. If any of the streams finishes
308-- early the other stream continues alone until it too finishes.
310-- >>> :set -XOverloadedStrings
311-- >>> import Data.Functor.Identity (Identity)
312-- >>> Stream.interleave "ab" ",,,," :: Stream.SerialT Identity Char
315-- >>> Stream.interleave "abcd" ",," :: Stream.SerialT Identity Char
318-- 'interleave' is dual to 'interleaveMin', it can be called @interleaveMax@.
320-- Do not use at scale in concatMapWith.
323{-# INLINE interleave #-}
324interleave ::(IsStream t, Monad m) => t m b -> t m b -> t m b
325interleave m1 m2 = fromStreamD $ D.interleave (toStreamD m1) (toStreamD m2)
327-- | Interleaves the outputs of two streams, yielding elements from each stream
328-- alternately, starting from the first stream. As soon as the first stream
329-- finishes, the output stops, discarding the remaining part of the second
330-- stream. In this case, the last element in the resulting stream would be from
331-- the second stream. If the second stream finishes early then the first stream
332-- still continues to yield elements until it finishes.
334-- >>> :set -XOverloadedStrings
335-- >>> import Data.Functor.Identity (Identity)
336-- >>> Stream.interleaveSuffix "abc" ",,,," :: Stream.SerialT Identity Char
338-- >>> Stream.interleaveSuffix "abc" "," :: Stream.SerialT Identity Char
341-- 'interleaveSuffix' is a dual of 'interleaveInfix'.
343-- Do not use at scale in concatMapWith.
346{-# INLINE interleaveSuffix #-}
347interleaveSuffix ::(IsStream t, Monad m) => t m b -> t m b -> t m b
348interleaveSuffix m1 m2 =
349 fromStreamD $ D.interleaveFstSuffix (toStreamD m1) (toStreamD m2)
351-- | Interleaves the outputs of two streams, yielding elements from each stream
352-- alternately, starting from the first stream and ending at the first stream.
353-- If the second stream is longer than the first, elements from the second
354-- stream are infixed with elements from the first stream. If the first stream
355-- is longer then it continues yielding elements even after the second stream
358-- >>> :set -XOverloadedStrings
359-- >>> import Data.Functor.Identity (Identity)
360-- >>> Stream.interleaveInfix "abc" ",,,," :: Stream.SerialT Identity Char
362-- >>> Stream.interleaveInfix "abc" "," :: Stream.SerialT Identity Char
365-- 'interleaveInfix' is a dual of 'interleaveSuffix'.
367-- Do not use at scale in concatMapWith.
370{-# INLINE interleaveInfix #-}
371interleaveInfix ::(IsStream t, Monad m) => t m b -> t m b -> t m b
372interleaveInfix m1 m2 =
373 fromStreamD $ D.interleaveFst (toStreamD m1) (toStreamD m2)
375-- | Interleaves the outputs of two streams, yielding elements from each stream
376-- alternately, starting from the first stream. The output stops as soon as any
377-- of the two streams finishes, discarding the remaining part of the other
378-- stream. The last element of the resulting stream would be from the longer
381-- >>> :set -XOverloadedStrings
382-- >>> import Data.Functor.Identity (Identity)
383-- >>> Stream.interleaveMin "ab" ",,,," :: Stream.SerialT Identity Char
385-- >>> Stream.interleaveMin "abcd" ",," :: Stream.SerialT Identity Char
388-- 'interleaveMin' is dual to 'interleave'.
390-- Do not use at scale in concatMapWith.
393{-# INLINE interleaveMin #-}
394interleaveMin ::(IsStream t, Monad m) => t m b -> t m b -> t m b
395interleaveMin m1 m2 = fromStreamD $ D.interleaveMin (toStreamD m1) (toStreamD m2)
397-------------------------------------------------------------------------------
399-------------------------------------------------------------------------------
401-- | Schedule the execution of two streams in a fair round-robin manner,
402-- executing each stream once, alternately. Execution of a stream may not
403-- necessarily result in an output, a stream may chose to @Skip@ producing an
404-- element until later giving the other stream a chance to run. Therefore, this
405-- combinator fairly interleaves the execution of two streams rather than
406-- fairly interleaving the output of the two streams. This can be useful in
407-- co-operative multitasking without using explicit threads. This can be used
408-- as an alternative to `async`.
410-- Do not use at scale in concatMapWith.
413{-# INLINE roundrobin #-}
414roundrobin ::(IsStream t, Monad m) => t m b -> t m b -> t m b
415roundrobin m1 m2 = fromStreamD $ D.roundRobin (toStreamD m1) (toStreamD m2)
419-- | Merges two streams, both the streams may be evaluated concurrently,
420-- outputs from both are used as they arrive:
422-- >>> import Streamly.Prelude (async)
423-- >>> stream1 = Stream.fromEffect (delay 4)
424-- >>> stream2 = Stream.fromEffect (delay 2)
425-- >>> Stream.toList $ stream1 `async` stream2
430-- Multiple streams can be combined. With enough threads, all of them can be
431-- scheduled simultaneously:
433-- >>> stream3 = Stream.fromEffect (delay 1)
434-- >>> Stream.toList $ stream1 `async` stream2 `async` stream3
438-- With 2 threads, only two can be scheduled at a time, when one of those
439-- finishes, the third one gets scheduled:
441-- >>> Stream.toList $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3
445-- With a single thread, it becomes serial:
447-- >>> Stream.toList $ Stream.maxThreads 1 $ stream1 `async` stream2 `async` stream3
451-- Only streams are scheduled for async evaluation, how actions within a
452-- stream are evaluated depends on the stream type. If it is a concurrent
453-- stream they will be evaluated concurrently.
455-- In the following example, both the streams are scheduled for concurrent
456-- evaluation but each individual stream is evaluated serially:
458-- >>> stream1 = Stream.fromListM $ Prelude.map delay [3,3] -- SerialT IO Int
459-- >>> stream2 = Stream.fromListM $ Prelude.map delay [1,1] -- SerialT IO Int
460-- >>> Stream.toList $ stream1 `async` stream2 -- IO [Int]
464-- If total threads are 2, the third stream is scheduled only after one of the
465-- first two has finished:
467-- > stream3 = Stream.fromListM $ Prelude.map delay [2,2] -- SerialT IO Int
468-- > Stream.toList $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3 -- IO [Int]
472-- Thus 'async' goes deep in first few streams rather than going wide in all
473-- streams. It prefers to evaluate the leftmost streams as much as possible.
474-- Because of this behavior, 'async' can be safely used to fold an infinite
475-- lazy container of streams.
477-- /Since: 0.2.0 ("Streamly")/
481async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
482async m1 m2 = fromStream $ asyncK (toStream m1) (toStream m2)
486-- | For singleton streams, 'wAsync' is the same as 'async'. See 'async' for
487-- singleton stream behavior. For multi-element streams, while 'async' is left
488-- biased i.e. it tries to evaluate the left side stream as much as possible,
489-- 'wAsync' tries to schedule them both fairly. In other words, 'async' goes
490-- deep while 'wAsync' goes wide. However, outputs are always used as they
493-- With a single thread, 'async' starts behaving like 'serial' while 'wAsync'
494-- starts behaving like 'wSerial'.
496-- >>> import Streamly.Prelude (async, wAsync)
497-- >>> stream1 = Stream.fromList [1,2,3]
498-- >>> stream2 = Stream.fromList [4,5,6]
499-- >>> Stream.toList $ Stream.fromAsync $ Stream.maxThreads 1 $ stream1 `async` stream2
502-- >>> Stream.toList $ Stream.fromWAsync $ Stream.maxThreads 1 $ stream1 `wAsync` stream2
505-- With two threads available, and combining three streams:
507-- >>> stream3 = Stream.fromList [7,8,9]
508-- >>> Stream.toList $ Stream.fromAsync $ Stream.maxThreads 2 $ stream1 `async` stream2 `async` stream3
509-- [1,2,3,4,5,6,7,8,9]
511-- >>> Stream.toList $ Stream.fromWAsync $ Stream.maxThreads 2 $ stream1 `wAsync` stream2 `wAsync` stream3
512-- [1,4,2,7,5,3,8,6,9]
514-- This operation cannot be used to fold an infinite lazy container of streams,
515-- because it schedules all the streams in a round robin manner.
517-- Note that 'WSerialT' and single threaded 'WAsyncT' both interleave streams
518-- but the exact scheduling is slightly different in both cases.
522-- /Since: 0.2.0 ("Streamly")/
524-- Scheduling details:
526-- This is how the execution of the above example proceeds:
528-- 1. The scheduler queue is initialized with @[S.fromList [1,2,3],
529-- (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])]@ assuming the head of the
530-- queue is represented by the rightmost item.
531-- 2. @S.fromList [1,2,3]@ is executed, yielding the element @1@ and putting
532-- @[2,3]@ at the back of the scheduler queue. The scheduler queue now looks
533-- like @[(S.fromList [4,5,6]) \<> (S.fromList [7,8,9]), S.fromList [2,3]]@.
534-- 3. Now @(S.fromList [4,5,6]) \<> (S.fromList [7,8,9])@ is picked up for
535-- execution, @S.fromList [7,8,9]@ is added at the back of the queue and
536-- @S.fromList [4,5,6]@ is executed, yielding the element @4@ and adding
537-- @S.fromList [5,6]@ at the back of the queue. The queue now looks like
538-- @[S.fromList [2,3], S.fromList [7,8,9], S.fromList [5,6]]@.
539-- 4. Note that the scheduler queue expands by one more stream component in
540-- every pass because one more @<>@ is broken down into two components. At this
541-- point there are no more @<>@ operations to be broken down further and the
542-- queue has reached its maximum size. Now these streams are scheduled in
543-- round-robin fashion yielding @[2,7,5,3,8,6,9]@.
545-- As we see above, in a right associated expression composed with @<>@, only
546-- one @<>@ operation is broken down into two components in one execution,
547-- therefore, if we have @n@ streams composed using @<>@ it will take @n@
548-- scheduler passes to expand the whole expression. By the time @n-th@
549-- component is added to the scheduler queue, the first component would have
550-- received @n@ scheduler passes.
553wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
554wAsync m1 m2 = fromStream $ wAsyncK (toStream m1) (toStream m2)
558-- | Appends two streams, both the streams may be evaluated concurrently but
559-- the outputs are used in the same order as the corresponding actions in the
560-- original streams, side effects will happen in the order in which the streams
563-- >>> import Streamly.Prelude (ahead, SerialT)
564-- >>> stream1 = Stream.fromEffect (delay 4) :: SerialT IO Int
565-- >>> stream2 = Stream.fromEffect (delay 2) :: SerialT IO Int
566-- >>> Stream.toList $ stream1 `ahead` stream2 :: IO [Int]
571-- Multiple streams can be combined. With enough threads, all of them can be
572-- scheduled simultaneously:
574-- >>> stream3 = Stream.fromEffect (delay 1)
575-- >>> Stream.toList $ stream1 `ahead` stream2 `ahead` stream3
581-- With 2 threads, only two can be scheduled at a time, when one of those
582-- finishes, the third one gets scheduled:
584-- >>> Stream.toList $ Stream.maxThreads 2 $ stream1 `ahead` stream2 `ahead` stream3
590-- Only streams are scheduled for ahead evaluation, how actions within a stream
591-- are evaluated depends on the stream type. If it is a concurrent stream they
592-- will be evaluated concurrently. It may not make much sense combining serial
593-- streams using 'ahead'.
595-- 'ahead' can be safely used to fold an infinite lazy container of streams.
597-- /Since: 0.3.0 ("Streamly")/
601ahead :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
602ahead m1 m2 = fromStream $ aheadK (toStream m1) (toStream m2)
606-- | Like 'Streamly.Prelude.async' except that the execution is much more
607-- strict. There is no limit on the number of threads. While
608-- 'Streamly.Prelude.async' may not schedule a stream if there is no demand
609-- from the consumer, 'parallel' always evaluates both the streams immediately.
610-- The only limit that applies to 'parallel' is 'Streamly.Prelude.maxBuffer'.
611-- Evaluation may block if the output buffer becomes full.
613-- >>> import Streamly.Prelude (parallel)
614-- >>> stream = Stream.fromEffect (delay 2) `parallel` Stream.fromEffect (delay 1)
615-- >>> Stream.toList stream -- IO [Int]
620-- 'parallel' guarantees that all the streams are scheduled for execution
621-- immediately, therefore, we could use things like starting timers inside the
622-- streams and relying on the fact that all timers were started at the same
625-- Unlike 'async' this operation cannot be used to fold an infinite lazy
626-- container of streams, because it schedules all the streams strictly
629-- /Since: 0.2.0 ("Streamly")/
632{-# INLINE parallel #-}
633parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
634parallel m1 m2 = fromStream $ Par.parallelK (toStream m1) (toStream m2)
636-- This is a race like combinator for streams.
638-- | Like `parallel` but stops the output as soon as any of the two streams
642{-# INLINE parallelMin #-}
643parallelMin :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
644parallelMin m1 m2 = fromStream $ Par.parallelMinK (toStream m1) (toStream m2)
646------------------------------------------------------------------------------
648------------------------------------------------------------------------------
650-- | Like 'zipAsyncWith' but with a monadic zipping function.
653{-# INLINE zipAsyncWithM #-}
654zipAsyncWithM :: (IsStream t, MonadAsync m)
655 => (a -> b -> m c) -> t m a -> t m b -> t m c
656zipAsyncWithM f m1 m2 =
657 fromStream $ ZipAsync.zipAsyncWithMK f (toStream m1) (toStream m2)
659-- XXX Should we rename this to zipParWith or zipParallelWith? This can happen
660-- along with the change of behvaior to end the stream concurrently.
662-- | Like 'zipWith' but zips concurrently i.e. both the streams being zipped
663-- are evaluated concurrently using the 'ParallelT' concurrent evaluation
664-- style. The maximum number of elements of each stream evaluated in advance
665-- can be controlled by 'maxBuffer'.
667-- The stream ends if stream @a@ or stream @b@ ends. However, if stream @b@
668-- ends while we are still evaluating stream @a@ and waiting for a result then
669-- stream will not end until after the evaluation of stream @a@ finishes. This
670-- behavior can potentially be changed in future to end the stream immediately
671-- as soon as any of the stream end is detected.
674{-# INLINE zipAsyncWith #-}
675zipAsyncWith :: (IsStream t, MonadAsync m)
676 => (a -> b -> c) -> t m a -> t m b -> t m c
677zipAsyncWith f = zipAsyncWithM (\a b -> return (f a b))
679------------------------------------------------------------------------------
680-- Merging (sorted streams)
681------------------------------------------------------------------------------
683-- | Merge two streams using a comparison function. The head elements of both
684-- the streams are compared and the smaller of the two elements is emitted, if
685-- both elements are equal then the element from the first stream is used
688-- If the streams are sorted in ascending order, the resulting stream would
689-- also remain sorted in ascending order.
692-- >>> Stream.toList $ Stream.mergeBy compare (Stream.fromList [1,3,5]) (Stream.fromList [2,4,6,8])
697-- See also: 'mergeByMFused'
700{-# INLINE mergeBy #-}
701mergeBy :: IsStream t => (a -> a -> Ordering) -> t m a -> t m a -> t m a
702mergeBy f m1 m2 = fromStream $ K.mergeBy f (toStream m1) (toStream m2)
704-- | Like 'mergeBy' but with a monadic comparison function.
706-- Merge two streams randomly:
709-- > randomly _ _ = randomIO >>= \x -> return $ if x then LT else GT
710-- > Stream.toList $ Stream.mergeByM randomly (Stream.fromList [1,1,1,1]) (Stream.fromList [2,2,2,2])
714-- Merge two streams in a proportion of 2:1:
719-- let proportionately m n = do
720-- ref <- newIORef $ cycle $ Prelude.concat [Prelude.replicate m LT, Prelude.replicate n GT]
721-- return $ \_ _ -> do
723-- writeIORef ref $ Prelude.tail r
724-- return $ Prelude.head r
725-- f <- proportionately 2 1
726-- xs <- Stream.toList $ Stream.mergeByM f (Stream.fromList [1,1,1,1,1,1]) (Stream.fromList [2,2,2])
729-- [1,1,2,1,1,2,1,1,2]
733-- See also: 'mergeByMFused'
736{-# INLINE mergeByM #-}
738 :: (IsStream t, Monad m)
739 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
740mergeByM f m1 m2 = fromStream $ K.mergeByM f (toStream m1) (toStream m2)
742-- XXX Fused versions should probably go to a separate module using the same
743-- names for the combinators.
745-- | Like 'mergeByM' but much faster, works best when merging statically known
746-- number of streams. When merging more than two streams try to merge pairs and
747-- pair pf pairs in a tree like structure.'mergeByM' works better with variable
748-- number of streams being merged using 'concatPairsWith'.
751{-# INLINE mergeByMFused #-}
753 :: (IsStream t, Monad m)
754 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
755mergeByMFused f m1 m2 =
756 fromStreamD $ D.mergeByM f (toStreamD m1) (toStreamD m2)
758-- | Like 'mergeByM' but stops merging as soon as any of the two streams stops.
761{-# INLINABLE mergeMinBy #-}
762mergeMinBy :: -- (IsStream t, Monad m) =>
763 (a -> a -> m Ordering) -> t m a -> t m a -> t m a
764mergeMinBy _f _m1 _m2 = undefined
765 -- fromStreamD $ D.mergeMinBy f (toStreamD m1) (toStreamD m2)
767-- | Like 'mergeByM' but stops merging as soon as the first stream stops.
770{-# INLINABLE mergeFstBy #-}
771mergeFstBy :: -- (IsStream t, Monad m) =>
772 (a -> a -> m Ordering) -> t m a -> t m a -> t m a
773mergeFstBy _f _m1 _m2 = undefined
774 -- fromStreamD $ D.mergeFstBy f (toStreamD m1) (toStreamD m2)
776-- XXX we may want to use the name "merge" differently
777-- | Same as @'mergeBy' 'compare'@.
779-- >>> Stream.toList $ Stream.merge (Stream.fromList [1,3,5]) (Stream.fromList [2,4,6,8])
784{-# INLINABLE merge #-}
786 (IsStream t, Ord a) => t m a -> t m a -> t m a
787merge = mergeBy compare
789-- | Like 'mergeBy' but merges concurrently (i.e. both the elements being
790-- merged are generated concurrently).
793{-# INLINE mergeAsyncBy #-}
794mergeAsyncBy :: (IsStream t, MonadAsync m)
795 => (a -> a -> Ordering) -> t m a -> t m a -> t m a
796mergeAsyncBy f = mergeAsyncByM (\a b -> return $ f a b)
798-- | Like 'mergeByM' but merges concurrently (i.e. both the elements being
799-- merged are generated concurrently).
802{-# INLINE mergeAsyncByM #-}
803mergeAsyncByM :: (IsStream t, MonadAsync m)
804 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
805mergeAsyncByM f m1 m2 =
807 let par = Par.mkParallelD . toStreamD
808 in D.mergeByM f (par m1) (par m2)
812{-# DEPRECATED concatUnfold "Please use unfoldMany instead." #-}
813{-# INLINE concatUnfold #-}
814concatUnfold ::(IsStream t, Monad m) => Unfold m a b -> t m a -> t m b
815concatUnfold u m = fromStreamD $ D.unfoldMany u (toStreamD m)
817------------------------------------------------------------------------------
818-- Combine N Streams - unfoldMany
819------------------------------------------------------------------------------
821-- | Like 'concatMap' but uses an 'Unfold' for stream generation. Unlike
822-- 'concatMap' this can fuse the 'Unfold' code with the inner loop and
823-- therefore provide many times better performance.
826{-# INLINE unfoldMany #-}
827unfoldMany ::(IsStream t, Monad m) => Unfold m a b -> t m a -> t m b
828unfoldMany u m = fromStreamD $ D.unfoldMany u (toStreamD m)
830-- | Like 'unfoldMany' but interleaves the streams in the same way as
831-- 'interleave' behaves instead of appending them.
834{-# INLINE unfoldManyInterleave #-}
835unfoldManyInterleave ::(IsStream t, Monad m)
836 => Unfold m a b -> t m a -> t m b
837unfoldManyInterleave u m =
838 fromStreamD $ D.unfoldInterleave u (toStreamD m)
840-- | Like 'unfoldMany' but executes the streams in the same way as
844{-# INLINE unfoldManyRoundRobin #-}
845unfoldManyRoundRobin ::(IsStream t, Monad m)
846 => Unfold m a b -> t m a -> t m b
847unfoldManyRoundRobin u m =
848 fromStreamD $ D.unfoldRoundRobin u (toStreamD m)
850------------------------------------------------------------------------------
851-- Combine N Streams - interpose
852------------------------------------------------------------------------------
854-- > interpose x unf str = gintercalate unf str UF.identity (repeat x)
856-- | Unfold the elements of a stream, intersperse the given element between the
857-- unfolded streams and then concat them into a single stream.
859-- > unwords = S.interpose ' '
862{-# INLINE interpose #-}
863interpose :: (IsStream t, Monad m)
864 => c -> Unfold m b c -> t m b -> t m c
866 fromStreamD $ D.interpose x unf (toStreamD str)
868-- interposeSuffix x unf str = gintercalateSuffix unf str UF.identity (repeat x)
870-- | Unfold the elements of a stream, append the given element after each
871-- unfolded stream and then concat them into a single stream.
873-- > unlines = S.interposeSuffix '\n'
876{-# INLINE interposeSuffix #-}
877interposeSuffix :: (IsStream t, Monad m)
878 => c -> Unfold m b c -> t m b -> t m c
879interposeSuffix x unf str =
880 fromStreamD $ D.interposeSuffix x unf (toStreamD str)
882------------------------------------------------------------------------------
883-- Combine N Streams - intercalate
884------------------------------------------------------------------------------
886-- XXX we can swap the order of arguments to gintercalate so that the
887-- definition of unfoldMany becomes simpler? The first stream should be
888-- infixed inside the second one. However, if we change the order in
889-- "interleave" as well similarly, then that will make it a bit unintuitive.
891-- > unfoldMany unf str =
892-- > gintercalate unf str (UF.nilM (\_ -> return ())) (repeat ())
894-- | 'interleaveInfix' followed by unfold and concat.
897{-# INLINE gintercalate #-}
899 :: (IsStream t, Monad m)
900 => Unfold m a c -> t m a -> Unfold m b c -> t m b -> t m c
901gintercalate unf1 str1 unf2 str2 =
902 fromStreamD $ D.gintercalate
903 unf1 (toStreamD str1)
904 unf2 (toStreamD str2)
906-- > intercalate unf seed str = gintercalate unf str unf (repeatM seed)
908-- | 'intersperse' followed by unfold and concat.
910-- > intercalate unf a str = unfoldMany unf $ intersperse a str
911-- > intersperse = intercalate (Unfold.function id)
912-- > unwords = intercalate Unfold.fromList " "
914-- >>> Stream.toList $ Stream.intercalate Unfold.fromList " " $ Stream.fromList ["abc", "def", "ghi"]
918{-# INLINE intercalate #-}
919intercalate :: (IsStream t, Monad m)
920 => Unfold m b c -> b -> t m b -> t m c
921intercalate unf seed str = fromStreamD $
922 D.unfoldMany unf $ D.intersperse seed (toStreamD str)
924-- | 'interleaveSuffix' followed by unfold and concat.
927{-# INLINE gintercalateSuffix #-}
929 :: (IsStream t, Monad m)
930 => Unfold m a c -> t m a -> Unfold m b c -> t m b -> t m c
931gintercalateSuffix unf1 str1 unf2 str2 =
932 fromStreamD $ D.gintercalateSuffix
933 unf1 (toStreamD str1)
934 unf2 (toStreamD str2)
936-- > intercalateSuffix unf seed str = gintercalateSuffix unf str unf (repeatM seed)
938-- | 'intersperseMSuffix' followed by unfold and concat.
940-- > intercalateSuffix unf a str = unfoldMany unf $ intersperseMSuffix a str
941-- > intersperseMSuffix = intercalateSuffix (Unfold.function id)
942-- > unlines = intercalateSuffix Unfold.fromList "\n"
944-- >>> Stream.toList $ Stream.intercalateSuffix Unfold.fromList "\n" $ Stream.fromList ["abc", "def", "ghi"]
948{-# INLINE intercalateSuffix #-}
949intercalateSuffix :: (IsStream t, Monad m)
950 => Unfold m b c -> b -> t m b -> t m c
951intercalateSuffix unf seed str = fromStreamD $ D.unfoldMany unf
952 $ D.intersperseMSuffix (return seed) (toStreamD str)
954------------------------------------------------------------------------------
955-- Combine N Streams - concatMap
956------------------------------------------------------------------------------
958-- | Flatten a stream of streams to a single stream.
961-- concat = concatMap id
966concat :: (IsStream t, Monad m) => t m (t m a) -> t m a
969------------------------------------------------------------------------------
970-- Combine N Streams - concatMap
971------------------------------------------------------------------------------
973-- | Like 'concatMapWith' but carries a state which can be used to share
974-- information across multiple steps of concat.
977-- concatSmapMWith combine f initial = concatMapWith combine id . smapM f initial
982{-# INLINE concatSmapMWith #-}
984 :: (IsStream t, Monad m)
985 => (t m b -> t m b -> t m b)
986 -> (s -> a -> m (s, t m b))
990concatSmapMWith combine f initial =
991 IsStream.concatMapWith combine id . smapM f initial
993-- XXX Implement a StreamD version for fusion.
995-- | Combine streams in pairs using a binary stream combinator, then combine
996-- the resulting streams in pairs recursively until we get to a single combined
999-- For example, you can sort a stream using merge sort like this:
1001-- >>> Stream.toList $ Stream.concatPairsWith (Stream.mergeBy compare) Stream.fromPure $ Stream.fromList [5,1,7,9,2]
1004-- /Caution: the stream of streams must be finite/
1008{-# INLINE concatPairsWith #-}
1009concatPairsWith :: IsStream t =>
1010 (t m b -> t m b -> t m b)
1014concatPairsWith par f m =
1017 (\s1 s2 -> toStream $ fromStream s1 `par` fromStream s2)
1021------------------------------------------------------------------------------
1022-- IterateMap - Map and flatten Trees of Streams
1023------------------------------------------------------------------------------
1025-- | Like 'iterateM' but iterates after mapping a stream generator on the
1028-- Yield an input element in the output stream, map a stream generator on it
1029-- and then do the same on the resulting stream. This can be used for a depth
1030-- first traversal of a tree like structure.
1032-- Note that 'iterateM' is a special case of 'iterateMapWith':
1035-- iterateM f = iterateMapWith serial (fromEffect . f) . fromEffect
1038-- It can be used to traverse a tree structure. For example, to list a
1042-- Stream.iterateMapWith Stream.serial
1043-- (either Dir.toEither (const nil))
1044-- (fromPure (Left "tmp"))
1049{-# INLINE iterateMapWith #-}
1052 => (t m a -> t m a -> t m a)
1056iterateMapWith combine f = IsStream.concatMapWith combine go
1058 go x = fromPure x `combine` IsStream.concatMapWith combine go (f x)
1060-- | Same as @iterateMapWith Stream.serial@ but more efficient due to stream
1064{-# INLINE iterateUnfold #-}
1065iterateUnfold :: -- (IsStream t, MonadAsync m) =>
1066 Unfold m a a -> t m a -> t m a
1067iterateUnfold = undefined
1069------------------------------------------------------------------------------
1071------------------------------------------------------------------------------
1073-- To traverse graphs we need a state to be carried around in the traversal.
1074-- For example, we can use a hashmap to store the visited status of nodes.
1076-- | Like 'iterateMap' but carries a state in the stream generation function.
1077-- This can be used to traverse graph like structures, we can remember the
1078-- visited nodes in the state to avoid cycles.
1080-- Note that a combination of 'iterateMap' and 'usingState' can also be used to
1081-- traverse graphs. However, this function provides a more localized state
1082-- instead of using a global state.
1088{-# INLINE iterateSmapMWith #-}
1090 :: (IsStream t, Monad m)
1091 => (t m a -> t m a -> t m a)
1092 -> (b -> a -> m (b, t m a))
1096iterateSmapMWith combine f initial stream =
1098 (\b -> IsStream.concatMapWith combine (go b) stream)
1099 (fromEffect initial)
1103 go b a = fromPure a `combine` feedback b a
1107 (\(b1, s) -> IsStream.concatMapWith combine (go b1) s)
1108 (fromEffect $ f b a)
1110------------------------------------------------------------------------------
1112------------------------------------------------------------------------------
1114-- Keep concating either streams as long as rights are generated, stop as soon
1115-- as a left is generated and concat the left stream.
1117-- See also: 'handle'
1123 :: -- (IsStream t, MonadAsync m) =>
1124 (forall x. t m x -> t m x -> t m x)
1125 -> (a -> t m (Either (t m b) b))
1128concatMapEitherWith = undefined
1131-- | In an 'Either' stream iterate on 'Left's. This is a special case of
1135-- iterateMapLeftsWith combine f = iterateMapWith combine (either f (const nil))
1138-- To traverse a directory tree:
1141-- iterateMapLeftsWith serial Dir.toEither (fromPure (Left "tmp"))
1146{-# INLINE iterateMapLeftsWith #-}
1148 :: (IsStream t, b ~ Either a c)
1149 => (t m b -> t m b -> t m b)
1153iterateMapLeftsWith combine f =
1154 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