9c85811 · streamly · git.iko.soy

git.iko.soy/streamly

rename "once" to "yieldM"

and some other related refactoring changes.

commit9c858119052d032f96f3f7e1eeb65b0bf23aff63 authorHarendra Kumar <harendra.kumar@gmail.com> dateSat 23 Jun 2018, 06:50 (8 years ago) parent5ee02ff

20 files changed+343−304

compare with master

unifiedsplit

Changelog.md+6

@@ -1,3 +1,9 @@

1+## Unreleased

3+### Deprecations

5+* `once` has been deprecated and renamed to `yieldM`

17## 0.3.0

39### Breaking changes

benchmark/LinearOps.hs+1−1

@@ -89,7 +89,7 @@ sourceFoldMapWith n = S.foldMapWith (S.<>) return [n..n+value]

8989{-# INLINE sourceFoldMapWithM #-}

9090sourceFoldMapWithM :: (S.IsStream t, Monad m, S.Semigroup (t m Int))

9191 => Int -> t m Int

92−sourceFoldMapWithM n = S.foldMapWith (S.<>) (S.once . return) [n..n+value]

92+sourceFoldMapWithM n = S.foldMapWith (S.<>) (S.yieldM . return) [n..n+value]

9393

9494{-# INLINE sourceUnfoldr #-}

9595sourceUnfoldr :: S.IsStream t => Int -> t m Int

examples/CirclingSquare.hs+3−3

@@ -9,7 +9,7 @@

99import Data.IORef

1010import Graphics.UI.SDL as SDL

1111import Streamly

12−import Streamly.Prelude (once)

12+import Streamly.Prelude (yieldM)

1313import Streamly.Time

1414

1515------------------------------------------------------------------------------

@@ -87,5 +87,5 @@ main :: IO ()

8787main = do

8888 sdlInit

8989 cref <- newIORef (0,0)

90− runStream $ once (updateController cref)

91− `parallel` once (updateDisplay cref)

90+ runStream $ yieldM (updateController cref)

91+ `parallel` yieldM (updateDisplay cref)

examples/MergeSort.hs+6−6

@@ -4,29 +4,29 @@ import Data.Word

44import System.Random (getStdGen, randoms)

55import Data.List (sort)

66import Streamly

7−import Streamly.Prelude (once)

7+import Streamly.Prelude (yieldM)

88import qualified Streamly.Prelude as A

1010getSorted :: Serial Word16

1111getSorted = do

12− g <- once getStdGen

12+ g <- yieldM getStdGen

1313 let ls = take 100000 (randoms g) :: [Word16]

1414 foldMap return (sort ls)

1515

1616-- | merge two streams generating the elements from each in parallel

1717mergeAsync :: Ord a => Serial a -> Serial a -> Serial a

1818mergeAsync a b = do

19− x <- once $ mkAsync a

20− y <- once $ mkAsync b

19+ x <- yieldM $ mkAsync a

20+ y <- yieldM $ mkAsync b

2121 merge x y

2222

2323merge :: Ord a => Serial a -> Serial a -> Serial a

2424merge a b = do

25− a1 <- once $ A.uncons a

25+ a1 <- yieldM $ A.uncons a

2626 case a1 of

2727 Nothing -> b

2828 Just (x, ma) -> do

29− b1 <- once $ A.uncons b

29+ b1 <- yieldM $ A.uncons b

3030 case b1 of

3131 Nothing -> return x <> ma

3232 Just (y, mb) ->

examples/SearchQuery.hs+4−3

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

11import Streamly

2−import Streamly.Prelude (nil, once, (|:))

2+import Streamly.Prelude (nil, yieldM, (|:))

33import Network.HTTP.Simple

55-- | Runs three search engine queries in parallel and prints the search engine

@@ -13,10 +13,11 @@ main = do

1313 runStream . parallely $ google |: bing |: duckduckgo |: nil

1414

1515 putStrLn "\nUsing parallel semigroup composition"

16− runStream . parallely $ once google <> once bing <> once duckduckgo

16+ runStream . parallely $ yieldM google <> yieldM bing <> yieldM duckduckgo

1717

1818 putStrLn "\nUsing parallel applicative zip"

19− runStream . zipAsyncly $ (,,) <$> once google <*> once bing <*> once duckduckgo

19+ runStream . zipAsyncly $

20+ (,,) <$> yieldM google <*> yieldM bing <*> yieldM duckduckgo

2021

2122 where

2223 get :: String -> IO ()

src/Streamly/Prelude.hs+62−61

@@ -53,7 +53,7 @@ module Streamly.Prelude

5353 -- * Special Generation

5454 -- | Generate a monadic stream from an input structure, a seed or a

5555 -- generation function.

56− , once

56+ , K.yieldM

5757 , replicateM

5858 , repeatM

5959 , iterate

@@ -121,6 +121,7 @@ module Streamly.Prelude

121121 , toHandle

122122

123123 -- * Deprecated

124+ , once

124125 , each

125126 , scan

126127 , foldl

@@ -217,7 +218,7 @@ unfoldrM step = go

217218 case mayb of

218219 Nothing -> stp

219220 Just (a, b) ->

220− K.runStream (toStream (return a |: go b)) svr stp sng yld

221+ K.unStream (toStream (return a |: go b)) svr stp sng yld

221222

222223-- | Construct a stream from a 'Foldable' containing pure values.

223224--

@@ -310,7 +311,7 @@ iterateM step = go

310311 where

311312 go s = fromStream $ K.Stream $ \svr stp sng yld -> do

312313 next <- step s

313− K.runStream (toStream (return s |: go next)) svr stp sng yld

314+ K.unStream (toStream (return s |: go next)) svr stp sng yld

314315

315316-- | Read lines from an IO Handle into a stream of Strings.

316317--

@@ -344,8 +345,8 @@ foldr step acc m = go (toStream m)

344345 go m1 =

345346 let stop = return acc

346347 single a = return (step a acc)

347− yield a r = go r >>= \b -> return (step a b)

348− in (K.runStream m1) Nothing stop single yield

348+ yieldk a r = go r >>= \b -> return (step a b)

349+ in (K.unStream m1) Nothing stop single yieldk

349350

350351-- | Lazy right fold with a monadic step function. For example, to fold a

351352-- stream into a list:

@@ -363,8 +364,8 @@ foldrM step acc m = go (toStream m)

363364 go m1 =

364365 let stop = return acc

365366 single a = step a acc

366− yield a r = go r >>= step a

367− in (K.runStream m1) Nothing stop single yield

367+ yieldk a r = go r >>= step a

368+ in (K.unStream m1) Nothing stop single yieldk

368369

369370-- | Strict left scan with an extraction function. Like 'scanl'', but applies a

370371-- user supplied extraction function (the third argument) at each step. This is

@@ -379,10 +380,10 @@ scanx step begin done m =

379380 where

380381 go m1 !acc = K.Stream $ \_ stp sng yld ->

381382 let single a = sng (done $ step acc a)

382− yield a r =

383+ yieldk a r =

383384 let s = step acc a

384385 in yld (done s) (go r s)

385− in K.runStream m1 Nothing stp single yield

386+ in K.unStream m1 Nothing stp single yieldk

386387

387388-- |

388389-- @since 0.1.1

@@ -413,7 +414,7 @@ foldx step begin done m = get $ go (toStream m) begin

413414 {-# NOINLINE get #-}

414415 get m1 =

415416 let single = return . done

416− in (K.runStream m1) Nothing undefined single undefined

417+ in (K.unStream m1) Nothing undefined single undefined

417418

418419 -- Note, this can be implemented by making a recursive call to "go",

419420 -- however that is more expensive because of unnecessary recursion

@@ -422,10 +423,10 @@ foldx step begin done m = get $ go (toStream m) begin

422423 go m1 !acc = K.Stream $ \_ _ sng yld ->

423424 let stop = sng acc

424425 single a = sng $ step acc a

425− yield a r =

426+ yieldk a r =

426427 let stream = go r (step acc a)

427− in (K.runStream stream) Nothing undefined sng yld

428− in (K.runStream m1) Nothing stop single yield

428+ in (K.unStream stream) Nothing undefined sng yld

429+ in (K.unStream m1) Nothing stop single yieldk

429430

430431-- |

431432-- @since 0.1.0

@@ -450,8 +451,8 @@ foldxM step begin done m = go begin (toStream m)

450451 go !acc m1 =

451452 let stop = acc >>= done

452453 single a = acc >>= \b -> step b a >>= done

453− yield a r = acc >>= \b -> go (step b a) r

454− in (K.runStream m1) Nothing stop single yield

454+ yieldk a r = acc >>= \b -> go (step b a) r

455+ in (K.unStream m1) Nothing stop single yieldk

455456

456457-- |

457458-- @since 0.1.0

@@ -474,8 +475,8 @@ uncons :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (a, t m a))

474475uncons m =

475476 let stop = return Nothing

476477 single a = return (Just (a, nil))

477− yield a r = return (Just (a, fromStream r))

478− in (K.runStream (toStream m)) Nothing stop single yield

478+ yieldk a r = return (Just (a, fromStream r))

479+ in (K.unStream (toStream m)) Nothing stop single yieldk

479480

480481-- | Write a stream of Strings to an IO Handle.

481482--

@@ -486,8 +487,8 @@ toHandle h m = go (toStream m)

486487 go m1 =

487488 let stop = return ()

488489 single a = liftIO (IO.hPutStrLn h a)

489− yield a r = liftIO (IO.hPutStrLn h a) >> go r

490− in (K.runStream m1) Nothing stop single yield

490+ yieldk a r = liftIO (IO.hPutStrLn h a) >> go r

491+ in (K.unStream m1) Nothing stop single yieldk

491492

492493------------------------------------------------------------------------------

493494-- Special folds

@@ -508,8 +509,8 @@ take :: IsStream t => Int -> t m a -> t m a

508509take n m = fromStream $ go n (toStream m)

509510 where

510511 go n1 m1 = K.Stream $ \_ stp sng yld ->

511− let yield a r = yld a (go (n1 - 1) r)

512− in if n1 <= 0 then stp else (K.runStream m1) Nothing stp sng yield

512+ let yieldk a r = yld a (go (n1 - 1) r)

513+ in if n1 <= 0 then stp else (K.unStream m1) Nothing stp sng yieldk

513514

514515-- | Include only those elements that pass a predicate.

515516--

@@ -521,9 +522,9 @@ filter p m = fromStream $ go (toStream m)

521522 go m1 = K.Stream $ \_ stp sng yld ->

522523 let single a | p a = sng a

523524 | otherwise = stp

524− yield a r | p a = yld a (go r)

525− | otherwise = (K.runStream r) Nothing stp single yield

526− in (K.runStream m1) Nothing stp single yield

525+ yieldk a r | p a = yld a (go r)

526+ | otherwise = (K.unStream r) Nothing stp single yieldk

527+ in (K.unStream m1) Nothing stp single yieldk

527528

528529-- | End the stream as soon as the predicate fails on an element.

529530--

@@ -535,9 +536,9 @@ takeWhile p m = fromStream $ go (toStream m)

535536 go m1 = K.Stream $ \_ stp sng yld ->

536537 let single a | p a = sng a

537538 | otherwise = stp

538− yield a r | p a = yld a (go r)

539+ yieldk a r | p a = yld a (go r)

539540 | otherwise = stp

540− in (K.runStream m1) Nothing stp single yield

541+ in (K.unStream m1) Nothing stp single yieldk

541542

542543-- | Discard first 'n' elements from the stream and take the rest.

543544--

@@ -547,11 +548,11 @@ drop n m = fromStream $ go n (toStream m)

547548 where

548549 go n1 m1 = K.Stream $ \_ stp sng yld ->

549550 let single _ = stp

550− yield _ r = (K.runStream $ go (n1 - 1) r) Nothing stp sng yld

551+ yieldk _ r = (K.unStream $ go (n1 - 1) r) Nothing stp sng yld

551552 -- Somehow "<=" check performs better than a ">"

552553 in if n1 <= 0

553− then (K.runStream m1) Nothing stp sng yld

554− else (K.runStream m1) Nothing stp single yield

554+ then (K.unStream m1) Nothing stp sng yld

555+ else (K.unStream m1) Nothing stp single yieldk

555556

556557-- | Drop elements in the stream as long as the predicate succeeds and then

557558-- take the rest of the stream.

@@ -564,9 +565,9 @@ dropWhile p m = fromStream $ go (toStream m)

564565 go m1 = K.Stream $ \_ stp sng yld ->

565566 let single a | p a = stp

566567 | otherwise = sng a

567− yield a r | p a = (K.runStream r) Nothing stp single yield

568+ yieldk a r | p a = (K.unStream r) Nothing stp single yieldk

568569 | otherwise = yld a r

569− in (K.runStream m1) Nothing stp single yield

570+ in (K.unStream m1) Nothing stp single yieldk

570571

571572-- | Determine whether all elements of a stream satisfy a predicate.

572573--

@@ -577,9 +578,9 @@ all p m = go (toStream m)

577578 go m1 =

578579 let single a | p a = return True

579580 | otherwise = return False

580− yield a r | p a = go r

581+ yieldk a r | p a = go r

581582 | otherwise = return False

582− in (K.runStream m1) Nothing (return True) single yield

583+ in (K.unStream m1) Nothing (return True) single yieldk

583584

584585-- | Determine whether any of the elements of a stream satisfy a predicate.

585586--

@@ -590,9 +591,9 @@ any p m = go (toStream m)

590591 go m1 =

591592 let single a | p a = return True

592593 | otherwise = return False

593− yield a r | p a = return True

594+ yieldk a r | p a = return True

594595 | otherwise = go r

595− in (K.runStream m1) Nothing (return False) single yield

596+ in (K.unStream m1) Nothing (return False) single yieldk

596597

597598-- | Determine the sum of all elements of a stream of numbers

598599--

@@ -613,8 +614,8 @@ head :: Monad m => SerialT m a -> m (Maybe a)

613614head m =

614615 let stop = return Nothing

615616 single a = return (Just a)

616− yield a _ = return (Just a)

617− in (K.runStream (toStream m)) Nothing stop single yield

617+ yieldk a _ = return (Just a)

618+ in (K.unStream (toStream m)) Nothing stop single yieldk

618619

619620-- | Extract all but the first element of the stream, if any.

620621--

@@ -623,8 +624,8 @@ tail :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (t m a))

623624tail m =

624625 let stop = return Nothing

625626 single _ = return $ Just nil

626− yield _ r = return $ Just $ fromStream r

627− in (K.runStream (toStream m)) Nothing stop single yield

627+ yieldk _ r = return $ Just $ fromStream r

628+ in (K.unStream (toStream m)) Nothing stop single yieldk

628629

629630-- | Extract the last element of the stream, if any.

630631--

@@ -640,8 +641,8 @@ null :: Monad m => SerialT m a -> m Bool

640641null m =

641642 let stop = return True

642643 single _ = return False

643− yield _ _ = return False

644− in (K.runStream (toStream m)) Nothing stop single yield

644+ yieldk _ _ = return False

645+ in (K.unStream (toStream m)) Nothing stop single yieldk

645646

646647-- | Determine whether an element is present in the stream.

647648--

@@ -652,8 +653,8 @@ elem e m = go (toStream m)

652653 go m1 =

653654 let stop = return False

654655 single a = return (a == e)

655− yield a r = if a == e then return True else go r

656− in (K.runStream m1) Nothing stop single yield

656+ yieldk a r = if a == e then return True else go r

657+ in (K.unStream m1) Nothing stop single yieldk

657658

658659-- | Determine whether an element is not present in the stream.

659660--

@@ -664,8 +665,8 @@ notElem e m = go (toStream m)

664665 go m1 =

665666 let stop = return True

666667 single a = return (a /= e)

667− yield a r = if a == e then return False else go r

668− in (K.runStream m1) Nothing stop single yield

668+ yieldk a r = if a == e then return False else go r

669+ in (K.unStream m1) Nothing stop single yieldk

669670

670671-- | Determine the length of the stream.

671672--

@@ -681,11 +682,11 @@ reverse :: (IsStream t) => t m a -> t m a

681682reverse m = fromStream $ go K.nil (toStream m)

682683 where

683684 go rev rest = K.Stream $ \_ stp sng yld ->

684− let runIt x = K.runStream x Nothing stp sng yld

685+ let runIt x = K.unStream x Nothing stp sng yld

685686 stop = runIt rev

686687 single a = runIt $ a `K.cons` rev

687− yield a r = runIt $ go (a `K.cons` rev) r

688− in K.runStream rest Nothing stop single yield

688+ yieldk a r = runIt $ go (a `K.cons` rev) r

689+ in K.unStream rest Nothing stop single yieldk

689690

690691-- XXX replace the recursive "go" with continuation

691692-- | Determine the minimum element in a stream.

@@ -697,8 +698,8 @@ minimum m = go Nothing (toStream m)

697698 go res m1 =

698699 let stop = return res

699700 single a = return $ min_ a res

700− yield a r = go (min_ a res) r

701− in (K.runStream m1) Nothing stop single yield

701+ yieldk a r = go (min_ a res) r

702+ in (K.unStream m1) Nothing stop single yieldk

702703

703704 min_ a res = case res of

704705 Nothing -> Just a

@@ -714,8 +715,8 @@ maximum m = go Nothing (toStream m)

714715 go res m1 =

715716 let stop = return res

716717 single a = return $ max_ a res

717− yield a r = go (max_ a res) r

718− in (K.runStream m1) Nothing stop single yield

718+ yieldk a r = go (max_ a res) r

719+ in (K.unStream m1) Nothing stop single yieldk

719720

720721 max_ a res = case res of

721722 Nothing -> Just a

@@ -745,8 +746,8 @@ mapM f m = go (toStream m)

745746 where

746747 go m1 = fromStream $ K.Stream $ \svr stp sng yld ->

747748 let single a = f a >>= sng

748− yield a r = K.runStream (toStream (f a |: (go r))) svr stp sng yld

749− in (K.runStream m1) Nothing stp single yield

749+ yieldk a r = K.unStream (toStream (f a |: (go r))) svr stp sng yld

750+ in (K.unStream m1) Nothing stp single yieldk

750751

751752-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'

752753-- elements, and return a stream of values extracted from 'Just'.

@@ -760,10 +761,10 @@ mapMaybe f m = go (toStream m)

760761 let single a = case f a of

761762 Just b -> sng b

762763 Nothing -> stp

763− yield a r = case f a of

764+ yieldk a r = case f a of

764765 Just b -> yld b (toStream $ go r)

765− Nothing -> (K.runStream r) Nothing stp single yield

766− in (K.runStream m1) Nothing stp single yield

766+ Nothing -> (K.unStream r) Nothing stp single yieldk

767+ in (K.unStream m1) Nothing stp single yieldk

767768

768769-- | Like 'mapMaybe' but maps a monadic function.

769770--

@@ -786,8 +787,8 @@ mapM_ f m = go (toStream m)

786787 go m1 =

787788 let stop = return ()

788789 single a = void (f a)

789− yield a r = f a >> go r

790− in (K.runStream m1) Nothing stop single yield

790+ yieldk a r = f a >> go r

791+ in (K.unStream m1) Nothing stop single yieldk

791792

792793-- | Reduce a stream of monadic actions to a stream of the output of those

793794-- actions.

@@ -808,5 +809,5 @@ sequence m = go (toStream m)

808809 where

809810 go m1 = fromStream $ K.Stream $ \svr stp sng yld ->

810811 let single ma = ma >>= sng

811− yield ma r = K.runStream (toStream $ ma |: go r) svr stp sng yld

812− in (K.runStream m1) Nothing stp single yield

812+ yieldk ma r = K.unStream (toStream $ ma |: go r) svr stp sng yld

813+ in (K.unStream m1) Nothing stp single yieldk

src/Streamly/Streams/Ahead.hs+27−27

@@ -141,9 +141,9 @@ workLoopAhead sv q heap = runHeap

141141 yieldOutput seqNo a r = do

142142 continue <- liftIO $ send sv (ChildYield a)

143143 if continue

144− then (runStream r) (Just sv) (runQueueToken seqNo)

145− (singleOutput seqNo)

146− (yieldOutput seqNo)

144+ then unStream r (Just sv) (runQueueToken seqNo)

145+ (singleOutput seqNo)

146+ (yieldOutput seqNo)

147147 else liftIO $ do

148148 atomicModifyIORefCAS_ heap $ \(h, _) ->

149149 (H.insert (Entry seqNo (AheadEntryStream r)) h, seqNo)

@@ -160,15 +160,15 @@ workLoopAhead sv q heap = runHeap

160160 Just (m, seqNo) -> do

161161 if seqNo == prevSeqNo + 1

162162 then

163− (runStream m) (Just sv) (runQueueToken seqNo)

164− (singleOutput seqNo)

165− (yieldOutput seqNo)

163+ unStream m (Just sv) (runQueueToken seqNo)

164+ (singleOutput seqNo)

165+ (yieldOutput seqNo)

166166 else do

167167 liftIO $ atomicModifyIORefCAS_ heap $ \(h, _) ->

168168 (h, prevSeqNo + 1)

169− (runStream m) (Just sv) runHeap

170− (singleToHeap seqNo)

171− (yieldToHeap seqNo)

169+ unStream m (Just sv) runHeap

170+ (singleToHeap seqNo)

171+ (yieldToHeap seqNo)

172172 runQueueNoToken = do

173173 work <- dequeueAhead q

174174 case work of

@@ -176,13 +176,13 @@ workLoopAhead sv q heap = runHeap

176176 Just (m, seqNo) -> do

177177 if seqNo == 0

178178 then

179− (runStream m) (Just sv) (runQueueToken seqNo)

180− (singleOutput seqNo)

181− (yieldOutput seqNo)

179+ unStream m (Just sv) (runQueueToken seqNo)

180+ (singleOutput seqNo)

181+ (yieldOutput seqNo)

182182 else

183− (runStream m) (Just sv) runHeap

184− (singleToHeap seqNo)

185− (yieldToHeap seqNo)

183+ unStream m (Just sv) runHeap

184+ (singleToHeap seqNo)

185+ (yieldToHeap seqNo)

186186

187187 {-# NOINLINE runHeap #-}

188188 runHeap = do

@@ -203,9 +203,9 @@ workLoopAhead sv q heap = runHeap

203203 case hent of

204204 AheadEntryPure a -> singleOutput seqNo a

205205 AheadEntryStream r ->

206− (runStream r) (Just sv) (runQueueToken seqNo)

207− (singleOutput seqNo)

208− (yieldOutput seqNo)

206+ unStream r (Just sv) (runQueueToken seqNo)

207+ (singleOutput seqNo)

208+ (yieldOutput seqNo)

209209

210210-------------------------------------------------------------------------------

211211-- WAhead

@@ -220,11 +220,11 @@ workLoopAhead sv q heap = runHeap

220220forkSVarAhead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

221221forkSVarAhead m1 m2 = Stream $ \_ stp sng yld -> do

222222 sv <- newAheadVar (concurrently m1 m2) workLoopAhead

223− (runStream (fromSVar sv)) Nothing stp sng yld

223+ unStream (fromSVar sv) Nothing stp sng yld

224224 where

225225 concurrently ma mb = Stream $ \svr stp sng yld -> do

226226 liftIO $ enqueue (fromJust svr) mb

227− (runStream ma) Nothing stp sng yld

227+ unStream ma Nothing stp sng yld

228228

229229{-# INLINE aheadS #-}

230230aheadS :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

@@ -235,14 +235,14 @@ aheadS m1 m2 = Stream $ \svr stp sng yld -> do

235235 -- Always run the left side on a new SVar to avoid complexity in

236236 -- sequencing results. This means the left side cannot further

237237 -- split into more ahead computations on the same SVar.

238− (runStream m1) Nothing stp sng yld

239− _ -> runStream (forkSVarAhead m1 m2) Nothing stp sng yld

238+ unStream m1 Nothing stp sng yld

239+ _ -> unStream (forkSVarAhead m1 m2) Nothing stp sng yld

240240

241241-- | XXX we can implement it more efficienty by directly implementing instead

242242-- of combining streams using ahead.

243243{-# INLINE consMAhead #-}

244244consMAhead :: MonadAsync m => m a -> Stream m a -> Stream m a

245−consMAhead m r = once m `aheadS` r

245+consMAhead m r = yieldM m `aheadS` r

246246

247247------------------------------------------------------------------------------

248248-- AheadT

@@ -356,10 +356,10 @@ aheadbind m f = go m

356356 where

357357 go (Stream g) =

358358 Stream $ \ctx stp sng yld ->

359− let runIt x = (runStream x) ctx stp sng yld

360− single a = runIt $ f a

361− yield a r = runIt $ f a `aheadS` go r

362− in g Nothing stp single yield

359+ let run x = unStream x ctx stp sng yld

360+ single a = run $ f a

361+ yieldk a r = run $ f a `aheadS` go r

362+ in g Nothing stp single yieldk

363363

364364instance MonadAsync m => Monad (AheadT m) where

365365 return = pure

src/Streamly/Streams/Async.hs+12−12

@@ -64,15 +64,15 @@ import Streamly.SVar

6464{-# INLINE runStreamLIFO #-}

6565runStreamLIFO :: MonadIO m

6666 => SVar Stream m a -> IORef [Stream m a] -> Stream m a -> m () -> m ()

67−runStreamLIFO sv q m stop = runStream m (Just sv) stop single yield

67+runStreamLIFO sv q m stop = unStream m (Just sv) stop single yieldk

6868 where

6969 single a = do

7070 res <- liftIO $ send sv (ChildYield a)

7171 if res then stop else liftIO $ sendStop sv

72− yield a r = do

72+ yieldk a r = do

7373 res <- liftIO $ send sv (ChildYield a)

7474 if res

75− then (runStream r) (Just sv) stop single yield

75+ then (unStream r) (Just sv) stop single yieldk

7676 else liftIO $ enqueueLIFO sv q r >> sendStop sv

7777

7878-------------------------------------------------------------------------------

@@ -82,12 +82,12 @@ runStreamLIFO sv q m stop = runStream m (Just sv) stop single yield

8282{-# INLINE runStreamFIFO #-}

8383runStreamFIFO :: MonadIO m

8484 => SVar Stream m a -> LinkedQueue (Stream m a) -> Stream m a -> m () -> m ()

85−runStreamFIFO sv q m stop = runStream m (Just sv) stop single yield

85+runStreamFIFO sv q m stop = unStream m (Just sv) stop single yieldk

8686 where

8787 single a = do

8888 res <- liftIO $ send sv (ChildYield a)

8989 if res then stop else liftIO $ sendStop sv

90− yield a r = do

90+ yieldk a r = do

9191 res <- liftIO $ send sv (ChildYield a)

9292 liftIO (enqueueFIFO sv q r)

9393 if res then stop else liftIO $ sendStop sv

@@ -278,11 +278,11 @@ forkSVarAsync style m1 m2 = Stream $ \_ stp sng yld -> do

278278 AsyncVar -> newAsyncVar (concurrently m1 m2)

279279 WAsyncVar -> newWAsyncVar (concurrently m1 m2)

280280 _ -> error "illegal svar type"

281− runStream (fromSVar sv) Nothing stp sng yld

281+ unStream (fromSVar sv) Nothing stp sng yld

282282 where

283283 concurrently ma mb = Stream $ \svr stp sng yld -> do

284284 liftIO $ enqueue (fromJust svr) mb

285− (runStream ma) svr stp sng yld

285+ unStream ma svr stp sng yld

286286

287287{-# INLINE joinStreamVarAsync #-}

288288joinStreamVarAsync :: MonadAsync m

@@ -290,8 +290,8 @@ joinStreamVarAsync :: MonadAsync m

290290joinStreamVarAsync style m1 m2 = Stream $ \svr stp sng yld ->

291291 case svr of

292292 Just sv | svarStyle sv == style ->

293− liftIO (enqueue sv m2) >> (runStream m1) svr stp sng yld

294− _ -> runStream (forkSVarAsync style m1 m2) Nothing stp sng yld

293+ liftIO (enqueue sv m2) >> unStream m1 svr stp sng yld

294+ _ -> unStream (forkSVarAsync style m1 m2) Nothing stp sng yld

295295

296296------------------------------------------------------------------------------

297297-- Semigroup and Monoid style compositions for parallel actions

@@ -323,7 +323,7 @@ async m1 m2 = fromStream $

323323-- of combining streams using async.

324324{-# INLINE consMAsync #-}

325325consMAsync :: MonadAsync m => m a -> Stream m a -> Stream m a

326−consMAsync m r = once m `asyncS` r

326+consMAsync m r = yieldM m `asyncS` r

327327

328328------------------------------------------------------------------------------

329329-- AsyncT

@@ -447,7 +447,7 @@ wAsyncS = joinStreamVarAsync WAsyncVar

447447-- of combining streams using wAsync.

448448{-# INLINE consMWAsync #-}

449449consMWAsync :: MonadAsync m => m a -> Stream m a -> Stream m a

450−consMWAsync m r = once m `wAsyncS` r

450+consMWAsync m r = yieldM m `wAsyncS` r

451451

452452-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WAsyncT'.

453453-- Merges two streams concurrently choosing elements from both fairly.

@@ -567,4 +567,4 @@ MONAD_COMMON_INSTANCES(WAsyncT, MONADPARALLEL)

567567-- @since 0.1.0

568568{-# DEPRECATED runAsyncT "Please use 'runStream . asyncly' instead." #-}

569569runAsyncT :: Monad m => AsyncT m a -> m ()

570−runAsyncT = run

570+runAsyncT = runStream

src/Streamly/Streams/Instances.hs+1−1

@@ -9,7 +9,7 @@

1010#define MONAD_APPLICATIVE_INSTANCE(STREAM,CONSTRAINT) \

1111instance (Monad m CONSTRAINT) => Applicative (STREAM m) where { \

12− pure = STREAM . singleton; \

12+ pure = STREAM . yield; \

1313 (<*>) = ap }

1414

1515#define MONAD_COMMON_INSTANCES(STREAM,CONSTRAINT) \

src/Streamly/Streams/Parallel.hs+8−8

@@ -58,7 +58,7 @@ import Streamly.SVar

5858

5959{-# NOINLINE runOne #-}

6060runOne :: MonadIO m => SVar Stream m a -> Stream m a -> m ()

61−runOne sv m = (runStream m) (Just sv) stop single yield

61+runOne sv m = (unStream m) (Just sv) stop single yieldk

6262

6363 where

6464

@@ -69,7 +69,7 @@ runOne sv m = (runStream m) (Just sv) stop single yield

6969 -- queue and queue it back on that and exit the thread when the outputQueue

7070 -- overflows. Parallel is dangerous because it can accumulate unbounded

7171 -- output in the buffer.

72− yield a r = void (sendit a) >> runOne sv r

72+ yieldk a r = void (sendit a) >> runOne sv r

7373

7474{-# NOINLINE forkSVarPar #-}

7575forkSVarPar :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

@@ -77,7 +77,7 @@ forkSVarPar m r = Stream $ \_ stp sng yld -> do

7777 sv <- newParallelVar

7878 pushWorkerPar sv (runOne sv m)

7979 pushWorkerPar sv (runOne sv r)

80− (runStream (fromSVar sv)) Nothing stp sng yld

80+ (unStream (fromSVar sv)) Nothing stp sng yld

8181

8282{-# INLINE joinStreamVarPar #-}

8383joinStreamVarPar :: MonadAsync m

@@ -85,8 +85,8 @@ joinStreamVarPar :: MonadAsync m

8585joinStreamVarPar style m1 m2 = Stream $ \svr stp sng yld ->

8686 case svr of

8787 Just sv | svarStyle sv == style -> do

88− pushWorkerPar sv (runOne sv m1) >> (runStream m2) svr stp sng yld

89− _ -> runStream (forkSVarPar m1 m2) Nothing stp sng yld

88+ pushWorkerPar sv (runOne sv m1) >> (unStream m2) svr stp sng yld

89+ _ -> unStream (forkSVarPar m1 m2) Nothing stp sng yld

9090

9191{-# INLINE parallelStream #-}

9292parallelStream :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

@@ -96,7 +96,7 @@ parallelStream = joinStreamVarPar ParallelVar

9696-- of combining streams using parallel.

9797{-# INLINE consMParallel #-}

9898consMParallel :: MonadAsync m => m a -> Stream m a -> Stream m a

99−consMParallel m r = once m `parallelStream` r

99+consMParallel m r = yieldM m `parallelStream` r

100100

101101-- | Polymorphic version of the 'Semigroup' operation '<>' of 'ParallelT'

102102-- Merges two streams concurrently.

@@ -125,7 +125,7 @@ applyWith :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b

125125applyWith f m = fromStream $ Stream $ \svr stp sng yld -> do

126126 sv <- newParallelVar

127127 pushWorkerPar sv (runOne sv (toStream m))

128− runStream (toStream $ f $ fromSVar sv) svr stp sng yld

128+ unStream (toStream $ f $ fromSVar sv) svr stp sng yld

129129

130130------------------------------------------------------------------------------

131131-- Stream runner concurrent function application

@@ -369,4 +369,4 @@ MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)

369369-- @since 0.1.0

370370{-# DEPRECATED runParallelT "Please use 'runStream . parallely' instead." #-}

371371runParallelT :: Monad m => ParallelT m a -> m ()

372−runParallelT = run

372+runParallelT = runStream

src/Streamly/Streams/Prelude.hs+20−66

@@ -21,13 +21,8 @@

2121--

2222module Streamly.Streams.Prelude

2323 (

24− -- * Construction

25− streamBuild

26− , fromCallback

27−

2824 -- * Elimination

29− , streamFold

30− , runStream

25+ runStream

3126 , runStreaming -- deprecated

3227

3328 -- * Fold Utilities

@@ -38,11 +33,28 @@ module Streamly.Streams.Prelude

3833where

3934

4035import Streamly.Streams.Serial (SerialT)

41−import Streamly.SVar (SVar)

42−

4336import Streamly.Streams.StreamK hiding (runStream)

4437import qualified Streamly.Streams.StreamK as K

4538

39+------------------------------------------------------------------------------

40+-- Eliminating a stream

41+------------------------------------------------------------------------------

42+

43+-- | Run a streaming composition, discard the results. By default it interprets

44+-- the stream as 'SerialT', to run other types of streams use the type adapting

45+-- combinators for example @runStream . 'asyncly'@.

46+--

47+-- @since 0.2.0

48+runStream :: Monad m => SerialT m a -> m ()

49+runStream = K.runStream

50+

51+-- | Same as 'runStream'

52+--

53+-- @since 0.1.0

54+{-# DEPRECATED runStreaming "Please use runStream instead." #-}

55+runStreaming :: (Monad m, IsStream t) => t m a -> m ()

56+runStreaming = runStream . adapt

57+

4658------------------------------------------------------------------------------

4759-- Fold Utilities

4860------------------------------------------------------------------------------

@@ -79,30 +91,6 @@ forEachWith :: (IsStream t, Foldable f)

7991 => (t m b -> t m b -> t m b) -> f a -> (a -> t m b) -> t m b

8092forEachWith f xs g = foldr (f . g) nil xs

8193

82−------------------------------------------------------------------------------

83−-- Generation

84−------------------------------------------------------------------------------

85−

86−-- | Build a stream from its church encoding. The function passed maps

87−-- directly to the underlying representation of the stream type. The second

88−-- parameter to the function is the "yield" function yielding a value and the

89−-- remaining stream if any otherwise 'Nothing'. The third parameter is to

90−-- represent an "empty" stream.

91−streamBuild :: IsStream t

92− => (forall r. Maybe (SVar Stream m a)

93− -> (a -> t m a -> m r)

94− -> (a -> m r)

95− -> m r

96− -> m r)

97− -> t m a

98−streamBuild k = fromStream $ Stream $ \svr stp sng yld ->

99− let yield a r = yld a (toStream r)

100− in k svr yield sng stp

101−

102−-- | Build a singleton stream from a callback function.

103−fromCallback :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a

104−fromCallback k = fromStream $ Stream $ \_ _ sng _ -> k sng

105−

10694-------------------------------------------------------------------------------

10795-- Generation by unfold

10896-------------------------------------------------------------------------------

@@ -128,37 +116,3 @@ unfoldrM step = go

128116 Just (a, b) ->

129117 K.runStream (a `cons` go b) svr stp sng yld

130118-}

131−

132−------------------------------------------------------------------------------

133−-- Destroying a stream

134−------------------------------------------------------------------------------

135−

136−-- | Fold a stream using its church encoding. The second argument is the "step"

137−-- function consuming an element and the remaining stream, if any. The third

138−-- argument is for consuming an "empty" stream that yields nothing.

139−streamFold

140− :: IsStream t

141− => Maybe (SVar Stream m a)

142− -> (a -> t m a -> m r)

143− -> (a -> m r)

144− -> m r

145− -> t m a

146− -> m r

147−streamFold svr step single blank m =

148− let yield a x = step a (fromStream x)

149− in (K.runStream (toStream m)) svr blank single yield

150−

151−-- | Run a streaming composition, discard the results. By default it interprets

152−-- the stream as 'SerialT', to run other types of streams use the type adapting

153−-- combinators for example @runStream . 'asyncly'@.

154−--

155−-- @since 0.2.0

156−runStream :: Monad m => SerialT m a -> m ()

157−runStream = run

158−

159−-- | Same as 'runStream'

160−--

161−-- @since 0.1.0

162−{-# DEPRECATED runStreaming "Please use runStream instead." #-}

163−runStreaming :: (Monad m, IsStream t) => t m a -> m ()

164−runStreaming = runStream . adapt

src/Streamly/Streams/SVar.hs+3−3

@@ -44,7 +44,7 @@ fromStreamVar sv = Stream $ \_ stp sng yld -> do

4444 -- Reversing the output is important to guarantee that we process the

4545 -- outputs in the same order as they were generated by the constituent

4646 -- streams.

47− runStream (processEvents $ reverse list) Nothing stp sng yld

47+ unStream (processEvents $ reverse list) Nothing stp sng yld

4848

4949 where

5050

@@ -62,7 +62,7 @@ fromStreamVar sv = Stream $ \_ stp sng yld -> do

6262 done <- postProcess sv

6363 if done

6464 then allDone stp

65− else runStream (fromStreamVar sv) Nothing stp sng yld

65+ else unStream (fromStreamVar sv) Nothing stp sng yld

6666

6767 processEvents (ev : es) = Stream $ \_ stp sng yld -> do

6868 let rest = processEvents es

@@ -71,7 +71,7 @@ fromStreamVar sv = Stream $ \_ stp sng yld -> do

7171 ChildStop tid e -> do

7272 accountThread sv tid

7373 case e of

74− Nothing -> runStream rest Nothing stp sng yld

74+ Nothing -> unStream rest Nothing stp sng yld

7575 Just ex -> throwM ex

7676

7777{-# INLINE fromSVar #-}

src/Streamly/Streams/Serial.hs+14−14

@@ -171,10 +171,10 @@ serial m1 m2 = fromStream $ C.serial (toStream m1) (toStream m2)

171171instance Monad m => Monad (SerialT m) where

172172 return = pure

173173 (SerialT (Stream m)) >>= f = SerialT $ Stream $ \_ stp sng yld ->

174− let runIt x = (runStream x) Nothing stp sng yld

175− single a = runIt $ toStream (f a)

176− yield a r = runIt $ toStream $ f a <> (fromStream r >>= f)

177− in m Nothing stp single yield

174+ let run x = (unStream x) Nothing stp sng yld

175+ single a = run $ toStream (f a)

176+ yieldk a r = run $ toStream $ f a <> (fromStream r >>= f)

177+ in m Nothing stp single yieldk

178178

179179------------------------------------------------------------------------------

180180-- Other instances

@@ -272,10 +272,10 @@ instance IsStream WSerialT where

272272{-# INLINE interleave #-}

273273interleave :: Stream m a -> Stream m a -> Stream m a

274274interleave m1 m2 = Stream $ \_ stp sng yld -> do

275− let stop = (runStream m2) Nothing stp sng yld

276− single a = yld a m2

277− yield a r = yld a (interleave m2 r)

278− (runStream m1) Nothing stop single yield

275+ let stop = (unStream m2) Nothing stp sng yld

276+ single a = yld a m2

277+ yieldk a r = yld a (interleave m2 r)

278+ (unStream m1) Nothing stop single yieldk

279279

280280-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WSerialT'.

281281-- Interleaves two streams, yielding one element from each stream alternately.

@@ -313,10 +313,10 @@ instance Monoid (WSerialT m a) where

313313instance Monad m => Monad (WSerialT m) where

314314 return = pure

315315 (WSerialT (Stream m)) >>= f = WSerialT $ Stream $ \_ stp sng yld ->

316− let runIt x = (runStream x) Nothing stp sng yld

317− single a = runIt $ toStream (f a)

318− yield a r = runIt $ toStream $ f a <> (fromStream r >>= f)

319− in m Nothing stp single yield

316+ let run x = (unStream x) Nothing stp sng yld

317+ single a = run $ toStream (f a)

318+ yieldk a r = run $ toStream $ f a <> (fromStream r >>= f)

319+ in m Nothing stp single yieldk

320320

321321------------------------------------------------------------------------------

322322-- Other instances

@@ -334,11 +334,11 @@ MONAD_COMMON_INSTANCES(WSerialT,)

334334-- @since 0.1.0

335335{-# DEPRECATED runStreamT "Please use runStream instead." #-}

336336runStreamT :: Monad m => SerialT m a -> m ()

337−runStreamT = run

337+runStreamT = runStream

338338

339339-- | Same as @runStream . wSerially@.

340340--

341341-- @since 0.1.0

342342{-# DEPRECATED runInterleavedT "Please use 'runStream . interleaving' instead." #-}

343343runInterleavedT :: Monad m => InterleavedT m a -> m ()

344−runInterleavedT = run

344+runInterleavedT = runStream

src/Streamly/Streams/StreamK.hs+124−47

@@ -20,32 +20,47 @@

2020-- Portability : GHC

2121--

2222--

23+-- Continuation passing style stream implementation.

24+--

25+-- import qualified Streamly.Streams.StreamK as K

26+--

2327module Streamly.Streams.StreamK

2428 (

2529 -- * Streams

2630 IsStream (..)

2731 , adapt

28− , Streaming -- deprecated

32+

2933 , Stream (..)

30− , run

34+ , mkStream

35+ , foldStream

36+ , runStream

3137

3238 -- * Construction

3339 , nil

40+ , yield

41+ , yieldM

3442 , cons

3543 , (.:)

36− , consMSerial

44+

45+ -- * Asynchronous construction

46+ , nilK

47+ , yieldK

48+ , consK

3749

3850 -- * Generation

39− , singleton

40− , once

4151 , repeat

4252

4353 -- * Semigroup Style Composition

4454 , serial

4555

4656 -- * Utilities

57+ , consMSerial

4758 , bindWith

4859 , withLocal

60+

61+ -- * Deprecated

62+ , Streaming -- deprecated

63+ , once -- deprecated

4964 )

5065where

5166

@@ -77,7 +92,7 @@ import Streamly.SVar

7792--

7893newtype Stream m a =

7994 Stream {

80− runStream :: forall r.

95+ unStream :: forall r.

8196 Maybe (SVar Stream m a) -- local state

8297 -> m r -- stop

8398 -> (a -> m r) -- singleton

@@ -152,6 +167,23 @@ type Streaming = IsStream

152167adapt :: (IsStream t1, IsStream t2) => t1 m a -> t2 m a

153168adapt = fromStream . toStream

154169

170+------------------------------------------------------------------------------

171+-- Building a stream

172+------------------------------------------------------------------------------

173+

174+-- | Build a stream from an 'SVar', a stop continuation, a singleton stream

175+-- continuation and a yield continuation.

176+mkStream:: IsStream t

177+ => (forall r. Maybe (SVar Stream m a)

178+ -> m r

179+ -> (a -> m r)

180+ -> (a -> t m a -> m r)

181+ -> m r)

182+ -> t m a

183+mkStream k = fromStream $ Stream $ \svr stp sng yld ->

184+ let yieldk a r = yld a (toStream r)

185+ in k svr stp sng yieldk

186+

155187------------------------------------------------------------------------------

156188-- Construction

157189------------------------------------------------------------------------------

@@ -167,11 +199,41 @@ adapt = fromStream . toStream

167199nil :: IsStream t => t m a

168200nil = fromStream $ Stream $ \_ stp _ _ -> stp

169201

202+-- faster than yieldM because there is no bind.

203+-- | Create a singleton stream from a pure value. Same as @yieldM (return a)@

204+-- but more efficient.

205+yield :: IsStream t => a -> t m a

206+yield a = fromStream $ Stream $ \_ _ single _ -> single a

207+

208+-- | Create a singleton stream from a monadic action. Same as @m \`consM` nil@

209+-- but more efficient.

210+--

211+-- @

212+-- > toList $ yieldM getLine

213+-- hello

214+-- ["hello"]

215+-- @

216+--

217+-- @since 0.4.0

218+{-# INLINE yieldM #-}

219+yieldM :: (Monad m, IsStream t) => m a -> t m a

220+yieldM m = fromStream $ Stream $ \_ _ single _ -> m >>= single

221+

222+-- | Same as yieldM

223+--

224+-- @since 0.2.0

225+{-# DEPRECATED once "Please use yieldM instead." #-}

226+{-# INLINE once #-}

227+once :: (Monad m, IsStream t) => m a -> t m a

228+once = yieldM

229+

170230infixr 5 `cons`

171231

172232-- faster than consM because there is no bind.

173233-- | Construct a stream by adding a pure value at the head of an existing

174−-- stream. For pure values it can be faster than 'consM'. For example:

234+-- stream. For serial streams this is the same as @(return a) \`consM` r@ but

235+-- more efficient. For concurrent streams this is not concurrent whereas

236+-- 'consM' is concurrent. For example:

175237--

176238-- @

177239-- > toList $ 1 \`cons` 2 \`cons` 3 \`cons` nil

@@ -210,6 +272,25 @@ infixr 5 .:

210272consMSerial :: (Monad m) => m a -> Stream m a -> Stream m a

211273consMSerial m r = Stream $ \_ _ _ yld -> m >>= \a -> yld a r

212274

275+------------------------------------------------------------------------------

276+-- Asynchronous construction

277+------------------------------------------------------------------------------

278+

279+-- | Make an empty stream from a callback function.

280+nilK :: IsStream t => (forall r. m r -> m r) -> t m a

281+nilK k = fromStream $ Stream $ \_ stp _ _ -> k stp

282+

283+-- | Make a singleton stream from a one shot callback function.

284+yieldK :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a

285+yieldK k = fromStream $ Stream $ \_ _ sng _ -> k sng

286+

287+-- | Construct a stream from a callback function.

288+consK :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a -> t m a

289+consK k r = fromStream $ Stream $ \_ _ _ yld -> k (\x -> yld x (toStream r))

290+

291+-- XXX consK with concurrent callbacks

292+-- XXX Build a stream from a repeating callback function.

293+

213294-------------------------------------------------------------------------------

214295-- IsStream Stream

215296-------------------------------------------------------------------------------

@@ -228,37 +309,33 @@ instance IsStream Stream where

228309 (|:) :: Monad m => m a -> Stream m a -> Stream m a

229310 (|:) = consMSerial

230311

231−run :: (Monad m, IsStream t) => t m a -> m ()

232−run m = go (toStream m)

312+-- | Fold a stream by providing an SVar, a stop continuation, a singleton

313+-- continuation and a yield continuation.

314+foldStream

315+ :: IsStream t

316+ => Maybe (SVar Stream m a)

317+ -> m r

318+ -> (a -> m r)

319+ -> (a -> t m a -> m r)

320+ -> t m a

321+ -> m r

322+foldStream svr blank single step m =

323+ let yieldk a x = step a (fromStream x)

324+ in (unStream (toStream m)) svr blank single yieldk

325+

326+runStream :: (Monad m, IsStream t) => t m a -> m ()

327+runStream m = go (toStream m)

233328 where

234329 go m1 =

235330 let stop = return ()

236331 single _ = return ()

237− yield _ r = go (toStream r)

238− in (runStream m1) Nothing stop single yield

332+ yieldk _ r = go (toStream r)

333+ in (unStream m1) Nothing stop single yieldk

239334

240335-------------------------------------------------------------------------------

241336-- Special generation

242337-------------------------------------------------------------------------------

243338

244−-- | Same as @once . return@ but may be faster because there is no bind

245−singleton :: IsStream t => a -> t m a

246−singleton a = fromStream $ Stream $ \_ _ single _ -> single a

247−

248−-- | Create a singleton stream by executing a monadic action once. Same as

249−-- @m \`consM` nil@ but more efficient.

250−--

251−-- @

252−-- > toList $ once getLine

253−-- hello

254−-- ["hello"]

255−-- @

256−--

257−-- @since 0.2.0

258−{-# INLINE once #-}

259−once :: (Monad m, IsStream t) => m a -> t m a

260−once m = fromStream $ Stream $ \_ _ single _ -> m >>= single

261−

262339repeat :: IsStream t => a -> t m a

263340repeat a = let x = cons a x in x

264341

@@ -273,10 +350,10 @@ serial :: Stream m a -> Stream m a -> Stream m a

273350serial m1 m2 = go m1

274351 where

275352 go (Stream m) = Stream $ \_ stp sng yld ->

276− let stop = (runStream m2) Nothing stp sng yld

277− single a = yld a m2

278− yield a r = yld a (go r)

279− in m Nothing stop single yield

353+ let stop = (unStream m2) Nothing stp sng yld

354+ single a = yld a m2

355+ yieldk a r = yld a (go r)

356+ in m Nothing stop single yieldk

280357

281358instance Semigroup (Stream m a) where

282359 (<>) = serial

@@ -295,9 +372,9 @@ instance Monoid (Stream m a) where

295372

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

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

298− let single = sng . f

299− yield a r = yld (f a) (fmap f r)

300− in (runStream m) Nothing stp single yield

375+ let single = sng . f

376+ yieldk a r = yld (f a) (fmap f r)

377+ in (unStream m) Nothing stp single yieldk

301378

302379-------------------------------------------------------------------------------

303380-- Bind utility

@@ -313,10 +390,10 @@ bindWith par m f = go m

313390 where

314391 go (Stream g) =

315392 Stream $ \ctx stp sng yld ->

316− let runIt x = (runStream x) ctx stp sng yld

317− single a = runIt $ f a

318− yield a r = runIt $ f a `par` go r

319− in g Nothing stp single yield

393+ let run x = (unStream x) ctx stp sng yld

394+ single a = run $ f a

395+ yieldk a r = run $ f a `par` go r

396+ in g Nothing stp single yieldk

320397

321398------------------------------------------------------------------------------

322399-- Alternative & MonadPlus

@@ -324,8 +401,8 @@ bindWith par m f = go m

324401

325402_alt :: Stream m a -> Stream m a -> Stream m a

326403_alt m1 m2 = Stream $ \_ stp sng yld ->

327− let stop = runStream m2 Nothing stp sng yld

328− in runStream m1 Nothing stop sng yld

404+ let stop = unStream m2 Nothing stp sng yld

405+ in unStream m1 Nothing stop sng yld

329406

330407------------------------------------------------------------------------------

331408-- MonadReader

@@ -335,8 +412,8 @@ withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a

335412withLocal f m =

336413 Stream $ \_ stp sng yld ->

337414 let single = local f . sng

338− yield a r = local f $ yld a (withLocal f r)

339− in (runStream m) Nothing (local f stp) single yield

415+ yieldk a r = local f $ yld a (withLocal f r)

416+ in (unStream m) Nothing (local f stp) single yieldk

340417

341418------------------------------------------------------------------------------

342419-- MonadError

@@ -349,9 +426,9 @@ withCatchError

349426 => Stream m a -> (e -> Stream m a) -> Stream m a

350427withCatchError m h =

351428 Stream $ \_ stp sng yld ->

352− let run x = runStream x Nothing stp sng yield

429+ let run x = unStream x Nothing stp sng yieldk

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

354− yield a r = yld a (withCatchError r h)

431+ yieldk a r = yld a (withCatchError r h)

355432 in handle $ run m

356433-}

357434

@@ -360,4 +437,4 @@ withCatchError m h =

360437-------------------------------------------------------------------------------

361438

362439instance MonadTrans Stream where

363− lift = once

440+ lift = yieldM

src/Streamly/Streams/Zip.hs+9−9

@@ -61,10 +61,10 @@ zipWithS f m1 m2 = go m1 m2

6161 let merge a ra =

6262 let single2 b = sng (f a b)

6363 yield2 b rb = yld (f a b) (go ra rb)

64− in (runStream my) Nothing stp single2 yield2

64+ in unStream my Nothing stp single2 yield2

6565 let single1 a = merge a nil

6666 yield1 a ra = merge a ra

67− (runStream mx) Nothing stp single1 yield1

67+ unStream mx Nothing stp single1 yield1

6868

6969-- | Zip two streams serially using a pure zipping function.

7070--

@@ -81,13 +81,13 @@ zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

8181 where

8282 go mx my = Stream $ \_ stp sng yld -> do

8383 let merge a ra =

84− let runIt x = runStream x Nothing stp sng yld

84+ let runIt x = unStream x Nothing stp sng yld

8585 single2 b = runIt $ toStream (f a b)

8686 yield2 b rb = runIt $ toStream (f a b) <> go ra rb

87− in (runStream my) Nothing stp single2 yield2

87+ in unStream my Nothing stp single2 yield2

8888 let single1 a = merge a nil

8989 yield1 a ra = merge a ra

90− (runStream mx) Nothing stp single1 yield1

90+ unStream mx Nothing stp single1 yield1

9191

9292------------------------------------------------------------------------------

9393-- Serially Zipping Streams

@@ -168,7 +168,7 @@ zipAsyncWith :: (IsStream t, MonadAsync m)

168168zipAsyncWith f m1 m2 = fromStream $ Stream $ \_ stp sng yld -> do

169169 ma <- mkAsync m1

170170 mb <- mkAsync m2

171− (runStream (toStream (zipWith f ma mb))) Nothing stp sng yld

171+ unStream (toStream (zipWith f ma mb)) Nothing stp sng yld

172172

173173-- | Zip two streams asyncly (i.e. both the elements being zipped are generated

174174-- concurrently) using a monadic zipping function.

@@ -179,7 +179,7 @@ zipAsyncWithM :: (IsStream t, MonadAsync m)

179179zipAsyncWithM f m1 m2 = fromStream $ Stream $ \_ stp sng yld -> do

180180 ma <- mkAsync m1

181181 mb <- mkAsync m2

182− (runStream (toStream (zipWithM f ma mb))) Nothing stp sng yld

182+ unStream (toStream (zipWithM f ma mb)) Nothing stp sng yld

183183

184184------------------------------------------------------------------------------

185185-- Parallely Zipping Streams

@@ -245,11 +245,11 @@ instance MonadAsync m => Applicative (ZipAsyncM m) where

245245-- @since 0.1.0

246246{-# DEPRECATED runZipStream "Please use 'runStream . zipSerially instead." #-}

247247runZipStream :: Monad m => ZipSerialM m a -> m ()

248−runZipStream = run

248+runZipStream = runStream

249249

250250-- | Same as @runStream . zippingAsync@.

251251--

252252-- @since 0.1.0

253253{-# DEPRECATED runZipAsync "Please use 'runStream . zipAsyncly instead." #-}

254254runZipAsync :: Monad m => ZipAsyncM m a -> m ()

255−runZipAsync = run

255+runZipAsync = runStream

test/Main.hs+19−19

@@ -40,7 +40,7 @@ main = hspec $ do

4040 it "simple serially" $

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

4242 it "simple serially with IO" $

43− (runStream . serially) (A.once $ putStrLn "hello") `shouldReturn` ()

43+ (runStream . serially) (A.yieldM $ putStrLn "hello") `shouldReturn` ()

4444

4545 describe "Empty" $ do

4646 it "Monoid - mempty" $

@@ -381,12 +381,12 @@ main = hspec $ do

381381

382382 it "asyncly crosses thread limit (2000 threads)" $

383383 runStream (asyncly $ fold $

384− replicate 2000 $ A.once $ threadDelay 1000000)

384+ replicate 2000 $ A.yieldM $ threadDelay 1000000)

385385 `shouldReturn` ()

386386

387387 it "aheadly crosses thread limit (4000 threads)" $

388388 runStream (aheadly $ fold $

389− replicate 4000 $ A.once $ threadDelay 1000000)

389+ replicate 4000 $ A.yieldM $ threadDelay 1000000)

390390 `shouldReturn` ()

391391

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

@@ -578,7 +578,7 @@ nestTwoParallelApp =

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

579579

580580timed :: (IsStream t, Monad (t IO)) => Int -> t IO Int

581−timed x = A.once (threadDelay (x * 100000)) >> return x

581+timed x = A.yieldM (threadDelay (x * 100000)) >> return x

582582

583583interleaveCheck :: IsStream t

584584 => (t IO Int -> SerialT IO Int)

@@ -602,7 +602,7 @@ parallelCheck t f = do

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

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

604604

605− where event n = (A.once $ threadDelay (n * 100000)) >> (return n)

605+ where event n = (A.yieldM $ threadDelay (n * 100000)) >> (return n)

606606

607607compose :: (IsStream t, Semigroup (t IO Int))

608608 => (t IO Int -> SerialT IO Int) -> t IO Int -> ([Int] -> [Int]) -> Spec

@@ -682,12 +682,12 @@ loops t tsrt hsrt = do

682682 where

683683 loopHead x = do

684684 -- this print line is important for the test (causes a bind)

685− A.once $ putStrLn "LoopHead..."

685+ A.yieldM $ putStrLn "LoopHead..."

686686 t $ (if x < 3 then loopHead (x + 1) else nil) <> return x

687687

688688 loopTail x = do

689689 -- this print line is important for the test (causes a bind)

690− A.once $ putStrLn "LoopTail..."

690+ A.yieldM $ putStrLn "LoopTail..."

691691 t $ return x <> (if x < 3 then loopTail (x + 1) else nil)

692692

693693bindAndComposeSimple

@@ -755,21 +755,21 @@ mixedOps = do

755755

756756 composeMixed :: SerialT IO Int

757757 composeMixed = do

758− A.once $ return ()

759− A.once $ putStr ""

758+ A.yieldM $ return ()

759+ A.yieldM $ putStr ""

760760 x <- return 1

761761 y <- return 2

762762 z <- do

763763 x1 <- wAsyncly $ return 1 <> return 2

764− A.once $ return ()

765− A.once $ putStr ""

764+ A.yieldM $ return ()

765+ A.yieldM $ putStr ""

766766 y1 <- asyncly $ return 1 <> return 2

767767 z1 <- do

768768 x11 <- return 1 <> return 2

769769 y11 <- asyncly $ return 1 <> return 2

770770 z11 <- wSerially $ return 1 <> return 2

771− A.once $ return ()

772− A.once $ putStr ""

771+ A.yieldM $ return ()

772+ A.yieldM $ putStr ""

773773 return (x11 + y11 + z11)

774774 return (x1 + y1 + z1)

775775 return (x + y + z)

@@ -785,21 +785,21 @@ mixedOpsAheadly = do

785785

786786 composeMixed :: SerialT IO Int

787787 composeMixed = do

788− A.once $ return ()

789− A.once $ putStr ""

788+ A.yieldM $ return ()

789+ A.yieldM $ putStr ""

790790 x <- return 1

791791 y <- return 2

792792 z <- do

793793 x1 <- wAsyncly $ return 1 <> return 2

794− A.once $ return ()

795− A.once $ putStr ""

794+ A.yieldM $ return ()

795+ A.yieldM $ putStr ""

796796 y1 <- aheadly $ return 1 <> return 2

797797 z1 <- do

798798 x11 <- return 1 <> return 2

799799 y11 <- aheadly $ return 1 <> return 2

800800 z11 <- parallely $ return 1 <> return 2

801− A.once $ return ()

802− A.once $ putStr ""

801+ A.yieldM $ return ()

802+ A.yieldM $ putStr ""

803803 return (x11 + y11 + z11)

804804 return (x1 + y1 + z1)

805805 return (x + y + z)

test/Prop.hs+3−3

@@ -159,10 +159,10 @@ concurrentUnfoldrM eq op n =

159159 -- results may not be yielded in order, in case of

160160 -- Async/WAsync/Parallel. So we use an increasing count

161161 -- instead.

162− i <- A.once $ readIORef cnt

163− A.once $ modifyIORef cnt (+1)

162+ i <- A.yieldM $ readIORef cnt

163+ A.yieldM $ modifyIORef cnt (+1)

164164 let msg = show i ++ "/" ++ show n

165− A.once $ do

165+ A.yieldM $ do

166166 if even i

167167 then do

168168 dbgMVar ("first take concurrentUnfoldrM " ++ msg)

test/loops.hs+11−11

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

11import Streamly

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

3−import Streamly.Prelude (nil, once)

3+import Streamly.Prelude (nil, yieldM)

55main = do

66 hSetBuffering stdout LineBuffering

@@ -8,32 +8,32 @@ main = do

88 putStrLn $ "\nloopTail:\n"

99 runStream $ do

1010 x <- loopTail 0

11− once $ print (x :: Int)

11+ yieldM $ print (x :: Int)

1212

1313 putStrLn $ "\nloopHead:\n"

1414 runStream $ do

1515 x <- loopHead 0

16− once $ print (x :: Int)

16+ yieldM $ print (x :: Int)

1717

1818 putStrLn $ "\nloopTailA:\n"

1919 runStream $ do

2020 x <- loopTailA 0

21− once $ print (x :: Int)

21+ yieldM $ print (x :: Int)

2222

2323 putStrLn $ "\nloopHeadA:\n"

2424 runStream $ do

2525 x <- loopHeadA 0

26− once $ print (x :: Int)

26+ yieldM $ print (x :: Int)

2727

2828 putStrLn $ "\nwSerial:\n"

2929 runStream $ do

3030 x <- (return 0 <> return 1) `wSerial` (return 100 <> return 101)

31− once $ print (x :: Int)

31+ yieldM $ print (x :: Int)

3232

3333 putStrLn $ "\nParallel interleave:\n"

3434 runStream $ do

3535 x <- (return 0 <> return 1) `wAsync` (return 100 <> return 101)

36− once $ print (x :: Int)

36+ yieldM $ print (x :: Int)

3737

3838 where

3939

@@ -45,7 +45,7 @@ main = do

4545 -- stream. Interleaves the generator and the consumer.

4646 loopTail :: Int -> Serial Int

4747 loopTail x = do

48− once $ putStrLn "LoopTail..."

48+ yieldM $ putStrLn "LoopTail..."

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

5050

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

@@ -53,7 +53,7 @@ main = do

5353 -- at all.

5454 loopHead :: Int -> Serial Int

5555 loopHead x = do

56− once $ putStrLn "LoopHead..."

56+ yieldM $ putStrLn "LoopHead..."

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

5858

5959-------------------------------------------------------------------------------

@@ -62,12 +62,12 @@ main = do

6262

6363 loopTailA :: Int -> Serial Int

6464 loopTailA x = do

65− once $ putStrLn "LoopTailA..."

65+ yieldM $ putStrLn "LoopTailA..."

6666 return x `async` (if x < 3 then loopTailA (x + 1) else nil)

6767

6868 loopHeadA :: Int -> Serial Int

6969 loopHeadA x = do

70− once $ putStrLn "LoopHeadA..."

70+ yieldM $ putStrLn "LoopHeadA..."

7171 (if x < 3 then loopHeadA (x + 1) else nil) `async` return x

7272

7373-------------------------------------------------------------------------------

test/nested-loops.hs+5−5

@@ -2,23 +2,23 @@ import Control.Concurrent (myThreadId)

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

33import System.Random (randomIO)

44import Streamly

5−import Streamly.Prelude (nil, once)

5+import Streamly.Prelude (nil, yieldM)

77main = runStream $ do

8− once $ hSetBuffering stdout LineBuffering

8+ yieldM $ hSetBuffering stdout LineBuffering

99 x <- loop "A " 2

1010 y <- loop "B " 2

11− once $ myThreadId >>= putStr . show

11+ yieldM $ myThreadId >>= putStr . show

1212 >> putStr " "

1313 >> print (x, y)

1414

1515 where

1616

1717 -- we can just use

18− -- parallely $ mconcat $ replicate n $ once (...)

18+ -- parallely $ mconcat $ replicate n $ yieldM (...)

1919 loop :: String -> Int -> SerialT IO String

2020 loop name n = do

21− rnd <- once (randomIO :: IO Int)

21+ rnd <- yieldM (randomIO :: IO Int)

2222 let result = (name ++ show rnd)

2323 repeat = if n > 1 then loop name (n - 1) else nil

2424 in (return result) `wAsync` repeat

test/parallel-loops.hs+5−5

@@ -8,19 +8,19 @@ main = do

88 hSetBuffering stdout LineBuffering

99 runStream $ do

1010 x <- S.take 10 $ loop "A" `parallel` loop "B"

11− S.once $ myThreadId >>= putStr . show

11+ S.yieldM $ myThreadId >>= putStr . show

1212 >> putStr " got "

1313 >> print x

1414

1515 where

1616

1717 -- we can just use

18− -- parallely $ cycle1 $ once (...)

18+ -- parallely $ cycle1 $ yieldM (...)

1919 loop :: String -> Serial (String, Int)

2020 loop name = do

21− S.once $ threadDelay 1000000

22− rnd <- S.once (randomIO :: IO Int)

23− S.once $ myThreadId >>= putStr . show

21+ S.yieldM $ threadDelay 1000000

22+ rnd <- S.yieldM (randomIO :: IO Int)

23+ S.yieldM $ myThreadId >>= putStr . show

2424 >> putStr " yielding "

2525 >> print rnd

2626 return (name, rnd) `parallel` loop name