@@ -54,14 +54,17 @@ import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)
5454import Streamly.Internal.Data.Stream.Serial (SerialT(..))
5555import Streamly.Internal.Data.Stream.StreamK.Type (Stream)
5656
57−import qualified Streamly.Internal.Data.Stream.Parallel as Par
5857import qualified Streamly.Internal.Data.Stream.Prelude as P
5958 (cmpBy, eqBy, foldl', foldr, fromList, toList)
6059import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
60+import qualified Streamly.Internal.Data.Stream.StreamK as K
6161import qualified Streamly.Internal.Data.Stream.StreamD as D
6262import 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
6365
6466import Prelude hiding (map, repeat, zipWith, errorWithoutStackTrace)
67+import Streamly.Internal.Data.SVar
6568
6669#include "Instances.hs"
6770
@@ -96,9 +99,10 @@ zipWithK f = zipWithMK (\a b -> return (f a b))
9699{-# INLINE zipAsyncWithMK #-}
97100zipAsyncWithMK :: MonadAsync m
98101 => (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c
99−zipAsyncWithMK f m1 m2 = D.toStreamK $
100− D.zipWithM f (Par.mkParallelD $ D.fromStreamK m1)
101− (Par.mkParallelD $ D.fromStreamK m2)
102+zipAsyncWithMK f m1 m2 = K.mkStream $ \st yld sng stp -> do
103+ sv <- newParallelVar StopNone (adaptState st)
104+ SVar.toSVarParallel (adaptState st) sv $ D.fromStreamK m2
105+ K.foldStream st yld sng stp $ K.zipWithM f m1 (getSerialT (SVar.fromSVar sv))
102106
103107-- XXX Should we rename this to zipParWith or zipParallelWith? This can happen
104108-- along with the change of behvaior to end the stream concurrently.