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
69import Control.Monad.IO.Class (MonadIO)
70import Data.Word (Word8)
71import Foreign.Storable (Storable)
72import GHC.Exts (Addr#, Ptr (Ptr))
73import Streamly.Internal.Data.Stream.Bottom
74 (absTimesWith, relTimesWith, timesWith)
75import Streamly.Internal.Data.Stream.Enumerate
76 (Enumerable(..), enumerate, enumerateTo)
77import Streamly.Internal.Data.Stream.Type
78 (Stream, fromStreamD, fromStreamK, toStreamK)
79import Streamly.Internal.Data.Time.Units (AbsTime, RelTime64, addToAbsTime64)
80import Streamly.Internal.Data.Unboxed (Unboxed)
81import Streamly.Internal.Data.Unfold.Type (Unfold)
83import qualified Streamly.Internal.Data.IORef.Unboxed as Unboxed
84import qualified Streamly.Internal.Data.Stream.Bottom as Bottom
85import qualified Streamly.Internal.Data.Stream.StreamD as D
86import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
87import qualified Streamly.Internal.Data.Stream.Type as Stream
89import Prelude hiding (iterate, replicate, repeat)
93-- >>> import Control.Concurrent (threadDelay)
94-- >>> import Data.Function ((&))
95-- >>> import qualified Streamly.Data.Fold as Fold
96-- >>> import qualified Streamly.Data.Unfold as Unfold
97-- >>> import qualified Streamly.Internal.Data.Stream as Stream
98-- >>> import GHC.Exts (Ptr (Ptr))
100------------------------------------------------------------------------------
102------------------------------------------------------------------------------
104-- | Convert an 'Unfold' into a stream by supplying it an input seed.
106-- >>> s = Stream.unfold Unfold.replicateM (3, putStrLn "hello")
107-- >>> Stream.fold Fold.drain s
114unfold :: Monad m => Unfold m a b -> a -> Stream m b
115unfold unf = Stream.fromStreamD . D.unfold unf
117------------------------------------------------------------------------------
118-- Generation by Unfolding
119------------------------------------------------------------------------------
125-- Nothing -> Stream.nil
126-- Just (a, b) -> a `Stream.cons` unfoldr step b
129-- Build a stream by unfolding a /pure/ step function @step@ starting from a
130-- seed @s@. The step function returns the next element in the stream and the
131-- next seed value. When it is done it returns 'Nothing' and the stream ends.
138-- else Just (b, b + 1)
139-- in Stream.fold Fold.toList $ Stream.unfoldr f 0
144{-# INLINE_EARLY unfoldr #-}
145unfoldr :: Monad m => (b -> Maybe (a, b)) -> b -> Stream m a
146unfoldr step seed = fromStreamD (D.unfoldr step seed)
147{-# RULES "unfoldr fallback to StreamK" [1]
148 forall a b. D.toStreamK (D.unfoldr a b) = K.unfoldr a b #-}
150-- | Build a stream by unfolding a /monadic/ step function starting from a
151-- seed. The step function returns the next element in the stream and the next
152-- seed value. When it is done it returns 'Nothing' and the stream ends. For
158-- then return Nothing
159-- else return (Just (b, b + 1))
160-- in Stream.fold Fold.toList $ Stream.unfoldrM f 0
165{-# INLINE unfoldrM #-}
166unfoldrM :: Monad m => (b -> m (Maybe (a, b))) -> b -> Stream m a
167unfoldrM step = fromStreamD . D.unfoldrM step
169------------------------------------------------------------------------------
171------------------------------------------------------------------------------
174-- 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-- >>> replicate n = Stream.take n . Stream.repeat
184-- Generate a stream of length @n@ by repeating a value @n@ times.
187{-# INLINE_NORMAL replicate #-}
188replicate :: Monad m => Int -> a -> Stream m a
189replicate n = fromStreamD . D.replicate n
191------------------------------------------------------------------------------
193------------------------------------------------------------------------------
195-- | @times@ returns a stream of time value tuples with clock of 10 ms
196-- granularity. The first component of the tuple is an absolute time reference
197-- (epoch) denoting the start of the stream and the second component is a time
198-- relative to the reference.
200-- >>> f = Fold.drainBy (\x -> print x >> threadDelay 1000000)
201-- >>> Stream.fold f $ Stream.take 3 $ Stream.times
202-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
203-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
204-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
206-- Note: This API is not safe on 32-bit machines.
211times :: MonadIO m => Stream m (AbsTime, RelTime64)
212times = timesWith 0.01
214-- | @absTimes@ returns a stream of absolute timestamps using a clock of 10 ms
217-- >>> f = Fold.drainBy print
218-- >>> Stream.fold f $ Stream.delayPre 1 $ Stream.take 3 $ Stream.absTimes
219-- AbsTime (TimeSpec {sec = ..., nsec = ...})
220-- AbsTime (TimeSpec {sec = ..., nsec = ...})
221-- AbsTime (TimeSpec {sec = ..., nsec = ...})
223-- Note: This API is not safe on 32-bit machines.
227{-# INLINE absTimes #-}
228absTimes :: MonadIO m => Stream m AbsTime
229absTimes = fmap (uncurry addToAbsTime64) times
231-- | @relTimes@ returns a stream of relative time values starting from 0,
232-- using a clock of granularity 10 ms.
234-- >>> f = Fold.drainBy print
235-- >>> Stream.fold f $ Stream.delayPre 1 $ Stream.take 3 $ Stream.relTimes
236-- RelTime64 (NanoSecond64 ...)
237-- RelTime64 (NanoSecond64 ...)
238-- RelTime64 (NanoSecond64 ...)
240-- Note: This API is not safe on 32-bit machines.
244{-# INLINE relTimes #-}
245relTimes :: MonadIO m => Stream m RelTime64
246relTimes = fmap snd times
248-- | @durations g@ returns a stream of relative time values measuring the time
249-- elapsed since the immediate predecessor element of the stream was generated.
250-- The first element of the stream is always 0. @durations@ uses a clock of
251-- granularity @g@ specified in seconds. A low granularity clock is more
252-- expensive in terms of CPU usage. The minimum granularity is 1 millisecond.
253-- Durations lower than 1 ms will be 0.
255-- Note: This API is not safe on 32-bit machines.
259{-# INLINE durations #-}
260durations :: -- Monad m =>
261 Double -> t m RelTime64
264-- | Generate a singleton event at or after the specified absolute time. Note
265-- that this is different from a threadDelay, a threadDelay starts from the
266-- time when the action is evaluated, whereas if we use AbsTime based timeout
267-- it will immediately expire if the action is evaluated too late.
271{-# INLINE timeout #-}
272timeout :: -- Monad m =>
276------------------------------------------------------------------------------
277-- Iterating functions
278------------------------------------------------------------------------------
281-- >>> iterate f x = x `Stream.cons` iterate f x
283-- Generate an infinite stream with @x@ as the first element and each
284-- successive element derived by applying the function @f@ on the previous
287-- >>> Stream.fold Fold.toList $ Stream.take 5 $ Stream.iterate (+1) 1
291{-# INLINE_NORMAL iterate #-}
292iterate :: Monad m => (a -> a) -> a -> Stream m a
293iterate step = fromStreamD . D.iterate step
296-- >>> iterateM f m = m >>= \a -> return a `Stream.consM` iterateM f (f a)
298-- Generate an infinite stream with the first element generated by the action
299-- @m@ and each successive element derived by applying the monadic function
300-- @f@ on the previous element.
303-- Stream.iterateM (\x -> print x >> return (x + 1)) (return 0)
305-- & Stream.fold Fold.toList
312{-# INLINE iterateM #-}
313iterateM :: Monad m => (a -> m a) -> m a -> Stream m a
314iterateM step = fromStreamD . D.iterateM step
316-- | We can define cyclic structures using @let@:
318-- >>> let (a, b) = ([1, b], head a) in (a, b)
321-- The function @fix@ defined as:
323-- >>> fix f = let x = f x in x
325-- ensures that the argument of a function and its output refer to the same
326-- lazy value @x@ i.e. the same location in memory. Thus @x@ can be defined
327-- in terms of itself, creating structures with cyclic references.
329-- >>> f ~(a, b) = ([1, b], head a)
333-- 'Control.Monad.mfix' is essentially the same as @fix@ but for monadic
336-- Using 'mfix' for streams we can construct a stream in which each element of
337-- the stream is defined in a cyclic fashion. The argument of the function
338-- being fixed represents the current element of the stream which is being
339-- returned by the stream monad. Thus, we can use the argument to construct
342-- In the following example, the argument @action@ of the function @f@
343-- represents the tuple @(x,y)@ returned by it in a given iteration. We define
344-- the first element of the tuple in terms of the second.
346-- >>> import Streamly.Internal.Data.Stream as Stream
347-- >>> import System.IO.Unsafe (unsafeInterleaveIO)
350-- main = Stream.fold (Fold.drainBy print) $ Stream.mfix f
353-- let incr n act = fmap ((+n) . snd) $ unsafeInterleaveIO act
354-- x <- Stream.sequence $ Stream.fromList [incr 1 action, incr 2 action]
355-- y <- Stream.fromList [4,5]
359-- Note: you cannot achieve this by just changing the order of the monad
360-- statements because that would change the order in which the stream elements
363-- Note that the function @f@ must be lazy in its argument, that's why we use
364-- 'unsafeInterleaveIO' on @action@ because IO monad is strict.
370mfix :: Monad m => (m a -> Stream m a) -> Stream m a
371mfix f = fromStreamK $ K.mfix (toStreamK . f)
373------------------------------------------------------------------------------
375------------------------------------------------------------------------------
378-- >>> fromFoldable = Prelude.foldr Stream.cons Stream.nil
380-- Construct a stream from a 'Foldable' containing pure values:
385{-# INLINE fromFoldable #-}
386fromFoldable :: Foldable f => f a -> Stream m a
387fromFoldable = fromStreamK . K.fromFoldable
389------------------------------------------------------------------------------
391------------------------------------------------------------------------------
393-- | Keep reading 'Storable' elements from 'Ptr' onwards.
395-- /Unsafe:/ The caller is responsible for safe addressing.
398{-# INLINE fromPtr #-}
399fromPtr :: (MonadIO m, Storable a) => Ptr a -> Stream m a
400fromPtr = Stream.fromStreamD . D.fromPtr
402-- | Take @n@ 'Storable' elements starting from 'Ptr' onwards.
404-- >>> fromPtrN n = Stream.take n . Stream.fromPtr
406-- /Unsafe:/ The caller is responsible for safe addressing.
409{-# INLINE fromPtrN #-}
410fromPtrN :: (MonadIO m, Storable a) => Int -> Ptr a -> Stream m a
411fromPtrN n = Stream.fromStreamD . D.take n . D.fromPtr
413-- | Read bytes from an 'Addr#' until a 0 byte is encountered, the 0 byte is
414-- not included in the stream.
416-- >>> fromByteStr# addr = Stream.takeWhile (/= 0) $ Stream.fromPtr $ Ptr addr
418-- /Unsafe:/ The caller is responsible for safe addressing.
420-- Note that this is completely safe when reading from Haskell string
421-- literals because they are guaranteed to be NULL terminated:
423-- >>> Stream.fold Fold.toList $ Stream.fromByteStr# "\1\2\3\0"#
426{-# INLINE fromByteStr# #-}
427fromByteStr# :: MonadIO m => Addr# -> Stream m Word8
429 Stream.fromStreamD $ D.takeWhile (/= 0) $ D.fromPtr $ Ptr addr
431-- | Construct a stream by reading an 'Unboxed' 'IORef' repeatedly.
435{-# INLINE fromUnboxedIORef #-}
436fromUnboxedIORef :: (MonadIO m, Unboxed a) => Unboxed.IORef a -> Stream m a
437fromUnboxedIORef = 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