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.Stream.Zip (ZipSerialM)
106import Streamly.Internal.Data.Time.Units (AbsTime , RelTime64, addToAbsTime64)
107import Streamly.Internal.Data.Unboxed (Unbox)
109import qualified Streamly.Internal.Data.IORef.Unboxed as Unboxed
111import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
112import qualified Streamly.Internal.Data.Stream.Parallel as Par
113import qualified Streamly.Internal.Data.Stream.Serial as Serial
114import qualified Streamly.Internal.Data.Stream.StreamD as D
115 (unfold, unfoldr, toStreamK, unfoldr, repeat, replicate, replicateM
116 , fromIndices, fromIndicesM, iterate, iterateM, toStreamK, fromListM
118import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
119 (unfoldr, unfoldrMWith, replicateMWith, fromIndicesMWith, iterateMWith
120 , mfix, fromFoldable, fromFoldableM)
121import qualified Streamly.Internal.Data.Stream.Type as Stream (fromStreamK)
122import qualified System.IO as IO
124import Prelude hiding (iterate, replicate, repeat)
128-- >>> :set -fno-warn-deprecations
129-- >>> import Data.Function ((&))
130-- >>> import Prelude hiding (iterate, replicate, repeat)
131-- >>> import qualified Streamly.Prelude as Stream
132-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream
133-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
134-- >>> import Control.Concurrent (threadDelay)
135-- >>> import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
137-- >>> hSetBuffering stdout LineBuffering
139------------------------------------------------------------------------------
141------------------------------------------------------------------------------
143-- | Convert an 'Unfold' into a stream by supplying it an input seed.
145-- >>> Stream.drain $ Stream.unfold Unfold.replicateM (3, putStrLn "hello")
152unfold :: (IsStream t, Monad m) => Unfold m a b -> a -> t m b
153unfold unf x = fromStreamD $ D.unfold unf x
155-- | Convert an 'Unfold' with a closed input end into a stream.
158{-# INLINE unfold0 #-}
159unfold0 :: (IsStream t, Monad m) => Unfold m Void b -> t m b
160unfold0 unf = unfold unf (error "unfold0: unexpected void evaluation")
162------------------------------------------------------------------------------
163-- Generation by Unfolding
164------------------------------------------------------------------------------
170-- Nothing -> Stream.nil
171-- Just (a, b) -> a `Stream.cons` unfoldr step b
174-- Build a stream by unfolding a /pure/ step function @step@ starting from a
175-- seed @s@. The step function returns the next element in the stream and the
176-- next seed value. When it is done it returns 'Nothing' and the stream ends.
183-- else Just (b, b + 1)
184-- in Stream.toList $ Stream.unfoldr f 0
189{-# INLINE_EARLY unfoldr #-}
190unfoldr :: (Monad m, IsStream t) => (b -> Maybe (a, b)) -> b -> t m a
191unfoldr step seed = fromStreamD (D.unfoldr step seed)
192{-# RULES "unfoldr fallback to StreamK" [1]
193 forall a b. D.toStreamK (D.unfoldr a b) = K.unfoldr a b #-}
195-- | Build a stream by unfolding a /monadic/ step function starting from a
196-- seed. The step function returns the next element in the stream and the next
197-- seed value. When it is done it returns 'Nothing' and the stream ends. For
203-- then return Nothing
204-- else return (Just (b, b + 1))
205-- in Stream.toList $ Stream.unfoldrM f 0
209-- When run concurrently, the next unfold step can run concurrently with the
210-- processing of the output of the previous step. Note that more than one step
211-- cannot run concurrently as the next step depends on the output of the
217-- then return Nothing
218-- else threadDelay 1000000 >> return (Just (b, b + 1))
219-- in Stream.toList $ Stream.delay 1 $ Stream.fromAsync $ Stream.unfoldrM f 0
226{-# INLINE_EARLY unfoldrM #-}
227unfoldrM :: forall t m b a. (IsStream t, MonadAsync m) =>
228 (b -> m (Maybe (a, b))) -> b -> t m a
229unfoldrM step = fromStream . K.unfoldrMWith (IsStream.toConsK (consM @t)) step
231{-# RULES "unfoldrM serial" unfoldrM = unfoldrMSerial #-}
232{-# INLINE_EARLY unfoldrMSerial #-}
233unfoldrMSerial :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> SerialT m a
234unfoldrMSerial = Serial.unfoldrM
236{-# RULES "unfoldrM wSerial" unfoldrM = unfoldrMWSerial #-}
237{-# INLINE_EARLY unfoldrMWSerial #-}
238unfoldrMWSerial :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> WSerialT m a
239unfoldrMWSerial f = fromSerial . Serial.unfoldrM f
241{-# RULES "unfoldrM zipSerial" unfoldrM = unfoldrMZipSerial #-}
242{-# INLINE_EARLY unfoldrMZipSerial #-}
243unfoldrMZipSerial :: MonadAsync m =>
244 (b -> m (Maybe (a, b))) -> b -> ZipSerialM m a
245unfoldrMZipSerial f = fromSerial . Serial.unfoldrM f
247------------------------------------------------------------------------------
249------------------------------------------------------------------------------
252-- Generate an infinite stream by repeating a pure value.
255{-# INLINE_NORMAL repeat #-}
256repeat :: (IsStream t, Monad m) => a -> t m a
257repeat = fromStreamD . D.repeat
260-- >>> replicate n = Stream.take n . Stream.repeat
262-- Generate a stream of length @n@ by repeating a value @n@ times.
265{-# INLINE_NORMAL replicate #-}
266replicate :: (IsStream t, Monad m) => Int -> a -> t m a
267replicate n = fromStreamD . D.replicate n
270-- >>> replicateM n = Stream.take n . Stream.repeatM
272-- Generate a stream by performing a monadic action @n@ times. Same as:
274-- >>> pr n = threadDelay 1000000 >> print n
276-- This runs serially and takes 3 seconds:
278-- >>> Stream.drain $ Stream.fromSerial $ Stream.replicateM 3 $ pr 1
283-- This runs concurrently and takes just 1 second:
285-- >>> Stream.drain $ Stream.fromAsync $ Stream.replicateM 3 $ pr 1
293{-# INLINE_EARLY replicateM #-}
294replicateM :: forall t m a. (IsStream t, MonadAsync m) => Int -> m a -> t m a
296 fromStream . K.replicateMWith (IsStream.toConsK (consM @t)) count
298{-# RULES "replicateM serial" replicateM = replicateMSerial #-}
299{-# INLINE replicateMSerial #-}
300replicateMSerial :: MonadAsync m => Int -> m a -> SerialT m a
301replicateMSerial n = fromStreamD . D.replicateM n
303------------------------------------------------------------------------------
305------------------------------------------------------------------------------
307-- | @times@ returns a stream of time value tuples with clock of 10 ms
308-- granularity. The first component of the tuple is an absolute time reference
309-- (epoch) denoting the start of the stream and the second component is a time
310-- relative to the reference.
312-- >>> Stream.mapM_ (\x -> print x >> threadDelay 1000000) $ Stream.take 3 $ Stream.times
313-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
314-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
315-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
317-- Note: This API is not safe on 32-bit machines.
322times :: (IsStream t, MonadAsync m) => t m (AbsTime, RelTime64)
323times = timesWith 0.01
325-- | @absTimes@ returns a stream of absolute timestamps using a clock of 10 ms
328-- >>> Stream.mapM_ print $ Stream.delayPre 1 $ Stream.take 3 $ Stream.absTimes
329-- AbsTime (TimeSpec {sec = ..., nsec = ...})
330-- AbsTime (TimeSpec {sec = ..., nsec = ...})
331-- AbsTime (TimeSpec {sec = ..., nsec = ...})
333-- Note: This API is not safe on 32-bit machines.
337{-# INLINE absTimes #-}
338absTimes :: (IsStream t, MonadAsync m, Functor (t m)) => t m AbsTime
339absTimes = fmap (uncurry addToAbsTime64) times
341{-# DEPRECATED currentTime "Please use absTimes instead" #-}
342{-# INLINE currentTime #-}
343currentTime :: (IsStream t, MonadAsync m, Functor (t m))
344 => Double -> t m AbsTime
345currentTime = absTimesWith
347-- | @relTimes@ returns a stream of relative time values starting from 0,
348-- using a clock of granularity 10 ms.
350-- >>> Stream.mapM_ print $ Stream.delayPre 1 $ Stream.take 3 $ Stream.relTimes
351-- RelTime64 (NanoSecond64 ...)
352-- RelTime64 (NanoSecond64 ...)
353-- RelTime64 (NanoSecond64 ...)
355-- Note: This API is not safe on 32-bit machines.
359{-# INLINE relTimes #-}
360relTimes :: (IsStream t, MonadAsync m, Functor (t m)) => t m RelTime64
361relTimes = fmap snd times
363-- | @durations g@ returns a stream of relative time values measuring the time
364-- elapsed since the immediate predecessor element of the stream was generated.
365-- The first element of the stream is always 0. @durations@ uses a clock of
366-- granularity @g@ specified in seconds. A low granularity clock is more
367-- expensive in terms of CPU usage. The minimum granularity is 1 millisecond.
368-- Durations lower than 1 ms will be 0.
370-- Note: This API is not safe on 32-bit machines.
374{-# INLINE durations #-}
375durations :: -- (IsStream t, MonadAsync m) =>
376 Double -> t m RelTime64
379-- | Generate ticks at the specified rate. The rate is adaptive, the tick
380-- generation speed can be increased or decreased at different times to achieve
381-- the specified rate. The specific behavior for different styles of 'Rate'
382-- specifications is documented under 'Rate'. The effective maximum rate
383-- achieved by a stream is governed by the processor speed.
388ticks :: -- (IsStream t, MonadAsync m) =>
392-- | Generate a singleton event at or after the specified absolute time. Note
393-- that this is different from a threadDelay, a threadDelay starts from the
394-- time when the action is evaluated, whereas if we use AbsTime based timeout
395-- it will immediately expire if the action is evaluated too late.
399{-# INLINE timeout #-}
400timeout :: -- (IsStream t, MonadAsync m) =>
404------------------------------------------------------------------------------
405-- From Generator functions
406------------------------------------------------------------------------------
408-- XXX we can remove it and recommend the definition in terms of enumerate and
409-- map. Check performance equivalence.
412-- >>> fromIndices f = fmap f $ Stream.enumerateFrom 0
413-- >>> fromIndices f = let g i = f i `Stream.cons` g (i + 1) in g 0
415-- Generate an infinite stream, whose values are the output of a function @f@
416-- applied on the corresponding index. Index starts at 0.
418-- >>> Stream.toList $ Stream.take 5 $ Stream.fromIndices id
422{-# INLINE fromIndices #-}
423fromIndices :: (IsStream t, Monad m) => (Int -> a) -> t m a
424fromIndices = fromStreamD . D.fromIndices
428-- >>> fromIndicesM f = Stream.mapM f $ Stream.enumerateFrom 0
429-- >>> fromIndicesM f = let g i = f i `Stream.consM` g (i + 1) in g 0
431-- Generate an infinite stream, whose values are the output of a monadic
432-- function @f@ applied on the corresponding index. Index starts at 0.
437{-# INLINE_EARLY fromIndicesM #-}
438fromIndicesM :: forall t m a. (IsStream t, MonadAsync m) =>
439 (Int -> m a) -> t m a
440fromIndicesM = fromStream . K.fromIndicesMWith (IsStream.toConsK (consM @t))
442{-# RULES "fromIndicesM serial" fromIndicesM = fromIndicesMSerial #-}
443{-# INLINE fromIndicesMSerial #-}
444fromIndicesMSerial :: MonadAsync m => (Int -> m a) -> SerialT m a
445fromIndicesMSerial = fromStreamD . D.fromIndicesM
447------------------------------------------------------------------------------
448-- Iterating functions
449------------------------------------------------------------------------------
452-- >>> iterate f x = x `Stream.cons` iterate f x
454-- Generate an infinite stream with @x@ as the first element and each
455-- successive element derived by applying the function @f@ on the previous
458-- >>> Stream.toList $ Stream.take 5 $ Stream.iterate (+1) 1
462{-# INLINE_NORMAL iterate #-}
463iterate :: (IsStream t, Monad m) => (a -> a) -> a -> t m a
464iterate step = fromStreamD . D.iterate step
467-- >>> iterateM f m = m >>= \a -> return a `Stream.consM` iterateM f (f a)
469-- Generate an infinite stream with the first element generated by the action
470-- @m@ and each successive element derived by applying the monadic function
471-- @f@ on the previous element.
473-- >>> pr n = threadDelay 1000000 >> print n
475-- Stream.iterateM (\x -> pr x >> return (x + 1)) (return 0)
477-- & Stream.fromSerial
484-- When run concurrently, the next iteration can run concurrently with the
485-- processing of the previous iteration. Note that more than one iteration
486-- cannot run concurrently as the next iteration depends on the output of the
487-- previous iteration.
490-- Stream.iterateM (\x -> pr x >> return (x + 1)) (return 0)
504-- /Since: 0.7.0 (signature change)/
505{-# INLINE_EARLY iterateM #-}
506iterateM :: forall t m a. (IsStream t, MonadAsync m) =>
507 (a -> m a) -> m a -> t m a
508iterateM f = fromStream . K.iterateMWith (IsStream.toConsK (consM @t)) f
510{-# RULES "iterateM serial" iterateM = iterateMSerial #-}
511{-# INLINE iterateMSerial #-}
512iterateMSerial :: MonadAsync m => (a -> m a) -> m a -> SerialT m a
513iterateMSerial step = fromStreamD . D.iterateM step
515-- | We can define cyclic structures using @let@:
517-- >>> let (a, b) = ([1, b], head a) in (a, b)
520-- The function @fix@ defined as:
522-- >>> fix f = let x = f x in x
524-- ensures that the argument of a function and its output refer to the same
525-- lazy value @x@ i.e. the same location in memory. Thus @x@ can be defined
526-- in terms of itself, creating structures with cyclic references.
528-- >>> f ~(a, b) = ([1, b], head a)
532-- 'Control.Monad.mfix' is essentially the same as @fix@ but for monadic
535-- Using 'mfix' for streams we can construct a stream in which each element of
536-- the stream is defined in a cyclic fashion. The argument of the function
537-- being fixed represents the current element of the stream which is being
538-- returned by the stream monad. Thus, we can use the argument to construct
543mfix :: (IsStream t, Monad m) => (m a -> t m a) -> t m a
544mfix f = fromStream $ K.mfix (toStream . f)
546------------------------------------------------------------------------------
548------------------------------------------------------------------------------
551-- >>> fromFoldable = Prelude.foldr Stream.cons Stream.nil
553-- Construct a stream from a 'Foldable' containing pure values:
556{-# INLINE fromFoldable #-}
557fromFoldable :: (IsStream t, Foldable f) => f a -> t m a
558fromFoldable = fromStream . K.fromFoldable
561-- >>> fromFoldableM = Prelude.foldr Stream.consM Stream.nil
563-- Construct a stream from a 'Foldable' containing monadic actions.
565-- >>> pr n = threadDelay 1000000 >> print n
566-- >>> Stream.drain $ Stream.fromSerial $ Stream.fromFoldableM $ map pr [1,2,3]
571-- >>> Stream.drain $ Stream.fromAsync $ Stream.fromFoldableM $ map pr [1,2,3]
576-- /Concurrent (do not use with 'fromParallel' on infinite containers)/
579{-# INLINE fromFoldableM #-}
580fromFoldableM :: (IsStream t, MonadAsync m, Foldable f) => f (m a) -> t m a
581fromFoldableM = Prelude.foldr consM IsStream.nil
584-- >>> fromListM = Stream.fromFoldableM
585-- >>> fromListM = Stream.sequence . Stream.fromList
586-- >>> fromListM = Stream.mapM id . Stream.fromList
587-- >>> fromListM = Prelude.foldr Stream.consM Stream.nil
589-- Construct a stream from a list of monadic actions. This is more efficient
590-- than 'fromFoldableM' for serial streams.
593{-# INLINE_EARLY fromListM #-}
594fromListM :: (MonadAsync m, IsStream t) => [m a] -> t m a
595fromListM = fromFoldableM
596{-# RULES "fromListM fallback to StreamK" [1]
597 forall a. D.toStreamK (D.fromListM a) = K.fromFoldableM a #-}
599{-# RULES "fromListM serial" fromListM = fromListMSerial #-}
600{-# INLINE_EARLY fromListMSerial #-}
601fromListMSerial :: MonadAsync m => [m a] -> SerialT m a
602fromListMSerial = fromStreamD . D.fromListM
604-- | Read lines from an IO Handle into a stream of Strings.
607{-# DEPRECATED fromHandle
608 "Please use Streamly.FileSystem.Handle module (see the changelog)" #-}
609fromHandle :: (IsStream t, MonadIO m) => IO.Handle -> t m String
612 go = IsStream.mkStream $ \_ yld _ stp -> do
613 eof <- liftIO $ IO.hIsEOF h
617 str <- liftIO $ IO.hGetLine h
620-- XXX This should perhaps be moved to Parallel
622-- | Takes a callback setter function and provides it with a callback. The
623-- callback when invoked adds a value at the tail of the stream. Returns a
624-- stream of values generated by the callback.
628{-# INLINE fromCallback #-}
629fromCallback :: MonadAsync m => ((a -> m ()) -> m ()) -> SerialT m a
630fromCallback setCallback = concatM $ do
631 (callback, stream) <- Par.newCallbackStream
633 return $ Stream.fromStreamK stream
635-- | Construct a stream by reading an 'Unboxed' 'IORef' repeatedly.
639{-# INLINE fromPrimIORef #-}
640fromPrimIORef :: (IsStream t, MonadIO m, Unbox a) => Unboxed.IORef a -> t m a
641fromPrimIORef = 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