2-- Module : Streamly.Internal.Data.Stream.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.Expand
14 -- * Binary Combinators (Linear)
15 -- | Functions ending in the shape:
17 -- @Stream m a -> Stream m a -> Stream 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
29 -- * Binary Combinators (Pair Wise)
30 -- | Like the functions in the section above these functions also combine
31 -- two streams into a single stream but when used @n@ times linearly they
32 -- exhibit O(n^2) complexity. They are best combined in a binary tree
33 -- fashion using 'concatPairsWith' giving a @n * log n@ complexity. Avoid
34 -- using these with 'concatMapWith' when combining a large or infinite
56 -- * Combine Streams and Unfolds
58 -- Expand a stream by repeatedly using an unfold and merging the resulting
59 -- streams. Functions generally ending in the shape:
61 -- @Unfold m a b -> Stream m a -> Stream m b@
63 -- ** Unfold and combine streams
64 -- | Unfold and flatten streams.
65 , unfoldMany -- XXX Rename to unfoldAppend
70 -- | Insert effects between streams. Like unfoldMany but intersperses an
71 -- effect between the streams. A special case of gintercalate.
77 -- | Insert Streams between Streams.
78 -- Like unfoldMany but intersperses streams from another source between
79 -- the streams from the first source.
85 -- * Combine Streams of Streams
86 -- | Map and serially append streams. 'concatMapM' is a generalization of
87 -- the binary append operation to append many streams.
94 -- | Map and flatten a stream like 'concatMap' but using a custom binary
95 -- stream merging combinator instead of just appending the streams. The
96 -- merging occurs sequentially, it works efficiently for 'serial', 'async',
97 -- 'ahead' like merge operations where we consume one stream before the
98 -- next or in case of 'wAsync' or 'parallel' where we consume all streams
99 -- simultaneously anyway.
101 -- However, in cases where the merging consumes streams in a round robin
102 -- fashion, a pair wise merging using 'concatPairsWith' would be more
103 -- efficient. These cases include operations like 'mergeBy' or 'zipWith'.
110 -- | See the notes about suitable merge functions in the 'concatMapWith'
115 -- | Map and flatten Trees of Streams
118 , iterateMapLeftsWith
125import Streamly.Internal.Data.Stream.Bottom
126 ( concatM, concatMapM, concatMap, smapM, fromPure, fromEffect
128import Streamly.Internal.Data.Stream.Type
129 ( Stream, fromStreamD, fromStreamK, toStreamD, toStreamK
130 , bindWith, concatMapWith, nil)
131import Streamly.Internal.Data.Unfold.Type (Unfold)
133import qualified Streamly.Internal.Data.Stream.StreamD as D
134import qualified Streamly.Internal.Data.Stream.StreamK as K (mergeBy, mergeByM)
135import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
137import Prelude hiding (concat, concatMap, zipWith)
141-- >>> import Data.Either (either)
142-- >>> import Data.IORef
143-- >>> import Prelude hiding (zipWith, concatMap, concat)
144-- >>> import qualified Streamly.Data.Array.Unboxed as Array
145-- >>> import qualified Streamly.Internal.Data.Fold as Fold
146-- >>> import qualified Streamly.Internal.Data.Stream as Stream
147-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
148-- >>> import qualified Streamly.Internal.Data.Parser as Parser
149-- >>> import qualified Streamly.Internal.FileSystem.Dir as Dir
152------------------------------------------------------------------------------
154------------------------------------------------------------------------------
158-- | This is fused version of 'append'. It could be up to 100x faster than
159-- 'append' when combining two fusible streams. However, it slows down
160-- quadratically with the number of streams being appended. Therefore, it is
161-- suitable for ad-hoc append of a few streams, and should not be used with
162-- 'concatMapWith' or 'concatPairsWith'.
167{-# INLINE append2 #-}
168append2 ::Monad m => Stream m b -> Stream m b -> Stream m b
169append2 m1 m2 = fromStreamD $ D.append (toStreamD m1) (toStreamD m2)
173-- | Appends two streams sequentially, yielding all elements from the first
174-- stream, and then all elements from the second stream.
176-- >>> s1 = Stream.fromList [1,2]
177-- >>> s2 = Stream.fromList [3,4]
178-- >>> Stream.fold Fold.toList $ s1 `Stream.append` s2
181-- This has O(n) append performance where @n@ is the number of streams. It can
182-- be used to efficiently fold an infinite lazy container of streams
183-- 'concatMapWith' et. al.
189append :: Stream m a -> Stream m a -> Stream m a
192------------------------------------------------------------------------------
194------------------------------------------------------------------------------
196-- | Interleaves two streams, yielding one element from each stream
197-- alternately. When one stream stops the rest of the other stream is used in
200-- When joining many streams in a left associative manner earlier streams will
201-- get exponential priority than the ones joining later. Because of exponential
202-- weighting it can be used with 'concatMapWith' even on a large number of
208{-# INLINE interleave #-}
209interleave :: Stream m a -> Stream m a -> Stream m a
210interleave s1 s2 = fromStreamK $ K.interleave (toStreamK s1) (toStreamK s2)
212-- | Like `interleave` but stops interleaving as soon as the first stream
216{-# INLINE interleaveFst #-}
217interleaveFst :: Stream m a -> Stream m a -> Stream m a
219 fromStreamK $ K.interleaveFst (toStreamK s1) (toStreamK s2)
221-- | Like `interleave` but stops interleaving as soon as any of the two streams
225{-# INLINE interleaveMin #-}
226interleaveMin :: Stream m a -> Stream m a -> Stream m a
228 fromStreamK $ K.interleaveMin (toStreamK s1) (toStreamK s2)
230------------------------------------------------------------------------------
231-- Merging (sorted streams)
232------------------------------------------------------------------------------
234-- | Merge two streams using a comparison function. The head elements of both
235-- the streams are compared and the smaller of the two elements is emitted, if
236-- both elements are equal then the element from the first stream is used
239-- If the streams are sorted in ascending order, the resulting stream would
240-- also remain sorted in ascending order.
242-- >>> s1 = Stream.fromList [1,3,5]
243-- >>> s2 = Stream.fromList [2,4,6,8]
244-- >>> Stream.fold Fold.toList $ Stream.mergeBy compare s1 s2
247-- See also: 'mergeByM2'
250{-# INLINE mergeBy #-}
251mergeBy :: (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a
252mergeBy f m1 m2 = fromStreamK $ K.mergeBy f (toStreamK m1) (toStreamK m2)
254-- | Like 'mergeBy' but with a monadic comparison function.
256-- Merge two streams randomly:
259-- > randomly _ _ = randomIO >>= \x -> return $ if x then LT else GT
260-- > Stream.toList $ Stream.mergeByM randomly (Stream.fromList [1,1,1,1]) (Stream.fromList [2,2,2,2])
264-- Merge two streams in a proportion of 2:1:
268-- let s1 = Stream.fromList [1,1,1,1,1,1]
269-- s2 = Stream.fromList [2,2,2]
270-- let proportionately m n = do
271-- ref <- newIORef $ cycle $ Prelude.concat [Prelude.replicate m LT, Prelude.replicate n GT]
272-- return $ \_ _ -> do
274-- writeIORef ref $ Prelude.tail r
275-- return $ Prelude.head r
276-- f <- proportionately 2 1
277-- xs <- Stream.fold Fold.toList $ Stream.mergeByM f s1 s2
280-- [1,1,2,1,1,2,1,1,2]
282-- See also: 'mergeByM2'
285{-# INLINE mergeByM #-}
288 => (a -> a -> m Ordering) -> Stream m a -> Stream m a -> Stream m a
289mergeByM f m1 m2 = fromStreamK $ K.mergeByM f (toStreamK m1) (toStreamK m2)
291-- | Like 'mergeByM' but much faster, works best when merging statically known
292-- number of streams. When merging more than two streams try to merge pairs and
293-- pair of pairs in a tree like structure.'mergeByM' works better with variable
294-- number of streams being merged using 'concatPairsWith'.
297{-# INLINE mergeByM2 #-}
300 => (a -> a -> m Ordering) -> Stream m a -> Stream m a -> Stream m a
302 fromStreamD $ D.mergeByM f (toStreamD m1) (toStreamD m2)
304-- | Like 'mergeByM' but stops merging as soon as any of the two streams stops.
307{-# INLINABLE mergeMinBy #-}
308mergeMinBy :: -- Monad m =>
309 (a -> a -> m Ordering) -> Stream m a -> Stream m a -> Stream m a
310mergeMinBy _f _m1 _m2 = undefined
311 -- fromStreamD $ D.mergeMinBy f (toStreamD m1) (toStreamD m2)
313-- | Like 'mergeByM' but stops merging as soon as the first stream stops.
316{-# INLINABLE mergeFstBy #-}
317mergeFstBy :: -- Monad m =>
318 (a -> a -> m Ordering) -> Stream m a -> Stream m a -> Stream m a
319mergeFstBy _f _m1 _m2 = undefined
320 -- fromStreamK $ D.mergeFstBy f (toStreamD m1) (toStreamD m2)
322------------------------------------------------------------------------------
323-- Combine N Streams - unfoldMany
324------------------------------------------------------------------------------
326-- | Like 'concatMap' but uses an 'Unfold' for stream generation. Unlike
327-- 'concatMap' this can fuse the 'Unfold' code with the inner loop and
328-- therefore provide many times better performance.
331{-# INLINE unfoldMany #-}
332unfoldMany ::Monad m => Unfold m a b -> Stream m a -> Stream m b
333unfoldMany u m = fromStreamD $ D.unfoldMany u (toStreamD m)
335-- | This does not pair streams like concatPairsWith, instead, it goes through
336-- each stream one by one and yields one element from each stream. After it
337-- goes to the last stream it reverses the traversal to come back to the first
338-- stream yielding elements from each stream on its way back to the first
341-- >>> lists = Stream.fromList [[1,1],[2,2],[3,3],[4,4],[5,5]]
342-- >>> interleaved = Stream.unfoldInterleave Unfold.fromList lists
343-- >>> Stream.fold Fold.toList interleaved
344-- [1,2,3,4,5,5,4,3,2,1]
346-- Note that this is order of magnitude more efficient than "concatPairsWith
347-- wSerial" because of fusion.
350{-# INLINE unfoldInterleave #-}
351unfoldInterleave ::Monad m => Unfold m a b -> Stream m a -> Stream m b
352unfoldInterleave u m =
353 fromStreamD $ D.unfoldManyInterleave u (toStreamD m)
355-- | 'unfoldInterleave' switches to the next stream whenever a value from a
356-- stream is yielded, it does not switch on a 'Skip'. So if a stream keeps
357-- skipping for long time other streams won't get a chance to run.
358-- 'unfoldRoundRobin' switches on Skip as well. So it basically schedules each
359-- stream fairly irrespective of whether it produces a value or not.
361{-# INLINE unfoldRoundRobin #-}
362unfoldRoundRobin ::Monad m => Unfold m a b -> Stream m a -> Stream m b
363unfoldRoundRobin u m =
364 fromStreamD $ D.unfoldManyInterleave u (toStreamD m)
366------------------------------------------------------------------------------
367-- Combine N Streams - interpose
368------------------------------------------------------------------------------
370-- > interpose x unf str = gintercalate unf str UF.identity (repeat x)
372-- | Unfold the elements of a stream, intersperse the given element between the
373-- unfolded streams and then concat them into a single stream.
375-- >>> unwords = Stream.interpose ' '
378{-# INLINE interpose #-}
380 => c -> Unfold m b c -> Stream m b -> Stream m c
382 fromStreamD $ D.interpose (return x) unf (toStreamD str)
384-- interposeSuffix x unf str = gintercalateSuffix unf str UF.identity (repeat x)
386-- | Unfold the elements of a stream, append the given element after each
387-- unfolded stream and then concat them into a single stream.
389-- >>> unlines = Stream.interposeSuffix '\n'
392{-# INLINE interposeSuffix #-}
393interposeSuffix :: Monad m
394 => c -> Unfold m b c -> Stream m b -> Stream m c
395interposeSuffix x unf str =
396 fromStreamD $ D.interposeSuffix (return x) unf (toStreamD str)
398------------------------------------------------------------------------------
399-- Combine N Streams - intercalate
400------------------------------------------------------------------------------
402-- XXX we can swap the order of arguments to gintercalate so that the
403-- definition of unfoldMany becomes simpler? The first stream should be
404-- infixed inside the second one. However, if we change the order in
405-- "interleave" as well similarly, then that will make it a bit unintuitive.
407-- > unfoldMany unf str =
408-- > gintercalate unf str (UF.nilM (\_ -> return ())) (repeat ())
410-- | 'interleaveInfix' followed by unfold and concat.
413{-# INLINE gintercalate #-}
416 => Unfold m a c -> Stream m a -> Unfold m b c -> Stream m b -> Stream m c
417gintercalate unf1 str1 unf2 str2 =
418 fromStreamD $ D.gintercalate
419 unf1 (toStreamD str1)
420 unf2 (toStreamD str2)
422-- > intercalate unf seed str = gintercalate unf str unf (repeatM seed)
424-- | 'intersperse' followed by unfold and concat.
426-- >>> intercalate u a = Stream.unfoldMany u . Stream.intersperse a
427-- >>> intersperse = Stream.intercalate Unfold.identity
428-- >>> unwords = Stream.intercalate Unfold.fromList " "
430-- >>> input = Stream.fromList ["abc", "def", "ghi"]
431-- >>> Stream.fold Fold.toList $ Stream.intercalate Unfold.fromList " " input
435{-# INLINE intercalate #-}
436intercalate :: Monad m
437 => Unfold m b c -> b -> Stream m b -> Stream m c
438intercalate unf seed str = fromStreamD $
439 D.unfoldMany unf $ D.intersperse seed (toStreamD str)
441-- | 'interleaveSuffix' followed by unfold and concat.
444{-# INLINE gintercalateSuffix #-}
447 => Unfold m a c -> Stream m a -> Unfold m b c -> Stream m b -> Stream m c
448gintercalateSuffix unf1 str1 unf2 str2 =
449 fromStreamD $ D.gintercalateSuffix
450 unf1 (toStreamD str1)
451 unf2 (toStreamD str2)
453-- > intercalateSuffix unf seed str = gintercalateSuffix unf str unf (repeatM seed)
455-- | 'intersperseSuffix' followed by unfold and concat.
457-- >>> intercalateSuffix u a = Stream.unfoldMany u . Stream.intersperseSuffix a
458-- >>> intersperseSuffix = Stream.intercalateSuffix Unfold.identity
459-- >>> unlines = Stream.intercalateSuffix Unfold.fromList "\n"
461-- >>> input = Stream.fromList ["abc", "def", "ghi"]
462-- >>> Stream.fold Fold.toList $ Stream.intercalateSuffix Unfold.fromList "\n" input
466{-# INLINE intercalateSuffix #-}
467intercalateSuffix :: Monad m
468 => Unfold m b c -> b -> Stream m b -> Stream m c
469intercalateSuffix unf seed str = fromStreamD $ D.unfoldMany unf
470 $ D.intersperseSuffix (return seed) (toStreamD str)
472------------------------------------------------------------------------------
473-- Combine N Streams - concatMap
474------------------------------------------------------------------------------
476-- | Flatten a stream of streams to a single stream.
478-- >>> concat = Stream.concatMap id
482concat :: Monad m => Stream m (Stream m a) -> Stream m a
485------------------------------------------------------------------------------
486-- Combine N Streams - concatMap
487------------------------------------------------------------------------------
489-- | Like 'concatMapWith' but carries a state which can be used to share
490-- information across multiple steps of concat.
492-- >>> concatSmapMWith combine f initial = Stream.concatMapWith combine id . Stream.smapM f initial
496{-# INLINE concatSmapMWith #-}
499 => (Stream m b -> Stream m b -> Stream m b)
500 -> (s -> a -> m (s, Stream m b))
504concatSmapMWith combine f initial =
505 concatMapWith combine id . smapM f initial
507-- XXX Implement a StreamD version for fusion.
509-- | Combine streams in pairs using a binary stream combinator, then combine
510-- the resulting streams in pairs recursively until we get to a single combined
513-- For example, you can sort a stream using merge sort like this:
515-- >>> s = Stream.fromList [5,1,7,9,2]
516-- >>> generate = Stream.fromPure
517-- >>> combine = Stream.mergeBy compare
518-- >>> Stream.fold Fold.toList $ Stream.concatPairsWith combine generate s
521-- /Caution: the stream of streams must be finite/
525{-# INLINE concatPairsWith #-}
527 (Stream m b -> Stream m b -> Stream m b)
531concatPairsWith par f m =
534 (\s1 s2 -> toStreamK $ fromStreamK s1 `par` fromStreamK s2)
538------------------------------------------------------------------------------
539-- IterateMap - Map and flatten Trees of Streams
540------------------------------------------------------------------------------
542-- | Like 'iterateM' but iterates after mapping a stream generator on the
545-- Yield an input element in the output stream, map a stream generator on it
546-- and then do the same on the resulting stream. This can be used for a depth
547-- first traversal of a tree like structure.
549-- Note that 'iterateM' is a special case of 'iterateMapWith':
551-- >>> iterateM f = Stream.iterateMapWith Stream.append (Stream.fromEffect . f) . Stream.fromEffect
553-- It can be used to traverse a tree structure. For example, to list a
556-- >> input = Stream.fromPure (Left ".")
557-- >> f = either Dir.toEither (const Stream.nil)
558-- >> ls = Stream.iterateMapWith Stream.append f input
562{-# INLINE iterateMapWith #-}
564 (Stream m a -> Stream m a -> Stream m a)
568iterateMapWith combine f = concatMapWith combine go
570 go x = fromPure x `combine` concatMapWith combine go (f x)
572-- | Same as @iterateMapWith Stream.serial@ but more efficient due to stream
576{-# INLINE iterateUnfold #-}
577iterateUnfold :: -- (MonadAsync m) =>
578 Unfold m a a -> Stream m a -> Stream m a
579iterateUnfold = undefined
581------------------------------------------------------------------------------
583------------------------------------------------------------------------------
585-- To traverse graphs we need a state to be carried around in the traversal.
586-- For example, we can use a hashmap to store the visited status of nodes.
588-- | Like 'iterateMap' but carries a state in the stream generation function.
589-- This can be used to traverse graph like structures, we can remember the
590-- visited nodes in the state to avoid cycles.
592-- Note that a combination of 'iterateMap' and 'usingState' can also be used to
593-- traverse graphs. However, this function provides a more localized state
594-- instead of using a global state.
600{-# INLINE iterateSmapMWith #-}
603 => (Stream m a -> Stream m a -> Stream m a)
604 -> (b -> a -> m (b, Stream m a))
608iterateSmapMWith combine f initial stream =
610 (\b -> concatMapWith combine (go b) stream)
615 go b a = fromPure a `combine` feedback b a
619 (\(b1, s) -> concatMapWith combine (go b1) s)
622------------------------------------------------------------------------------
624------------------------------------------------------------------------------
626-- Keep concating either streams as long as rights are generated, stop as soon
627-- as a left is generated and concat the left stream.
635 :: -- (MonadAsync m) =>
636 (forall x. t m x -> t m x -> t m x)
637 -> (a -> t m (Either (Stream m b) b))
640concatMapEitherWith = undefined
643-- | In an 'Either' stream iterate on 'Left's. This is a special case of
646-- >>> iterateMapLeftsWith combine f = Stream.iterateMapWith combine (either f (const Stream.nil))
648-- To traverse a directory tree:
650-- >> input = fromPure (Left ".")
651-- >> ls = Stream.iterateMapLeftsWith Stream.append Dir.toEither input
655{-# INLINE iterateMapLeftsWith #-}
658 => (Stream m b -> Stream m b -> Stream m b)
662iterateMapLeftsWith combine f =
663 iterateMapWith combine (either f (const 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