1{-# OPTIONS_GHC -Wno-orphans #-}
4-- Module : Streamly.Internal.Data.Stream.Generate
5-- Copyright : (c) 2017 Composewell Technologies
6-- License : BSD-3-Clause
7-- Maintainer : streamly@composewell.com
8-- Stability : experimental
11module Streamly.Internal.Data.Stream.Generate
71import Control.Monad.IO.Class (MonadIO)
72import Data.Word (Word8)
73import Foreign.Storable (Storable)
74import GHC.Exts (Addr#, Ptr (Ptr))
75import Streamly.Internal.Data.Stream.Bottom
76 (absTimesWith, relTimesWith, timesWith)
77import Streamly.Internal.Data.Stream.Enumerate
78 (Enumerable(..), enumerate, enumerateTo)
79import Streamly.Internal.Data.Stream.Type
80 (Stream, fromStreamD, fromStreamK, toStreamK)
81import Streamly.Internal.Data.Time.Units (AbsTime, RelTime64, addToAbsTime64)
82import Streamly.Internal.Data.Unboxed (Unboxed)
83import Streamly.Internal.Data.Unfold.Type (Unfold)
85import qualified Streamly.Internal.Data.IORef.Unboxed as Unboxed
86import qualified Streamly.Internal.Data.Stream.Bottom as Bottom
87import qualified Streamly.Internal.Data.Stream.StreamD as D
88import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
89import qualified Streamly.Internal.Data.Stream.Type as Stream
90import qualified Streamly.Internal.Data.Stream.Transform as Stream (sequence)
92import Prelude hiding (iterate, replicate, repeat, take)
96-- >>> import Control.Concurrent (threadDelay)
97-- >>> import Data.Function (fix, (&))
98-- >>> import Data.Semigroup (cycle1)
99-- >>> import qualified Streamly.Data.Fold as Fold
100-- >>> import qualified Streamly.Data.Unfold as Unfold
101-- >>> import qualified Streamly.Internal.Data.Stream as Stream
102-- >>> import GHC.Exts (Ptr (Ptr))
104------------------------------------------------------------------------------
106------------------------------------------------------------------------------
108-- | Convert an 'Unfold' into a stream by supplying it an input seed.
110-- >>> s = Stream.unfold Unfold.replicateM (3, putStrLn "hello")
111-- >>> Stream.fold Fold.drain s
118unfold :: Monad m => Unfold m a b -> a -> Stream m b
119unfold unf = Stream.fromStreamD . D.unfold unf
121------------------------------------------------------------------------------
122-- Generation by Unfolding
123------------------------------------------------------------------------------
129-- Nothing -> Stream.nil
130-- Just (a, b) -> a `Stream.cons` unfoldr step b
133-- Build a stream by unfolding a /pure/ step function @step@ starting from a
134-- seed @s@. The step function returns the next element in the stream and the
135-- next seed value. When it is done it returns 'Nothing' and the stream ends.
142-- else Just (b, b + 1)
143-- in Stream.fold Fold.toList $ Stream.unfoldr f 0
148{-# INLINE_EARLY unfoldr #-}
149unfoldr :: Monad m => (b -> Maybe (a, b)) -> b -> Stream m a
150unfoldr step seed = fromStreamD (D.unfoldr step seed)
151{-# RULES "unfoldr fallback to StreamK" [1]
152 forall a b. D.toStreamK (D.unfoldr a b) = K.unfoldr a b #-}
154-- | Build a stream by unfolding a /monadic/ step function starting from a
155-- seed. The step function returns the next element in the stream and the next
156-- seed value. When it is done it returns 'Nothing' and the stream ends. For
162-- then return Nothing
163-- else return (Just (b, b + 1))
164-- in Stream.fold Fold.toList $ Stream.unfoldrM f 0
169{-# INLINE unfoldrM #-}
170unfoldrM :: Monad m => (b -> m (Maybe (a, b))) -> b -> Stream m a
171unfoldrM step = fromStreamD . D.unfoldrM step
173------------------------------------------------------------------------------
175------------------------------------------------------------------------------
178-- Generate an infinite stream by repeating a pure value.
181{-# INLINE_NORMAL repeat #-}
182repeat :: Monad m => a -> Stream m a
183repeat = fromStreamD . D.repeat
186-- >>> repeatM = Stream.sequence . Stream.repeat
187-- >>> repeatM = fix . Stream.consM
188-- >>> repeatM = cycle1 . Stream.fromEffect
190-- Generate a stream by repeatedly executing a monadic action forever.
194-- Stream.repeatM (threadDelay 1000000 >> print 1)
196-- & Stream.fold Fold.drain
201{-# INLINE_NORMAL repeatM #-}
202repeatM :: Monad m => m a -> Stream m a
203repeatM = Stream.sequence . repeat
206-- >>> replicate n = Stream.take n . Stream.repeat
208-- Generate a stream of length @n@ by repeating a value @n@ times.
211{-# INLINE_NORMAL replicate #-}
212replicate :: Monad m => Int -> a -> Stream m a
213replicate n = fromStreamD . D.replicate n
216-- >>> replicateM n = Stream.sequence . Stream.replicate n
218-- Generate a stream by performing a monadic action @n@ times.
220{-# INLINE_NORMAL replicateM #-}
221replicateM :: Monad m => Int -> m a -> Stream m a
222replicateM n = Stream.sequence . replicate n
224------------------------------------------------------------------------------
226------------------------------------------------------------------------------
228-- | @times@ returns a stream of time value tuples with clock of 10 ms
229-- granularity. The first component of the tuple is an absolute time reference
230-- (epoch) denoting the start of the stream and the second component is a time
231-- relative to the reference.
233-- >>> f = Fold.drainBy (\x -> print x >> threadDelay 1000000)
234-- >>> Stream.fold f $ Stream.take 3 $ Stream.times
235-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
236-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
237-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
239-- Note: This API is not safe on 32-bit machines.
244times :: MonadIO m => Stream m (AbsTime, RelTime64)
245times = timesWith 0.01
247-- | @absTimes@ returns a stream of absolute timestamps using a clock of 10 ms
250-- >>> f = Fold.drainBy print
251-- >>> Stream.fold f $ Stream.delayPre 1 $ Stream.take 3 $ Stream.absTimes
252-- AbsTime (TimeSpec {sec = ..., nsec = ...})
253-- AbsTime (TimeSpec {sec = ..., nsec = ...})
254-- AbsTime (TimeSpec {sec = ..., nsec = ...})
256-- Note: This API is not safe on 32-bit machines.
260{-# INLINE absTimes #-}
261absTimes :: MonadIO m => Stream m AbsTime
262absTimes = fmap (uncurry addToAbsTime64) times
264-- | @relTimes@ returns a stream of relative time values starting from 0,
265-- using a clock of granularity 10 ms.
267-- >>> f = Fold.drainBy print
268-- >>> Stream.fold f $ Stream.delayPre 1 $ Stream.take 3 $ Stream.relTimes
269-- RelTime64 (NanoSecond64 ...)
270-- RelTime64 (NanoSecond64 ...)
271-- RelTime64 (NanoSecond64 ...)
273-- Note: This API is not safe on 32-bit machines.
277{-# INLINE relTimes #-}
278relTimes :: MonadIO m => Stream m RelTime64
279relTimes = fmap snd times
281-- | @durations g@ returns a stream of relative time values measuring the time
282-- elapsed since the immediate predecessor element of the stream was generated.
283-- The first element of the stream is always 0. @durations@ uses a clock of
284-- granularity @g@ specified in seconds. A low granularity clock is more
285-- expensive in terms of CPU usage. The minimum granularity is 1 millisecond.
286-- Durations lower than 1 ms will be 0.
288-- Note: This API is not safe on 32-bit machines.
292{-# INLINE durations #-}
293durations :: -- Monad m =>
294 Double -> t m RelTime64
297-- | Generate a singleton event at or after the specified absolute time. Note
298-- that this is different from a threadDelay, a threadDelay starts from the
299-- time when the action is evaluated, whereas if we use AbsTime based timeout
300-- it will immediately expire if the action is evaluated too late.
304{-# INLINE timeout #-}
305timeout :: -- Monad m =>
309------------------------------------------------------------------------------
310-- Iterating functions
311------------------------------------------------------------------------------
314-- >>> iterate f x = x `Stream.cons` iterate f x
316-- Generate an infinite stream with @x@ as the first element and each
317-- successive element derived by applying the function @f@ on the previous
320-- >>> Stream.fold Fold.toList $ Stream.take 5 $ Stream.iterate (+1) 1
324{-# INLINE_NORMAL iterate #-}
325iterate :: Monad m => (a -> a) -> a -> Stream m a
326iterate step = fromStreamD . D.iterate step
329-- >>> iterateM f m = m >>= \a -> return a `Stream.consM` iterateM f (f a)
331-- Generate an infinite stream with the first element generated by the action
332-- @m@ and each successive element derived by applying the monadic function
333-- @f@ on the previous element.
336-- Stream.iterateM (\x -> print x >> return (x + 1)) (return 0)
338-- & Stream.fold Fold.toList
345{-# INLINE iterateM #-}
346iterateM :: Monad m => (a -> m a) -> m a -> Stream m a
347iterateM step = fromStreamD . D.iterateM step
349-- | We can define cyclic structures using @let@:
351-- >>> let (a, b) = ([1, b], head a) in (a, b)
354-- The function @fix@ defined as:
356-- >>> fix f = let x = f x in x
358-- ensures that the argument of a function and its output refer to the same
359-- lazy value @x@ i.e. the same location in memory. Thus @x@ can be defined
360-- in terms of itself, creating structures with cyclic references.
362-- >>> f ~(a, b) = ([1, b], head a)
366-- 'Control.Monad.mfix' is essentially the same as @fix@ but for monadic
369-- Using 'mfix' for streams we can construct a stream in which each element of
370-- the stream is defined in a cyclic fashion. The argument of the function
371-- being fixed represents the current element of the stream which is being
372-- returned by the stream monad. Thus, we can use the argument to construct
375-- In the following example, the argument @action@ of the function @f@
376-- represents the tuple @(x,y)@ returned by it in a given iteration. We define
377-- the first element of the tuple in terms of the second.
379-- >>> import Streamly.Internal.Data.Stream as Stream
380-- >>> import System.IO.Unsafe (unsafeInterleaveIO)
383-- main = Stream.fold (Fold.drainBy print) $ Stream.mfix f
386-- let incr n act = fmap ((+n) . snd) $ unsafeInterleaveIO act
387-- x <- Stream.sequence $ Stream.fromList [incr 1 action, incr 2 action]
388-- y <- Stream.fromList [4,5]
392-- Note: you cannot achieve this by just changing the order of the monad
393-- statements because that would change the order in which the stream elements
396-- Note that the function @f@ must be lazy in its argument, that's why we use
397-- 'unsafeInterleaveIO' on @action@ because IO monad is strict.
403mfix :: Monad m => (m a -> Stream m a) -> Stream m a
404mfix f = fromStreamK $ K.mfix (toStreamK . f)
406------------------------------------------------------------------------------
408------------------------------------------------------------------------------
411-- >>> fromFoldable = Prelude.foldr Stream.cons Stream.nil
413-- Construct a stream from a 'Foldable' containing pure values:
418{-# INLINE fromFoldable #-}
419fromFoldable :: Foldable f => f a -> Stream m a
420fromFoldable = fromStreamK . K.fromFoldable
422------------------------------------------------------------------------------
424------------------------------------------------------------------------------
426-- | Keep reading 'Storable' elements from 'Ptr' onwards.
428-- /Unsafe:/ The caller is responsible for safe addressing.
431{-# INLINE fromPtr #-}
432fromPtr :: (MonadIO m, Storable a) => Ptr a -> Stream m a
433fromPtr = Stream.fromStreamD . D.fromPtr
435-- | Take @n@ 'Storable' elements starting from 'Ptr' onwards.
437-- >>> fromPtrN n = Stream.take n . Stream.fromPtr
439-- /Unsafe:/ The caller is responsible for safe addressing.
442{-# INLINE fromPtrN #-}
443fromPtrN :: (MonadIO m, Storable a) => Int -> Ptr a -> Stream m a
444fromPtrN n = Stream.fromStreamD . D.take n . D.fromPtr
446-- | Read bytes from an 'Addr#' until a 0 byte is encountered, the 0 byte is
447-- not included in the stream.
449-- >>> fromByteStr# addr = Stream.takeWhile (/= 0) $ Stream.fromPtr $ Ptr addr
451-- /Unsafe:/ The caller is responsible for safe addressing.
453-- Note that this is completely safe when reading from Haskell string
454-- literals because they are guaranteed to be NULL terminated:
456-- >>> Stream.fold Fold.toList $ Stream.fromByteStr# "\1\2\3\0"#
459{-# INLINE fromByteStr# #-}
460fromByteStr# :: MonadIO m => Addr# -> Stream m Word8
462 Stream.fromStreamD $ D.takeWhile (/= 0) $ D.fromPtr $ Ptr addr
464-- | Construct a stream by reading an 'Unboxed' 'IORef' repeatedly.
468{-# INLINE fromUnboxedIORef #-}
469fromUnboxedIORef :: (MonadIO m, Unboxed a) => Unboxed.IORef a -> Stream m a
470fromUnboxedIORef = fromStreamD . Unboxed.toStreamD
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