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.Unbox (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 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 Streamly.Internal.Data.Stream.Cross (CrossStream(..))
100-- >>> import qualified Streamly.Data.Fold as Fold
101-- >>> import qualified Streamly.Data.Unfold as Unfold
102-- >>> import qualified Streamly.Internal.Data.Stream as Stream
103-- >>> import GHC.Exts (Ptr (Ptr))
105------------------------------------------------------------------------------
107------------------------------------------------------------------------------
109-- | Convert an 'Unfold' into a stream by supplying it an input seed.
111-- >>> s = Stream.unfold Unfold.replicateM (3, putStrLn "hello")
112-- >>> 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
147{-# INLINE_EARLY unfoldr #-}
148unfoldr :: Monad m => (b -> Maybe (a, b)) -> b -> Stream m a
149unfoldr step seed = fromStreamD (D.unfoldr step seed)
150{-# RULES "unfoldr fallback to StreamK" [1]
151 forall a b. D.toStreamK (D.unfoldr a b) = K.unfoldr a b #-}
153-- | Build a stream by unfolding a /monadic/ step function starting from a
154-- seed. The step function returns the next element in the stream and the next
155-- seed value. When it is done it returns 'Nothing' and the stream ends. For
161-- then return Nothing
162-- else return (Just (b, b + 1))
163-- in Stream.fold Fold.toList $ Stream.unfoldrM f 0
167{-# INLINE unfoldrM #-}
168unfoldrM :: Monad m => (b -> m (Maybe (a, b))) -> b -> Stream m a
169unfoldrM step = fromStreamD . D.unfoldrM step
171------------------------------------------------------------------------------
173------------------------------------------------------------------------------
176-- Generate an infinite stream by repeating a pure value.
178{-# INLINE_NORMAL repeat #-}
179repeat :: Monad m => a -> Stream m a
180repeat = fromStreamD . D.repeat
183-- >>> repeatM = Stream.sequence . Stream.repeat
184-- >>> repeatM = fix . Stream.consM
185-- >>> repeatM = cycle1 . Stream.fromEffect
187-- Generate a stream by repeatedly executing a monadic action forever.
191-- Stream.repeatM (threadDelay 1000000 >> print 1)
193-- & Stream.fold Fold.drain
196{-# INLINE_NORMAL repeatM #-}
197repeatM :: Monad m => m a -> Stream m a
198repeatM = Stream.sequence . repeat
201-- >>> replicate n = Stream.take n . Stream.repeat
203-- Generate a stream of length @n@ by repeating a value @n@ times.
205{-# INLINE_NORMAL replicate #-}
206replicate :: Monad m => Int -> a -> Stream m a
207replicate n = fromStreamD . D.replicate n
210-- >>> replicateM n = Stream.sequence . Stream.replicate n
212-- Generate a stream by performing a monadic action @n@ times.
213{-# INLINE_NORMAL replicateM #-}
214replicateM :: Monad m => Int -> m a -> Stream m a
215replicateM n = Stream.sequence . replicate n
217------------------------------------------------------------------------------
219------------------------------------------------------------------------------
221-- | @times@ returns a stream of time value tuples with clock of 10 ms
222-- granularity. The first component of the tuple is an absolute time reference
223-- (epoch) denoting the start of the stream and the second component is a time
224-- relative to the reference.
226-- >>> f = Fold.drainMapM (\x -> print x >> threadDelay 1000000)
227-- >>> Stream.fold f $ Stream.take 3 $ Stream.times
228-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
229-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
230-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
232-- Note: This API is not safe on 32-bit machines.
237times :: MonadIO m => Stream m (AbsTime, RelTime64)
238times = timesWith 0.01
240-- | @absTimes@ returns a stream of absolute timestamps using a clock of 10 ms
243-- >>> f = Fold.drainMapM print
244-- >>> Stream.fold f $ Stream.delayPre 1 $ Stream.take 3 $ Stream.absTimes
245-- AbsTime (TimeSpec {sec = ..., nsec = ...})
246-- AbsTime (TimeSpec {sec = ..., nsec = ...})
247-- AbsTime (TimeSpec {sec = ..., nsec = ...})
249-- Note: This API is not safe on 32-bit machines.
253{-# INLINE absTimes #-}
254absTimes :: MonadIO m => Stream m AbsTime
255absTimes = fmap (uncurry addToAbsTime64) times
257-- | @relTimes@ returns a stream of relative time values starting from 0,
258-- using a clock of granularity 10 ms.
260-- >>> f = Fold.drainMapM print
261-- >>> Stream.fold f $ Stream.delayPre 1 $ Stream.take 3 $ Stream.relTimes
262-- RelTime64 (NanoSecond64 ...)
263-- RelTime64 (NanoSecond64 ...)
264-- RelTime64 (NanoSecond64 ...)
266-- Note: This API is not safe on 32-bit machines.
270{-# INLINE relTimes #-}
271relTimes :: MonadIO m => Stream m RelTime64
272relTimes = fmap snd times
274-- | @durations g@ returns a stream of relative time values measuring the time
275-- elapsed since the immediate predecessor element of the stream was generated.
276-- The first element of the stream is always 0. @durations@ uses a clock of
277-- granularity @g@ specified in seconds. A low granularity clock is more
278-- expensive in terms of CPU usage. The minimum granularity is 1 millisecond.
279-- Durations lower than 1 ms will be 0.
281-- Note: This API is not safe on 32-bit machines.
285{-# INLINE durations #-}
286durations :: -- Monad m =>
287 Double -> t m RelTime64
290-- | Generate a singleton event at or after the specified absolute time. Note
291-- that this is different from a threadDelay, a threadDelay starts from the
292-- time when the action is evaluated, whereas if we use AbsTime based timeout
293-- it will immediately expire if the action is evaluated too late.
297{-# INLINE timeout #-}
298timeout :: -- Monad m =>
302------------------------------------------------------------------------------
303-- Iterating functions
304------------------------------------------------------------------------------
307-- >>> iterate f x = x `Stream.cons` iterate f x
309-- Generate an infinite stream with @x@ as the first element and each
310-- successive element derived by applying the function @f@ on the previous
313-- >>> Stream.fold Fold.toList $ Stream.take 5 $ Stream.iterate (+1) 1
316{-# INLINE_NORMAL iterate #-}
317iterate :: Monad m => (a -> a) -> a -> Stream m a
318iterate step = fromStreamD . D.iterate step
321-- >>> iterateM f m = m >>= \a -> return a `Stream.consM` iterateM f (f a)
323-- Generate an infinite stream with the first element generated by the action
324-- @m@ and each successive element derived by applying the monadic function
325-- @f@ on the previous element.
328-- Stream.iterateM (\x -> print x >> return (x + 1)) (return 0)
330-- & Stream.fold Fold.toList
336{-# INLINE iterateM #-}
337iterateM :: Monad m => (a -> m a) -> m a -> Stream m a
338iterateM step = fromStreamD . D.iterateM step
340-- | We can define cyclic structures using @let@:
342-- >>> let (a, b) = ([1, b], head a) in (a, b)
345-- The function @fix@ defined as:
347-- >>> fix f = let x = f x in x
349-- ensures that the argument of a function and its output refer to the same
350-- lazy value @x@ i.e. the same location in memory. Thus @x@ can be defined
351-- in terms of itself, creating structures with cyclic references.
353-- >>> f ~(a, b) = ([1, b], head a)
357-- 'Control.Monad.mfix' is essentially the same as @fix@ but for monadic
360-- Using 'mfix' for streams we can construct a stream in which each element of
361-- the stream is defined in a cyclic fashion. The argument of the function
362-- being fixed represents the current element of the stream which is being
363-- returned by the stream monad. Thus, we can use the argument to construct
366-- In the following example, the argument @action@ of the function @f@
367-- represents the tuple @(x,y)@ returned by it in a given iteration. We define
368-- the first element of the tuple in terms of the second.
370-- >>> import Streamly.Internal.Data.Stream as Stream
371-- >>> import System.IO.Unsafe (unsafeInterleaveIO)
374-- main = Stream.fold (Fold.drainMapM print) $ Stream.mfix f
376-- f action = unCrossStream $ do
377-- let incr n act = fmap ((+n) . snd) $ unsafeInterleaveIO act
378-- x <- CrossStream (Stream.sequence $ Stream.fromList [incr 1 action, incr 2 action])
379-- y <- CrossStream (Stream.fromList [4,5])
383-- Note: you cannot achieve this by just changing the order of the monad
384-- statements because that would change the order in which the stream elements
387-- Note that the function @f@ must be lazy in its argument, that's why we use
388-- 'unsafeInterleaveIO' on @action@ because IO monad is strict.
394mfix :: Monad m => (m a -> Stream m a) -> Stream m a
395mfix f = fromStreamK $ K.mfix (toStreamK . f)
397------------------------------------------------------------------------------
399------------------------------------------------------------------------------
402-- >>> fromFoldable = Prelude.foldr Stream.cons Stream.nil
404-- Construct a stream from a 'Foldable' containing pure values:
408{-# INLINE fromFoldable #-}
409fromFoldable :: Foldable f => f a -> Stream m a
410fromFoldable = fromStreamK . K.fromFoldable
412------------------------------------------------------------------------------
414------------------------------------------------------------------------------
416-- | Keep reading 'Storable' elements from 'Ptr' onwards.
418-- /Unsafe:/ The caller is responsible for safe addressing.
421{-# INLINE fromPtr #-}
422fromPtr :: (MonadIO m, Storable a) => Ptr a -> Stream m a
423fromPtr = Stream.fromStreamD . D.fromPtr
425-- | Take @n@ 'Storable' elements starting from 'Ptr' onwards.
427-- >>> fromPtrN n = Stream.take n . Stream.fromPtr
429-- /Unsafe:/ The caller is responsible for safe addressing.
432{-# INLINE fromPtrN #-}
433fromPtrN :: (MonadIO m, Storable a) => Int -> Ptr a -> Stream m a
434fromPtrN n = Stream.fromStreamD . D.take n . D.fromPtr
436-- | Read bytes from an 'Addr#' until a 0 byte is encountered, the 0 byte is
437-- not included in the stream.
439-- >>> fromByteStr# addr = Stream.takeWhile (/= 0) $ Stream.fromPtr $ Ptr addr
441-- /Unsafe:/ The caller is responsible for safe addressing.
443-- Note that this is completely safe when reading from Haskell string
444-- literals because they are guaranteed to be NULL terminated:
446-- >>> Stream.fold Fold.toList $ Stream.fromByteStr# "\1\2\3\0"#
449{-# INLINE fromByteStr# #-}
450fromByteStr# :: MonadIO m => Addr# -> Stream m Word8
452 Stream.fromStreamD $ D.takeWhile (/= 0) $ D.fromPtr $ Ptr addr
454-- | Construct a stream by reading an 'Unboxed' 'IORef' repeatedly.
458{-# INLINE fromUnboxedIORef #-}
459fromUnboxedIORef :: (MonadIO m, Unbox a) => Unboxed.IORef a -> Stream m a
460fromUnboxedIORef = 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