1{-# OPTIONS_GHC -Wno-orphans #-}
4-- Module : Streamly.Internal.Data.Stream.IsStream.Generate
5-- Copyright : (c) 2017 Composewell Technologies
6-- License : BSD-3-Clause
7-- Maintainer : streamly@composewell.com
8-- Stability : experimental
11-- Most of the combinators in this module can be implemented as unfolds. Some
12-- of them however can only be expressed in terms StreamK e.g. cons/consM,
13-- fromFoldable, mfix. We can possibly remove those from this module which can
14-- be expressed as unfolds. Unless we want to use rewrite rules to rewrite them
15-- as StreamK when StreamK is used, avoiding conversion to StreamD. Will that
16-- help? Are there any other reasons to keep these and not use unfolds?
18module Streamly.Internal.Data.Stream.IsStream.Generate
92import Control.Monad.IO.Class (MonadIO(..))
93import Data.Void (Void)
94import Streamly.Internal.Control.Concurrent (MonadAsync)
95import Streamly.Internal.Data.Unfold.Type (Unfold)
96import Streamly.Internal.Data.SVar (Rate (..))
97import Streamly.Internal.Data.Stream.IsStream.Enumeration
98 (Enumerable(..), enumerate, enumerateTo)
99import Streamly.Internal.Data.Stream.IsStream.Common
100 ( absTimesWith, concatM, relTimesWith, timesWith, fromPure, fromEffect
101 , yield, yieldM, repeatM)
102import Streamly.Internal.Data.Stream.Prelude (fromStreamS)
103import Streamly.Internal.Data.Stream.StreamD (fromStreamD)
104import Streamly.Internal.Data.Stream.StreamK (IsStream((|:), consM))
105import Streamly.Internal.Data.Stream.Serial (SerialT, WSerialT)
106import Streamly.Internal.Data.Stream.Zip (ZipSerialM)
107import Streamly.Internal.Data.Time.Units (AbsTime , RelTime64, addToAbsTime64)
109import qualified Streamly.Internal.Data.Stream.Parallel as Par
110import qualified Streamly.Internal.Data.Stream.Prelude as P
111import qualified Streamly.Internal.Data.Stream.Serial as Serial
112import qualified Streamly.Internal.Data.Stream.StreamD.Generate as D
113import qualified Streamly.Internal.Data.Stream.StreamK as K
114#ifdef USE_STREAMK_ONLY
115import qualified Streamly.Internal.Data.Stream.StreamK as S
117import qualified Streamly.Internal.Data.Stream.StreamD.Generate as S
119import qualified System.IO as IO
121import Prelude hiding (iterate, replicate, repeat)
125-- >>> import Prelude hiding (iterate, replicate, repeat)
126-- >>> import qualified Streamly.Prelude as Stream
127-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream
128-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
129-- >>> import Control.Concurrent (threadDelay)
131------------------------------------------------------------------------------
133------------------------------------------------------------------------------
135-- | Convert an 'Unfold' into a stream by supplying it an input seed.
137-- >>> Stream.drain $ Stream.unfold (Unfold.replicateM 3) (putStrLn "hello")
144unfold :: (IsStream t, Monad m) => Unfold m a b -> a -> t m b
145unfold unf x = fromStreamD $ D.unfold unf x
147-- | Convert an 'Unfold' with a closed input end into a stream.
150{-# INLINE unfold0 #-}
151unfold0 :: (IsStream t, Monad m) => Unfold m Void b -> t m b
152unfold0 unf = unfold unf (error "unfold0: unexpected void evaluation")
154------------------------------------------------------------------------------
155-- Generation by Unfolding
156------------------------------------------------------------------------------
163-- Just (a, b) -> a \`cons` unfoldr step b
166-- Build a stream by unfolding a /pure/ step function @step@ starting from a
167-- seed @s@. The step function returns the next element in the stream and the
168-- next seed value. When it is done it returns 'Nothing' and the stream ends.
176-- else Just (b, b + 1)
177-- in Stream.toList $ Stream.unfoldr f 0
184{-# INLINE_EARLY unfoldr #-}
185unfoldr :: (Monad m, IsStream t) => (b -> Maybe (a, b)) -> b -> t m a
186unfoldr step seed = fromStreamS (S.unfoldr step seed)
187{-# RULES "unfoldr fallback to StreamK" [1]
188 forall a b. S.toStreamK (S.unfoldr a b) = K.unfoldr a b #-}
190-- | Build a stream by unfolding a /monadic/ step function starting from a
191-- seed. The step function returns the next element in the stream and the next
192-- seed value. When it is done it returns 'Nothing' and the stream ends. For
199-- then return Nothing
200-- else return (Just (b, b + 1))
201-- in Stream.toList $ Stream.unfoldrM f 0
206-- When run concurrently, the next unfold step can run concurrently with the
207-- processing of the output of the previous step. Note that more than one step
208-- cannot run concurrently as the next step depends on the output of the
212-- (fromAsync $ S.unfoldrM (\\n -> liftIO (threadDelay 1000000) >> return (Just (n, n + 1))) 0)
213-- & S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
219{-# INLINE_EARLY unfoldrM #-}
220unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a
223{-# RULES "unfoldrM serial" unfoldrM = unfoldrMSerial #-}
224{-# INLINE_EARLY unfoldrMSerial #-}
225unfoldrMSerial :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> SerialT m a
226unfoldrMSerial = Serial.unfoldrM
228{-# RULES "unfoldrM wSerial" unfoldrM = unfoldrMWSerial #-}
229{-# INLINE_EARLY unfoldrMWSerial #-}
230unfoldrMWSerial :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> WSerialT m a
231unfoldrMWSerial = Serial.unfoldrM
233{-# RULES "unfoldrM zipSerial" unfoldrM = unfoldrMZipSerial #-}
234{-# INLINE_EARLY unfoldrMZipSerial #-}
235unfoldrMZipSerial :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> ZipSerialM m a
236unfoldrMZipSerial = Serial.unfoldrM
238------------------------------------------------------------------------------
240------------------------------------------------------------------------------
243-- Generate an infinite stream by repeating a pure value.
246{-# INLINE_NORMAL repeat #-}
247repeat :: (IsStream t, Monad m) => a -> t m a
248repeat = fromStreamS . S.repeat
252-- replicate = take n . repeat
255-- Generate a stream of length @n@ by repeating a value @n@ times.
258{-# INLINE_NORMAL replicate #-}
259replicate :: (IsStream t, Monad m) => Int -> a -> t m a
260replicate n = fromStreamS . S.replicate n
264-- replicateM = take n . repeatM
267-- Generate a stream by performing a monadic action @n@ times. Same as:
270-- drain $ fromSerial $ S.replicateM 10 $ (threadDelay 1000000 >> print 1)
271-- drain $ fromAsync $ S.replicateM 10 $ (threadDelay 1000000 >> print 1)
277{-# INLINE_EARLY replicateM #-}
278replicateM :: (IsStream t, MonadAsync m) => Int -> m a -> t m a
279replicateM = K.replicateM
281{-# RULES "replicateM serial" replicateM = replicateMSerial #-}
282{-# INLINE replicateMSerial #-}
283replicateMSerial :: MonadAsync m => Int -> m a -> SerialT m a
284replicateMSerial n = fromStreamS . S.replicateM n
286------------------------------------------------------------------------------
288------------------------------------------------------------------------------
290-- | @times@ returns a stream of time value tuples with clock of 10 ms
291-- granularity. The first component of the tuple is an absolute time reference
292-- (epoch) denoting the start of the stream and the second component is a time
293-- relative to the reference.
295-- >>> Stream.mapM_ (\x -> print x >> threadDelay 1000000) $ Stream.take 3 $ Stream.times
296-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
297-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
298-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
300-- Note: This API is not safe on 32-bit machines.
305times :: (IsStream t, MonadAsync m) => t m (AbsTime, RelTime64)
306times = timesWith 0.01
308-- | @absTimes@ returns a stream of absolute timestamps using a clock of 10 ms
311-- >>> Stream.mapM_ print $ Stream.delayPre 1 $ Stream.take 3 $ Stream.absTimes
312-- AbsTime (TimeSpec {sec = ..., nsec = ...})
313-- AbsTime (TimeSpec {sec = ..., nsec = ...})
314-- AbsTime (TimeSpec {sec = ..., nsec = ...})
316-- Note: This API is not safe on 32-bit machines.
320{-# INLINE absTimes #-}
321absTimes :: (IsStream t, MonadAsync m, Functor (t m)) => t m AbsTime
322absTimes = fmap (uncurry addToAbsTime64) times
324{-# DEPRECATED currentTime "Please use absTimes instead" #-}
325{-# INLINE currentTime #-}
326currentTime :: (IsStream t, MonadAsync m, Functor (t m))
327 => Double -> t m AbsTime
328currentTime = absTimesWith
330-- | @relTimes@ returns a stream of relative time values starting from 0,
331-- using a clock of granularity 10 ms.
333-- >>> Stream.mapM_ print $ Stream.delayPre 1 $ Stream.take 3 $ Stream.relTimes
334-- RelTime64 (NanoSecond64 ...)
335-- RelTime64 (NanoSecond64 ...)
336-- RelTime64 (NanoSecond64 ...)
338-- Note: This API is not safe on 32-bit machines.
342{-# INLINE relTimes #-}
343relTimes :: (IsStream t, MonadAsync m, Functor (t m)) => t m RelTime64
344relTimes = fmap snd times
346-- | @durations g@ returns a stream of relative time values measuring the time
347-- elapsed since the immediate predecessor element of the stream was generated.
348-- The first element of the stream is always 0. @durations@ uses a clock of
349-- granularity @g@ specified in seconds. A low granularity clock is more
350-- expensive in terms of CPU usage. The minimum granularity is 1 millisecond.
351-- Durations lower than 1 ms will be 0.
353-- Note: This API is not safe on 32-bit machines.
357{-# INLINE durations #-}
358durations :: -- (IsStream t, MonadAsync m) =>
359 Double -> t m RelTime64
362-- | Generate ticks at the specified rate. The rate is adaptive, the tick
363-- generation speed can be increased or decreased at different times to achieve
364-- the specified rate. The specific behavior for different styles of 'Rate'
365-- specifications is documented under 'Rate'. The effective maximum rate
366-- achieved by a stream is governed by the processor speed.
371ticks :: -- (IsStream t, MonadAsync m) =>
375-- | Generate a singleton event at or after the specified absolute time. Note
376-- that this is different from a threadDelay, a threadDelay starts from the
377-- time when the action is evaluated, whereas if we use AbsTime based timeout
378-- it will immediately expire if the action is evaluated too late.
382{-# INLINE timeout #-}
383timeout :: -- (IsStream t, MonadAsync m) =>
387------------------------------------------------------------------------------
388-- From Generator functions
389------------------------------------------------------------------------------
391-- XXX we can remove it and recommend the definition in terms of enumerate and
392-- map. Check performance equivalence.
396-- fromIndices f = fmap f $ Stream.enumerateFrom 0
397-- fromIndices f = let g i = f i \`cons` g (i + 1) in g 0
400-- Generate an infinite stream, whose values are the output of a function @f@
401-- applied on the corresponding index. Index starts at 0.
403-- >>> Stream.toList $ Stream.take 5 $ Stream.fromIndices id
407{-# INLINE fromIndices #-}
408fromIndices :: (IsStream t, Monad m) => (Int -> a) -> t m a
409fromIndices = fromStreamS . S.fromIndices
414-- fromIndicesM f = Stream.mapM f $ Stream.enumerateFrom 0
415-- fromIndicesM f = let g i = f i \`consM` g (i + 1) in g 0
418-- Generate an infinite stream, whose values are the output of a monadic
419-- function @f@ applied on the corresponding index. Index starts at 0.
424{-# INLINE_EARLY fromIndicesM #-}
425fromIndicesM :: (IsStream t, MonadAsync m) => (Int -> m a) -> t m a
426fromIndicesM = K.fromIndicesM
428{-# RULES "fromIndicesM serial" fromIndicesM = fromIndicesMSerial #-}
429{-# INLINE fromIndicesMSerial #-}
430fromIndicesMSerial :: MonadAsync m => (Int -> m a) -> SerialT m a
431fromIndicesMSerial = fromStreamS . S.fromIndicesM
433------------------------------------------------------------------------------
434-- Iterating functions
435------------------------------------------------------------------------------
439-- iterate f x = x \`cons` iterate f x
442-- Generate an infinite stream with @x@ as the first element and each
443-- successive element derived by applying the function @f@ on the previous
447-- >>> Stream.toList $ Stream.take 5 $ Stream.iterate (+1) 1
453{-# INLINE_NORMAL iterate #-}
454iterate :: (IsStream t, Monad m) => (a -> a) -> a -> t m a
455iterate step = fromStreamS . S.iterate step
459-- iterateM f m = m >>= \a -> return a \`consM` iterateM f (f a)
462-- Generate an infinite stream with the first element generated by the action
463-- @m@ and each successive element derived by applying the monadic function
464-- @f@ on the previous element.
466-- When run concurrently, the next iteration can run concurrently with the
467-- processing of the previous iteration. Note that more than one iteration
468-- cannot run concurrently as the next iteration depends on the output of the
469-- previous iteration.
472-- drain $ fromSerial $ S.take 10 $ S.iterateM
473-- (\\x -> threadDelay 1000000 >> print x >> return (x + 1)) (return 0)
475-- drain $ fromAsync $ S.take 10 $ S.iterateM
476-- (\\x -> threadDelay 1000000 >> print x >> return (x + 1)) (return 0)
483-- /Since: 0.7.0 (signature change)/
484{-# INLINE_EARLY iterateM #-}
485iterateM :: (IsStream t, MonadAsync m) => (a -> m a) -> m a -> t m a
488{-# RULES "iterateM serial" iterateM = iterateMSerial #-}
489{-# INLINE iterateMSerial #-}
490iterateMSerial :: MonadAsync m => (a -> m a) -> m a -> SerialT m a
491iterateMSerial step = fromStreamS . S.iterateM step
493------------------------------------------------------------------------------
495------------------------------------------------------------------------------
499-- fromFoldableM = 'Prelude.foldr' 'consM' 'K.nil'
502-- Construct a stream from a 'Foldable' containing monadic actions.
505-- drain $ fromSerial $ S.fromFoldableM $ replicateM 10 (threadDelay 1000000 >> print 1)
506-- drain $ fromAsync $ S.fromFoldableM $ replicateM 10 (threadDelay 1000000 >> print 1)
509-- /Concurrent (do not use with 'fromParallel' on infinite containers)/
512{-# INLINE fromFoldableM #-}
513fromFoldableM :: (IsStream t, MonadAsync m, Foldable f) => f (m a) -> t m a
514fromFoldableM = Prelude.foldr consM K.nil
518-- fromListM = 'Prelude.foldr' 'K.consM' 'K.nil'
521-- Construct a stream from a list of monadic actions. This is more efficient
522-- than 'fromFoldableM' for serial streams.
525{-# INLINE_EARLY fromListM #-}
526fromListM :: (MonadAsync m, IsStream t) => [m a] -> t m a
527fromListM = fromFoldableM
528{-# RULES "fromListM fallback to StreamK" [1]
529 forall a. D.toStreamK (D.fromListM a) = fromFoldableM a #-}
531{-# RULES "fromListM serial" fromListM = fromListMSerial #-}
532{-# INLINE_EARLY fromListMSerial #-}
533fromListMSerial :: MonadAsync m => [m a] -> SerialT m a
534fromListMSerial = fromStreamD . D.fromListM
536-- | Same as 'fromFoldable'.
539{-# DEPRECATED each "Please use fromFoldable instead." #-}
541each :: (IsStream t, Foldable f) => f a -> t m a
544-- | Read lines from an IO Handle into a stream of Strings.
547{-# DEPRECATED fromHandle
548 "Please use Streamly.FileSystem.Handle module (see the changelog)" #-}
549fromHandle :: (IsStream t, MonadIO m) => IO.Handle -> t m String
552 go = K.mkStream $ \_ yld _ stp -> do
553 eof <- liftIO $ IO.hIsEOF h
557 str <- liftIO $ IO.hGetLine h
560-- XXX This should perhaps be moved to Parallel
562-- | Takes a callback setter function and provides it with a callback. The
563-- callback when invoked adds a value at the tail of the stream. Returns a
564-- stream of values generated by the callback.
568{-# INLINE fromCallback #-}
569fromCallback :: MonadAsync m => ((a -> m ()) -> m ()) -> SerialT m a
570fromCallback setCallback = concatM $ do
571 (callback, stream) <- Par.newCallbackStream
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