2-- Module : Streamly.Internal.Data.Stream.Bottom
3-- Copyright : (c) 2017 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
9-- Bottom level Stream module that can be used by all other upper level
12module Streamly.Internal.Data.Stream.Bottom
71import Control.Monad.IO.Class (MonadIO(..))
72import GHC.Types (SPEC(..))
73import Streamly.Internal.Data.Fold.Type (Fold (..))
74import Streamly.Internal.Data.Time.Units (AbsTime, RelTime64, addToAbsTime64)
75import Streamly.Internal.Data.Unbox (Unbox)
76import Streamly.Internal.Data.Producer.Type (Producer(..))
77import Streamly.Internal.System.IO (defaultChunkSize)
78import Streamly.Internal.Data.SVar.Type (defState)
80import qualified Streamly.Internal.Data.Array.Type as A
81import qualified Streamly.Internal.Data.Fold as Fold
82import qualified Streamly.Internal.Data.StreamK as K
83import qualified Streamly.Internal.Data.Stream as D
85import Prelude hiding (take, takeWhile, drop, reverse, concatMap, map, zipWith)
87import Streamly.Internal.Data.Stream.Type
92-- >>> import Control.Monad (join, (>=>), (<=<))
93-- >>> import Data.Function (fix, (&))
94-- >>> import Data.Functor.Identity (Identity)
95-- >>> import Data.Maybe (fromJust, isJust)
96-- >>> import Prelude hiding (take, takeWhile, drop, reverse)
97-- >>> import Streamly.Data.Array (Array)
98-- >>> import Streamly.Data.Fold (Fold)
99-- >>> import Streamly.Data.Stream (Stream)
100-- >>> import System.IO.Unsafe (unsafePerformIO)
101-- >>> import qualified Streamly.Data.Array as Array
102-- >>> import qualified Streamly.Data.MutArray as MArray
103-- >>> import qualified Streamly.Data.Fold as Fold
104-- >>> import qualified Streamly.Data.Parser as Parser
105-- >>> import qualified Streamly.Data.Unfold as Unfold
106-- >>> import qualified Streamly.Internal.Data.Fold as Fold (toStream)
107-- >>> import Streamly.Internal.Data.Stream as Stream
109------------------------------------------------------------------------------
110-- Generation - Time related
111------------------------------------------------------------------------------
113-- | @timesWith g@ returns a stream of time value tuples. The first component
114-- of the tuple is an absolute time reference (epoch) denoting the start of the
115-- stream and the second component is a time relative to the reference.
117-- The argument @g@ specifies the granularity of the relative time in seconds.
118-- A lower granularity clock gives higher precision but is more expensive in
119-- terms of CPU usage. Any granularity lower than 1 ms is treated as 1 ms.
121-- >>> import Control.Concurrent (threadDelay)
122-- >>> f = Fold.drainMapM (\x -> print x >> threadDelay 1000000)
123-- >>> Stream.fold f $ Stream.take 3 $ Stream.timesWith 0.01
124-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
125-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
126-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
128-- Note: This API is not safe on 32-bit machines.
132{-# INLINE timesWith #-}
133timesWith :: MonadIO m => Double -> Stream m (AbsTime, RelTime64)
134timesWith g = fromStreamD $ D.timesWith g
136-- | @absTimesWith g@ returns a stream of absolute timestamps using a clock of
137-- granularity @g@ specified in seconds. A low granularity clock is more
138-- expensive in terms of CPU usage. Any granularity lower than 1 ms is treated
141-- >>> f = Fold.drainMapM print
142-- >>> Stream.fold f $ Stream.delayPre 1 $ Stream.take 3 $ Stream.absTimesWith 0.01
143-- AbsTime (TimeSpec {sec = ..., nsec = ...})
144-- AbsTime (TimeSpec {sec = ..., nsec = ...})
145-- AbsTime (TimeSpec {sec = ..., nsec = ...})
147-- Note: This API is not safe on 32-bit machines.
151{-# INLINE absTimesWith #-}
152absTimesWith :: MonadIO m => Double -> Stream m AbsTime
153absTimesWith = fmap (uncurry addToAbsTime64) . timesWith
155-- | @relTimesWith g@ returns a stream of relative time values starting from 0,
156-- using a clock of granularity @g@ specified in seconds. A low granularity
157-- clock is more expensive in terms of CPU usage. Any granularity lower than 1
158-- ms is treated as 1 ms.
160-- >>> f = Fold.drainMapM print
161-- >>> Stream.fold f $ Stream.delayPre 1 $ Stream.take 3 $ Stream.relTimesWith 0.01
162-- RelTime64 (NanoSecond64 ...)
163-- RelTime64 (NanoSecond64 ...)
164-- RelTime64 (NanoSecond64 ...)
166-- Note: This API is not safe on 32-bit machines.
170{-# INLINE relTimesWith #-}
171relTimesWith :: MonadIO m => Double -> Stream m RelTime64
172relTimesWith = fmap snd . timesWith
174------------------------------------------------------------------------------
175-- Elimination - Running a Fold
176------------------------------------------------------------------------------
178-- | Append a stream to a fold lazily to build an accumulator incrementally.
180-- Example, to continue folding a list of streams on the same sum fold:
182-- >>> streams = [Stream.fromList [1..5], Stream.fromList [6..10]]
183-- >>> f = Prelude.foldl Stream.foldAddLazy Fold.sum streams
184-- >>> Stream.fold f Stream.nil
187{-# INLINE foldAddLazy #-}
188foldAddLazy :: Monad m => Fold m a b -> Stream m a -> Fold m a b
189foldAddLazy f s = D.foldAddLazy f $ toStreamD s
191-- >>> foldAdd f = Stream.foldAddLazy f >=> Fold.reduce
194-- >>> foldAdd = flip Fold.addStream
196foldAdd :: Monad m => Fold m a b -> Stream m a -> m (Fold m a b)
197foldAdd f = fold (Fold.duplicate f)
199-- >>> fold f = Fold.extractM . Stream.foldAddLazy f
200-- >>> fold f = Stream.fold Fold.one . Stream.foldManyPost f
201-- >>> fold f = Fold.extractM <=< Stream.foldAdd f
203-- | Fold a stream using the supplied left 'Fold' and reducing the resulting
204-- expression strictly at each step. The behavior is similar to 'foldl''. A
205-- 'Fold' can terminate early without consuming the full stream. See the
206-- documentation of individual 'Fold's for termination behavior.
210-- >>> fold f = fmap fst . Stream.foldBreak f
211-- >>> fold f = Stream.parse (Parser.fromFold f)
215-- >>> Stream.fold Fold.sum (Stream.enumerateFromTo 1 100)
219fold :: Monad m => Fold m a b -> Stream m a -> m b
220fold fl strm = D.fold fl $ D.fromStreamK $ toStreamK strm
222-- Alternative name foldSome, but may be confused vs foldMany.
224-- | Like 'fold' but also returns the remaining stream. The resulting stream
225-- would be 'Stream.nil' if the stream finished before the fold.
229{-# INLINE foldBreak #-}
230foldBreak :: Monad m => Fold m a b -> Stream m a -> m (b, Stream m a)
231foldBreak fl strm = fmap f $ K.foldBreak fl (toStreamK strm)
235 f (b, str) = (b, fromStreamK str)
237-- XXX The quadratic slowdown in recursive use is because recursive function
238-- cannot be inlined and StreamD/StreamK conversions pile up and cannot be
239-- eliminated by rewrite rules.
241-- | Like 'foldBreak' but fuses.
243-- /Note:/ Unlike 'foldBreak', recursive application on the resulting stream
244-- would lead to quadratic slowdown. If you need recursion with fusion (within
245-- one iteration of recursion) use StreamD.foldBreak directly.
248{-# INLINE foldBreak2 #-}
249foldBreak2 :: Monad m => Fold m a b -> Stream m a -> m (b, Stream m a)
250foldBreak2 fl strm = fmap f $ D.foldBreak fl $ toStreamD strm
254 f (b, str) = (b, fromStreamD str)
256-- | Fold resulting in either breaking the stream or continuation of the fold.
257-- Instead of supplying the input stream in one go we can run the fold multiple
258-- times, each time supplying the next segment of the input stream. If the fold
259-- has not yet finished it returns a fold that can be run again otherwise it
260-- returns the fold result and the residual stream.
263{-# INLINE foldEither #-}
264foldEither :: Monad m =>
265 Fold m a b -> Stream m a -> m (Either (Fold m a b) (b, Stream m a))
266foldEither fl strm = fmap (fmap f) $ K.foldEither fl $ toStreamK strm
270 f (b, str) = (b, fromStreamK str)
272-- | Like 'foldEither' but fuses. However, recursive application on resulting
273-- stream would lead to quadratic slowdown.
276{-# INLINE foldEither2 #-}
277foldEither2 :: Monad m =>
278 Fold m a b -> Stream m a -> m (Either (Fold m a b) (b, Stream m a))
279foldEither2 fl strm = fmap (fmap f) $ D.foldEither fl $ toStreamD strm
283 f (b, str) = (b, fromStreamD str)
285-- XXX Array folds can be implemented using this.
286-- foldContainers? Specialized to foldArrays.
288-- | Generate streams from individual elements of a stream and fold the
289-- concatenation of those streams using the supplied fold. Return the result of
290-- the fold and residual stream.
292-- For example, this can be used to efficiently fold an Array Word8 stream
295-- The outer stream forces CPS to allow scalable appends and the inner stream
296-- forces direct style for stream fusion.
299{-# INLINE foldConcat #-}
300foldConcat :: Monad m =>
301 Producer m a b -> Fold m b c -> Stream m a -> m (c, Stream m a)
303 (Producer pstep pinject pextract)
304 (Fold fstep begin done)
309 Fold.Partial fs -> go fs streamK
310 Fold.Done fb -> return (fb, fromStreamK streamK)
314 streamK = toStreamK stream
319 return (r, fromStreamK K.nil)
322 res <- go1 SPEC acc st
326 return (r, fromStreamK K.nil)
329 return (b, fromStreamK (K.fromPure x))
332 res <- go1 SPEC acc st
337 return (b, fromStreamK (x `K.cons` r))
338 in K.foldStream defState yieldk single stop m1
347 Fold.Done b -> return $ Right (b, s)
348 Fold.Partial fs1 -> go1 SPEC fs1 s
349 D.Skip s -> go1 SPEC fs s
350 D.Stop -> return $ Left fs
352------------------------------------------------------------------------------
354------------------------------------------------------------------------------
362-- >>> Stream.fold Fold.toList $ fmap (+1) $ Stream.fromList [1,2,3]
366map :: Monad m => (a -> b) -> Stream m a -> Stream m b
367map f = fromStreamD . D.map f . toStreamD
370-- | Postscan a stream using the given monadic fold.
372-- The following example extracts the input stream up to a point where the
373-- running average of elements is no more than 10:
375-- >>> import Data.Maybe (fromJust)
376-- >>> let avg = Fold.teeWith (/) Fold.sum (fmap fromIntegral Fold.length)
377-- >>> s = Stream.enumerateFromTo 1.0 100.0
379-- Stream.fold Fold.toList
380-- $ fmap (fromJust . fst)
381-- $ Stream.takeWhile (\(_,x) -> x <= 10)
382-- $ Stream.postscan (Fold.tee Fold.latest avg) s
384-- [1.0,2.0,3.0,4.0,5.0,6.0,7.0,8.0,9.0,10.0,11.0,12.0,13.0,14.0,15.0,16.0,17.0,18.0,19.0]
386{-# INLINE postscan #-}
387postscan :: Monad m => Fold m a b -> Stream m a -> Stream m b
388postscan fld = fromStreamD . D.postscan fld . toStreamD
392-- The stateful step function can be simplified to @(s -> a -> m b)@ to provide
393-- a read-only environment. However, that would just be 'mapM'.
395-- The initial action could be @m (s, Maybe b)@, and we can also add a final
396-- action @s -> m (Maybe b)@. This can be used to get pre/post scan like
397-- functionality and also to flush the state in the end like scanlMAfter'.
398-- We can also use it along with a fusible version of bracket to get
399-- scanlMAfter' like functionality. See issue #677.
401-- This can be further generalized to a type similar to Fold/Parser, giving it
402-- filtering and parsing capability as well (this is in fact equivalent to
405-- smapM :: (s -> a -> m (Step s b)) -> m s -> Stream m a -> Stream m b
408-- | A stateful 'mapM', equivalent to a left scan, more like mapAccumL.
409-- Hopefully, this is a better alternative to @scan@. Separation of state from
410-- the output makes it easier to think in terms of a shared state, and also
411-- makes it easier to keep the state fully strict and the output lazy.
413-- See also: 'postscan'
423smapM step initial stream =
424 -- XXX implement this directly instead of using postscan
426 (\(s, _) a -> step s a)
427 (fmap (,undefined) initial)
428 in fmap snd $ postscan f stream
430-- | In a stream of 'Maybe's, discard 'Nothing's and unwrap 'Just's.
432-- >>> catMaybes = Stream.mapMaybe id
433-- >>> catMaybes = fmap fromJust . Stream.filter isJust
437{-# INLINE catMaybes #-}
438catMaybes :: Monad m => Stream m (Maybe a) -> Stream m a
439-- catMaybes = fmap fromJust . filter isJust
440catMaybes = fromStreamD . D.catMaybes . toStreamD
442-- | Use a filtering fold on a stream.
444-- >>> scanMaybe f = Stream.catMaybes . Stream.postscan f
446{-# INLINE scanMaybe #-}
447scanMaybe :: Monad m => Fold m a (Maybe b) -> Stream m a -> Stream m b
448scanMaybe p = catMaybes . postscan p
450------------------------------------------------------------------------------
451-- Transformation - Trimming
452------------------------------------------------------------------------------
454-- | Take first 'n' elements from the stream and discard the rest.
457take :: Monad m => Int -> Stream m a -> Stream m a
458-- take n = scanMaybe (Fold.taking n)
459take n m = fromStreamD $ D.take n $ toStreamD m
461-- | End the stream as soon as the predicate fails on an element.
463{-# INLINE takeWhile #-}
464takeWhile :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
465-- takeWhile p = scanMaybe (Fold.takingEndBy_ (not . p))
466takeWhile p m = fromStreamD $ D.takeWhile p $ toStreamD m
468{-# INLINE takeEndBy #-}
469takeEndBy :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
470-- takeEndBy p = scanMaybe (Fold.takingEndBy p)
471takeEndBy p m = fromStreamD $ D.takeEndBy p $ toStreamD m
473-- | Discard first 'n' elements from the stream and take the rest.
476drop :: Monad m => Int -> Stream m a -> Stream m a
477-- drop n = scanMaybe (Fold.dropping n)
478drop n m = fromStreamD $ D.drop n $ toStreamD m
480------------------------------------------------------------------------------
482------------------------------------------------------------------------------
484-- | Find all the indices where the element in the stream satisfies the given
487-- >>> findIndices p = Stream.scanMaybe (Fold.findIndices p)
489{-# INLINE findIndices #-}
490findIndices :: Monad m => (a -> Bool) -> Stream m a -> Stream m Int
491-- findIndices p = scanMaybe (Fold.findIndices p)
492findIndices p m = fromStreamD $ D.findIndices p (toStreamD m)
494------------------------------------------------------------------------------
495-- Transformation by Inserting
496------------------------------------------------------------------------------
498-- intersperseM = intersperseMWith 1
500-- | Insert an effect and its output before consuming an element of a stream
501-- except the first one.
503-- >>> input = Stream.fromList "hello"
504-- >>> Stream.fold Fold.toList $ Stream.trace putChar $ Stream.intersperseM (putChar '.' >> return ',') input
505-- h.,e.,l.,l.,o"h,e,l,l,o"
507-- Be careful about the order of effects. In the above example we used trace
508-- after the intersperse, if we use it before the intersperse the output would
509-- be he.l.l.o."h,e,l,l,o".
511-- >>> Stream.fold Fold.toList $ Stream.intersperseM (putChar '.' >> return ',') $ Stream.trace putChar input
512-- he.l.l.o."h,e,l,l,o"
514{-# INLINE intersperseM #-}
515intersperseM :: Monad m => m a -> Stream m a -> Stream m a
516intersperseM m = fromStreamD . D.intersperseM m . toStreamD
518------------------------------------------------------------------------------
519-- Transformation by Reordering
520------------------------------------------------------------------------------
522-- XXX Use a compact region list to temporarily store the list, in both reverse
523-- as well as in reverse'.
525-- /Note:/ 'reverse'' is much faster than this, use that when performance
528-- | Returns the elements of the stream in reverse order. The stream must be
529-- finite. Note that this necessarily buffers the entire stream in memory.
531-- >>> reverse = Stream.foldlT (flip Stream.cons) Stream.nil
533{-# INLINE reverse #-}
534reverse :: Stream m a -> Stream m a
535reverse s = fromStreamK $ K.reverse $ toStreamK s
537-- | Like 'reverse' but several times faster, requires a 'Storable' instance.
542{-# INLINE reverse' #-}
543reverse' :: (MonadIO m, Unbox a) => Stream m a -> Stream m a
544-- reverse' s = fromStreamD $ D.reverse' $ toStreamD s
547 . A.flattenArraysRev -- unfoldMany A.readRev
551 . A.chunksOf defaultChunkSize
554------------------------------------------------------------------------------
555-- Combine streams and flatten
556------------------------------------------------------------------------------
558-- | Map a stream producing monadic function on each element of the stream
559-- and then flatten the results into a single stream. Since the stream
560-- generation function is monadic, unlike 'concatMap', it can produce an
561-- effect at the beginning of each iteration of the inner loop.
563-- See 'unfoldMany' for a fusible alternative.
565{-# INLINE concatMapM #-}
566concatMapM :: Monad m => (a -> m (Stream m b)) -> Stream m a -> Stream m b
567concatMapM f m = fromStreamD $ D.concatMapM (fmap toStreamD . f) (toStreamD m)
569-- | Map a stream producing function on each element of the stream and then
570-- flatten the results into a single stream.
572-- >>> concatMap f = Stream.concatMapM (return . f)
573-- >>> concatMap f = Stream.concatMapWith Stream.append f
574-- >>> concatMap f = Stream.concat . fmap f
575-- >>> concatMap f = Stream.unfoldMany (Unfold.lmap f Unfold.fromStream)
577-- See 'unfoldMany' for a fusible alternative.
579{-# INLINE concatMap #-}
580concatMap ::Monad m => (a -> Stream m b) -> Stream m a -> Stream m b
581concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)
583-- >>> concatEffect = Stream.concat . lift -- requires (MonadTrans t)
584-- >>> concatEffect = join . lift -- requires (MonadTrans t, Monad (Stream m))
586-- | Given a stream value in the underlying monad, lift and join the underlying
587-- monad with the stream monad.
589-- >>> concatEffect = Stream.concat . Stream.fromEffect
591-- See also: 'concat', 'sequence'
593-- See 'concatEffect2' for a fusible alternative.
597{-# INLINE concatEffect #-}
598concatEffect :: Monad m => m (Stream m a) -> Stream m a
599concatEffect generator =
600 fromStreamK $ K.concatEffect $ fmap toStreamK generator
602{-# INLINE concatEffect2 #-}
603concatEffect2 :: Monad m => m (Stream m a) -> Stream m a
604-- concatEffect generator = concatMapM (\() -> generator) (fromPure ())
605concatEffect2 generator =
606 fromStreamD $ D.concatEffect $ fmap toStreamD generator
608-- XXX Need a more intuitive name, and need to reconcile the names
609-- foldMany/fold/parse/parseMany/parseManyPost etc.
611-- | Like 'foldMany' but evaluates the fold before the stream, and yields its
612-- output even if the stream is empty, therefore, always results in a non-empty
613-- output even on an empty stream (default result of the fold).
615-- Example, empty stream:
617-- >>> f = Fold.take 2 Fold.sum
618-- >>> fmany = Stream.fold Fold.toList . Stream.foldManyPost f
619-- >>> fmany $ Stream.fromList []
622-- Example, last fold empty:
624-- >>> fmany $ Stream.fromList [1..4]
627-- Example, last fold non-empty:
629-- >>> fmany $ Stream.fromList [1..5]
632-- Note that using a closed fold e.g. @Fold.take 0@, would result in an
633-- infinite stream without consuming the input.
637{-# INLINE foldManyPost #-}
643foldManyPost f m = fromStreamD $ D.foldManyPost f (toStreamD m)
645------------------------------------------------------------------------------
647------------------------------------------------------------------------------
649-- | Like 'zipWith' but using a monadic zipping function.
651{-# INLINE zipWithM #-}
652zipWithM :: Monad m =>
653 (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c
654zipWithM f m1 m2 = fromStreamK $ K.zipWithM f (toStreamK m1) (toStreamK m2)
656-- | Stream @a@ is evaluated first, followed by stream @b@, the resulting
657-- elements @a@ and @b@ are then zipped using the supplied zip function and the
658-- result @c@ is yielded to the consumer.
660-- If stream @a@ or stream @b@ ends, the zipped stream ends. If stream @b@ ends
661-- first, the element @a@ from previous evaluation of stream @a@ is discarded.
663-- >>> s1 = Stream.fromList [1,2,3]
664-- >>> s2 = Stream.fromList [4,5,6]
665-- >>> Stream.fold Fold.toList $ Stream.zipWith (+) s1 s2
668{-# INLINE zipWith #-}
669zipWith :: Monad m => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
670zipWith f m1 m2 = fromStreamK $ K.zipWith f (toStreamK m1) (toStreamK m2)
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list