git.iko.soy/streamly
on master help sign in
Modify the behaviour of Zip.zipAsyncWithMK
4210138adithyaov
src/Streamly/Internal/Data/Stream/Zip.hs+8−4
compare with master
unifiedsplit
@@ -54,14 +54,17 @@ import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)
54import Streamly.Internal.Data.Stream.Serial (SerialT(..))
54import Streamly.Internal.Data.Stream.Serial (SerialT(..))
55import Streamly.Internal.Data.Stream.StreamK.Type (Stream)
55import Streamly.Internal.Data.Stream.StreamK.Type (Stream)
56
56
57−import qualified Streamly.Internal.Data.Stream.Parallel as Par
58import qualified Streamly.Internal.Data.Stream.Prelude as P
57import qualified Streamly.Internal.Data.Stream.Prelude as P
59 (cmpBy, eqBy, foldl', foldr, fromList, toList)
58 (cmpBy, eqBy, foldl', foldr, fromList, toList)
60import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
59import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
60+import qualified Streamly.Internal.Data.Stream.StreamK as K
61import qualified Streamly.Internal.Data.Stream.StreamD as D
61import qualified Streamly.Internal.Data.Stream.StreamD as D
62import qualified Streamly.Internal.Data.Stream.Serial as Serial
62import qualified Streamly.Internal.Data.Stream.Serial as Serial
63+import qualified Streamly.Internal.Data.Stream.SVar.Eliminate as SVar
64+import qualified Streamly.Internal.Data.Stream.SVar.Generate as SVar
63
65
64import Prelude hiding (map, repeat, zipWith, errorWithoutStackTrace)
66import Prelude hiding (map, repeat, zipWith, errorWithoutStackTrace)
67+import Streamly.Internal.Data.SVar
65
68
66#include "Instances.hs"
69#include "Instances.hs"
67
70
@@ -96,9 +99,10 @@ zipWithK f = zipWithMK (\a b -> return (f a b))
96{-# INLINE zipAsyncWithMK #-}
99{-# INLINE zipAsyncWithMK #-}
97zipAsyncWithMK :: MonadAsync m
100zipAsyncWithMK :: MonadAsync m
98 => (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c
101 => (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c
99−zipAsyncWithMK f m1 m2 = D.toStreamK $
102+zipAsyncWithMK f m1 m2 = K.mkStream $ \st yld sng stp -> do
100− D.zipWithM f (Par.mkParallelD $ D.fromStreamK m1)
103+ sv <- newParallelVar StopNone (adaptState st)
101− (Par.mkParallelD $ D.fromStreamK m2)
104+ SVar.toSVarParallel (adaptState st) sv $ D.fromStreamK m2
105+ K.foldStream st yld sng stp $ K.zipWithM f m1 (getSerialT (SVar.fromSVar sv))
102
106
103-- XXX Should we rename this to zipParWith or zipParallelWith? This can happen
107-- XXX Should we rename this to zipParWith or zipParallelWith? This can happen
104-- along with the change of behvaior to end the stream concurrently.
108-- along with the change of behvaior to end the stream concurrently.