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 (Unbox)
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
117unfold :: Monad m => Unfold m a b -> a -> Stream m b
118unfold unf = Stream.fromStreamD . D.unfold unf
120------------------------------------------------------------------------------
121-- Generation by Unfolding
122------------------------------------------------------------------------------
128-- Nothing -> Stream.nil
129-- Just (a, b) -> a `Stream.cons` unfoldr step b
132-- Build a stream by unfolding a /pure/ step function @step@ starting from a
133-- seed @s@. The step function returns the next element in the stream and the
134-- next seed value. When it is done it returns 'Nothing' and the stream ends.
141-- else Just (b, b + 1)
142-- in Stream.fold Fold.toList $ Stream.unfoldr f 0
146{-# INLINE_EARLY unfoldr #-}
147unfoldr :: Monad m => (b -> Maybe (a, b)) -> b -> Stream m a
148unfoldr step seed = fromStreamD (D.unfoldr step seed)
149{-# RULES "unfoldr fallback to StreamK" [1]
150 forall a b. D.toStreamK (D.unfoldr a b) = K.unfoldr a b #-}
152-- | Build a stream by unfolding a /monadic/ step function starting from a
153-- seed. The step function returns the next element in the stream and the next
154-- seed value. When it is done it returns 'Nothing' and the stream ends. For
160-- then return Nothing
161-- else return (Just (b, b + 1))
162-- in Stream.fold Fold.toList $ Stream.unfoldrM f 0
166{-# INLINE unfoldrM #-}
167unfoldrM :: Monad m => (b -> m (Maybe (a, b))) -> b -> Stream m a
168unfoldrM step = fromStreamD . D.unfoldrM step
170------------------------------------------------------------------------------
172------------------------------------------------------------------------------
175-- Generate an infinite stream by repeating a pure value.
177{-# INLINE_NORMAL repeat #-}
178repeat :: Monad m => a -> Stream m a
179repeat = fromStreamD . D.repeat
182-- >>> repeatM = Stream.sequence . Stream.repeat
183-- >>> repeatM = fix . Stream.consM
184-- >>> repeatM = cycle1 . Stream.fromEffect
186-- Generate a stream by repeatedly executing a monadic action forever.
190-- Stream.repeatM (threadDelay 1000000 >> print 1)
192-- & Stream.fold Fold.drain
197{-# INLINE_NORMAL repeatM #-}
198repeatM :: Monad m => m a -> Stream m a
199repeatM = Stream.sequence . repeat
202-- >>> replicate n = Stream.take n . Stream.repeat
204-- Generate a stream of length @n@ by repeating a value @n@ times.
206{-# INLINE_NORMAL replicate #-}
207replicate :: Monad m => Int -> a -> Stream m a
208replicate n = fromStreamD . D.replicate n
211-- >>> replicateM n = Stream.sequence . Stream.replicate n
213-- Generate a stream by performing a monadic action @n@ times.
215{-# INLINE_NORMAL replicateM #-}
216replicateM :: Monad m => Int -> m a -> Stream m a
217replicateM n = Stream.sequence . replicate n
219------------------------------------------------------------------------------
221------------------------------------------------------------------------------
223-- | @times@ returns a stream of time value tuples with clock of 10 ms
224-- granularity. The first component of the tuple is an absolute time reference
225-- (epoch) denoting the start of the stream and the second component is a time
226-- relative to the reference.
228-- >>> f = Fold.drainBy (\x -> print x >> threadDelay 1000000)
229-- >>> Stream.fold f $ Stream.take 3 $ Stream.times
230-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
231-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
232-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
234-- Note: This API is not safe on 32-bit machines.
239times :: MonadIO m => Stream m (AbsTime, RelTime64)
240times = timesWith 0.01
242-- | @absTimes@ returns a stream of absolute timestamps using a clock of 10 ms
245-- >>> f = Fold.drainBy print
246-- >>> Stream.fold f $ Stream.delayPre 1 $ Stream.take 3 $ Stream.absTimes
247-- AbsTime (TimeSpec {sec = ..., nsec = ...})
248-- AbsTime (TimeSpec {sec = ..., nsec = ...})
249-- AbsTime (TimeSpec {sec = ..., nsec = ...})
251-- Note: This API is not safe on 32-bit machines.
255{-# INLINE absTimes #-}
256absTimes :: MonadIO m => Stream m AbsTime
257absTimes = fmap (uncurry addToAbsTime64) times
259-- | @relTimes@ returns a stream of relative time values starting from 0,
260-- using a clock of granularity 10 ms.
262-- >>> f = Fold.drainBy print
263-- >>> Stream.fold f $ Stream.delayPre 1 $ Stream.take 3 $ Stream.relTimes
264-- RelTime64 (NanoSecond64 ...)
265-- RelTime64 (NanoSecond64 ...)
266-- RelTime64 (NanoSecond64 ...)
268-- Note: This API is not safe on 32-bit machines.
272{-# INLINE relTimes #-}
273relTimes :: MonadIO m => Stream m RelTime64
274relTimes = fmap snd times
276-- | @durations g@ returns a stream of relative time values measuring the time
277-- elapsed since the immediate predecessor element of the stream was generated.
278-- The first element of the stream is always 0. @durations@ uses a clock of
279-- granularity @g@ specified in seconds. A low granularity clock is more
280-- expensive in terms of CPU usage. The minimum granularity is 1 millisecond.
281-- Durations lower than 1 ms will be 0.
283-- Note: This API is not safe on 32-bit machines.
287{-# INLINE durations #-}
288durations :: -- Monad m =>
289 Double -> t m RelTime64
292-- | Generate a singleton event at or after the specified absolute time. Note
293-- that this is different from a threadDelay, a threadDelay starts from the
294-- time when the action is evaluated, whereas if we use AbsTime based timeout
295-- it will immediately expire if the action is evaluated too late.
299{-# INLINE timeout #-}
300timeout :: -- Monad m =>
304------------------------------------------------------------------------------
305-- Iterating functions
306------------------------------------------------------------------------------
309-- >>> iterate f x = x `Stream.cons` iterate f x
311-- Generate an infinite stream with @x@ as the first element and each
312-- successive element derived by applying the function @f@ on the previous
315-- >>> Stream.fold Fold.toList $ Stream.take 5 $ Stream.iterate (+1) 1
318{-# INLINE_NORMAL iterate #-}
319iterate :: Monad m => (a -> a) -> a -> Stream m a
320iterate step = fromStreamD . D.iterate step
323-- >>> iterateM f m = m >>= \a -> return a `Stream.consM` iterateM f (f a)
325-- Generate an infinite stream with the first element generated by the action
326-- @m@ and each successive element derived by applying the monadic function
327-- @f@ on the previous element.
330-- Stream.iterateM (\x -> print x >> return (x + 1)) (return 0)
332-- & Stream.fold Fold.toList
338{-# INLINE iterateM #-}
339iterateM :: Monad m => (a -> m a) -> m a -> Stream m a
340iterateM step = fromStreamD . D.iterateM step
342-- | We can define cyclic structures using @let@:
344-- >>> let (a, b) = ([1, b], head a) in (a, b)
347-- The function @fix@ defined as:
349-- >>> fix f = let x = f x in x
351-- ensures that the argument of a function and its output refer to the same
352-- lazy value @x@ i.e. the same location in memory. Thus @x@ can be defined
353-- in terms of itself, creating structures with cyclic references.
355-- >>> f ~(a, b) = ([1, b], head a)
359-- 'Control.Monad.mfix' is essentially the same as @fix@ but for monadic
362-- Using 'mfix' for streams we can construct a stream in which each element of
363-- the stream is defined in a cyclic fashion. The argument of the function
364-- being fixed represents the current element of the stream which is being
365-- returned by the stream monad. Thus, we can use the argument to construct
368-- In the following example, the argument @action@ of the function @f@
369-- represents the tuple @(x,y)@ returned by it in a given iteration. We define
370-- the first element of the tuple in terms of the second.
372-- >>> import Streamly.Internal.Data.Stream as Stream
373-- >>> import System.IO.Unsafe (unsafeInterleaveIO)
376-- main = Stream.fold (Fold.drainBy print) $ Stream.mfix f
379-- let incr n act = fmap ((+n) . snd) $ unsafeInterleaveIO act
380-- x <- Stream.sequence $ Stream.fromList [incr 1 action, incr 2 action]
381-- y <- Stream.fromList [4,5]
385-- Note: you cannot achieve this by just changing the order of the monad
386-- statements because that would change the order in which the stream elements
389-- Note that the function @f@ must be lazy in its argument, that's why we use
390-- 'unsafeInterleaveIO' on @action@ because IO monad is strict.
396mfix :: Monad m => (m a -> Stream m a) -> Stream m a
397mfix f = fromStreamK $ K.mfix (toStreamK . f)
399------------------------------------------------------------------------------
401------------------------------------------------------------------------------
404-- >>> fromFoldable = Prelude.foldr Stream.cons Stream.nil
406-- Construct a stream from a 'Foldable' containing pure values:
410{-# INLINE fromFoldable #-}
411fromFoldable :: Foldable f => f a -> Stream m a
412fromFoldable = fromStreamK . K.fromFoldable
414------------------------------------------------------------------------------
416------------------------------------------------------------------------------
418-- | Keep reading 'Storable' elements from 'Ptr' onwards.
420-- /Unsafe:/ The caller is responsible for safe addressing.
423{-# INLINE fromPtr #-}
424fromPtr :: (MonadIO m, Storable a) => Ptr a -> Stream m a
425fromPtr = Stream.fromStreamD . D.fromPtr
427-- | Take @n@ 'Storable' elements starting from 'Ptr' onwards.
429-- >>> fromPtrN n = Stream.take n . Stream.fromPtr
431-- /Unsafe:/ The caller is responsible for safe addressing.
434{-# INLINE fromPtrN #-}
435fromPtrN :: (MonadIO m, Storable a) => Int -> Ptr a -> Stream m a
436fromPtrN n = Stream.fromStreamD . D.take n . D.fromPtr
438-- | Read bytes from an 'Addr#' until a 0 byte is encountered, the 0 byte is
439-- not included in the stream.
441-- >>> fromByteStr# addr = Stream.takeWhile (/= 0) $ Stream.fromPtr $ Ptr addr
443-- /Unsafe:/ The caller is responsible for safe addressing.
445-- Note that this is completely safe when reading from Haskell string
446-- literals because they are guaranteed to be NULL terminated:
448-- >>> Stream.fold Fold.toList $ Stream.fromByteStr# "\1\2\3\0"#
451{-# INLINE fromByteStr# #-}
452fromByteStr# :: MonadIO m => Addr# -> Stream m Word8
454 Stream.fromStreamD $ D.takeWhile (/= 0) $ D.fromPtr $ Ptr addr
456-- | Construct a stream by reading an 'Unboxed' 'IORef' repeatedly.
460{-# INLINE fromUnboxedIORef #-}
461fromUnboxedIORef :: (MonadIO m, Unbox a) => Unboxed.IORef a -> Stream m a
462fromUnboxedIORef = 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