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
13-- >>> fromEffect = Stream.sequence . Stream.fromPure
14-- >>> fromIndices f = fmap f $ Stream.enumerateFrom 0
15-- >>> repeatM = Stream.sequence . Stream.repeat
16-- >>> replicateM n = Stream.sequence . Stream.replicate n
17-- >>> fromListM = Stream.sequence . Stream.fromList
18-- >>> fromFoldableM = Stream.sequence . Stream.fromFoldable
20module Streamly.Internal.Data.Stream.Generate
78import Control.Monad.IO.Class (MonadIO)
79import Data.Word (Word8)
80import Foreign.Storable (Storable)
81import GHC.Exts (Addr#, Ptr (Ptr))
82import Streamly.Internal.Data.Stream.Bottom
83 (absTimesWith, relTimesWith, timesWith)
84import Streamly.Internal.Data.Stream.Enumerate
85 (Enumerable(..), enumerate, enumerateTo)
86import Streamly.Internal.Data.Stream.Type
87 (Stream, fromStreamD, fromStreamK, toStreamK)
88import Streamly.Internal.Data.Time.Units (AbsTime, RelTime64, addToAbsTime64)
89import Streamly.Internal.Data.Unboxed (Unboxed)
90import Streamly.Internal.Data.Unfold.Type (Unfold)
92import qualified Streamly.Internal.Data.IORef.Unboxed as Unboxed
93import qualified Streamly.Internal.Data.Stream.Bottom as Bottom
94import qualified Streamly.Internal.Data.Stream.StreamD as D
95import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
96import qualified Streamly.Internal.Data.Stream.Type as Stream
98import Prelude hiding (iterate, replicate, repeat)
102-- >>> import Control.Concurrent (threadDelay)
103-- >>> import Data.Function ((&))
104-- >>> import qualified Streamly.Data.Fold as Fold
105-- >>> import qualified Streamly.Data.Unfold as Unfold
106-- >>> import qualified Streamly.Internal.Data.Stream as Stream
107-- >>> import GHC.Exts (Ptr (Ptr))
109------------------------------------------------------------------------------
111------------------------------------------------------------------------------
113-- | Convert an 'Unfold' into a stream by supplying it an input seed.
115-- >>> s = Stream.unfold (Unfold.replicateM 3) (putStrLn "hello")
116-- >>> Stream.fold Fold.drain s
123unfold :: Monad m => Unfold m a b -> a -> Stream m b
124unfold unf = Stream.fromStreamD . D.unfold unf
126------------------------------------------------------------------------------
127-- Generation by Unfolding
128------------------------------------------------------------------------------
134-- Nothing -> Stream.nil
135-- Just (a, b) -> a `Stream.cons` unfoldr step b
138-- Build a stream by unfolding a /pure/ step function @step@ starting from a
139-- seed @s@. The step function returns the next element in the stream and the
140-- next seed value. When it is done it returns 'Nothing' and the stream ends.
147-- else Just (b, b + 1)
148-- in Stream.fold Fold.toList $ Stream.unfoldr f 0
152{-# INLINE_EARLY unfoldr #-}
153unfoldr :: Monad m => (b -> Maybe (a, b)) -> b -> Stream m a
154unfoldr step seed = fromStreamD (D.unfoldr step seed)
155{-# RULES "unfoldr fallback to StreamK" [1]
156 forall a b. D.toStreamK (D.unfoldr a b) = K.unfoldr a b #-}
158-- | Build a stream by unfolding a /monadic/ step function starting from a
159-- seed. The step function returns the next element in the stream and the next
160-- seed value. When it is done it returns 'Nothing' and the stream ends. For
166-- then return Nothing
167-- else return (Just (b, b + 1))
168-- in Stream.fold Fold.toList $ Stream.unfoldrM f 0
173{-# INLINE unfoldrM #-}
174unfoldrM :: Monad m => (b -> m (Maybe (a, b))) -> b -> Stream m a
175unfoldrM step = fromStreamD . D.unfoldrM step
177------------------------------------------------------------------------------
179------------------------------------------------------------------------------
182-- Generate an infinite stream by repeating a pure value.
184{-# INLINE_NORMAL repeat #-}
185repeat :: Monad m => a -> Stream m a
186repeat = fromStreamD . D.repeat
189-- >>> replicate n = Stream.take n . Stream.repeat
191-- Generate a stream of length @n@ by repeating a value @n@ times.
193{-# INLINE_NORMAL replicate #-}
194replicate :: Monad m => Int -> a -> Stream m a
195replicate n = fromStreamD . D.replicate n
197------------------------------------------------------------------------------
199------------------------------------------------------------------------------
201-- | @times@ returns a stream of time value tuples with clock of 10 ms
202-- granularity. The first component of the tuple is an absolute time reference
203-- (epoch) denoting the start of the stream and the second component is a time
204-- relative to the reference.
206-- >>> f = Fold.drainBy (\x -> print x >> threadDelay 1000000)
207-- >>> Stream.fold f $ Stream.take 3 $ Stream.times
208-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
209-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
210-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
212-- Note: This API is not safe on 32-bit machines.
217times :: MonadIO m => Stream m (AbsTime, RelTime64)
218times = timesWith 0.01
220-- | @absTimes@ returns a stream of absolute timestamps using a clock of 10 ms
223-- >>> f = Fold.drainBy print
224-- >>> Stream.fold f $ Stream.delayPre 1 $ Stream.take 3 $ Stream.absTimes
225-- AbsTime (TimeSpec {sec = ..., nsec = ...})
226-- AbsTime (TimeSpec {sec = ..., nsec = ...})
227-- AbsTime (TimeSpec {sec = ..., nsec = ...})
229-- Note: This API is not safe on 32-bit machines.
233{-# INLINE absTimes #-}
234absTimes :: MonadIO m => Stream m AbsTime
235absTimes = fmap (uncurry addToAbsTime64) times
237-- | @relTimes@ returns a stream of relative time values starting from 0,
238-- using a clock of granularity 10 ms.
240-- >>> f = Fold.drainBy print
241-- >>> Stream.fold f $ Stream.delayPre 1 $ Stream.take 3 $ Stream.relTimes
242-- RelTime64 (NanoSecond64 ...)
243-- RelTime64 (NanoSecond64 ...)
244-- RelTime64 (NanoSecond64 ...)
246-- Note: This API is not safe on 32-bit machines.
250{-# INLINE relTimes #-}
251relTimes :: MonadIO m => Stream m RelTime64
252relTimes = fmap snd times
254-- | @durations g@ returns a stream of relative time values measuring the time
255-- elapsed since the immediate predecessor element of the stream was generated.
256-- The first element of the stream is always 0. @durations@ uses a clock of
257-- granularity @g@ specified in seconds. A low granularity clock is more
258-- expensive in terms of CPU usage. The minimum granularity is 1 millisecond.
259-- Durations lower than 1 ms will be 0.
261-- Note: This API is not safe on 32-bit machines.
265{-# INLINE durations #-}
266durations :: -- Monad m =>
267 Double -> t m RelTime64
270-- | Generate a singleton event at or after the specified absolute time. Note
271-- that this is different from a threadDelay, a threadDelay starts from the
272-- time when the action is evaluated, whereas if we use AbsTime based timeout
273-- it will immediately expire if the action is evaluated too late.
277{-# INLINE timeout #-}
278timeout :: -- Monad m =>
282------------------------------------------------------------------------------
283-- Iterating functions
284------------------------------------------------------------------------------
287-- >>> iterate f x = x `Stream.cons` iterate f x
289-- Generate an infinite stream with @x@ as the first element and each
290-- successive element derived by applying the function @f@ on the previous
293-- >>> Stream.fold Fold.toList $ Stream.take 5 $ Stream.iterate (+1) 1
296{-# INLINE_NORMAL iterate #-}
297iterate :: Monad m => (a -> a) -> a -> Stream m a
298iterate step = fromStreamD . D.iterate step
301-- >>> iterateM f m = m >>= \a -> return a `Stream.consM` iterateM f (f a)
303-- Generate an infinite stream with the first element generated by the action
304-- @m@ and each successive element derived by applying the monadic function
305-- @f@ on the previous element.
308-- Stream.iterateM (\x -> print x >> return (x + 1)) (return 0)
310-- & Stream.fold Fold.toList
316{-# INLINE iterateM #-}
317iterateM :: Monad m => (a -> m a) -> m a -> Stream m a
318iterateM step = fromStreamD . D.iterateM step
320-- | We can define cyclic structures using @let@:
322-- >>> let (a, b) = ([1, b], head a) in (a, b)
325-- The function @fix@ defined as:
327-- >>> fix f = let x = f x in x
329-- ensures that the argument of a function and its output refer to the same
330-- lazy value @x@ i.e. the same location in memory. Thus @x@ can be defined
331-- in terms of itself, creating structures with cyclic references.
333-- >>> f ~(a, b) = ([1, b], head a)
337-- 'Control.Monad.mfix' is essentially the same as @fix@ but for monadic
340-- Using 'mfix' for streams we can construct a stream in which each element of
341-- the stream is defined in a cyclic fashion. The argument of the function
342-- being fixed represents the current element of the stream which is being
343-- returned by the stream monad. Thus, we can use the argument to construct
346-- In the following example, the argument @action@ of the function @f@
347-- represents the tuple @(x,y)@ returned by it in a given iteration. We define
348-- the first element of the tuple in terms of the second.
350-- >>> import Streamly.Internal.Data.Stream as Stream
351-- >>> import System.IO.Unsafe (unsafeInterleaveIO)
354-- main = Stream.fold (Fold.drainBy print) $ Stream.mfix f
357-- let incr n act = fmap ((+n) . snd) $ unsafeInterleaveIO act
358-- x <- Stream.sequence $ Stream.fromList [incr 1 action, incr 2 action]
359-- y <- Stream.fromList [4,5]
363-- Note: you cannot achieve this by just changing the order of the monad
364-- statements because that would change the order in which the stream elements
367-- Note that the function @f@ must be lazy in its argument, that's why we use
368-- 'unsafeInterleaveIO' on @action@ because IO monad is strict.
374mfix :: Monad m => (m a -> Stream m a) -> Stream m a
375mfix f = fromStreamK $ K.mfix (toStreamK . f)
377------------------------------------------------------------------------------
379------------------------------------------------------------------------------
382-- >>> fromFoldable = Prelude.foldr Stream.cons Stream.nil
384-- Construct a stream from a 'Foldable' containing pure values:
388{-# INLINE fromFoldable #-}
389fromFoldable :: Foldable f => f a -> Stream m a
390fromFoldable = fromStreamK . K.fromFoldable
392------------------------------------------------------------------------------
394------------------------------------------------------------------------------
396-- | Keep reading 'Storable' elements from 'Ptr' onwards.
398-- /Unsafe:/ The caller is responsible for safe addressing.
401{-# INLINE fromPtr #-}
402fromPtr :: (MonadIO m, Storable a) => Ptr a -> Stream m a
403fromPtr = Stream.fromStreamD . D.fromPtr
405-- | Take @n@ 'Storable' elements starting from 'Ptr' onwards.
407-- >>> fromPtrN n = Stream.take n . Stream.fromPtr
409-- /Unsafe:/ The caller is responsible for safe addressing.
412{-# INLINE fromPtrN #-}
413fromPtrN :: (MonadIO m, Storable a) => Int -> Ptr a -> Stream m a
414fromPtrN n = Stream.fromStreamD . D.take n . D.fromPtr
416-- | Read bytes from an 'Addr#' until a 0 byte is encountered, the 0 byte is
417-- not included in the stream.
419-- >>> fromByteStr# addr = Stream.takeWhile (/= 0) $ Stream.fromPtr $ Ptr addr
421-- /Unsafe:/ The caller is responsible for safe addressing.
423-- Note that this is completely safe when reading from Haskell string
424-- literals because they are guaranteed to be NULL terminated:
426-- >>> Stream.fold Fold.toList $ Stream.fromByteStr# "\1\2\3\0"#
429{-# INLINE fromByteStr# #-}
430fromByteStr# :: MonadIO m => Addr# -> Stream m Word8
432 Stream.fromStreamD $ D.takeWhile (/= 0) $ D.fromPtr $ Ptr addr
434-- | Construct a stream by reading an 'Unboxed' 'IORef' repeatedly.
438{-# INLINE fromUnboxedIORef #-}
439fromUnboxedIORef :: (MonadIO m, Unboxed a) => Unboxed.IORef a -> Stream m a
440fromUnboxedIORef = 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