901567d · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add repeatM and replicateM to concurrent streams

901567dAdithya Kumar4 years ago

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

compare with master

unifiedsplit

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

192-- & Stream.fold Fold.drain

193-- :}

194--

195−-- /Pre-release/

196−--

197{-# INLINE_NORMAL repeatM #-}

195{-# INLINE_NORMAL repeatM #-}

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

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

199repeatM = Stream.sequence . repeat

197repeatM = Stream.sequence . repeat

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

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

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

212--

210--

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

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

214−-- /Pre-release/

215{-# INLINE_NORMAL replicateM #-}

212{-# INLINE_NORMAL replicateM #-}

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

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

217replicateM n = Stream.sequence . replicate n

214replicateM n = Stream.sequence . replicate n

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

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

45 , eval

46 , evalWith

48+ -- ** Generate

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

50+ , repeatM

51+ , replicateM

52+

48 -- ** Map

53 -- ** Map

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

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

50 , mapM

55 , mapM

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

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

51 , evalWith

52 -- Add unfoldrM/iterateM?

54+ -- ** Generate

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

56+ , repeatM

57+ , replicateM

58+

54 -- ** Map

59 -- ** Map

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

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

56 , mapM

61 , mapM

57 , mapMWith

62 , mapMWith

58 , sequence

63 , sequence

59 , sequenceWith

64 , sequenceWith

60− -- Add repeatM/replicateM?

62 -- ** Combine two

66 -- ** Combine two

63 -- | Use a channel for each pair.

67 -- | Use a channel for each pair.

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

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

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

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

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

657

661

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+

658-------------------------------------------------------------------------------

683-------------------------------------------------------------------------------

659-- Reactive

684-- Reactive

660-------------------------------------------------------------------------------

685-------------------------------------------------------------------------------

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

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

113-- /Pre-release/

114{-# INLINE interjectSuffix #-}

115interjectSuffix :: 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]

117

118 where

119

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

121− repeatM = Stream.sequence . Stream.repeat

122

121

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

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

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

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