@@ -1,3 +1,4 @@
1+{-# LANGUAGE CPP #-}
12{-# LANGUAGE ConstraintKinds #-}
23{-# LANGUAGE FlexibleContexts #-}
34{-# LANGUAGE FlexibleInstances #-}
@@ -102,15 +103,14 @@ module Streamly.Streams
102103 )
103104where
104105
105−import Control.Applicative (Alternative (..), liftA2)
106−import Control.Monad (MonadPlus(..), ap)
107−import Control.Monad.Base (MonadBase (..))
108−import Control.Monad.Catch (MonadThrow)
109−import Control.Monad.Error.Class (MonadError(..))
106+import Control.Monad (ap)
107+import Control.Monad.Base (MonadBase (..), liftBaseDefault)
108+import Control.Monad.Catch (MonadThrow, throwM)
109+-- import Control.Monad.Error.Class (MonadError(..))
110110import Control.Monad.IO.Class (MonadIO(..))
111111import Control.Monad.Reader.Class (MonadReader(..))
112112import Control.Monad.State.Class (MonadState(..))
113−import Control.Monad.Trans.Class (MonadTrans)
113+import Control.Monad.Trans.Class (MonadTrans (lift))
114114import Data.Semigroup (Semigroup(..))
115115import Prelude hiding (zipWith)
116116import Streamly.Core ( MonadParallel, Stream(Stream)
@@ -266,6 +266,46 @@ async m = do
266266 sv <- S.newStreamVar1 (SVarStyle Disjunction LIFO) (toStream m)
267267 return $ fromSVar sv
268268
269+------------------------------------------------------------------------------
270+-- CPP macros for common instances
271+------------------------------------------------------------------------------
272+
273+-- XXX use template haskell instead and include Monoid and IsStream instances
274+-- as well.
275+
276+#define MONADPARALLEL , MonadParallel m
277+
278+#define MONAD_APPLICATIVE_INSTANCE(STREAM,CONSTRAINT) \
279+instance (Monad m CONSTRAINT) => Applicative (STREAM m) where { \
280+ pure = STREAM . singleton; \
281+ (<*>) = ap }
282+
283+#define MONAD_COMMON_INSTANCES(STREAM,CONSTRAINT) \
284+instance (MonadBase b m, Monad m CONSTRAINT) => MonadBase b (STREAM m) where {\
285+ liftBase = liftBaseDefault }; \
286+ \
287+instance (MonadIO m CONSTRAINT) => MonadIO (STREAM m) where { \
288+ liftIO = lift . liftIO }; \
289+ \
290+instance (MonadThrow m CONSTRAINT) => MonadThrow (STREAM m) where { \
291+ throwM = lift . throwM }; \
292+ \
293+{- \
294+instance (MonadError e m CONSTRAINT) => MonadError e (STREAM m) where { \
295+ throwError = lift . throwError; \
296+ catchError m h = \
297+ fromStream $ S.withCatchError (toStream m) (\e -> toStream $ h e) }; \
298+-} \
299+ \
300+instance (MonadReader r m CONSTRAINT) => MonadReader r (STREAM m) where { \
301+ ask = lift ask; \
302+ local f m = fromStream $ S.withLocal f (toStream m) }; \
303+ \
304+instance (MonadState s m CONSTRAINT) => MonadState s (STREAM m) where { \
305+ get = lift get; \
306+ put x = lift (put x); \
307+ state k = lift (state k) }
308+
269309------------------------------------------------------------------------------
270310-- SerialT
271311------------------------------------------------------------------------------
@@ -321,14 +361,7 @@ async m = do
321361-- streams as it explores only one stream at a time.
322362--
323363newtype SerialT m a = SerialT {getSerialT :: Stream m a}
324− deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)
325−
326−deriving instance MonadParallel m => Alternative (SerialT m)
327−deriving instance MonadParallel m => MonadPlus (SerialT m)
328−deriving instance (MonadBase b m, Monad m) => MonadBase b (SerialT m)
329−deriving instance MonadError e m => MonadError e (SerialT m)
330−deriving instance MonadReader r m => MonadReader r (SerialT m)
331−deriving instance MonadState s m => MonadState s (SerialT m)
364+ deriving (Semigroup, Monoid, Functor, MonadTrans)
332365
333366instance IsStream SerialT where
334367 toStream = getSerialT
@@ -338,11 +371,6 @@ instance IsStream SerialT where
338371{-# Deprecated StreamT "Please use SerialT instead." #-}
339372type StreamT = SerialT
340373
341−-- XXX The Functor/Applicative/Num instances for all the types are exactly the
342−-- same, how can we reduce this boilerplate (use TH)? We cannot derive them
343−-- from a single base type because they depend on the Monad instance which is
344−-- different for each type.
345−
346374------------------------------------------------------------------------------
347375-- Semigroup
348376------------------------------------------------------------------------------
@@ -367,66 +395,11 @@ instance Monad m => Monad (SerialT m) where
367395 in m Nothing stp single yield
368396
369397------------------------------------------------------------------------------
370−-- Applicative
371−------------------------------------------------------------------------------
372−
373−instance Monad m => Applicative (SerialT m) where
374− pure = SerialT . singleton
375− (<*>) = ap
376−
377−------------------------------------------------------------------------------
378−-- Functor
379−------------------------------------------------------------------------------
380−
381−instance Monad m => Functor (SerialT m) where
382− fmap f m = fromStream $ Stream $ \_ stp sng yld ->
383− let single = sng . f
384− yield a r = yld (f a) (toStream $ fmap f (SerialT r))
385− in (S.runStream (toStream m)) Nothing stp single yield
386−
387−------------------------------------------------------------------------------
388−-- Num
398+-- Other instances
389399------------------------------------------------------------------------------
390400
391−instance (Monad m, Num a) => Num (SerialT m a) where
392− fromInteger n = pure (fromInteger n)
393−
394− negate = fmap negate
395− abs = fmap abs
396− signum = fmap signum
397−
398− (+) = liftA2 (+)
399− (*) = liftA2 (*)
400− (-) = liftA2 (-)
401−
402−instance (Monad m, Fractional a) => Fractional (SerialT m a) where
403− fromRational n = pure (fromRational n)
404−
405− recip = fmap recip
406−
407− (/) = liftA2 (/)
408−
409−instance (Monad m, Floating a) => Floating (SerialT m a) where
410− pi = pure pi
411−
412− exp = fmap exp
413− sqrt = fmap sqrt
414− log = fmap log
415− sin = fmap sin
416− tan = fmap tan
417− cos = fmap cos
418− asin = fmap asin
419− atan = fmap atan
420− acos = fmap acos
421− sinh = fmap sinh
422− tanh = fmap tanh
423− cosh = fmap cosh
424− asinh = fmap asinh
425− atanh = fmap atanh
426− acosh = fmap acosh
427−
428− (**) = liftA2 (**)
429− logBase = liftA2 logBase
401+MONAD_APPLICATIVE_INSTANCE(SerialT,)
402+MONAD_COMMON_INSTANCES(SerialT,)
430403
431404------------------------------------------------------------------------------
432405-- CoserialT
@@ -463,14 +436,7 @@ instance (Monad m, Floating a) => Floating (SerialT m a) where
463436-- streams as it needs to retain state for each unfinished stream.
464437--
465438newtype CoserialT m a = CoserialT {getCoserialT :: Stream m a}
466− deriving (Monoid, MonadTrans, MonadIO, MonadThrow)
467−
468−deriving instance MonadParallel m => Alternative (CoserialT m)
469−deriving instance MonadParallel m => MonadPlus (CoserialT m)
470−deriving instance (MonadBase b m, Monad m) => MonadBase b (CoserialT m)
471−deriving instance MonadError e m => MonadError e (CoserialT m)
472−deriving instance MonadReader r m => MonadReader r (CoserialT m)
473−deriving instance MonadState s m => MonadState s (CoserialT m)
439+ deriving (Functor, MonadTrans)
474440
475441{-# DEPRECATED InterleavedT "Please use 'CoserialT' instead." #-}
476442type InterleavedT = CoserialT
@@ -500,6 +466,14 @@ infixr 5 <=>
500466(<=>) :: IsStream t => t m a -> t m a -> t m a
501467(<=>) = coserial
502468
469+------------------------------------------------------------------------------
470+-- Monoid
471+------------------------------------------------------------------------------
472+
473+instance Monoid (CoserialT m a) where
474+ mempty = nil
475+ mappend = (<>)
476+
503477------------------------------------------------------------------------------
504478-- Monad
505479------------------------------------------------------------------------------
@@ -513,66 +487,11 @@ instance Monad m => Monad (CoserialT m) where
513487 in m Nothing stp single yield
514488
515489------------------------------------------------------------------------------
516−-- Applicative
517−------------------------------------------------------------------------------
518−
519−instance Monad m => Applicative (CoserialT m) where
520− pure = CoserialT . singleton
521− (<*>) = ap
522−
523−------------------------------------------------------------------------------
524−-- Functor
525−------------------------------------------------------------------------------
526−
527−instance Monad m => Functor (CoserialT m) where
528− fmap f m = fromStream $ Stream $ \_ stp sng yld ->
529− let single = sng . f
530− yield a r = yld (f a) (toStream $ fmap f (CoserialT r))
531− in (S.runStream (toStream m)) Nothing stp single yield
532−
533−------------------------------------------------------------------------------
534−-- Num
490+-- Other instances
535491------------------------------------------------------------------------------
536492
537−instance (Monad m, Num a) => Num (CoserialT m a) where
538− fromInteger n = pure (fromInteger n)
539−
540− negate = fmap negate
541− abs = fmap abs
542− signum = fmap signum
543−
544− (+) = liftA2 (+)
545− (*) = liftA2 (*)
546− (-) = liftA2 (-)
547−
548−instance (Monad m, Fractional a) => Fractional (CoserialT m a) where
549− fromRational n = pure (fromRational n)
550−
551− recip = fmap recip
552−
553− (/) = liftA2 (/)
554−
555−instance (Monad m, Floating a) => Floating (CoserialT m a) where
556− pi = pure pi
557−
558− exp = fmap exp
559− sqrt = fmap sqrt
560− log = fmap log
561− sin = fmap sin
562− tan = fmap tan
563− cos = fmap cos
564− asin = fmap asin
565− atan = fmap atan
566− acos = fmap acos
567− sinh = fmap sinh
568− tanh = fmap tanh
569− cosh = fmap cosh
570− asinh = fmap asinh
571− atanh = fmap atanh
572− acosh = fmap acosh
573−
574− (**) = liftA2 (**)
575− logBase = liftA2 logBase
493+MONAD_APPLICATIVE_INSTANCE(CoserialT,)
494+MONAD_COMMON_INSTANCES(CoserialT,)
576495
577496------------------------------------------------------------------------------
578497-- CoparallelT
@@ -618,17 +537,7 @@ instance (Monad m, Floating a) => Floating (CoserialT m a) where
618537-- as it explores only a bounded number of streams at a time.
619538--
620539newtype CoparallelT m a = CoparallelT {getCoparallelT :: Stream m a}
621− deriving (MonadTrans)
622−
623−deriving instance MonadParallel m => Monoid (CoparallelT m a)
624−deriving instance MonadParallel m => Alternative (CoparallelT m)
625−deriving instance MonadParallel m => MonadPlus (CoparallelT m)
626−deriving instance MonadParallel m => MonadIO (CoparallelT m)
627−deriving instance MonadParallel m => MonadThrow (CoparallelT m)
628−deriving instance (MonadBase b m, MonadParallel m) => MonadBase b (CoparallelT m)
629−deriving instance (MonadError e m, MonadParallel m) => MonadError e (CoparallelT m)
630−deriving instance (MonadReader r m, MonadParallel m) => MonadReader r (CoparallelT m)
631−deriving instance (MonadState s m, MonadParallel m) => MonadState s (CoparallelT m)
540+ deriving (Functor, MonadTrans)
632541
633542{-# DEPRECATED AsyncT "Please use 'CoparallelT' instead." #-}
634543type AsyncT = CoparallelT
@@ -657,6 +566,14 @@ instance MonadParallel m => Semigroup (CoparallelT m a) where
657566(<|) :: (IsStream t, MonadParallel m) => t m a -> t m a -> t m a
658567(<|) = coparallel
659568
569+------------------------------------------------------------------------------
570+-- Monoid
571+------------------------------------------------------------------------------
572+
573+instance MonadParallel m => Monoid (CoparallelT m a) where
574+ mempty = nil
575+ mappend = (<>)
576+
660577------------------------------------------------------------------------------
661578-- Monad
662579------------------------------------------------------------------------------
@@ -682,66 +599,11 @@ instance MonadParallel m => Monad (CoparallelT m) where
682599 where par = S.joinStreamVar2 (SVarStyle Conjunction LIFO)
683600
684601------------------------------------------------------------------------------
685−-- Applicative
602+-- Other instances
686603------------------------------------------------------------------------------
687604
688−instance MonadParallel m => Applicative (CoparallelT m) where
689− pure = CoparallelT . singleton
690− (<*>) = ap
691−
692−------------------------------------------------------------------------------
693−-- Functor
694−------------------------------------------------------------------------------
695−
696−instance Monad m => Functor (CoparallelT m) where
697− fmap f m = fromStream $ Stream $ \_ stp sng yld ->
698− let single = sng . f
699− yield a r = yld (f a) (toStream $ fmap f (CoparallelT r))
700− in (S.runStream (toStream m)) Nothing stp single yield
701−
702−------------------------------------------------------------------------------
703−-- Num
704−------------------------------------------------------------------------------
705−
706−instance (MonadParallel m, Num a) => Num (CoparallelT m a) where
707− fromInteger n = pure (fromInteger n)
708−
709− negate = fmap negate
710− abs = fmap abs
711− signum = fmap signum
712−
713− (+) = liftA2 (+)
714− (*) = liftA2 (*)
715− (-) = liftA2 (-)
716−
717−instance (MonadParallel m, Fractional a) => Fractional (CoparallelT m a) where
718− fromRational n = pure (fromRational n)
719−
720− recip = fmap recip
721−
722− (/) = liftA2 (/)
723−
724−instance (MonadParallel m, Floating a) => Floating (CoparallelT m a) where
725− pi = pure pi
726−
727− exp = fmap exp
728− sqrt = fmap sqrt
729− log = fmap log
730− sin = fmap sin
731− tan = fmap tan
732− cos = fmap cos
733− asin = fmap asin
734− atan = fmap atan
735− acos = fmap acos
736− sinh = fmap sinh
737− tanh = fmap tanh
738− cosh = fmap cosh
739− asinh = fmap asinh
740− atanh = fmap atanh
741− acosh = fmap acosh
742−
743− (**) = liftA2 (**)
744− logBase = liftA2 logBase
605+MONAD_APPLICATIVE_INSTANCE(CoparallelT,MONADPARALLEL)
606+MONAD_COMMON_INSTANCES(CoparallelT, MONADPARALLEL)
745607
746608------------------------------------------------------------------------------
747609-- ParallelT
@@ -785,17 +647,7 @@ instance (MonadParallel m, Floating a) => Floating (CoparallelT m a) where
785647-- streams as it needs to retain state for each unfinished stream.
786648--
787649newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
788− deriving (MonadTrans)
789−
790−deriving instance MonadParallel m => Monoid (ParallelT m a)
791−deriving instance MonadParallel m => Alternative (ParallelT m)
792−deriving instance MonadParallel m => MonadPlus (ParallelT m)
793−deriving instance MonadParallel m => MonadIO (ParallelT m)
794−deriving instance MonadParallel m => MonadThrow (ParallelT m)
795−deriving instance (MonadBase b m, MonadParallel m) => MonadBase b (ParallelT m)
796−deriving instance (MonadError e m, MonadParallel m) => MonadError e (ParallelT m)
797−deriving instance (MonadReader r m, MonadParallel m) => MonadReader r (ParallelT m)
798−deriving instance (MonadState s m, MonadParallel m) => MonadState s (ParallelT m)
650+ deriving (Functor, MonadTrans)
799651
800652instance IsStream ParallelT where
801653 toStream = getParallelT
@@ -814,6 +666,14 @@ parallel m1 m2 = fromStream $ S.parallel (toStream m1) (toStream m2)
814666instance MonadParallel m => Semigroup (ParallelT m a) where
815667 (<>) = parallel
816668
669+------------------------------------------------------------------------------
670+-- Monoid
671+------------------------------------------------------------------------------
672+
673+instance MonadParallel m => Monoid (ParallelT m a) where
674+ mempty = nil
675+ mappend = (<>)
676+
817677------------------------------------------------------------------------------
818678-- Monad
819679------------------------------------------------------------------------------
@@ -824,66 +684,11 @@ instance MonadParallel m => Monad (ParallelT m) where
824684 where par = S.joinStreamVar2 (SVarStyle Conjunction FIFO)
825685
826686------------------------------------------------------------------------------
827−-- Applicative
828−------------------------------------------------------------------------------
829−
830−instance MonadParallel m => Applicative (ParallelT m) where
831− pure = ParallelT . singleton
832− (<*>) = ap
833−
834−------------------------------------------------------------------------------
835−-- Functor
687+-- Other instances
836688------------------------------------------------------------------------------
837689
838−instance Monad m => Functor (ParallelT m) where
839− fmap f m = fromStream $ Stream $ \_ stp sng yld ->
840− let single = sng . f
841− yield a r = yld (f a) (toStream $ fmap f (ParallelT r))
842− in (S.runStream (toStream m)) Nothing stp single yield
843−
844−------------------------------------------------------------------------------
845−-- Num
846−------------------------------------------------------------------------------
847−
848−instance (MonadParallel m, Num a) => Num (ParallelT m a) where
849− fromInteger n = pure (fromInteger n)
850−
851− negate = fmap negate
852− abs = fmap abs
853− signum = fmap signum
854−
855− (+) = liftA2 (+)
856− (*) = liftA2 (*)
857− (-) = liftA2 (-)
858−
859−instance (MonadParallel m, Fractional a) => Fractional (ParallelT m a) where
860− fromRational n = pure (fromRational n)
861−
862− recip = fmap recip
863−
864− (/) = liftA2 (/)
865−
866−instance (MonadParallel m, Floating a) => Floating (ParallelT m a) where
867− pi = pure pi
868−
869− exp = fmap exp
870− sqrt = fmap sqrt
871− log = fmap log
872− sin = fmap sin
873− tan = fmap tan
874− cos = fmap cos
875− asin = fmap asin
876− atan = fmap atan
877− acos = fmap acos
878− sinh = fmap sinh
879− tanh = fmap tanh
880− cosh = fmap cosh
881− asinh = fmap asinh
882− atanh = fmap atanh
883− acosh = fmap acosh
884−
885− (**) = liftA2 (**)
886− logBase = liftA2 logBase
690+MONAD_APPLICATIVE_INSTANCE(ParallelT,MONADPARALLEL)
691+MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)
887692
888693------------------------------------------------------------------------------
889694-- Serially Zipping Streams
@@ -920,20 +725,12 @@ zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
920725-- 'SerialT'.
921726--
922727newtype ZipSerial m a = ZipSerial {getZipSerial :: Stream m a}
923− deriving (Semigroup, Monoid)
728+ deriving (Functor, Semigroup, Monoid)
924729
925730{-# Deprecated ZipStream "Please use ZipSerial instead." #-}
926731-- | Same as ZipSerial.
927732type ZipStream = ZipSerial
928733
929−deriving instance MonadParallel m => Alternative (ZipSerial m)
930−
931−instance Monad m => Functor (ZipSerial m) where
932− fmap f m = fromStream $ Stream $ \_ stp sng yld ->
933− let single = sng . f
934− yield a r = yld (f a) (toStream $ fmap f (ZipSerial r))
935− in (S.runStream (toStream m)) Nothing stp single yield
936−
937734instance Monad m => Applicative (ZipSerial m) where
938735 pure = ZipSerial . S.repeat
939736 (<*>) = zipWith id
@@ -942,46 +739,6 @@ instance IsStream ZipSerial where
942739 toStream = getZipSerial
943740 fromStream = ZipSerial
944741
945−instance (Monad m, Num a) => Num (ZipSerial m a) where
946− fromInteger n = pure (fromInteger n)
947−
948− negate = fmap negate
949− abs = fmap abs
950− signum = fmap signum
951−
952− (+) = liftA2 (+)
953− (*) = liftA2 (*)
954− (-) = liftA2 (-)
955−
956−instance (Monad m, Fractional a) => Fractional (ZipSerial m a) where
957− fromRational n = pure (fromRational n)
958−
959− recip = fmap recip
960−
961− (/) = liftA2 (/)
962−
963−instance (Monad m, Floating a) => Floating (ZipSerial m a) where
964− pi = pure pi
965−
966− exp = fmap exp
967− sqrt = fmap sqrt
968− log = fmap log
969− sin = fmap sin
970− tan = fmap tan
971− cos = fmap cos
972− asin = fmap asin
973− atan = fmap atan
974− acos = fmap acos
975− sinh = fmap sinh
976− tanh = fmap tanh
977− cosh = fmap cosh
978− asinh = fmap asinh
979− atanh = fmap atanh
980− acosh = fmap acosh
981−
982− (**) = liftA2 (**)
983− logBase = liftA2 logBase
984−
985742------------------------------------------------------------------------------
986743-- Parallely Zipping Streams
987744------------------------------------------------------------------------------
@@ -1017,19 +774,11 @@ zipAsyncWith = zipParallelWith
1017774-- 'SerialT'.
1018775--
1019776newtype ZipParallel m a = ZipParallel {getZipParallel :: Stream m a}
1020− deriving (Semigroup, Monoid)
777+ deriving (Functor, Semigroup, Monoid)
1021778
1022779{-# DEPRECATED ZipAsync "Please use ZipParallel instead." #-}
1023780type ZipAsync = ZipParallel
1024781
1025−deriving instance MonadParallel m => Alternative (ZipParallel m)
1026−
1027−instance Monad m => Functor (ZipParallel m) where
1028− fmap f m = fromStream $ Stream $ \_ stp sng yld ->
1029− let single = sng . f
1030− yield a r = yld (f a) (toStream $ fmap f (ZipParallel r))
1031− in (S.runStream (toStream m)) Nothing stp single yield
1032−
1033782instance MonadParallel m => Applicative (ZipParallel m) where
1034783 pure = ZipParallel . S.repeat
1035784 (<*>) = zipParallelWith id
@@ -1038,46 +787,6 @@ instance IsStream ZipParallel where
1038787 toStream = getZipParallel
1039788 fromStream = ZipParallel
1040789
1041−instance (MonadParallel m, Num a) => Num (ZipParallel m a) where
1042− fromInteger n = pure (fromInteger n)
1043−
1044− negate = fmap negate
1045− abs = fmap abs
1046− signum = fmap signum
1047−
1048− (+) = liftA2 (+)
1049− (*) = liftA2 (*)
1050− (-) = liftA2 (-)
1051−
1052−instance (MonadParallel m, Fractional a) => Fractional (ZipParallel m a) where
1053− fromRational n = pure (fromRational n)
1054−
1055− recip = fmap recip
1056−
1057− (/) = liftA2 (/)
1058−
1059−instance (MonadParallel m, Floating a) => Floating (ZipParallel m a) where
1060− pi = pure pi
1061−
1062− exp = fmap exp
1063− sqrt = fmap sqrt
1064− log = fmap log
1065− sin = fmap sin
1066− tan = fmap tan
1067− cos = fmap cos
1068− asin = fmap asin
1069− atan = fmap atan
1070− acos = fmap acos
1071− sinh = fmap sinh
1072− tanh = fmap tanh
1073− cosh = fmap cosh
1074− asinh = fmap asinh
1075− atanh = fmap atanh
1076− acosh = fmap acosh
1077−
1078− (**) = liftA2 (**)
1079− logBase = liftA2 logBase
1080−
1081790-------------------------------------------------------------------------------
1082791-- Type adapting combinators
1083792-------------------------------------------------------------------------------