{-# LANGUAGE CPP                       #-}
{-# LANGUAGE ConstraintKinds           #-}
{-# LANGUAGE FlexibleContexts          #-}
{-# LANGUAGE FlexibleInstances         #-}
{-# LANGUAGE GeneralizedNewtypeDeriving#-}
{-# LANGUAGE InstanceSigs              #-}
{-# LANGUAGE MultiParamTypeClasses     #-}
{-# LANGUAGE TypeFamilies              #-}
{-# LANGUAGE UndecidableInstances      #-} -- XXX

-- |
-- Module      : Streamly.Internal.Data.Stream.Zip
-- Copyright   : (c) 2017 Harendra Kumar
--
-- License     : BSD3
-- Maintainer  : streamly@composewell.com
-- Stability   : experimental
-- Portability : GHC
--
--
module Streamly.Internal.Data.Stream.Zip
    (
      ZipSerialM
    , ZipSerial
    , zipSerially

    , ZipAsyncM
    , ZipAsync
    , zipAsyncly

    , zipWith
    , zipWithM
    , zipAsyncWith
    , zipAsyncWithM

    -- * Deprecated
    , ZipStream
    , zipping
    , zippingAsync
    )
where

import Control.Applicative (liftA2)
import Control.DeepSeq (NFData(..))
#if MIN_VERSION_deepseq(1,4,3)
import Control.DeepSeq (NFData1(..))
#endif
import Data.Foldable (Foldable(foldl'), fold)
import Data.Functor.Identity (Identity(..), runIdentity)
import Data.Maybe (fromMaybe)
import Data.Semigroup (Endo(..))
#if __GLASGOW_HASKELL__ < 808
import Data.Semigroup (Semigroup(..))
#endif
import GHC.Exts (IsList(..), IsString(..))
import Text.Read (Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec,
                  readListPrecDefault)
import Prelude hiding (map, repeat, zipWith, errorWithoutStackTrace)

import Streamly.Internal.BaseCompat ((#.), errorWithoutStackTrace)
import Streamly.Internal.Data.Stream.StreamK (IsStream(..), Stream)
import Streamly.Internal.Data.Strict (Maybe'(..), toMaybe)
import Streamly.Internal.Data.SVar (MonadAsync)

import qualified Streamly.Internal.Data.Stream.Prelude as P
import qualified Streamly.Internal.Data.Stream.StreamK as K
import qualified Streamly.Internal.Data.Stream.StreamD as D

#ifdef USE_STREAMK_ONLY
import qualified Streamly.Internal.Data.Stream.StreamK as S
#else
import qualified Streamly.Internal.Data.Stream.StreamD as S
#endif

#include "Instances.hs"

-- | Like 'zipWith' but using a monadic zipping function.
--
-- @since 0.4.0
{-# INLINABLE zipWithM #-}
zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c
zipWithM :: (a -> b -> m c) -> t m a -> t m b -> t m c
zipWithM a -> b -> m c
f t m a
m1 t m b
m2 = Stream m c -> t m c
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(IsStream t, Monad m) =>
Stream m a -> t m a
P.fromStreamS (Stream m c -> t m c) -> Stream m c -> t m c
forall a b. (a -> b) -> a -> b
$ (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c
forall (m :: * -> *) a b c.
Monad m =>
(a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c
S.zipWithM a -> b -> m c
f (t m a -> Stream m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(IsStream t, Monad m) =>
t m a -> Stream m a
P.toStreamS t m a
m1) (t m b -> Stream m b
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(IsStream t, Monad m) =>
t m a -> Stream m a
P.toStreamS t m b
m2)

-- | Zip two streams serially using a pure zipping function.
--
-- @
-- > S.toList $ S.zipWith (+) (S.fromList [1,2,3]) (S.fromList [4,5,6])
-- [5,7,9]
-- @
--
-- @since 0.1.0
{-# INLINABLE zipWith #-}
zipWith :: (IsStream t, Monad m) => (a -> b -> c) -> t m a -> t m b -> t m c
zipWith :: (a -> b -> c) -> t m a -> t m b -> t m c
zipWith a -> b -> c
f t m a
m1 t m b
m2 = Stream m c -> t m c
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(IsStream t, Monad m) =>
Stream m a -> t m a
P.fromStreamS (Stream m c -> t m c) -> Stream m c -> t m c
forall a b. (a -> b) -> a -> b
$ (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
forall (m :: * -> *) a b c.
Monad m =>
(a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
S.zipWith a -> b -> c
f (t m a -> Stream m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(IsStream t, Monad m) =>
t m a -> Stream m a
P.toStreamS t m a
m1) (t m b -> Stream m b
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(IsStream t, Monad m) =>
t m a -> Stream m a
P.toStreamS t m b
m2)

------------------------------------------------------------------------------
-- Parallel Zipping
------------------------------------------------------------------------------

-- | Like 'zipWithM' but zips concurrently i.e. both the streams being zipped
-- are generated concurrently.
--
-- @since 0.4.0
{-# INLINE zipAsyncWithM #-}
zipAsyncWithM :: (IsStream t, MonadAsync m)
    => (a -> b -> m c) -> t m a -> t m b -> t m c
zipAsyncWithM :: (a -> b -> m c) -> t m a -> t m b -> t m c
zipAsyncWithM a -> b -> m c
f t m a
m1 t m b
m2 = Stream m c -> t m c
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(IsStream t, Monad m) =>
Stream m a -> t m a
D.fromStreamD (Stream m c -> t m c) -> Stream m c -> t m c
forall a b. (a -> b) -> a -> b
$
    (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c
forall (m :: * -> *) a b c.
Monad m =>
(a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c
D.zipWithM a -> b -> m c
f (Stream m a -> Stream m a
forall (m :: * -> *) a. MonadAsync m => Stream m a -> Stream m a
D.mkParallelD (Stream m a -> Stream m a) -> Stream m a -> Stream m a
forall a b. (a -> b) -> a -> b
$ t m a -> Stream m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(IsStream t, Monad m) =>
t m a -> Stream m a
D.toStreamD t m a
m1)
                 (Stream m b -> Stream m b
forall (m :: * -> *) a. MonadAsync m => Stream m a -> Stream m a
D.mkParallelD (Stream m b -> Stream m b) -> Stream m b -> Stream m b
forall a b. (a -> b) -> a -> b
$ t m b -> Stream m b
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(IsStream t, Monad m) =>
t m a -> Stream m a
D.toStreamD t m b
m2)

-- | Like 'zipWith' but zips concurrently i.e. both the streams being zipped
-- are generated concurrently.
--
-- @since 0.1.0
{-# INLINE zipAsyncWith #-}
zipAsyncWith :: (IsStream t, MonadAsync m)
    => (a -> b -> c) -> t m a -> t m b -> t m c
zipAsyncWith :: (a -> b -> c) -> t m a -> t m b -> t m c
zipAsyncWith a -> b -> c
f = (a -> b -> m c) -> t m a -> t m b -> t m c
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a b c.
(IsStream t, MonadAsync m) =>
(a -> b -> m c) -> t m a -> t m b -> t m c
zipAsyncWithM (\a
a b
b -> c -> m c
forall (m :: * -> *) a. Monad m => a -> m a
return (a -> b -> c
f a
a b
b))

------------------------------------------------------------------------------
-- Serially Zipping Streams
------------------------------------------------------------------------------

-- | The applicative instance of 'ZipSerialM' zips a number of streams serially
-- i.e. it produces one element from each stream serially and then zips all
-- those elements.
--
-- @
-- main = (toList . 'zipSerially' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
--     where s1 = fromFoldable [1, 2]
--           s2 = fromFoldable [3, 4]
--           s3 = fromFoldable [5, 6]
-- @
-- @
-- [(1,3,5),(2,4,6)]
-- @
--
-- The 'Semigroup' instance of this type works the same way as that of
-- 'SerialT'.
--
-- @since 0.2.0
newtype ZipSerialM m a = ZipSerialM {ZipSerialM m a -> Stream m a
getZipSerialM :: Stream m a}
        deriving (b -> ZipSerialM m a -> ZipSerialM m a
NonEmpty (ZipSerialM m a) -> ZipSerialM m a
ZipSerialM m a -> ZipSerialM m a -> ZipSerialM m a
(ZipSerialM m a -> ZipSerialM m a -> ZipSerialM m a)
-> (NonEmpty (ZipSerialM m a) -> ZipSerialM m a)
-> (forall b. Integral b => b -> ZipSerialM m a -> ZipSerialM m a)
-> Semigroup (ZipSerialM m a)
forall b. Integral b => b -> ZipSerialM m a -> ZipSerialM m a
forall a.
(a -> a -> a)
-> (NonEmpty a -> a)
-> (forall b. Integral b => b -> a -> a)
-> Semigroup a
forall (m :: * -> *) a. NonEmpty (ZipSerialM m a) -> ZipSerialM m a
forall (m :: * -> *) a.
ZipSerialM m a -> ZipSerialM m a -> ZipSerialM m a
forall (m :: * -> *) a b.
Integral b =>
b -> ZipSerialM m a -> ZipSerialM m a
stimes :: b -> ZipSerialM m a -> ZipSerialM m a
$cstimes :: forall (m :: * -> *) a b.
Integral b =>
b -> ZipSerialM m a -> ZipSerialM m a
sconcat :: NonEmpty (ZipSerialM m a) -> ZipSerialM m a
$csconcat :: forall (m :: * -> *) a. NonEmpty (ZipSerialM m a) -> ZipSerialM m a
<> :: ZipSerialM m a -> ZipSerialM m a -> ZipSerialM m a
$c<> :: forall (m :: * -> *) a.
ZipSerialM m a -> ZipSerialM m a -> ZipSerialM m a
Semigroup, Semigroup (ZipSerialM m a)
ZipSerialM m a
Semigroup (ZipSerialM m a)
-> ZipSerialM m a
-> (ZipSerialM m a -> ZipSerialM m a -> ZipSerialM m a)
-> ([ZipSerialM m a] -> ZipSerialM m a)
-> Monoid (ZipSerialM m a)
[ZipSerialM m a] -> ZipSerialM m a
ZipSerialM m a -> ZipSerialM m a -> ZipSerialM m a
forall a.
Semigroup a -> a -> (a -> a -> a) -> ([a] -> a) -> Monoid a
forall (m :: * -> *) a. Semigroup (ZipSerialM m a)
forall (m :: * -> *) a. ZipSerialM m a
forall (m :: * -> *) a. [ZipSerialM m a] -> ZipSerialM m a
forall (m :: * -> *) a.
ZipSerialM m a -> ZipSerialM m a -> ZipSerialM m a
mconcat :: [ZipSerialM m a] -> ZipSerialM m a
$cmconcat :: forall (m :: * -> *) a. [ZipSerialM m a] -> ZipSerialM m a
mappend :: ZipSerialM m a -> ZipSerialM m a -> ZipSerialM m a
$cmappend :: forall (m :: * -> *) a.
ZipSerialM m a -> ZipSerialM m a -> ZipSerialM m a
mempty :: ZipSerialM m a
$cmempty :: forall (m :: * -> *) a. ZipSerialM m a
$cp1Monoid :: forall (m :: * -> *) a. Semigroup (ZipSerialM m a)
Monoid)

-- |
-- @since 0.1.0
{-# DEPRECATED ZipStream "Please use 'ZipSerialM' instead." #-}
type ZipStream = ZipSerialM

-- | An IO stream whose applicative instance zips streams serially.
--
-- @since 0.2.0
type ZipSerial = ZipSerialM IO

-- | Fix the type of a polymorphic stream as 'ZipSerialM'.
--
-- @since 0.2.0
zipSerially :: IsStream t => ZipSerialM m a -> t m a
zipSerially :: ZipSerialM m a -> t m a
zipSerially = ZipSerialM m a -> t m a
forall (t1 :: (* -> *) -> * -> *) (t2 :: (* -> *) -> * -> *)
       (m :: * -> *) a.
(IsStream t1, IsStream t2) =>
t1 m a -> t2 m a
K.adapt

-- | Same as 'zipSerially'.
--
-- @since 0.1.0
{-# DEPRECATED zipping "Please use zipSerially instead." #-}
zipping :: IsStream t => ZipSerialM m a -> t m a
zipping :: ZipSerialM m a -> t m a
zipping = ZipSerialM m a -> t m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
IsStream t =>
ZipSerialM m a -> t m a
zipSerially

consMZip :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
consMZip :: m a -> ZipSerialM m a -> ZipSerialM m a
consMZip m a
m ZipSerialM m a
ms = Stream m a -> ZipSerialM m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
IsStream t =>
Stream m a -> t m a
fromStream (Stream m a -> ZipSerialM m a) -> Stream m a -> ZipSerialM m a
forall a b. (a -> b) -> a -> b
$ m a -> Stream m a -> Stream m a
forall (m :: * -> *) a. Monad m => m a -> Stream m a -> Stream m a
K.consMStream m a
m (ZipSerialM m a -> Stream m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
IsStream t =>
t m a -> Stream m a
toStream ZipSerialM m a
ms)

instance IsStream ZipSerialM where
    toStream :: ZipSerialM m a -> Stream m a
toStream = ZipSerialM m a -> Stream m a
forall (m :: * -> *) a. ZipSerialM m a -> Stream m a
getZipSerialM
    fromStream :: Stream m a -> ZipSerialM m a
fromStream = Stream m a -> ZipSerialM m a
forall (m :: * -> *) a. Stream m a -> ZipSerialM m a
ZipSerialM

    {-# INLINE consM #-}
    {-# SPECIALIZE consM :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
    consM :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
    consM :: m a -> ZipSerialM m a -> ZipSerialM m a
consM = m a -> ZipSerialM m a -> ZipSerialM m a
forall (m :: * -> *) a.
Monad m =>
m a -> ZipSerialM m a -> ZipSerialM m a
consMZip

    {-# INLINE (|:) #-}
    {-# SPECIALIZE (|:) :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
    (|:) :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
    |: :: m a -> ZipSerialM m a -> ZipSerialM m a
(|:) = m a -> ZipSerialM m a -> ZipSerialM m a
forall (m :: * -> *) a.
Monad m =>
m a -> ZipSerialM m a -> ZipSerialM m a
consMZip

LIST_INSTANCES(ZipSerialM)
NFDATA1_INSTANCE(ZipSerialM)

instance Monad m => Functor (ZipSerialM m) where
    {-# INLINE fmap #-}
    fmap :: (a -> b) -> ZipSerialM m a -> ZipSerialM m b
fmap a -> b
f (ZipSerialM Stream m a
m) = Stream m b -> ZipSerialM m b
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(IsStream t, Monad m) =>
Stream m a -> t m a
D.fromStreamD (Stream m b -> ZipSerialM m b) -> Stream m b -> ZipSerialM m b
forall a b. (a -> b) -> a -> b
$ (a -> m b) -> Stream m a -> Stream m b
forall (m :: * -> *) a b.
Monad m =>
(a -> m b) -> Stream m a -> Stream m b
D.mapM (b -> m b
forall (m :: * -> *) a. Monad m => a -> m a
return (b -> m b) -> (a -> b) -> a -> m b
forall b c a. (b -> c) -> (a -> b) -> a -> c
. a -> b
f) (Stream m a -> Stream m b) -> Stream m a -> Stream m b
forall a b. (a -> b) -> a -> b
$ Stream m a -> Stream m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(IsStream t, Monad m) =>
t m a -> Stream m a
D.toStreamD Stream m a
m

instance Monad m => Applicative (ZipSerialM m) where
    pure :: a -> ZipSerialM m a
pure = Stream m a -> ZipSerialM m a
forall (m :: * -> *) a. Stream m a -> ZipSerialM m a
ZipSerialM (Stream m a -> ZipSerialM m a)
-> (a -> Stream m a) -> a -> ZipSerialM m a
forall b c a. (b -> c) -> (a -> b) -> a -> c
. a -> Stream m a
forall (t :: (* -> *) -> * -> *) a (m :: * -> *).
IsStream t =>
a -> t m a
K.repeat
    {-# INLINE (<*>) #-}
    <*> :: ZipSerialM m (a -> b) -> ZipSerialM m a -> ZipSerialM m b
(<*>) = ((a -> b) -> a -> b)
-> ZipSerialM m (a -> b) -> ZipSerialM m a -> ZipSerialM m b
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a b c.
(IsStream t, Monad m) =>
(a -> b -> c) -> t m a -> t m b -> t m c
zipWith (a -> b) -> a -> b
forall a. a -> a
id

FOLDABLE_INSTANCE(ZipSerialM)
TRAVERSABLE_INSTANCE(ZipSerialM)

------------------------------------------------------------------------------
-- Parallely Zipping Streams
------------------------------------------------------------------------------
--
-- | Like 'ZipSerialM' but zips in parallel, it generates all the elements to
-- be zipped concurrently.
--
-- @
-- main = (toList . 'zipAsyncly' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
--     where s1 = fromFoldable [1, 2]
--           s2 = fromFoldable [3, 4]
--           s3 = fromFoldable [5, 6]
-- @
-- @
-- [(1,3,5),(2,4,6)]
-- @
--
-- The 'Semigroup' instance of this type works the same way as that of
-- 'SerialT'.
--
-- @since 0.2.0
newtype ZipAsyncM m a = ZipAsyncM {ZipAsyncM m a -> Stream m a
getZipAsyncM :: Stream m a}
        deriving (b -> ZipAsyncM m a -> ZipAsyncM m a
NonEmpty (ZipAsyncM m a) -> ZipAsyncM m a
ZipAsyncM m a -> ZipAsyncM m a -> ZipAsyncM m a
(ZipAsyncM m a -> ZipAsyncM m a -> ZipAsyncM m a)
-> (NonEmpty (ZipAsyncM m a) -> ZipAsyncM m a)
-> (forall b. Integral b => b -> ZipAsyncM m a -> ZipAsyncM m a)
-> Semigroup (ZipAsyncM m a)
forall b. Integral b => b -> ZipAsyncM m a -> ZipAsyncM m a
forall a.
(a -> a -> a)
-> (NonEmpty a -> a)
-> (forall b. Integral b => b -> a -> a)
-> Semigroup a
forall (m :: * -> *) a. NonEmpty (ZipAsyncM m a) -> ZipAsyncM m a
forall (m :: * -> *) a.
ZipAsyncM m a -> ZipAsyncM m a -> ZipAsyncM m a
forall (m :: * -> *) a b.
Integral b =>
b -> ZipAsyncM m a -> ZipAsyncM m a
stimes :: b -> ZipAsyncM m a -> ZipAsyncM m a
$cstimes :: forall (m :: * -> *) a b.
Integral b =>
b -> ZipAsyncM m a -> ZipAsyncM m a
sconcat :: NonEmpty (ZipAsyncM m a) -> ZipAsyncM m a
$csconcat :: forall (m :: * -> *) a. NonEmpty (ZipAsyncM m a) -> ZipAsyncM m a
<> :: ZipAsyncM m a -> ZipAsyncM m a -> ZipAsyncM m a
$c<> :: forall (m :: * -> *) a.
ZipAsyncM m a -> ZipAsyncM m a -> ZipAsyncM m a
Semigroup, Semigroup (ZipAsyncM m a)
ZipAsyncM m a
Semigroup (ZipAsyncM m a)
-> ZipAsyncM m a
-> (ZipAsyncM m a -> ZipAsyncM m a -> ZipAsyncM m a)
-> ([ZipAsyncM m a] -> ZipAsyncM m a)
-> Monoid (ZipAsyncM m a)
[ZipAsyncM m a] -> ZipAsyncM m a
ZipAsyncM m a -> ZipAsyncM m a -> ZipAsyncM m a
forall a.
Semigroup a -> a -> (a -> a -> a) -> ([a] -> a) -> Monoid a
forall (m :: * -> *) a. Semigroup (ZipAsyncM m a)
forall (m :: * -> *) a. ZipAsyncM m a
forall (m :: * -> *) a. [ZipAsyncM m a] -> ZipAsyncM m a
forall (m :: * -> *) a.
ZipAsyncM m a -> ZipAsyncM m a -> ZipAsyncM m a
mconcat :: [ZipAsyncM m a] -> ZipAsyncM m a
$cmconcat :: forall (m :: * -> *) a. [ZipAsyncM m a] -> ZipAsyncM m a
mappend :: ZipAsyncM m a -> ZipAsyncM m a -> ZipAsyncM m a
$cmappend :: forall (m :: * -> *) a.
ZipAsyncM m a -> ZipAsyncM m a -> ZipAsyncM m a
mempty :: ZipAsyncM m a
$cmempty :: forall (m :: * -> *) a. ZipAsyncM m a
$cp1Monoid :: forall (m :: * -> *) a. Semigroup (ZipAsyncM m a)
Monoid)

-- | An IO stream whose applicative instance zips streams wAsyncly.
--
-- @since 0.2.0
type ZipAsync = ZipAsyncM IO

-- | Fix the type of a polymorphic stream as 'ZipAsyncM'.
--
-- @since 0.2.0
zipAsyncly :: IsStream t => ZipAsyncM m a -> t m a
zipAsyncly :: ZipAsyncM m a -> t m a
zipAsyncly = ZipAsyncM m a -> t m a
forall (t1 :: (* -> *) -> * -> *) (t2 :: (* -> *) -> * -> *)
       (m :: * -> *) a.
(IsStream t1, IsStream t2) =>
t1 m a -> t2 m a
K.adapt

-- | Same as 'zipAsyncly'.
--
-- @since 0.1.0
{-# DEPRECATED zippingAsync "Please use zipAsyncly instead." #-}
zippingAsync :: IsStream t => ZipAsyncM m a -> t m a
zippingAsync :: ZipAsyncM m a -> t m a
zippingAsync = ZipAsyncM m a -> t m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
IsStream t =>
ZipAsyncM m a -> t m a
zipAsyncly

consMZipAsync :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
consMZipAsync :: m a -> ZipAsyncM m a -> ZipAsyncM m a
consMZipAsync m a
m ZipAsyncM m a
ms = Stream m a -> ZipAsyncM m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
IsStream t =>
Stream m a -> t m a
fromStream (Stream m a -> ZipAsyncM m a) -> Stream m a -> ZipAsyncM m a
forall a b. (a -> b) -> a -> b
$ m a -> Stream m a -> Stream m a
forall (m :: * -> *) a. Monad m => m a -> Stream m a -> Stream m a
K.consMStream m a
m (ZipAsyncM m a -> Stream m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
IsStream t =>
t m a -> Stream m a
toStream ZipAsyncM m a
ms)

instance IsStream ZipAsyncM where
    toStream :: ZipAsyncM m a -> Stream m a
toStream = ZipAsyncM m a -> Stream m a
forall (m :: * -> *) a. ZipAsyncM m a -> Stream m a
getZipAsyncM
    fromStream :: Stream m a -> ZipAsyncM m a
fromStream = Stream m a -> ZipAsyncM m a
forall (m :: * -> *) a. Stream m a -> ZipAsyncM m a
ZipAsyncM

    {-# INLINE consM #-}
    {-# SPECIALIZE consM :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
    consM :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
    consM :: m a -> ZipAsyncM m a -> ZipAsyncM m a
consM = m a -> ZipAsyncM m a -> ZipAsyncM m a
forall (m :: * -> *) a.
Monad m =>
m a -> ZipAsyncM m a -> ZipAsyncM m a
consMZipAsync

    {-# INLINE (|:) #-}
    {-# SPECIALIZE (|:) :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
    (|:) :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
    |: :: m a -> ZipAsyncM m a -> ZipAsyncM m a
(|:) = m a -> ZipAsyncM m a -> ZipAsyncM m a
forall (m :: * -> *) a.
Monad m =>
m a -> ZipAsyncM m a -> ZipAsyncM m a
consMZipAsync

instance Monad m => Functor (ZipAsyncM m) where
    {-# INLINE fmap #-}
    fmap :: (a -> b) -> ZipAsyncM m a -> ZipAsyncM m b
fmap a -> b
f (ZipAsyncM Stream m a
m) = Stream m b -> ZipAsyncM m b
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(IsStream t, Monad m) =>
Stream m a -> t m a
D.fromStreamD (Stream m b -> ZipAsyncM m b) -> Stream m b -> ZipAsyncM m b
forall a b. (a -> b) -> a -> b
$ (a -> m b) -> Stream m a -> Stream m b
forall (m :: * -> *) a b.
Monad m =>
(a -> m b) -> Stream m a -> Stream m b
D.mapM (b -> m b
forall (m :: * -> *) a. Monad m => a -> m a
return (b -> m b) -> (a -> b) -> a -> m b
forall b c a. (b -> c) -> (a -> b) -> a -> c
. a -> b
f) (Stream m a -> Stream m b) -> Stream m a -> Stream m b
forall a b. (a -> b) -> a -> b
$ Stream m a -> Stream m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(IsStream t, Monad m) =>
t m a -> Stream m a
D.toStreamD Stream m a
m

instance MonadAsync m => Applicative (ZipAsyncM m) where
    pure :: a -> ZipAsyncM m a
pure = Stream m a -> ZipAsyncM m a
forall (m :: * -> *) a. Stream m a -> ZipAsyncM m a
ZipAsyncM (Stream m a -> ZipAsyncM m a)
-> (a -> Stream m a) -> a -> ZipAsyncM m a
forall b c a. (b -> c) -> (a -> b) -> a -> c
. a -> Stream m a
forall (t :: (* -> *) -> * -> *) a (m :: * -> *).
IsStream t =>
a -> t m a
K.repeat
    {-# INLINE (<*>) #-}
    ZipAsyncM m (a -> b)
m1 <*> :: ZipAsyncM m (a -> b) -> ZipAsyncM m a -> ZipAsyncM m b
<*> ZipAsyncM m a
m2 = ((a -> b) -> a -> b)
-> ZipAsyncM m (a -> b) -> ZipAsyncM m a -> ZipAsyncM m b
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a b c.
(IsStream t, MonadAsync m) =>
(a -> b -> c) -> t m a -> t m b -> t m c
zipAsyncWith (a -> b) -> a -> b
forall a. a -> a
id ZipAsyncM m (a -> b)
m1 ZipAsyncM m a
m2