9fe6dc1 · streamly · git.iko.soy

Changelog.md+1

compare with master

unifiedsplit

@@ -2,6 +2,7 @@

2

3### Breaking changes

4

5+* Signatures of `zipWithM` and `zipAsyncWithM` have changed

5* Some functions in prelude now require an additional Monad constraint for the

6* Some functions in prelude now require an additional Monad constraint for the

6 underlying type of the stream.

7 underlying type of the stream.

7

8

benchmark/Linear.hs+2

@@ -97,6 +97,7 @@ main = do

97 , benchIO "dropWhileM-true" Ops.dropWhileMTrue

98 ]

99 , benchIO "zip" $ Ops.zip

100+ , benchIO "zipM" $ Ops.zipM

100 , bgroup "compose"

101 , bgroup "compose"

101 [ benchIO "mapM" Ops.composeMapM

102 [ benchIO "mapM" Ops.composeMapM

102 , benchIO "map-with-all-in-filter" Ops.composeMapAllInFilter

103 , benchIO "map-with-all-in-filter" Ops.composeMapAllInFilter

@@ -151,5 +152,6 @@ main = do

151 , benchIO "mapM" $ Ops.mapM parallely

152 , benchIO "mapM" $ Ops.mapM parallely

152 -- Zip has only one parallel flavor

153 -- Zip has only one parallel flavor

153 , benchIO "zip" $ Ops.zipAsync

154 , benchIO "zip" $ Ops.zipAsync

155+ , benchIO "zipM" $ Ops.zipAsyncM

154 ]

156 ]

155 ]

157 ]

benchmark/LinearOps.hs+21−4

@@ -9,6 +9,7 @@

9

10module LinearOps where

11

12+import Data.Maybe (fromJust)

12import Prelude

13import Prelude

13 (Monad, Int, (+), ($), (.), return, fmap, even, (>), (<=),

14 (Monad, Int, (+), ($), (.), return, fmap, even, (>), (<=),

14 subtract, undefined, Maybe(..), odd, Bool, not)

15 subtract, undefined, Maybe(..), odd, Bool, not)

@@ -42,13 +43,14 @@ maxValue = value + 1000

42{-# INLINE dropWhileTrue #-}

43{-# INLINE dropWhileTrue #-}

43{-# INLINE dropWhileMTrue #-}

44{-# INLINE dropWhileMTrue #-}

44{-# INLINE zip #-}

45{-# INLINE zip #-}

46+{-# INLINE zipM #-}

45{-# INLINE concat #-}

47{-# INLINE concat #-}

46{-# INLINE composeAllInFilters #-}

48{-# INLINE composeAllInFilters #-}

47{-# INLINE composeAllOutFilters #-}

49{-# INLINE composeAllOutFilters #-}

48{-# INLINE composeMapAllInFilter #-}

50{-# INLINE composeMapAllInFilter #-}

49uncons, nullHeadTail, scan, mapM_, map, fmap, mapMaybe, filterEven, filterAllOut,

51uncons, nullHeadTail, scan, mapM_, map, fmap, mapMaybe, filterEven, filterAllOut,

50 filterAllIn, takeOne, takeAll, takeWhileTrue, takeWhileMTrue, dropAll,

52 filterAllIn, takeOne, takeAll, takeWhileTrue, takeWhileMTrue, dropAll,

51− dropWhileTrue, dropWhileMTrue, zip,

53+ dropWhileTrue, dropWhileMTrue, zip, zipM,

52 concat, composeAllInFilters, composeAllOutFilters,

54 concat, composeAllInFilters, composeAllOutFilters,

53 composeMapAllInFilter

55 composeMapAllInFilter

54 :: Monad m

56 :: Monad m

@@ -56,8 +58,9 @@ uncons, nullHeadTail, scan, mapM_, map, fmap, mapMaybe, filterEven, filterAllOut

56

58

57{-# INLINE composeMapM #-}

59{-# INLINE composeMapM #-}

58{-# INLINE zipAsync #-}

60{-# INLINE zipAsync #-}

61+{-# INLINE zipAsyncM #-}

59{-# INLINE mapMaybeM #-}

62{-# INLINE mapMaybeM #-}

60−composeMapM, zipAsync, mapMaybeM :: S.MonadAsync m => Stream m Int -> m ()

63+composeMapM, zipAsync, zipAsyncM, mapMaybeM :: S.MonadAsync m => Stream m Int -> m ()

61

64

62{-# INLINE toList #-}

65{-# INLINE toList #-}

63{-# INLINE foldr #-}

66{-# INLINE foldr #-}

@@ -228,8 +231,22 @@ dropWhileMTrue = transform . S.dropWhileM (return . (<= maxValue))

228-- Zipping and concat

231-- Zipping and concat

229-------------------------------------------------------------------------------

232-------------------------------------------------------------------------------

230

233

234+zip src = do

235+ r <- S.tail src

236+ let src1 = fromJust r

231−zip src = transform $ (S.zipWith (,) src src)

237+ transform $ (S.zipWith (,) src src1)

238+zipM src = do

239+ r <- S.tail src

240+ let src1 = fromJust r

241+ transform $ (S.zipWithM (\a b -> return (a,b)) src src1)

242+zipAsync src = do

243+ r <- S.tail src

244+ let src1 = fromJust r

232−zipAsync src = transform $ (S.zipAsyncWith (,) src src)

245+ transform $ (S.zipAsyncWith (,) src src1)

246+zipAsyncM src = do

247+ r <- S.tail src

248+ let src1 = fromJust r

249+ transform $ (S.zipAsyncWithM (\a b -> return (a,b)) src src1)

233concat _n = return ()

250concat _n = return ()

234

251

235-------------------------------------------------------------------------------

252-------------------------------------------------------------------------------

src/Streamly/Prelude.hs+22−3

@@ -141,8 +141,8 @@ module Streamly.Prelude

141 -- * Zipping

142 , zipWith

143 , zipWithM

144− , zipAsyncWith

144+ , Z.zipAsyncWith

145− , zipAsyncWithM

145+ , Z.zipAsyncWithM

146

147 -- * Deprecated

148 , K.once

@@ -166,13 +166,14 @@ import qualified System.IO as IO

166import Streamly.SVar (MonadAsync)

167import Streamly.Streams.StreamK (IsStream(..))

168import Streamly.Streams.Serial (SerialT)

169−import Streamly.Streams.Zip (zipWith, zipWithM, zipAsyncWith, zipAsyncWithM)

170

169

171import qualified Streamly.Streams.StreamK as K

170import qualified Streamly.Streams.StreamK as K

172import qualified Streamly.Streams.StreamD as D

171import qualified Streamly.Streams.StreamD as D

172+import qualified Streamly.Streams.Zip as Z

173

174#ifdef USE_STREAMK_ONLY

175import qualified Streamly.Streams.StreamK as S

176+import qualified Streamly.Streams.Zip as S

176#else

177#else

177import qualified Streamly.Streams.StreamD as S

178import qualified Streamly.Streams.StreamD as S

178#endif

179#endif

@@ -801,3 +802,21 @@ reverse m = fromStream $ go K.nil (toStream m)

801 single a = runIt $ a `K.cons` rev

802 single a = runIt $ a `K.cons` rev

802 yieldk a r = runIt $ go (a `K.cons` rev) r

803 yieldk a r = runIt $ go (a `K.cons` rev) r

803 in K.unStream rest Nothing stop single yieldk

804 in K.unStream rest Nothing stop single yieldk

805+

806+------------------------------------------------------------------------------

807+-- Zipping

808+------------------------------------------------------------------------------

809+

810+-- | Zip two streams serially using a monadic zipping function.

811+--

812+-- @since 0.4.0

813+{-# INLINABLE zipWithM #-}

814+zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c

815+zipWithM f m1 m2 = fromStreamS $ S.zipWithM f (toStreamS m1) (toStreamS m2)

816+

817+-- | Zip two streams serially using a pure zipping function.

818+--

819+-- @since 0.1.0

820+{-# INLINABLE zipWith #-}

821+zipWith :: (IsStream t, Monad m) => (a -> b -> c) -> t m a -> t m b -> t m c

822+zipWith f m1 m2 = fromStreamS $ S.zipWith f (toStreamS m1) (toStreamS m2)

src/Streamly/Streams/StreamD.hs+37−1

@@ -114,6 +114,10 @@ module Streamly.Streams.StreamD

114 -- ** Map and Filter

115 , mapMaybe

116 , mapMaybeM

117+

118+ -- * Zipping

119+ , zipWith

120+ , zipWithM

117 )

121 )

118where

122where

119

123

@@ -122,7 +126,7 @@ import GHC.Types ( SPEC(..) )

122import Prelude

126import Prelude

123 hiding (map, mapM, mapM_, repeat, foldr, last, take, filter,

127 hiding (map, mapM, mapM_, repeat, foldr, last, take, filter,

124 takeWhile, drop, dropWhile, all, any, maximum, minimum, elem,

128 takeWhile, drop, dropWhile, all, any, maximum, minimum, elem,

125− notElem, null, head, tail)

129+ notElem, null, head, tail, zipWith)

126

130

127import Streamly.SVar (MonadAsync)

131import Streamly.SVar (MonadAsync)

128import qualified Streamly.Streams.StreamK as K

132import qualified Streamly.Streams.StreamK as K

@@ -624,6 +628,7 @@ map f = mapM (return . f)

624-- Transformation by Map and Filter

628-- Transformation by Map and Filter

625------------------------------------------------------------------------------

629------------------------------------------------------------------------------

626

630

631+-- XXX Will this always fuse properly?

627{-# INLINE_NORMAL mapMaybe #-}

632{-# INLINE_NORMAL mapMaybe #-}

628mapMaybe :: Monad m => (a -> Maybe b) -> Stream m a -> Stream m b

633mapMaybe :: Monad m => (a -> Maybe b) -> Stream m a -> Stream m b

629mapMaybe f = fmap fromJust . filter isJust . map f

634mapMaybe f = fmap fromJust . filter isJust . map f

@@ -636,6 +641,37 @@ mapMaybeM f = fmap fromJust . filter isJust . mapM f

636-- Instances

641-- Instances

637------------------------------------------------------------------------------

642------------------------------------------------------------------------------

638

643

644+{-# INLINE_NORMAL zipWithM #-}

645+zipWithM :: Monad m

646+ => (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c

647+zipWithM f (Stream stepa ta) (Stream stepb tb) = Stream step (ta, tb, Nothing)

648+ where

649+ {-# INLINE_LATE step #-}

650+ step (sa, sb, Nothing) = do

651+ r <- stepa sa

652+ case r of

653+ Yield x sa' -> step (sa', sb, Just x)

654+ Stop -> return Stop

655+

656+ step (sa, sb, Just x) = do

657+ r <- stepb sb

658+ case r of

659+ Yield y sb' -> do

660+ z <- f x y

661+ return $ Yield z (sa, sb', Nothing)

662+ Stop -> return Stop

663+

664+{-# RULES "zipWithM xs xs"

665+ forall f xs. zipWithM f xs xs = mapM (\x -> f x x) xs #-}

666+

667+{-# INLINE zipWith #-}

668+zipWith :: Monad m => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c

669+zipWith f = zipWithM (\a b -> return (f a b))

670+

671+------------------------------------------------------------------------------

672+-- Instances

673+------------------------------------------------------------------------------

674+

639instance Monad m => Functor (Stream m) where

675instance Monad m => Functor (Stream m) where

640 {-# INLINE fmap #-}

676 {-# INLINE fmap #-}

641 fmap = map

677 fmap = map

src/Streamly/Streams/Zip.hs+5−5

@@ -77,14 +77,14 @@ zipWith f m1 m2 = fromStream $ zipWithS f (toStream m1) (toStream m2)

77-- | Zip two streams serially using a monadic zipping function.

78--

79-- @since 0.1.0

80−zipWithM :: IsStream t => (a -> b -> t m c) -> t m a -> t m b -> t m c

80+zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c

81zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

82 where

83 go mx my = Stream $ \_ stp sng yld -> do

84 let merge a ra =

85 let runIt x = unStream x Nothing stp sng yld

86− single2 b = runIt $ toStream (f a b)

86+ single2 b = f a b >>= sng

87− yield2 b rb = runIt $ toStream (f a b) <> go ra rb

87+ yield2 b rb = f a b >>= \x -> runIt (x `K.cons` go ra rb)

88 in unStream my Nothing stp single2 yield2

89 let single1 a = merge a K.nil

90 yield1 a ra = merge a ra

@@ -177,9 +177,9 @@ zipAsyncWith f m1 m2 = fromStream $ Stream $ \_ stp sng yld -> do

177-- | Zip two streams asyncly (i.e. both the elements being zipped are generated

178-- concurrently) using a monadic zipping function.

179--

180−-- @since 0.1.0

180+-- @since 0.4.0

181zipAsyncWithM :: (IsStream t, MonadAsync m)

182− => (a -> b -> t m c) -> t m a -> t m b -> t m c

182+ => (a -> b -> m c) -> t m a -> t m b -> t m c

183zipAsyncWithM f m1 m2 = fromStream $ Stream $ \_ stp sng yld -> do

184 ma <- mkAsync m1

185 mb <- mkAsync m2

test/Prop.hs+1−1

@@ -525,7 +525,7 @@ zipApplicative constr t eq (a, b) = withMaxSuccess maxTestCount $

525 equals eq stream3 list

526

527zipMonadic

528− :: (IsStream t, Monad (t IO))

528+ :: IsStream t

529 => ([Int] -> t IO Int)

530 -> (t IO (Int, Int) -> SerialT IO (Int, Int))

531 -> ([(Int, Int)] -> [(Int, Int)] -> Bool)