@@ -39,6 +39,11 @@ module Asyncly.Prelude
3939 , drop
4040 , dropWhile
4141
42+ -- * Transformation
43+ , mapM
44+ , mapM_
45+ , sequence
46+
4247 -- * Zipping
4348 , zipWith
4449 , zipWithM
@@ -51,7 +56,8 @@ import Control.Monad (liftM)
5156import Control.Monad.IO.Class (MonadIO(..))
5257import Data.Semigroup (Semigroup(..))
5358import Prelude hiding (filter, drop, dropWhile, take,
54− takeWhile, zipWith, foldr, foldl)
59+ takeWhile, zipWith, foldr, foldl,
60+ mapM, mapM_, sequence)
5561import qualified System.IO as IO
5662
5763import Asyncly.Core
@@ -217,6 +223,42 @@ dropWhile p m = fromStream $ go (toStream m)
217223 | otherwise = yld a (Just x)
218224 in (runStream m1) ctx stp yield
219225
226+------------------------------------------------------------------------------
227+-- Transformation
228+------------------------------------------------------------------------------
229+
230+-- | Apply a monadic action to each element of the stream and replace it with
231+-- the result.
232+mapM :: (Streaming t, Monad m) => (a -> m b) -> t m a -> t m b
233+mapM f m = fromStream $ go (toStream m)
234+ where
235+ go m1 = Stream $ \_ stp yld -> do
236+ let stop = stp
237+ yield a Nothing = f a >>= \b -> yld b Nothing
238+ yield a (Just x) = f a >>= \b -> yld b (Just (go x))
239+ in (runStream m1) Nothing stop yield
240+
241+-- | Apply a monadic action to each element of the stream and discard the
242+-- results.
243+mapM_ :: (Streaming t, Monad m) => (a -> m b) -> t m a -> m ()
244+mapM_ f m = go (toStream m)
245+ where
246+ go m1 =
247+ let stop = return ()
248+ yield a Nothing = f a >> return ()
249+ yield a (Just x) = f a >> go x
250+ in (runStream m1) Nothing stop yield
251+
252+-- | Reduce a stream of monadic actions to a stream of their results.
253+sequence :: (Streaming t, Monad m) => t m (m a) -> t m a
254+sequence m = fromStream $ go (toStream m)
255+ where
256+ go m1 = Stream $ \_ stp yld -> do
257+ let stop = stp
258+ yield a Nothing = a >>= \b -> yld b Nothing
259+ yield a (Just x) = a >>= \b -> yld b (Just (go x))
260+ in (runStream m1) Nothing stop yield
261+
220262------------------------------------------------------------------------------
221263-- Serially Zipping Streams
222264------------------------------------------------------------------------------