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
60import Control.Monad.IO.Class (MonadIO(..))
61import Streamly.Internal.Data.Fold.Type (Fold (..))
62import Streamly.Internal.Data.Time.Units (AbsTime, RelTime64, addToAbsTime64)
63import Streamly.Internal.Data.Unboxed (Unboxed)
64import Streamly.Internal.System.IO (defaultChunkSize)
66import qualified Streamly.Internal.Data.Array.Unboxed.Type as A
67import qualified Streamly.Internal.Data.Fold as Fold
68import qualified Streamly.Internal.Data.Stream.Common as Common
69import qualified Streamly.Internal.Data.Stream.StreamK as K
70import qualified Streamly.Internal.Data.Stream.StreamD as D
72import Prelude hiding (take, takeWhile, drop, reverse, concatMap, map, zipWith)
74import Streamly.Internal.Data.Stream.Type
79-- >>> import Control.Monad (join)
80-- >>> import Control.Monad.Trans.Class (lift)
81-- >>> import Data.Function (fix, (&))
82-- >>> import Data.Semigroup (cycle1)
83-- >>> import Prelude hiding (take, takeWhile, drop, reverse)
84-- >>> import System.IO.Unsafe (unsafePerformIO)
85-- >>> import Streamly.Internal.Data.Stream as Stream
86-- >>> import qualified Streamly.Data.Array.Unboxed as Array
87-- >>> import qualified Streamly.Internal.Data.Fold as Fold
88-- >>> import qualified Streamly.Internal.Data.Parser as Parser
89-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
91------------------------------------------------------------------------------
93------------------------------------------------------------------------------
96-- >>> fromList = Prelude.foldr Stream.cons Stream.nil
98-- Construct a stream from a list of pure values. This is more efficient than
101{-# INLINE fromList #-}
102fromList :: Monad m => [a] -> Stream m a
103fromList = fromStreamK . Common.fromList
105------------------------------------------------------------------------------
106-- Generation - Time related
107------------------------------------------------------------------------------
109-- | @timesWith g@ returns a stream of time value tuples. The first component
110-- of the tuple is an absolute time reference (epoch) denoting the start of the
111-- stream and the second component is a time relative to the reference.
113-- The argument @g@ specifies the granularity of the relative time in seconds.
114-- A lower granularity clock gives higher precision but is more expensive in
115-- terms of CPU usage. Any granularity lower than 1 ms is treated as 1 ms.
117-- >>> import Control.Concurrent (threadDelay)
118-- >>> f = Fold.drainBy (\x -> print x >> threadDelay 1000000)
119-- >>> Stream.fold f $ Stream.take 3 $ Stream.timesWith 0.01
120-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
121-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
122-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
124-- Note: This API is not safe on 32-bit machines.
128{-# INLINE timesWith #-}
129timesWith :: MonadIO m => Double -> Stream m (AbsTime, RelTime64)
130timesWith g = fromStreamD $ D.times g
132-- | @absTimesWith g@ returns a stream of absolute timestamps using a clock of
133-- granularity @g@ specified in seconds. A low granularity clock is more
134-- expensive in terms of CPU usage. Any granularity lower than 1 ms is treated
137-- >>> f = Fold.drainBy print
138-- >>> Stream.fold f $ Stream.delayPre 1 $ Stream.take 3 $ Stream.absTimesWith 0.01
139-- AbsTime (TimeSpec {sec = ..., nsec = ...})
140-- AbsTime (TimeSpec {sec = ..., nsec = ...})
141-- AbsTime (TimeSpec {sec = ..., nsec = ...})
143-- Note: This API is not safe on 32-bit machines.
147{-# INLINE absTimesWith #-}
148absTimesWith :: MonadIO m => Double -> Stream m AbsTime
149absTimesWith = fmap (uncurry addToAbsTime64) . timesWith
151-- | @relTimesWith g@ returns a stream of relative time values starting from 0,
152-- using a clock of granularity @g@ specified in seconds. A low granularity
153-- clock is more expensive in terms of CPU usage. Any granularity lower than 1
154-- ms is treated as 1 ms.
156-- >>> f = Fold.drainBy print
157-- >>> Stream.fold f $ Stream.delayPre 1 $ Stream.take 3 $ Stream.relTimesWith 0.01
158-- RelTime64 (NanoSecond64 ...)
159-- RelTime64 (NanoSecond64 ...)
160-- RelTime64 (NanoSecond64 ...)
162-- Note: This API is not safe on 32-bit machines.
166{-# INLINE relTimesWith #-}
167relTimesWith :: MonadIO m => Double -> Stream m RelTime64
168relTimesWith = fmap snd . timesWith
170------------------------------------------------------------------------------
171-- Elimination - Running a Fold
172------------------------------------------------------------------------------
174-- | We can create higher order folds using 'foldContinue'. We can fold a
175-- number of streams to a given fold efficiently with full stream fusion. For
176-- example, to fold a list of streams on the same sum fold:
178-- >>> concatFold = Prelude.foldl Stream.foldContinue Fold.sum
180-- >>> fold f = Fold.finish . Stream.foldContinue f
183{-# INLINE foldContinue #-}
184foldContinue :: Monad m => Fold m a b -> Stream m a -> Fold m a b
185foldContinue f s = D.foldContinue f $ toStreamD s
187-- | Fold a stream using the supplied left 'Fold' and reducing the resulting
188-- expression strictly at each step. The behavior is similar to 'foldl''. A
189-- 'Fold' can terminate early without consuming the full stream. See the
190-- documentation of individual 'Fold's for termination behavior.
192-- >>> Stream.fold Fold.sum (Stream.enumerateFromTo 1 100)
195-- Folds never fail, therefore, they produce a default value even when no input
196-- is provided. It means we can always fold an empty stream and get a valid
197-- result. For example:
199-- >>> Stream.fold Fold.sum Stream.nil
202-- >>> fold f = Stream.parse (Parser.fromFold f)
206fold :: Monad m => Fold m a b -> Stream m a -> m b
207fold fl strm = D.fold fl $ D.fromStreamK $ toStreamK strm
209-- | Like 'fold' but also returns the remaining stream.
213{-# INLINE foldBreak #-}
214foldBreak :: Monad m => Fold m a b -> Stream m a -> m (b, Stream m a)
216-- XXX This shows quadratic performance when used recursively perhaps because
217-- of StreamK to StreamD conversions not getting eliminated sue to recursion.
218foldBreak fl (Stream strm) = fmap f $ D.foldBreak fl $ D.fromStreamK strm
222 f (b, str) = (b, Stream (D.toStreamK str))
224foldBreak fl strm = fmap f $ K.foldBreak fl (toStreamK strm)
228 f (b, str) = (b, fromStreamK str)
230------------------------------------------------------------------------------
232------------------------------------------------------------------------------
240-- >>> Stream.fold Fold.toList $ fmap (+1) $ Stream.fromList [1,2,3]
244map :: Monad m => (a -> b) -> Stream m a -> Stream m b
245map f = fromStreamD . D.map f . toStreamD
248-- | Postscan a stream using the given monadic fold.
250-- The following example extracts the input stream up to a point where the
251-- running average of elements is no more than 10:
253-- >>> import Data.Maybe (fromJust)
254-- >>> let avg = Fold.teeWith (/) Fold.sum (fmap fromIntegral Fold.length)
255-- >>> s = Stream.enumerateFromTo 1.0 100.0
257-- Stream.fold Fold.toList
258-- $ fmap (fromJust . fst)
259-- $ Stream.takeWhile (\(_,x) -> x <= 10)
260-- $ Stream.postscan (Fold.tee Fold.last avg) s
262-- [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]
264{-# INLINE postscan #-}
265postscan :: Monad m => Fold m a b -> Stream m a -> Stream m b
266postscan fld = fromStreamD . D.postscanOnce fld . toStreamD
270-- The stateful step function can be simplified to @(s -> a -> m b)@ to provide
271-- a read-only environment. However, that would just be 'mapM'.
273-- The initial action could be @m (s, Maybe b)@, and we can also add a final
274-- action @s -> m (Maybe b)@. This can be used to get pre/post scan like
275-- functionality and also to flush the state in the end like scanlMAfter'.
276-- We can also use it along with a fusible version of bracket to get
277-- scanlMAfter' like functionality. See issue #677.
279-- This can be further generalized to a type similar to Fold/Parser, giving it
280-- filtering and parsing capability as well (this is in fact equivalent to
283-- smapM :: (s -> a -> m (Step s b)) -> m s -> Stream m a -> Stream m b
286-- | A stateful 'mapM', equivalent to a left scan, more like mapAccumL.
287-- Hopefully, this is a better alternative to @scan@. Separation of state from
288-- the output makes it easier to think in terms of a shared state, and also
289-- makes it easier to keep the state fully strict and the output lazy.
291-- See also: 'postscan'
301smapM step initial stream =
302 -- XXX implement this directly instead of using postscan
304 (\(s, _) a -> step s a)
305 (fmap (,undefined) initial)
306 in fmap snd $ postscan f stream
308------------------------------------------------------------------------------
309-- Transformation - Trimming
310------------------------------------------------------------------------------
312-- | Take first 'n' elements from the stream and discard the rest.
315take :: Monad m => Int -> Stream m a -> Stream m a
316take n m = fromStreamD $ D.take n $ toStreamD m
318-- | End the stream as soon as the predicate fails on an element.
320{-# INLINE takeWhile #-}
321takeWhile :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
322takeWhile p m = fromStreamD $ D.takeWhile p $ toStreamD m
324{-# INLINE takeEndBy #-}
325takeEndBy :: Monad m => (a -> Bool) -> Stream m a -> Stream m a
326takeEndBy p m = fromStreamD $ D.takeEndBy p $ toStreamD m
328-- | Discard first 'n' elements from the stream and take the rest.
331drop :: Monad m => Int -> Stream m a -> Stream m a
332drop n m = fromStreamD $ D.drop n $ toStreamD m
334------------------------------------------------------------------------------
336------------------------------------------------------------------------------
338-- | Find all the indices where the element in the stream satisfies the given
341-- >>> findIndices p = Stream.fold (Fold.findIndices p)
343{-# INLINE findIndices #-}
344findIndices :: Monad m => (a -> Bool) -> Stream m a -> Stream m Int
345findIndices p m = fromStreamD $ D.findIndices p (toStreamD m)
347------------------------------------------------------------------------------
348-- Transformation by Inserting
349------------------------------------------------------------------------------
351-- intersperseM = intersperseBySpan 1
353-- | Insert an effect and its output before consuming an element of a stream
354-- except the first one.
356-- >>> input = Stream.fromList "hello"
357-- >>> Stream.fold Fold.toList $ Stream.trace putChar $ Stream.intersperseM (putChar '.' >> return ',') input
358-- h.,e.,l.,l.,o"h,e,l,l,o"
360-- Be careful about the order of effects. In the above example we used trace
361-- after the intersperse, if we use it before the intersperse the output would
362-- be he.l.l.o."h,e,l,l,o".
364-- >>> Stream.fold Fold.toList $ Stream.intersperseM (putChar '.' >> return ',') $ Stream.trace putChar input
365-- he.l.l.o."h,e,l,l,o"
367{-# INLINE intersperseM #-}
368intersperseM :: Monad m => m a -> Stream m a -> Stream m a
369intersperseM m = fromStreamD . D.intersperseM m . toStreamD
371------------------------------------------------------------------------------
372-- Transformation by Reordering
373------------------------------------------------------------------------------
375-- XXX Use a compact region list to temporarily store the list, in both reverse
376-- as well as in reverse'.
378-- /Note:/ 'reverse'' is much faster than this, use that when performance
381-- | Returns the elements of the stream in reverse order. The stream must be
382-- finite. Note that this necessarily buffers the entire stream in memory.
384-- >>> reverse = Stream.foldlT (flip Stream.cons) Stream.nil
386{-# INLINE reverse #-}
387reverse :: Monad m => Stream m a -> Stream m a
388reverse s = fromStreamD $ D.reverse $ toStreamD s
390-- | Like 'reverse' but several times faster, requires a 'Storable' instance.
393{-# INLINE reverse' #-}
394reverse' :: (MonadIO m, Unboxed a) => Stream m a -> Stream m a
395-- reverse' s = fromStreamD $ D.reverse' $ toStreamD s
398 . A.flattenArraysRev -- unfoldMany A.readRev
402 . A.arraysOf defaultChunkSize
405------------------------------------------------------------------------------
406-- Combine streams and flatten
407------------------------------------------------------------------------------
409-- | Map a stream producing monadic function on each element of the stream
410-- and then flatten the results into a single stream. Since the stream
411-- generation function is monadic, unlike 'concatMap', it can produce an
412-- effect at the beginning of each iteration of the inner loop.
414{-# INLINE concatMapM #-}
415concatMapM :: Monad m => (a -> m (Stream m b)) -> Stream m a -> Stream m b
416concatMapM f m = fromStreamD $ D.concatMapM (fmap toStreamD . f) (toStreamD m)
418-- | Map a stream producing function on each element of the stream and then
419-- flatten the results into a single stream.
421-- >>> concatMap f = Stream.concatMapM (return . f)
422-- >>> concatMap f = Stream.concatMapWith Stream.append f
423-- >>> concatMap f = Stream.concat . fmap f
424-- >>> concatMap f = Stream.unfoldMany (Unfold.lmap f Unfold.fromStream)
426{-# INLINE concatMap #-}
427concatMap ::Monad m => (a -> Stream m b) -> Stream m a -> Stream m b
428concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)
430-- | Given a stream value in the underlying monad, lift and join the underlying
431-- monad with the stream monad.
433-- >>> concatM = Stream.concat . Stream.fromEffect
434-- >>> concatM = Stream.concat . lift -- requires (MonadTrans t)
435-- >>> concatM = join . lift -- requires (MonadTrans t, Monad (Stream m))
437-- See also: 'concat', 'sequence'
441{-# INLINE concatM #-}
442concatM :: Monad m => m (Stream m a) -> Stream m a
443concatM generator = concatMapM (\() -> generator) (fromPure ())
445-- | Like 'foldMany' but appends empty fold output if the fold and stream
446-- termination aligns:
448-- >>> f = Fold.take 2 Fold.sum
449-- >>> Stream.fold Fold.toList $ Stream.foldManyPost f $ Stream.fromList []
451-- >>> Stream.fold Fold.toList $ Stream.foldManyPost f $ Stream.fromList [1..9]
453-- >>> Stream.fold Fold.toList $ Stream.foldManyPost f $ Stream.fromList [1..10]
458{-# INLINE foldManyPost #-}
464foldManyPost f m = fromStreamD $ D.foldManyPost f (toStreamD m)
466------------------------------------------------------------------------------
468------------------------------------------------------------------------------
470-- | Like 'zipWith' but using a monadic zipping function.
472{-# INLINE zipWithM #-}
473zipWithM :: Monad m =>
474 (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c
475zipWithM f m1 m2 = fromStreamD $ D.zipWithM f (toStreamD m1) (toStreamD m2)
477-- | Stream @a@ is evaluated first, followed by stream @b@, the resulting
478-- elements @a@ and @b@ are then zipped using the supplied zip function and the
479-- result @c@ is yielded to the consumer.
481-- If stream @a@ or stream @b@ ends, the zipped stream ends. If stream @b@ ends
482-- first, the element @a@ from previous evaluation of stream @a@ is discarded.
484-- >>> s1 = Stream.fromList [1,2,3]
485-- >>> s2 = Stream.fromList [4,5,6]
486-- >>> Stream.fold Fold.toList $ Stream.zipWith (+) s1 s2
489{-# INLINE zipWith #-}
490zipWith :: Monad m => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
491zipWith 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