@@ -1,1056 +0,0 @@
1−-- |
2−-- Module : Streamly.Internal.Data.Stream.Transform
3−-- Copyright : (c) 2017 Composewell Technologies
4−-- License : BSD-3-Clause
5−-- Maintainer : streamly@composewell.com
6−-- Stability : experimental
7−-- Portability : GHC
8−
9−module Streamly.Internal.Data.Stream.Transform
10− (
11− -- * Piping
12− -- | Pass through a 'Pipe'.
13− transform
14−
15− -- * Folding
16− , foldrS
17−
18− -- * Mapping
19− -- | Stateless one-to-one maps.
20− , sequence
21− , mapM
22−
23− -- * Mapping Side Effects (Observation)
24− -- | See also the intersperse*_ combinators.
25− , trace
26− , trace_
27− , tap
28−
29− -- * Scanning
30− , scan
31− , scanMany
32− , postscan
33− , smapM
34− , scanlMAfter'
35−
36− -- * Filtering
37− -- | Produce a subset of the stream using criteria based on the values of
38− -- the elements. We can use a concatMap and scan for filtering but these
39− -- combinators are more efficient and convenient.
40−
41− -- mapMaybeM is a general filtering combinator as we can map the stream to
42− -- Just/Nothing using any stateful fold and then use this to filter out.
43− , mapMaybeM
44− , mapMaybe
45− , catMaybes
46− , scanMaybe
47−
48− , with
49− , deleteBy
50− , filter
51− , filterM
52−
53− -- Stateful/scanning filters
54− , uniq
55− , uniqBy
56− , prune
57− , repeated
58−
59− -- * Trimming
60− -- | Produce a subset of the stream trimmed at ends.
61−
62− , take
63− , takeWhile
64− , takeWhileM
65− , takeWhileLast
66− , takeWhileAround
67− , drop
68− , dropLast
69− , dropWhile
70− , dropWhileM
71− , dropWhileLast
72− , dropWhileAround
73−
74− -- * Position Indexing
75− , indexed
76− , indexedR
77−
78− -- * Time Indexing
79− , timestamped
80− , timestampWith
81− , timeIndexed
82− , timeIndexWith
83−
84− -- * Searching
85− , findIndices -- XXX indicesBy
86− , elemIndices -- XXX indicesOf
87−
88− -- * Rolling map
89− -- | Map using the previous element.
90− , rollingMapM
91− , rollingMap
92− , rollingMap2
93−
94− -- Merge
95−
96− -- * Inserting Elements
97− -- | Produce a superset of the stream. This is the opposite of
98− -- filtering/sampling. We can always use concatMap and scan for inserting
99− -- but these combinators are more efficient and convenient.
100−
101− -- Element agnostic (Opposite of sampling)
102− , intersperse
103− , intersperseM -- XXX naming
104− , intersperseMWith
105−
106− , intersperseMSuffix
107− , intersperseMSuffixWith
108−
109− -- , interspersePrefix
110− -- , interspersePrefixBySpan
111−
112− -- * Inserting Side Effects/Time
113− , intersperseM_ -- XXX naming
114− , delay
115− , intersperseMSuffix_
116− , delayPost
117− , intersperseMPrefix_
118− , delayPre
119−
120− -- * Element Aware Insertion
121− -- | Opposite of filtering
122− , insertBy
123− -- , intersperseByBefore
124− -- , intersperseByAfter
125−
126− -- Fold and Unfold, Buffering
127−
128− -- * Reordering
129− , reverse
130− , reverse'
131− , reassembleBy
132−
133− -- * Either Streams
134− -- Move these to Streamly.Data.Either.Stream?
135− , catLefts
136− , catRights
137− , catEithers
138− )
139−where
140−
141−#include "inline.hs"
142−
143−import Control.Concurrent (threadDelay)
144−import Control.Monad (void)
145−import Control.Monad.IO.Class (MonadIO (liftIO))
146−import Data.Either (fromLeft, isLeft, isRight, fromRight)
147−import Data.Maybe (isJust, fromJust)
148−
149−import Streamly.Internal.Data.Fold.Type (Fold)
150−import Streamly.Internal.Data.Pipe (Pipe)
151−import Streamly.Internal.Data.Time.Units (AbsTime, RelTime64)
152−
153−import qualified Streamly.Internal.Data.Fold as FL
154−-- import qualified Streamly.Internal.Data.Fold.Window as Window
155−import qualified Streamly.Internal.Data.Stream.StreamD.Transform as D
156−import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
157−import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
158−
159−
160−import Streamly.Internal.Data.Stream.Bottom
161−import Streamly.Internal.Data.Stream.Type
162−
163−import Prelude hiding
164− ( filter, drop, dropWhile, take, takeWhile, foldr, map, mapM, sequence
165− , reverse, foldr1 , repeat, scanl, scanl1, zipWith)
166−
167−--
168−-- $setup
169−-- >>> :m
170−-- >>> import Control.Concurrent (threadDelay)
171−-- >>> import Control.Monad (void)
172−-- >>> import Control.Monad.IO.Class (MonadIO (liftIO))
173−-- >>> import Data.Either (fromLeft, fromRight, isLeft, isRight, either)
174−-- >>> import Data.Function ((&))
175−-- >>> import Data.Maybe (fromJust, isJust)
176−-- >>> import Prelude hiding (filter, drop, dropWhile, take, takeWhile, foldr, map, mapM, sequence, reverse, foldr1 , scanl, scanl1)
177−-- >>> import Streamly.Internal.Data.Stream (Stream)
178−-- >>> import qualified Streamly.Data.Fold as Fold
179−-- >>> import qualified Streamly.Data.Unfold as Unfold
180−-- >>> import qualified Streamly.Internal.Data.Fold as Fold (filtering)
181−-- >>> import qualified Streamly.Internal.Data.Fold.Window as Window
182−-- >>> import qualified Streamly.Internal.Data.Stream as Stream
183−-- >>> import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
184−--
185−-- >>> hSetBuffering stdout LineBuffering
186−
187−-- XXX because of the use of D.cons for appending, folds and scans have
188−-- quadratic complexity when iterated over a stream. We should use StreamK for
189−-- linear performance on iteration.
190−
191−------------------------------------------------------------------------------
192−-- Piping
193−------------------------------------------------------------------------------
194−
195−-- | Use a 'Pipe' to transform a stream.
196−--
197−-- /Pre-release/
198−--
199−{-# INLINE transform #-}
200−transform :: Monad m => Pipe m a b -> Stream m a -> Stream m b
201−transform pipe xs = fromStreamD $ D.transform pipe (toStreamD xs)
202−
203−------------------------------------------------------------------------------
204−-- Transformation Folds
205−------------------------------------------------------------------------------
206−
207−-- | Right fold to a streaming monad.
208−--
209−-- > foldrS Stream.cons Stream.nil === id
210−--
211−-- 'foldrS' can be used to perform stateless stream to stream transformations
212−-- like map and filter in general. It can be coupled with a scan to perform
213−-- stateful transformations. However, note that the custom map and filter
214−-- routines can be much more efficient than this due to better stream fusion.
215−--
216−-- >>> input = Stream.fromList [1..5]
217−-- >>> Stream.fold Fold.toList $ Stream.foldrS Stream.cons Stream.nil input
218−-- [1,2,3,4,5]
219−--
220−-- Find if any element in the stream is 'True':
221−--
222−-- >>> step x xs = if odd x then Stream.fromPure True else xs
223−-- >>> input = Stream.fromList (2:4:5:undefined) :: Stream IO Int
224−-- >>> Stream.fold Fold.toList $ Stream.foldrS step (Stream.fromPure False) input
225−-- [True]
226−--
227−-- Map (+2) on odd elements and filter out the even elements:
228−--
229−-- >>> step x xs = if odd x then (x + 2) `Stream.cons` xs else xs
230−-- >>> input = Stream.fromList [1..5] :: Stream IO Int
231−-- >>> Stream.fold Fold.toList $ Stream.foldrS step Stream.nil input
232−-- [3,5,7]
233−--
234−-- /Pre-release/
235−{-# INLINE foldrS #-}
236−foldrS ::
237− (a -> Stream m b -> Stream m b)
238− -> Stream m b
239− -> Stream m a
240− -> Stream m b
241−foldrS f z xs =
242− fromStreamK
243− $ K.foldrS
244− (\y ys -> toStreamK $ f y (fromStreamK ys))
245− (toStreamK z)
246− (toStreamK xs)
247−
248−------------------------------------------------------------------------------
249−-- Transformation by Mapping
250−------------------------------------------------------------------------------
251−
252−-- |
253−-- >>> mapM f = Stream.sequence . fmap f
254−--
255−-- Apply a monadic function to each element of the stream and replace it with
256−-- the output of the resulting action.
257−--
258−-- >>> s = Stream.fromList ["a", "b", "c"]
259−-- >>> Stream.fold Fold.drain $ Stream.mapM putStr s
260−-- abc
261−--
262−{-# INLINE mapM #-}
263−mapM :: Monad m => (a -> m b) -> Stream m a -> Stream m b
264−mapM f m = fromStreamK $ D.toStreamK $ D.mapM f $ toStreamD m
265−
266−-- |
267−-- >>> sequence = Stream.mapM id
268−--
269−-- Replace the elements of a stream of monadic actions with the outputs of
270−-- those actions.
271−--
272−-- >>> s = Stream.fromList [putStr "a", putStr "b", putStrLn "c"]
273−-- >>> Stream.fold Fold.drain $ Stream.sequence s
274−-- abc
275−--
276−{-# INLINE sequence #-}
277−sequence :: Monad m => Stream m (m a) -> Stream m a
278−sequence = mapM id
279−
280−------------------------------------------------------------------------------
281−-- Mapping side effects
282−------------------------------------------------------------------------------
283−
284−-- | Tap the data flowing through a stream into a 'Fold'. For example, you may
285−-- add a tap to log the contents flowing through the stream. The fold is used
286−-- only for effects, its result is discarded.
287−--
288−-- @
289−-- Fold m a b
290−-- |
291−-- -----stream m a ---------------stream m a-----
292−--
293−-- @
294−--
295−-- >>> s = Stream.enumerateFromTo 1 2
296−-- >>> Stream.fold Fold.drain $ Stream.tap (Fold.drainMapM print) s
297−-- 1
298−-- 2
299−--
300−-- Compare with 'trace'.
301−--
302−{-# INLINE tap #-}
303−tap :: Monad m => FL.Fold m a b -> Stream m a -> Stream m a
304−tap f xs = fromStreamD $ D.tap f (toStreamD xs)
305−
306−-- | Apply a monadic function to each element flowing through the stream and
307−-- discard the results.
308−--
309−-- >>> s = Stream.enumerateFromTo 1 2
310−-- >>> Stream.fold Fold.drain $ Stream.trace print s
311−-- 1
312−-- 2
313−--
314−-- Compare with 'tap'.
315−--
316−{-# INLINE trace #-}
317−trace :: Monad m => (a -> m b) -> Stream m a -> Stream m a
318−trace f = mapM (\x -> void (f x) >> return x)
319−
320−-- | Perform a side effect before yielding each element of the stream and
321−-- discard the results.
322−--
323−-- >>> s = Stream.enumerateFromTo 1 2
324−-- >>> Stream.fold Fold.drain $ Stream.trace_ (print "got here") s
325−-- "got here"
326−-- "got here"
327−--
328−-- Same as 'intersperseMPrefix_' but always serial.
329−--
330−-- See also: 'trace'
331−--
332−-- /Pre-release/
333−{-# INLINE trace_ #-}
334−trace_ :: Monad m => m b -> Stream m a -> Stream m a
335−trace_ eff = fromStreamD . D.mapM (\x -> eff >> return x) . toStreamD
336−
337−-------------------------------------------------------------------------------
338−-- Scanning
339−-------------------------------------------------------------------------------
340−
341−-- | @scanlMAfter' accumulate initial done stream@ is like 'scanlM'' except
342−-- that it provides an additional @done@ function to be applied on the
343−-- accumulator when the stream stops. The result of @done@ is also emitted in
344−-- the stream.
345−--
346−-- This function can be used to allocate a resource in the beginning of the
347−-- scan and release it when the stream ends or to flush the internal state of
348−-- the scan at the end.
349−--
350−-- /Pre-release/
351−--
352−{-# INLINE scanlMAfter' #-}
353−scanlMAfter' ::
354− Monad m
355− => (b -> a -> m b)
356− -> m b
357− -> (b -> m b)
358− -> Stream m a
359− -> Stream m b
360−scanlMAfter' step initial done stream =
361− fromStreamD $ D.scanlMAfter' step initial done $ toStreamD stream
362−
363−------------------------------------------------------------------------------
364−-- Scanning with a Fold
365−------------------------------------------------------------------------------
366−
367−-- XXX It may be useful to have a version of scan where we can keep the
368−-- accumulator independent of the value emitted. So that we do not necessarily
369−-- have to keep a value in the accumulator which we are not using. We can pass
370−-- an extraction function that will take the accumulator and the current value
371−-- of the element and emit the next value in the stream. That will also make it
372−-- possible to modify the accumulator after using it. In fact, the step function
373−-- can return new accumulator and the value to be emitted. The signature would
374−-- be more like mapAccumL.
375−
376−-- | Strict left scan. Scan a stream using the given monadic fold.
377−--
378−-- >>> s = Stream.fromList [1..10]
379−-- >>> Stream.fold Fold.toList $ Stream.takeWhile (< 10) $ Stream.scan Fold.sum s
380−-- [0,1,3,6]
381−--
382−-- See also: 'usingStateT'
383−--
384−
385−-- EXPLANATION:
386−-- >>> scanl' step z = Stream.scan (Fold.foldl' step z)
387−--
388−-- Like 'map', 'scanl'' too is a one to one transformation,
389−-- however it adds an extra element.
390−--
391−-- >>> s = Stream.fromList [1,2,3,4]
392−-- >>> Stream.fold Fold.toList $ scanl' (+) 0 s
393−-- [0,1,3,6,10]
394−--
395−-- >>> Stream.fold Fold.toList $ scanl' (flip (:)) [] s
396−-- [[],[1],[2,1],[3,2,1],[4,3,2,1]]
397−--
398−-- The output of 'scanl'' is the initial value of the accumulator followed by
399−-- all the intermediate steps and the final result of 'foldl''.
400−--
401−-- By streaming the accumulated state after each fold step, we can share the
402−-- state across multiple stages of stream composition. Each stage can modify or
403−-- extend the state, do some processing with it and emit it for the next stage,
404−-- thus modularizing the stream processing. This can be useful in
405−-- stateful or event-driven programming.
406−--
407−-- Consider the following monolithic example, computing the sum and the product
408−-- of the elements in a stream in one go using a @foldl'@:
409−--
410−-- >>> foldl' step z = Stream.fold (Fold.foldl' step z)
411−-- >>> foldl' (\(s, p) x -> (s + x, p * x)) (0,1) s
412−-- (10,24)
413−--
414−-- Using @scanl'@ we can make it modular by computing the sum in the first
415−-- stage and passing it down to the next stage for computing the product:
416−--
417−-- >>> :{
418−-- foldl' (\(_, p) (s, x) -> (s, p * x)) (0,1)
419−-- $ scanl' (\(s, _) x -> (s + x, x)) (0,1)
420−-- $ Stream.fromList [1,2,3,4]
421−-- :}
422−-- (10,24)
423−--
424−-- IMPORTANT: 'scanl'' evaluates the accumulator to WHNF. To avoid building
425−-- lazy expressions inside the accumulator, it is recommended that a strict
426−-- data structure is used for accumulator.
427−--
428−{-# INLINE scan #-}
429−scan :: Monad m => Fold m a b -> Stream m a -> Stream m b
430−scan fld m = fromStreamD $ D.scan fld $ toStreamD m
431−
432−-- | Like 'scan' but restarts scanning afresh when the scanning fold
433−-- terminates.
434−--
435−{-# INLINE scanMany #-}
436−scanMany :: Monad m => Fold m a b -> Stream m a -> Stream m b
437−scanMany fld m = fromStreamD $ D.scanMany fld $ toStreamD m
438−
439−------------------------------------------------------------------------------
440−-- Filtering
441−------------------------------------------------------------------------------
442−
443−-- | Modify a @Stream m a -> Stream m a@ stream transformation that accepts a
444−-- predicate @(a -> b)@ to accept @((s, a) -> b)@ instead, provided a
445−-- transformation @Stream m a -> Stream m (s, a)@. Convenient to filter with
446−-- index or time.
447−--
448−-- >>> filterWithIndex = Stream.with Stream.indexed Stream.filter
449−--
450−-- /Pre-release/
451−{-# INLINE with #-}
452−with :: Monad m =>
453− (Stream m a -> Stream m (s, a))
454− -> (((s, a) -> b) -> Stream m (s, a) -> Stream m (s, a))
455− -> (((s, a) -> b) -> Stream m a -> Stream m a)
456−with f comb g = fmap snd . comb g . f
457−
458−-- | Include only those elements that pass a predicate.
459−--
460−-- >>> filter p = Stream.filterM (return . p)
461−-- >>> filter p = Stream.mapMaybe (\x -> if p x then Just x else Nothing)
462−-- >>> filter p = Stream.scanMaybe (Fold.filtering p)
463−--
464−{-# INLINE filter #-}
465−filter :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
466−-- filter p = scanMaybe (FL.filtering p)
467−filter p m = fromStreamD $ D.filter p $ toStreamD m
468−
469−-- | Same as 'filter' but with a monadic predicate.
470−--
471−-- >>> f p x = p x >>= \r -> return $ if r then Just x else Nothing
472−-- >>> filterM p = Stream.mapMaybeM (f p)
473−--
474−{-# INLINE filterM #-}
475−filterM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
476−filterM p m = fromStreamD $ D.filterM p $ toStreamD m
477−
478−-- | Drop repeated elements that are adjacent to each other using the supplied
479−-- comparison function.
480−--
481−-- >>> uniq = Stream.uniqBy (==)
482−--
483−-- To strip duplicate path separators:
484−--
485−-- >>> input = Stream.fromList "//a//b"
486−-- >>> f x y = x == '/' && y == '/'
487−-- >>> Stream.fold Fold.toList $ Stream.uniqBy f input
488−-- "/a/b"
489−--
490−-- Space: @O(1)@
491−--
492−-- /Pre-release/
493−--
494−{-# INLINE uniqBy #-}
495−uniqBy :: Monad m =>
496− (a -> a -> Bool) -> Stream m a -> Stream m a
497−-- uniqBy eq = scanMaybe (FL.uniqBy eq)
498−uniqBy eq = catMaybes . rollingMap f
499−
500− where
501−
502− f pre curr =
503− case pre of
504− Nothing -> Just curr
505− Just x -> if x `eq` curr then Nothing else Just curr
506−
507−-- | Drop repeated elements that are adjacent to each other.
508−--
509−-- >>> uniq = Stream.uniqBy (==)
510−--
511−{-# INLINE uniq #-}
512−uniq :: (Eq a, Monad m) => Stream m a -> Stream m a
513−-- uniq = scanMaybe FL.uniq
514−uniq = fromStreamD . D.uniq . toStreamD
515−
516−-- | Strip all leading and trailing occurrences of an element passing a
517−-- predicate and make all other consecutive occurrences uniq.
518−--
519−-- >> prune p = Stream.dropWhileAround p $ Stream.uniqBy (x y -> p x && p y)
520−--
521−-- @
522−-- > Stream.prune isSpace (Stream.fromList " hello world! ")
523−-- "hello world!"
524−--
525−-- @
526−--
527−-- Space: @O(1)@
528−--
529−-- /Unimplemented/
530−{-# INLINE prune #-}
531−prune ::
532− -- (Monad m, Eq a) =>
533− (a -> Bool) -> Stream m a -> Stream m a
534−prune = error "Not implemented yet!"
535−
536−-- Possible implementation:
537−-- @repeated =
538−-- Stream.catMaybes . Stream.parseMany (Parser.groupBy (==) Fold.repeated)@
539−--
540−-- 'Fold.repeated' should return 'Just' when repeated, and 'Nothing' for a
541−-- single element.
542−
543−-- | Emit only repeated elements, once.
544−--
545−-- /Unimplemented/
546−repeated :: -- (Monad m, Eq a) =>
547− Stream m a -> Stream m a
548−repeated = undefined
549−
550−-- | Deletes the first occurrence of the element in the stream that satisfies
551−-- the given equality predicate.
552−--
553−-- >>> input = Stream.fromList [1,3,3,5]
554−-- >>> Stream.fold Fold.toList $ Stream.deleteBy (==) 3 input
555−-- [1,3,5]
556−--
557−{-# INLINE deleteBy #-}
558−deleteBy :: Monad m => (a -> a -> Bool) -> a -> Stream m a -> Stream m a
559−-- deleteBy cmp x = scanMaybe (FL.deleteBy cmp x)
560−deleteBy cmp x m = fromStreamD $ D.deleteBy cmp x (toStreamD m)
561−
562−------------------------------------------------------------------------------
563−-- Trimming
564−------------------------------------------------------------------------------
565−
566−-- | Same as 'takeWhile' but with a monadic predicate.
567−--
568−{-# INLINE takeWhileM #-}
569−takeWhileM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
570−-- takeWhileM p = scanMaybe (FL.takingEndByM_ (\x -> not <$> p x))
571−takeWhileM p m = fromStreamD $ D.takeWhileM p $ toStreamD m
572−
573−-- | Take all consecutive elements at the end of the stream for which the
574−-- predicate is true.
575−--
576−-- O(n) space, where n is the number elements taken.
577−--
578−-- /Unimplemented/
579−{-# INLINE takeWhileLast #-}
580−takeWhileLast :: -- Monad m =>
581− (a -> Bool) -> Stream m a -> Stream m a
582−takeWhileLast = undefined -- fromStreamD $ D.takeWhileLast n $ toStreamD m
583−
584−-- | Like 'takeWhile' and 'takeWhileLast' combined.
585−--
586−-- O(n) space, where n is the number elements taken from the end.
587−--
588−-- /Unimplemented/
589−{-# INLINE takeWhileAround #-}
590−takeWhileAround :: -- Monad m =>
591− (a -> Bool) -> Stream m a -> Stream m a
592−takeWhileAround = undefined -- fromStreamD $ D.takeWhileAround n $ toStreamD m
593−
594−-- | Drop elements in the stream as long as the predicate succeeds and then
595−-- take the rest of the stream.
596−--
597−{-# INLINE dropWhile #-}
598−dropWhile :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
599−-- dropWhile p = scanMaybe (FL.droppingWhile p)
600−dropWhile p m = fromStreamD $ D.dropWhile p $ toStreamD m
601−
602−-- | Same as 'dropWhile' but with a monadic predicate.
603−--
604−{-# INLINE dropWhileM #-}
605−dropWhileM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a
606−-- dropWhileM p = scanMaybe (FL.droppingWhileM p)
607−dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m
608−
609−-- | Drop @n@ elements at the end of the stream.
610−--
611−-- O(n) space, where n is the number elements dropped.
612−--
613−-- /Unimplemented/
614−{-# INLINE dropLast #-}
615−dropLast :: -- Monad m =>
616− Int -> Stream m a -> Stream m a
617−dropLast = undefined -- fromStreamD $ D.dropLast n $ toStreamD m
618−
619−-- | Drop all consecutive elements at the end of the stream for which the
620−-- predicate is true.
621−--
622−-- O(n) space, where n is the number elements dropped.
623−--
624−-- /Unimplemented/
625−{-# INLINE dropWhileLast #-}
626−dropWhileLast :: -- Monad m =>
627− (a -> Bool) -> Stream m a -> Stream m a
628−dropWhileLast = undefined -- fromStreamD $ D.dropWhileLast n $ toStreamD m
629−
630−-- | Like 'dropWhile' and 'dropWhileLast' combined.
631−--
632−-- O(n) space, where n is the number elements dropped from the end.
633−--
634−-- /Unimplemented/
635−{-# INLINE dropWhileAround #-}
636−dropWhileAround :: -- Monad m =>
637− (a -> Bool) -> Stream m a -> Stream m a
638−dropWhileAround = undefined -- fromStreamD $ D.dropWhileAround n $ toStreamD m
639−
640−------------------------------------------------------------------------------
641−-- Inserting Elements
642−------------------------------------------------------------------------------
643−
644−-- | @insertBy cmp elem stream@ inserts @elem@ before the first element in
645−-- @stream@ that is less than @elem@ when compared using @cmp@.
646−--
647−-- >>> insertBy cmp x = Stream.mergeBy cmp (Stream.fromPure x)
648−--
649−-- >>> input = Stream.fromList [1,3,5]
650−-- >>> Stream.fold Fold.toList $ Stream.insertBy compare 2 input
651−-- [1,2,3,5]
652−--
653−{-# INLINE insertBy #-}
654−insertBy ::Monad m => (a -> a -> Ordering) -> a -> Stream m a -> Stream m a
655−insertBy cmp x m = fromStreamD $ D.insertBy cmp x (toStreamD m)
656−
657−-- | Insert a pure value between successive elements of a stream.
658−--
659−-- >>> input = Stream.fromList "hello"
660−-- >>> Stream.fold Fold.toList $ Stream.intersperse ',' input
661−-- "h,e,l,l,o"
662−--
663−{-# INLINE intersperse #-}
664−intersperse :: Monad m => a -> Stream m a -> Stream m a
665−intersperse a = fromStreamD . D.intersperse a . toStreamD
666−
667−-- | Insert a side effect before consuming an element of a stream except the
668−-- first one.
669−--
670−-- >>> input = Stream.fromList "hello"
671−-- >>> Stream.fold Fold.drain $ Stream.trace putChar $ Stream.intersperseM_ (putChar '.') input
672−-- h.e.l.l.o
673−--
674−-- /Pre-release/
675−{-# INLINE intersperseM_ #-}
676−intersperseM_ :: Monad m => m b -> Stream m a -> Stream m a
677−intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
678−
679−-- | Intersperse a monadic action into the input stream after every @n@
680−-- elements.
681−--
682−-- >> input = Stream.fromList "hello"
683−-- >> Stream.fold Fold.toList $ Stream.intersperseMWith 2 (return ',') input
684−-- "he,ll,o"
685−--
686−-- /Unimplemented/
687−{-# INLINE intersperseMWith #-}
688−intersperseMWith :: -- Monad m =>
689− Int -> m a -> Stream m a -> Stream m a
690−intersperseMWith _n _f _xs = undefined
691−
692−-- | Insert an effect and its output after consuming an element of a stream.
693−--
694−-- >>> input = Stream.fromList "hello"
695−-- >>> Stream.fold Fold.toList $ Stream.trace putChar $ Stream.intersperseMSuffix (putChar '.' >> return ',') input
696−-- h.,e.,l.,l.,o.,"h,e,l,l,o,"
697−--
698−-- /Pre-release/
699−{-# INLINE intersperseMSuffix #-}
700−intersperseMSuffix :: Monad m => m a -> Stream m a -> Stream m a
701−intersperseMSuffix m = fromStreamD . D.intersperseMSuffix m . toStreamD
702−
703−-- | Insert a side effect after consuming an element of a stream.
704−--
705−-- >>> input = Stream.fromList "hello"
706−-- >>> Stream.fold Fold.toList $ Stream.intersperseMSuffix_ (threadDelay 1000000) input
707−-- "hello"
708−--
709−-- /Pre-release/
710−--
711−{-# INLINE intersperseMSuffix_ #-}
712−intersperseMSuffix_ :: Monad m => m b -> Stream m a -> Stream m a
713−intersperseMSuffix_ m = fromStreamD . D.intersperseMSuffix_ m . toStreamD
714−
715−-- XXX Use an offset argument, like tapOffsetEvery
716−
717−-- | Like 'intersperseMSuffix' but intersperses an effectful action into the
718−-- input stream after every @n@ elements and after the last element.
719−--
720−-- >>> input = Stream.fromList "hello"
721−-- >>> Stream.fold Fold.toList $ Stream.intersperseMSuffixWith 2 (return ',') input
722−-- "he,ll,o,"
723−--
724−-- /Pre-release/
725−--
726−{-# INLINE intersperseMSuffixWith #-}
727−intersperseMSuffixWith :: Monad m
728− => Int -> m a -> Stream m a -> Stream m a
729−intersperseMSuffixWith n eff =
730− fromStreamD . D.intersperseMSuffixWith n eff . toStreamD
731−
732−-- | Insert a side effect before consuming an element of a stream.
733−--
734−-- Definition:
735−--
736−-- >>> intersperseMPrefix_ m = Stream.mapM (\x -> void m >> return x)
737−--
738−-- >>> input = Stream.fromList "hello"
739−-- >>> Stream.fold Fold.toList $ Stream.trace putChar $ Stream.intersperseMPrefix_ (putChar '.' >> return ',') input
740−-- .h.e.l.l.o"hello"
741−--
742−-- Same as 'trace_'.
743−--
744−-- /Pre-release/
745−--
746−{-# INLINE intersperseMPrefix_ #-}
747−intersperseMPrefix_ :: Monad m => m b -> Stream m a -> Stream m a
748−intersperseMPrefix_ m = mapM (\x -> void m >> return x)
749−
750−------------------------------------------------------------------------------
751−-- Inserting Time
752−------------------------------------------------------------------------------
753−
754−-- XXX This should be in Prelude, should we export this as a helper function?
755−
756−-- | Block the current thread for specified number of seconds.
757−{-# INLINE sleep #-}
758−sleep :: MonadIO m => Double -> m ()
759−sleep n = liftIO $ threadDelay $ round $ n * 1000000
760−
761−-- | Introduce a delay of specified seconds between elements of the stream.
762−--
763−-- Definition:
764−--
765−-- >>> sleep n = liftIO $ threadDelay $ round $ n * 1000000
766−-- >>> delay = Stream.intersperseM_ . sleep
767−--
768−-- Example:
769−--
770−-- >>> input = Stream.enumerateFromTo 1 3
771−-- >>> Stream.fold (Fold.drainMapM print) $ Stream.delay 1 input
772−-- 1
773−-- 2
774−-- 3
775−--
776−{-# INLINE delay #-}
777−delay :: MonadIO m => Double -> Stream m a -> Stream m a
778−delay = intersperseM_ . sleep
779−
780−-- | Introduce a delay of specified seconds after consuming an element of a
781−-- stream.
782−--
783−-- Definition:
784−--
785−-- >>> sleep n = liftIO $ threadDelay $ round $ n * 1000000
786−-- >>> delayPost = Stream.intersperseMSuffix_ . sleep
787−--
788−-- Example:
789−--
790−-- >>> input = Stream.enumerateFromTo 1 3
791−-- >>> Stream.fold (Fold.drainMapM print) $ Stream.delayPost 1 input
792−-- 1
793−-- 2
794−-- 3
795−--
796−-- /Pre-release/
797−--
798−{-# INLINE delayPost #-}
799−delayPost :: MonadIO m => Double -> Stream m a -> Stream m a
800−delayPost n = intersperseMSuffix_ $ liftIO $ threadDelay $ round $ n * 1000000
801−
802−-- | Introduce a delay of specified seconds before consuming an element of a
803−-- stream.
804−--
805−-- Definition:
806−--
807−-- >>> sleep n = liftIO $ threadDelay $ round $ n * 1000000
808−-- >>> delayPre = Stream.intersperseMPrefix_. sleep
809−--
810−-- Example:
811−--
812−-- >>> input = Stream.enumerateFromTo 1 3
813−-- >>> Stream.fold (Fold.drainMapM print) $ Stream.delayPre 1 input
814−-- 1
815−-- 2
816−-- 3
817−--
818−-- /Pre-release/
819−--
820−{-# INLINE delayPre #-}
821−delayPre :: MonadIO m => Double -> Stream m a -> Stream m a
822−delayPre = intersperseMPrefix_. sleep
823−
824−------------------------------------------------------------------------------
825−-- Reorder in sequence
826−------------------------------------------------------------------------------
827−
828−-- | Buffer until the next element in sequence arrives. The function argument
829−-- determines the difference in sequence numbers. This could be useful in
830−-- implementing sequenced streams, for example, TCP reassembly.
831−--
832−-- /Unimplemented/
833−--
834−{-# INLINE reassembleBy #-}
835−reassembleBy
836− :: -- Monad m =>
837− Fold m a b
838− -> (a -> a -> Int)
839− -> Stream m a
840− -> Stream m b
841−reassembleBy = undefined
842−
843−------------------------------------------------------------------------------
844−-- Position Indexing
845−------------------------------------------------------------------------------
846−
847−-- |
848−-- >>> f = Fold.foldl' (\(i, _) x -> (i + 1, x)) (-1,undefined)
849−-- >>> indexed = Stream.postscan f
850−-- >>> indexed = Stream.zipWith (,) (Stream.enumerateFrom 0)
851−-- >>> indexedR n = fmap (\(i, a) -> (n - i, a)) . indexed
852−--
853−-- Pair each element in a stream with its index, starting from index 0.
854−--
855−-- >>> Stream.fold Fold.toList $ Stream.indexed $ Stream.fromList "hello"
856−-- [(0,'h'),(1,'e'),(2,'l'),(3,'l'),(4,'o')]
857−--
858−{-# INLINE indexed #-}
859−indexed :: Monad m => Stream m a -> Stream m (Int, a)
860−-- indexed = scanMaybe FL.indexing
861−indexed = fromStreamD . D.indexed . toStreamD
862−
863−-- |
864−-- >>> f n = Fold.foldl' (\(i, _) x -> (i - 1, x)) (n + 1,undefined)
865−-- >>> indexedR n = Stream.postscan (f n)
866−--
867−-- >>> s n = Stream.enumerateFromThen n (n - 1)
868−-- >>> indexedR n = Stream.zipWith (,) (s n)
869−--
870−-- Pair each element in a stream with its index, starting from the
871−-- given index @n@ and counting down.
872−--
873−-- >>> Stream.fold Fold.toList $ Stream.indexedR 10 $ Stream.fromList "hello"
874−-- [(10,'h'),(9,'e'),(8,'l'),(7,'l'),(6,'o')]
875−--
876−{-# INLINE indexedR #-}
877−indexedR :: Monad m => Int -> Stream m a -> Stream m (Int, a)
878−-- indexedR n = scanMaybe (FL.indexingRev n)
879−indexedR n = fromStreamD . D.indexedR n . toStreamD
880−
881−-------------------------------------------------------------------------------
882−-- Time Indexing
883−-------------------------------------------------------------------------------
884−
885−-- Note: The timestamp stream must be the second stream in the zip so that the
886−-- timestamp is generated after generating the stream element and not before.
887−-- If we do not do that then the following example will generate the same
888−-- timestamp for first two elements:
889−--
890−-- Stream.fold Fold.toList $ Stream.timestamped $ Stream.delay $ Stream.enumerateFromTo 1 3
891−--
892−-- | Pair each element in a stream with an absolute timestamp, using a clock of
893−-- specified granularity. The timestamp is generated just before the element
894−-- is consumed.
895−--
896−-- >>> Stream.fold Fold.toList $ Stream.timestampWith 0.01 $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
897−-- [(AbsTime (TimeSpec {sec = ..., nsec = ...}),1),(AbsTime (TimeSpec {sec = ..., nsec = ...}),2),(AbsTime (TimeSpec {sec = ..., nsec = ...}),3)]
898−--
899−-- /Pre-release/
900−--
901−{-# INLINE timestampWith #-}
902−timestampWith :: (MonadIO m)
903− => Double -> Stream m a -> Stream m (AbsTime, a)
904−timestampWith g stream = zipWith (flip (,)) stream (absTimesWith g)
905−
906−-- TBD: check performance vs a custom implementation without using zipWith.
907−--
908−-- /Pre-release/
909−--
910−{-# INLINE timestamped #-}
911−timestamped :: (MonadIO m)
912− => Stream m a -> Stream m (AbsTime, a)
913−timestamped = timestampWith 0.01
914−
915−-- | Pair each element in a stream with relative times starting from 0, using a
916−-- clock with the specified granularity. The time is measured just before the
917−-- element is consumed.
918−--
919−-- >>> Stream.fold Fold.toList $ Stream.timeIndexWith 0.01 $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
920−-- [(RelTime64 (NanoSecond64 ...),1),(RelTime64 (NanoSecond64 ...),2),(RelTime64 (NanoSecond64 ...),3)]
921−--
922−-- /Pre-release/Monad
923−--
924−{-# INLINE timeIndexWith #-}
925−timeIndexWith :: (MonadIO m)
926− => Double -> Stream m a -> Stream m (RelTime64, a)
927−timeIndexWith g stream = zipWith (flip (,)) stream (relTimesWith g)
928−
929−-- | Pair each element in a stream with relative times starting from 0, using a
930−-- 10 ms granularity clock. The time is measured just before the element is
931−-- consumed.
932−--
933−-- >>> Stream.fold Fold.toList $ Stream.timeIndexed $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
934−-- [(RelTime64 (NanoSecond64 ...),1),(RelTime64 (NanoSecond64 ...),2),(RelTime64 (NanoSecond64 ...),3)]
935−--
936−-- /Pre-release/
937−--
938−{-# INLINE timeIndexed #-}
939−timeIndexed :: (MonadIO m)
940− => Stream m a -> Stream m (RelTime64, a)
941−timeIndexed = timeIndexWith 0.01
942−
943−------------------------------------------------------------------------------
944−-- Searching
945−------------------------------------------------------------------------------
946−
947−-- | Find all the indices where the value of the element in the stream is equal
948−-- to the given value.
949−--
950−-- >>> elemIndices a = Stream.findIndices (== a)
951−--
952−{-# INLINE elemIndices #-}
953−elemIndices :: (Monad m, Eq a) => a -> Stream m a -> Stream m Int
954−elemIndices a = findIndices (== a)
955−
956−------------------------------------------------------------------------------
957−-- Rolling map
958−------------------------------------------------------------------------------
959−
960−-- XXX this is not a one-to-one map so calling it map may not be right.
961−-- We can perhaps call it zipWithTail or rollWith.
962−
963−-- | Apply a function on every two successive elements of a stream. The first
964−-- argument of the map function is the previous element and the second argument
965−-- is the current element. When the current element is the first element, the
966−-- previous element is 'Nothing'.
967−--
968−-- /Pre-release/
969−--
970−{-# INLINE rollingMap #-}
971−rollingMap :: Monad m => (Maybe a -> a -> b) -> Stream m a -> Stream m b
972−-- rollingMap f = scanMaybe (FL.slide2 $ Window.rollingMap f)
973−rollingMap f m = fromStreamD $ D.rollingMap f $ toStreamD m
974−
975−-- | Like 'rollingMap' but with an effectful map function.
976−--
977−-- /Pre-release/
978−--
979−{-# INLINE rollingMapM #-}
980−rollingMapM :: Monad m => (Maybe a -> a -> m b) -> Stream m a -> Stream m b
981−-- rollingMapM f = scanMaybe (FL.slide2 $ Window.rollingMapM f)
982−rollingMapM f m = fromStreamD $ D.rollingMapM f $ toStreamD m
983−
984−-- | Like 'rollingMap' but requires at least two elements in the stream,
985−-- returns an empty stream otherwise.
986−--
987−-- This is the stream equivalent of the list idiom @zipWith f xs (tail xs)@.
988−--
989−-- /Pre-release/
990−--
991−{-# INLINE rollingMap2 #-}
992−rollingMap2 :: Monad m => (a -> a -> b) -> Stream m a -> Stream m b
993−rollingMap2 f m = fromStreamD $ D.rollingMap2 f $ toStreamD m
994−
995−------------------------------------------------------------------------------
996−-- Maybe Streams
997−------------------------------------------------------------------------------
998−
999−-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'
1000−-- elements, and return a stream of values extracted from 'Just'.
1001−--
1002−-- Equivalent to:
1003−--
1004−-- >>> mapMaybe f = Stream.catMaybes . fmap f
1005−--
1006−{-# INLINE mapMaybe #-}
1007−mapMaybe :: Monad m => (a -> Maybe b) -> Stream m a -> Stream m b
1008−mapMaybe f m = fromStreamD $ D.mapMaybe f $ toStreamD m
1009−
1010−-- | Like 'mapMaybe' but maps a monadic function.
1011−--
1012−-- Equivalent to:
1013−--
1014−-- >>> mapMaybeM f = Stream.catMaybes . Stream.mapM f
1015−--
1016−-- >>> mapM f = Stream.mapMaybeM (\x -> Just <$> f x)
1017−--
1018−{-# INLINE_EARLY mapMaybeM #-}
1019−mapMaybeM :: Monad m
1020− => (a -> m (Maybe b)) -> Stream m a -> Stream m b
1021−mapMaybeM f = fmap fromJust . filter isJust . mapM f
1022−
1023−------------------------------------------------------------------------------
1024−-- Either streams
1025−------------------------------------------------------------------------------
1026−
1027−-- | Discard 'Right's and unwrap 'Left's in an 'Either' stream.
1028−--
1029−-- >>> catLefts = fmap (fromLeft undefined) . Stream.filter isLeft
1030−--
1031−-- /Pre-release/
1032−--
1033−{-# INLINE catLefts #-}
1034−catLefts :: Monad m => Stream m (Either a b) -> Stream m a
1035−catLefts = fmap (fromLeft undefined) . filter isLeft
1036−
1037−-- | Discard 'Left's and unwrap 'Right's in an 'Either' stream.
1038−--
1039−-- >>> catRights = fmap (fromRight undefined) . Stream.filter isRight
1040−--
1041−-- /Pre-release/
1042−--
1043−{-# INLINE catRights #-}
1044−catRights :: Monad m => Stream m (Either a b) -> Stream m b
1045−catRights = fmap (fromRight undefined) . filter isRight
1046−
1047−-- | Remove the either wrapper and flatten both lefts and as well as rights in
1048−-- the output stream.
1049−--
1050−-- >>> catEithers = fmap (either id id)
1051−--
1052−-- /Pre-release/
1053−--
1054−{-# INLINE catEithers #-}
1055−catEithers :: Monad m => Stream m (Either a a) -> Stream m a
1056−catEithers = fmap (either id id)