@@ -21,7 +21,8 @@ module Streamly.Streams
2121 (
2222 IsStream (..)
2323 , Streaming -- deprecated
24− , MonadAsync
24+ , MonadParallel
25+ , S.MonadAsync -- deprecated
2526
2627 -- * SVars
2728 , SVarSched (..)
@@ -48,9 +49,9 @@ module Streamly.Streams
4849 , async
4950
5051 -- * Merging Streams
51− , append
52− , interleave
53− , aparallel
52+ , serial
53+ , coserial
54+ , coparallel
5455 , parallel
5556 , (<=>) --deprecated
5657 , (<|) --deprecated
@@ -58,22 +59,27 @@ module Streamly.Streams
5859 -- * Stream Styles
5960 , SerialT
6061 , StreamT -- deprecated
61− , InterleavedT
62− , AParallelT
62+ , CoserialT
63+ , InterleavedT -- deprecated
64+ , CoparallelT
6365 , AsyncT -- deprecated
6466 , ParallelT
6567 , ZipSerial
6668 , ZipStream -- deprecated
67− , ZipAsync
69+ , ZipParallel
70+ , ZipAsync -- deprecated
6871
6972 -- * Type Adapters
7073 , serially
71− , interleaving
72− , aparallely
74+ , coserially
75+ , interleaving -- deprecated
76+ , coparallely
7377 , asyncly -- deprecated
7478 , parallely
75− , zipping
76− , zippingAsync
79+ , zipSerially
80+ , zipping -- deprecated
81+ , zipParallely
82+ , zippingAsync -- deprecated
7783 , adapt
7884
7985 -- * Running Streams
@@ -86,10 +92,10 @@ module Streamly.Streams
8692
8793 -- * Zipping
8894 , zipWith
89− , zipAsyncWith
95+ , zipParallelWith
96+ , zipAsyncWith -- deprecated
9097
9198 -- * Fold Utilities
92− -- $foldutils
9399 , foldWith
94100 , foldMapWith
95101 , forEachWith
@@ -107,7 +113,7 @@ import Control.Monad.State.Class (MonadState(..))
107113import Control.Monad.Trans.Class (MonadTrans)
108114import Data.Semigroup (Semigroup(..))
109115import Prelude hiding (zipWith)
110−import Streamly.Core ( MonadAsync, Stream(Stream)
116+import Streamly.Core ( MonadParallel, Stream(Stream)
111117 , SVar, SVarStyle(..)
112118 , SVarTag(..), SVarSched(..))
113119import qualified Streamly.Core as S
@@ -199,7 +205,7 @@ fromCallback :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a
199205fromCallback k = fromStream $ Stream $ \_ _ yld -> k (\a -> yld a Nothing)
200206
201207-- | Read an SVar to get a stream.
202−fromSVar :: (MonadAsync m, IsStream t) => SVar m a -> t m a
208+fromSVar :: (MonadParallel m, IsStream t) => SVar m a -> t m a
203209fromSVar sv = fromStream $ S.fromStreamVar sv
204210
205211------------------------------------------------------------------------------
@@ -216,7 +222,9 @@ streamFold sv step blank m =
216222 yield a (Just x) = step a (Just (fromStream x))
217223 in (S.runStream (toStream m)) sv blank yield
218224
219−-- | Run a streaming composition, discard the results.
225+-- | Run a streaming composition, discard the results. By default it interprets
226+-- the stream as 'SerialT', to run other types of streams use the type adapting
227+-- combinators for example @runStream . 'parallely'@.
220228runStream :: Monad m => SerialT m a -> m ()
221229runStream m = go (toStream m)
222230 where
@@ -233,7 +241,7 @@ runStreaming = runStream . adapt
233241
234242-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
235243-- be read back from the SVar using 'fromSVar'.
236−toSVar :: (IsStream t, MonadAsync m) => SVar m a -> t m a -> m ()
244+toSVar :: (IsStream t, MonadParallel m) => SVar m a -> t m a -> m ()
237245toSVar sv m = S.toStreamVar sv (toStream m)
238246
239247------------------------------------------------------------------------------
@@ -247,7 +255,7 @@ toSVar sv m = S.toStreamVar sv (toStream m)
247255-- not drained fully we may have a thread blocked forever and once exhausted it
248256-- will always return 'empty'.
249257
250−async :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
258+async :: (IsStream t, MonadParallel m) => t m a -> m (t m a)
251259async m = do
252260 sv <- S.newStreamVar1 (SVarStyle Disjunction LIFO) (toStream m)
253261 return $ fromSVar sv
@@ -309,8 +317,8 @@ async m = do
309317newtype SerialT m a = SerialT {getSerialT :: Stream m a}
310318 deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)
311319
312−deriving instance MonadAsync m => Alternative (SerialT m)
313−deriving instance MonadAsync m => MonadPlus (SerialT m)
320+deriving instance MonadParallel m => Alternative (SerialT m)
321+deriving instance MonadParallel m => MonadPlus (SerialT m)
314322deriving instance (MonadBase b m, Monad m) => MonadBase b (SerialT m)
315323deriving instance MonadError e m => MonadError e (SerialT m)
316324deriving instance MonadReader r m => MonadReader r (SerialT m)
@@ -336,9 +344,9 @@ type StreamT = SerialT
336344-- | Polymorphic version of the 'Semigroup' operation '<>' of 'SerialT'.
337345-- Appends two streams sequentially, yielding all elements from the first
338346-- stream, and then all elements from the second stream.
339−{-# INLINE append #-}
340−append :: IsStream t => t m a -> t m a -> t m a
341−append m1 m2 = fromStream $ S.append (toStream m1) (toStream m2)
347+{-# INLINE serial #-}
348+serial :: IsStream t => t m a -> t m a -> t m a
349+serial m1 m2 = fromStream $ S.serial (toStream m1) (toStream m2)
342350
343351------------------------------------------------------------------------------
344352-- Monad
@@ -416,10 +424,10 @@ instance (Monad m, Floating a) => Floating (SerialT m a) where
416424 logBase = liftA2 logBase
417425
418426------------------------------------------------------------------------------
419−-- InterleavedT
427+-- CoserialT
420428------------------------------------------------------------------------------
421429
422−-- | The 'Semigroup' instance of 'InterleavedT' interleaves two streams,
430+-- | The 'Semigroup' instance of 'CoserialT' interleaves two streams,
423431-- yielding one element from each stream alternately.
424432--
425433-- @
@@ -449,48 +457,51 @@ instance (Monad m, Floating a) => Floating (SerialT m a) where
449457-- Note that interleaving composition can only combine a finite number of
450458-- streams as it needs to retain state for each unfinished stream.
451459--
452−newtype InterleavedT m a = InterleavedT {getInterleavedT :: Stream m a}
460+newtype CoserialT m a = CoserialT {getCoserialT :: Stream m a}
453461 deriving (Monoid, MonadTrans, MonadIO, MonadThrow)
454462
455−deriving instance MonadAsync m => Alternative (InterleavedT m)
456−deriving instance MonadAsync m => MonadPlus (InterleavedT m)
457−deriving instance (MonadBase b m, Monad m) => MonadBase b (InterleavedT m)
458−deriving instance MonadError e m => MonadError e (InterleavedT m)
459−deriving instance MonadReader r m => MonadReader r (InterleavedT m)
460−deriving instance MonadState s m => MonadState s (InterleavedT m)
463+deriving instance MonadParallel m => Alternative (CoserialT m)
464+deriving instance MonadParallel m => MonadPlus (CoserialT m)
465+deriving instance (MonadBase b m, Monad m) => MonadBase b (CoserialT m)
466+deriving instance MonadError e m => MonadError e (CoserialT m)
467+deriving instance MonadReader r m => MonadReader r (CoserialT m)
468+deriving instance MonadState s m => MonadState s (CoserialT m)
461469
462−instance IsStream InterleavedT where
463− toStream = getInterleavedT
464− fromStream = InterleavedT
470+{-# DEPRECATED InterleavedT "Please use 'CoserialT' instead." #-}
471+type InterleavedT = CoserialT
472+
473+instance IsStream CoserialT where
474+ toStream = getCoserialT
475+ fromStream = CoserialT
465476
466477------------------------------------------------------------------------------
467478-- Semigroup
468479------------------------------------------------------------------------------
469480
470−-- | Polymorphic version of the 'Semigroup' operation '<>' of 'InterleavedT'.
481+-- | Polymorphic version of the 'Semigroup' operation '<>' of 'CoserialT'.
471482-- Interleaves two streams, yielding one element from each stream alternately.
472−{-# INLINE interleave #-}
473−interleave :: IsStream t => t m a -> t m a -> t m a
474−interleave m1 m2 = fromStream $ S.interleave (toStream m1) (toStream m2)
483+{-# INLINE coserial #-}
484+coserial :: IsStream t => t m a -> t m a -> t m a
485+coserial m1 m2 = fromStream $ S.coserial (toStream m1) (toStream m2)
475486
476−instance Semigroup (InterleavedT m a) where
477− (<>) = interleave
487+instance Semigroup (CoserialT m a) where
488+ (<>) = coserial
478489
479490infixr 5 <=>
480491
481−-- | Same as 'interleave'.
482−{-# Deprecated (<=>) "Please use '<>' of InterleavedT or 'interleave' instead." #-}
492+-- | Same as 'coserial'.
493+{-# DEPRECATED (<=>) "Please use 'coserial' instead." #-}
483494{-# INLINE (<=>) #-}
484495(<=>) :: IsStream t => t m a -> t m a -> t m a
485−(<=>) = interleave
496+(<=>) = coserial
486497
487498------------------------------------------------------------------------------
488499-- Monad
489500------------------------------------------------------------------------------
490501
491−instance Monad m => Monad (InterleavedT m) where
502+instance Monad m => Monad (CoserialT m) where
492503 return = pure
493− (InterleavedT (Stream m)) >>= f = InterleavedT $ Stream $ \_ stp yld ->
504+ (CoserialT (Stream m)) >>= f = CoserialT $ Stream $ \_ stp yld ->
494505 let run x = (S.runStream x) Nothing stp yld
495506 yield a Nothing = run $ toStream (f a)
496507 yield a (Just r) = run $ toStream $ f a <> (fromStream r >>= f)
@@ -500,26 +511,26 @@ instance Monad m => Monad (InterleavedT m) where
500511-- Applicative
501512------------------------------------------------------------------------------
502513
503−instance Monad m => Applicative (InterleavedT m) where
504− pure a = InterleavedT $ S.cons a Nothing
514+instance Monad m => Applicative (CoserialT m) where
515+ pure a = CoserialT $ S.cons a Nothing
505516 (<*>) = ap
506517
507518------------------------------------------------------------------------------
508519-- Functor
509520------------------------------------------------------------------------------
510521
511−instance Monad m => Functor (InterleavedT m) where
512− fmap f (InterleavedT (Stream m)) = InterleavedT $ Stream $ \_ stp yld ->
522+instance Monad m => Functor (CoserialT m) where
523+ fmap f (CoserialT (Stream m)) = CoserialT $ Stream $ \_ stp yld ->
513524 let yield a Nothing = yld (f a) Nothing
514525 yield a (Just r) =
515− yld (f a) (Just (getInterleavedT (fmap f (InterleavedT r))))
526+ yld (f a) (Just (getCoserialT (fmap f (CoserialT r))))
516527 in m Nothing stp yield
517528
518529------------------------------------------------------------------------------
519530-- Num
520531------------------------------------------------------------------------------
521532
522−instance (Monad m, Num a) => Num (InterleavedT m a) where
533+instance (Monad m, Num a) => Num (CoserialT m a) where
523534 fromInteger n = pure (fromInteger n)
524535
525536 negate = fmap negate
@@ -530,14 +541,14 @@ instance (Monad m, Num a) => Num (InterleavedT m a) where
530541 (*) = liftA2 (*)
531542 (-) = liftA2 (-)
532543
533−instance (Monad m, Fractional a) => Fractional (InterleavedT m a) where
544+instance (Monad m, Fractional a) => Fractional (CoserialT m a) where
534545 fromRational n = pure (fromRational n)
535546
536547 recip = fmap recip
537548
538549 (/) = liftA2 (/)
539550
540−instance (Monad m, Floating a) => Floating (InterleavedT m a) where
551+instance (Monad m, Floating a) => Floating (CoserialT m a) where
541552 pi = pure pi
542553
543554 exp = fmap exp
@@ -560,7 +571,7 @@ instance (Monad m, Floating a) => Floating (InterleavedT m a) where
560571 logBase = liftA2 logBase
561572
562573------------------------------------------------------------------------------
563−-- AParallelT
574+-- CoparallelT
564575------------------------------------------------------------------------------
565576
566577-- | Adaptive parallel, in a left to right 'Semigroup' composition it tries to
@@ -570,7 +581,7 @@ instance (Monad m, Floating a) => Floating (InterleavedT m a) where
570581-- enough for the consumer.
571582--
572583-- @
573−-- main = ('toList' . 'aparallely' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
584+-- main = ('toList' . 'coparallely' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
574585-- @
575586-- @
576587-- [1,2,3,4]
@@ -585,7 +596,7 @@ instance (Monad m, Floating a) => Floating (InterleavedT m a) where
585596-- import "Streamly"
586597-- import Control.Concurrent
587598--
588−-- main = 'runStream' . 'aparallely' $ do
599+-- main = 'runStream' . 'coparallely' $ do
589600-- n <- return 3 \<\> return 2 \<\> return 1
590601-- liftIO $ do
591602-- threadDelay (n * 1000000)
@@ -602,45 +613,45 @@ instance (Monad m, Floating a) => Floating (InterleavedT m a) where
602613-- Note that this composition can be used to combine infinite number of streams
603614-- as it explores only a bounded number of streams at a time.
604615--
605−newtype AParallelT m a = AParallelT {getAParallelT :: Stream m a}
616+newtype CoparallelT m a = CoparallelT {getCoparallelT :: Stream m a}
606617 deriving (MonadTrans)
607618
608−deriving instance MonadAsync m => Monoid (AParallelT m a)
609−deriving instance MonadAsync m => Alternative (AParallelT m)
610−deriving instance MonadAsync m => MonadPlus (AParallelT m)
611−deriving instance MonadAsync m => MonadIO (AParallelT m)
612−deriving instance MonadAsync m => MonadThrow (AParallelT m)
613−deriving instance (MonadBase b m, MonadAsync m) => MonadBase b (AParallelT m)
614−deriving instance (MonadError e m, MonadAsync m) => MonadError e (AParallelT m)
615−deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (AParallelT m)
616−deriving instance (MonadState s m, MonadAsync m) => MonadState s (AParallelT m)
619+deriving instance MonadParallel m => Monoid (CoparallelT m a)
620+deriving instance MonadParallel m => Alternative (CoparallelT m)
621+deriving instance MonadParallel m => MonadPlus (CoparallelT m)
622+deriving instance MonadParallel m => MonadIO (CoparallelT m)
623+deriving instance MonadParallel m => MonadThrow (CoparallelT m)
624+deriving instance (MonadBase b m, MonadParallel m) => MonadBase b (CoparallelT m)
625+deriving instance (MonadError e m, MonadParallel m) => MonadError e (CoparallelT m)
626+deriving instance (MonadReader r m, MonadParallel m) => MonadReader r (CoparallelT m)
627+deriving instance (MonadState s m, MonadParallel m) => MonadState s (CoparallelT m)
617628
618−{-# DEPRECATED AsyncT "Please use 'AParallelT' instead." #-}
619−type AsyncT = AParallelT
629+{-# DEPRECATED AsyncT "Please use 'CoparallelT' instead." #-}
630+type AsyncT = CoparallelT
620631
621−instance IsStream AParallelT where
622− toStream = getAParallelT
623− fromStream = AParallelT
632+instance IsStream CoparallelT where
633+ toStream = getCoparallelT
634+ fromStream = CoparallelT
624635
625636------------------------------------------------------------------------------
626637-- Semigroup
627638------------------------------------------------------------------------------
628639
629−-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AParallelT', but
640+-- | Polymorphic version of the 'Semigroup' operation '<>' of 'CoparallelT', but
630641-- polymorphic. Merges two streams possibly concurrently, preferring the
631642-- elements from the left one when available.
632−{-# INLINE aparallel #-}
633−aparallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
634−aparallel m1 m2 = fromStream $ S.aparallel (toStream m1) (toStream m2)
643+{-# INLINE coparallel #-}
644+coparallel :: (IsStream t, MonadParallel m) => t m a -> t m a -> t m a
645+coparallel m1 m2 = fromStream $ S.coparallel (toStream m1) (toStream m2)
635646
636−instance MonadAsync m => Semigroup (AParallelT m a) where
637− (<>) = aparallel
647+instance MonadParallel m => Semigroup (CoparallelT m a) where
648+ (<>) = coparallel
638649
639−-- | Same as 'aparallel'.
640−{-# DEPRECATED (<|) "Please use 'aparallel' instead." #-}
650+-- | Same as 'coparallel'.
651+{-# DEPRECATED (<|) "Please use 'coparallel' instead." #-}
641652{-# INLINE (<|) #-}
642−(<|) :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
643−(<|) = aparallel
653+(<|) :: (IsStream t, MonadParallel m) => t m a -> t m a -> t m a
654+(<|) = coparallel
644655
645656------------------------------------------------------------------------------
646657-- Monad
@@ -661,35 +672,35 @@ parbind par m f = go m
661672 yield a (Just r) = run $ f a `par` go r
662673 in g Nothing stp yield
663674
664−instance MonadAsync m => Monad (AParallelT m) where
675+instance MonadParallel m => Monad (CoparallelT m) where
665676 return = pure
666− (AParallelT m) >>= f = AParallelT $ parbind par m g
667− where g x = getAParallelT (f x)
677+ (CoparallelT m) >>= f = CoparallelT $ parbind par m g
678+ where g x = getCoparallelT (f x)
668679 par = S.joinStreamVar2 (SVarStyle Conjunction LIFO)
669680
670681------------------------------------------------------------------------------
671682-- Applicative
672683------------------------------------------------------------------------------
673684
674−instance MonadAsync m => Applicative (AParallelT m) where
675− pure a = AParallelT $ S.cons a Nothing
685+instance MonadParallel m => Applicative (CoparallelT m) where
686+ pure a = CoparallelT $ S.cons a Nothing
676687 (<*>) = ap
677688
678689------------------------------------------------------------------------------
679690-- Functor
680691------------------------------------------------------------------------------
681692
682−instance Monad m => Functor (AParallelT m) where
683− fmap f (AParallelT (Stream m)) = AParallelT $ Stream $ \_ stp yld ->
693+instance Monad m => Functor (CoparallelT m) where
694+ fmap f (CoparallelT (Stream m)) = CoparallelT $ Stream $ \_ stp yld ->
684695 let yield a Nothing = yld (f a) Nothing
685− yield a (Just r) = yld (f a) (Just (getAParallelT (fmap f (AParallelT r))))
696+ yield a (Just r) = yld (f a) (Just (getCoparallelT (fmap f (CoparallelT r))))
686697 in m Nothing stp yield
687698
688699------------------------------------------------------------------------------
689700-- Num
690701------------------------------------------------------------------------------
691702
692−instance (MonadAsync m, Num a) => Num (AParallelT m a) where
703+instance (MonadParallel m, Num a) => Num (CoparallelT m a) where
693704 fromInteger n = pure (fromInteger n)
694705
695706 negate = fmap negate
@@ -700,14 +711,14 @@ instance (MonadAsync m, Num a) => Num (AParallelT m a) where
700711 (*) = liftA2 (*)
701712 (-) = liftA2 (-)
702713
703−instance (MonadAsync m, Fractional a) => Fractional (AParallelT m a) where
714+instance (MonadParallel m, Fractional a) => Fractional (CoparallelT m a) where
704715 fromRational n = pure (fromRational n)
705716
706717 recip = fmap recip
707718
708719 (/) = liftA2 (/)
709720
710−instance (MonadAsync m, Floating a) => Floating (AParallelT m a) where
721+instance (MonadParallel m, Floating a) => Floating (CoparallelT m a) where
711722 pi = pure pi
712723
713724 exp = fmap exp
@@ -764,7 +775,7 @@ instance (MonadAsync m, Floating a) => Floating (AParallelT m a) where
764775-- ThreadId 38: Delay 3
765776-- @
766777--
767−-- Unlike 'AParallelT' all iterations are guaranteed to run fairly
778+-- Unlike 'CoparallelT' all iterations are guaranteed to run fairly
768779-- concurrently, unconditionally.
769780--
770781-- Note that round robin composition can only combine a finite number of
@@ -773,15 +784,15 @@ instance (MonadAsync m, Floating a) => Floating (AParallelT m a) where
773784newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
774785 deriving (MonadTrans)
775786
776−deriving instance MonadAsync m => Monoid (ParallelT m a)
777−deriving instance MonadAsync m => Alternative (ParallelT m)
778−deriving instance MonadAsync m => MonadPlus (ParallelT m)
779−deriving instance MonadAsync m => MonadIO (ParallelT m)
780−deriving instance MonadAsync m => MonadThrow (ParallelT m)
781−deriving instance (MonadBase b m, MonadAsync m) => MonadBase b (ParallelT m)
782−deriving instance (MonadError e m, MonadAsync m) => MonadError e (ParallelT m)
783−deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (ParallelT m)
784−deriving instance (MonadState s m, MonadAsync m) => MonadState s (ParallelT m)
787+deriving instance MonadParallel m => Monoid (ParallelT m a)
788+deriving instance MonadParallel m => Alternative (ParallelT m)
789+deriving instance MonadParallel m => MonadPlus (ParallelT m)
790+deriving instance MonadParallel m => MonadIO (ParallelT m)
791+deriving instance MonadParallel m => MonadThrow (ParallelT m)
792+deriving instance (MonadBase b m, MonadParallel m) => MonadBase b (ParallelT m)
793+deriving instance (MonadError e m, MonadParallel m) => MonadError e (ParallelT m)
794+deriving instance (MonadReader r m, MonadParallel m) => MonadReader r (ParallelT m)
795+deriving instance (MonadState s m, MonadParallel m) => MonadState s (ParallelT m)
785796
786797instance IsStream ParallelT where
787798 toStream = getParallelT
@@ -794,17 +805,17 @@ instance IsStream ParallelT where
794805-- | Polymorphic version of the 'Semigroup' operation '<>' of 'ParallelT'.
795806-- Merges two streams concurrently choosing elements from both fairly.
796807{-# INLINE parallel #-}
797−parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
808+parallel :: (IsStream t, MonadParallel m) => t m a -> t m a -> t m a
798809parallel m1 m2 = fromStream $ S.parallel (toStream m1) (toStream m2)
799810
800−instance MonadAsync m => Semigroup (ParallelT m a) where
811+instance MonadParallel m => Semigroup (ParallelT m a) where
801812 (<>) = parallel
802813
803814------------------------------------------------------------------------------
804815-- Monad
805816------------------------------------------------------------------------------
806817
807−instance MonadAsync m => Monad (ParallelT m) where
818+instance MonadParallel m => Monad (ParallelT m) where
808819 return = pure
809820 (ParallelT m) >>= f = ParallelT $ parbind par m g
810821 where g x = getParallelT (f x)
@@ -814,7 +825,7 @@ instance MonadAsync m => Monad (ParallelT m) where
814825-- Applicative
815826------------------------------------------------------------------------------
816827
817−instance MonadAsync m => Applicative (ParallelT m) where
828+instance MonadParallel m => Applicative (ParallelT m) where
818829 pure a = ParallelT $ S.cons a Nothing
819830 (<*>) = ap
820831
@@ -833,7 +844,7 @@ instance Monad m => Functor (ParallelT m) where
833844-- Num
834845------------------------------------------------------------------------------
835846
836−instance (MonadAsync m, Num a) => Num (ParallelT m a) where
847+instance (MonadParallel m, Num a) => Num (ParallelT m a) where
837848 fromInteger n = pure (fromInteger n)
838849
839850 negate = fmap negate
@@ -844,14 +855,14 @@ instance (MonadAsync m, Num a) => Num (ParallelT m a) where
844855 (*) = liftA2 (*)
845856 (-) = liftA2 (-)
846857
847−instance (MonadAsync m, Fractional a) => Fractional (ParallelT m a) where
858+instance (MonadParallel m, Fractional a) => Fractional (ParallelT m a) where
848859 fromRational n = pure (fromRational n)
849860
850861 recip = fmap recip
851862
852863 (/) = liftA2 (/)
853864
854−instance (MonadAsync m, Floating a) => Floating (ParallelT m a) where
865+instance (MonadParallel m, Floating a) => Floating (ParallelT m a) where
855866 pi = pure pi
856867
857868 exp = fmap exp
@@ -892,11 +903,10 @@ zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
892903
893904-- | The applicative instance of 'ZipSerial' zips a number of streams serially
894905-- i.e. it produces one element from each stream serially and then zips all
895−-- those elements. Note, for convenience we have used the 'zipping' combinator
896−-- in the following example instead of using a 'ZipSerial' type annotation.
906+-- those elements.
897907--
898908-- @
899−-- main = (toList . 'zipping' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
909+-- main = (toList . 'zipSerially' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
900910-- where s1 = fromFoldable [1, 2]
901911-- s2 = fromFoldable [3, 4]
902912-- s3 = fromFoldable [5, 6]
@@ -915,7 +925,7 @@ newtype ZipSerial m a = ZipSerial {getZipSerial :: Stream m a}
915925-- | Same as ZipSerial.
916926type ZipStream = ZipSerial
917927
918−deriving instance MonadAsync m => Alternative (ZipSerial m)
928+deriving instance MonadParallel m => Alternative (ZipSerial m)
919929
920930instance Monad m => Functor (ZipSerial m) where
921931 fmap f (ZipSerial (Stream m)) = ZipSerial $ Stream $ \_ stp yld ->
@@ -976,20 +986,25 @@ instance (Monad m, Floating a) => Floating (ZipSerial m a) where
976986-- Parallely Zipping Streams
977987------------------------------------------------------------------------------
978988
979−-- | Zip two streams asyncly (i.e. both the elements being zipped are generated
980−-- concurrently) using a pure zipping function.
981−zipAsyncWith :: (IsStream t, MonadAsync m)
989+-- | Zip two streams concurrently (i.e. both the elements being zipped are
990+-- generated concurrently) using a pure zipping function.
991+zipParallelWith :: (IsStream t, MonadParallel m)
982992 => (a -> b -> c) -> t m a -> t m b -> t m c
983−zipAsyncWith f m1 m2 = fromStream $ Stream $ \_ stp yld -> do
993+zipParallelWith f m1 m2 = fromStream $ Stream $ \_ stp yld -> do
984994 ma <- async m1
985995 mb <- async m2
986996 (S.runStream (toStream (zipWith f ma mb))) Nothing stp yld
987997
998+{-# DEPRECATED zipAsyncWith "Please use zipParallelWith instead." #-}
999+zipAsyncWith :: (IsStream t, MonadParallel m)
1000+ => (a -> b -> c) -> t m a -> t m b -> t m c
1001+zipAsyncWith = zipParallelWith
1002+
9881003-- | Like 'ZipSerial' but zips in parallel, it generates all the elements to
9891004-- be zipped concurrently.
9901005--
9911006-- @
992−-- main = (toList . 'zippingAsync' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
1007+-- main = (toList . 'zipParallely' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
9931008-- where s1 = fromFoldable [1, 2]
9941009-- s2 = fromFoldable [3, 4]
9951010-- s3 = fromFoldable [5, 6]
@@ -1001,27 +1016,30 @@ zipAsyncWith f m1 m2 = fromStream $ Stream $ \_ stp yld -> do
10011016-- The 'Semigroup' instance of this type works the same way as that of
10021017-- 'SerialT'.
10031018--
1004−newtype ZipAsync m a = ZipAsync {getZipAsync :: Stream m a}
1019+newtype ZipParallel m a = ZipParallel {getZipParallel :: Stream m a}
10051020 deriving (Semigroup, Monoid)
10061021
1007−deriving instance MonadAsync m => Alternative (ZipAsync m)
1022+{-# DEPRECATED ZipAsync "Please use ZipParallel instead." #-}
1023+type ZipAsync = ZipParallel
1024+
1025+deriving instance MonadParallel m => Alternative (ZipParallel m)
10081026
1009−instance Monad m => Functor (ZipAsync m) where
1010− fmap f (ZipAsync (Stream m)) = ZipAsync $ Stream $ \_ stp yld ->
1027+instance Monad m => Functor (ZipParallel m) where
1028+ fmap f (ZipParallel (Stream m)) = ZipParallel $ Stream $ \_ stp yld ->
10111029 let yield a Nothing = yld (f a) Nothing
10121030 yield a (Just r) = yld (f a)
1013− (Just (getZipAsync (fmap f (ZipAsync r))))
1031+ (Just (getZipParallel (fmap f (ZipParallel r))))
10141032 in m Nothing stp yield
10151033
1016−instance MonadAsync m => Applicative (ZipAsync m) where
1017− pure = ZipAsync . S.repeat
1018− (<*>) = zipAsyncWith id
1034+instance MonadParallel m => Applicative (ZipParallel m) where
1035+ pure = ZipParallel . S.repeat
1036+ (<*>) = zipParallelWith id
10191037
1020−instance IsStream ZipAsync where
1021− toStream = getZipAsync
1022− fromStream = ZipAsync
1038+instance IsStream ZipParallel where
1039+ toStream = getZipParallel
1040+ fromStream = ZipParallel
10231041
1024−instance (MonadAsync m, Num a) => Num (ZipAsync m a) where
1042+instance (MonadParallel m, Num a) => Num (ZipParallel m a) where
10251043 fromInteger n = pure (fromInteger n)
10261044
10271045 negate = fmap negate
@@ -1032,14 +1050,14 @@ instance (MonadAsync m, Num a) => Num (ZipAsync m a) where
10321050 (*) = liftA2 (*)
10331051 (-) = liftA2 (-)
10341052
1035−instance (MonadAsync m, Fractional a) => Fractional (ZipAsync m a) where
1053+instance (MonadParallel m, Fractional a) => Fractional (ZipParallel m a) where
10361054 fromRational n = pure (fromRational n)
10371055
10381056 recip = fmap recip
10391057
10401058 (/) = liftA2 (/)
10411059
1042−instance (MonadAsync m, Floating a) => Floating (ZipAsync m a) where
1060+instance (MonadParallel m, Floating a) => Floating (ZipParallel m a) where
10431061 pi = pure pi
10441062
10451063 exp = fmap exp
@@ -1073,30 +1091,45 @@ adapt = fromStream . toStream
10731091serially :: IsStream t => SerialT m a -> t m a
10741092serially = adapt
10751093
1076−-- | Fix the type of a polymorphic stream as 'InterleavedT'.
1077−interleaving :: IsStream t => InterleavedT m a -> t m a
1078−interleaving = adapt
1094+-- | Fix the type of a polymorphic stream as 'CoserialT'.
1095+coserially :: IsStream t => CoserialT m a -> t m a
1096+coserially = adapt
10791097
1080−-- | Fix the type of a polymorphic stream as 'AParallelT'.
1081−aparallely :: IsStream t => AParallelT m a -> t m a
1082−aparallely = adapt
1098+-- | Same as 'coserially'.
1099+{-# DEPRECATED interleaving "Please use coserially instead." #-}
1100+interleaving :: IsStream t => CoserialT m a -> t m a
1101+interleaving = coserially
10831102
1084−-- | Same as 'aparallely'.
1085−{-# DEPRECATED asyncly "Please use aparallely instead." #-}
1086−asyncly :: IsStream t => AParallelT m a -> t m a
1087−asyncly = aparallely
1103+-- | Fix the type of a polymorphic stream as 'CoparallelT'.
1104+coparallely :: IsStream t => CoparallelT m a -> t m a
1105+coparallely = adapt
1106+
1107+-- | Same as 'coparallely'.
1108+{-# DEPRECATED asyncly "Please use coparallely instead." #-}
1109+asyncly :: IsStream t => CoparallelT m a -> t m a
1110+asyncly = coparallely
10881111
10891112-- | Fix the type of a polymorphic stream as 'ParallelT'.
10901113parallely :: IsStream t => ParallelT m a -> t m a
10911114parallely = adapt
10921115
10931116-- | Fix the type of a polymorphic stream as 'ZipSerial'.
1117+zipSerially :: IsStream t => ZipSerial m a -> t m a
1118+zipSerially = adapt
1119+
1120+-- | Same as 'zipParallely'.
1121+{-# DEPRECATED zipping "Please use zipSerially instead." #-}
10941122zipping :: IsStream t => ZipSerial m a -> t m a
1095−zipping = adapt
1123+zipping = zipSerially
10961124
1097−-- | Fix the type of a polymorphic stream as 'ZipAsync'.
1098−zippingAsync :: IsStream t => ZipAsync m a -> t m a
1099−zippingAsync = adapt
1125+-- | Fix the type of a polymorphic stream as 'ZipParallel'.
1126+zipParallely :: IsStream t => ZipParallel m a -> t m a
1127+zipParallely = adapt
1128+
1129+-- | Same as 'zipParallely'.
1130+{-# DEPRECATED zippingAsync "Please use zipParallely instead." #-}
1131+zippingAsync :: IsStream t => ZipParallel m a -> t m a
1132+zippingAsync = zipParallely
11001133
11011134-------------------------------------------------------------------------------
11021135-- Running Streams, convenience functions specialized to types
@@ -1107,15 +1140,15 @@ zippingAsync = adapt
11071140runStreamT :: Monad m => SerialT m a -> m ()
11081141runStreamT = runStream
11091142
1110−-- | Same as @runStream . interleaving@.
1143+-- | Same as @runStream . coserially@.
11111144{-# DEPRECATED runInterleavedT "Please use 'runStream . interleaving' instead." #-}
11121145runInterleavedT :: Monad m => InterleavedT m a -> m ()
1113−runInterleavedT = runStream . interleaving
1146+runInterleavedT = runStream . coserially
11141147
1115−-- | Same as @runStream . aparallely@.
1116−{-# DEPRECATED runAsyncT "Please use 'runStream . aparallely' instead." #-}
1117−runAsyncT :: Monad m => AParallelT m a -> m ()
1118−runAsyncT = runStream . aparallely
1148+-- | Same as @runStream . coparallely@.
1149+{-# DEPRECATED runAsyncT "Please use 'runStream . coparallely' instead." #-}
1150+runAsyncT :: Monad m => CoparallelT m a -> m ()
1151+runAsyncT = runStream . coparallely
11191152
11201153-- | Same as @runStream . parallely@.
11211154{-# DEPRECATED runParallelT "Please use 'runStream . parallely' instead." #-}
@@ -1123,38 +1156,33 @@ runParallelT :: Monad m => ParallelT m a -> m ()
11231156runParallelT = runStream . parallely
11241157
11251158-- | Same as @runStream . zipping@.
1126−{-# DEPRECATED runZipStream "Please use 'runStream . zipping instead." #-}
1159+{-# DEPRECATED runZipStream "Please use 'runStream . zipSerially instead." #-}
11271160runZipStream :: Monad m => ZipSerial m a -> m ()
1128−runZipStream = runStream . zipping
1161+runZipStream = runStream . zipSerially
11291162
11301163-- | Same as @runStream . zippingAsync@.
1131−{-# DEPRECATED runZipAsync "Please use 'runStream . zippingAsync instead." #-}
1132−runZipAsync :: Monad m => ZipAsync m a -> m ()
1133−runZipAsync = runStream . zippingAsync
1164+{-# DEPRECATED runZipAsync "Please use 'runStream . zipParallely instead." #-}
1165+runZipAsync :: Monad m => ZipParallel m a -> m ()
1166+runZipAsync = runStream . zipParallely
11341167
11351168------------------------------------------------------------------------------
11361169-- Fold Utilities
11371170------------------------------------------------------------------------------
11381171
1139−-- $foldutils
1140−-- These utilities are designed to pass the first argument as one of the sum
1141−-- style composition operators (i.e. '<>', '<=>', '<|', '<|>') to conveniently
1142−-- fold a container using any style of stream composition.
1143−
1144−-- | Like the 'Prelude' 'fold' but allows you to specify a binary sum style
1145−-- stream composition operator to fold a container of streams.
1172+-- | A variant of 'Data.Foldable.fold' that allows you to fold a 'Foldable'
1173+-- container of streams using the specified stream sum operation.
11461174--
1147−-- @foldWith (<>) $ map return [1..3]@
1175+-- @foldWith 'parallel' $ map return [1..3]@
11481176{-# INLINABLE foldWith #-}
11491177foldWith :: (IsStream t, Foldable f)
11501178 => (t m a -> t m a -> t m a) -> f (t m a) -> t m a
11511179foldWith f = foldr f nil
11521180
1153−-- | Like 'foldMap' but allows you to specify a binary sum style composition
1154−-- operator to fold a container of streams. Maps a monadic streaming action on
1155−-- the container before folding it.
1181+-- | A variant of 'foldMap' that allows you to map a monadic streaming action
1182+-- on a 'Foldable' container and then fold it using the specified stream sum
1183+-- operation.
11561184--
1157−-- @foldMapWith (<>) return [1..3]@
1185+-- @foldMapWith 'parallel' return [1..3]@
11581186{-# INLINABLE foldMapWith #-}
11591187foldMapWith :: (IsStream t, Foldable f)
11601188 => (t m b -> t m b -> t m b) -> (a -> t m b) -> f a -> t m b