@@ -1,171 +1,43 @@
11-- |
2−-- Module : Streamly.Internal.Data.Stream.IsStream.Nesting
2+-- Module : Streamly.Internal.Data.Stream.IsStream.Reduce
33-- Copyright : (c) 2017 Composewell Technologies
44-- License : BSD-3-Clause
55-- Maintainer : streamly@composewell.com
66-- Stability : experimental
77-- Portability : GHC
88--
9−-- Transformations involving multiple streams, streams and unfolds or streams
10−-- and folds.
9+-- Reduce streams by streams, folds or parsers.
1110
12−module Streamly.Internal.Data.Stream.IsStream.Nesting
11+module Streamly.Internal.Data.Stream.IsStream.Reduce
1312 (
14− -- * Generate
15− -- | Combining streams to generate streams.
16−
17− -- ** Combine Two Streams
18− -- | Functions ending in the shape:
19− --
20− -- @t m a -> t m a -> t m a@.
21−
22− -- *** Appending
23− serial
24− , append
25−
26− -- *** Interleaving
27− , interleave
28− , interleaveMin
29− , interleaveSuffix
30− , interleaveInfix
31−
32− , wSerial
33− , Serial.wSerialFst
34− , Serial.wSerialMin
35−
36− -- *** Scheduling
37− , ahead
38− , async
39− , wAsync
40− , roundrobin
41−
42− -- *** Parallel
43− , parallel
44− , Par.parallelFst
45− , Par.parallelMin
46−
47− -- *** Zipping
48− , Z.zipWith
49− , Z.zipWithM
50− , Z.zipAsyncWith
51− , Z.zipAsyncWithM
52−
53− -- *** Merging
54− -- , merge
55− , mergeBy
56− , mergeByM
57− , mergeAsyncBy
58− , mergeAsyncByM
59−
60− -- *** Trimming
61− , dropPrefix
13+ -- * Reduce By Streams
14+ dropPrefix
6215 , dropInfix
6316 , dropSuffix
6417
65− -- ** Combine N Streams
66− -- | Functions generally ending in these shapes:
67− --
68− -- @
69− -- concat: f (t m a) -> t m a
70− -- concatMap: (a -> t m b) -> t m a -> t m b
71− -- concatUnfold: Unfold m a b -> t m a -> t m b
72− -- @
73−
74− -- *** Flatten Containers of Streams
75− , concatM
76− , concat
77− , concatFoldableWith
78−
79− -- *** ConcatMap
80− -- | Map and flatten streams.
81− , concatMapFoldableWith
82− , concatForFoldableWith
83− , concatMap
84− , concatMapM
85− , concatMapWith
86− , concatSmapMWith
87− , K.bindWith
88− , concatPairsWith
89−
90− -- *** ConcatUnfold
91− -- | Unfold and flatten streams.
92− , concatUnfold
93− , concatUnfoldInterleave
94− , concatUnfoldRoundrobin
95−
96− -- *** Interpose
97− -- | Insert effects between streams. Like concatUnfold but intersperses an
98− -- effect between the streams. A
99− -- special case of gintercalate.
100− , interpose
101− , interposeSuffix
102− -- , interposeBy
103−
104− -- *** Intercalate
105− -- | Insert Streams between Streams.
106− -- Like concatUnfold but intersperses streams from another source between
107− -- the streams from the first source.
108− , intercalate
109− , intercalateSuffix
110− , gintercalate
111− , gintercalateSuffix
112−
113− -- *** IterateMap
114− -- | Map and flatten Trees of Streams
115− , iterateMapWith
116− , iterateSmapMWith
117− , iterateMapLeftsWith
118−
119− -- * Eliminate
120− -- | Folding and Parsing chunks of streams to eliminate nested streams.
121− -- Functions generally ending in these shapes:
18+ -- * Reduce By Folds
19+ -- |
20+ -- Reduce a stream by folding or parsing chunks of the stream. Functions
21+ -- generally ending in these shapes:
12222 --
12323 -- @
12424 -- f (Fold m a b) -> t m a -> t m b
12525 -- f (Parser m a b) -> t m a -> t m b
12626 -- @
12727
128− -- ** Folding
28+ -- ** Generic Folding
12929 -- | Apply folds on a stream.
13030 , foldMany
13131 , foldManyPost
13232 , foldSequence
13333 , foldIterateM
13434
135− -- ** Parsing
136− -- | Apply parsers on a stream.
137− , parseMany
138− , parseManyD
139− , parseManyTill
140− , parseSequence
141− , parseIterate
142−
14335 -- ** Chunking
14436 -- | Element unaware grouping.
14537 , chunksOf
146− , chunksOf2
14738 , arraysOf
14839 , intervalsOf
14940
150− -- ** Grouping
151− -- In imperative terms, grouped folding can be considered as a nested loop
152− -- where we loop over the stream to group elements and then loop over
153− -- individual groups to fold them to a single value that is yielded in the
154− -- output stream.
155−
156− , groups
157− , groupsBy
158− , groupsByRolling
159−
160− -- -- *** Searching Sequences
161− -- , seqIndices -- search a sequence in the stream
162−
163− -- -- *** Searching Multiple Sequences
164− -- , seqIndicesAny -- search any of the given sequence in the stream
165−
166− -- -- -- ** Searching Streams
167− -- -- | Finding a stream within another stream.
168−
16941 -- ** Splitting
17042 -- | Streams can be sliced into segments in space or in time. We use the
17143 -- term @chunk@ to refer to a spatial length of the stream (spatial window)
@@ -179,7 +51,6 @@ module Streamly.Internal.Data.Stream.IsStream.Nesting
17951 -- , splitBy
18052 , splitWithSuffix
18153 -- , splitByPrefix
182− , wordsBy -- stripAndCompactBy
18354
18455 -- -- *** Splitting By Sequences
18556 , splitBySeq
@@ -235,10 +106,41 @@ module Streamly.Internal.Data.Stream.IsStream.Nesting
235106 -- , slidingSessionBuffer
236107 -}
237108
238− -- * Transform (Nested Containers)
239− -- Nested splitting
109+ -- * Reduce By Parsers
110+ -- ** Generic Parsing
111+ -- | Apply parsers on a stream.
112+ , parseMany
113+ , parseManyD
114+ , parseManyTill
115+ , parseSequence
116+ , parseIterate
117+
118+ -- ** Grouping
119+ -- In imperative terms, grouped folding can be considered as a nested loop
120+ -- where we loop over the stream to group elements and then loop over
121+ -- individual groups to fold them to a single value that is yielded in the
122+ -- output stream.
123+
124+ , wordsBy -- stripAndCompactBy
125+ , groups
126+ , groupsBy
127+ , groupsByRolling
128+
129+ -- -- *** Searching Sequences
130+ -- , seqIndices -- search a sequence in the stream
131+
132+ -- -- *** Searching Multiple Sequences
133+ -- , seqIndicesAny -- search any of the given sequence in the stream
134+
135+ -- -- -- ** Searching Streams
136+ -- -- | Finding a stream within another stream.
137+
138+ -- * Nested splitting
240139 , splitInnerBy
241140 , splitInnerBySuffix
141+
142+ -- * Fold2
143+ , chunksOf2
242144 )
243145where
244146
@@ -254,30 +156,18 @@ import Data.Maybe (isNothing)
254156import Foreign.Storable (Storable)
255157import Streamly.Internal.Data.Fold.Type (Fold (..), Fold2 (..))
256158import Streamly.Internal.Data.Parser (Parser (..))
257−import Streamly.Internal.Data.Unfold.Type (Unfold)
258159import Streamly.Internal.Data.Array.Foreign.Type (Array)
259160import Streamly.Internal.Data.SVar (MonadAsync)
260−import Streamly.Internal.Data.Stream.Ahead (ahead)
261−import Streamly.Internal.Data.Stream.Async (async, wAsync)
262161import Streamly.Internal.Data.Stream.IsStream.Common
263− ( concatM
264− , concatMapM
162+ ( concatMap
265163 , fold
266164 , interjectSuffix
267165 , intersperseM
268166 , repeatM
269167 , scanlMAfter'
270− , smapM
271168 , splitOnSeq
272− , yield
273− , yieldM)
274−import Streamly.Internal.Data.Stream.Parallel (parallel)
275−import Streamly.Internal.Data.Stream.Prelude
276− ( fromStreamS, toStreamS, concatFoldableWith, concatMapFoldableWith
277− , concatForFoldableWith)
278−import Streamly.Internal.Data.Stream.StreamD (fromStreamD, toStreamD)
169+ , yield)
279170import Streamly.Internal.Data.Stream.StreamK (IsStream)
280−import Streamly.Internal.Data.Stream.Serial (serial, wSerial)
281171import Streamly.Internal.Data.Time.Units
282172 ( AbsTime, MilliSecond64(..), addToAbsTime, toRelTime
283173 , toAbsTime)
@@ -293,14 +183,8 @@ import qualified Streamly.Internal.Data.Stream.Parallel as Par
293183import qualified Streamly.Internal.Data.Stream.Serial as Serial
294184import qualified Streamly.Internal.Data.Stream.StreamD as D
295185import qualified Streamly.Internal.Data.Stream.StreamK as K
296−import qualified Streamly.Internal.Data.Stream.Zip as Z
297−#ifdef USE_STREAMK_ONLY
298−import qualified Streamly.Internal.Data.Stream.StreamK as S
299−#else
300−import qualified Streamly.Internal.Data.Stream.StreamD as S
301−#endif
302186
303−import Prelude hiding (zipWith, concatMap, concat)
187+import Prelude hiding (concatMap)
304188
305189-- $setup
306190-- >>> :m
@@ -313,624 +197,44 @@ import Prelude hiding (zipWith, concatMap, concat)
313197-- >>> import qualified Streamly.Internal.Data.Parser as Parser
314198-- >>> import qualified Streamly.Data.Array.Foreign as Array
315199
316−-------------------------------------------------------------------------------
317−-- Appending
318−-------------------------------------------------------------------------------
319−
320−-- | Append the outputs of two streams, yielding all the elements from the
321−-- first stream and then yielding all the elements from the second stream.
322−--
323−-- IMPORTANT NOTE: This could be 100x faster than @serial/<>@ for appending a
324−-- few (say 100) streams because it can fuse via stream fusion. However, it
325−-- does not scale for a large number of streams (say 1000s) and becomes
326−-- qudartically slow. Therefore use this for custom appending of a few streams
327−-- but use 'concatMap' or 'concatMapWith serial' for appending @n@ streams or
328−-- infinite containers of streams.
329−--
330−-- /Pre-release/
331−{-# INLINE append #-}
332−append ::(IsStream t, Monad m) => t m b -> t m b -> t m b
333−append m1 m2 = fromStreamD $ D.append (toStreamD m1) (toStreamD m2)
334−
335−-------------------------------------------------------------------------------
336−-- Interleaving
337−-------------------------------------------------------------------------------
338−
339−-- XXX Same as 'wSerial'. We should perhaps rename wSerial to interleave.
340−-- XXX Document the interleaving behavior of side effects in all the
341−-- interleaving combinators.
342−-- XXX Write time-domain equivalents of these. In the time domain we can
343−-- interleave two streams such that the value of second stream is always taken
344−-- from its last value even if no new value is being yielded, like
345−-- zipWithLatest. It would be something like interleaveWithLatest.
346−--
347−-- | Interleaves the outputs of two streams, yielding elements from each stream
348−-- alternately, starting from the first stream. If any of the streams finishes
349−-- early the other stream continues alone until it too finishes.
350−--
351−-- >>> :set -XOverloadedStrings
352−-- >>> import Data.Functor.Identity (Identity)
353−-- >>> Stream.interleave "ab" ",,,," :: Stream.SerialT Identity Char
354−-- fromList "a,b,,,"
355−-- >>> Stream.interleave "abcd" ",," :: Stream.SerialT Identity Char
356−-- fromList "a,b,cd"
357−--
358−-- 'interleave' is dual to 'interleaveMin', it can be called @interleaveMax@.
359−--
360−-- Do not use at scale in concatMapWith.
361−--
362−-- /Pre-release/
363−{-# INLINE interleave #-}
364−interleave ::(IsStream t, Monad m) => t m b -> t m b -> t m b
365−interleave m1 m2 = fromStreamD $ D.interleave (toStreamD m1) (toStreamD m2)
366−
367−-- | Interleaves the outputs of two streams, yielding elements from each stream
368−-- alternately, starting from the first stream. As soon as the first stream
369−-- finishes, the output stops, discarding the remaining part of the second
370−-- stream. In this case, the last element in the resulting stream would be from
371−-- the second stream. If the second stream finishes early then the first stream
372−-- still continues to yield elements until it finishes.
373−--
374−-- >>> :set -XOverloadedStrings
375−-- >>> import Data.Functor.Identity (Identity)
376−-- >>> Stream.interleaveSuffix "abc" ",,,," :: Stream.SerialT Identity Char
377−-- fromList "a,b,c,"
378−-- >>> Stream.interleaveSuffix "abc" "," :: Stream.SerialT Identity Char
379−-- fromList "a,bc"
380−--
381−-- 'interleaveSuffix' is a dual of 'interleaveInfix'.
382−--
383−-- Do not use at scale in concatMapWith.
384−--
385−-- /Pre-release/
386−{-# INLINE interleaveSuffix #-}
387−interleaveSuffix ::(IsStream t, Monad m) => t m b -> t m b -> t m b
388−interleaveSuffix m1 m2 =
389− fromStreamD $ D.interleaveSuffix (toStreamD m1) (toStreamD m2)
390−
391−-- | Interleaves the outputs of two streams, yielding elements from each stream
392−-- alternately, starting from the first stream and ending at the first stream.
393−-- If the second stream is longer than the first, elements from the second
394−-- stream are infixed with elements from the first stream. If the first stream
395−-- is longer then it continues yielding elements even after the second stream
396−-- has finished.
397−--
398−-- >>> :set -XOverloadedStrings
399−-- >>> import Data.Functor.Identity (Identity)
400−-- >>> Stream.interleaveInfix "abc" ",,,," :: Stream.SerialT Identity Char
401−-- fromList "a,b,c"
402−-- >>> Stream.interleaveInfix "abc" "," :: Stream.SerialT Identity Char
403−-- fromList "a,bc"
404−--
405−-- 'interleaveInfix' is a dual of 'interleaveSuffix'.
406−--
407−-- Do not use at scale in concatMapWith.
408−--
409−-- /Pre-release/
410−{-# INLINE interleaveInfix #-}
411−interleaveInfix ::(IsStream t, Monad m) => t m b -> t m b -> t m b
412−interleaveInfix m1 m2 =
413− fromStreamD $ D.interleaveInfix (toStreamD m1) (toStreamD m2)
414−
415−-- | Interleaves the outputs of two streams, yielding elements from each stream
416−-- alternately, starting from the first stream. The output stops as soon as any
417−-- of the two streams finishes, discarding the remaining part of the other
418−-- stream. The last element of the resulting stream would be from the longer
419−-- stream.
420−--
421−-- >>> :set -XOverloadedStrings
422−-- >>> import Data.Functor.Identity (Identity)
423−-- >>> Stream.interleaveMin "ab" ",,,," :: Stream.SerialT Identity Char
424−-- fromList "a,b,"
425−-- >>> Stream.interleaveMin "abcd" ",," :: Stream.SerialT Identity Char
426−-- fromList "a,b,c"
427−--
428−-- 'interleaveMin' is dual to 'interleave'.
429−--
430−-- Do not use at scale in concatMapWith.
431−--
432−-- /Pre-release/
433−{-# INLINE interleaveMin #-}
434−interleaveMin ::(IsStream t, Monad m) => t m b -> t m b -> t m b
435−interleaveMin m1 m2 = fromStreamD $ D.interleaveMin (toStreamD m1) (toStreamD m2)
436−
437−-------------------------------------------------------------------------------
438−-- Scheduling
439−-------------------------------------------------------------------------------
440−
441−-- | Schedule the execution of two streams in a fair round-robin manner,
442−-- executing each stream once, alternately. Execution of a stream may not
443−-- necessarily result in an output, a stream may chose to @Skip@ producing an
444−-- element until later giving the other stream a chance to run. Therefore, this
445−-- combinator fairly interleaves the execution of two streams rather than
446−-- fairly interleaving the output of the two streams. This can be useful in
447−-- co-operative multitasking without using explicit threads. This can be used
448−-- as an alternative to `async`.
449−--
450−-- Do not use at scale in concatMapWith.
451−--
452−-- /Pre-release/
453−{-# INLINE roundrobin #-}
454−roundrobin ::(IsStream t, Monad m) => t m b -> t m b -> t m b
455−roundrobin m1 m2 = fromStreamD $ D.roundRobin (toStreamD m1) (toStreamD m2)
456−
457−------------------------------------------------------------------------------
458−-- Merging (sorted streams)
459−------------------------------------------------------------------------------
460−
461−-- | Merge two streams using a comparison function. The head elements of both
462−-- the streams are compared and the smaller of the two elements is emitted, if
463−-- both elements are equal then the element from the first stream is used
464−-- first.
465−--
466−-- If the streams are sorted in ascending order, the resulting stream would
467−-- also remain sorted in ascending order.
468−--
469−-- @
470−-- >>> Stream.toList $ Stream.mergeBy compare (Stream.fromList [1,3,5]) (Stream.fromList [2,4,6,8])
471−-- [1,2,3,4,5,6,8]
472−--
473−-- @
474−--
475−-- @since 0.6.0
476−{-# INLINABLE mergeBy #-}
477−mergeBy ::
478− (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> t m a -> t m a
479−mergeBy f m1 m2 = fromStreamS $ S.mergeBy f (toStreamS m1) (toStreamS m2)
480−
481−-- | Like 'mergeBy' but with a monadic comparison function.
482−--
483−-- Merge two streams randomly:
484−--
485−-- @
486−-- > randomly _ _ = randomIO >>= \x -> return $ if x then LT else GT
487−-- > S.toList $ S.mergeByM randomly (S.fromList [1,1,1,1]) (S.fromList [2,2,2,2])
488−-- [2,1,2,2,2,1,1,1]
489−-- @
490−--
491−-- Merge two streams in a proportion of 2:1:
492−--
493−-- @
494−-- proportionately m n = do
495−-- ref <- newIORef $ cycle $ concat [replicate m LT, replicate n GT]
496−-- return $ \\_ _ -> do
497−-- r <- readIORef ref
498−-- writeIORef ref $ tail r
499−-- return $ head r
500−--
501−-- main = do
502−-- f <- proportionately 2 1
503−-- xs <- S.toList $ S.mergeByM f (S.fromList [1,1,1,1,1,1]) (S.fromList [2,2,2])
504−-- print xs
505−-- @
506−-- @
507−-- [1,1,2,1,1,2,1,1,2]
508−-- @
509−--
510−-- @since 0.6.0
511−{-# INLINABLE mergeByM #-}
512−mergeByM
513− :: (IsStream t, Monad m)
514− => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
515−mergeByM f m1 m2 = fromStreamS $ S.mergeByM f (toStreamS m1) (toStreamS m2)
516−
517−{-
518−-- | Like 'mergeByM' but stops merging as soon as any of the two streams stops.
519−{-# INLINABLE mergeEndByAny #-}
520−mergeEndByAny
521− :: (IsStream t, Monad m)
522− => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
523−mergeEndByAny f m1 m2 = fromStreamD $
524− D.mergeEndByAny f (toStreamD m1) (toStreamD m2)
525−
526−-- Like 'mergeByM' but stops merging as soon as the first stream stops.
527−{-# INLINABLE mergeEndByFirst #-}
528−mergeEndByFirst
529− :: (IsStream t, Monad m)
530− => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
531−mergeEndByFirst f m1 m2 = fromStreamS $
532− D.mergeEndByFirst f (toStreamD m1) (toStreamD m2)
533−
534−-- Holding this back for now, we may want to use the name "merge" differently
535−-- | Same as @'mergeBy' 'compare'@.
536−--
537−-- @
538−-- > S.toList $ S.merge (S.fromList [1,3,5]) (S.fromList [2,4,6,8])
539−-- [1,2,3,4,5,6,8]
540−-- @
541−--
542−-- @since 0.6.0
543−{-# INLINABLE merge #-}
544−merge ::
545− (IsStream t, Monad m, Ord a) => t m a -> t m a -> t m a
546−merge = mergeBy compare
547−-}
548−
549−-- | Like 'mergeBy' but merges concurrently (i.e. both the elements being
550−-- merged are generated concurrently).
551−--
552−-- @since 0.6.0
553−{-# INLINE mergeAsyncBy #-}
554−mergeAsyncBy :: (IsStream t, MonadAsync m)
555− => (a -> a -> Ordering) -> t m a -> t m a -> t m a
556−mergeAsyncBy f = mergeAsyncByM (\a b -> return $ f a b)
557−
558−-- | Like 'mergeByM' but merges concurrently (i.e. both the elements being
559−-- merged are generated concurrently).
560−--
561−-- @since 0.6.0
562−{-# INLINE mergeAsyncByM #-}
563−mergeAsyncByM :: (IsStream t, MonadAsync m)
564− => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
565−mergeAsyncByM f m1 m2 =
566− fromStreamD $
567− let par = Par.mkParallelD . toStreamD
568− in D.mergeByM f (par m1) (par m2)
569−
570−------------------------------------------------------------------------------
571−-- Combine N Streams - concatMap
572−------------------------------------------------------------------------------
573−
574−-- | @concatMapWith merge map stream@ is a two dimensional looping combinator.
575−-- The first argument specifies a merge or concat function that is used to
576−-- merge the streams generated by applying the second argument i.e. the @map@
577−-- function to each element of the input stream. The concat function could be
578−-- 'serial', 'parallel', 'async', 'ahead' or any other zip or merge function
579−-- and the second argument could be any stream generation function using a
580−-- seed.
581−--
582−-- /Compare 'foldMapWith'/
583−--
584−-- /Since: 0.7.0/
585−--
586−-- /Since: 0.8.0 (signature change)/
587−{-# INLINE concatMapWith #-}
588−concatMapWith
589− :: IsStream t
590− => (t m b -> t m b -> t m b)
591− -> (a -> t m b)
592− -> t m a
593− -> t m b
594−concatMapWith = K.concatMapBy
595−
596−-- | Like 'concatMapWith' but carries a state which can be used to share
597−-- information across multiple steps of concat.
598−--
599−-- @
600−-- concatSmapMWith combine f initial = concatMapWith combine id . smapM f initial
601−-- @
602−--
603−-- /Pre-release/
604−--
605−{-# INLINE concatSmapMWith #-}
606−concatSmapMWith
607− :: (IsStream t, Monad m)
608− => (t m b -> t m b -> t m b)
609− -> (s -> a -> m (s, t m b))
610− -> m s
611− -> t m a
612− -> t m b
613−concatSmapMWith combine f initial = concatMapWith combine id . smapM f initial
614−
615−-- | Map a stream producing function on each element of the stream and then
616−-- flatten the results into a single stream.
617−--
618−-- @
619−-- concatMap f = 'concat . map f'
620−-- concatMap = 'concatMapWith' 'Serial.serial'
621−-- concatMap f = 'concatMapM' (return . f)
622−-- concatMap f = 'concatUnfold' (UF.lmap f UF.fromStream)
623−-- @
624−--
625−-- @since 0.6.0
626−{-# INLINE concatMap #-}
627−concatMap ::(IsStream t, Monad m) => (a -> t m b) -> t m a -> t m b
628−concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)
629−
630−-- | Flatten a stream of streams to a single stream.
631−--
632−-- @
633−-- concat = concatMap id
634−-- @
635−--
636−-- /Pre-release/
637−{-# INLINE concat #-}
638−concat :: (IsStream t, Monad m) => t m (t m a) -> t m a
639−concat = concatMap id
640−
641−-- Keep concating either streams as long as rights are generated, stop as soon
642−-- as a left is generated and concat the left stream.
643−--
644−-- See also: 'handle'
645−--
646−-- /Unimplemented/
647−--
648−{-
649−concatMapEitherWith
650− :: -- (IsStream t, MonadAsync m) =>
651− (forall x. t m x -> t m x -> t m x)
652− -> (a -> t m (Either (t m b) b))
653− -> t m a
654− -> t m b
655−concatMapEitherWith = undefined
656−-}
657−
658−-- XXX Implement a StreamD version for fusion.
659−--
660−-- | Combine streams in pairs using a binary stream combinator, then combine
661−-- the resulting streams in pairs recursively until we get to a single combined
662−-- stream.
663−--
664−-- For example, you can sort a stream using merge sort like this:
665−--
666−-- >>> Stream.toList $ Stream.concatPairsWith (Stream.mergeBy compare) Stream.yield $ Stream.fromList [5,1,7,9,2]
667−-- [1,2,5,7,9]
668−--
669−-- /Caution: the stream of streams must be finite/
670−--
671−-- /Pre-release/
672−--
673−{-# INLINE concatPairsWith #-}
674−concatPairsWith :: IsStream t =>
675− (t m b -> t m b -> t m b)
676− -> (a -> t m b)
677− -> t m a
678− -> t m b
679−concatPairsWith = K.concatPairsWith
680−
681−------------------------------------------------------------------------------
682−-- Combine N Streams - concatUnfold
683−------------------------------------------------------------------------------
684−
685−-- | Like 'concatMap' but uses an 'Unfold' for stream generation. Unlike
686−-- 'concatMap' this can fuse the 'Unfold' code with the inner loop and
687−-- therefore provide many times better performance.
688−--
689−-- @since 0.7.0
690−{-# INLINE concatUnfold #-}
691−concatUnfold ::(IsStream t, Monad m) => Unfold m a b -> t m a -> t m b
692−concatUnfold u m = fromStreamD $ D.concatUnfold u (toStreamD m)
693−
694−-- | Like 'concatUnfold' but interleaves the streams in the same way as
695−-- 'interleave' behaves instead of appending them.
696−--
697−-- /Pre-release/
698−{-# INLINE concatUnfoldInterleave #-}
699−concatUnfoldInterleave ::(IsStream t, Monad m)
700− => Unfold m a b -> t m a -> t m b
701−concatUnfoldInterleave u m =
702− fromStreamD $ D.concatUnfoldInterleave u (toStreamD m)
703−
704−-- | Like 'concatUnfold' but executes the streams in the same way as
705−-- 'roundrobin'.
706−--
707−-- /Pre-release/
708−{-# INLINE concatUnfoldRoundrobin #-}
709−concatUnfoldRoundrobin ::(IsStream t, Monad m)
710− => Unfold m a b -> t m a -> t m b
711−concatUnfoldRoundrobin u m =
712− fromStreamD $ D.concatUnfoldRoundrobin u (toStreamD m)
713−
714−------------------------------------------------------------------------------
715−-- Combine N Streams - interpose
716−------------------------------------------------------------------------------
717−
718−-- > interpose x unf str = gintercalate unf str UF.identity (repeat x)
719−--
720−-- | Unfold the elements of a stream, intersperse the given element between the
721−-- unfolded streams and then concat them into a single stream.
722−--
723−-- > unwords = S.interpose ' '
724−--
725−-- /Pre-release/
726−{-# INLINE interpose #-}
727−interpose :: (IsStream t, Monad m)
728− => c -> Unfold m b c -> t m b -> t m c
729−interpose x unf str =
730− D.fromStreamD $ D.interpose (return x) unf (D.toStreamD str)
731−
732−-- interposeSuffix x unf str = gintercalateSuffix unf str UF.identity (repeat x)
733−--
734−-- | Unfold the elements of a stream, append the given element after each
735−-- unfolded stream and then concat them into a single stream.
736−--
737−-- > unlines = S.interposeSuffix '\n'
738−--
739−-- /Pre-release/
740−{-# INLINE interposeSuffix #-}
741−interposeSuffix :: (IsStream t, Monad m)
742− => c -> Unfold m b c -> t m b -> t m c
743−interposeSuffix x unf str =
744− D.fromStreamD $ D.interposeSuffix (return x) unf (D.toStreamD str)
745−
746200------------------------------------------------------------------------------
747−-- Combine N Streams - intercalate
748−------------------------------------------------------------------------------
749−
750−-- XXX we can swap the order of arguments to gintercalate so that the
751−-- definition of concatUnfold becomes simpler? The first stream should be
752−-- infixed inside the second one. However, if we change the order in
753−-- "interleave" as well similarly, then that will make it a bit unintuitive.
754−--
755−-- > concatUnfold unf str =
756−-- > gintercalate unf str (UF.nilM (\_ -> return ())) (repeat ())
757−--
758−-- | 'interleaveInfix' followed by unfold and concat.
759−--
760−-- /Pre-release/
761−{-# INLINE gintercalate #-}
762−gintercalate
763− :: (IsStream t, Monad m)
764− => Unfold m a c -> t m a -> Unfold m b c -> t m b -> t m c
765−gintercalate unf1 str1 unf2 str2 =
766− D.fromStreamD $ D.gintercalate
767− unf1 (D.toStreamD str1)
768− unf2 (D.toStreamD str2)
769−
770−-- XXX The order of arguments in "intercalate" is consistent with the list
771−-- intercalate but inconsistent with gintercalate and other stream interleaving
772−-- combinators. We can change the order of the arguments in other combinators
773−-- but then 'interleave' combinator may become a bit unintuitive because we
774−-- will be starting with the second stream.
775−
776−-- > intercalate seed unf str = gintercalate unf str unf (repeatM seed)
777−-- > intercalate a unf str = concatUnfold unf $ intersperse a str
778−--
779−-- | 'intersperse' followed by unfold and concat.
780−--
781−-- > unwords = intercalate " " Unfold.fromList
782−--
783−-- >>> Stream.toList $ Stream.intercalate " " Unfold.fromList $ Stream.fromList ["abc", "def", "ghi"]
784−-- "abc def ghi"
785−--
786−-- /Pre-release/
787−{-# INLINE intercalate #-}
788−intercalate :: (IsStream t, Monad m)
789− => b -> Unfold m b c -> t m b -> t m c
790−intercalate seed unf str = D.fromStreamD $
791− D.concatUnfold unf $ D.intersperse seed (toStreamD str)
792−
793−-- | 'interleaveSuffix' followed by unfold and concat.
794−--
795−-- /Pre-release/
796−{-# INLINE gintercalateSuffix #-}
797−gintercalateSuffix
798− :: (IsStream t, Monad m)
799− => Unfold m a c -> t m a -> Unfold m b c -> t m b -> t m c
800−gintercalateSuffix unf1 str1 unf2 str2 =
801− D.fromStreamD $ D.gintercalateSuffix
802− unf1 (D.toStreamD str1)
803− unf2 (D.toStreamD str2)
804−
805−-- > intercalateSuffix seed unf str = gintercalateSuffix unf str unf (repeatM seed)
806−-- > intercalateSuffix a unf str = concatUnfold unf $ intersperseSuffix a str
807−--
808−-- | 'intersperseSuffix' followed by unfold and concat.
809−--
810−-- > unlines = intercalateSuffix "\n" Unfold.fromList
811−--
812−-- >>> Stream.toList $ Stream.intercalateSuffix "\n" Unfold.fromList $ Stream.fromList ["abc", "def", "ghi"]
813−-- "abc\ndef\nghi\n"
814−--
815−-- /Pre-release/
816−{-# INLINE intercalateSuffix #-}
817−intercalateSuffix :: (IsStream t, Monad m)
818− => b -> Unfold m b c -> t m b -> t m c
819−intercalateSuffix seed unf str = fromStreamD $ D.concatUnfold unf
820− $ D.intersperseSuffix (return seed) (D.toStreamD str)
821−
822−------------------------------------------------------------------------------
823−-- IterateMap - Map and flatten Trees of Streams
201+-- Trimming
824202------------------------------------------------------------------------------
825203
826−-- | Like 'iterateM' but iterates after mapping a stream generator on the
827−-- output.
828−--
829−-- Yield an input element in the output stream, map a stream generator on it
830−-- and then do the same on the resulting stream. This can be used for a depth
831−-- first traversal of a tree like structure.
832−--
833−-- Note that 'iterateM' is a special case of 'iterateMapWith':
834−--
835−-- @
836−-- iterateM f = iterateMapWith serial (yieldM . f) . yieldM
837−-- @
838−--
839−-- It can be used to traverse a tree structure. For example, to list a
840−-- directory tree:
841−--
842−-- @
843−-- Stream.iterateMapWith Stream.serial
844−-- (either Dir.toEither (const nil))
845−-- (yield (Left "tmp"))
846−-- @
204+-- | Drop prefix from the input stream if present.
847205--
848−-- /Pre-release/
206+-- Space: @O(1)@
849207--
850−{-# INLINE iterateMapWith #-}
851−iterateMapWith
852− :: IsStream t
853− => (t m a -> t m a -> t m a)
854− -> (a -> t m a)
855− -> t m a
856− -> t m a
857−iterateMapWith combine f = concatMapWith combine go
858− where
859− go x = yield x `combine` concatMapWith combine go (f x)
860−
861−{-
862−{-# INLINE iterateUnfold #-}
863−iterateUnfold :: (IsStream t, MonadAsync m)
864− => Unfold m a a -> t m a -> t m a
865−iterateUnfold unf xs = undefined
866−-}
867−
868−------------------------------------------------------------------------------
869−-- Flattening Graphs
870−------------------------------------------------------------------------------
871−
872−-- To traverse graphs we need a state to be carried around in the traversal.
873−-- For example, we can use a hashmap to store the visited status of nodes.
208+-- /Unimplemented/ - Help wanted.
209+{-# INLINE dropPrefix #-}
210+dropPrefix ::
211+ -- (Eq a, IsStream t, Monad m) =>
212+ t m a -> t m a -> t m a
213+dropPrefix = error "Not implemented yet!"
874214
875−-- | Like 'iterateMap' but carries a state in the stream generation function.
876−-- This can be used to traverse graph like structures, we can remember the
877−-- visited nodes in the state to avoid cycles.
878−--
879−-- Note that a combination of 'iterateMap' and 'usingState' can also be used to
880−-- traverse graphs. However, this function provides a more localized state
881−-- instead of using a global state.
882−--
883−-- See also: 'mfix'
215+-- | Drop all matching infix from the input stream if present. Infix stream
216+-- may be consumed multiple times.
884217--
885−-- /Pre-release/
218+-- Space: @O(n)@ where n is the length of the infix.
886219--
887−{-# INLINE iterateSmapMWith #-}
888−iterateSmapMWith
889− :: (IsStream t, Monad m)
890− => (t m a -> t m a -> t m a)
891− -> (b -> a -> m (b, t m a))
892− -> m b
893− -> t m a
894− -> t m a
895−iterateSmapMWith combine f initial stream =
896− concatMap (\b -> concatMapWith combine (go b) stream) (yieldM initial)
897−
898− where
899−
900− go b a = yield a `combine` feedback b a
901−
902− feedback b a =
903− concatMap
904− (\(b1, s) -> concatMapWith combine (go b1) s)
905− (yieldM $ f b a)
906−
907−------------------------------------------------------------------------------
908−-- iterateMap - Either streams
909−------------------------------------------------------------------------------
220+-- /Unimplemented/ - Help wanted.
221+{-# INLINE dropInfix #-}
222+dropInfix ::
223+ -- (Eq a, IsStream t, Monad m) =>
224+ t m a -> t m a -> t m a
225+dropInfix = error "Not implemented yet!"
910226
911−-- | In an 'Either' stream iterate on 'Left's. This is a special case of
912−-- 'iterateMapWith':
913−--
914−-- @
915−-- iterateMapLeftsWith combine f = iterateMapWith combine (either f (const nil))
916−-- @
917−--
918−-- To traverse a directory tree:
919−--
920−-- @
921−-- iterateMapLeftsWith serial Dir.toEither (yield (Left "tmp"))
922−-- @
227+-- | Drop suffix from the input stream if present. Suffix stream may be
228+-- consumed multiple times.
923229--
924−-- /Pre-release/
230+-- Space: @O(n)@ where n is the length of the suffix.
925231--
926−{-# INLINE iterateMapLeftsWith #-}
927−iterateMapLeftsWith
928− :: IsStream t
929− => (t m (Either a b) -> t m (Either a b) -> t m (Either a b))
930− -> (a -> t m (Either a b))
931− -> t m (Either a b)
932− -> t m (Either a b)
933−iterateMapLeftsWith combine f = iterateMapWith combine (either f (const K.nil))
232+-- /Unimplemented/ - Help wanted.
233+{-# INLINE dropSuffix #-}
234+dropSuffix ::
235+ -- (Eq a, IsStream t, Monad m) =>
236+ t m a -> t m a -> t m a
237+dropSuffix = error "Not implemented yet!"
934238
935239------------------------------------------------------------------------------
936240-- Folding
@@ -1085,6 +389,8 @@ parseSequence
1085389 -> t m b
1086390parseSequence _f _m = undefined
1087391
392+-- XXX Change the parser arguments' order
393+--
1088394-- | @parseManyTill collect test stream@ tries the parser @test@ on the input,
1089395-- if @test@ fails it backtracks and tries @collect@, after @collect@ succeeds
1090396-- @test@ is tried again and so on. The parser stops when @test@ succeeds. The
@@ -1125,45 +431,6 @@ parseIterate
1125431parseIterate f i m = D.fromStreamD $
1126432 D.parseIterate (PRK.fromParserK . f) i (D.toStreamD m)
1127433
1128−------------------------------------------------------------------------------
1129−-- Trimming
1130−------------------------------------------------------------------------------
1131−
1132−-- | Drop prefix from the input stream if present.
1133−--
1134−-- Space: @O(1)@
1135−--
1136−-- /Unimplemented/ - Help wanted.
1137−{-# INLINE dropPrefix #-}
1138−dropPrefix ::
1139− -- (Eq a, IsStream t, Monad m) =>
1140− t m a -> t m a -> t m a
1141−dropPrefix = error "Not implemented yet!"
1142−
1143−-- | Drop all matching infix from the input stream if present. Infix stream
1144−-- may be consumed multiple times.
1145−--
1146−-- Space: @O(n)@ where n is the length of the infix.
1147−--
1148−-- /Unimplemented/ - Help wanted.
1149−{-# INLINE dropInfix #-}
1150−dropInfix ::
1151− -- (Eq a, IsStream t, Monad m) =>
1152− t m a -> t m a -> t m a
1153−dropInfix = error "Not implemented yet!"
1154−
1155−-- | Drop suffix from the input stream if present. Suffix stream may be
1156−-- consumed multiple times.
1157−--
1158−-- Space: @O(n)@ where n is the length of the suffix.
1159−--
1160−-- /Unimplemented/ - Help wanted.
1161−{-# INLINE dropSuffix #-}
1162−dropSuffix ::
1163− -- (Eq a, IsStream t, Monad m) =>
1164− t m a -> t m a -> t m a
1165−dropSuffix = error "Not implemented yet!"
1166−
1167434------------------------------------------------------------------------------
1168435-- Generalized grouping
1169436------------------------------------------------------------------------------