99afd31 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Merge pull request #69 from composewell/stream-type

99afd31Harendra Kumar9 years ago

Changelog.md+4

compare with master

unifiedsplit

@@ -28,6 +28,10 @@

28 Stream types may have to be changed or type combinators may have to be added

29 or removed to adapt to this change.

30* Change the type of `foldrM` to make it consistent with `foldrM` in base.

31+* Remove the MonadError instance as it was not working correctly for

32+ parallel compositions. Use MonadThrow instead for error propagation.

33+* Remove Num/Fractional/Floating instances as they are not very useful. Use

34+ `fmap` and `liftA2` instead.

32### Deprecations

36### Deprecations

33* Deprecate and rename the following symbols:

37* Deprecate and rename the following symbols:

benchmark/Main.hs+7−3

@@ -95,6 +95,10 @@ main = do

95map :: Monad m => (a -> Int) -> a -> m Int

96map f x = return $ f x

98+guard' :: (A.IsStream t, Applicative (t m)) => Bool -> t m ()

99+guard' True = pure ()

100+guard' False = A.nil

101+

98{-# INLINABLE filter #-}

102{-# INLINABLE filter #-}

99filter :: (Monad m, Alternative m) => (a -> Bool) -> a -> m a

103filter :: (Monad m, Alternative m) => (a -> Bool) -> a -> m a

100filter cond x = guard (not $ cond x) >> return x

104filter cond x = guard (not $ cond x) >> return x

@@ -102,12 +106,12 @@ filter cond x = guard (not $ cond x) >> return x

102amap :: Monad (s IO) => (Int -> Int) -> Int -> s IO Int

106amap :: Monad (s IO) => (Int -> Int) -> Int -> s IO Int

103amap = Main.map

107amap = Main.map

104

108

105−afilter :: (Alternative (s IO), Monad (s IO)) => (Int -> Bool) -> Int -> s IO Int

109+afilter :: (A.IsStream s, Monad (s IO)) => (Int -> Bool) -> Int -> s IO Int

106−afilter = Main.filter

110+afilter cond x = guard' (not $ cond x) >> return x

107

111

108{-# INLINE streamly_basic #-}

112{-# INLINE streamly_basic #-}

109streamly_basic

113streamly_basic

110− :: (Alternative (t IO), Monad (t IO), A.IsStream t)

114+ :: (Monad (t IO), A.IsStream t)

111 => (forall a. t IO a -> IO [a])

115 => (forall a. t IO a -> IO [a])

112 -> (t IO Int -> t IO Int -> t IO Int)

116 -> (t IO Int -> t IO Int -> t IO Int)

113 -> IO Int

117 -> IO Int

examples/loops.hs+5−4

@@ -1,5 +1,6 @@

1import Streamly

2import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))

3+import Streamly.Prelude (nil)

4main = do

5main = do

5 liftIO $ hSetBuffering stdout LineBuffering

6 liftIO $ hSetBuffering stdout LineBuffering

@@ -49,7 +50,7 @@ main = do

49 loopTail :: Int -> SerialT IO Int

50 loopTail :: Int -> SerialT IO Int

50 loopTail x = do

51 loopTail x = do

51 liftIO $ putStrLn "LoopTail..."

52 liftIO $ putStrLn "LoopTail..."

52− return x <> (if x < 3 then loopTail (x + 1) else empty)

53+ return x <> (if x < 3 then loopTail (x + 1) else nil)

54 -- Loops and then generates a value. The consumer can run only after the

55 -- Loops and then generates a value. The consumer can run only after the

55 -- loop has finished. An infinite generator will not let the consumer run

56 -- loop has finished. An infinite generator will not let the consumer run

@@ -57,7 +58,7 @@ main = do

57 loopHead :: Int -> SerialT IO Int

58 loopHead :: Int -> SerialT IO Int

58 loopHead x = do

59 loopHead x = do

59 liftIO $ putStrLn "LoopHead..."

60 liftIO $ putStrLn "LoopHead..."

60− (if x < 3 then loopHead (x + 1) else empty) <> return x

61+ (if x < 3 then loopHead (x + 1) else nil) <> return x

62-------------------------------------------------------------------------------

63-------------------------------------------------------------------------------

63-- Concurrent (multi-threaded) adaptive demand-based stream generator loops

64-- Concurrent (multi-threaded) adaptive demand-based stream generator loops

@@ -71,12 +72,12 @@ main = do

71 loopTailA :: Int -> SerialT IO Int

72 loopTailA :: Int -> SerialT IO Int

72 loopTailA x = do

73 loopTailA x = do

73 liftIO $ putStrLn "LoopTailA..."

74 liftIO $ putStrLn "LoopTailA..."

74− return x `coparallel` (if x < 3 then loopTailA (x + 1) else empty)

75+ return x `coparallel` (if x < 3 then loopTailA (x + 1) else nil)

76 loopHeadA :: Int -> SerialT IO Int

77 loopHeadA :: Int -> SerialT IO Int

77 loopHeadA x = do

78 loopHeadA x = do

78 liftIO $ putStrLn "LoopHeadA..."

79 liftIO $ putStrLn "LoopHeadA..."

79− (if x < 3 then loopHeadA (x + 1) else empty) `coparallel` return x

80+ (if x < 3 then loopHeadA (x + 1) else nil) `coparallel` return x

81-------------------------------------------------------------------------------

82-------------------------------------------------------------------------------

82-- Parallel (fairly scheduled, multi-threaded) stream generator loops

83-- Parallel (fairly scheduled, multi-threaded) stream generator loops

src/Streamly/Core.hs+25−67

@@ -39,6 +39,10 @@ module Streamly.Core

39 -- * Alternative

40 , alt

42+ -- * Transformers

43+ , withLocal

44+ , withCatchError

45+

42 -- * Concurrent Stream Vars (SVars)

46 -- * Concurrent Stream Vars (SVars)

43 , SVar

47 , SVar

44 , SVarSched (..)

48 , SVarSched (..)

@@ -53,20 +57,17 @@ module Streamly.Core

53 )

57 )

54where

58where

56−import Control.Applicative (Alternative (..))

57import Control.Concurrent (ThreadId, forkIO,

60import Control.Concurrent (ThreadId, forkIO,

58 myThreadId, threadDelay)

61 myThreadId, threadDelay)

59import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,

62import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,

60 tryPutMVar, takeMVar)

63 tryPutMVar, takeMVar)

61import Control.Exception (SomeException (..))

64import Control.Exception (SomeException (..))

62import qualified Control.Exception.Lifted as EL

65import qualified Control.Exception.Lifted as EL

63−import Control.Monad (MonadPlus(..), mzero, when)

66+import Control.Monad (when)

64−import Control.Monad.Base (MonadBase (..), liftBaseDefault)

65import Control.Monad.Catch (MonadThrow, throwM)

67import Control.Monad.Catch (MonadThrow, throwM)

66import Control.Monad.Error.Class (MonadError(..))

68import Control.Monad.Error.Class (MonadError(..))

67import Control.Monad.IO.Class (MonadIO(..))

69import Control.Monad.IO.Class (MonadIO(..))

68import Control.Monad.Reader.Class (MonadReader(..))

70import Control.Monad.Reader.Class (MonadReader(..))

69−import Control.Monad.State.Class (MonadState(..))

70import Control.Monad.Trans.Class (MonadTrans (lift))

71import Control.Monad.Trans.Class (MonadTrans (lift))

71import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)

72import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)

72import Data.Atomics (atomicModifyIORefCAS,

73import Data.Atomics (atomicModifyIORefCAS,

@@ -184,15 +185,6 @@ cons a r = Stream $ \_ _ _ yld -> yld a r

184repeat :: a -> Stream m a

185repeat :: a -> Stream m a

185repeat a = let x = cons a x in x

186repeat a = let x = cons a x in x

186

187

187−------------------------------------------------------------------------------

188−-- Composing streams

189−------------------------------------------------------------------------------

190−

191−-- Streams can be composed sequentially or in parallel; in product style

192−-- compositions (monadic bind multiplies streams in a ListT fashion) or in sum

193−-- style compositions like 'Semigroup', 'Monoid', 'Alternative' or variants of

194−-- these.

195−

196------------------------------------------------------------------------------

188------------------------------------------------------------------------------

197-- Semigroup

189-- Semigroup

198------------------------------------------------------------------------------

190------------------------------------------------------------------------------

@@ -582,25 +574,14 @@ parallel :: MonadParallel m => Stream m a -> Stream m a -> Stream m a

582parallel = joinStreamVar2 (SVarStyle Disjunction FIFO)

574parallel = joinStreamVar2 (SVarStyle Disjunction FIFO)

583

575

584-------------------------------------------------------------------------------

576-------------------------------------------------------------------------------

585−-- Instances (only used for deriving newtype instances)

577+-- Functor instace is the same for all types

586-------------------------------------------------------------------------------

578-------------------------------------------------------------------------------

587

579

588−-- Stream type is not exposed, these instances are only for deriving instances

589−-- for the newtype wrappers based on Stream.

590−

591−-- Dummy Instances, defined to enable the definition of other instances that

592−-- require a Monad constraint. Must be defined by the newtypes.

593−

594instance Monad m => Functor (Stream m) where

580instance Monad m => Functor (Stream m) where

595− fmap = undefined

581+ fmap f m = Stream $ \_ stp sng yld ->

596−

582+ let single = sng . f

597−instance Monad m => Applicative (Stream m) where

583+ yield a r = yld (f a) (fmap f r)

598− pure = undefined

584+ in (runStream m) Nothing stp single yield

599− (<*>) = undefined

600−

601−instance Monad m => Monad (Stream m) where

602− return = pure

603− (>>=) = undefined

604

585

605------------------------------------------------------------------------------

586------------------------------------------------------------------------------

606-- Alternative & MonadPlus

587-- Alternative & MonadPlus

@@ -611,50 +592,27 @@ alt m1 m2 = Stream $ \_ stp sng yld ->

611 let stop = runStream m2 Nothing stp sng yld

592 let stop = runStream m2 Nothing stp sng yld

612 in runStream m1 Nothing stop sng yld

593 in runStream m1 Nothing stop sng yld

613

594

614−instance MonadParallel m => Alternative (Stream m) where

615− empty = nil

616− (<|>) = alt

617−

618−instance MonadParallel m => MonadPlus (Stream m) where

619− mzero = nil

620− mplus = (<|>)

621−

622-------------------------------------------------------------------------------

595-------------------------------------------------------------------------------

623−-- Transformer

596+-- Transformers

624-------------------------------------------------------------------------------

597-------------------------------------------------------------------------------

625

598

626instance MonadTrans Stream where

599instance MonadTrans Stream where

627 lift mx = Stream $ \_ _ single _ -> mx >>= single

600 lift mx = Stream $ \_ _ single _ -> mx >>= single

628

601

629−instance (MonadBase b m, Monad m) => MonadBase b (Stream m) where

602+withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a

630− liftBase = liftBaseDefault

603+withLocal f m =

631−

632−------------------------------------------------------------------------------

633−-- Standard transformer instances

634−------------------------------------------------------------------------------

635−

636−instance MonadIO m => MonadIO (Stream m) where

637− liftIO = lift . liftIO

638−

639−instance MonadThrow m => MonadThrow (Stream m) where

640− throwM = lift . throwM

641−

642−-- XXX handle and test cross thread state transfer

643−instance MonadError e m => MonadError e (Stream m) where

644− throwError = lift . throwError

645− catchError m h = Stream $ \svr stp sng yld ->

646− let handle r = r `catchError` \e -> (runStream (h e)) svr stp sng yld

647− yield a r = yld a (catchError r h)

648− in handle $ (runStream m) svr stp sng yield

649−

650−instance MonadReader r m => MonadReader r (Stream m) where

651− ask = lift ask

652− local f m = Stream $ \svr stp sng yld ->

604+ Stream $ \svr stp sng yld ->

653 let single = local f . sng

605 let single = local f . sng

654− yield a r = local f $ yld a (local f r)

606+ yield a r = local f $ yld a (withLocal f r)

655 in (runStream m) svr (local f stp) single yield

607 in (runStream m) svr (local f stp) single yield

656

608

657−instance MonadState s m => MonadState s (Stream m) where

609+-- XXX handle and test cross thread state transfer

658− get = lift get

610+withCatchError

659− put x = lift (put x)

611+ :: MonadError e m

660− state k = lift (state k)

612+ => Stream m a -> (e -> Stream m a) -> Stream m a

613+withCatchError m h =

614+ Stream $ \svr stp sng yld ->

615+ let run x = runStream x svr stp sng yield

616+ handle r = r `catchError` \e -> run $ h e

617+ yield a r = yld a (withCatchError r h)

618+ in handle $ run m

src/Streamly/Streams.hs+88−379

@@ -1,3 +1,4 @@

1+{-# LANGUAGE CPP #-}

1{-# LANGUAGE ConstraintKinds #-}

2{-# LANGUAGE ConstraintKinds #-}

2{-# LANGUAGE FlexibleContexts #-}

3{-# LANGUAGE FlexibleContexts #-}

3{-# LANGUAGE FlexibleInstances #-}

4{-# LANGUAGE FlexibleInstances #-}

@@ -102,15 +103,14 @@ module Streamly.Streams

102 )

103 )

103where

104where

104

105

105−import Control.Applicative (Alternative (..), liftA2)

106+import Control.Monad (ap)

106−import Control.Monad (MonadPlus(..), ap)

107−import Control.Monad.Base (MonadBase (..))

107+import Control.Monad.Base (MonadBase (..), liftBaseDefault)

108−import Control.Monad.Catch (MonadThrow)

108+import Control.Monad.Catch (MonadThrow, throwM)

109−import Control.Monad.Error.Class (MonadError(..))

109+-- import Control.Monad.Error.Class (MonadError(..))

110import Control.Monad.IO.Class (MonadIO(..))

111import Control.Monad.Reader.Class (MonadReader(..))

112import Control.Monad.State.Class (MonadState(..))

113−import Control.Monad.Trans.Class (MonadTrans)

113+import Control.Monad.Trans.Class (MonadTrans (lift))

114import Data.Semigroup (Semigroup(..))

115import Prelude hiding (zipWith)

116import Streamly.Core ( MonadParallel, Stream(Stream)

@@ -266,6 +266,46 @@ async m = do

266 sv <- S.newStreamVar1 (SVarStyle Disjunction LIFO) (toStream m)

267 return $ fromSVar sv

268

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+

269------------------------------------------------------------------------------

309------------------------------------------------------------------------------

270-- SerialT

310-- SerialT

271------------------------------------------------------------------------------

311------------------------------------------------------------------------------

@@ -321,14 +361,7 @@ async m = do

321-- streams as it explores only one stream at a time.

361-- streams as it explores only one stream at a time.

322--

362--

323newtype SerialT m a = SerialT {getSerialT :: Stream m a}

363newtype SerialT m a = SerialT {getSerialT :: Stream m a}

324− deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)

364+ deriving (Semigroup, Monoid, Functor, MonadTrans)

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)

332

365

333instance IsStream SerialT where

366instance IsStream SerialT where

334 toStream = getSerialT

367 toStream = getSerialT

@@ -338,11 +371,6 @@ instance IsStream SerialT where

338{-# Deprecated StreamT "Please use SerialT instead." #-}

371{-# Deprecated StreamT "Please use SerialT instead." #-}

339type StreamT = SerialT

372type StreamT = SerialT

340

373

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−

346------------------------------------------------------------------------------

374------------------------------------------------------------------------------

347-- Semigroup

375-- Semigroup

348------------------------------------------------------------------------------

376------------------------------------------------------------------------------

@@ -367,66 +395,11 @@ instance Monad m => Monad (SerialT m) where

367 in m Nothing stp single yield

395 in m Nothing stp single yield

368

396

369------------------------------------------------------------------------------

397------------------------------------------------------------------------------

370−-- Applicative

398+-- Other instances

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

389------------------------------------------------------------------------------

399------------------------------------------------------------------------------

390

400

391−instance (Monad m, Num a) => Num (SerialT m a) where

401+MONAD_APPLICATIVE_INSTANCE(SerialT,)

392− fromInteger n = pure (fromInteger n)

402+MONAD_COMMON_INSTANCES(SerialT,)

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

430

403

431------------------------------------------------------------------------------

404------------------------------------------------------------------------------

432-- CoserialT

405-- CoserialT

@@ -463,14 +436,7 @@ instance (Monad m, Floating a) => Floating (SerialT m a) where

463-- streams as it needs to retain state for each unfinished stream.

436-- streams as it needs to retain state for each unfinished stream.

464--

437--

465newtype CoserialT m a = CoserialT {getCoserialT :: Stream m a}

438newtype CoserialT m a = CoserialT {getCoserialT :: Stream m a}

466− deriving (Monoid, MonadTrans, MonadIO, MonadThrow)

439+ deriving (Functor, MonadTrans)

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)

474

440

475{-# DEPRECATED InterleavedT "Please use 'CoserialT' instead." #-}

441{-# DEPRECATED InterleavedT "Please use 'CoserialT' instead." #-}

476type InterleavedT = CoserialT

442type InterleavedT = CoserialT

@@ -500,6 +466,14 @@ infixr 5 <=>

500(<=>) :: IsStream t => t m a -> t m a -> t m a

466(<=>) :: IsStream t => t m a -> t m a -> t m a

501(<=>) = coserial

467(<=>) = coserial

502

468

469+------------------------------------------------------------------------------

470+-- Monoid

471+------------------------------------------------------------------------------

472+

473+instance Monoid (CoserialT m a) where

474+ mempty = nil

475+ mappend = (<>)

476+

503------------------------------------------------------------------------------

477------------------------------------------------------------------------------

504-- Monad

478-- Monad

505------------------------------------------------------------------------------

479------------------------------------------------------------------------------

@@ -513,66 +487,11 @@ instance Monad m => Monad (CoserialT m) where

513 in m Nothing stp single yield

487 in m Nothing stp single yield

514

488

515------------------------------------------------------------------------------

489------------------------------------------------------------------------------

516−-- Applicative

490+-- Other instances

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

535------------------------------------------------------------------------------

491------------------------------------------------------------------------------

536

492

537−instance (Monad m, Num a) => Num (CoserialT m a) where

493+MONAD_APPLICATIVE_INSTANCE(CoserialT,)

538− fromInteger n = pure (fromInteger n)

494+MONAD_COMMON_INSTANCES(CoserialT,)

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

576

495

577------------------------------------------------------------------------------

496------------------------------------------------------------------------------

578-- CoparallelT

497-- CoparallelT

@@ -618,17 +537,7 @@ instance (Monad m, Floating a) => Floating (CoserialT m a) where

618-- as it explores only a bounded number of streams at a time.

537-- as it explores only a bounded number of streams at a time.

619--

538--

620newtype CoparallelT m a = CoparallelT {getCoparallelT :: Stream m a}

539newtype CoparallelT m a = CoparallelT {getCoparallelT :: Stream m a}

621− deriving (MonadTrans)

540+ deriving (Functor, 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)

632

541

633{-# DEPRECATED AsyncT "Please use 'CoparallelT' instead." #-}

542{-# DEPRECATED AsyncT "Please use 'CoparallelT' instead." #-}

634type AsyncT = CoparallelT

543type AsyncT = CoparallelT

@@ -657,6 +566,14 @@ instance MonadParallel m => Semigroup (CoparallelT m a) where

657(<|) :: (IsStream t, MonadParallel m) => t m a -> t m a -> t m a

566(<|) :: (IsStream t, MonadParallel m) => t m a -> t m a -> t m a

658(<|) = coparallel

567(<|) = coparallel

659

568

569+------------------------------------------------------------------------------

570+-- Monoid

571+------------------------------------------------------------------------------

572+

573+instance MonadParallel m => Monoid (CoparallelT m a) where

574+ mempty = nil

575+ mappend = (<>)

576+

660------------------------------------------------------------------------------

577------------------------------------------------------------------------------

661-- Monad

578-- Monad

662------------------------------------------------------------------------------

579------------------------------------------------------------------------------

@@ -682,66 +599,11 @@ instance MonadParallel m => Monad (CoparallelT m) where

682 where par = S.joinStreamVar2 (SVarStyle Conjunction LIFO)

599 where par = S.joinStreamVar2 (SVarStyle Conjunction LIFO)

683

600

684------------------------------------------------------------------------------

601------------------------------------------------------------------------------

685−-- Applicative

602+-- Other instances

686------------------------------------------------------------------------------

603------------------------------------------------------------------------------

687

604

688−instance MonadParallel m => Applicative (CoparallelT m) where

605+MONAD_APPLICATIVE_INSTANCE(CoparallelT,MONADPARALLEL)

689− pure = CoparallelT . singleton

606+MONAD_COMMON_INSTANCES(CoparallelT, MONADPARALLEL)

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

745

607

746------------------------------------------------------------------------------

608------------------------------------------------------------------------------

747-- ParallelT

609-- ParallelT

@@ -785,17 +647,7 @@ instance (MonadParallel m, Floating a) => Floating (CoparallelT m a) where

785-- streams as it needs to retain state for each unfinished stream.

647-- streams as it needs to retain state for each unfinished stream.

786--

648--

787newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}

649newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}

788− deriving (MonadTrans)

650+ deriving (Functor, 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)

799

651

800instance IsStream ParallelT where

652instance IsStream ParallelT where

801 toStream = getParallelT

653 toStream = getParallelT

@@ -814,6 +666,14 @@ parallel m1 m2 = fromStream $ S.parallel (toStream m1) (toStream m2)

814instance MonadParallel m => Semigroup (ParallelT m a) where

666instance MonadParallel m => Semigroup (ParallelT m a) where

815 (<>) = parallel

667 (<>) = parallel

816

668

669+------------------------------------------------------------------------------

670+-- Monoid

671+------------------------------------------------------------------------------

672+

673+instance MonadParallel m => Monoid (ParallelT m a) where

674+ mempty = nil

675+ mappend = (<>)

676+

817------------------------------------------------------------------------------

677------------------------------------------------------------------------------

818-- Monad

678-- Monad

819------------------------------------------------------------------------------

679------------------------------------------------------------------------------

@@ -824,66 +684,11 @@ instance MonadParallel m => Monad (ParallelT m) where

824 where par = S.joinStreamVar2 (SVarStyle Conjunction FIFO)

684 where par = S.joinStreamVar2 (SVarStyle Conjunction FIFO)

825

685

826------------------------------------------------------------------------------

686------------------------------------------------------------------------------

827−-- Applicative

687+-- Other instances

828−------------------------------------------------------------------------------

829−

830−instance MonadParallel m => Applicative (ParallelT m) where

831− pure = ParallelT . singleton

832− (<*>) = ap

833−

834−------------------------------------------------------------------------------

835−-- Functor

836------------------------------------------------------------------------------

688------------------------------------------------------------------------------

837

689

838−instance Monad m => Functor (ParallelT m) where

690+MONAD_APPLICATIVE_INSTANCE(ParallelT,MONADPARALLEL)

839− fmap f m = fromStream $ Stream $ \_ stp sng yld ->

691+MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)

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

887

692

888------------------------------------------------------------------------------

693------------------------------------------------------------------------------

889-- Serially Zipping Streams

694-- Serially Zipping Streams

@@ -920,20 +725,12 @@ zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

920-- 'SerialT'.

725-- 'SerialT'.

921--

726--

922newtype ZipSerial m a = ZipSerial {getZipSerial :: Stream m a}

727newtype ZipSerial m a = ZipSerial {getZipSerial :: Stream m a}

923− deriving (Semigroup, Monoid)

728+ deriving (Functor, Semigroup, Monoid)

924

729

925{-# Deprecated ZipStream "Please use ZipSerial instead." #-}

730{-# Deprecated ZipStream "Please use ZipSerial instead." #-}

926-- | Same as ZipSerial.

731-- | Same as ZipSerial.

927type ZipStream = ZipSerial

732type ZipStream = ZipSerial

928

733

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−

937instance Monad m => Applicative (ZipSerial m) where

734instance Monad m => Applicative (ZipSerial m) where

938 pure = ZipSerial . S.repeat

735 pure = ZipSerial . S.repeat

939 (<*>) = zipWith id

736 (<*>) = zipWith id

@@ -942,46 +739,6 @@ instance IsStream ZipSerial where

942 toStream = getZipSerial

739 toStream = getZipSerial

943 fromStream = ZipSerial

740 fromStream = ZipSerial

944

741

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−

985------------------------------------------------------------------------------

742------------------------------------------------------------------------------

986-- Parallely Zipping Streams

743-- Parallely Zipping Streams

987------------------------------------------------------------------------------

744------------------------------------------------------------------------------

@@ -1017,19 +774,11 @@ zipAsyncWith = zipParallelWith

1017-- 'SerialT'.

774-- 'SerialT'.

1018--

775--

1019newtype ZipParallel m a = ZipParallel {getZipParallel :: Stream m a}

776newtype ZipParallel m a = ZipParallel {getZipParallel :: Stream m a}

1020− deriving (Semigroup, Monoid)

777+ deriving (Functor, Semigroup, Monoid)

1021

778

1022{-# DEPRECATED ZipAsync "Please use ZipParallel instead." #-}

779{-# DEPRECATED ZipAsync "Please use ZipParallel instead." #-}

1023type ZipAsync = ZipParallel

780type ZipAsync = ZipParallel

1024

781

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−

1033instance MonadParallel m => Applicative (ZipParallel m) where

782instance MonadParallel m => Applicative (ZipParallel m) where

1034 pure = ZipParallel . S.repeat

783 pure = ZipParallel . S.repeat

1035 (<*>) = zipParallelWith id

784 (<*>) = zipParallelWith id

@@ -1038,46 +787,6 @@ instance IsStream ZipParallel where

1038 toStream = getZipParallel

787 toStream = getZipParallel

1039 fromStream = ZipParallel

788 fromStream = ZipParallel

1040

789

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−

1081-------------------------------------------------------------------------------

790-------------------------------------------------------------------------------

1082-- Type adapting combinators

791-- Type adapting combinators

1083-------------------------------------------------------------------------------

792-------------------------------------------------------------------------------

stack.yaml+2−2

@@ -25,5 +25,5 @@ flags: {}

25extra-package-dbs: []

26rebuild-ghc-options: true

27# For mac ports installed SDL library on Mac OS X

28−#extra-include-dirs:

28+extra-include-dirs:

29−#- /opt/local/include

29+- /opt/local/include

streamly.cabal+3−3

@@ -149,9 +149,9 @@ test-suite test

149 , base >= 4.8 && < 5

150 , hspec >= 2.0 && < 3

151 , containers >= 0.5 && < 0.6

152− if impl(ghc < 8.0)

153− build-depends:

154− transformers >= 0.4 && < 0.6

152+ , transformers >= 0.4 && < 0.6

153+ , mtl >= 2.2 && < 3

154+ , exceptions >= 0.8 && < 0.11

155 default-language: Haskell2010

156

157test-suite properties

test/Main.hs+138−55

@@ -5,6 +5,10 @@

5module Main (main) where

7import Control.Concurrent (threadDelay)

8+import Control.Exception (Exception, try)

9+import Control.Monad.Catch (throwM, MonadThrow)

10+import Control.Monad.Error.Class (throwError, MonadError)

11+import Control.Monad.Trans.Except (runExceptT, ExceptT)

8import Data.Foldable (forM_)

12import Data.Foldable (forM_)

9import Data.List (sort)

13import Data.List (sort)

10import Test.Hspec

14import Test.Hspec

@@ -32,6 +36,7 @@ main :: IO ()

32main = hspec $ do

36main = hspec $ do

33 describe "Runners" $ do

37 describe "Runners" $ do

34 -- XXX move these to property tests

38 -- XXX move these to property tests

39+ -- XXX use an IORef to store and check the side effects

35 it "simple serially" $

40 it "simple serially" $

36 (runStream . serially) (return (0 :: Int)) `shouldReturn` ()

41 (runStream . serially) (return (0 :: Int)) `shouldReturn` ()

37 it "simple serially with IO" $

42 it "simple serially with IO" $

@@ -40,10 +45,10 @@ main = hspec $ do

40 describe "Empty" $ do

45 describe "Empty" $ do

41 it "Monoid - mempty" $

46 it "Monoid - mempty" $

42 (toListSerial mempty) `shouldReturn` ([] :: [Int])

47 (toListSerial mempty) `shouldReturn` ([] :: [Int])

43− it "Alternative - empty" $

48+ -- it "Alternative - empty" $

44− (toListSerial empty) `shouldReturn` ([] :: [Int])

49+ -- (toListSerial empty) `shouldReturn` ([] :: [Int])

45− it "MonadPlus - mzero" $

50+ -- it "MonadPlus - mzero" $

46− (toListSerial mzero) `shouldReturn` ([] :: [Int])

51+ -- (toListSerial mzero) `shouldReturn` ([] :: [Int])

48 ---------------------------------------------------------------------------

53 ---------------------------------------------------------------------------

49 -- Functor

54 -- Functor

@@ -84,7 +89,26 @@ main = hspec $ do

84 `shouldReturn` ([(1,2),(1,3)] :: [(Int, Int)])

89 `shouldReturn` ([(1,2),(1,3)] :: [(Int, Int)])

86 ---------------------------------------------------------------------------

91 ---------------------------------------------------------------------------

92+ -- Semigroup/Monoidal Composition strict ordering checks

93+ ---------------------------------------------------------------------------

94+

95+ -- test both (<>) and mappend to make sure we are using correct instance

96+ -- for Monoid that is using the right version of semigroup. Instance

97+ -- deriving can cause us to pick wrong instances sometimes.

98+

99+ describe "Serial interleaved (<>) ordering check" $ interleaveCheck coserially (<>)

100+ describe "Serial interleaved mappend ordering check" $ interleaveCheck coserially mappend

101+

102+ describe "Parallel interleaved (<>) ordering check" $ interleaveCheck parallely (<>)

103+ describe "Parallel interleaved mappend ordering check" $ interleaveCheck parallely mappend

104+

105+ describe "Coparallel (<>) time order check" $ parallelCheck coparallely (<>)

106+ describe "Coparallel mappend time order check" $ parallelCheck coparallely mappend

107+ describe "Parallel (<>) time order check" $ parallelCheck parallely (<>)

108+ describe "Parallel mappend time order check" $ parallelCheck parallely mappend

109+

110+ ---------------------------------------------------------------------------

87− -- Monoidal Compositions

111+ -- Monoidal Compositions, multiset equality checks

88 ---------------------------------------------------------------------------

112 ---------------------------------------------------------------------------

113

90 describe "Serial Composition" $ compose serially mempty id

114 describe "Serial Composition" $ compose serially mempty id

@@ -94,14 +118,6 @@ main = hspec $ do

94 describe "Semigroup Composition for ZipSerial" $ compose zipSerially mempty id

118 describe "Semigroup Composition for ZipSerial" $ compose zipSerially mempty id

95 describe "Semigroup Composition for ZipAsync" $ compose zipParallely mempty id

119 describe "Semigroup Composition for ZipAsync" $ compose zipParallely mempty id

96 -- XXX need to check alternative compositions as well

120 -- XXX need to check alternative compositions as well

97− ---------------------------------------------------------------------------

98− -- Monoidal Composition ordering checks

99− ---------------------------------------------------------------------------

100−

101− describe "Serial interleaved ordering check" $ interleaveCheck coserially

102− describe "Parallel interleaved ordering check" $ interleaveCheck parallely

103− describe "Left biased parallel time order check" $ parallelCheck coparallely

104− describe "Fair parallel time order check" $ parallelCheck parallely

105

121

106 ---------------------------------------------------------------------------

122 ---------------------------------------------------------------------------

107 -- TBD Monoidal composition combinations

123 -- TBD Monoidal composition combinations

@@ -261,17 +277,108 @@ main = hspec $ do

261 it "Nests two streams using applicative async composition" nestTwoAsyncApp

277 it "Nests two streams using applicative async composition" nestTwoAsyncApp

262 it "Nests two streams using applicative parallel composition" nestTwoParallelApp

278 it "Nests two streams using applicative parallel composition" nestTwoParallelApp

263

279

264− it "Nests two streams using Num serial composition" nestTwoSerialNum

265− it "Nests two streams using Num interleaved composition" nestTwoInterleavedNum

266− it "Nests two streams using Num async composition" nestTwoAsyncNum

267− it "Nests two streams using Num parallel composition" nestTwoParallelNum

268−

269 ---------------------------------------------------------------------------

280 ---------------------------------------------------------------------------

270 -- TBD Bind and Bind combinations

281 -- TBD Bind and Bind combinations

271 ---------------------------------------------------------------------------

282 ---------------------------------------------------------------------------

272

283

273 -- TBD combine all binds and all compose in one example

284 -- TBD combine all binds and all compose in one example

274 describe "Miscellaneous combined examples" mixedOps

285 describe "Miscellaneous combined examples" mixedOps

286+ describe "Simple MonadError and MonadThrow" simpleMonadError

287+

288+ {-

289+ describe "Composed MonadError serially" $ composeWithMonadError serially

290+ describe "Composed MonadError coserially" $ composeWithMonadError coserially

291+ describe "Composed MonadError coparallely" $ composeWithMonadError coparallely

292+ describe "Composed MonadError parallely" $ composeWithMonadError parallely

293+ -}

294+

295+ describe "Composed MonadThrow serially" $ composeWithMonadThrow serially

296+ describe "Composed MonadThrow coserially" $ composeWithMonadThrow coserially

297+ describe "Composed MonadThrow coparallely" $ composeWithMonadThrow coparallely

298+ describe "Composed MonadThrow parallely" $ composeWithMonadThrow parallely

299+

300+-- XXX need to test that we have promptly cleaned up everything after the error

301+-- XXX We can also check the output that we are expected to get before the

302+-- error occurs.

303+

304+data ExampleException = ExampleException String deriving (Eq, Show)

305+

306+instance Exception ExampleException

307+

308+simpleMonadError :: Spec

309+simpleMonadError = do

310+{-

311+ it "simple runExceptT" $ do

312+ (runExceptT $ runStream $ return ())

313+ `shouldReturn` (Right () :: Either String ())

314+ it "simple runExceptT with error" $ do

315+ (runExceptT $ runStream $ throwError "E") `shouldReturn` Left "E"

316+ -}

317+ it "simple try" $ do

318+ (try $ runStream $ return ())

319+ `shouldReturn` (Right () :: Either ExampleException ())

320+ it "simple try with throw error" $ do

321+ (try $ runStream $ throwM $ ExampleException "E")

322+ `shouldReturn` (Left (ExampleException "E") :: Either ExampleException ())

323+

324+composeWithMonadThrow

325+ :: ( IsStream t

326+ , Semigroup (t IO Int)

327+ , MonadThrow (t IO)

328+ )

329+ => (t IO Int -> SerialT IO Int) -> Spec

330+composeWithMonadThrow t = do

331+ it "Compose throwM, nil" $

332+ (try $ tl (throwM (ExampleException "E") <> A.nil))

333+ `shouldReturn` (Left (ExampleException "E") :: Either ExampleException [Int])

334+ it "Compose nil, throwM" $

335+ (try $ tl (A.nil <> throwM (ExampleException "E")))

336+ `shouldReturn` (Left (ExampleException "E") :: Either ExampleException [Int])

337+ oneLevelNestedSum "serially" serially

338+ oneLevelNestedSum "coserially" coserially

339+ oneLevelNestedSum "coparallely" coparallely

340+ oneLevelNestedSum "parallely" parallely

341+ -- XXX add two level nesting

342+

343+ oneLevelNestedProduct "serially" serially

344+ oneLevelNestedProduct "coserially" coserially

345+ oneLevelNestedProduct "coparallely" coparallely

346+ oneLevelNestedProduct "parallely" parallely

347+

348+ where

349+ tl = A.toList . t

350+ oneLevelNestedSum desc t1 =

351+ it ("One level nested sum " ++ desc) $ do

352+ let nested = (A.fromFoldable [1..10] <> throwM (ExampleException "E")

353+ <> A.fromFoldable [1..10])

354+ (try $ tl (A.nil <> t1 nested <> A.fromFoldable [1..10]))

355+ `shouldReturn` (Left (ExampleException "E") :: Either ExampleException [Int])

356+

357+ oneLevelNestedProduct desc t1 =

358+ it ("One level nested product" ++ desc) $ do

359+ let s1 = t $ foldMapWith (<>) return [1..4]

360+ s2 = t1 $ foldMapWith (<>) return [5..8]

361+ try $ tl (do

362+ x <- adapt s1

363+ y <- s2

364+ if (x + y > 10)

365+ then throwM (ExampleException "E")

366+ else return (x + y)

367+ )

368+ `shouldReturn` (Left (ExampleException "E") :: Either ExampleException [Int])

369+

370+_composeWithMonadError

371+ :: ( IsStream t

372+ , Semigroup (t (ExceptT String IO) Int)

373+ , MonadError String (t (ExceptT String IO))

374+ )

375+ => (t (ExceptT String IO) Int -> SerialT (ExceptT String IO) Int) -> Spec

376+_composeWithMonadError t = do

377+ let tl = A.toList . t

378+ it "Compose throwError, nil" $

379+ (runExceptT $ tl (throwError "E" <> A.nil)) `shouldReturn` Left "E"

380+ it "Compose nil, error" $

381+ (runExceptT $ tl (A.nil <> throwError "E")) `shouldReturn` Left "E"

275

382

276nestTwoSerial :: Expectation

383nestTwoSerial :: Expectation

277nestTwoSerial =

384nestTwoSerial =

@@ -290,13 +397,6 @@ nestTwoSerialApp =

290 in toListSerial ((+) <$> s1 <*> s2)

397 in toListSerial ((+) <$> s1 <*> s2)

291 `shouldReturn` ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])

398 `shouldReturn` ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])

292

399

293−nestTwoSerialNum :: Expectation

294−nestTwoSerialNum =

295− let s1 = foldMapWith (<>) return [1..4]

296− s2 = foldMapWith (<>) return [5..8]

297− in toListSerial (s1 + s2)

298− `shouldReturn` ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])

299−

300nestTwoInterleaved :: Expectation

400nestTwoInterleaved :: Expectation

301nestTwoInterleaved =

401nestTwoInterleaved =

302 let s1 = foldMapWith (<>) return [1..4]

402 let s1 = foldMapWith (<>) return [1..4]

@@ -314,13 +414,6 @@ nestTwoInterleavedApp =

314 in toListInterleaved ((+) <$> s1 <*> s2)

414 in toListInterleaved ((+) <$> s1 <*> s2)

315 `shouldReturn` ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])

415 `shouldReturn` ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])

316

416

317−nestTwoInterleavedNum :: Expectation

318−nestTwoInterleavedNum =

319− let s1 = foldMapWith (<>) return [1..4]

320− s2 = foldMapWith (<>) return [5..8]

321− in toListInterleaved (s1 + s2)

322− `shouldReturn` ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])

323−

324nestTwoAsync :: Expectation

417nestTwoAsync :: Expectation

325nestTwoAsync =

418nestTwoAsync =

326 let s1 = foldMapWith (<>) return [1..4]

419 let s1 = foldMapWith (<>) return [1..4]

@@ -339,13 +432,6 @@ nestTwoAsyncApp =

339 in (toListAsync ((+) <$> s1 <*> s2) >>= return . sort)

432 in (toListAsync ((+) <$> s1 <*> s2) >>= return . sort)

340 `shouldReturn` sort ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])

433 `shouldReturn` sort ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])

341

434

342−nestTwoAsyncNum :: Expectation

343−nestTwoAsyncNum =

344− let s1 = foldMapWith (<>) return [1..4]

345− s2 = foldMapWith (<>) return [5..8]

346− in (toListAsync (s1 + s2) >>= return . sort)

347− `shouldReturn` sort ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])

348−

349nestTwoParallel :: Expectation

435nestTwoParallel :: Expectation

350nestTwoParallel =

436nestTwoParallel =

351 let s1 = foldMapWith (<>) return [1..4]

437 let s1 = foldMapWith (<>) return [1..4]

@@ -364,32 +450,29 @@ nestTwoParallelApp =

364 in (toListParallel ((+) <$> s1 <*> s2) >>= return . sort)

450 in (toListParallel ((+) <$> s1 <*> s2) >>= return . sort)

365 `shouldReturn` sort ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])

451 `shouldReturn` sort ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])

366

452

367−nestTwoParallelNum :: Expectation

368−nestTwoParallelNum =

369− let s1 = foldMapWith (<>) return [1..4]

370− s2 = foldMapWith (<>) return [5..8]

371− in (toListParallel (s1 + s2) >>= return . sort)

372− `shouldReturn` sort ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])

373−

374timed :: MonadIO (t IO) => Int -> t IO Int

453timed :: MonadIO (t IO) => Int -> t IO Int

375timed x = liftIO (threadDelay (x * 100000)) >> return x

454timed x = liftIO (threadDelay (x * 100000)) >> return x

376

455

377−interleaveCheck :: (IsStream t, Semigroup (t IO Int))

456+interleaveCheck :: IsStream t

378− => (t IO Int -> SerialT IO Int) -> Spec

457+ => (t IO Int -> SerialT IO Int)

458+ -> (t IO Int -> t IO Int -> t IO Int)

459+ -> Spec

379−interleaveCheck t =

460+interleaveCheck t f =

380 it "Interleave four" $

461 it "Interleave four" $

381− (A.toList . t) ((singleton 0 <> singleton 1) <> (singleton 100 <> singleton 101))

462+ (A.toList . t) ((singleton 0 `f` singleton 1) `f` (singleton 100 `f` singleton 101))

382 `shouldReturn` ([0, 100, 1, 101])

463 `shouldReturn` ([0, 100, 1, 101])

383

464

384−parallelCheck :: (Semigroup (t IO Int), MonadIO (t IO))

465+parallelCheck :: MonadIO (t IO)

385− => (t IO Int -> SerialT IO Int) -> Spec

466+ => (t IO Int -> SerialT IO Int)

467+ -> (t IO Int -> t IO Int -> t IO Int)

468+ -> Spec

386−parallelCheck t = do

469+parallelCheck t f = do

387 it "Parallel ordering left associated" $

470 it "Parallel ordering left associated" $

388− (A.toList . t) (((event 4 <> event 3) <> event 2) <> event 1)

471+ (A.toList . t) (((event 4 `f` event 3) `f` event 2) `f` event 1)

389 `shouldReturn` ([1..4])

472 `shouldReturn` ([1..4])

390

473

391 it "Parallel ordering right associated" $

474 it "Parallel ordering right associated" $

392− (A.toList . t) (event 4 <> (event 3 <> (event 2 <> event 1)))

475+ (A.toList . t) (event 4 `f` (event 3 `f` (event 2 `f` event 1)))

393 `shouldReturn` ([1..4])

476 `shouldReturn` ([1..4])

394

477

395 where event n = (liftIO $ threadDelay (n * 100000)) >> (return n)

478 where event n = (liftIO $ threadDelay (n * 100000)) >> (return n)