34cbc34 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Define functor instances for Zip types instead of deriving

34cbc34Harendra Kumar9 years ago

src/Asyncly/AsyncT.hs+60−62

compare with master

unifiedsplit

@@ -32,6 +32,7 @@ module Asyncly.AsyncT

32 , yielding

33 , interleave

34 , parallel

35+ , parAlt

35 , parLeft

36 , parLeft

36 , CtxType (..)

37 , CtxType (..)

37 , SchedPolicy (..)

38 , SchedPolicy (..)

@@ -157,37 +158,7 @@ interleave m1 m2 = Stream $ \_ stp yld -> do

157 (runStream m1) Nothing stop yield

158 (runStream m1) Nothing stop yield

158

159

159------------------------------------------------------------------------------

160------------------------------------------------------------------------------

160−-- Functor

161+-- Streaming threads for parallel composition

161−------------------------------------------------------------------------------

162−

163−instance Monad m => Functor (Stream m) where

164− fmap = liftM

165−

166−------------------------------------------------------------------------------

167−-- Applicative

168−------------------------------------------------------------------------------

169−

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

171− pure = yielding

172− (<*>) = ap

173−

174−------------------------------------------------------------------------------

175−-- Monad

176−------------------------------------------------------------------------------

177−

178−-- | Execute a monadic action sequentially for each element in the 'AsyncT'

179−-- stream, i.e. an iteration finishes completely before the next one starts.

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

181− return = pure

182−

183− Stream m >>= f = Stream $ \_ stp yld ->

184− let run x = (runStream x) Nothing stp yld

185− yield a Nothing = run $ f a

186− yield a (Just r) = run $ f a <> (r >>= f)

187− in m Nothing stp yield

188−

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

190−-- Alternative

191------------------------------------------------------------------------------

162------------------------------------------------------------------------------

192

163

193{-# INLINE doFork #-}

164{-# INLINE doFork #-}

@@ -468,19 +439,28 @@ parallel ct m1 m2 = Stream $ \ctx stp yld -> do

468 (runStream (pushPullFork ct m1 m2)) Nothing stp yld

439 (runStream (pushPullFork ct m1 m2)) Nothing stp yld

469 Just c -> liftIO ((enqueue c) m2) >> (runStream m1) ctx stp yld

440 Just c -> liftIO ((enqueue c) m2) >> (runStream m1) ctx stp yld

470

441

471−-- | `empty` represents an action that takes non-zero time to complete. Since

442+-- | The computation specified in the argument is pushed to a new thread and

472−-- all actions take non-zero time, an `Alternative` composition ('<|>') is a

443+-- the context is returned. The returned context can be used to pull the output

473−-- monoidal composition executing all actions in parallel, it is similar to

444+-- of the computation.

474−-- '<>' except that it runs all the actions in parallel and interleaves their

475−-- results fairly.

476−instance MonadAsync m => Alternative (Stream m) where

445+pushOneToCtx :: MonadAsync m => CtxType -> Stream m a -> m (Context m a)

477− empty = mempty

446+pushOneToCtx ctype m = do

478−

447+ ctx <- liftIO $

479− {-# INLINE (<|>) #-}

448+ case ctype of

480− (<|>) = parallel (CtxType Disjunction FIFO)

449+ CtxType _ FIFO -> do

450+ c <- getFifoCtx ctype

451+ return c

452+ CtxType _ LIFO -> do

453+ c <- getLifoCtx ctype

454+ return c

455+ -- Note: We must have all the work on the queue before sending the

456+ -- pushworker, otherwise the pushworker may exit before we even get a

457+ -- chance to push.

458+ liftIO $ (enqueue ctx) m

459+ pushWorker ctx

460+ return ctx

481

461

482------------------------------------------------------------------------------

462------------------------------------------------------------------------------

483−-- Other monoidal compositions for parallel actions

463+-- Semigroup and Monoid style compositions for parallel actions

484------------------------------------------------------------------------------

464------------------------------------------------------------------------------

485

465

486{-

466{-

@@ -495,36 +475,54 @@ parAhead = undefined

495(<>|) = parAhead

475(<>|) = parAhead

496-}

476-}

497

477

478+-- | Same as '<|>'.

479+{-# INLINE parAlt #-}

480+parAlt :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

481+parAlt = parallel (CtxType Disjunction FIFO)

482+

498-- | Same as '<|'.

483-- | Same as '<|'.

499{-# INLINE parLeft #-}

484{-# INLINE parLeft #-}

500parLeft :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

485parLeft :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

501parLeft = parallel (CtxType Disjunction LIFO)

486parLeft = parallel (CtxType Disjunction LIFO)

502

487

488+-------------------------------------------------------------------------------

489+-- Stream type is not exposed, these instances are only for deriving instances

490+-- for the newtype wrappers based on Stream.

491+-------------------------------------------------------------------------------

492+

493+-- Dummy Instances, defined to enable the definition of other instances that

494+-- require a Monad constraint. Must be defined by the newtypes.

495+

496+instance Monad m => Functor (Stream m) where

497+ fmap = undefined

498+

499+instance Monad m => Applicative (Stream m) where

500+ pure = undefined

501+ (<*>) = undefined

502+

503+instance Monad m => Monad (Stream m) where

504+ return = pure

505+ (>>=) = undefined

506+

507+------------------------------------------------------------------------------

508+-- Alternative & MonadPlus

509+------------------------------------------------------------------------------

510+

511+-- | `empty` represents an action that takes non-zero time to complete. Since

512+-- all actions take non-zero time, an `Alternative` composition ('<|>') is a

513+-- monoidal composition executing all actions in parallel, it is similar to

514+-- '<>' except that it runs all the actions in parallel and interleaves their

515+-- results fairly.

516+instance MonadAsync m => Alternative (Stream m) where

517+ empty = mempty

518+ (<|>) = parAlt

519+

503instance MonadAsync m => MonadPlus (Stream m) where

520instance MonadAsync m => MonadPlus (Stream m) where

504 mzero = empty

521 mzero = empty

505 mplus = (<|>)

522 mplus = (<|>)

506

523

507−-- | The computation specified in the argument is pushed to a new thread and

508−-- the context is computation pulls output from the thread.

509−pushOneToCtx :: MonadAsync m => CtxType -> Stream m a -> m (Context m a)

510−pushOneToCtx ctype m = do

511− ctx <- liftIO $

512− case ctype of

513− CtxType _ FIFO -> do

514− c <- getFifoCtx ctype

515− return c

516− CtxType _ LIFO -> do

517− c <- getLifoCtx ctype

518− return c

519− -- Note: We must have all the work on the queue before sending the

520− -- pushworker, otherwise the pushworker may exit before we even get a

521− -- chance to push.

522− liftIO $ (enqueue ctx) m

523− pushWorker ctx

524− return ctx

525−

526-------------------------------------------------------------------------------

524-------------------------------------------------------------------------------

527−-- AsyncT transformer

525+-- Transformer

528-------------------------------------------------------------------------------

526-------------------------------------------------------------------------------

529

527

530instance MonadTrans Stream where

528instance MonadTrans Stream where

src/Asyncly/Prelude.hs+148−44

@@ -19,6 +19,7 @@

19module Asyncly.Prelude

20 (

21 -- * Product Style Composition

22+ -- $product

22 StreamT

23 StreamT

23 , InterleavedT

24 , InterleavedT

24 , AsyncT

25 , AsyncT

@@ -66,13 +67,58 @@ import Data.Semigroup (Semigroup(..), cycle1)

66import Prelude hiding (drop, take, zipWith)

67import Prelude hiding (drop, take, zipWith)

67import Asyncly.AsyncT

68import Asyncly.AsyncT

70+-- $product

71+--

72+-- Different styles of streams dictate the manner in which a stream is

73+-- processed in monadic or applicative composition. Each style is represented

74+-- by a different type. The types have the same underlying representation and

75+-- therefore can be converted back and forth easily to switch the composition

76+-- style at any point. In other words, processing order of the elements or

77+-- processing parallelism is decided by the type of the stream.

78+--

79+-- 'StreamT' and 'InterleavedT' process the streams serially but differ in the

80+-- order in which the elements of the streams are processed. 'StreamT' goes

81+-- depth first that is it combine one element of the first stream to every

82+-- element of the second stream before it comes back to process th second

83+-- element of the first stream. 'InterleavedT' takes turns when processing

84+-- elements from each stream and therefore all iterations finish at the same

85+-- time rather than one iteration at a time as in the case of 'StreamT'.

86+-- Depending on the cache behavior or IO behavior 'InterleavedT' might be

87+-- slightly more efficient than 'StreamT' or vice-versa.

88+--

89+-- 'AsyncT' and 'ParallelT' can process the individual elements of the streams

90+-- in parallel. 'AsyncT' is the parallel version of 'StreamT', it follows the

91+-- order of 'StreamT' processing multiple elements of a stream concurrently.

92+-- 'ParallelT' follows the order of 'InterleavedT' processing and interleaving

93+-- both streams concurrently. Also, note that 'fmap' on these streams would

94+-- also work in a concurrent fashion i.e. multiple elements in a stream or

95+-- multiple streams can be mapped concurrently.

96+--

97+-- Zipping types have a different style of composition altogether, they zip the

98+-- streams rather than multiplying them. These types provide only Applicative

99+-- composition and no Monad instances. 'ZipStream' zips serially i.e. it

100+-- produces one element from each stream in a serial manner and then zips the

101+-- two items. 'ZipAsync' on the other hand zips in parallel, it produces one

102+-- element from each stream concurrently and then zips the two.

103+--

104+-- It may help develop a better understanding if we think in terms of stream

105+-- generation (production) side parallelism versus stream processing

106+-- (consumption) side parallelism. The sum style operators (e.g. '<|' or '<|>')

107+-- generate streams in parallel and then combine the results into a unified

108+-- stream. On the other hand the product style composition (e.g. 'AsyncT' or

109+-- 'ParallelT' or 'ZipAsync') dictate how the elements of a stream are

110+-- processed when it is consumed.

111+

69------------------------------------------------------------------------------

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

70-- StreamT iterates serially in the Monad and Applicative compositions

113-- StreamT iterates serially in the Monad and Applicative compositions

71------------------------------------------------------------------------------

114------------------------------------------------------------------------------

115

116+-- | Note that 'StreamT' style composition works well operationally for

117+-- infinite streams. However if the second stream is infinite we will never

118+-- come back to the second iteration of the first stream.

119+

73newtype StreamT m a = StreamT {getStreamT :: Stream m a}

120newtype StreamT m a = StreamT {getStreamT :: Stream m a}

74− deriving (Functor, Applicative, Monad, Semigroup, Monoid,

75− MonadTrans, MonadIO, MonadThrow)

121+ deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)

122

77deriving instance MonadAsync m => Alternative (StreamT m)

123deriving instance MonadAsync m => Alternative (StreamT m)

78deriving instance MonadAsync m => MonadPlus (StreamT m)

124deriving instance MonadAsync m => MonadPlus (StreamT m)

@@ -90,6 +136,36 @@ instance Streaming StreamT where

90-- from a single base type because they depend on the Monad instance which is

136-- from a single base type because they depend on the Monad instance which is

91-- different for each type.

137-- different for each type.

138

139+------------------------------------------------------------------------------

140+-- Monad

141+------------------------------------------------------------------------------

142+

143+-- | Execute a monadic action sequentially for each element in the 'AsyncT'

144+-- stream, i.e. an iteration finishes completely before the next one starts.

145+instance Monad m => Monad (StreamT m) where

146+ return = pure

147+ (StreamT (Stream m)) >>= f = StreamT $ Stream $ \_ stp yld ->

148+ let run x = (runStream x) Nothing stp yld

149+ yield a Nothing = run $ getStreamT (f a)

150+ yield a (Just r) = run $ getStreamT (f a)

151+ <> getStreamT (StreamT r >>= f)

152+ in m Nothing stp yield

153+

154+------------------------------------------------------------------------------

155+-- Applicative

156+------------------------------------------------------------------------------

157+

158+instance Monad m => Applicative (StreamT m) where

159+ pure = StreamT . yielding

160+ (<*>) = ap

161+

162+------------------------------------------------------------------------------

163+-- Functor

164+------------------------------------------------------------------------------

165+

166+instance Monad m => Functor (StreamT m) where

167+ fmap = liftM

168+

93------------------------------------------------------------------------------

169------------------------------------------------------------------------------

94-- Num

170-- Num

95------------------------------------------------------------------------------

171------------------------------------------------------------------------------

@@ -136,9 +212,13 @@ instance (Monad m, Floating a) => Floating (StreamT m a) where

136

212

137------------------------------------------------------------------------------

213------------------------------------------------------------------------------

138-- InterleavedT interleaves the iterations in the Monad and Applicative

214-- InterleavedT interleaves the iterations in the Monad and Applicative

139−-- compositions. It does not make much sense for infinite streams.

215+-- compositions.

140------------------------------------------------------------------------------

216------------------------------------------------------------------------------

141

217

218+-- | Note that 'InterleavedT' style composition cannot work well operationally

219+-- on infinite streams. It will keep accumulating the elements of the stream

220+-- resulting in continuous increase in space consumption.

221+--

142newtype InterleavedT m a = InterleavedT {getInterleavedT :: Stream m a}

222newtype InterleavedT m a = InterleavedT {getInterleavedT :: Stream m a}

143 deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)

223 deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)

144

224

@@ -162,16 +242,9 @@ instance Monad m => Monad (InterleavedT m) where

162 yield a Nothing = run $ getInterleavedT (f a)

242 yield a Nothing = run $ getInterleavedT (f a)

163 yield a (Just r) = run $ getInterleavedT (f a)

243 yield a (Just r) = run $ getInterleavedT (f a)

164 `interleave`

244 `interleave`

165− getInterleavedT ((InterleavedT r) >>= f)

245+ getInterleavedT (InterleavedT r >>= f)

166 in m Nothing stp yield

246 in m Nothing stp yield

167

247

168−------------------------------------------------------------------------------

169−-- Functor

170−------------------------------------------------------------------------------

171−

172−instance Monad m => Functor (InterleavedT m) where

173− fmap = liftM

174−

175------------------------------------------------------------------------------

248------------------------------------------------------------------------------

176-- Applicative

249-- Applicative

177------------------------------------------------------------------------------

250------------------------------------------------------------------------------

@@ -180,6 +253,13 @@ instance Monad m => Applicative (InterleavedT m) where

180 pure = InterleavedT . yielding

253 pure = InterleavedT . yielding

181 (<*>) = ap

254 (<*>) = ap

182

255

256+------------------------------------------------------------------------------

257+-- Functor

258+------------------------------------------------------------------------------

259+

260+instance Monad m => Functor (InterleavedT m) where

261+ fmap = liftM

262+

183------------------------------------------------------------------------------

263------------------------------------------------------------------------------

184-- Num

264-- Num

185------------------------------------------------------------------------------

265------------------------------------------------------------------------------

@@ -230,14 +310,20 @@ instance (Monad m, Floating a) => Floating (InterleavedT m a) where

230-- with infinite streams.

310-- with infinite streams.

231------------------------------------------------------------------------------

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

232

312

313+-- | Just like 'StreamT', 'AsyncT' style composition works well operationally

314+-- for infinite streams. However, if the second stream is infinite we will

315+-- never be able to finish the first stream.

316+--

317+-- Note that 'fmap' on 'AsyncT' would map on multiple elements concurrently.

318+

233newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}

319newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}

234 deriving (Semigroup, Monoid, MonadTrans)

320 deriving (Semigroup, Monoid, MonadTrans)

235

321

322+deriving instance MonadAsync m => Alternative (AsyncT m)

323+deriving instance MonadAsync m => MonadPlus (AsyncT m)

236deriving instance MonadAsync m => MonadIO (AsyncT m)

324deriving instance MonadAsync m => MonadIO (AsyncT m)

237deriving instance MonadAsync m => MonadThrow (AsyncT m)

325deriving instance MonadAsync m => MonadThrow (AsyncT m)

238deriving instance (MonadBase b m, MonadAsync m) => MonadBase b (AsyncT m)

326deriving instance (MonadBase b m, MonadAsync m) => MonadBase b (AsyncT m)

239−deriving instance MonadAsync m => Alternative (AsyncT m)

240−deriving instance MonadAsync m => MonadPlus (AsyncT m)

241deriving instance (MonadError e m, MonadAsync m) => MonadError e (AsyncT m)

327deriving instance (MonadError e m, MonadAsync m) => MonadError e (AsyncT m)

242deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (AsyncT m)

328deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (AsyncT m)

243deriving instance (MonadState s m, MonadAsync m) => MonadState s (AsyncT m)

329deriving instance (MonadState s m, MonadAsync m) => MonadState s (AsyncT m)

@@ -270,13 +356,6 @@ instance MonadAsync m => Monad (AsyncT m) where

270 where g x = getAsyncT (f x)

356 where g x = getAsyncT (f x)

271 par = parallel (CtxType Conjunction LIFO)

357 par = parallel (CtxType Conjunction LIFO)

272

358

273−------------------------------------------------------------------------------

274−-- Functor

275−------------------------------------------------------------------------------

276−

277−instance MonadAsync m => Functor (AsyncT m) where

278− fmap = liftM

279−

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

359------------------------------------------------------------------------------

281-- Applicative

360-- Applicative

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

361------------------------------------------------------------------------------

@@ -285,6 +364,13 @@ instance MonadAsync m => Applicative (AsyncT m) where

285 pure = AsyncT . yielding

364 pure = AsyncT . yielding

286 (<*>) = ap

365 (<*>) = ap

287

366

367+------------------------------------------------------------------------------

368+-- Functor

369+------------------------------------------------------------------------------

370+

371+instance MonadAsync m => Functor (AsyncT m) where

372+ fmap = liftM

373+

288------------------------------------------------------------------------------

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

289-- Num

375-- Num

290------------------------------------------------------------------------------

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

@@ -331,18 +417,24 @@ instance (MonadAsync m, Floating a) => Floating (AsyncT m a) where

331

417

332------------------------------------------------------------------------------

418------------------------------------------------------------------------------

333-- ParallelT runs all iterations in parallel in the Monad and Applicative

419-- ParallelT runs all iterations in parallel in the Monad and Applicative

334−-- compositions. Iterations are scheduled in a round-robin fashion. It cannot

420+-- compositions. Iterations are scheduled in a round-robin fashion.

335−-- work well when the streams being composed are infinite.

336------------------------------------------------------------------------------

421------------------------------------------------------------------------------

337

422

423+-- | Note that, just like 'InterleavedT', 'ParallelT' style composition cannot

424+-- work well operationally on infinite streams. It will keep accumulating the

425+-- elements of the stream resulting in continuous increase in space

426+-- consumption.

427+--

428+-- Note that 'fmap' on 'ParallelT' would map on both streams concurrently.

429+--

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

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

339 deriving (Semigroup, Monoid, MonadTrans)

431 deriving (Semigroup, Monoid, MonadTrans)

340

432

433+deriving instance MonadAsync m => Alternative (ParallelT m)

434+deriving instance MonadAsync m => MonadPlus (ParallelT m)

341deriving instance MonadAsync m => MonadIO (ParallelT m)

435deriving instance MonadAsync m => MonadIO (ParallelT m)

342deriving instance MonadAsync m => MonadThrow (ParallelT m)

436deriving instance MonadAsync m => MonadThrow (ParallelT m)

343deriving instance (MonadBase b m, MonadAsync m) => MonadBase b (ParallelT m)

437deriving instance (MonadBase b m, MonadAsync m) => MonadBase b (ParallelT m)

344−deriving instance MonadAsync m => Alternative (ParallelT m)

345−deriving instance MonadAsync m => MonadPlus (ParallelT m)

346deriving instance (MonadError e m, MonadAsync m) => MonadError e (ParallelT m)

438deriving instance (MonadError e m, MonadAsync m) => MonadError e (ParallelT m)

347deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (ParallelT m)

439deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (ParallelT m)

348deriving instance (MonadState s m, MonadAsync m) => MonadState s (ParallelT m)

440deriving instance (MonadState s m, MonadAsync m) => MonadState s (ParallelT m)

@@ -360,13 +452,6 @@ instance MonadAsync m => Monad (ParallelT m) where

360 where g x = getParallelT (f x)

452 where g x = getParallelT (f x)

361 par = parallel (CtxType Conjunction FIFO)

453 par = parallel (CtxType Conjunction FIFO)

362

454

363−------------------------------------------------------------------------------

364−-- Functor

365−------------------------------------------------------------------------------

366−

367−instance MonadAsync m => Functor (ParallelT m) where

368− fmap = liftM

369−

370------------------------------------------------------------------------------

455------------------------------------------------------------------------------

371-- Applicative

456-- Applicative

372------------------------------------------------------------------------------

457------------------------------------------------------------------------------

@@ -375,6 +460,13 @@ instance MonadAsync m => Applicative (ParallelT m) where

375 pure = ParallelT . yielding

460 pure = ParallelT . yielding

376 (<*>) = ap

461 (<*>) = ap

377

462

463+------------------------------------------------------------------------------

464+-- Functor

465+------------------------------------------------------------------------------

466+

467+instance MonadAsync m => Functor (ParallelT m) where

468+ fmap = liftM

469+

378------------------------------------------------------------------------------

470------------------------------------------------------------------------------

379-- Num

471-- Num

380------------------------------------------------------------------------------

472------------------------------------------------------------------------------

@@ -451,13 +543,19 @@ zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

451 yield1 a (Just ra) = merge a ra

543 yield1 a (Just ra) = merge a ra

452 (runStream mx) Nothing stp yield1

544 (runStream mx) Nothing stp yield1

453

545

454−-- | Wrapper around AsyncT type with a serial zipping Applicative instance.

546+-- | 'ZipStream' zips serially i.e. it produces one element from each stream in

455−-- Note that the binary function interleave (\<=>) is a special case of

547+-- a serial manner and then zips the two elements.

456−-- ZipStream Applicative.

457−--

458−-- > f <$> ZipStream xs1 <*> ... <*> ZipStream xsN

459newtype ZipStream m a = ZipStream {getZipStream :: Stream m a}

548newtype ZipStream m a = ZipStream {getZipStream :: Stream m a}

460− deriving (Functor)

549+ deriving (Semigroup, Monoid)

550+

551+deriving instance MonadAsync m => Alternative (ZipStream m)

552+

553+instance Monad m => Functor (ZipStream m) where

554+ fmap f (ZipStream (Stream m)) = ZipStream $ Stream $ \_ stp yld ->

555+ let yield a Nothing = yld (f a) Nothing

556+ yield a (Just r) = yld (f a)

557+ (Just (getZipStream (fmap f (ZipStream r))))

558+ in m Nothing stp yield

461

559

462instance Monad m => Applicative (ZipStream m) where

560instance Monad m => Applicative (ZipStream m) where

463 pure a = ZipStream $ cycle1 (pure a)

561 pure a = ZipStream $ cycle1 (pure a)

@@ -508,7 +606,7 @@ instance (Monad m, Floating a) => Floating (ZipStream m a) where

508 logBase = liftA2 logBase

606 logBase = liftA2 logBase

509

607

510------------------------------------------------------------------------------

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

511−-- Asyncly Zipping Streams

609+-- Parallely Zipping Streams

512------------------------------------------------------------------------------

610------------------------------------------------------------------------------

513

611

514-- | Zip two AsyncT streams asyncly (i.e. both the streams are generated

612-- | Zip two AsyncT streams asyncly (i.e. both the streams are generated

@@ -529,13 +627,19 @@ zipAsyncWith f m1 m2 = fromStream $ Stream $ \_ stp yld -> do

529 mb <- async m2

627 mb <- async m2

530 (runStream (toStream (zipWith f ma mb))) Nothing stp yld

628 (runStream (toStream (zipWith f ma mb))) Nothing stp yld

531

629

532−-- | Wrapper around AsyncT type with a parallel zipping Applicative instance.

630+-- | 'ZipAsync' zips in parallel, it produces one element from each stream

533−-- Note that the binary operator (\<|>) from the Alternative instance of AsyncT

631+-- concurrently and then zips the two.

534−-- is a special case of ZipAsync Applicative.

535−--

536−-- > f <$> ZipAsync xs1 <*> ... <*> ZipAsync xsN

537newtype ZipAsync m a = ZipAsync {getZipAsync :: Stream m a}

632newtype ZipAsync m a = ZipAsync {getZipAsync :: Stream m a}

538− deriving (Functor)

633+ deriving (Semigroup, Monoid)

634+

635+deriving instance MonadAsync m => Alternative (ZipAsync m)

636+

637+instance Monad m => Functor (ZipAsync m) where

638+ fmap f (ZipAsync (Stream m)) = ZipAsync $ Stream $ \_ stp yld ->

639+ let yield a Nothing = yld (f a) Nothing

640+ yield a (Just r) = yld (f a)

641+ (Just (getZipAsync (fmap f (ZipAsync r))))

642+ in m Nothing stp yield

539

643

540instance MonadAsync m => Applicative (ZipAsync m) where

644instance MonadAsync m => Applicative (ZipAsync m) where

541 pure a = ZipAsync $ cycle1 (pure a)

645 pure a = ZipAsync $ cycle1 (pure a)

src/Asyncly/Tutorial.hs+13

@@ -585,3 +585,16 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

585

586-- $statemachine

587-- State machine stuff

588+

589+-- $comparison

590+--

591+-- Even though asyncly covers all that is provided by the 'async' package or

592+-- most of what is provided by the 'streaming', 'pipes' or 'conduit' packages,

593+-- I would not say that it renders those useless. Asyncly is like monad if

594+-- 'async' is applicative and monads and applicatives both have their use

595+-- cases. It can completely repalce 'async', the ZipAsync type is equivalent to

596+-- the functionality provided by 'async'.

597+--

598+-- pipes and conduit are like Arrows and asyncly is like monad to them. You

599+-- would use pipes and conduit when you do not need the product style

600+-- composition and the implicit concurrency.