0717458 · streamly · git.iko.soy

139-- >>> import qualified Streamly.Data.Fold as Fold

140-- >>> import qualified Streamly.Data.Parser as Parser

141-- >>> import qualified Streamly.Data.Stream as Stream

142−-- >>> import qualified Streamly.Internal.Data.Stream.Concurrent as Async

142+-- >>> import qualified Streamly.Internal.Data.Stream.Concurrent as Concur

143-- >>> import Prelude hiding (concatMap, concat, zipWith)

144-- >>> :{

145-- delay n = do

185-- | Evaluate a stream asynchronously using a channel and serve the consumer

186-- from the evaluation buffer.

187--

188−-- >>> eval = Async.evalWith id

188+-- >>> eval = Concur.evalWith id

189--

190{-# INLINE eval #-}

191eval :: MonadAsync m => Stream m a -> Stream m a

270-- all streams. However, with this operation you can precisely control the

271-- scheduling by creating arbitrary shape expression trees.

272--

273−-- >>> append2 = Async.combineWith id

273+-- >>> append2 = Concur.combineWith id

274--

275-- The following code finishes in 4 seconds:

276--

277-- >>> stream1 = Stream.fromEffect (delay 4)

278-- >>> stream2 = Stream.fromEffect (delay 2)

279−-- >>> Stream.fold Fold.toList $ stream1 `Async.append2` stream2

279+-- >>> Stream.fold Fold.toList $ stream1 `Concur.append2` stream2

280-- 2 sec

281-- 4 sec

282-- [2,4]

289-- the left stream. This schedules all streams in a round robin fashion over

290-- limited number of threads.

291--

292−-- >>> interleave2 = Async.combineWith (Async.interleaved True)

292+-- >>> interleave2 = Concur.combineWith (Concur.interleaved True)

293--

294{-# INLINE interleave2 #-}

295interleave2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

297

298-- | Like 'append' but with 'ordered' on.

299--

300−-- >>> ahead2 = Async.combineWith (Async.ordered True)

300+-- >>> ahead2 = Concur.combineWith (Concur.ordered True)

301--

302{-# INLINE ahead2 #-}

303ahead2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

305

306-- | Like 'append2' but with 'eager' on.

307--

308−-- >>> parallel2 = Async.combineWith (Async.eager True)

308+-- >>> parallel2 = Concur.combineWith (Concur.eager True)

309--

310{-# INLINE parallel2 #-}

311parallel2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

313

314-- | Like 'parallel2' but stops the output as soon as the first stream stops.

315--

316−-- >>> parallelFst2 = Async.combineWith (Async.eager True . Async.stopWhen Async.FirstStops)

316+-- >>> parallelFst2 = Concur.combineWith (Concur.eager True . Concur.stopWhen Concur.FirstStops)

317--

318-- /Pre-release/

319{-# INLINE parallelFst2 #-}

323-- | Like 'parallel2' but stops the output as soon as any of the two streams

324-- stops.

325--

326−-- >>> parallelMin2 = Async.combineWith (Async.eager True . Async.stopWhen Async.AnyStops)

326+-- >>> parallelMin2 = Concur.combineWith (Concur.eager True . Concur.stopWhen Concur.AnyStops)

327--

328-- /Pre-release/

329{-# INLINE parallelMin2 #-}

440-- concurrently but earlier streams are perferred. Output from the streams are

441-- used as they arrive.

442--

443−-- >>> concatMap = Async.concatMapWith id

443+-- >>> concatMap = Concur.concatMapWith id

444--

445−-- >>> f cfg xs = Stream.fold Fold.toList $ Async.concatMapWith cfg id $ Stream.fromList xs

445+-- >>> f cfg xs = Stream.fold Fold.toList $ Concur.concatMapWith cfg id $ Stream.fromList xs

446--

447-- The following streams finish in 4 seconds:

448--

458-- Limiting threads to 2 schedules the third stream only after one of the first

459-- two has finished, releasing a thread:

460--

461−-- >>> f (Async.maxThreads 2) [stream1, stream2, stream3]

461+-- >>> f (Concur.maxThreads 2) [stream1, stream2, stream3]

462-- ...

463-- [2,1,4]

464--

465-- When used with a Single thread it behaves like serial concatMap:

466--

467−-- >>> f (Async.maxThreads 1) [stream1, stream2, stream3]

467+-- >>> f (Concur.maxThreads 1) [stream1, stream2, stream3]

468-- ...

469-- [4,2,1]

470--

471-- >>> stream1 = Stream.fromList [1,2,3]

472-- >>> stream2 = Stream.fromList [4,5,6]

473−-- >>> f (Async.maxThreads 1) [stream1, stream2]

473+-- >>> f (Concur.maxThreads 1) [stream1, stream2]

474-- [1,2,3,4,5,6]

475--

476{-# INLINE concatMap #-}

482-- evaluate the earlier stream first, this schedules all streams in a round

483-- robin fashion over the available threads.

484--

485−-- >>> concatMapInterleave = Async.concatMapWith (Async.interleaved True)

485+-- >>> concatMapInterleave = Concur.concatMapWith (Concur.interleaved True)

486--

487-- When used with a single thread it behaves like serial interleaving:

488--

489−-- >>> f cfg xs = Stream.fold Fold.toList $ Async.concatMapWith (Async.interleaved True . cfg) id $ Stream.fromList xs

489+-- >>> f cfg xs = Stream.fold Fold.toList $ Concur.concatMapWith (Concur.interleaved True . cfg) id $ Stream.fromList xs

490--

491-- >>> stream1 = Stream.fromList [1,2,3]

492-- >>> stream2 = Stream.fromList [4,5,6]

493−-- >>> f (Async.maxThreads 1) [stream1, stream2]

493+-- >>> f (Concur.maxThreads 1) [stream1, stream2]

494-- [1,4,2,5,3,6]

495--

496-- /Works only on finite number of streams/

502-- | Like 'concat' but we can also specify the concurrent channel's

503-- configuration parameters using Config modifiers.

504--

505−-- >>> concatWith modifier = Async.concatMapWith modifier id

505+-- >>> concatWith modifier = Concur.concatMapWith modifier id

506--

507{-# INLINE concatWith #-}

508concatWith :: MonadAsync m =>

511

512-- | Evaluate the streams in the input stream concurrently and combine them.

513--

514−-- >>> concat = Async.concatWith id

514+-- >>> concat = Concur.concatWith id

515--

516{-# INLINE concat #-}

517concat :: MonadAsync m => Stream m (Stream m a) -> Stream m a

523

524-- | Like 'concatWith' but works on a list of streams.

525--

526−-- >>> concatListWith modifier = Async.concatWith modifier . Stream.fromList

526+-- >>> concatListWith modifier = Concur.concatWith modifier . Stream.fromList

527--

528{-# INLINE concatListWith #-}

529concatListWith :: MonadAsync m => (Config -> Config) -> [Stream m a] -> Stream m a

531

532-- | Like 'concat' but works on a list of streams.

533--

534−-- >>> append = Async.concatListWith id

534+-- >>> append = Concur.concatListWith id

535--

536{-# INLINE append #-}

537append :: MonadAsync m => [Stream m a] -> Stream m a

541-- the left stream. This schedules all streams in a round robin fashion over

542-- limited number of threads.

543--

544−-- >>> interleave = Async.concatListWith (Async.interleaved True)

544+-- >>> interleave = Concur.concatListWith (Concur.interleaved True)

545--

546{-# INLINE interleave #-}

547interleave :: MonadAsync m => [Stream m a] -> Stream m a

549

550-- | Like 'append' but with 'ordered' on.

551--

552−-- >>> ahead = Async.concatListWith (Async.ordered True)

552+-- >>> ahead = Concur.concatListWith (Concur.ordered True)

553--

554{-# INLINE ahead #-}

555ahead :: MonadAsync m => [Stream m a] -> Stream m a

557

558-- | Like 'append' but with 'eager' on.

559--

560−-- >>> parallel = Async.concatListWith (Async.eager True)

560+-- >>> parallel = Concur.concatListWith (Concur.eager True)

561--

562{-# INLINE parallel #-}

563parallel :: MonadAsync m => [Stream m a] -> Stream m a

565

566-- | Like 'parallel' but stops the output as soon as the first stream stops.

567--

568−-- >>> parallelFst = Async.concatListWith (Async.eager True . Async.stopWhen Async.FirstStops)

568+-- >>> parallelFst = Concur.concatListWith (Concur.eager True . Concur.stopWhen Concur.FirstStops)

569--

570{-# INLINE parallelFst #-}

571parallelFst :: MonadAsync m => [Stream m a] -> Stream m a

574-- | Like 'parallel' but stops the output as soon as any of the two streams

575-- stops.

576--

577−-- >>> parallelMin = Async.concatListWith (Async.eager True . Async.stopWhen Async.AnyStops)

577+-- >>> parallelMin = Concur.concatListWith (Concur.eager True . Concur.stopWhen Concur.AnyStops)

578--

579{-# INLINE parallelMin #-}

580parallelMin :: MonadAsync m => [Stream m a] -> Stream m a

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

606

607-- |

608−-- >>> mapMWith modifier f = Async.concatMapWith modifier (Stream.fromEffect . f)

608+-- >>> mapMWith modifier f = Concur.concatMapWith modifier (Stream.fromEffect . f)

609--

610{-# INLINE mapMWith #-}

611mapMWith :: MonadAsync m =>

613mapMWith modifier f = concatMapWith modifier (Stream.fromEffect . f)

614

615-- |

616−-- >>> mapM = Async.mapMWith id

616+-- >>> mapM = Concur.mapMWith id

617{-# INLINE mapM #-}

618mapM :: MonadAsync m => (a -> m b) -> Stream m a -> Stream m b

619mapM = mapMWith id

620

621-- |

622−-- >>> sequenceWith modifier = Async.mapMWith modifier id

622+-- >>> sequenceWith modifier = Concur.mapMWith modifier id

623--

624{-# INLINE sequenceWith #-}

625sequenceWith :: MonadAsync m =>

627sequenceWith modifier = mapMWith modifier id

628

629-- |

630−-- >>> sequence = Async.sequenceWith id

630+-- >>> sequence = Concur.sequenceWith id

631--

632{-# INLINE sequence #-}

633sequence :: MonadAsync m =>

635sequence = sequenceWith id

636

637-- |

638−-- >>> zipWithM f m1 m2 = Stream.zipWithM f (Async.eval m1) (Async.eval m2)

638+-- >>> zipWithM f m1 m2 = Stream.zipWithM f (Concur.eval m1) (Concur.eval m2)

639--

640{-# INLINE zipWithM #-}

641zipWithM :: MonadAsync m

647--

648-- >>> m1 = Stream.fromList [1,2,3]

649-- >>> m2 = Stream.fromList [4,5,6]

650−-- >>> Stream.fold Fold.toList $ Async.zipWith (,) m1 m2

650+-- >>> Stream.fold Fold.toList $ Concur.zipWith (,) m1 m2

651-- [(1,4),(2,5),(3,6)]

652--

653{-# INLINE zipWith #-}