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
63import Control.Monad.IO.Class (MonadIO(..))
64import Streamly.Internal.Data.Fold.Type (Fold (..))
65import Streamly.Internal.Data.Time.Units (AbsTime, RelTime64, addToAbsTime64)
66import Streamly.Internal.Data.Unboxed (Unboxed)
67import Streamly.Internal.System.IO (defaultChunkSize)
69import qualified Streamly.Internal.Data.Array.Unboxed.Type as A
70import qualified Streamly.Internal.Data.Fold as Fold
71import qualified Streamly.Internal.Data.Stream.Common as Common
72import qualified Streamly.Internal.Data.Stream.StreamK as K
73import qualified Streamly.Internal.Data.Stream.StreamD as D
75import Prelude hiding (take, takeWhile, drop, reverse, concatMap, map, zipWith)
77import Streamly.Internal.Data.Stream.Type
82-- >>> import Control.Monad (join)
83-- >>> import Control.Monad.Trans.Class (lift)
84-- >>> import Data.Function (fix, (&))
85-- >>> import Data.Maybe (fromJust, isJust)
86-- >>> import Data.Semigroup (cycle1)
87-- >>> import Prelude hiding (take, takeWhile, drop, reverse)
88-- >>> import System.IO.Unsafe (unsafePerformIO)
89-- >>> import Streamly.Internal.Data.Stream as Stream
90-- >>> import qualified Streamly.Data.Array.Unboxed as Array
91-- >>> import qualified Streamly.Internal.Data.Fold as Fold
92-- >>> import qualified Streamly.Internal.Data.Parser as Parser
93-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
95------------------------------------------------------------------------------
97------------------------------------------------------------------------------
100-- >>> fromList = Prelude.foldr Stream.cons Stream.nil
102-- Construct a stream from a list of pure values. This is more efficient than
106{-# INLINE fromList #-}
107fromList :: Monad m => [a] -> Stream m a
108fromList = fromStreamK . Common.fromList
110------------------------------------------------------------------------------
111-- Generation - Time related
112------------------------------------------------------------------------------
114-- | @timesWith g@ returns a stream of time value tuples. The first component
115-- of the tuple is an absolute time reference (epoch) denoting the start of the
116-- stream and the second component is a time relative to the reference.
118-- The argument @g@ specifies the granularity of the relative time in seconds.
119-- A lower granularity clock gives higher precision but is more expensive in
120-- terms of CPU usage. Any granularity lower than 1 ms is treated as 1 ms.
122-- >>> import Control.Concurrent (threadDelay)
123-- >>> f = Fold.drainBy (\x -> print x >> threadDelay 1000000)
124-- >>> Stream.fold f $ Stream.take 3 $ Stream.timesWith 0.01
125-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
126-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
127-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
129-- Note: This API is not safe on 32-bit machines.
133{-# INLINE timesWith #-}
134timesWith :: MonadIO m => Double -> Stream m (AbsTime, RelTime64)
135timesWith g = fromStreamD $ D.times g
137-- | @absTimesWith g@ returns a stream of absolute timestamps using a clock of
138-- granularity @g@ specified in seconds. A low granularity clock is more
139-- expensive in terms of CPU usage. Any granularity lower than 1 ms is treated
142-- >>> f = Fold.drainBy print
143-- >>> Stream.fold f $ Stream.delayPre 1 $ Stream.take 3 $ Stream.absTimesWith 0.01
144-- AbsTime (TimeSpec {sec = ..., nsec = ...})
145-- AbsTime (TimeSpec {sec = ..., nsec = ...})
146-- AbsTime (TimeSpec {sec = ..., nsec = ...})
148-- Note: This API is not safe on 32-bit machines.
152{-# INLINE absTimesWith #-}
153absTimesWith :: MonadIO m => Double -> Stream m AbsTime
154absTimesWith = fmap (uncurry addToAbsTime64) . timesWith
156-- | @relTimesWith g@ returns a stream of relative time values starting from 0,
157-- using a clock of granularity @g@ specified in seconds. A low granularity
158-- clock is more expensive in terms of CPU usage. Any granularity lower than 1
159-- ms is treated as 1 ms.
161-- >>> f = Fold.drainBy print
162-- >>> Stream.fold f $ Stream.delayPre 1 $ Stream.take 3 $ Stream.relTimesWith 0.01
163-- RelTime64 (NanoSecond64 ...)
164-- RelTime64 (NanoSecond64 ...)
165-- RelTime64 (NanoSecond64 ...)
167-- Note: This API is not safe on 32-bit machines.
171{-# INLINE relTimesWith #-}
172relTimesWith :: MonadIO m => Double -> Stream m RelTime64
173relTimesWith = fmap snd . timesWith
175------------------------------------------------------------------------------
176-- Elimination - Running a Fold
177------------------------------------------------------------------------------
179-- | We can create higher order folds using 'foldContinue'. We can fold a
180-- number of streams to a given fold efficiently with full stream fusion. For
181-- example, to fold a list of streams on the same sum fold:
183-- >>> concatFold = Prelude.foldl Stream.foldContinue Fold.sum
185-- >>> fold f = Fold.finish . Stream.foldContinue f
188{-# INLINE foldContinue #-}
189foldContinue :: Monad m => Fold m a b -> Stream m a -> Fold m a b
190foldContinue f s = D.foldContinue f $ toStreamD s
192-- | Fold a stream using the supplied left 'Fold' and reducing the resulting
193-- expression strictly at each step. The behavior is similar to 'foldl''. A
194-- 'Fold' can terminate early without consuming the full stream. See the
195-- documentation of individual 'Fold's for termination behavior.
197-- >>> Stream.fold Fold.sum (Stream.enumerateFromTo 1 100)
200-- Folds never fail, therefore, they produce a default value even when no input
201-- is provided. It means we can always fold an empty stream and get a valid
202-- result. For example:
204-- >>> Stream.fold Fold.sum Stream.nil
207-- >>> fold f = Stream.parse (Parser.fromFold f)
211fold :: Monad m => Fold m a b -> Stream m a -> m b
212fold fl strm = D.fold fl $ D.fromStreamK $ toStreamK strm
214-- | Like 'fold' but also returns the remaining stream.
219{-# INLINE foldBreak #-}
220foldBreak :: Monad m => Fold m a b -> Stream m a -> m (b, Stream m a)
222-- XXX This shows quadratic performance when used recursively perhaps because
223-- of StreamK to StreamD conversions not getting eliminated sue to recursion.
224foldBreak fl (Stream strm) = fmap f $ D.foldBreak fl $ D.fromStreamK strm
228 f (b, str) = (b, Stream (D.toStreamK str))
230foldBreak fl strm = fmap f $ K.foldBreak fl (toStreamK strm)
234 f (b, str) = (b, fromStreamK str)
236------------------------------------------------------------------------------
238------------------------------------------------------------------------------
246-- >>> Stream.fold Fold.toList $ fmap (+1) $ Stream.fromList [1,2,3]
250map :: Monad m => (a -> b) -> Stream m a -> Stream m b
251map f = fromStreamD . D.map f . toStreamD
254-- | Postscan a stream using the given monadic fold.
256-- The following example extracts the input stream up to a point where the
257-- running average of elements is no more than 10:
259-- >>> import Data.Maybe (fromJust)
260-- >>> let avg = Fold.teeWith (/) Fold.sum (fmap fromIntegral Fold.length)
261-- >>> s = Stream.enumerateFromTo 1.0 100.0
263-- Stream.fold Fold.toList
264-- $ fmap (fromJust . fst)
265-- $ Stream.takeWhile (\(_,x) -> x <= 10)
266-- $ Stream.postscan (Fold.tee Fold.last avg) s
268-- [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]
271{-# INLINE postscan #-}
272postscan :: Monad m => Fold m a b -> Stream m a -> Stream m b
273postscan fld = fromStreamD . D.postscanOnce fld . toStreamD
277-- The stateful step function can be simplified to @(s -> a -> m b)@ to provide
278-- a read-only environment. However, that would just be 'mapM'.
280-- The initial action could be @m (s, Maybe b)@, and we can also add a final
281-- action @s -> m (Maybe b)@. This can be used to get pre/post scan like
282-- functionality and also to flush the state in the end like scanlMAfter'.
283-- We can also use it along with a fusible version of bracket to get
284-- scanlMAfter' like functionality. See issue #677.
286-- This can be further generalized to a type similar to Fold/Parser, giving it
287-- filtering and parsing capability as well (this is in fact equivalent to
290-- smapM :: (s -> a -> m (Step s b)) -> m s -> Stream m a -> Stream m b
293-- | A stateful 'mapM', equivalent to a left scan, more like mapAccumL.
294-- Hopefully, this is a better alternative to @scan@. Separation of state from
295-- the output makes it easier to think in terms of a shared state, and also
296-- makes it easier to keep the state fully strict and the output lazy.
298-- See also: 'postscan'
308smapM step initial stream =
309 -- XXX implement this directly instead of using postscan
311 (\(s, _) a -> step s a)
312 (fmap (,undefined) initial)
313 in fmap snd $ postscan f stream
315-- | In a stream of 'Maybe's, discard 'Nothing's and unwrap 'Just's.
317-- >>> catMaybes = Stream.mapMaybe id
318-- >>> catMaybes = fmap fromJust . Stream.filter isJust
322{-# INLINE catMaybes #-}
323catMaybes :: Monad m => Stream m (Maybe a) -> Stream m a
324-- catMaybes = fmap fromJust . filter isJust
325catMaybes = fromStreamD . D.catMaybes . toStreamD
327-- | Use a filtering fold on a stream.
329-- >>> scanMaybe f = Stream.catMaybes . Stream.postscan f
332{-# INLINE scanMaybe #-}
333scanMaybe :: Monad m => Fold m a (Maybe b) -> Stream m a -> Stream m b
334scanMaybe p = catMaybes . postscan p
336------------------------------------------------------------------------------
337-- Transformation - Trimming
338------------------------------------------------------------------------------
340-- | Take first 'n' elements from the stream and discard the rest.
344take :: Monad m => Int -> Stream m a -> Stream m a
345-- take n = scanMaybe (Fold.taking n)
346take n m = fromStreamD $ D.take n $ toStreamD m
348-- | End the stream as soon as the predicate fails on an element.
351{-# INLINE takeWhile #-}
352takeWhile :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
353-- takeWhile p = scanMaybe (Fold.takingEndBy_ (not . p))
354takeWhile p m = fromStreamD $ D.takeWhile p $ toStreamD m
356{-# INLINE takeEndBy #-}
357takeEndBy :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
358-- takeEndBy p = scanMaybe (Fold.takingEndBy p)
359takeEndBy p m = fromStreamD $ D.takeEndBy p $ toStreamD m
361-- | Discard first 'n' elements from the stream and take the rest.
365drop :: Monad m => Int -> Stream m a -> Stream m a
366-- drop n = scanMaybe (Fold.dropping n)
367drop n m = fromStreamD $ D.drop n $ toStreamD m
369------------------------------------------------------------------------------
371------------------------------------------------------------------------------
373-- | Find all the indices where the element in the stream satisfies the given
376-- >>> findIndices p = Stream.fold (Fold.findIndices p)
379{-# INLINE findIndices #-}
380findIndices :: Monad m => (a -> Bool) -> Stream m a -> Stream m Int
381-- findIndices p = scanMaybe (Fold.findIndices p)
382findIndices p m = fromStreamD $ D.findIndices p (toStreamD m)
384------------------------------------------------------------------------------
385-- Transformation by Inserting
386------------------------------------------------------------------------------
388-- intersperseM = intersperseBySpan 1
390-- | Insert an effect and its output before consuming an element of a stream
391-- except the first one.
393-- >>> input = Stream.fromList "hello"
394-- >>> Stream.fold Fold.toList $ Stream.trace putChar $ Stream.intersperseM (putChar '.' >> return ',') input
395-- h.,e.,l.,l.,o"h,e,l,l,o"
397-- Be careful about the order of effects. In the above example we used trace
398-- after the intersperse, if we use it before the intersperse the output would
399-- be he.l.l.o."h,e,l,l,o".
401-- >>> Stream.fold Fold.toList $ Stream.intersperseM (putChar '.' >> return ',') $ Stream.trace putChar input
402-- he.l.l.o."h,e,l,l,o"
405{-# INLINE intersperseM #-}
406intersperseM :: Monad m => m a -> Stream m a -> Stream m a
407intersperseM m = fromStreamD . D.intersperseM m . toStreamD
409------------------------------------------------------------------------------
410-- Transformation by Reordering
411------------------------------------------------------------------------------
413-- XXX Use a compact region list to temporarily store the list, in both reverse
414-- as well as in reverse'.
416-- /Note:/ 'reverse'' is much faster than this, use that when performance
419-- | Returns the elements of the stream in reverse order. The stream must be
420-- finite. Note that this necessarily buffers the entire stream in memory.
422-- >>> reverse = Stream.foldlT (flip Stream.cons) Stream.nil
425{-# INLINE reverse #-}
426reverse :: Monad m => Stream m a -> Stream m a
427reverse s = fromStreamD $ D.reverse $ toStreamD s
429-- | Like 'reverse' but several times faster, requires a 'Storable' instance.
432{-# INLINE reverse' #-}
433reverse' :: (MonadIO m, Unboxed a) => Stream m a -> Stream m a
434-- reverse' s = fromStreamD $ D.reverse' $ toStreamD s
437 . A.flattenArraysRev -- unfoldMany A.readRev
441 . A.arraysOf defaultChunkSize
444------------------------------------------------------------------------------
445-- Combine streams and flatten
446------------------------------------------------------------------------------
448-- | Map a stream producing monadic function on each element of the stream
449-- and then flatten the results into a single stream. Since the stream
450-- generation function is monadic, unlike 'concatMap', it can produce an
451-- effect at the beginning of each iteration of the inner loop.
454{-# INLINE concatMapM #-}
455concatMapM :: Monad m => (a -> m (Stream m b)) -> Stream m a -> Stream m b
456concatMapM f m = fromStreamD $ D.concatMapM (fmap toStreamD . f) (toStreamD m)
458-- | Map a stream producing function on each element of the stream and then
459-- flatten the results into a single stream.
461-- >>> concatMap f = Stream.concatMapM (return . f)
462-- >>> concatMap f = Stream.concatMapWith Stream.append f
463-- >>> concatMap f = Stream.concat . fmap f
464-- >>> concatMap f = Stream.unfoldMany (Unfold.lmap f Unfold.fromStream)
467{-# INLINE concatMap #-}
468concatMap ::Monad m => (a -> Stream m b) -> Stream m a -> Stream m b
469concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)
471-- | Given a stream value in the underlying monad, lift and join the underlying
472-- monad with the stream monad.
474-- >>> concatM = Stream.concat . Stream.fromEffect
475-- >>> concatM = Stream.concat . lift -- requires (MonadTrans t)
476-- >>> concatM = join . lift -- requires (MonadTrans t, Monad (Stream m))
478-- See also: 'concat', 'sequence'
482{-# INLINE concatM #-}
483concatM :: Monad m => m (Stream m a) -> Stream m a
484concatM generator = concatMapM (\() -> generator) (fromPure ())
486-- | Like 'foldMany' but appends empty fold output if the fold and stream
487-- termination aligns:
489-- >>> f = Fold.take 2 Fold.sum
490-- >>> Stream.fold Fold.toList $ Stream.foldManyPost f $ Stream.fromList []
492-- >>> Stream.fold Fold.toList $ Stream.foldManyPost f $ Stream.fromList [1..9]
494-- >>> Stream.fold Fold.toList $ Stream.foldManyPost f $ Stream.fromList [1..10]
499{-# INLINE foldManyPost #-}
505foldManyPost f m = fromStreamD $ D.foldManyPost f (toStreamD m)
507------------------------------------------------------------------------------
509------------------------------------------------------------------------------
511-- | Like 'zipWith' but using a monadic zipping function.
514{-# INLINE zipWithM #-}
515zipWithM :: Monad m =>
516 (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c
517zipWithM f m1 m2 = fromStreamD $ D.zipWithM f (toStreamD m1) (toStreamD m2)
519-- | Stream @a@ is evaluated first, followed by stream @b@, the resulting
520-- elements @a@ and @b@ are then zipped using the supplied zip function and the
521-- result @c@ is yielded to the consumer.
523-- If stream @a@ or stream @b@ ends, the zipped stream ends. If stream @b@ ends
524-- first, the element @a@ from previous evaluation of stream @a@ is discarded.
526-- >>> s1 = Stream.fromList [1,2,3]
527-- >>> s2 = Stream.fromList [4,5,6]
528-- >>> Stream.fold Fold.toList $ Stream.zipWith (+) s1 s2
532{-# INLINE zipWith #-}
533zipWith :: Monad m => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
534zipWith f m1 m2 = fromStreamD $ D.zipWith f (toStreamD m1) (toStreamD 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