901567d · streamly · git.iko.soy

git.iko.soy/streamly

Add repeatM and replicateM to concurrent streams

commit901567da87b3b31e37710ff61910cdfe3dd44f09 authorAdithya Kumar <adithya@composewell.com> dateTue 8 Nov 2022, 14:41 (4 years ago) parent1d68418

4 files changed+32−6

compare with master

unifiedsplit

core/src/Streamly/Internal/Data/Stream/Generate.hs−3 src/Streamly/Data/Stream/Concurrent.hs+5 src/Streamly/Internal/Data/Stream/Concurrent.hs+26−1 src/Streamly/Internal/Data/Stream/Time.hs+1−2

core/src/Streamly/Internal/Data/Stream/Generate.hs−3

@@ -192,8 +192,6 @@ repeat = fromStreamD . D.repeat

192192-- & Stream.fold Fold.drain

193193-- :}

194194--

195−-- /Pre-release/

196−--

197195{-# INLINE_NORMAL repeatM #-}

198196repeatM :: Monad m => m a -> Stream m a

199197repeatM = Stream.sequence . repeat

@@ -211,7 +209,6 @@ replicate n = fromStreamD . D.replicate n

211209-- >>> replicateM n = Stream.sequence . Stream.replicate n

212210--

213211-- Generate a stream by performing a monadic action @n@ times.

214−-- /Pre-release/

215212{-# INLINE_NORMAL replicateM #-}

216213replicateM :: Monad m => Int -> m a -> Stream m a

217214replicateM n = Stream.sequence . replicate n

src/Streamly/Data/Stream/Concurrent.hs+5

@@ -45,6 +45,11 @@ module Streamly.Data.Stream.Concurrent

4545 , eval

4646 , evalWith

4747

48+ -- ** Generate

49+ -- | Uses a single channel to evaluate mapped actions concurrently.

50+ , repeatM

51+ , replicateM

52+

4853 -- ** Map

4954 -- | Uses a single channel to evaluate mapped actions concurrently.

5055 , mapM

src/Streamly/Internal/Data/Stream/Concurrent.hs+26−1

@@ -51,13 +51,17 @@ module Streamly.Internal.Data.Stream.Concurrent

5151 , evalWith

5252 -- Add unfoldrM/iterateM?

5353

54+ -- ** Generate

55+ -- | Uses a single channel to evaluate all actions.

56+ , repeatM

57+ , replicateM

58+

5459 -- ** Map

5560 -- | Uses a single channel to evaluate all actions.

5661 , mapM

5762 , mapMWith

5863 , sequence

5964 , sequenceWith

60− -- Add repeatM/replicateM?

6165

6266 -- ** Combine two

6367 -- | Use a channel for each pair.

@@ -655,6 +659,27 @@ zipWith :: MonadAsync m

655659 => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c

656660zipWith f = zipWithM (\a b -> return $ f a b)

657661

662+-------------------------------------------------------------------------------

663+-- Generate

664+-------------------------------------------------------------------------------

665+

666+-- |

667+-- >>> repeatM = Concur.sequence . Stream.repeat

668+--

669+-- Generate a stream by repeatedly executing a monadic action forever. The

670+-- number of actions that are executed concurrently is limited by 'maxThreads'.

671+{-# INLINE repeatM #-}

672+repeatM :: MonadAsync m => m a -> Stream m a

673+repeatM = sequence . Stream.repeat

674+

675+-- |

676+-- >>> replicateM n = Concur.sequence . Stream.replicate n

677+--

678+-- Generate a stream by concurrently performing a monadic action @n@ times.

679+{-# INLINE replicateM #-}

680+replicateM :: MonadAsync m => Int -> m a -> Stream m a

681+replicateM n = sequence . Stream.replicate n

682+

658683-------------------------------------------------------------------------------

659684-- Reactive

660685-------------------------------------------------------------------------------

src/Streamly/Internal/Data/Stream/Time.hs+1−2

@@ -113,12 +113,11 @@ import Streamly.Internal.Data.Stream.Concurrent

113113-- /Pre-release/

114114{-# INLINE interjectSuffix #-}

115115interjectSuffix :: MonadAsync m => Double -> m a -> Stream m a -> Stream m a

116−interjectSuffix n f xs = parallelFst [xs, repeatM timed]

116+interjectSuffix n f xs = parallelFst [xs, Stream.repeatM timed]

117117

118118 where

119119

120120 timed = liftIO (threadDelay (round $ n * 1000000)) >> f

121− repeatM = Stream.sequence . Stream.repeat

122121

123122-- | Generate ticks at the specified rate. The rate is adaptive, the tick

124123-- generation speed can be increased or decreased at different times to achieve