@@ -27,6 +27,9 @@ module Asyncly.AsyncT
2727 , zipWithM
2828 , zipWith
2929 , ZipSerial (..)
30+ , zipAsyncWithM
31+ , zipAsyncWith
32+ , ZipAsync (..)
3033
3134 , interleave
3235 , (<=>)
@@ -711,7 +714,7 @@ drop n m = AsyncT $ \ctx stp yld -> do
711714-- Zipping Streams
712715------------------------------------------------------------------------------
713716
714−-- | Zip two AsyncT streams serially using a monadic function.
717+-- | Zip two AsyncT streams serially using a monadic zipping function.
715718zipWithM :: MonadAsync m =>
716719 (a -> b -> AsyncT m c) -> AsyncT m a -> AsyncT m b -> AsyncT m c
717720zipWithM f m1 m2 = AsyncT $ \_ stp yld -> do
@@ -724,7 +727,7 @@ zipWithM f m1 m2 = AsyncT $ \_ stp yld -> do
724727 yield1 a (Just ra) = merge a ra
725728 (runAsyncT m1) Nothing stp yield1
726729
727−-- | Zip two AsyncT streams serially using a pure function.
730+-- | Zip two AsyncT streams serially using a pure zipping function.
728731zipWith :: MonadAsync m
729732 => (a -> b -> c) -> AsyncT m a -> AsyncT m b -> AsyncT m c
730733zipWith f m1 m2 = AsyncT $ \_ stp yld -> do
@@ -746,6 +749,58 @@ instance MonadAsync m => Applicative (ZipSerial m) where
746749 pure a = ZipSerial $ cycle1 (pure a)
747750 (ZipSerial xs) <*> (ZipSerial ys) = ZipSerial (zipWith id xs ys)
748751
752+-- | The computation specified in the argument is pushed to a new thread and
753+-- the context is computation pulls output from the thread.
754+pushOneToCtx :: MonadAsync m => CtxType -> AsyncT m a -> m (Context m a)
755+pushOneToCtx ctype m = do
756+ ctx <- liftIO $
757+ case ctype of
758+ CtxType _ FIFO -> do
759+ c <- getFifoCtx ctype
760+ return c
761+ CtxType _ LIFO -> do
762+ c <- getLifoCtx ctype
763+ return c
764+ -- Note: We must have all the work on the queue before sending the
765+ -- pushworker, otherwise the pushworker may exit before we even get a
766+ -- chance to push.
767+ liftIO $ (enqueue ctx) m
768+ pushWorker ctx
769+ return ctx
770+
771+async :: MonadAsync m => AsyncT m a -> m (AsyncT m a)
772+async m = do
773+ ctx <- pushOneToCtx (CtxType Disjunction LIFO) m
774+ return $ pullFromCtx ctx
775+
776+-- | Zip two AsyncT streams asyncly (i.e. both the streams are generated
777+-- concurrently) using a monadic zipping function.
778+zipAsyncWithM :: MonadAsync m
779+ => (a -> b -> AsyncT m c) -> AsyncT m a -> AsyncT m b -> AsyncT m c
780+zipAsyncWithM f m1 m2 = AsyncT $ \_ stp yld -> do
781+ ma <- async m1
782+ mb <- async m2
783+ (runAsyncT (zipWithM f ma mb)) Nothing stp yld
784+
785+-- | Zip two AsyncT streams asyncly (i.e. both the streams are generated
786+-- concurrently) using a pure zipping function.
787+zipAsyncWith :: MonadAsync m
788+ => (a -> b -> c) -> AsyncT m a -> AsyncT m b -> AsyncT m c
789+zipAsyncWith f m1 m2 = AsyncT $ \_ stp yld -> do
790+ ma <- async m1
791+ mb <- async m2
792+ (runAsyncT (zipWith f ma mb)) Nothing stp yld
793+
794+-- | Wrapper around AsyncT type with a parallel zipping Applicative instance
795+--
796+-- > f <$> ZipAsync xs1 <*> ... <*> ZipAsync xsN
797+newtype ZipAsync m a = ZipAsync {getZipAsync :: AsyncT m a}
798+ deriving (Functor)
799+
800+instance MonadAsync m => Applicative (ZipAsync m) where
801+ pure a = ZipAsync $ cycle1 (pure a)
802+ (ZipAsync xs) <*> (ZipAsync ys) = ZipAsync (zipAsyncWith id xs ys)
803+
749804------------------------------------------------------------------------------
750805-- Utilities
751806------------------------------------------------------------------------------