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 (Unbox)
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
105{-# INLINE fromList #-}
106fromList :: Monad m => [a] -> Stream m a
107fromList = fromStreamK . Common.fromList
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.drainBy (\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.times 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.drainBy 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.drainBy 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-- | We can create higher order folds using 'foldContinue'. We can fold a
179-- number of streams to a given fold efficiently with full stream fusion. For
180-- example, to fold a list of streams on the same sum fold:
182-- >>> concatFold = Prelude.foldl Stream.foldContinue Fold.sum
184-- >>> fold f = Fold.finish . Stream.foldContinue f
186{-# INLINE foldContinue #-}
187foldContinue :: Monad m => Fold m a b -> Stream m a -> Fold m a b
188foldContinue f s = D.foldContinue f $ toStreamD s
190-- | Fold a stream using the supplied left 'Fold' and reducing the resulting
191-- expression strictly at each step. The behavior is similar to 'foldl''. A
192-- 'Fold' can terminate early without consuming the full stream. See the
193-- documentation of individual 'Fold's for termination behavior.
195-- >>> Stream.fold Fold.sum (Stream.enumerateFromTo 1 100)
198-- Folds never fail, therefore, they produce a default value even when no input
199-- is provided. It means we can always fold an empty stream and get a valid
200-- result. For example:
202-- >>> Stream.fold Fold.sum Stream.nil
205-- >>> fold f = Stream.parse (Parser.fromFold f)
208fold :: Monad m => Fold m a b -> Stream m a -> m b
209fold fl strm = D.fold fl $ D.fromStreamK $ toStreamK strm
211-- | Like 'fold' but also returns the remaining stream.
215{-# INLINE foldBreak #-}
216foldBreak :: Monad m => Fold m a b -> Stream m a -> m (b, Stream m a)
218-- XXX This shows quadratic performance when used recursively perhaps because
219-- of StreamK to StreamD conversions not getting eliminated sue to recursion.
220foldBreak fl (Stream strm) = fmap f $ D.foldBreak fl $ D.fromStreamK strm
224 f (b, str) = (b, Stream (D.toStreamK str))
226foldBreak fl strm = fmap f $ K.foldBreak fl (toStreamK strm)
230 f (b, str) = (b, fromStreamK str)
232------------------------------------------------------------------------------
234------------------------------------------------------------------------------
242-- >>> Stream.fold Fold.toList $ fmap (+1) $ Stream.fromList [1,2,3]
246map :: Monad m => (a -> b) -> Stream m a -> Stream m b
247map f = fromStreamD . D.map f . toStreamD
250-- | Postscan a stream using the given monadic fold.
252-- The following example extracts the input stream up to a point where the
253-- running average of elements is no more than 10:
255-- >>> import Data.Maybe (fromJust)
256-- >>> let avg = Fold.teeWith (/) Fold.sum (fmap fromIntegral Fold.length)
257-- >>> s = Stream.enumerateFromTo 1.0 100.0
259-- Stream.fold Fold.toList
260-- $ fmap (fromJust . fst)
261-- $ Stream.takeWhile (\(_,x) -> x <= 10)
262-- $ Stream.postscan (Fold.tee Fold.last avg) s
264-- [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]
266{-# INLINE postscan #-}
267postscan :: Monad m => Fold m a b -> Stream m a -> Stream m b
268postscan fld = fromStreamD . D.postscanOnce fld . toStreamD
272-- The stateful step function can be simplified to @(s -> a -> m b)@ to provide
273-- a read-only environment. However, that would just be 'mapM'.
275-- The initial action could be @m (s, Maybe b)@, and we can also add a final
276-- action @s -> m (Maybe b)@. This can be used to get pre/post scan like
277-- functionality and also to flush the state in the end like scanlMAfter'.
278-- We can also use it along with a fusible version of bracket to get
279-- scanlMAfter' like functionality. See issue #677.
281-- This can be further generalized to a type similar to Fold/Parser, giving it
282-- filtering and parsing capability as well (this is in fact equivalent to
285-- smapM :: (s -> a -> m (Step s b)) -> m s -> Stream m a -> Stream m b
288-- | A stateful 'mapM', equivalent to a left scan, more like mapAccumL.
289-- Hopefully, this is a better alternative to @scan@. Separation of state from
290-- the output makes it easier to think in terms of a shared state, and also
291-- makes it easier to keep the state fully strict and the output lazy.
293-- See also: 'postscan'
303smapM step initial stream =
304 -- XXX implement this directly instead of using postscan
306 (\(s, _) a -> step s a)
307 (fmap (,undefined) initial)
308 in fmap snd $ postscan f stream
310-- | In a stream of 'Maybe's, discard 'Nothing's and unwrap 'Just's.
312-- >>> catMaybes = Stream.mapMaybe id
313-- >>> catMaybes = fmap fromJust . Stream.filter isJust
317{-# INLINE catMaybes #-}
318catMaybes :: Monad m => Stream m (Maybe a) -> Stream m a
319-- catMaybes = fmap fromJust . filter isJust
320catMaybes = fromStreamD . D.catMaybes . toStreamD
322-- | Use a filtering fold on a stream.
324-- >>> scanMaybe f = Stream.catMaybes . Stream.postscan f
326{-# INLINE scanMaybe #-}
327scanMaybe :: Monad m => Fold m a (Maybe b) -> Stream m a -> Stream m b
328scanMaybe p = catMaybes . postscan p
330------------------------------------------------------------------------------
331-- Transformation - Trimming
332------------------------------------------------------------------------------
334-- | Take first 'n' elements from the stream and discard the rest.
337take :: Monad m => Int -> Stream m a -> Stream m a
338-- take n = scanMaybe (Fold.taking n)
339take n m = fromStreamD $ D.take n $ toStreamD m
341-- | End the stream as soon as the predicate fails on an element.
343{-# INLINE takeWhile #-}
344takeWhile :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
345-- takeWhile p = scanMaybe (Fold.takingEndBy_ (not . p))
346takeWhile p m = fromStreamD $ D.takeWhile p $ toStreamD m
348{-# INLINE takeEndBy #-}
349takeEndBy :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
350-- takeEndBy p = scanMaybe (Fold.takingEndBy p)
351takeEndBy p m = fromStreamD $ D.takeEndBy p $ toStreamD m
353-- | Discard first 'n' elements from the stream and take the rest.
356drop :: Monad m => Int -> Stream m a -> Stream m a
357-- drop n = scanMaybe (Fold.dropping n)
358drop n m = fromStreamD $ D.drop n $ toStreamD m
360------------------------------------------------------------------------------
362------------------------------------------------------------------------------
364-- | Find all the indices where the element in the stream satisfies the given
367-- >>> findIndices p = Stream.fold (Fold.findIndices p)
369{-# INLINE findIndices #-}
370findIndices :: Monad m => (a -> Bool) -> Stream m a -> Stream m Int
371-- findIndices p = scanMaybe (Fold.findIndices p)
372findIndices p m = fromStreamD $ D.findIndices p (toStreamD m)
374------------------------------------------------------------------------------
375-- Transformation by Inserting
376------------------------------------------------------------------------------
378-- intersperseM = intersperseBySpan 1
380-- | Insert an effect and its output before consuming an element of a stream
381-- except the first one.
383-- >>> input = Stream.fromList "hello"
384-- >>> Stream.fold Fold.toList $ Stream.trace putChar $ Stream.intersperseM (putChar '.' >> return ',') input
385-- h.,e.,l.,l.,o"h,e,l,l,o"
387-- Be careful about the order of effects. In the above example we used trace
388-- after the intersperse, if we use it before the intersperse the output would
389-- be he.l.l.o."h,e,l,l,o".
391-- >>> Stream.fold Fold.toList $ Stream.intersperseM (putChar '.' >> return ',') $ Stream.trace putChar input
392-- he.l.l.o."h,e,l,l,o"
394{-# INLINE intersperseM #-}
395intersperseM :: Monad m => m a -> Stream m a -> Stream m a
396intersperseM m = fromStreamD . D.intersperseM m . toStreamD
398------------------------------------------------------------------------------
399-- Transformation by Reordering
400------------------------------------------------------------------------------
402-- XXX Use a compact region list to temporarily store the list, in both reverse
403-- as well as in reverse'.
405-- /Note:/ 'reverse'' is much faster than this, use that when performance
408-- | Returns the elements of the stream in reverse order. The stream must be
409-- finite. Note that this necessarily buffers the entire stream in memory.
411-- >>> reverse = Stream.foldlT (flip Stream.cons) Stream.nil
413{-# INLINE reverse #-}
414reverse :: Monad m => Stream m a -> Stream m a
415reverse s = fromStreamD $ D.reverse $ toStreamD s
417-- | Like 'reverse' but several times faster, requires a 'Storable' instance.
420{-# INLINE reverse' #-}
421reverse' :: (MonadIO m, Unbox a) => Stream m a -> Stream m a
422-- reverse' s = fromStreamD $ D.reverse' $ toStreamD s
425 . A.flattenArraysRev -- unfoldMany A.readRev
429 . A.arraysOf defaultChunkSize
432------------------------------------------------------------------------------
433-- Combine streams and flatten
434------------------------------------------------------------------------------
436-- | Map a stream producing monadic function on each element of the stream
437-- and then flatten the results into a single stream. Since the stream
438-- generation function is monadic, unlike 'concatMap', it can produce an
439-- effect at the beginning of each iteration of the inner loop.
441{-# INLINE concatMapM #-}
442concatMapM :: Monad m => (a -> m (Stream m b)) -> Stream m a -> Stream m b
443concatMapM f m = fromStreamD $ D.concatMapM (fmap toStreamD . f) (toStreamD m)
445-- | Map a stream producing function on each element of the stream and then
446-- flatten the results into a single stream.
448-- >>> concatMap f = Stream.concatMapM (return . f)
449-- >>> concatMap f = Stream.concatMapWith Stream.append f
450-- >>> concatMap f = Stream.concat . fmap f
451-- >>> concatMap f = Stream.unfoldMany (Unfold.lmap f Unfold.fromStream)
453{-# INLINE concatMap #-}
454concatMap ::Monad m => (a -> Stream m b) -> Stream m a -> Stream m b
455concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)
457-- | Given a stream value in the underlying monad, lift and join the underlying
458-- monad with the stream monad.
460-- >>> concatM = Stream.concat . Stream.fromEffect
461-- >>> concatM = Stream.concat . lift -- requires (MonadTrans t)
462-- >>> concatM = join . lift -- requires (MonadTrans t, Monad (Stream m))
464-- See also: 'concat', 'sequence'
468{-# INLINE concatM #-}
469concatM :: Monad m => m (Stream m a) -> Stream m a
470concatM generator = concatMapM (\() -> generator) (fromPure ())
472-- | Like 'foldMany' but appends empty fold output if the fold and stream
473-- termination aligns:
475-- >>> f = Fold.take 2 Fold.sum
476-- >>> Stream.fold Fold.toList $ Stream.foldManyPost f $ Stream.fromList []
478-- >>> Stream.fold Fold.toList $ Stream.foldManyPost f $ Stream.fromList [1..9]
480-- >>> Stream.fold Fold.toList $ Stream.foldManyPost f $ Stream.fromList [1..10]
485{-# INLINE foldManyPost #-}
491foldManyPost f m = fromStreamD $ D.foldManyPost f (toStreamD m)
493------------------------------------------------------------------------------
495------------------------------------------------------------------------------
497-- | Like 'zipWith' but using a monadic zipping function.
499{-# INLINE zipWithM #-}
500zipWithM :: Monad m =>
501 (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c
502zipWithM f m1 m2 = fromStreamD $ D.zipWithM f (toStreamD m1) (toStreamD m2)
504-- | Stream @a@ is evaluated first, followed by stream @b@, the resulting
505-- elements @a@ and @b@ are then zipped using the supplied zip function and the
506-- result @c@ is yielded to the consumer.
508-- If stream @a@ or stream @b@ ends, the zipped stream ends. If stream @b@ ends
509-- first, the element @a@ from previous evaluation of stream @a@ is discarded.
511-- >>> s1 = Stream.fromList [1,2,3]
512-- >>> s2 = Stream.fromList [4,5,6]
513-- >>> Stream.fold Fold.toList $ Stream.zipWith (+) s1 s2
516{-# INLINE zipWith #-}
517zipWith :: Monad m => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
518zipWith 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