1{-# OPTIONS_GHC -Wno-deprecations #-}
2{-# OPTIONS_GHC -Wno-orphans #-}
5-- Module : Streamly.Internal.Data.Stream.IsStream.Generate
6-- Copyright : (c) 2017 Composewell Technologies
7-- License : BSD-3-Clause
8-- Maintainer : streamly@composewell.com
9-- Stability : experimental
12-- Most of the combinators in this module can be implemented as unfolds. Some
13-- of them however can only be expressed in terms StreamK e.g. cons/consM,
14-- fromFoldable, mfix. We can possibly remove those from this module which can
15-- be expressed as unfolds. Unless we want to use rewrite rules to rewrite them
16-- as StreamK when StreamK is used, avoiding conversion to StreamD. Will that
17-- help? Are there any other reasons to keep these and not use unfolds?
19module Streamly.Internal.Data.Stream.IsStream.Generate {-# DEPRECATED "Please use \"Streamly.Data.Stream.*\" instead." #-}
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.IsStream.Type
103 (IsStream (..), fromSerial, consM, fromStreamD)
104import Streamly.Internal.Data.Stream.Serial (SerialT, WSerialT)
105import Streamly.Internal.Data.Time.Units (AbsTime , RelTime64, addToAbsTime64)
106import Streamly.Internal.Data.Unboxed (Unbox)
108import 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 as D
114 (unfold, unfoldr, toStreamK, unfoldr, repeat, replicate, replicateM
115 , fromIndices, fromIndicesM, iterate, iterateM, toStreamK, fromListM
117import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
118 (unfoldr, unfoldrMWith, replicateMWith, fromIndicesMWith, iterateMWith
119 , mfix, fromFoldable, fromFoldableM)
120import qualified Streamly.Internal.Data.Stream.Serial as Stream (fromStreamK)
121import qualified System.IO as IO
123import Prelude hiding (iterate, replicate, repeat)
127-- >>> :set -fno-warn-deprecations
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 = fromStreamD (D.unfoldr step seed)
191{-# RULES "unfoldr fallback to StreamK" [1]
192 forall a b. D.toStreamK (D.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 -> IsStream.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 = fromStreamD . D.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 = fromStreamD . D.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 = fromStreamD . D.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 = fromStreamD . D.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 = fromStreamD . D.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 = fromStreamD . D.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 = fromStreamD . D.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
542mfix :: (IsStream t, Monad m) => (m a -> t m a) -> t m a
543mfix f = fromStream $ K.mfix (toStream . f)
545------------------------------------------------------------------------------
547------------------------------------------------------------------------------
550-- >>> fromFoldable = Prelude.foldr Stream.cons Stream.nil
552-- Construct a stream from a 'Foldable' containing pure values:
555{-# INLINE fromFoldable #-}
556fromFoldable :: (IsStream t, Foldable f) => f a -> t m a
557fromFoldable = fromStream . K.fromFoldable
560-- >>> fromFoldableM = Prelude.foldr Stream.consM Stream.nil
562-- Construct a stream from a 'Foldable' containing monadic actions.
564-- >>> pr n = threadDelay 1000000 >> print n
565-- >>> Stream.drain $ Stream.fromSerial $ Stream.fromFoldableM $ map pr [1,2,3]
570-- >>> Stream.drain $ Stream.fromAsync $ Stream.fromFoldableM $ map pr [1,2,3]
575-- /Concurrent (do not use with 'fromParallel' on infinite containers)/
578{-# INLINE fromFoldableM #-}
579fromFoldableM :: (IsStream t, MonadAsync m, Foldable f) => f (m a) -> t m a
580fromFoldableM = Prelude.foldr consM IsStream.nil
583-- >>> fromListM = Stream.fromFoldableM
584-- >>> fromListM = Stream.sequence . Stream.fromList
585-- >>> fromListM = Stream.mapM id . Stream.fromList
586-- >>> fromListM = Prelude.foldr Stream.consM Stream.nil
588-- Construct a stream from a list of monadic actions. This is more efficient
589-- than 'fromFoldableM' for serial streams.
592{-# INLINE_EARLY fromListM #-}
593fromListM :: (MonadAsync m, IsStream t) => [m a] -> t m a
594fromListM = fromFoldableM
595{-# RULES "fromListM fallback to StreamK" [1]
596 forall a. D.toStreamK (D.fromListM a) = K.fromFoldableM a #-}
598{-# RULES "fromListM serial" fromListM = fromListMSerial #-}
599{-# INLINE_EARLY fromListMSerial #-}
600fromListMSerial :: MonadAsync m => [m a] -> SerialT m a
601fromListMSerial = fromStreamD . D.fromListM
603-- | Read lines from an IO Handle into a stream of Strings.
606{-# DEPRECATED fromHandle
607 "Please use Streamly.FileSystem.Handle module (see the changelog)" #-}
608fromHandle :: (IsStream t, MonadIO m) => IO.Handle -> t m String
611 go = IsStream.mkStream $ \_ yld _ stp -> do
612 eof <- liftIO $ IO.hIsEOF h
616 str <- liftIO $ IO.hGetLine h
619-- XXX This should perhaps be moved to Parallel
621-- | Takes a callback setter function and provides it with a callback. The
622-- callback when invoked adds a value at the tail of the stream. Returns a
623-- stream of values generated by the callback.
627{-# INLINE fromCallback #-}
628fromCallback :: MonadAsync m => ((a -> m ()) -> m ()) -> SerialT m a
629fromCallback setCallback = concatM $ do
630 (callback, stream) <- Par.newCallbackStream
632 return $ Stream.fromStreamK stream
634-- | Construct a stream by reading an 'Unboxed' 'IORef' repeatedly.
638{-# INLINE fromPrimIORef #-}
639fromPrimIORef :: (IsStream t, MonadIO m, Unbox a) => Unboxed.IORef a -> t m a
640fromPrimIORef = 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