@@ -1,3 +1,4 @@
1+{-# LANGUAGE CPP #-}
12{-# LANGUAGE ConstraintKinds #-}
23{-# LANGUAGE FlexibleContexts #-}
34{-# LANGUAGE FlexibleInstances #-}
@@ -102,15 +103,15 @@ 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)
106+import Control.Applicative (liftA2)
107+import Control.Monad (ap)
108+import Control.Monad.Base (MonadBase (..), liftBaseDefault)
109+import Control.Monad.Catch (MonadThrow, throwM)
109110import Control.Monad.Error.Class (MonadError(..))
110111import Control.Monad.IO.Class (MonadIO(..))
111112import Control.Monad.Reader.Class (MonadReader(..))
112113import Control.Monad.State.Class (MonadState(..))
113−import Control.Monad.Trans.Class (MonadTrans)
114+import Control.Monad.Trans.Class (MonadTrans (lift))
114115import Data.Semigroup (Semigroup(..))
115116import Prelude hiding (zipWith)
116117import Streamly.Core ( MonadParallel, Stream(Stream)
@@ -266,6 +267,85 @@ async m = do
266267 sv <- S.newStreamVar1 (SVarStyle Disjunction LIFO) (toStream m)
267268 return $ fromSVar sv
268269
270+------------------------------------------------------------------------------
271+-- CPP macros for common instances
272+------------------------------------------------------------------------------
273+
274+-- XXX use template haskell instead and include Monoid and IsStream instances
275+-- as well.
276+
277+#define MONADPARALLEL , MonadParallel m
278+
279+#define MONAD_APPLICATIVE_INSTANCE(STREAM,CONSTRAINT) \
280+instance (Monad m CONSTRAINT) => Applicative (STREAM m) where { \
281+ pure = STREAM . singleton; \
282+ (<*>) = ap }
283+
284+#define MONAD_COMMON_INSTANCES(STREAM,CONSTRAINT) \
285+instance (MonadBase b m, Monad m CONSTRAINT) => MonadBase b (STREAM m) where {\
286+ liftBase = liftBaseDefault }; \
287+ \
288+instance (MonadIO m CONSTRAINT) => MonadIO (STREAM m) where { \
289+ liftIO = lift . liftIO }; \
290+ \
291+instance (MonadThrow m CONSTRAINT) => MonadThrow (STREAM m) where { \
292+ throwM = lift . throwM }; \
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+instance (MonadReader r m CONSTRAINT) => MonadReader r (STREAM m) where { \
300+ ask = lift ask; \
301+ local f m = fromStream $ S.withLocal f (toStream m) }; \
302+ \
303+instance (MonadState s m CONSTRAINT) => MonadState s (STREAM m) where { \
304+ get = lift get; \
305+ put x = lift (put x); \
306+ state k = lift (state k) }
307+
308+#define NUMERIC_COMMON_INSTANCES(STREAM,CONSTRAINT) \
309+instance (Monad m, Num a CONSTRAINT) => Num (STREAM m a) where { \
310+ fromInteger n = pure (fromInteger n); \
311+ \
312+ negate = fmap negate; \
313+ abs = fmap abs; \
314+ signum = fmap signum; \
315+ \
316+ (+) = liftA2 (+); \
317+ (*) = liftA2 (*); \
318+ (-) = liftA2 (-) }; \
319+ \
320+instance (Monad m, Fractional a CONSTRAINT) => Fractional (STREAM m a) where {\
321+ fromRational n = pure (fromRational n); \
322+ \
323+ recip = fmap recip; \
324+ \
325+ (/) = liftA2 (/) }; \
326+ \
327+instance (Monad m, Floating a CONSTRAINT) => Floating (STREAM m a) where { \
328+ pi = pure pi; \
329+ \
330+ exp = fmap exp; \
331+ sqrt = fmap sqrt; \
332+ log = fmap log; \
333+ sin = fmap sin; \
334+ tan = fmap tan; \
335+ cos = fmap cos; \
336+ asin = fmap asin; \
337+ atan = fmap atan; \
338+ acos = fmap acos; \
339+ sinh = fmap sinh; \
340+ tanh = fmap tanh; \
341+ cosh = fmap cosh; \
342+ asinh = fmap asinh; \
343+ atanh = fmap atanh; \
344+ acosh = fmap acosh; \
345+ \
346+ (**) = liftA2 (**); \
347+ logBase = liftA2 logBase }
348+
269349------------------------------------------------------------------------------
270350-- SerialT
271351------------------------------------------------------------------------------
@@ -321,14 +401,7 @@ async m = do
321401-- streams as it explores only one stream at a time.
322402--
323403newtype SerialT m a = SerialT {getSerialT :: Stream m a}
324− deriving (Functor, 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)
404+ deriving (Semigroup, Monoid, Functor, MonadTrans)
332405
333406instance IsStream SerialT where
334407 toStream = getSerialT
@@ -338,11 +411,6 @@ instance IsStream SerialT where
338411{-# Deprecated StreamT "Please use SerialT instead." #-}
339412type StreamT = SerialT
340413
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−
346414------------------------------------------------------------------------------
347415-- Semigroup
348416------------------------------------------------------------------------------
@@ -367,56 +435,12 @@ instance Monad m => Monad (SerialT m) where
367435 in m Nothing stp single yield
368436
369437------------------------------------------------------------------------------
370−-- Applicative
371−------------------------------------------------------------------------------
372−
373−instance Monad m => Applicative (SerialT m) where
374− pure = SerialT . singleton
375− (<*>) = ap
376−
377−------------------------------------------------------------------------------
378−-- Num
438+-- Other instances
379439------------------------------------------------------------------------------
380440
381−instance (Monad m, Num a) => Num (SerialT m a) where
382− fromInteger n = pure (fromInteger n)
383−
384− negate = fmap negate
385− abs = fmap abs
386− signum = fmap signum
387−
388− (+) = liftA2 (+)
389− (*) = liftA2 (*)
390− (-) = liftA2 (-)
391−
392−instance (Monad m, Fractional a) => Fractional (SerialT m a) where
393− fromRational n = pure (fromRational n)
394−
395− recip = fmap recip
396−
397− (/) = liftA2 (/)
398−
399−instance (Monad m, Floating a) => Floating (SerialT m a) where
400− pi = pure pi
401−
402− exp = fmap exp
403− sqrt = fmap sqrt
404− log = fmap log
405− sin = fmap sin
406− tan = fmap tan
407− cos = fmap cos
408− asin = fmap asin
409− atan = fmap atan
410− acos = fmap acos
411− sinh = fmap sinh
412− tanh = fmap tanh
413− cosh = fmap cosh
414− asinh = fmap asinh
415− atanh = fmap atanh
416− acosh = fmap acosh
417−
418− (**) = liftA2 (**)
419− logBase = liftA2 logBase
441+MONAD_APPLICATIVE_INSTANCE(SerialT,)
442+MONAD_COMMON_INSTANCES(SerialT,)
443+NUMERIC_COMMON_INSTANCES(SerialT,)
420444
421445------------------------------------------------------------------------------
422446-- CoserialT
@@ -453,14 +477,7 @@ instance (Monad m, Floating a) => Floating (SerialT m a) where
453477-- streams as it needs to retain state for each unfinished stream.
454478--
455479newtype CoserialT m a = CoserialT {getCoserialT :: Stream m a}
456− deriving (Functor, Monoid, MonadTrans, MonadIO, MonadThrow)
457−
458−deriving instance MonadParallel m => Alternative (CoserialT m)
459−deriving instance MonadParallel m => MonadPlus (CoserialT m)
460−deriving instance (MonadBase b m, Monad m) => MonadBase b (CoserialT m)
461−deriving instance MonadError e m => MonadError e (CoserialT m)
462−deriving instance MonadReader r m => MonadReader r (CoserialT m)
463−deriving instance MonadState s m => MonadState s (CoserialT m)
480+ deriving (Functor, MonadTrans)
464481
465482{-# DEPRECATED InterleavedT "Please use 'CoserialT' instead." #-}
466483type InterleavedT = CoserialT
@@ -490,6 +507,14 @@ infixr 5 <=>
490507(<=>) :: IsStream t => t m a -> t m a -> t m a
491508(<=>) = coserial
492509
510+------------------------------------------------------------------------------
511+-- Monoid
512+------------------------------------------------------------------------------
513+
514+instance Monoid (CoserialT m a) where
515+ mempty = nil
516+ mappend = (<>)
517+
493518------------------------------------------------------------------------------
494519-- Monad
495520------------------------------------------------------------------------------
@@ -503,56 +528,12 @@ instance Monad m => Monad (CoserialT m) where
503528 in m Nothing stp single yield
504529
505530------------------------------------------------------------------------------
506−-- Applicative
531+-- Other instances
507532------------------------------------------------------------------------------
508533
509−instance Monad m => Applicative (CoserialT m) where
510− pure = CoserialT . singleton
511− (<*>) = ap
512−
513−------------------------------------------------------------------------------
514−-- Num
515−------------------------------------------------------------------------------
516−
517−instance (Monad m, Num a) => Num (CoserialT m a) where
518− fromInteger n = pure (fromInteger n)
519−
520− negate = fmap negate
521− abs = fmap abs
522− signum = fmap signum
523−
524− (+) = liftA2 (+)
525− (*) = liftA2 (*)
526− (-) = liftA2 (-)
527−
528−instance (Monad m, Fractional a) => Fractional (CoserialT m a) where
529− fromRational n = pure (fromRational n)
530−
531− recip = fmap recip
532−
533− (/) = liftA2 (/)
534−
535−instance (Monad m, Floating a) => Floating (CoserialT m a) where
536− pi = pure pi
537−
538− exp = fmap exp
539− sqrt = fmap sqrt
540− log = fmap log
541− sin = fmap sin
542− tan = fmap tan
543− cos = fmap cos
544− asin = fmap asin
545− atan = fmap atan
546− acos = fmap acos
547− sinh = fmap sinh
548− tanh = fmap tanh
549− cosh = fmap cosh
550− asinh = fmap asinh
551− atanh = fmap atanh
552− acosh = fmap acosh
553−
554− (**) = liftA2 (**)
555− logBase = liftA2 logBase
534+MONAD_APPLICATIVE_INSTANCE(CoserialT,)
535+MONAD_COMMON_INSTANCES(CoserialT,)
536+NUMERIC_COMMON_INSTANCES(CoserialT,)
556537
557538------------------------------------------------------------------------------
558539-- CoparallelT
@@ -600,16 +581,6 @@ instance (Monad m, Floating a) => Floating (CoserialT m a) where
600581newtype CoparallelT m a = CoparallelT {getCoparallelT :: Stream m a}
601582 deriving (Functor, MonadTrans)
602583
603−deriving instance MonadParallel m => Monoid (CoparallelT m a)
604−deriving instance MonadParallel m => Alternative (CoparallelT m)
605−deriving instance MonadParallel m => MonadPlus (CoparallelT m)
606−deriving instance MonadParallel m => MonadIO (CoparallelT m)
607−deriving instance MonadParallel m => MonadThrow (CoparallelT m)
608−deriving instance (MonadBase b m, MonadParallel m) => MonadBase b (CoparallelT m)
609−deriving instance (MonadError e m, MonadParallel m) => MonadError e (CoparallelT m)
610−deriving instance (MonadReader r m, MonadParallel m) => MonadReader r (CoparallelT m)
611−deriving instance (MonadState s m, MonadParallel m) => MonadState s (CoparallelT m)
612−
613584{-# DEPRECATED AsyncT "Please use 'CoparallelT' instead." #-}
614585type AsyncT = CoparallelT
615586
@@ -637,6 +608,14 @@ instance MonadParallel m => Semigroup (CoparallelT m a) where
637608(<|) :: (IsStream t, MonadParallel m) => t m a -> t m a -> t m a
638609(<|) = coparallel
639610
611+------------------------------------------------------------------------------
612+-- Monoid
613+------------------------------------------------------------------------------
614+
615+instance MonadParallel m => Monoid (CoparallelT m a) where
616+ mempty = nil
617+ mappend = (<>)
618+
640619------------------------------------------------------------------------------
641620-- Monad
642621------------------------------------------------------------------------------
@@ -662,56 +641,12 @@ instance MonadParallel m => Monad (CoparallelT m) where
662641 where par = S.joinStreamVar2 (SVarStyle Conjunction LIFO)
663642
664643------------------------------------------------------------------------------
665−-- Applicative
644+-- Other instances
666645------------------------------------------------------------------------------
667646
668−instance MonadParallel m => Applicative (CoparallelT m) where
669− pure = CoparallelT . singleton
670− (<*>) = ap
671−
672−------------------------------------------------------------------------------
673−-- Num
674−------------------------------------------------------------------------------
675−
676−instance (MonadParallel m, Num a) => Num (CoparallelT m a) where
677− fromInteger n = pure (fromInteger n)
678−
679− negate = fmap negate
680− abs = fmap abs
681− signum = fmap signum
682−
683− (+) = liftA2 (+)
684− (*) = liftA2 (*)
685− (-) = liftA2 (-)
686−
687−instance (MonadParallel m, Fractional a) => Fractional (CoparallelT m a) where
688− fromRational n = pure (fromRational n)
689−
690− recip = fmap recip
691−
692− (/) = liftA2 (/)
693−
694−instance (MonadParallel m, Floating a) => Floating (CoparallelT m a) where
695− pi = pure pi
696−
697− exp = fmap exp
698− sqrt = fmap sqrt
699− log = fmap log
700− sin = fmap sin
701− tan = fmap tan
702− cos = fmap cos
703− asin = fmap asin
704− atan = fmap atan
705− acos = fmap acos
706− sinh = fmap sinh
707− tanh = fmap tanh
708− cosh = fmap cosh
709− asinh = fmap asinh
710− atanh = fmap atanh
711− acosh = fmap acosh
712−
713− (**) = liftA2 (**)
714− logBase = liftA2 logBase
647+MONAD_APPLICATIVE_INSTANCE(CoparallelT,MONADPARALLEL)
648+MONAD_COMMON_INSTANCES(CoparallelT, MONADPARALLEL)
649+NUMERIC_COMMON_INSTANCES(CoparallelT, MONADPARALLEL)
715650
716651------------------------------------------------------------------------------
717652-- ParallelT
@@ -757,16 +692,6 @@ instance (MonadParallel m, Floating a) => Floating (CoparallelT m a) where
757692newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
758693 deriving (Functor, MonadTrans)
759694
760−deriving instance MonadParallel m => Monoid (ParallelT m a)
761−deriving instance MonadParallel m => Alternative (ParallelT m)
762−deriving instance MonadParallel m => MonadPlus (ParallelT m)
763−deriving instance MonadParallel m => MonadIO (ParallelT m)
764−deriving instance MonadParallel m => MonadThrow (ParallelT m)
765−deriving instance (MonadBase b m, MonadParallel m) => MonadBase b (ParallelT m)
766−deriving instance (MonadError e m, MonadParallel m) => MonadError e (ParallelT m)
767−deriving instance (MonadReader r m, MonadParallel m) => MonadReader r (ParallelT m)
768−deriving instance (MonadState s m, MonadParallel m) => MonadState s (ParallelT m)
769−
770695instance IsStream ParallelT where
771696 toStream = getParallelT
772697 fromStream = ParallelT
@@ -784,6 +709,14 @@ parallel m1 m2 = fromStream $ S.parallel (toStream m1) (toStream m2)
784709instance MonadParallel m => Semigroup (ParallelT m a) where
785710 (<>) = parallel
786711
712+------------------------------------------------------------------------------
713+-- Monoid
714+------------------------------------------------------------------------------
715+
716+instance MonadParallel m => Monoid (ParallelT m a) where
717+ mempty = nil
718+ mappend = (<>)
719+
787720------------------------------------------------------------------------------
788721-- Monad
789722------------------------------------------------------------------------------
@@ -794,56 +727,12 @@ instance MonadParallel m => Monad (ParallelT m) where
794727 where par = S.joinStreamVar2 (SVarStyle Conjunction FIFO)
795728
796729------------------------------------------------------------------------------
797−-- Applicative
730+-- Other instances
798731------------------------------------------------------------------------------
799732
800−instance MonadParallel m => Applicative (ParallelT m) where
801− pure = ParallelT . singleton
802− (<*>) = ap
803−
804−------------------------------------------------------------------------------
805−-- Num
806−------------------------------------------------------------------------------
807−
808−instance (MonadParallel m, Num a) => Num (ParallelT m a) where
809− fromInteger n = pure (fromInteger n)
810−
811− negate = fmap negate
812− abs = fmap abs
813− signum = fmap signum
814−
815− (+) = liftA2 (+)
816− (*) = liftA2 (*)
817− (-) = liftA2 (-)
818−
819−instance (MonadParallel m, Fractional a) => Fractional (ParallelT m a) where
820− fromRational n = pure (fromRational n)
821−
822− recip = fmap recip
823−
824− (/) = liftA2 (/)
825−
826−instance (MonadParallel m, Floating a) => Floating (ParallelT m a) where
827− pi = pure pi
828−
829− exp = fmap exp
830− sqrt = fmap sqrt
831− log = fmap log
832− sin = fmap sin
833− tan = fmap tan
834− cos = fmap cos
835− asin = fmap asin
836− atan = fmap atan
837− acos = fmap acos
838− sinh = fmap sinh
839− tanh = fmap tanh
840− cosh = fmap cosh
841− asinh = fmap asinh
842− atanh = fmap atanh
843− acosh = fmap acosh
844−
845− (**) = liftA2 (**)
846− logBase = liftA2 logBase
733+MONAD_APPLICATIVE_INSTANCE(ParallelT,MONADPARALLEL)
734+MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)
735+NUMERIC_COMMON_INSTANCES(ParallelT, MONADPARALLEL)
847736
848737------------------------------------------------------------------------------
849738-- Serially Zipping Streams
@@ -886,8 +775,6 @@ newtype ZipSerial m a = ZipSerial {getZipSerial :: Stream m a}
886775-- | Same as ZipSerial.
887776type ZipStream = ZipSerial
888777
889−deriving instance MonadParallel m => Alternative (ZipSerial m)
890−
891778instance Monad m => Applicative (ZipSerial m) where
892779 pure = ZipSerial . S.repeat
893780 (<*>) = zipWith id
@@ -896,45 +783,7 @@ instance IsStream ZipSerial where
896783 toStream = getZipSerial
897784 fromStream = ZipSerial
898785
899−instance (Monad m, Num a) => Num (ZipSerial m a) where
900− fromInteger n = pure (fromInteger n)
901−
902− negate = fmap negate
903− abs = fmap abs
904− signum = fmap signum
905−
906− (+) = liftA2 (+)
907− (*) = liftA2 (*)
908− (-) = liftA2 (-)
909−
910−instance (Monad m, Fractional a) => Fractional (ZipSerial m a) where
911− fromRational n = pure (fromRational n)
912−
913− recip = fmap recip
914−
915− (/) = liftA2 (/)
916−
917−instance (Monad m, Floating a) => Floating (ZipSerial m a) where
918− pi = pure pi
919−
920− exp = fmap exp
921− sqrt = fmap sqrt
922− log = fmap log
923− sin = fmap sin
924− tan = fmap tan
925− cos = fmap cos
926− asin = fmap asin
927− atan = fmap atan
928− acos = fmap acos
929− sinh = fmap sinh
930− tanh = fmap tanh
931− cosh = fmap cosh
932− asinh = fmap asinh
933− atanh = fmap atanh
934− acosh = fmap acosh
935−
936− (**) = liftA2 (**)
937− logBase = liftA2 logBase
786+NUMERIC_COMMON_INSTANCES(ZipSerial,)
938787
939788------------------------------------------------------------------------------
940789-- Parallely Zipping Streams
@@ -976,8 +825,6 @@ newtype ZipParallel m a = ZipParallel {getZipParallel :: Stream m a}
976825{-# DEPRECATED ZipAsync "Please use ZipParallel instead." #-}
977826type ZipAsync = ZipParallel
978827
979−deriving instance MonadParallel m => Alternative (ZipParallel m)
980−
981828instance MonadParallel m => Applicative (ZipParallel m) where
982829 pure = ZipParallel . S.repeat
983830 (<*>) = zipParallelWith id
@@ -986,45 +833,7 @@ instance IsStream ZipParallel where
986833 toStream = getZipParallel
987834 fromStream = ZipParallel
988835
989−instance (MonadParallel m, Num a) => Num (ZipParallel m a) where
990− fromInteger n = pure (fromInteger n)
991−
992− negate = fmap negate
993− abs = fmap abs
994− signum = fmap signum
995−
996− (+) = liftA2 (+)
997− (*) = liftA2 (*)
998− (-) = liftA2 (-)
999−
1000−instance (MonadParallel m, Fractional a) => Fractional (ZipParallel m a) where
1001− fromRational n = pure (fromRational n)
1002−
1003− recip = fmap recip
1004−
1005− (/) = liftA2 (/)
1006−
1007−instance (MonadParallel m, Floating a) => Floating (ZipParallel m a) where
1008− pi = pure pi
1009−
1010− exp = fmap exp
1011− sqrt = fmap sqrt
1012− log = fmap log
1013− sin = fmap sin
1014− tan = fmap tan
1015− cos = fmap cos
1016− asin = fmap asin
1017− atan = fmap atan
1018− acos = fmap acos
1019− sinh = fmap sinh
1020− tanh = fmap tanh
1021− cosh = fmap cosh
1022− asinh = fmap asinh
1023− atanh = fmap atanh
1024− acosh = fmap acosh
1025−
1026− (**) = liftA2 (**)
1027− logBase = liftA2 logBase
836+NUMERIC_COMMON_INSTANCES(ZipParallel,MONADPARALLEL)
1028837
1029838-------------------------------------------------------------------------------
1030839-- Type adapting combinators