on acc37cf help sign in
768 lines 26 kB history raw
1{-# OPTIONS_GHC -Wno-orphans #-}
3-- |
4-- Module : Streamly.Internal.Data.Stream.IsStream.Common
5-- Copyright : (c) 2017 Composewell Technologies
6-- License : BSD-3-Clause
7-- Maintainer : streamly@composewell.com
8-- Stability : experimental
9-- Portability : GHC
10--
11-- Bottom level IsStream module that can be used by all other upper level
12-- IsStream modules.
14module Streamly.Internal.Data.Stream.IsStream.Common
15 (
16 -- * Generation
17 fromPure
18 , fromEffect
19 , repeatM
20 , timesWith
21 , absTimesWith
22 , relTimesWith
24 -- * Elimination
25 , foldContinue
26 , fold
27 , foldBreak
29 -- * Transformation
30 , map
31 , scanlMAfter'
32 , postscanlMAfter'
33 , postscanlM'
34 , smapM
35 , foldManyPost
36 -- $smapM_Notes
37 , take
38 , takeWhile
39 , takeEndBy
40 , drop
41 , findIndices
42 , intersperseM
43 , interjectSuffix
44 , reverse
45 , reverse'
47 -- * Concurrent
48 , mkAsync
49 , mkParallel
50 , parallelFst
52 -- * Nesting
53 , concatM
54 , concatMapM
55 , concatMap
56 , splitOnSeq
58 -- * Zipping
59 , zipWithM
60 , zipWith
62 -- * Deprecated
63 , yield
64 , yieldM
65 )
66where
68#include "inline.hs"
70import Control.Concurrent (threadDelay)
71import Control.Monad.IO.Class (MonadIO(..))
72import Foreign.Storable (Storable)
73import Streamly.Internal.Control.Concurrent (MonadAsync)
74import Streamly.Internal.Data.Array.Foreign.Type (Array)
75import Streamly.Internal.Data.Fold.Type (Fold (..))
76import Streamly.Internal.Data.Stream.IsStream.Combinators (maxYields)
77import Streamly.Internal.Data.Stream.IsStream.Type
78 (IsStream(..), fromStreamD, toStreamD, fromStreamS, toStreamS)
79import Streamly.Internal.Data.Stream.Serial (SerialT(..))
80import Streamly.Internal.Data.Time.Units (AbsTime, RelTime64, addToAbsTime64)
82import qualified Streamly.Internal.Data.Array.Foreign.Type as A
83import qualified Streamly.Internal.Data.Stream.Async as Async
84import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
85import qualified Streamly.Internal.Data.Stream.Parallel as Par
86import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
87import qualified Streamly.Internal.Data.Stream.StreamK as K
88import qualified Streamly.Internal.Data.Stream.StreamD as D
89#ifdef USE_STREAMK_ONLY
90import qualified Streamly.Internal.Data.Stream.StreamK as S
91#else
92import qualified Streamly.Internal.Data.Stream.StreamD as S
93#endif
94import Streamly.Internal.System.IO (defaultChunkSize)
96import Prelude hiding (take, takeWhile, drop, reverse, concatMap, map, zipWith)
98--
99-- $setup
100-- >>> :m
101-- >>> import Control.Concurrent (threadDelay)
102-- >>> import Control.Monad (join)
103-- >>> import Control.Monad.Trans.Class (lift)
104-- >>> import Data.Function (fix, (&))
105-- >>> import Data.Semigroup (cycle1)
106-- >>> import Prelude hiding (take, takeWhile, drop, reverse)
107-- >>> import System.IO.Unsafe (unsafePerformIO)
108-- >>> import qualified Streamly.Prelude as Stream
109-- >>> import Streamly.Internal.Data.Stream.IsStream as Stream
110-- >>> import qualified Streamly.Data.Array.Foreign as Array
111-- >>> import qualified Streamly.Data.Fold as Fold
112-- >>> import qualified Streamly.Internal.Data.Fold as Fold
113-- >>> import qualified Streamly.Data.Unfold as Unfold
115------------------------------------------------------------------------------
116-- Generation
117------------------------------------------------------------------------------
119-- Faster than fromEffect because there is no bind.
120--
121-- |
122-- @
123-- fromPure a = a \`cons` nil
124-- @
125--
126-- Create a singleton stream from a pure value.
127--
128-- The following holds in monadic streams, but not in Zip streams:
129--
130-- @
131-- fromPure = pure
132-- fromPure = fromEffect . pure
133-- @
134--
135-- In Zip applicative streams 'fromPure' is not the same as 'pure' because in that
136-- case 'pure' is equivalent to 'repeat' instead. 'fromPure' and 'pure' are
137-- equally efficient, in other cases 'fromPure' may be slightly more efficient
138-- than the other equivalent definitions.
139--
140-- /Since: 0.8.0 (Renamed yield to fromPure)/
141--
142{-# INLINE fromPure #-}
143fromPure :: IsStream t => a -> t m a
144fromPure = fromStream . K.fromPure
146-- | Same as 'fromPure'
147--
148-- @since 0.4.0
149{-# DEPRECATED yield "Please use fromPure instead." #-}
150{-# INLINE yield #-}
151yield :: IsStream t => a -> t m a
152yield = fromPure
154-- |
155-- @
156-- fromEffect m = m \`consM` nil
157-- @
158--
159-- Create a singleton stream from a monadic action.
160--
161-- @
162-- > Stream.toList $ Stream.fromEffect getLine
163-- hello
164-- ["hello"]
165-- @
166--
167-- /Since: 0.8.0 (Renamed yieldM to fromEffect)/
168--
169{-# INLINE fromEffect #-}
170fromEffect :: (Monad m, IsStream t) => m a -> t m a
171fromEffect = fromStream . K.fromEffect
173-- | Same as 'fromEffect'
174--
175-- @since 0.4.0
176{-# DEPRECATED yieldM "Please use fromEffect instead." #-}
177{-# INLINE yieldM #-}
178yieldM :: (Monad m, IsStream t) => m a -> t m a
179yieldM = fromEffect
180-- |
181-- >>> repeatM = fix . consM
182-- >>> repeatM = cycle1 . fromEffect
183--
184-- Generate a stream by repeatedly executing a monadic action forever.
185--
186-- >>> :{
187-- repeatAsync =
188-- Stream.repeatM (threadDelay 1000000 >> print 1)
189-- & Stream.take 10
190-- & Stream.fromAsync
191-- & Stream.drain
192-- :}
193--
194-- /Concurrent, infinite (do not use with 'fromParallel')/
195--
196-- @since 0.2.0
197{-# INLINE_EARLY repeatM #-}
198repeatM :: (IsStream t, MonadAsync m) => m a -> t m a
199repeatM = K.repeatMWith IsStream.consM
201{-# RULES "repeatM serial" repeatM = repeatMSerial #-}
202{-# INLINE repeatMSerial #-}
203repeatMSerial :: MonadAsync m => m a -> SerialT m a
204repeatMSerial = fromStreamS . S.repeatM
206------------------------------------------------------------------------------
207-- Generation - Time related
208------------------------------------------------------------------------------
210-- | @timesWith g@ returns a stream of time value tuples. The first component
211-- of the tuple is an absolute time reference (epoch) denoting the start of the
212-- stream and the second component is a time relative to the reference.
213--
214-- The argument @g@ specifies the granularity of the relative time in seconds.
215-- A lower granularity clock gives higher precision but is more expensive in
216-- terms of CPU usage. Any granularity lower than 1 ms is treated as 1 ms.
217--
218-- >>> import Control.Concurrent (threadDelay)
219-- >>> import Streamly.Internal.Data.Stream.IsStream.Common as Stream (timesWith)
220-- >>> Stream.mapM_ (\x -> print x >> threadDelay 1000000) $ Stream.take 3 $ Stream.timesWith 0.01
221-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
222-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
223-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
224--
225-- Note: This API is not safe on 32-bit machines.
226--
227-- /Pre-release/
228--
229{-# INLINE timesWith #-}
230timesWith :: (IsStream t, MonadAsync m) => Double -> t m (AbsTime, RelTime64)
231timesWith g = fromStreamD $ D.times g
233-- | @absTimesWith g@ returns a stream of absolute timestamps using a clock of
234-- granularity @g@ specified in seconds. A low granularity clock is more
235-- expensive in terms of CPU usage. Any granularity lower than 1 ms is treated
236-- as 1 ms.
237--
238-- >>> Stream.mapM_ print $ Stream.delayPre 1 $ Stream.take 3 $ absTimesWith 0.01
239-- AbsTime (TimeSpec {sec = ..., nsec = ...})
240-- AbsTime (TimeSpec {sec = ..., nsec = ...})
241-- AbsTime (TimeSpec {sec = ..., nsec = ...})
242--
243-- Note: This API is not safe on 32-bit machines.
244--
245-- /Pre-release/
246--
247{-# INLINE absTimesWith #-}
248absTimesWith :: (IsStream t, MonadAsync m, Functor (t m))
249 => Double -> t m AbsTime
250absTimesWith = fmap (uncurry addToAbsTime64) . timesWith
252-- | @relTimesWith g@ returns a stream of relative time values starting from 0,
253-- using a clock of granularity @g@ specified in seconds. A low granularity
254-- clock is more expensive in terms of CPU usage. Any granularity lower than 1
255-- ms is treated as 1 ms.
256--
257-- >>> Stream.mapM_ print $ Stream.delayPre 1 $ Stream.take 3 $ Stream.relTimesWith 0.01
258-- RelTime64 (NanoSecond64 ...)
259-- RelTime64 (NanoSecond64 ...)
260-- RelTime64 (NanoSecond64 ...)
261--
262-- Note: This API is not safe on 32-bit machines.
263--
264-- /Pre-release/
265--
266{-# INLINE relTimesWith #-}
267relTimesWith :: (IsStream t, MonadAsync m, Functor (t m))
268 => Double -> t m RelTime64
269relTimesWith = fmap snd . timesWith
271------------------------------------------------------------------------------
272-- Elimination - Running a Fold
273------------------------------------------------------------------------------
275-- | We can create higher order folds using 'foldContinue'. We can fold a
276-- number of streams to a given fold efficiently with full stream fusion. For
277-- example, to fold a list of streams on the same sum fold:
278--
279-- >>> concatFold = Prelude.foldl Stream.foldContinue Fold.sum
280--
281-- >>> fold f = Fold.finish . Stream.foldContinue f
282--
283-- /Internal/
284{-# INLINE foldContinue #-}
285foldContinue :: Monad m => Fold m a b -> SerialT m a -> Fold m a b
286foldContinue f s = D.foldContinue f $ IsStream.toStreamD s
288-- | Fold a stream using the supplied left 'Fold' and reducing the resulting
289-- expression strictly at each step. The behavior is similar to 'foldl''. A
290-- 'Fold' can terminate early without consuming the full stream. See the
291-- documentation of individual 'Fold's for termination behavior.
292--
293-- >>> Stream.fold Fold.sum (Stream.enumerateFromTo 1 100)
294-- 5050
295--
296-- Folds never fail, therefore, they produce a default value even when no input
297-- is provided. It means we can always fold an empty stream and get a valid
298-- result. For example:
299--
300-- >>> Stream.fold Fold.sum Stream.nil
301-- 0
302--
303-- However, 'foldMany' on an empty stream results in an empty stream.
304-- Therefore, @Stream.fold f@ is not the same as @Stream.head . Stream.foldMany
305-- f@.
306--
307-- @fold f = Stream.parse (Parser.fromFold f)@
308--
309-- @since 0.7.0
310{-# INLINE fold #-}
311fold :: Monad m => Fold m a b -> SerialT m a -> m b
312fold fl (SerialT strm) = D.fold fl $ D.fromStreamK strm
314-- | Like 'fold' but also returns the remaining stream.
315--
316-- /Inhibits stream fusion/
317--
318{-# INLINE foldBreak #-}
319foldBreak :: Monad m => Fold m a b -> SerialT m a -> m (b, SerialT m a)
320{-
321-- XXX This shows quadratic performance when used recursively perhaps because
322-- of StreamK to StreamD conversions not getting eliminated sue to recursion.
323foldBreak fl (SerialT strm) = fmap f $ D.foldBreak fl $ D.fromStreamK strm
325 where
327 f (b, str) = (b, SerialT (D.toStreamK str))
328-}
329foldBreak fl (SerialT strm) = fmap f $ K.foldBreak fl strm
331 where
333 f (b, str) = (b, SerialT str)
335------------------------------------------------------------------------------
336-- Transformation
337------------------------------------------------------------------------------
339-- |
340-- @
341-- map = fmap
342-- @
343--
344-- Same as 'fmap'.
345--
346-- @
347-- > S.toList $ S.map (+1) $ S.fromList [1,2,3]
348-- [2,3,4]
349-- @
350--
351-- @since 0.4.0
352{-# INLINE map #-}
353map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b
354map f = fromStreamD . D.map f . toStreamD
356-- | @scanlMAfter' accumulate initial done stream@ is like 'scanlM'' except
357-- that it provides an additional @done@ function to be applied on the
358-- accumulator when the stream stops. The result of @done@ is also emitted in
359-- the stream.
360--
361-- This function can be used to allocate a resource in the beginning of the
362-- scan and release it when the stream ends or to flush the internal state of
363-- the scan at the end.
364--
365-- /Pre-release/
366--
367{-# INLINE scanlMAfter' #-}
368scanlMAfter' :: (IsStream t, Monad m)
369 => (b -> a -> m b) -> m b -> (b -> m b) -> t m a -> t m b
370scanlMAfter' step initial done stream =
371 fromStreamD $ D.scanlMAfter' step initial done $ toStreamD stream
373{-# INLINE postscanlMAfter' #-}
374postscanlMAfter' :: (IsStream t, Monad m)
375 => (b -> a -> m b) -> m b -> (b -> m b) -> t m a -> t m b
376postscanlMAfter' step initial done stream =
377 fromStreamD $ D.postscanlMAfter' step initial done $ toStreamD stream
379-- XXX this needs to be concurrent
380-- | Like 'postscanl'' but with a monadic step function and a monadic seed.
381--
382-- >>> postscanlM' f z xs = Stream.drop 1 $ Stream.scanlM' f z xs
383--
384-- /Since: 0.7.0/
385--
386-- /Since: 0.8.0 (signature change)/
387{-# INLINE postscanlM' #-}
388postscanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
389postscanlM' step z m = fromStreamD $ D.postscanlM' step z $ toStreamD m
391-- $smapM_Notes
392--
393-- The stateful step function can be simplified to @(s -> a -> m b)@ to provide
394-- a read-only environment. However, that would just be 'mapM'.
395--
396-- The initial action could be @m (s, Maybe b)@, and we can also add a final
397-- action @s -> m (Maybe b)@. This can be used to get pre/post scan like
398-- functionality and also to flush the state in the end like scanlMAfter'.
399-- We can also use it along with a fusible version of bracket to get
400-- scanlMAfter' like functionality. See issue #677.
401--
402-- This can be further generalized to a type similar to Fold/Parser, giving it
403-- filtering and parsing capability as well (this is in fact equivalent to
404-- parseMany):
405--
406-- smapM :: (s -> a -> m (Step s b)) -> m s -> t m a -> t m b
407--
409-- | A stateful 'mapM', equivalent to a left scan, more like mapAccumL.
410-- Hopefully, this is a better alternative to @scan@. Separation of state from
411-- the output makes it easier to think in terms of a shared state, and also
412-- makes it easier to keep the state fully strict and the output lazy.
413--
414-- See also: 'scanlM''
415--
416-- /Pre-release/
417--
418{-# INLINE smapM #-}
419smapM :: (IsStream t, Monad m) =>
420 (s -> a -> m (s, b))
421 -> m s
422 -> t m a
423 -> t m b
424smapM step initial stream =
425 -- XXX implement this directly instead of using scanlM'
426 -- Once we have postscanlM' with monadic initial we can use this code
427 -- let r = postscanlM'
428 -- (\(s, _) a -> step s a)
429 -- (fmap (,undefined) initial)
430 -- stream
431 let r = postscanlM'
432 (\(s, _) a -> step s a)
433 (fmap (,undefined) initial)
434 stream
435 in map snd r
437------------------------------------------------------------------------------
438-- Transformation - Trimming
439------------------------------------------------------------------------------
441-- | Take first 'n' elements from the stream and discard the rest.
442--
443-- @since 0.1.0
444{-# INLINE take #-}
445take :: (IsStream t, Monad m) => Int -> t m a -> t m a
446take n m = fromStreamS $ S.take n $ toStreamS
447 (maxYields (Just (fromIntegral n)) m)
449-- | End the stream as soon as the predicate fails on an element.
450--
451-- @since 0.1.0
452{-# INLINE takeWhile #-}
453takeWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
454takeWhile p m = fromStreamS $ S.takeWhile p $ toStreamS m
456{-# INLINE takeEndBy #-}
457takeEndBy :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
458takeEndBy p m = fromStreamD $ D.takeEndBy p $ toStreamD m
460-- | Discard first 'n' elements from the stream and take the rest.
461--
462-- @since 0.1.0
463{-# INLINE drop #-}
464drop :: (IsStream t, Monad m) => Int -> t m a -> t m a
465drop n m = fromStreamS $ S.drop n $ toStreamS m
467------------------------------------------------------------------------------
468-- Searching
469------------------------------------------------------------------------------
471-- | Find all the indices where the element in the stream satisfies the given
472-- predicate.
473--
474-- > findIndices = fold Fold.findIndices
475--
476-- @since 0.5.0
477{-# INLINE findIndices #-}
478findIndices :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m Int
479findIndices p m = fromStreamS $ S.findIndices p (toStreamS m)
481------------------------------------------------------------------------------
482-- Transformation by Inserting
483------------------------------------------------------------------------------
485-- intersperseM = intersperseBySpan 1
487-- | Insert an effect and its output before consuming an element of a stream
488-- except the first one.
489--
490-- >>> Stream.toList $ Stream.trace putChar $ Stream.intersperseM (putChar '.' >> return ',') $ Stream.fromList "hello"
491-- h.,e.,l.,l.,o"h,e,l,l,o"
492--
493-- Be careful about the order of effects. In the above example we used trace
494-- after the intersperse, if we use it before the intersperse the output would
495-- be he.l.l.o."h,e,l,l,o".
496--
497-- >>> Stream.toList $ Stream.intersperseM (putChar '.' >> return ',') $ Stream.trace putChar $ Stream.fromList "hello"
498-- he.l.l.o."h,e,l,l,o"
499--
500-- @since 0.5.0
501{-# INLINE intersperseM #-}
502intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
503intersperseM m = fromStreamS . S.intersperseM m . toStreamS
505-- | Intersperse a monadic action into the input stream after every @n@
506-- seconds.
507--
508-- @
509-- > import Control.Concurrent (threadDelay)
510-- > Stream.drain $ Stream.interjectSuffix 1 (putChar ',') $ Stream.mapM (\x -> threadDelay 1000000 >> putChar x) $ Stream.fromList "hello"
511-- h,e,l,l,o
512-- @
513--
514-- /Pre-release/
515{-# INLINE interjectSuffix #-}
516interjectSuffix
517 :: (IsStream t, MonadAsync m)
518 => Double -> m a -> t m a -> t m a
519interjectSuffix n f xs = xs `parallelFst` repeatM timed
520 where timed = liftIO (threadDelay (round $ n * 1000000)) >> f
522------------------------------------------------------------------------------
523-- Transformation by Reordering
524------------------------------------------------------------------------------
526-- XXX Use a compact region list to temporarily store the list, in both reverse
527-- as well as in reverse'.
528--
529-- /Note:/ 'reverse'' is much faster than this, use that when performance
530-- matters.
531--
532-- | Returns the elements of the stream in reverse order. The stream must be
533-- finite. Note that this necessarily buffers the entire stream in memory.
534--
535-- >>> reverse = Stream.foldlT (flip Stream.cons) Stream.nil
536--
537-- /Since 0.7.0 (Monad m constraint)/
538--
539-- /Since: 0.1.1/
540{-# INLINE reverse #-}
541reverse :: (IsStream t, Monad m) => t m a -> t m a
542reverse s = fromStreamS $ S.reverse $ toStreamS s
544-- | Like 'reverse' but several times faster, requires a 'Storable' instance.
545--
546-- /Pre-release/
547{-# INLINE reverse' #-}
548reverse' :: (IsStream t, MonadIO m, Storable a) => t m a -> t m a
549-- reverse' s = fromStreamD $ D.reverse' $ toStreamD s
550reverse' =
551 fromStreamD
552 . A.flattenArraysRev -- unfoldMany A.readRev
553 . D.fromStreamK
554 . K.reverse
555 . D.toStreamK
556 . A.arraysOf defaultChunkSize
557 . toStreamD
559------------------------------------------------------------------------------
560-- Concurrent Transformations and Combining
561------------------------------------------------------------------------------
563-- | Make the stream producer and consumer run concurrently by introducing a
564-- buffer between them. The producer thread evaluates the input stream until
565-- the buffer fills, it terminates if the buffer is full and a worker thread is
566-- kicked off again to evaluate the remaining stream when there is space in the
567-- buffer. The consumer consumes the stream lazily from the buffer.
568--
569-- /Since: 0.2.0 (Streamly)/
570--
571-- @since 0.8.0
572--
573{-# INLINE_NORMAL mkAsync #-}
574mkAsync :: (IsStream t, MonadAsync m) => t m a -> t m a
575mkAsync = fromStreamD . Async.mkAsyncD . toStreamD
577-- Compare with mkAsync. mkAsync uses an Async style SVar whereas this uses a
578-- parallel style SVar for evaluation. Currently, parallel style cannot use
579-- rate control whereas Async style can use rate control. In async style SVar
580-- the worker thread terminates when the buffer is full whereas in Parallel
581-- style it blocks.
582--
583-- | Make the stream producer and consumer run concurrently by introducing a
584-- buffer between them. The producer thread evaluates the input stream until
585-- the buffer fills, it blocks if the buffer is full until there is space in
586-- the buffer. The consumer consumes the stream lazily from the buffer.
587--
588-- @mkParallel = IsStream.fromStreamD . mkParallelD . IsStream.toStreamD@
589--
590-- /Pre-release/
591--
592{-# INLINE_NORMAL mkParallel #-}
593mkParallel :: (IsStream t, MonadAsync m) => t m a -> t m a
594mkParallel = fromStreamD . Par.mkParallelD . toStreamD
596-- | Like `parallel` but stops the output as soon as the first stream stops.
597--
598-- /Pre-release/
599{-# INLINE parallelFst #-}
600parallelFst :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
601parallelFst m1 m2 = fromStream $ Par.parallelFstK (toStream m1) (toStream m2)
603------------------------------------------------------------------------------
604-- Combine streams and flatten
605------------------------------------------------------------------------------
607-- | Map a stream producing monadic function on each element of the stream
608-- and then flatten the results into a single stream. Since the stream
609-- generation function is monadic, unlike 'concatMap', it can produce an
610-- effect at the beginning of each iteration of the inner loop.
611--
612-- @since 0.6.0
613{-# INLINE concatMapM #-}
614concatMapM :: (IsStream t, Monad m) => (a -> m (t m b)) -> t m a -> t m b
615concatMapM f m = fromStreamD $ D.concatMapM (fmap toStreamD . f) (toStreamD m)
617-- | Map a stream producing function on each element of the stream and then
618-- flatten the results into a single stream.
619--
620-- >>> concatMap f = Stream.concatMapM (return . f)
621-- >>> concatMap f = Stream.concatMapWith Stream.serial f
622-- >>> concatMap f = Stream.concat . Stream.map f
623-- >>> concatMap f = Stream.unfoldMany (Unfold.lmap f Unfold.fromStream)
624--
625-- @since 0.6.0
626{-# INLINE concatMap #-}
627concatMap ::(IsStream t, Monad m) => (a -> t m b) -> t m a -> t m b
628concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)
630-- | Given a stream value in the underlying monad, lift and join the underlying
631-- monad with the stream monad.
632--
633-- >>> concatM = Stream.concat . Stream.fromEffect
634-- >>> concatM = Stream.concat . lift -- requires (MonadTrans t)
635-- >>> concatM = join . lift -- requires (MonadTrans t, Monad (t m))
636--
637-- See also: 'concat', 'sequence'
638--
639-- /Internal/
640--
641{-# INLINE concatM #-}
642concatM :: (IsStream t, Monad m) => m (t m a) -> t m a
643concatM generator = concatMapM (\() -> generator) (fromPure ())
645-- | Like 'foldMany' but appends empty fold output if the fold and stream
646-- termination aligns:
647--
648-- >>> f = Fold.take 2 Fold.sum
649-- >>> Stream.toList $ Stream.foldManyPost f $ Stream.fromList []
650-- [0]
651-- >>> Stream.toList $ Stream.foldManyPost f $ Stream.fromList [1..9]
652-- [3,7,11,15,9]
653-- >>> Stream.toList $ Stream.foldManyPost f $ Stream.fromList [1..10]
654-- [3,7,11,15,19,0]
655--
656-- /Pre-release/
657--
658{-# INLINE foldManyPost #-}
659foldManyPost
660 :: (IsStream t, Monad m)
661 => Fold m a b
662 -> t m a
663 -> t m b
664foldManyPost f m = fromStreamD $ D.foldManyPost f (toStreamD m)
666------------------------------------------------------------------------------
667-- Split stream and fold
668------------------------------------------------------------------------------
670-- $
671-- > intercalate . splitOnSeq == id
672-- prop> (\f -> unsafePerformIO . Stream.toList . f . Stream.fromList) (Stream.intercalate Unfold.fromList " " . Stream.splitOnSeq (Array.fromList " ") Fold.toList) xs == xs
674-- | Like 'splitOn' but the separator is a sequence of elements instead of a
675-- single element.
676--
677-- For illustration, let's define a function that operates on pure lists:
678--
679-- >>> splitOnSeq' pat xs = Stream.toList $ Stream.splitOnSeq (Array.fromList pat) Fold.toList (Stream.fromList xs)
680--
681-- >>> splitOnSeq' "" "hello"
682-- ["h","e","l","l","o"]
683--
684-- >>> splitOnSeq' "hello" ""
685-- [""]
686--
687-- >>> splitOnSeq' "hello" "hello"
688-- ["",""]
689--
690-- >>> splitOnSeq' "x" "hello"
691-- ["hello"]
692--
693-- >>> splitOnSeq' "h" "hello"
694-- ["","ello"]
695--
696-- >>> splitOnSeq' "o" "hello"
697-- ["hell",""]
698--
699-- >>> splitOnSeq' "e" "hello"
700-- ["h","llo"]
701--
702-- >>> splitOnSeq' "l" "hello"
703-- ["he","","o"]
704--
705-- >>> splitOnSeq' "ll" "hello"
706-- ["he","o"]
707--
708-- 'splitOnSeq' is an inverse of 'intercalate'. The following law always holds:
709--
710-- > intercalate . splitOnSeq == id
711--
712-- The following law holds when the separator is non-empty and contains none of
713-- the elements present in the input lists:
714--
715-- > splitOnSeq . intercalate == id
716--
717-- >>> splitOnSeq pat f = Stream.foldManyPost (Fold.takeEndBySeq_ pat f)
718--
719-- /Pre-release/
721-- XXX We can use a polymorphic vector implemented by Array# to represent the
722-- sequence, that way we can avoid the Storable constraint. If we still need
723-- Storable Array for performance, we can use a separate splitOnArray API for
724-- that. We can also have an API where the sequence itself is a lazy stream, so
725-- that we can search files in files for example.
726{-# INLINE splitOnSeq #-}
727splitOnSeq
728 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
729 => Array a -> Fold m a b -> t m a -> t m b
730splitOnSeq patt f m =
731 IsStream.fromStreamD $ D.splitOnSeq patt f (IsStream.toStreamD m)
732-- XXX This may have a problem if the stream terminates and we start extracting
733-- the fold and then the fold terminates before consuming all the buffered
734-- input. We have no way to supply the buffered input back to the driver.
735-- splitOnSeq patt f =
736 -- foldManyPost (Fold.takeEndBySeq_ patt f)
738------------------------------------------------------------------------------
739-- Zipping
740------------------------------------------------------------------------------
742-- | Like 'zipWith' but using a monadic zipping function.
743--
744-- @since 0.4.0
745{-# INLINE zipWithM #-}
746zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c
747zipWithM f m1 m2 =
748 IsStream.fromStreamS
749 $ S.zipWithM f (IsStream.toStreamS m1) (IsStream.toStreamS m2)
751-- | Stream @a@ is evaluated first, followed by stream @b@, the resulting
752-- elements @a@ and @b@ are then zipped using the supplied zip function and the
753-- result @c@ is yielded to the consumer.
754--
755-- If stream @a@ or stream @b@ ends, the zipped stream ends. If stream @b@ ends
756-- first, the element @a@ from previous evaluation of stream @a@ is discarded.
757--
758-- @
759-- > S.toList $ S.zipWith (+) (S.fromList [1,2,3]) (S.fromList [4,5,6])
760-- [5,7,9]
761-- @
762--
763-- @since 0.1.0
764{-# INLINE zipWith #-}
765zipWith :: (IsStream t, Monad m) => (a -> b -> c) -> t m a -> t m b -> t m c
766zipWith f m1 m2 =
767 IsStream.fromStreamS
768 $ S.zipWith f (IsStream.toStreamS m1) (IsStream.toStreamS m2)