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
91import Control.Monad.IO.Class (MonadIO(..))
92import Data.Void (Void)
93import Streamly.Internal.Control.Concurrent (MonadAsync)
94import Streamly.Internal.Data.Unfold.Type (Unfold)
95import Streamly.Internal.Data.SVar (Rate (..))
96import Streamly.Internal.Data.Stream.IsStream.Enumeration
97 (Enumerable(..), enumerate, enumerateTo)
98import Streamly.Internal.Data.Stream.IsStream.Common
99 ( absTimesWith, concatM, relTimesWith, timesWith, fromPure, fromEffect
100 , yield, yieldM, repeatM)
101import Streamly.Internal.Data.Stream.IsStream.Type
102 (IsStream (..), fromSerial, consM, fromStreamD, fromStreamS)
103import Streamly.Internal.Data.Stream.Serial (SerialT)
104import Streamly.Internal.Data.Stream.WSerial (WSerialT)
105import Streamly.Internal.Data.Stream.Zip (ZipSerialM)
106import Streamly.Internal.Data.Time.Units (AbsTime , RelTime64, addToAbsTime64)
107import Streamly.Internal.Data.Unboxed (Unboxed)
109import qualified Streamly.Internal.Data.IORef.Unboxed as Unboxed
110import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
111import qualified Streamly.Internal.Data.Stream.Parallel as Par
112import qualified Streamly.Internal.Data.Stream.Serial as Serial
113import qualified Streamly.Internal.Data.Stream.StreamD.Generate as D
114import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
115#ifdef USE_STREAMK_ONLY
116import qualified Streamly.Internal.Data.Stream.StreamK as S
117import qualified Streamly.Internal.Data.Stream.StreamK.Type as S
119import qualified Streamly.Internal.Data.Stream.StreamD.Generate as S
121import qualified Streamly.Internal.Data.Stream.Type as Stream
122import qualified System.IO as IO
124import Prelude hiding (iterate, replicate, repeat)
128-- >>> import Data.Function ((&))
129-- >>> import Prelude hiding (iterate, replicate, repeat)
130-- >>> import qualified Streamly.Prelude as Stream
131-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream
132-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
133-- >>> import Control.Concurrent (threadDelay)
134-- >>> import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
136-- >>> hSetBuffering stdout LineBuffering
138------------------------------------------------------------------------------
140------------------------------------------------------------------------------
142-- | Convert an 'Unfold' into a stream by supplying it an input seed.
144-- >>> Stream.drain $ Stream.unfold (Unfold.replicateM 3) (putStrLn "hello")
151unfold :: (IsStream t, Monad m) => Unfold m a b -> a -> t m b
152unfold unf x = fromStreamD $ D.unfold unf x
154-- | Convert an 'Unfold' with a closed input end into a stream.
157{-# INLINE unfold0 #-}
158unfold0 :: (IsStream t, Monad m) => Unfold m Void b -> t m b
159unfold0 unf = unfold unf (error "unfold0: unexpected void evaluation")
161------------------------------------------------------------------------------
162-- Generation by Unfolding
163------------------------------------------------------------------------------
169-- Nothing -> Stream.nil
170-- Just (a, b) -> a `Stream.cons` unfoldr step b
173-- Build a stream by unfolding a /pure/ step function @step@ starting from a
174-- seed @s@. The step function returns the next element in the stream and the
175-- next seed value. When it is done it returns 'Nothing' and the stream ends.
182-- else Just (b, b + 1)
183-- in Stream.toList $ Stream.unfoldr f 0
188{-# INLINE_EARLY unfoldr #-}
189unfoldr :: (Monad m, IsStream t) => (b -> Maybe (a, b)) -> b -> t m a
190unfoldr step seed = fromStreamS (S.unfoldr step seed)
191{-# RULES "unfoldr fallback to StreamK" [1]
192 forall a b. S.toStreamK (S.unfoldr a b) = K.unfoldr a b #-}
194-- | Build a stream by unfolding a /monadic/ step function starting from a
195-- seed. The step function returns the next element in the stream and the next
196-- seed value. When it is done it returns 'Nothing' and the stream ends. For
202-- then return Nothing
203-- else return (Just (b, b + 1))
204-- in Stream.toList $ Stream.unfoldrM f 0
208-- When run concurrently, the next unfold step can run concurrently with the
209-- processing of the output of the previous step. Note that more than one step
210-- cannot run concurrently as the next step depends on the output of the
216-- then return Nothing
217-- else threadDelay 1000000 >> return (Just (b, b + 1))
218-- in Stream.toList $ Stream.delay 1 $ Stream.fromAsync $ Stream.unfoldrM f 0
225{-# INLINE_EARLY unfoldrM #-}
226unfoldrM :: forall t m b a. (IsStream t, MonadAsync m) =>
227 (b -> m (Maybe (a, b))) -> b -> t m a
228unfoldrM step = fromStream . K.unfoldrMWith (IsStream.toConsK (consM @t)) step
230{-# RULES "unfoldrM serial" unfoldrM = unfoldrMSerial #-}
231{-# INLINE_EARLY unfoldrMSerial #-}
232unfoldrMSerial :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> SerialT m a
233unfoldrMSerial = Serial.unfoldrM
235{-# RULES "unfoldrM wSerial" unfoldrM = unfoldrMWSerial #-}
236{-# INLINE_EARLY unfoldrMWSerial #-}
237unfoldrMWSerial :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> WSerialT m a
238unfoldrMWSerial f = fromSerial . Serial.unfoldrM f
240{-# RULES "unfoldrM zipSerial" unfoldrM = unfoldrMZipSerial #-}
241{-# INLINE_EARLY unfoldrMZipSerial #-}
242unfoldrMZipSerial :: MonadAsync m =>
243 (b -> m (Maybe (a, b))) -> b -> ZipSerialM m a
244unfoldrMZipSerial f = fromSerial . Serial.unfoldrM f
246------------------------------------------------------------------------------
248------------------------------------------------------------------------------
251-- Generate an infinite stream by repeating a pure value.
254{-# INLINE_NORMAL repeat #-}
255repeat :: (IsStream t, Monad m) => a -> t m a
256repeat = fromStreamS . S.repeat
259-- >>> replicate n = Stream.take n . Stream.repeat
261-- Generate a stream of length @n@ by repeating a value @n@ times.
264{-# INLINE_NORMAL replicate #-}
265replicate :: (IsStream t, Monad m) => Int -> a -> t m a
266replicate n = fromStreamS . S.replicate n
269-- >>> replicateM n = Stream.take n . Stream.repeatM
271-- Generate a stream by performing a monadic action @n@ times. Same as:
273-- >>> pr n = threadDelay 1000000 >> print n
275-- This runs serially and takes 3 seconds:
277-- >>> Stream.drain $ Stream.fromSerial $ Stream.replicateM 3 $ pr 1
282-- This runs concurrently and takes just 1 second:
284-- >>> Stream.drain $ Stream.fromAsync $ Stream.replicateM 3 $ pr 1
292{-# INLINE_EARLY replicateM #-}
293replicateM :: forall t m a. (IsStream t, MonadAsync m) => Int -> m a -> t m a
295 fromStream . K.replicateMWith (IsStream.toConsK (consM @t)) count
297{-# RULES "replicateM serial" replicateM = replicateMSerial #-}
298{-# INLINE replicateMSerial #-}
299replicateMSerial :: MonadAsync m => Int -> m a -> SerialT m a
300replicateMSerial n = fromStreamS . S.replicateM n
302------------------------------------------------------------------------------
304------------------------------------------------------------------------------
306-- | @times@ returns a stream of time value tuples with clock of 10 ms
307-- granularity. The first component of the tuple is an absolute time reference
308-- (epoch) denoting the start of the stream and the second component is a time
309-- relative to the reference.
311-- >>> Stream.mapM_ (\x -> print x >> threadDelay 1000000) $ Stream.take 3 $ Stream.times
312-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
313-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
314-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
316-- Note: This API is not safe on 32-bit machines.
321times :: (IsStream t, MonadAsync m) => t m (AbsTime, RelTime64)
322times = timesWith 0.01
324-- | @absTimes@ returns a stream of absolute timestamps using a clock of 10 ms
327-- >>> Stream.mapM_ print $ Stream.delayPre 1 $ Stream.take 3 $ Stream.absTimes
328-- AbsTime (TimeSpec {sec = ..., nsec = ...})
329-- AbsTime (TimeSpec {sec = ..., nsec = ...})
330-- AbsTime (TimeSpec {sec = ..., nsec = ...})
332-- Note: This API is not safe on 32-bit machines.
336{-# INLINE absTimes #-}
337absTimes :: (IsStream t, MonadAsync m, Functor (t m)) => t m AbsTime
338absTimes = fmap (uncurry addToAbsTime64) times
340{-# DEPRECATED currentTime "Please use absTimes instead" #-}
341{-# INLINE currentTime #-}
342currentTime :: (IsStream t, MonadAsync m, Functor (t m))
343 => Double -> t m AbsTime
344currentTime = absTimesWith
346-- | @relTimes@ returns a stream of relative time values starting from 0,
347-- using a clock of granularity 10 ms.
349-- >>> Stream.mapM_ print $ Stream.delayPre 1 $ Stream.take 3 $ Stream.relTimes
350-- RelTime64 (NanoSecond64 ...)
351-- RelTime64 (NanoSecond64 ...)
352-- RelTime64 (NanoSecond64 ...)
354-- Note: This API is not safe on 32-bit machines.
358{-# INLINE relTimes #-}
359relTimes :: (IsStream t, MonadAsync m, Functor (t m)) => t m RelTime64
360relTimes = fmap snd times
362-- | @durations g@ returns a stream of relative time values measuring the time
363-- elapsed since the immediate predecessor element of the stream was generated.
364-- The first element of the stream is always 0. @durations@ uses a clock of
365-- granularity @g@ specified in seconds. A low granularity clock is more
366-- expensive in terms of CPU usage. The minimum granularity is 1 millisecond.
367-- Durations lower than 1 ms will be 0.
369-- Note: This API is not safe on 32-bit machines.
373{-# INLINE durations #-}
374durations :: -- (IsStream t, MonadAsync m) =>
375 Double -> t m RelTime64
378-- | Generate ticks at the specified rate. The rate is adaptive, the tick
379-- generation speed can be increased or decreased at different times to achieve
380-- the specified rate. The specific behavior for different styles of 'Rate'
381-- specifications is documented under 'Rate'. The effective maximum rate
382-- achieved by a stream is governed by the processor speed.
387ticks :: -- (IsStream t, MonadAsync m) =>
391-- | Generate a singleton event at or after the specified absolute time. Note
392-- that this is different from a threadDelay, a threadDelay starts from the
393-- time when the action is evaluated, whereas if we use AbsTime based timeout
394-- it will immediately expire if the action is evaluated too late.
398{-# INLINE timeout #-}
399timeout :: -- (IsStream t, MonadAsync m) =>
403------------------------------------------------------------------------------
404-- From Generator functions
405------------------------------------------------------------------------------
407-- XXX we can remove it and recommend the definition in terms of enumerate and
408-- map. Check performance equivalence.
411-- >>> fromIndices f = fmap f $ Stream.enumerateFrom 0
412-- >>> fromIndices f = let g i = f i `Stream.cons` g (i + 1) in g 0
414-- Generate an infinite stream, whose values are the output of a function @f@
415-- applied on the corresponding index. Index starts at 0.
417-- >>> Stream.toList $ Stream.take 5 $ Stream.fromIndices id
421{-# INLINE fromIndices #-}
422fromIndices :: (IsStream t, Monad m) => (Int -> a) -> t m a
423fromIndices = fromStreamS . S.fromIndices
427-- >>> fromIndicesM f = Stream.mapM f $ Stream.enumerateFrom 0
428-- >>> fromIndicesM f = let g i = f i `Stream.consM` g (i + 1) in g 0
430-- Generate an infinite stream, whose values are the output of a monadic
431-- function @f@ applied on the corresponding index. Index starts at 0.
436{-# INLINE_EARLY fromIndicesM #-}
437fromIndicesM :: forall t m a. (IsStream t, MonadAsync m) =>
438 (Int -> m a) -> t m a
439fromIndicesM = fromStream . K.fromIndicesMWith (IsStream.toConsK (consM @t))
441{-# RULES "fromIndicesM serial" fromIndicesM = fromIndicesMSerial #-}
442{-# INLINE fromIndicesMSerial #-}
443fromIndicesMSerial :: MonadAsync m => (Int -> m a) -> SerialT m a
444fromIndicesMSerial = fromStreamS . S.fromIndicesM
446------------------------------------------------------------------------------
447-- Iterating functions
448------------------------------------------------------------------------------
451-- >>> iterate f x = x `Stream.cons` iterate f x
453-- Generate an infinite stream with @x@ as the first element and each
454-- successive element derived by applying the function @f@ on the previous
457-- >>> Stream.toList $ Stream.take 5 $ Stream.iterate (+1) 1
461{-# INLINE_NORMAL iterate #-}
462iterate :: (IsStream t, Monad m) => (a -> a) -> a -> t m a
463iterate step = fromStreamS . S.iterate step
466-- >>> iterateM f m = m >>= \a -> return a `Stream.consM` iterateM f (f a)
468-- Generate an infinite stream with the first element generated by the action
469-- @m@ and each successive element derived by applying the monadic function
470-- @f@ on the previous element.
472-- >>> pr n = threadDelay 1000000 >> print n
474-- Stream.iterateM (\x -> pr x >> return (x + 1)) (return 0)
476-- & Stream.fromSerial
483-- When run concurrently, the next iteration can run concurrently with the
484-- processing of the previous iteration. Note that more than one iteration
485-- cannot run concurrently as the next iteration depends on the output of the
486-- previous iteration.
489-- Stream.iterateM (\x -> pr x >> return (x + 1)) (return 0)
503-- /Since: 0.7.0 (signature change)/
504{-# INLINE_EARLY iterateM #-}
505iterateM :: forall t m a. (IsStream t, MonadAsync m) =>
506 (a -> m a) -> m a -> t m a
507iterateM f = fromStream . K.iterateMWith (IsStream.toConsK (consM @t)) f
509{-# RULES "iterateM serial" iterateM = iterateMSerial #-}
510{-# INLINE iterateMSerial #-}
511iterateMSerial :: MonadAsync m => (a -> m a) -> m a -> SerialT m a
512iterateMSerial step = fromStreamS . S.iterateM step
514-- | We can define cyclic structures using @let@:
516-- >>> let (a, b) = ([1, b], head a) in (a, b)
519-- The function @fix@ defined as:
521-- >>> fix f = let x = f x in x
523-- ensures that the argument of a function and its output refer to the same
524-- lazy value @x@ i.e. the same location in memory. Thus @x@ can be defined
525-- in terms of itself, creating structures with cyclic references.
527-- >>> f ~(a, b) = ([1, b], head a)
531-- 'Control.Monad.mfix' is essentially the same as @fix@ but for monadic
534-- Using 'mfix' for streams we can construct a stream in which each element of
535-- the stream is defined in a cyclic fashion. The argument of the function
536-- being fixed represents the current element of the stream which is being
537-- returned by the stream monad. Thus, we can use the argument to construct
540-- In the following example, the argument @action@ of the function @f@
541-- represents the tuple @(x,y)@ returned by it in a given iteration. We define
542-- the first element of the tuple in terms of the second.
544-- >>> import Streamly.Internal.Data.Stream.IsStream as Stream
545-- >>> import System.IO.Unsafe (unsafeInterleaveIO)
548-- main = Stream.mapM_ print $ Stream.mfix f
551-- let incr n act = fmap ((+n) . snd) $ unsafeInterleaveIO act
552-- x <- Stream.fromListM [incr 1 action, incr 2 action]
553-- y <- Stream.fromList [4,5]
557-- Note: you cannot achieve this by just changing the order of the monad
558-- statements because that would change the order in which the stream elements
561-- Note that the function @f@ must be lazy in its argument, that's why we use
562-- 'unsafeInterleaveIO' on @action@ because IO monad is strict.
566mfix :: (IsStream t, Monad m) => (m a -> t m a) -> t m a
567mfix f = fromStream $ K.mfix (toStream . f)
569------------------------------------------------------------------------------
571------------------------------------------------------------------------------
574-- >>> fromFoldable = Prelude.foldr Stream.cons Stream.nil
576-- Construct a stream from a 'Foldable' containing pure values:
579{-# INLINE fromFoldable #-}
580fromFoldable :: (IsStream t, Foldable f) => f a -> t m a
581fromFoldable = fromStream . K.fromFoldable
584-- >>> fromFoldableM = Prelude.foldr Stream.consM Stream.nil
586-- Construct a stream from a 'Foldable' containing monadic actions.
588-- >>> pr n = threadDelay 1000000 >> print n
589-- >>> Stream.drain $ Stream.fromSerial $ Stream.fromFoldableM $ map pr [1,2,3]
594-- >>> Stream.drain $ Stream.fromAsync $ Stream.fromFoldableM $ map pr [1,2,3]
599-- /Concurrent (do not use with 'fromParallel' on infinite containers)/
602{-# INLINE fromFoldableM #-}
603fromFoldableM :: (IsStream t, MonadAsync m, Foldable f) => f (m a) -> t m a
604fromFoldableM = Prelude.foldr consM IsStream.nil
607-- >>> fromListM = Stream.fromFoldableM
608-- >>> fromListM = Stream.sequence . Stream.fromList
609-- >>> fromListM = Stream.mapM id . Stream.fromList
610-- >>> fromListM = Prelude.foldr Stream.consM Stream.nil
612-- Construct a stream from a list of monadic actions. This is more efficient
613-- than 'fromFoldableM' for serial streams.
616{-# INLINE_EARLY fromListM #-}
617fromListM :: (MonadAsync m, IsStream t) => [m a] -> t m a
618fromListM = fromFoldableM
619{-# RULES "fromListM fallback to StreamK" [1]
620 forall a. D.toStreamK (D.fromListM a) = K.fromFoldableM a #-}
622{-# RULES "fromListM serial" fromListM = fromListMSerial #-}
623{-# INLINE_EARLY fromListMSerial #-}
624fromListMSerial :: MonadAsync m => [m a] -> SerialT m a
625fromListMSerial = fromStreamD . D.fromListM
627-- | Read lines from an IO Handle into a stream of Strings.
630{-# DEPRECATED fromHandle
631 "Please use Streamly.FileSystem.Handle module (see the changelog)" #-}
632fromHandle :: (IsStream t, MonadIO m) => IO.Handle -> t m String
635 go = IsStream.mkStream $ \_ yld _ stp -> do
636 eof <- liftIO $ IO.hIsEOF h
640 str <- liftIO $ IO.hGetLine h
643-- XXX This should perhaps be moved to Parallel
645-- | Takes a callback setter function and provides it with a callback. The
646-- callback when invoked adds a value at the tail of the stream. Returns a
647-- stream of values generated by the callback.
651{-# INLINE fromCallback #-}
652fromCallback :: MonadAsync m => ((a -> m ()) -> m ()) -> SerialT m a
653fromCallback setCallback = concatM $ do
654 (callback, stream) <- Par.newCallbackStream
656 return $ Stream.fromStreamK stream
658-- | Construct a stream by reading an 'Unboxed' 'IORef' repeatedly.
662{-# INLINE fromPrimIORef #-}
663fromPrimIORef :: (IsStream t, MonadIO m, Unboxed a) => Unboxed.IORef a -> t m a
664fromPrimIORef = 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