2ba5f01 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

rename unstream et. al. and make them polymorphic

2ba5f01Harendra Kumar8 years ago

src/Streamly/Prelude.hs+12−12

compare with master

unifiedsplit

@@ -840,9 +840,9 @@ iterate step = fromStream . go

840iterateM :: (IsStream t, MonadAsync m) => (a -> m a) -> a -> t m a

841iterateM step = go

842 where

843− go s = fromStream $ K.mkStream $ \svr stp sng yld -> do

843+ go s = K.mkStream $ \st stp sng yld -> do

844 next <- step s

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

845+ K.foldStreamShared st stp sng yld (return s |: go next)

846

847------------------------------------------------------------------------------

848-- Conversions

@@ -894,7 +894,7 @@ each = K.fromFoldable

894--

895-- @since 0.1.0

896fromHandle :: (IsStream t, MonadIO m) => IO.Handle -> t m String

897−fromHandle h = fromStream go

897+fromHandle h = go

898 where

899 go = K.mkStream $ \_ stp _ yld -> do

900 eof <- liftIO $ IO.hIsEOF h

@@ -1443,13 +1443,13 @@ toList = P.toList

1443--

1444-- @since 0.1.0

1445toHandle :: MonadIO m => IO.Handle -> SerialT m String -> m ()

1446−toHandle h m = go (toStream m)

1446+toHandle h m = go m

1447 where

1448 go m1 =

1449 let stop = return ()

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

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

1452− in K.unStream m1 defState stop single yieldk

1452+ in K.foldStream defState stop single yieldk m1

1453

1454------------------------------------------------------------------------------

1455-- Transformation by Folding (Scans)

@@ -1768,14 +1768,14 @@ mapMaybeMSerial f m = fromStreamD $ D.mapMaybeM f $ toStreamD m

1768--

1769-- @since 0.1.1

1770reverse :: (IsStream t) => t m a -> t m a

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

1771+reverse m = go K.nil m

1772 where

1773 go rev rest = K.mkStream $ \st stp sng yld ->

1774− let runIt x = K.unStream x st stp sng yld

1774+ let runIt x = K.foldStream st stp sng yld x

1775 stop = runIt rev

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

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

1778− in K.unStream rest st stop single yieldk

1778+ in K.foldStream st stop single yieldk rest

1779

1780------------------------------------------------------------------------------

1781-- Transformation by Inserting

@@ -1986,10 +1986,10 @@ merge = mergeBy compare

1986-- @since 0.6.0

1987mergeAsyncBy :: (IsStream t, MonadAsync m)

1988 => (a -> a -> Ordering) -> t m a -> t m a -> t m a

1989−mergeAsyncBy f m1 m2 = K.fromStream $ K.mkStream $ \st stp sng yld -> do

1989+mergeAsyncBy f m1 m2 = K.mkStream $ \st stp sng yld -> do

1990 ma <- mkAsync' st m1

1991 mb <- mkAsync' st m2

1992− K.unStream (K.toStream (K.mergeBy f ma mb)) st stp sng yld

1992+ K.foldStream st stp sng yld (K.mergeBy f ma mb)

1993

1994-- | Like 'mergeByM' but merges concurrently (i.e. both the elements being

1995-- merged are generated concurrently).

@@ -1997,7 +1997,7 @@ mergeAsyncBy f m1 m2 = K.fromStream $ K.mkStream $ \st stp sng yld -> do

1997-- @since 0.6.0

1998mergeAsyncByM :: (IsStream t, MonadAsync m)

1999 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a

2000−mergeAsyncByM f m1 m2 = K.fromStream $ K.mkStream $ \st stp sng yld -> do

2000+mergeAsyncByM f m1 m2 = K.mkStream $ \st stp sng yld -> do

2001 ma <- mkAsync' st m1

2002 mb <- mkAsync' st m2

2003− K.unStream (K.toStream (K.mergeByM f ma mb)) st stp sng yld

2003+ K.foldStream st stp sng yld (K.mergeByM f ma mb)

src/Streamly/SVar.hs+1−1

@@ -434,7 +434,7 @@ defState = State

434-- results in slightly better perf for zip/zipM but the performance of scan

435-- worsens a lot, it does not fuse.

436--

437−-- This has a side effect of clearing the SVar and yieldLimit, therefore it

437+-- XXX This has a side effect of clearing the SVar and yieldLimit, therefore it

438-- should not be used to convert from the same type to the same type, unless

439-- you want to clear the SVar. For clearing the SVar you should be using the

440-- appropriate unStream functions instead.

src/Streamly/Streams/Ahead.hs+35−36

@@ -48,8 +48,8 @@ import Streamly.Streams.SVar (fromSVar)

48import Streamly.Streams.Serial (map)

49import Streamly.SVar

50import Streamly.Streams.StreamK

51− (IsStream(..), Stream, mkStream, unStream, unStreamShared,

51+ (IsStream(..), Stream, mkStream, foldStream, foldStreamShared,

52− runStreamSVar)

52+ foldStreamSVar, unShare)

53import qualified Streamly.Streams.StreamK as K

55import Prelude hiding (map)

@@ -298,9 +298,10 @@ processHeap q heap st sv winfo entry sno stopping = loopHeap sno entry

298 let stop = do

299 liftIO (incrementYieldLimit sv)

300 nextHeap seqNo

301− runStreamSVar sv r st stop

301+ foldStreamSVar sv st stop

302 (singleStreamFromHeap seqNo)

303 (yieldStreamFromHeap seqNo)

304+ r

304 else liftIO $ do

305 else liftIO $ do

305 let ent = Entry seqNo (AheadEntryStream r)

306 let ent = Entry seqNo (AheadEntryStream r)

306 liftIO $ requeueOnHeapTop heap ent seqNo

307 liftIO $ requeueOnHeapTop heap ent seqNo

@@ -348,9 +349,10 @@ processWithoutToken q heap st sv winfo m seqNo = do

348 -- we stop.

349 -- we stop.

349 toHeap AheadEntryNull

350 toHeap AheadEntryNull

350

351

351− runStreamSVar sv m st stop

352+ foldStreamSVar sv st stop

352 (toHeap . AheadEntryPure)

353 (toHeap . AheadEntryPure)

353 (\a r -> toHeap $ AheadEntryStream $ K.cons a r)

354 (\a r -> toHeap $ AheadEntryStream $ K.cons a r)

355+ m

354

356

355 where

357 where

356

358

@@ -408,7 +410,7 @@ processWithToken q heap st sv winfo action sno = do

408 liftIO (incrementYieldLimit sv)

410 liftIO (incrementYieldLimit sv)

409 loopWithToken (sno + 1)

411 loopWithToken (sno + 1)

410

412

411− runStreamSVar sv action st stop (singleOutput sno) (yieldOutput sno)

413+ foldStreamSVar sv st stop (singleOutput sno) (yieldOutput sno) action

412

414

413 where

415 where

414

416

@@ -431,9 +433,10 @@ processWithToken q heap st sv winfo action sno = do

431 let stop = do

433 let stop = do

432 liftIO (incrementYieldLimit sv)

434 liftIO (incrementYieldLimit sv)

433 loopWithToken (seqNo + 1)

435 loopWithToken (seqNo + 1)

434− runStreamSVar sv r st stop

436+ foldStreamSVar sv st stop

435 (singleOutput seqNo)

437 (singleOutput seqNo)

436 (yieldOutput seqNo)

438 (yieldOutput seqNo)

439+ r

437 else do

440 else do

438 let ent = Entry seqNo (AheadEntryStream r)

441 let ent = Entry seqNo (AheadEntryStream r)

439 liftIO $ requeueOnHeapTop heap ent seqNo

442 liftIO $ requeueOnHeapTop heap ent seqNo

@@ -460,9 +463,10 @@ processWithToken q heap st sv winfo action sno = do

460 let stop = do

463 let stop = do

461 liftIO (incrementYieldLimit sv)

464 liftIO (incrementYieldLimit sv)

462 loopWithToken (seqNo + 1)

465 loopWithToken (seqNo + 1)

463− runStreamSVar sv m st stop

466+ foldStreamSVar sv st stop

464 (singleOutput seqNo)

467 (singleOutput seqNo)

465 (yieldOutput seqNo)

468 (yieldOutput seqNo)

469+ m

466 else

470 else

467 -- To avoid a race when another thread puts something

471 -- To avoid a race when another thread puts something

468 -- on the heap and goes away, the consumer will not get

472 -- on the heap and goes away, the consumer will not get

@@ -543,32 +547,37 @@ workLoopAhead q heap st sv winfo = do

543

547

544-- The only difference between forkSVarAsync and this is that we run the left

548-- The only difference between forkSVarAsync and this is that we run the left

545-- computation without a shared SVar.

549-- computation without a shared SVar.

546−forkSVarAhead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

550+forkSVarAhead :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

547forkSVarAhead m1 m2 = mkStream $ \st stp sng yld -> do

551forkSVarAhead m1 m2 = mkStream $ \st stp sng yld -> do

548− sv <- newAheadVar st (concurrently m1 m2) workLoopAhead

552+ sv <- newAheadVar st (concurrently (toStream m1) (toStream m2))

549− unStream (fromSVar sv) st stp sng yld

553+ workLoopAhead

554+ foldStream st stp sng yld (fromSVar sv)

550 where

555 where

551 concurrently ma mb = mkStream $ \st stp sng yld -> do

556 concurrently ma mb = mkStream $ \st stp sng yld -> do

552 liftIO $ enqueue (fromJust $ streamVar st) mb

557 liftIO $ enqueue (fromJust $ streamVar st) mb

553− unStream ma st stp sng yld

558+ foldStream st stp sng yld ma

554

559

560+-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AheadT'.

561+-- Merges two streams sequentially but with concurrent lookahead.

562+--

563+-- @since 0.3.0

555−{-# INLINE aheadS #-}

564+{-# INLINE ahead #-}

556−aheadS :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

565+ahead :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

557−aheadS m1 m2 = mkStream $ \st stp sng yld ->

566+ahead m1 m2 = mkStream $ \st stp sng yld ->

558 case streamVar st of

567 case streamVar st of

559 Just sv | svarStyle sv == AheadVar -> do

568 Just sv | svarStyle sv == AheadVar -> do

560− liftIO $ enqueue sv m2

569+ liftIO $ enqueue sv (toStream m2)

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

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

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

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

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

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

564− unStream m1 st stp sng yld

573+ foldStream st stp sng yld m1

565− _ -> unStreamShared (forkSVarAhead m1 m2) st stp sng yld

574+ _ -> foldStreamShared st stp sng yld (forkSVarAhead m1 m2)

566

575

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

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

568-- of combining streams using ahead.

577-- of combining streams using ahead.

569{-# INLINE consMAhead #-}

578{-# INLINE consMAhead #-}

570−consMAhead :: MonadAsync m => m a -> Stream m a -> Stream m a

579+consMAhead :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a

571−consMAhead m r = K.yieldM m `aheadS` r

580+consMAhead m r = K.yieldM m `ahead` r

572

581

573------------------------------------------------------------------------------

582------------------------------------------------------------------------------

574-- AheadT

583-- AheadT

@@ -639,7 +648,7 @@ instance IsStream AheadT where

639

648

640 {-# INLINE consM #-}

649 {-# INLINE consM #-}

641 {-# SPECIALIZE consM :: IO a -> AheadT IO a -> AheadT IO a #-}

650 {-# SPECIALIZE consM :: IO a -> AheadT IO a -> AheadT IO a #-}

642− consM m r = fromStream $ consMAhead m (toStream r)

651+ consM = consMAhead

643

652

644 {-# INLINE (|:) #-}

653 {-# INLINE (|:) #-}

645 {-# SPECIALIZE (|:) :: IO a -> AheadT IO a -> AheadT IO a #-}

654 {-# SPECIALIZE (|:) :: IO a -> AheadT IO a -> AheadT IO a #-}

@@ -649,15 +658,6 @@ instance IsStream AheadT where

649-- Semigroup

658-- Semigroup

650------------------------------------------------------------------------------

659------------------------------------------------------------------------------

651

660

652−-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AheadT'.

653−-- Merges two streams sequentially but with concurrent lookahead.

654−--

655−-- @since 0.3.0

656−{-# INLINE ahead #-}

657−ahead :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

658−ahead m1 m2 = fromStream $ mkStream $ \st stp sng yld ->

659− unStreamShared (aheadS (toStream m1) (toStream m2)) st stp sng yld

660−

661instance MonadAsync m => Semigroup (AheadT m a) where

662 (<>) = ahead

663

@@ -673,6 +673,8 @@ instance MonadAsync m => Monoid (AheadT m a) where

673-- Monad

674------------------------------------------------------------------------------

675

676+-- XXX use bindWith instead

677+

676{-# INLINE aheadbind #-}

678{-# INLINE aheadbind #-}

677aheadbind

679aheadbind

678 :: MonadAsync m

680 :: MonadAsync m

@@ -683,13 +685,10 @@ aheadbind m f = go m

683 where

685 where

684 go g =

686 go g =

685 mkStream $ \st stp sng yld ->

687 mkStream $ \st stp sng yld ->

686− let runShared x = unStreamShared x st stp sng yld

688+ let foldShared = foldStreamShared st stp sng yld

687− runIsolated x = unStream x st stp sng yld

689+ single a = foldShared $ unShare (f a)

688−

689− single a = runIsolated $ f a

690− yieldk a r = runShared $

691− K.unShare (f a) `aheadS` go r

690+ yieldk a r = foldShared $ unShare (f a) `ahead` go r

692− in unStream g st stp single yieldk

691+ in foldStream (adaptState st) stp single yieldk g

693

692

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

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

695 return = pure

694 return = pure

src/Streamly/Streams/Async.hs+32−43

@@ -58,8 +58,8 @@ import Streamly.Streams.SVar (fromSVar)

58import Streamly.Streams.Serial (map)

59import Streamly.SVar

60import Streamly.Streams.StreamK

61− (IsStream(..), Stream, mkStream, unStream, adapt, unStreamShared,

61+ (IsStream(..), Stream, mkStream, foldStream, adapt, foldStreamShared,

62− runStreamSVar)

62+ foldStreamSVar)

63import qualified Streamly.Streams.StreamK as K

65#include "Instances.hs"

@@ -84,7 +84,7 @@ workLoopLIFO q st sv winfo = run

84 work <- dequeue

85 case work of

86 Nothing -> liftIO $ sendStop sv winfo

87− Just m -> runStreamSVar sv m st run single yieldk

87+ Just m -> foldStreamSVar sv st run single yieldk m

89 single a = do

90 res <- liftIO $ sendYield sv winfo (ChildYield a)

@@ -93,7 +93,7 @@ workLoopLIFO q st sv winfo = run

93 yieldk a r = do

94 res <- liftIO $ sendYield sv winfo (ChildYield a)

95 if res

96− then runStreamSVar sv r st run single yieldk

96+ then foldStreamSVar sv st run single yieldk r

97 else liftIO $ do

98 enqueueLIFO sv q r

99 sendStop sv winfo

@@ -134,7 +134,7 @@ workLoopLIFOLimited q st sv winfo = run

134 if yieldLimitOk

135 then do

136 let stop = liftIO (incrementYieldLimit sv) >> run

137− runStreamSVar sv m st stop single yieldk

137+ foldStreamSVar sv st stop single yieldk m

138 -- Avoid any side effects, undo the yield limit decrement if we

139 -- never yielded anything.

140 else liftIO $ do

@@ -153,7 +153,7 @@ workLoopLIFOLimited q st sv winfo = run

153 yieldLimitOk <- liftIO $ decrementYieldLimit sv

154 let stop = liftIO (incrementYieldLimit sv) >> run

155 if res && yieldLimitOk

156− then runStreamSVar sv r st stop single yieldk

156+ then foldStreamSVar sv st stop single yieldk r

157 else liftIO $ do

158 incrementYieldLimit sv

159 enqueueLIFO sv q r

@@ -185,7 +185,7 @@ workLoopFIFO q st sv winfo = run

185 work <- liftIO $ tryPopR q

186 case work of

187 Nothing -> liftIO $ sendStop sv winfo

188− Just m -> runStreamSVar sv m st run single yieldk

188+ Just m -> foldStreamSVar sv st run single yieldk m

189

190 single a = do

191 res <- liftIO $ sendYield sv winfo (ChildYield a)

@@ -194,7 +194,7 @@ workLoopFIFO q st sv winfo = run

194 yieldk a r = do

195 res <- liftIO $ sendYield sv winfo (ChildYield a)

196 if res

197− then runStreamSVar sv r st run single yieldk

197+ then foldStreamSVar sv st run single yieldk r

198 else liftIO $ do

199 enqueueFIFO sv q r

200 sendStop sv winfo

@@ -220,7 +220,7 @@ workLoopFIFOLimited q st sv winfo = run

220 if yieldLimitOk

221 then do

222 let stop = liftIO (incrementYieldLimit sv) >> run

223− runStreamSVar sv m st stop single yieldk

223+ foldStreamSVar sv st stop single yieldk m

224 else liftIO $ do

225 enqueueFIFO sv q m

226 incrementYieldLimit sv

@@ -235,7 +235,7 @@ workLoopFIFOLimited q st sv winfo = run

235 yieldLimitOk <- liftIO $ decrementYieldLimit sv

236 let stop = liftIO (incrementYieldLimit sv) >> run

237 if res && yieldLimitOk

238− then runStreamSVar sv r st stop single yieldk

238+ then foldStreamSVar sv st stop single yieldk r

239 else liftIO $ do

240 incrementYieldLimit sv

241 enqueueFIFO sv q r

@@ -528,36 +528,33 @@ newWAsyncVar st m = do

528-- composition and vice-versa we create a new SVar to isolate the scheduling

529-- of the two.

530

531−forkSVarAsync :: MonadAsync m

531+forkSVarAsync :: (IsStream t, MonadAsync m)

532− => SVarStyle -> Stream m a -> Stream m a -> Stream m a

532+ => SVarStyle -> t m a -> t m a -> t m a

533forkSVarAsync style m1 m2 = mkStream $ \st stp sng yld -> do

534 sv <- case style of

535− AsyncVar -> newAsyncVar st (concurrently m1 m2)

535+ AsyncVar -> newAsyncVar st (concurrently (toStream m1) (toStream m2))

536− WAsyncVar -> newWAsyncVar st (concurrently m1 m2)

536+ WAsyncVar -> newWAsyncVar st (concurrently (toStream m1) (toStream m2))

537 _ -> error "illegal svar type"

538− unStream (fromSVar sv) st stp sng yld

538+ foldStream st stp sng yld $ fromSVar sv

539 where

540 concurrently ma mb = mkStream $ \st stp sng yld -> do

541 liftIO $ enqueue (fromJust $ streamVar st) mb

542− unStreamShared ma st stp sng yld

542+ foldStreamShared st stp sng yld ma

543

544{-# INLINE joinStreamVarAsync #-}

545−joinStreamVarAsync :: MonadAsync m

545+joinStreamVarAsync :: (IsStream t, MonadAsync m)

546− => SVarStyle -> Stream m a -> Stream m a -> Stream m a

546+ => SVarStyle -> t m a -> t m a -> t m a

547joinStreamVarAsync style m1 m2 = mkStream $ \st stp sng yld ->

548 case streamVar st of

549− Just sv | svarStyle sv == style ->

549+ Just sv | svarStyle sv == style -> do

550− liftIO (enqueue sv m2) >> unStreamShared m1 st stp sng yld

550+ liftIO $ enqueue sv (toStream m2)

551+ foldStreamShared st stp sng yld m1

551− _ -> unStreamShared (forkSVarAsync style m1 m2) st stp sng yld

552+ _ -> foldStreamShared st stp sng yld (forkSVarAsync style m1 m2)

552

553

553------------------------------------------------------------------------------

554------------------------------------------------------------------------------

554-- Semigroup and Monoid style compositions for parallel actions

555-- Semigroup and Monoid style compositions for parallel actions

555------------------------------------------------------------------------------

556------------------------------------------------------------------------------

556

557

557−{-# INLINE asyncS #-}

558−asyncS :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

559−asyncS = joinStreamVarAsync AsyncVar

560−

561-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AsyncT'.

558-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AsyncT'.

562-- Merges two streams possibly concurrently, preferring the

559-- Merges two streams possibly concurrently, preferring the

563-- elements from the left one when available.

560-- elements from the left one when available.

@@ -565,9 +562,7 @@ asyncS = joinStreamVarAsync AsyncVar

565-- @since 0.2.0

562-- @since 0.2.0

566{-# INLINE async #-}

563{-# INLINE async #-}

567async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

564async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

568−async m1 m2 = fromStream $ mkStream $ \st stp sng yld ->

569− unStreamShared (joinStreamVarAsync AsyncVar (toStream m1) (toStream m2))

565+async = joinStreamVarAsync AsyncVar

570− st stp sng yld

571

566

572-- | Same as 'async'.

567-- | Same as 'async'.

573--

568--

@@ -580,8 +575,8 @@ async m1 m2 = fromStream $ mkStream $ \st stp sng yld ->

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

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

581-- of combining streams using async.

576-- of combining streams using async.

582{-# INLINE consMAsync #-}

577{-# INLINE consMAsync #-}

583−consMAsync :: MonadAsync m => m a -> Stream m a -> Stream m a

578+consMAsync :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a

584−consMAsync m r = K.yieldM m `asyncS` r

579+consMAsync m r = K.yieldM m `async` r

585

580

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

581------------------------------------------------------------------------------

587-- AsyncT

582-- AsyncT

@@ -657,7 +652,7 @@ instance IsStream AsyncT where

657

652

658 {-# INLINE consM #-}

653 {-# INLINE consM #-}

659 {-# SPECIALIZE consM :: IO a -> AsyncT IO a -> AsyncT IO a #-}

654 {-# SPECIALIZE consM :: IO a -> AsyncT IO a -> AsyncT IO a #-}

660− consM m r = fromStream $ consMAsync m (toStream r)

655+ consM = consMAsync

661

656

662 {-# INLINE (|:) #-}

657 {-# INLINE (|:) #-}

663 {-# SPECIALIZE (|:) :: IO a -> AsyncT IO a -> AsyncT IO a #-}

658 {-# SPECIALIZE (|:) :: IO a -> AsyncT IO a -> AsyncT IO a #-}

@@ -684,7 +679,7 @@ instance MonadAsync m => Monoid (AsyncT m a) where

684

679

685instance MonadAsync m => Monad (AsyncT m) where

680instance MonadAsync m => Monad (AsyncT m) where

686 return = pure

681 return = pure

687− (AsyncT m) >>= f = AsyncT $ K.bindWith asyncS m (getAsyncT . f)

682+ (>>=) = K.bindWith async

688

683

689------------------------------------------------------------------------------

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

690-- Other instances

685-- Other instances

@@ -697,15 +692,11 @@ MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)

697-- WAsyncT

692-- WAsyncT

698------------------------------------------------------------------------------

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

699

694

700−{-# INLINE wAsyncS #-}

701−wAsyncS :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

702−wAsyncS = joinStreamVarAsync WAsyncVar

703−

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

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

705-- of combining streams using wAsync.

696-- of combining streams using wAsync.

706{-# INLINE consMWAsync #-}

697{-# INLINE consMWAsync #-}

707−consMWAsync :: MonadAsync m => m a -> Stream m a -> Stream m a

698+consMWAsync :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a

708−consMWAsync m r = K.yieldM m `wAsyncS` r

699+consMWAsync m r = K.yieldM m `wAsync` r

709

700

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

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

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

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

@@ -713,8 +704,7 @@ consMWAsync m r = K.yieldM m `wAsyncS` r

713-- @since 0.2.0

704-- @since 0.2.0

714{-# INLINE wAsync #-}

705{-# INLINE wAsync #-}

715wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

706wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

716−wAsync m1 m2 = fromStream $ mkStream $ \st stp sng yld ->

707+wAsync = joinStreamVarAsync WAsyncVar

717− unStreamShared (wAsyncS (toStream m1) (toStream m2)) st stp sng yld

718

708

719-- | Wide async composition or async composition with breadth first traversal.

709-- | Wide async composition or async composition with breadth first traversal.

720-- The Semigroup instance of 'WAsyncT' concurrently /traverses/ the composed

710-- The Semigroup instance of 'WAsyncT' concurrently /traverses/ the composed

@@ -784,7 +774,7 @@ instance IsStream WAsyncT where

784

774

785 {-# INLINE consM #-}

775 {-# INLINE consM #-}

786 {-# SPECIALIZE consM :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}

776 {-# SPECIALIZE consM :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}

787− consM m r = fromStream $ consMWAsync m (toStream r)

777+ consM = consMWAsync

788

778

789 {-# INLINE (|:) #-}

779 {-# INLINE (|:) #-}

790 {-# SPECIALIZE (|:) :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}

780 {-# SPECIALIZE (|:) :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}

@@ -811,8 +801,7 @@ instance MonadAsync m => Monoid (WAsyncT m a) where

811

801

812instance MonadAsync m => Monad (WAsyncT m) where

802instance MonadAsync m => Monad (WAsyncT m) where

813 return = pure

803 return = pure

814− (WAsyncT m) >>= f =

804+ (>>=) = K.bindWith wAsync

815− WAsyncT $ K.bindWith wAsyncS m (getWAsyncT . f)

816

805

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

806------------------------------------------------------------------------------

818-- Other instances

807-- Other instances

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

@@ -52,8 +52,8 @@ import Streamly.Streams.Serial (SerialT)

52-- @since 0.4.0

53{-# INLINE_NORMAL maxThreads #-}

54maxThreads :: IsStream t => Int -> t m a -> t m a

55−maxThreads n m = fromStream $ mkStream $ \st stp sng yld ->

55+maxThreads n m = mkStream $ \st stp sng yld ->

56− unStreamShared (toStream m) (setMaxThreads n st) stp sng yld

56+ foldStreamShared (setMaxThreads n st) stp sng yld m

58{-

59{-# RULES "maxThreadsSerial serial" maxThreads = maxThreadsSerial #-}

@@ -77,8 +77,8 @@ maxThreadsSerial _ = id

77-- @since 0.4.0

78{-# INLINE_NORMAL maxBuffer #-}

79maxBuffer :: IsStream t => Int -> t m a -> t m a

80−maxBuffer n m = fromStream $ mkStream $ \st stp sng yld ->

80+maxBuffer n m = mkStream $ \st stp sng yld ->

81− unStreamShared (toStream m) (setMaxBuffer n st) stp sng yld

81+ foldStreamShared (setMaxBuffer n st) stp sng yld m

83{-

84{-# RULES "maxBuffer serial" maxBuffer = maxBufferSerial #-}

@@ -101,7 +101,7 @@ maxBufferSerial _ = id

101-- @since 0.5.0

102{-# INLINE_NORMAL rate #-}

103rate :: IsStream t => Maybe Rate -> t m a -> t m a

104−rate r m = fromStream $ mkStream $ \st stp sng yld ->

104+rate r m = mkStream $ \st stp sng yld ->

105 case r of

106 Just (Rate low goal _ _) | goal < low ->

107 error "rate: Target rate cannot be lower than minimum rate."

@@ -109,7 +109,7 @@ rate r m = fromStream $ mkStream $ \st stp sng yld ->

109 error "rate: Target rate cannot be greater than maximum rate."

110 Just (Rate low _ high _) | low > high ->

111 error "rate: Minimum rate cannot be greater than maximum rate."

112− _ -> unStreamShared (toStream m) (setStreamRate r st) stp sng yld

112+ _ -> foldStreamShared (setStreamRate r st) stp sng yld m

113

114-- XXX implement for serial streams as well, as a simple delay

115

@@ -181,8 +181,8 @@ constRate r = rate (Just $ Rate r r r 0)

181--

182{-# INLINE_NORMAL _serialLatency #-}

183_serialLatency :: IsStream t => Int -> t m a -> t m a

184−_serialLatency n m = fromStream $ mkStream $ \st stp sng yld ->

184+_serialLatency n m = mkStream $ \st stp sng yld ->

185− unStreamShared (toStream m) (setStreamLatency n st) stp sng yld

185+ foldStreamShared (setStreamLatency n st) stp sng yld m

186

187{-

188{-# RULES "serialLatency serial" _serialLatency = serialLatencySerial #-}

@@ -196,8 +196,8 @@ serialLatencySerial _ = id

196-- inherited by everything in enclosed scope.

197{-# INLINE_NORMAL maxYields #-}

198maxYields :: IsStream t => Maybe Int64 -> t m a -> t m a

199−maxYields n m = fromStream $ mkStream $ \st stp sng yld ->

199+maxYields n m = mkStream $ \st stp sng yld ->

200− unStreamShared (toStream m) (setYieldLimit n st) stp sng yld

200+ foldStreamShared (setYieldLimit n st) stp sng yld m

201

202{-# RULES "maxYields serial" maxYields = maxYieldsSerial #-}

203maxYieldsSerial :: Maybe Int64 -> SerialT m a -> SerialT m a

@@ -212,5 +212,5 @@ printState st = liftIO $ do

212

213-- | Print debug information about an SVar when the stream ends

214inspectMode :: IsStream t => t m a -> t m a

215−inspectMode m = fromStream $ mkStream $ \st stp sng yld ->

215+inspectMode m = mkStream $ \st stp sng yld ->

216− unStreamShared (toStream m) (setInspectMode st) stp sng yld

216+ foldStreamShared (setInspectMode st) stp sng yld m

src/Streamly/Streams/Parallel.hs+19−26

@@ -49,8 +49,8 @@ import Prelude hiding (map)

49import Streamly.Streams.SVar (fromSVar)

50import Streamly.Streams.Serial (map)

51import Streamly.SVar

52−import Streamly.Streams.StreamK (IsStream(..), Stream, mkStream, unStream,

52+import Streamly.Streams.StreamK (IsStream(..), Stream, mkStream, foldStream,

53− unStreamShared, adapt)

53+ foldStreamShared, adapt)

54import qualified Streamly.Streams.StreamK as K

56#include "Instances.hs"

@@ -63,7 +63,7 @@ import qualified Streamly.Streams.StreamK as K

63runOne

64 :: MonadIO m

65 => State Stream m a -> Stream m a -> Maybe WorkerInfo -> m ()

66−runOne st m winfo = unStreamShared m st stop single yieldk

66+runOne st m winfo = foldStreamShared st stop single yieldk m

68 where

@@ -88,32 +88,28 @@ runOne st m winfo = unStreamShared m st stop single yieldk

88 >> withLimitCheck (void $ liftIO $ mrun $ runOne st r winfo)

90{-# NOINLINE forkSVarPar #-}

91−forkSVarPar :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

91+forkSVarPar :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

92forkSVarPar m r = mkStream $ \st stp sng yld -> do

93 sv <- newParallelVar st

94− pushWorkerPar sv (runOne st{streamVar = Just sv} m)

94+ pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream m)

95− pushWorkerPar sv (runOne st{streamVar = Just sv} r)

95+ pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream r)

96− unStream (fromSVar sv) st stp sng yld

96+ foldStream st stp sng yld (fromSVar sv)

98{-# INLINE joinStreamVarPar #-}

99−joinStreamVarPar :: MonadAsync m

99+joinStreamVarPar :: (IsStream t, MonadAsync m)

100− => SVarStyle -> Stream m a -> Stream m a -> Stream m a

100+ => SVarStyle -> t m a -> t m a -> t m a

101joinStreamVarPar style m1 m2 = mkStream $ \st stp sng yld ->

102 case streamVar st of

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

104− pushWorkerPar sv (runOne st m1)

104+ pushWorkerPar sv (runOne st $ toStream m1)

105− unStreamShared m2 st stp sng yld

105+ foldStreamShared st stp sng yld m2

106− _ -> unStreamShared (forkSVarPar m1 m2) st stp sng yld

106+ _ -> foldStreamShared st stp sng yld (forkSVarPar m1 m2)

107−

108−{-# INLINE parallelStream #-}

109−parallelStream :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

110−parallelStream = joinStreamVarPar ParallelVar

111

107

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

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

113-- of combining streams using parallel.

109-- of combining streams using parallel.

114{-# INLINE consMParallel #-}

110{-# INLINE consMParallel #-}

115−consMParallel :: MonadAsync m => m a -> Stream m a -> Stream m a

111+consMParallel :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a

116−consMParallel m r = K.yieldM m `parallelStream` r

112+consMParallel m r = K.yieldM m `parallel` r

117

113

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

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

119-- Merges two streams concurrently.

115-- Merges two streams concurrently.

@@ -121,9 +117,7 @@ consMParallel m r = K.yieldM m `parallelStream` r

121-- @since 0.2.0

117-- @since 0.2.0

122{-# INLINE parallel #-}

118{-# INLINE parallel #-}

123parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

119parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

124−parallel m1 m2 = fromStream $ mkStream $ \st stp sng yld ->

120+parallel = joinStreamVarPar ParallelVar

125− unStreamShared (parallelStream (toStream m1) (toStream m2))

126− st stp sng yld

127

121

128------------------------------------------------------------------------------

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

129-- Convert a stream to parallel

123-- Convert a stream to parallel

@@ -141,10 +135,10 @@ mkParallel m = do

141

135

142{-# INLINE applyWith #-}

136{-# INLINE applyWith #-}

143applyWith :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b

137applyWith :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b

144−applyWith f m = fromStream $ mkStream $ \st stp sng yld -> do

138+applyWith f m = mkStream $ \st stp sng yld -> do

145 sv <- newParallelVar (adaptState st)

139 sv <- newParallelVar (adaptState st)

146 pushWorkerPar sv (runOne st{streamVar = Just sv} (toStream m))

140 pushWorkerPar sv (runOne st{streamVar = Just sv} (toStream m))

147− unStream (toStream $ f $ fromSVar sv) st stp sng yld

141+ foldStream st stp sng yld $ f $ fromSVar sv

148

142

149------------------------------------------------------------------------------

143------------------------------------------------------------------------------

150-- Stream runner concurrent function application

144-- Stream runner concurrent function application

@@ -346,7 +340,7 @@ instance IsStream ParallelT where

346

340

347 {-# INLINE consM #-}

341 {-# INLINE consM #-}

348 {-# SPECIALIZE consM :: IO a -> ParallelT IO a -> ParallelT IO a #-}

342 {-# SPECIALIZE consM :: IO a -> ParallelT IO a -> ParallelT IO a #-}

349− consM m r = fromStream $ consMParallel m (toStream r)

343+ consM = consMParallel

350

344

351 {-# INLINE (|:) #-}

345 {-# INLINE (|:) #-}

352 {-# SPECIALIZE (|:) :: IO a -> ParallelT IO a -> ParallelT IO a #-}

346 {-# SPECIALIZE (|:) :: IO a -> ParallelT IO a -> ParallelT IO a #-}

@@ -373,8 +367,7 @@ instance MonadAsync m => Monoid (ParallelT m a) where

373

367

374instance MonadAsync m => Monad (ParallelT m) where

368instance MonadAsync m => Monad (ParallelT m) where

375 return = pure

369 return = pure

376− (ParallelT m) >>= f

370+ (>>=) = K.bindWith parallel

377− = ParallelT $ K.bindWith parallelStream m (getParallelT . f)

378

371

379------------------------------------------------------------------------------

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

380-- Other instances

373-- Other instances

src/Streamly/Streams/SVar.hs+7−6

@@ -43,7 +43,7 @@ fromStreamVar sv = mkStream $ \st stp sng yld -> do

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

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

45 -- streams.

46− unStream (processEvents $ reverse list) st stp sng yld

46+ foldStream st stp sng yld $ processEvents $ reverse list

48 where

@@ -59,7 +59,7 @@ fromStreamVar sv = mkStream $ \st stp sng yld -> do

59 done <- postProcess sv

60 if done

61 then allDone stp

62− else unStream (fromStreamVar sv) st stp sng yld

62+ else foldStream st stp sng yld $ fromStreamVar sv

64 processEvents (ev : es) = mkStream $ \st stp sng yld -> do

65 let rest = processEvents es

@@ -68,23 +68,24 @@ fromStreamVar sv = mkStream $ \st stp sng yld -> do

68 ChildStop tid e -> do

69 accountThread sv tid

70 case e of

71− Nothing -> unStream rest st stp sng yld

71+ Nothing -> foldStream st stp sng yld rest

72 Just ex ->

73 case fromException ex of

74 Just ThreadAbort ->

75− unStream rest st stp sng yld

75+ foldStream st stp sng yld rest

76 Nothing -> liftIO (cleanupSVar sv) >> throwM ex

78{-# INLINE fromSVar #-}

79fromSVar :: (MonadAsync m, IsStream t) => SVar Stream m a -> t m a

80fromSVar sv =

81− fromStream $ mkStream $ \st stp sng yld -> do

81+ mkStream $ \st stp sng yld -> do

82 ref <- liftIO $ newIORef ()

83 _ <- liftIO $ mkWeakIORef ref hook

84 -- We pass a copy of sv to fromStreamVar, so that we know that it has

85 -- no other references, when that copy gets garbage collected "ref"

86 -- will get garbage collected and our hook will be called.

87+ foldStreamShared st stp sng yld $

87− unStreamShared (fromStreamVar sv{svarRef = Just ref}) st stp sng yld

88+ fromStream $ fromStreamVar sv{svarRef = Just ref}

88 where

89 where

90 hook = do

91 hook = do

src/Streamly/Streams/Serial.hs+30−42

@@ -24,7 +24,7 @@ module Streamly.Streams.Serial

24 SerialT

25 , StreamT -- deprecated

26 , Serial

27− , serial

27+ , K.serial

28 , serially

30 -- * Serial interleaving stream

@@ -60,8 +60,9 @@ import Text.Read (Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec,

60 readListPrecDefault)

61import Prelude hiding (map, mapM)

63+import Streamly.SVar (adaptState)

63import Streamly.Streams.StreamK (IsStream(..), adapt, Stream, mkStream,

64import Streamly.Streams.StreamK (IsStream(..), adapt, Stream, mkStream,

64− unStream)

65+ foldStream)

65import qualified Streamly.Streams.Prelude as P

66import qualified Streamly.Streams.Prelude as P

66import qualified Streamly.Streams.StreamK as K

67import qualified Streamly.Streams.StreamK as K

67import qualified Streamly.Streams.StreamD as D

68import qualified Streamly.Streams.StreamD as D

@@ -156,39 +157,28 @@ instance IsStream SerialT where

156 {-# INLINE consM #-}

157 {-# INLINE consM #-}

157 {-# SPECIALIZE consM :: IO a -> SerialT IO a -> SerialT IO a #-}

158 {-# SPECIALIZE consM :: IO a -> SerialT IO a -> SerialT IO a #-}

158 consM :: Monad m => m a -> SerialT m a -> SerialT m a

159 consM :: Monad m => m a -> SerialT m a -> SerialT m a

159− consM m r = fromStream $ K.consMSerial m (toStream r)

160+ consM = K.consMSerial

160

161

161 {-# INLINE (|:) #-}

162 {-# INLINE (|:) #-}

162 {-# SPECIALIZE (|:) :: IO a -> SerialT IO a -> SerialT IO a #-}

163 {-# SPECIALIZE (|:) :: IO a -> SerialT IO a -> SerialT IO a #-}

163 (|:) :: Monad m => m a -> SerialT m a -> SerialT m a

164 (|:) :: Monad m => m a -> SerialT m a -> SerialT m a

164− m |: r = fromStream $ K.consMSerial m (toStream r)

165+ (|:) = K.consMSerial

165−

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

167−-- Semigroup

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

169−

170−-- | Polymorphic version of the 'Semigroup' operation '<>' of 'SerialT'.

171−-- Appends two streams sequentially, yielding all elements from the first

172−-- stream, and then all elements from the second stream.

173−--

174−-- @since 0.2.0

175−{-# INLINE serial #-}

176−serial :: IsStream t => t m a -> t m a -> t m a

177−serial m1 m2 = fromStream $ mkStream $ \st stp sng yld ->

178− unStream (K.serial (toStream m1) (toStream m2))

179− st stp sng yld

180

166

181------------------------------------------------------------------------------

167------------------------------------------------------------------------------

182-- Monad

168-- Monad

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

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

184

170

171+-- XXX if foldStreamShared/unShare/adaptState are not efficient in serial case

172+-- we can use a rewrite rule for that for the serial case and then just use a

173+-- bindWith serial here.

174+--

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

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

186 return = pure

176 return = pure

187− (SerialT m) >>= f = SerialT $ mkStream $ \st stp sng yld ->

177+ m >>= f = mkStream $ \st stp sng yld ->

188− let run x = unStream x st stp sng yld

178+ let run x = foldStream st stp sng yld x

189− single a = run $ toStream (f a)

179+ single a = run $ f a

190− yieldk a r = run $ toStream $ f a <> (fromStream r >>= f)

180+ yieldk a r = run $ f a <> (r >>= f)

191− in unStream m st stp single yieldk

181+ in foldStream (adaptState st) stp single yieldk m

192

182

193------------------------------------------------------------------------------

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

194-- Other instances

184-- Other instances

@@ -293,34 +283,28 @@ instance IsStream WSerialT where

293 {-# INLINE consM #-}

283 {-# INLINE consM #-}

294 {-# SPECIALIZE consM :: IO a -> WSerialT IO a -> WSerialT IO a #-}

284 {-# SPECIALIZE consM :: IO a -> WSerialT IO a -> WSerialT IO a #-}

295 consM :: Monad m => m a -> WSerialT m a -> WSerialT m a

285 consM :: Monad m => m a -> WSerialT m a -> WSerialT m a

296− consM m r = fromStream $ K.consMSerial m (toStream r)

286+ consM = K.consMSerial

297

287

298 {-# INLINE (|:) #-}

288 {-# INLINE (|:) #-}

299 {-# SPECIALIZE (|:) :: IO a -> WSerialT IO a -> WSerialT IO a #-}

289 {-# SPECIALIZE (|:) :: IO a -> WSerialT IO a -> WSerialT IO a #-}

300 (|:) :: Monad m => m a -> WSerialT m a -> WSerialT m a

290 (|:) :: Monad m => m a -> WSerialT m a -> WSerialT m a

301− m |: r = fromStream $ K.consMSerial m (toStream r)

291+ (|:) = K.consMSerial

302

292

303------------------------------------------------------------------------------

293------------------------------------------------------------------------------

304-- Semigroup

294-- Semigroup

305------------------------------------------------------------------------------

295------------------------------------------------------------------------------

306

296

307−{-# INLINE interleave #-}

308−interleave :: Stream m a -> Stream m a -> Stream m a

309−interleave m1 m2 = mkStream $ \st stp sng yld -> do

310− let stop = unStream m2 st stp sng yld

311− single a = yld a m2

312− yieldk a r = yld a (interleave m2 r)

313− unStream m1 st stop single yieldk

314−

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

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

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

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

317--

299--

318-- @since 0.2.0

300-- @since 0.2.0

319{-# INLINE wSerial #-}

301{-# INLINE wSerial #-}

320wSerial :: IsStream t => t m a -> t m a -> t m a

302wSerial :: IsStream t => t m a -> t m a -> t m a

321−wSerial m1 m2 = fromStream $ mkStream $ \st stp sng yld ->

303+wSerial m1 m2 = mkStream $ \st stp sng yld -> do

322− unStream (interleave (toStream m1) (toStream m2))

304+ let stop = foldStream st stp sng yld m2

323− st stp sng yld

305+ single a = yld a m2

306+ yieldk a r = yld a (wSerial m2 r)

307+ foldStream st stop single yieldk m1

324

308

325instance Semigroup (WSerialT m a) where

309instance Semigroup (WSerialT m a) where

326 (<>) = wSerial

310 (<>) = wSerial

@@ -347,13 +331,17 @@ instance Monoid (WSerialT m a) where

347-- Monad

331-- Monad

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

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

349

333

334+-- XXX if foldStreamShared/unShare/adaptState are not efficient in serial case

335+-- we can use a rewrite rule for that for the serial case and then just use a

336+-- bindWith interleave here.

337+--

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

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

351 return = pure

339 return = pure

352− (WSerialT m) >>= f = WSerialT $ mkStream $ \st stp sng yld ->

340+ m >>= f = mkStream $ \st stp sng yld ->

353− let run x = unStream x st stp sng yld

341+ let run x = foldStream st stp sng yld x

354− single a = run $ toStream (f a)

342+ single a = run $ f a

355− yieldk a r = run $ toStream $ f a <> (fromStream r >>= f)

343+ yieldk a r = run $ f a <> (r >>= f)

356− in unStream m st stp single yieldk

344+ in foldStream (adaptState st) stp single yieldk m

357

345

358------------------------------------------------------------------------------

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

359-- Other instances

347-- Other instances

src/Streamly/Streams/StreamD.hs+2−2

@@ -483,7 +483,7 @@ fromStreamK = Stream step

483 let stop = return Stop

484 single a = return $ Yield a K.nil

485 yieldk a r = return $ Yield a r

486− in K.unStreamShared m1 gst stop single yieldk

486+ in K.foldStreamShared gst stop single yieldk m1

487

488{-# INLINE toStreamD #-}

489toStreamD :: (K.IsStream t, Monad m) => t m a -> Stream m a

@@ -883,7 +883,7 @@ toStreamK (Stream step state) = go state

883 r <- step gst st

884 case r of

885 Yield x s -> yld x (go s)

886− Skip s -> K.unStreamShared (go s) gst stp sng yld

886+ Skip s -> K.foldStreamShared gst stp sng yld $ go s

887 Stop -> stp

888

889#ifndef DISABLE_FUSION

src/Streamly/Streams/StreamK.hs+161−165

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

6{-# LANGUAGE InstanceSigs #-}

7{-# LANGUAGE MultiParamTypeClasses #-}

8{-# LANGUAGE RankNTypes #-}

9+{-# LANGUAGE ScopedTypeVariables #-}

9{-# LANGUAGE UndecidableInstances #-} -- XXX

10{-# LANGUAGE UndecidableInstances #-} -- XXX

11-- |

12-- |

@@ -33,17 +34,21 @@ module Streamly.Streams.StreamK

34 -- * The stream type

35 -- * The stream type

35 , Stream

36 , Stream

36− , unStream

37− , unStreamShared

38− , unShare

39− , runStreamSVar

41− -- * Construction

38+ -- * Construction Primitives

42 , mkStream

39 , mkStream

43 , nil

40 , nil

44 , cons

41 , cons

45 , (.:)

42 , (.:)

44+ -- * Elimination Primitives

45+ , foldStream

46+ , foldStreamShared

47+ , foldStreamSVar

48+

49+ -- * Transformation Primitives

50+ , unShare

51+

47 -- * Asynchronous construction

52 -- * Asynchronous construction

48 , nilK

53 , nilK

49 , yieldK

54 , yieldK

@@ -71,7 +76,6 @@ module Streamly.Streams.StreamK

72 -- * Elimination

77 -- * Elimination

73 -- ** General Folds

78 -- ** General Folds

74− , foldStream

75 , foldr

79 , foldr

76 , foldrM

80 , foldrM

77 , foldr1

81 , foldr1

@@ -174,9 +178,9 @@ import Streamly.Streams.StreamKType

174

178

175-- | Detach a stream from an SVar

179-- | Detach a stream from an SVar

176{-# INLINE unShare #-}

180{-# INLINE unShare #-}

177−unShare :: Stream m a -> Stream m a

181+unShare :: IsStream t => t m a -> t m a

178unShare x = mkStream $ \st stp sng yld ->

182unShare x = mkStream $ \st stp sng yld ->

179− unStream x st stp sng yld

183+ foldStream st stp sng yld x

180

184

181------------------------------------------------------------------------------

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

182-- Construction

186-- Construction

@@ -197,7 +201,7 @@ infixr 5 `cons`

197--

201--

198-- @since 0.1.0

202-- @since 0.1.0

199cons :: IsStream t => a -> t m a -> t m a

203cons :: IsStream t => a -> t m a -> t m a

200−cons a r = fromStream $ mkStream $ \_ _ _ yld -> yld a (toStream r)

204+cons a r = mkStream $ \_ _ _ yld -> yld a r

201

205

202infixr 5 .:

206infixr 5 .:

203

207

@@ -218,15 +222,15 @@ infixr 5 .:

218

222

219-- | Make an empty stream from a callback function.

223-- | Make an empty stream from a callback function.

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

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

221−nilK k = fromStream $ mkStream $ \_ stp _ _ -> k stp

225+nilK k = mkStream $ \_ stp _ _ -> k stp

222

226

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

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

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

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

225−yieldK k = fromStream $ mkStream $ \_ _ sng _ -> k sng

229+yieldK k = mkStream $ \_ _ sng _ -> k sng

226

230

227-- | Construct a stream from a callback function.

231-- | Construct a stream from a callback function.

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

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

229−consK k r = fromStream $ mkStream $ \_ _ _ yld -> k (\x -> yld x (toStream r))

233+consK k r = mkStream $ \_ _ _ yld -> k (\x -> yld x r)

230

234

231-- XXX consK with concurrent callbacks

235-- XXX consK with concurrent callbacks

232-- XXX Build a stream from a repeating callback function.

236-- XXX Build a stream from a repeating callback function.

@@ -240,8 +244,8 @@ uncons :: (IsStream t, Monad m) => t m a -> m (Maybe (a, t m a))

240uncons m =

244uncons m =

241 let stop = return Nothing

245 let stop = return Nothing

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

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

243− yieldk a r = return (Just (a, fromStream r))

247+ yieldk a r = return (Just (a, r))

244− in unStream (toStream m) defState stop single yieldk

248+ in foldStream defState stop single yieldk m

245

249

246-------------------------------------------------------------------------------

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

247-- Generation

251-- Generation

@@ -249,7 +253,7 @@ uncons m =

249

253

250{-# INLINE unfoldr #-}

254{-# INLINE unfoldr #-}

251unfoldr :: IsStream t => (b -> Maybe (a, b)) -> b -> t m a

255unfoldr :: IsStream t => (b -> Maybe (a, b)) -> b -> t m a

252−unfoldr step = fromStream . go

256+unfoldr step = go

253 where

257 where

254 go s = mkStream $ \_ stp _ yld ->

258 go s = mkStream $ \_ stp _ yld ->

255 case step s of

259 case step s of

@@ -260,19 +264,19 @@ unfoldr step = fromStream . go

260unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a

264unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a

261unfoldrM step = go

265unfoldrM step = go

262 where

266 where

263− go s = fromStream $ mkStream $ \svr stp sng yld -> do

267+ go s = mkStream $ \st stp sng yld -> do

264 mayb <- step s

268 mayb <- step s

265 case mayb of

269 case mayb of

266 Nothing -> stp

270 Nothing -> stp

267 Just (a, b) ->

271 Just (a, b) ->

268− unStreamShared (toStream (return a |: go b)) svr stp sng yld

272+ foldStreamShared st stp sng yld $ return a |: go b

269

273

270-------------------------------------------------------------------------------

274-------------------------------------------------------------------------------

271-- Special generation

275-- Special generation

272-------------------------------------------------------------------------------

276-------------------------------------------------------------------------------

273

277

274yield :: IsStream t => a -> t m a

278yield :: IsStream t => a -> t m a

275−yield a = fromStream $ mkStream $ \_ _ single _ -> single a

279+yield a = mkStream $ \_ _ single _ -> single a

276

280

277-- | Same as yieldM

281-- | Same as yieldM

278--

282--

@@ -325,8 +329,8 @@ fromList :: IsStream t => [a] -> t m a

325fromList = fromFoldable

329fromList = fromFoldable

326

330

327{-# INLINE fromStreamK #-}

331{-# INLINE fromStreamK #-}

328−fromStreamK :: Stream m a -> Stream m a

332+fromStreamK :: IsStream t => Stream m a -> t m a

329−fromStreamK = id

333+fromStreamK = fromStream

330

334

331-------------------------------------------------------------------------------

335-------------------------------------------------------------------------------

332-- Elimination by Folding

336-- Elimination by Folding

@@ -334,24 +338,24 @@ fromStreamK = id

334

338

335-- | Lazy right associative fold.

339-- | Lazy right associative fold.

336foldr :: (IsStream t, Monad m) => (a -> b -> b) -> b -> t m a -> m b

340foldr :: (IsStream t, Monad m) => (a -> b -> b) -> b -> t m a -> m b

337−foldr step acc m = go (toStream m)

341+foldr step acc m = go m

338 where

342 where

339 go m1 =

343 go m1 =

340 let stop = return acc

344 let stop = return acc

341 single a = return (step a acc)

345 single a = return (step a acc)

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

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

343− in unStream m1 defState stop single yieldk

347+ in foldStream defState stop single yieldk m1

344

348

345-- | Lazy right fold with a monadic step function.

349-- | Lazy right fold with a monadic step function.

346{-# INLINE foldrM #-}

350{-# INLINE foldrM #-}

347foldrM :: (IsStream t, Monad m) => (a -> b -> m b) -> b -> t m a -> m b

351foldrM :: (IsStream t, Monad m) => (a -> b -> m b) -> b -> t m a -> m b

348−foldrM step acc m = go (toStream m)

352+foldrM step acc m = go m

349 where

353 where

350 go m1 =

354 go m1 =

351 let stop = return acc

355 let stop = return acc

352 single a = step a acc

356 single a = step a acc

353 yieldk a r = go r >>= step a

357 yieldk a r = go r >>= step a

354− in unStream m1 defState stop single yieldk

358+ in foldStream defState stop single yieldk m1

355

359

356{-# INLINE foldr1 #-}

360{-# INLINE foldr1 #-}

357foldr1 :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> m (Maybe a)

361foldr1 :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> m (Maybe a)

@@ -359,13 +363,13 @@ foldr1 step m = do

359 r <- uncons m

363 r <- uncons m

360 case r of

364 case r of

361 Nothing -> return Nothing

365 Nothing -> return Nothing

362− Just (h, t) -> fmap Just (go h (toStream t))

366+ Just (h, t) -> fmap Just (go h t)

363 where

367 where

364 go p m1 =

368 go p m1 =

365 let stp = return p

369 let stp = return p

366 single a = return $ step a p

370 single a = return $ step a p

367 yieldk a r = fmap (step p) (go a r)

371 yieldk a r = fmap (step p) (go a r)

368− in unStream m1 defState stp single yieldk

372+ in foldStream defState stp single yieldk m1

369

373

370-- | Strict left fold with an extraction function. Like the standard strict

374-- | Strict left fold with an extraction function. Like the standard strict

371-- left fold, but applies a user supplied extraction function (the third

375-- left fold, but applies a user supplied extraction function (the third

@@ -374,28 +378,31 @@ foldr1 step m = do

374--

378--

375-- Note that the accumulator is always evaluated including the initial value.

379-- Note that the accumulator is always evaluated including the initial value.

376{-# INLINE foldx #-}

380{-# INLINE foldx #-}

377−foldx :: (IsStream t, Monad m)

381+foldx :: forall t m a b x. (IsStream t, Monad m)

378 => (x -> a -> x) -> x -> (x -> b) -> t m a -> m b

382 => (x -> a -> x) -> x -> (x -> b) -> t m a -> m b

379−foldx step begin done m = get $ go (toStream m) begin

383+foldx step begin done m = get $ go m begin

380 where

384 where

381 {-# NOINLINE get #-}

385 {-# NOINLINE get #-}

386+ get :: t m x -> m b

382 get m1 =

387 get m1 =

383 -- XXX we are not strictly evaluating the accumulator here. Is this

388 -- XXX we are not strictly evaluating the accumulator here. Is this

384 -- okay?

389 -- okay?

385 let single = return . done

390 let single = return . done

391+ -- XXX this is foldSingleton. why foldStreamShared?

386− in unStreamShared m1 undefined undefined single undefined

392+ in foldStreamShared undefined undefined single undefined m1

387

393

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

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

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

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

390 -- that cannot be tail call optimized. Unfolding recursion explicitly via

396 -- that cannot be tail call optimized. Unfolding recursion explicitly via

391 -- continuations is much more efficient.

397 -- continuations is much more efficient.

398+ go :: t m a -> x -> t m x

392 go m1 !acc = mkStream $ \_ _ sng yld ->

399 go m1 !acc = mkStream $ \_ _ sng yld ->

393 let stop = sng acc

400 let stop = sng acc

394 single a = sng $ step acc a

401 single a = sng $ step acc a

395− yieldk a r =

402+ -- XXX this is foldNonEmptyStream

403+ yieldk a r = foldStream defState undefined sng yld $

396− let stream = go r (step acc a)

404+ go r (step acc a)

397− in unStream stream defState undefined sng yld

398− in unStream m1 defState stop single yieldk

405+ in foldStream defState stop single yieldk m1

399

406

400-- | Strict left associative fold.

407-- | Strict left associative fold.

401{-# INLINE foldl' #-}

408{-# INLINE foldl' #-}

@@ -406,13 +413,13 @@ foldl' step begin = foldx step begin id

406-- | Like 'foldx', but with a monadic step function.

413-- | Like 'foldx', but with a monadic step function.

407foldxM :: (IsStream t, Monad m)

414foldxM :: (IsStream t, Monad m)

408 => (x -> a -> m x) -> m x -> (x -> m b) -> t m a -> m b

415 => (x -> a -> m x) -> m x -> (x -> m b) -> t m a -> m b

409−foldxM step begin done m = go begin (toStream m)

416+foldxM step begin done m = go begin m

410 where

417 where

411 go !acc m1 =

418 go !acc m1 =

412 let stop = acc >>= done

419 let stop = acc >>= done

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

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

414 yieldk a r = acc >>= \b -> step b a >>= \x -> go (return x) r

421 yieldk a r = acc >>= \b -> step b a >>= \x -> go (return x) r

415− in unStream m1 defState stop single yieldk

422+ in foldStream defState stop single yieldk m1

416

423

417-- | Like 'foldl'' but with a monadic step function.

424-- | Like 'foldl'' but with a monadic step function.

418foldlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> m b

425foldlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> m b

@@ -424,13 +431,13 @@ foldlM' step begin = foldxM step (return begin) return

424

431

425{-# INLINE runStream #-}

432{-# INLINE runStream #-}

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

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

427−runStream m = go (toStream m)

434+runStream = go

428 where

435 where

429 go m1 =

436 go m1 =

430 let stop = return ()

437 let stop = return ()

431 single _ = return ()

438 single _ = return ()

432− yieldk _ r = go (toStream r)

439+ yieldk _ r = go r

433− in unStream m1 defState stop single yieldk

440+ in foldStream defState stop single yieldk m1

434

441

435{-# INLINE null #-}

442{-# INLINE null #-}

436null :: (IsStream t, Monad m) => t m a -> m Bool

443null :: (IsStream t, Monad m) => t m a -> m Bool

@@ -438,7 +445,7 @@ null m =

438 let stop = return True

445 let stop = return True

439 single _ = return False

446 single _ = return False

440 yieldk _ _ = return False

447 yieldk _ _ = return False

441− in unStream (toStream m) defState stop single yieldk

448+ in foldStream defState stop single yieldk m

442

449

443{-# INLINE head #-}

450{-# INLINE head #-}

444head :: (IsStream t, Monad m) => t m a -> m (Maybe a)

451head :: (IsStream t, Monad m) => t m a -> m (Maybe a)

@@ -446,69 +453,69 @@ head m =

446 let stop = return Nothing

453 let stop = return Nothing

447 single a = return (Just a)

454 single a = return (Just a)

448 yieldk a _ = return (Just a)

455 yieldk a _ = return (Just a)

449− in unStream (toStream m) defState stop single yieldk

456+ in foldStream defState stop single yieldk m

450

457

451{-# INLINE tail #-}

458{-# INLINE tail #-}

452tail :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))

459tail :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))

453tail m =

460tail m =

454 let stop = return Nothing

461 let stop = return Nothing

455 single _ = return $ Just nil

462 single _ = return $ Just nil

456− yieldk _ r = return $ Just $ fromStream r

463+ yieldk _ r = return $ Just r

457− in unStream (toStream m) defState stop single yieldk

464+ in foldStream defState stop single yieldk m

458

465

459{-# INLINE init #-}

466{-# INLINE init #-}

460init :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))

467init :: (IsStream t, Monad m) => t m a -> m (Maybe (t m a))

461−init m = go1 (toStream m)

468+init m = go1 m

462 where

469 where

463 go1 m1 = do

470 go1 m1 = do

464 r <- uncons m1

471 r <- uncons m1

465 case r of

472 case r of

466 Nothing -> return Nothing

473 Nothing -> return Nothing

467− Just (h, t) -> return . Just . fromStream $ go h t

474+ Just (h, t) -> return . Just $ go h t

468 go p m1 = mkStream $ \_ stp sng yld ->

475 go p m1 = mkStream $ \_ stp sng yld ->

469 let single _ = sng p

476 let single _ = sng p

470 yieldk a x = yld p $ go a x

477 yieldk a x = yld p $ go a x

471− in unStream m1 defState stp single yieldk

478+ in foldStream defState stp single yieldk m1

472

479

473{-# INLINE elem #-}

480{-# INLINE elem #-}

474elem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool

481elem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool

475−elem e m = go (toStream m)

482+elem e m = go m

476 where

483 where

477 go m1 =

484 go m1 =

478 let stop = return False

485 let stop = return False

479 single a = return (a == e)

486 single a = return (a == e)

480 yieldk a r = if a == e then return True else go r

487 yieldk a r = if a == e then return True else go r

481− in unStream m1 defState stop single yieldk

488+ in foldStream defState stop single yieldk m1

482

489

483{-# INLINE notElem #-}

490{-# INLINE notElem #-}

484notElem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool

491notElem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool

485−notElem e m = go (toStream m)

492+notElem e m = go m

486 where

493 where

487 go m1 =

494 go m1 =

488 let stop = return True

495 let stop = return True

489 single a = return (a /= e)

496 single a = return (a /= e)

490 yieldk a r = if a == e then return False else go r

497 yieldk a r = if a == e then return False else go r

491− in unStream m1 defState stop single yieldk

498+ in foldStream defState stop single yieldk m1

492

499

493all :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool

500all :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool

494−all p m = go (toStream m)

501+all p m = go m

495 where

502 where

496 go m1 =

503 go m1 =

497 let single a | p a = return True

504 let single a | p a = return True

498 | otherwise = return False

505 | otherwise = return False

499 yieldk a r | p a = go r

506 yieldk a r | p a = go r

500 | otherwise = return False

507 | otherwise = return False

501− in unStream m1 defState (return True) single yieldk

508+ in foldStream defState (return True) single yieldk m1

502

509

503any :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool

510any :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m Bool

504−any p m = go (toStream m)

511+any p m = go m

505 where

512 where

506 go m1 =

513 go m1 =

507 let single a | p a = return True

514 let single a | p a = return True

508 | otherwise = return False

515 | otherwise = return False

509 yieldk a r | p a = return True

516 yieldk a r | p a = return True

510 | otherwise = go r

517 | otherwise = go r

511− in unStream m1 defState (return False) single yieldk

518+ in foldStream defState (return False) single yieldk m1

512

519

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

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

514{-# INLINE last #-}

521{-# INLINE last #-}

@@ -517,13 +524,13 @@ last = foldx (\_ y -> Just y) Nothing id

517

524

518{-# INLINE minimum #-}

525{-# INLINE minimum #-}

519minimum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)

526minimum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)

520−minimum m = go Nothing (toStream m)

527+minimum m = go Nothing m

521 where

528 where

522 go Nothing m1 =

529 go Nothing m1 =

523 let stop = return Nothing

530 let stop = return Nothing

524 single a = return (Just a)

531 single a = return (Just a)

525 yieldk a r = go (Just a) r

532 yieldk a r = go (Just a) r

526− in unStream m1 defState stop single yieldk

533+ in foldStream defState stop single yieldk m1

527

534

528 go (Just res) m1 =

535 go (Just res) m1 =

529 let stop = return (Just res)

536 let stop = return (Just res)

@@ -535,19 +542,19 @@ minimum m = go Nothing (toStream m)

535 if res <= a

542 if res <= a

536 then go (Just res) r

543 then go (Just res) r

537 else go (Just a) r

544 else go (Just a) r

538− in unStream m1 defState stop single yieldk

545+ in foldStream defState stop single yieldk m1

539

546

540{-# INLINE minimumBy #-}

547{-# INLINE minimumBy #-}

541minimumBy

548minimumBy

542 :: (IsStream t, Monad m)

549 :: (IsStream t, Monad m)

543 => (a -> a -> Ordering) -> t m a -> m (Maybe a)

550 => (a -> a -> Ordering) -> t m a -> m (Maybe a)

544−minimumBy cmp m = go Nothing (toStream m)

551+minimumBy cmp m = go Nothing m

545 where

552 where

546 go Nothing m1 =

553 go Nothing m1 =

547 let stop = return Nothing

554 let stop = return Nothing

548 single a = return (Just a)

555 single a = return (Just a)

549 yieldk a r = go (Just a) r

556 yieldk a r = go (Just a) r

550− in unStream m1 defState stop single yieldk

557+ in foldStream defState stop single yieldk m1

551

558

552 go (Just res) m1 =

559 go (Just res) m1 =

553 let stop = return (Just res)

560 let stop = return (Just res)

@@ -557,17 +564,17 @@ minimumBy cmp m = go Nothing (toStream m)

557 yieldk a r = case cmp res a of

564 yieldk a r = case cmp res a of

558 GT -> go (Just a) r

565 GT -> go (Just a) r

559 _ -> go (Just res) r

566 _ -> go (Just res) r

560− in unStream m1 defState stop single yieldk

567+ in foldStream defState stop single yieldk m1

561

568

562{-# INLINE maximum #-}

569{-# INLINE maximum #-}

563maximum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)

570maximum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)

564−maximum m = go Nothing (toStream m)

571+maximum m = go Nothing m

565 where

572 where

566 go Nothing m1 =

573 go Nothing m1 =

567 let stop = return Nothing

574 let stop = return Nothing

568 single a = return (Just a)

575 single a = return (Just a)

569 yieldk a r = go (Just a) r

576 yieldk a r = go (Just a) r

570− in unStream m1 defState stop single yieldk

577+ in foldStream defState stop single yieldk m1

571

578

572 go (Just res) m1 =

579 go (Just res) m1 =

573 let stop = return (Just res)

580 let stop = return (Just res)

@@ -579,17 +586,17 @@ maximum m = go Nothing (toStream m)

579 if res <= a

586 if res <= a

580 then go (Just a) r

587 then go (Just a) r

581 else go (Just res) r

588 else go (Just res) r

582− in unStream m1 defState stop single yieldk

589+ in foldStream defState stop single yieldk m1

583

590

584{-# INLINE maximumBy #-}

591{-# INLINE maximumBy #-}

585maximumBy :: (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> m (Maybe a)

592maximumBy :: (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> m (Maybe a)

586−maximumBy cmp m = go Nothing (toStream m)

593+maximumBy cmp m = go Nothing m

587 where

594 where

588 go Nothing m1 =

595 go Nothing m1 =

589 let stop = return Nothing

596 let stop = return Nothing

590 single a = return (Just a)

597 single a = return (Just a)

591 yieldk a r = go (Just a) r

598 yieldk a r = go (Just a) r

592− in unStream m1 defState stop single yieldk

599+ in foldStream defState stop single yieldk m1

593

600

594 go (Just res) m1 =

601 go (Just res) m1 =

595 let stop = return (Just res)

602 let stop = return (Just res)

@@ -599,11 +606,11 @@ maximumBy cmp m = go Nothing (toStream m)

599 yieldk a r = case cmp res a of

606 yieldk a r = case cmp res a of

600 GT -> go (Just res) r

607 GT -> go (Just res) r

601 _ -> go (Just a) r

608 _ -> go (Just a) r

602− in unStream m1 defState stop single yieldk

609+ in foldStream defState stop single yieldk m1

603

610

604{-# INLINE (!!) #-}

611{-# INLINE (!!) #-}

605(!!) :: (IsStream t, Monad m) => t m a -> Int -> m (Maybe a)

612(!!) :: (IsStream t, Monad m) => t m a -> Int -> m (Maybe a)

606−m !! i = go i (toStream m)

613+m !! i = go i m

607 where

614 where

608 go n m1 =

615 go n m1 =

609 let single a | n == 0 = return $ Just a

616 let single a | n == 0 = return $ Just a

@@ -611,22 +618,22 @@ m !! i = go i (toStream m)

611 yieldk a x | n < 0 = return Nothing

618 yieldk a x | n < 0 = return Nothing

612 | n == 0 = return $ Just a

619 | n == 0 = return $ Just a

613 | otherwise = go (n - 1) x

620 | otherwise = go (n - 1) x

614− in unStream m1 defState (return Nothing) single yieldk

621+ in foldStream defState (return Nothing) single yieldk m1

615

622

616{-# INLINE lookup #-}

623{-# INLINE lookup #-}

617lookup :: (IsStream t, Monad m, Eq a) => a -> t m (a, b) -> m (Maybe b)

624lookup :: (IsStream t, Monad m, Eq a) => a -> t m (a, b) -> m (Maybe b)

618−lookup e m = go (toStream m)

625+lookup e m = go m

619 where

626 where

620 go m1 =

627 go m1 =

621 let single (a, b) | a == e = return $ Just b

628 let single (a, b) | a == e = return $ Just b

622 | otherwise = return Nothing

629 | otherwise = return Nothing

623 yieldk (a, b) x | a == e = return $ Just b

630 yieldk (a, b) x | a == e = return $ Just b

624 | otherwise = go x

631 | otherwise = go x

625− in unStream m1 defState (return Nothing) single yieldk

632+ in foldStream defState (return Nothing) single yieldk m1

626

633

627{-# INLINE findM #-}

634{-# INLINE findM #-}

628findM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> m (Maybe a)

635findM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> m (Maybe a)

629−findM p m = go (toStream m)

636+findM p m = go m

630 where

637 where

631 go m1 =

638 go m1 =

632 let single a = do

639 let single a = do

@@ -635,7 +642,7 @@ findM p m = go (toStream m)

635 yieldk a x = do

642 yieldk a x = do

636 b <- p a

643 b <- p a

637 if b then return $ Just a else go x

644 if b then return $ Just a else go x

638− in unStream m1 defState (return Nothing) single yieldk

645+ in foldStream defState (return Nothing) single yieldk m1

639

646

640{-# INLINE find #-}

647{-# INLINE find #-}

641find :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m (Maybe a)

648find :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m (Maybe a)

@@ -643,14 +650,15 @@ find p = findM (return . p)

643

650

644{-# INLINE findIndices #-}

651{-# INLINE findIndices #-}

645findIndices :: IsStream t => (a -> Bool) -> t m a -> t m Int

652findIndices :: IsStream t => (a -> Bool) -> t m a -> t m Int

646−findIndices p = fromStream . go 0 . toStream

653+findIndices p = go 0

647 where

654 where

648 go offset m1 = mkStream $ \st stp sng yld ->

655 go offset m1 = mkStream $ \st stp sng yld ->

649 let single a | p a = sng offset

656 let single a | p a = sng offset

650 | otherwise = stp

657 | otherwise = stp

651 yieldk a x | p a = yld offset $ go (offset + 1) x

658 yieldk a x | p a = yld offset $ go (offset + 1) x

659+ | otherwise = foldStream (adaptState st) stp sng yld $

652− | otherwise = unStream (go (offset + 1) x) st stp sng yld

660+ go (offset + 1) x

653− in unStream m1 st stp single yieldk

661+ in foldStream (adaptState st) stp single yieldk m1

654

662

655------------------------------------------------------------------------------

663------------------------------------------------------------------------------

656-- Map and Fold

664-- Map and Fold

@@ -659,13 +667,13 @@ findIndices p = fromStream . go 0 . toStream

659-- | Apply a monadic action to each element of the stream and discard the

667-- | Apply a monadic action to each element of the stream and discard the

660-- output of the action.

668-- output of the action.

661mapM_ :: (IsStream t, Monad m) => (a -> m b) -> t m a -> m ()

669mapM_ :: (IsStream t, Monad m) => (a -> m b) -> t m a -> m ()

662−mapM_ f m = go (toStream m)

670+mapM_ f m = go m

663 where

671 where

664 go m1 =

672 go m1 =

665 let stop = return ()

673 let stop = return ()

666 single a = void (f a)

674 single a = void (f a)

667 yieldk a r = f a >> go r

675 yieldk a r = f a >> go r

668− in unStream m1 defState stop single yieldk

676+ in foldStream defState stop single yieldk m1

669

677

670------------------------------------------------------------------------------

678------------------------------------------------------------------------------

671-- Converting folds

679-- Converting folds

@@ -686,14 +694,14 @@ toStreamK = id

686{-# INLINE scanx #-}

694{-# INLINE scanx #-}

687scanx :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b

695scanx :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b

688scanx step begin done m =

696scanx step begin done m =

689− cons (done begin) $ fromStream $ go (toStream m) begin

697+ cons (done begin) $ go m begin

690 where

698 where

691 go m1 !acc = mkStream $ \st stp sng yld ->

699 go m1 !acc = mkStream $ \st stp sng yld ->

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

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

693 yieldk a r =

701 yieldk a r =

694 let s = step acc a

702 let s = step acc a

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

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

696− in unStream m1 st stp single yieldk

704+ in foldStream (adaptState st) stp single yieldk m1

697

705

698{-# INLINE scanl' #-}

706{-# INLINE scanl' #-}

699scanl' :: IsStream t => (b -> a -> b) -> b -> t m a -> t m b

707scanl' :: IsStream t => (b -> a -> b) -> b -> t m a -> t m b

@@ -705,58 +713,58 @@ scanl' step begin = scanx step begin id

705

713

706{-# INLINE filter #-}

714{-# INLINE filter #-}

707filter :: IsStream t => (a -> Bool) -> t m a -> t m a

715filter :: IsStream t => (a -> Bool) -> t m a -> t m a

708−filter p m = fromStream $ go (toStream m)

716+filter p m = go m

709 where

717 where

710 go m1 = mkStream $ \st stp sng yld ->

718 go m1 = mkStream $ \st stp sng yld ->

711 let single a | p a = sng a

719 let single a | p a = sng a

712 | otherwise = stp

720 | otherwise = stp

713 yieldk a r | p a = yld a (go r)

721 yieldk a r | p a = yld a (go r)

714− | otherwise = unStream r st stp single yieldk

722+ | otherwise = foldStream st stp single yieldk r

715− in unStream m1 st stp single yieldk

723+ in foldStream st stp single yieldk m1

716

724

717{-# INLINE take #-}

725{-# INLINE take #-}

718take :: IsStream t => Int -> t m a -> t m a

726take :: IsStream t => Int -> t m a -> t m a

719−take n m = fromStream $ go n (toStream m)

727+take n m = go n m

720 where

728 where

721 go n1 m1 = mkStream $ \st stp sng yld ->

729 go n1 m1 = mkStream $ \st stp sng yld ->

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

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

723 in if n1 <= 0

731 in if n1 <= 0

724 then stp

732 then stp

725− else unStream m1 st stp sng yieldk

733+ else foldStream st stp sng yieldk m1

726

734

727{-# INLINE takeWhile #-}

735{-# INLINE takeWhile #-}

728takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a

736takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a

729−takeWhile p m = fromStream $ go (toStream m)

737+takeWhile p m = go m

730 where

738 where

731 go m1 = mkStream $ \st stp sng yld ->

739 go m1 = mkStream $ \st stp sng yld ->

732 let single a | p a = sng a

740 let single a | p a = sng a

733 | otherwise = stp

741 | otherwise = stp

734 yieldk a r | p a = yld a (go r)

742 yieldk a r | p a = yld a (go r)

735 | otherwise = stp

743 | otherwise = stp

736− in unStream m1 st stp single yieldk

744+ in foldStream st stp single yieldk m1

737

745

738drop :: IsStream t => Int -> t m a -> t m a

746drop :: IsStream t => Int -> t m a -> t m a

739−drop n m = fromStream $ mkStream $ \st stp sng yld ->

747+drop n m = mkStream $ \st stp sng yld ->

740− unStream (go n (toStream m)) st stp sng yld

748+ foldStream st stp sng yld $ go n m

741 where

749 where

742 go n1 m1 = mkStream $ \st stp sng yld ->

750 go n1 m1 = mkStream $ \st stp sng yld ->

743 let single _ = stp

751 let single _ = stp

744− yieldk _ r = (unStream $ go (n1 - 1) r) st stp sng yld

752+ yieldk _ r = foldStream st stp sng yld $ go (n1 - 1) r

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

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

746 in if n1 <= 0

754 in if n1 <= 0

747− then unStream m1 st stp sng yld

755+ then foldStream st stp sng yld m1

748− else unStream m1 st stp single yieldk

756+ else foldStream st stp single yieldk m1

749

757

750{-# INLINE dropWhile #-}

758{-# INLINE dropWhile #-}

751dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a

759dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a

752−dropWhile p m = fromStream $ go (toStream m)

760+dropWhile p m = go m

753 where

761 where

754 go m1 = mkStream $ \st stp sng yld ->

762 go m1 = mkStream $ \st stp sng yld ->

755 let single a | p a = stp

763 let single a | p a = stp

756 | otherwise = sng a

764 | otherwise = sng a

757− yieldk a r | p a = unStream r st stp single yieldk

765+ yieldk a r | p a = foldStream st stp single yieldk r

758 | otherwise = yld a r

766 | otherwise = yld a r

759− in unStream m1 st stp single yieldk

767+ in foldStream st stp single yieldk m1

760

768

761-------------------------------------------------------------------------------

769-------------------------------------------------------------------------------

762-- Mapping

770-- Mapping

@@ -766,23 +774,23 @@ dropWhile p m = fromStream $ go (toStream m)

766-- other stream types to overload it.

774-- other stream types to overload it.

767{-# INLINE mapM #-}

775{-# INLINE mapM #-}

768mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b

776mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b

769−mapM f m = go (toStream m)

777+mapM f m = go m

770 where

778 where

771− go m1 = fromStream $ mkStream $ \st stp sng yld ->

779+ go m1 = mkStream $ \st stp sng yld ->

772 let single a = f a >>= sng

780 let single a = f a >>= sng

773− yieldk a r = unStreamShared (toStream (f a |: go r)) st stp sng yld

781+ yieldk a r = foldStreamShared st stp sng yld $ f a |: go r

774− in unStream m1 st stp single yieldk

782+ in foldStream (adaptState st) stp single yieldk m1

775

783

776-- Be careful when modifying this, this uses a consM (|:) deliberately to allow

784-- Be careful when modifying this, this uses a consM (|:) deliberately to allow

777-- other stream types to overload it.

785-- other stream types to overload it.

778{-# INLINE sequence #-}

786{-# INLINE sequence #-}

779sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a

787sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a

780−sequence m = go (toStream m)

788+sequence m = go m

781 where

789 where

782− go m1 = fromStream $ mkStream $ \st stp sng yld ->

790+ go m1 = mkStream $ \st stp sng yld ->

783 let single ma = ma >>= sng

791 let single ma = ma >>= sng

784− yieldk ma r = unStreamShared (toStream $ ma |: go r) st stp sng yld

792+ yieldk ma r = foldStreamShared st stp sng yld $ ma |: go r

785− in unStream m1 st stp single yieldk

793+ in foldStream (adaptState st) stp single yieldk m1

786

794

787-------------------------------------------------------------------------------

795-------------------------------------------------------------------------------

788-- Inserting

796-- Inserting

@@ -790,19 +798,19 @@ sequence m = go (toStream m)

790

798

791{-# INLINE intersperseM #-}

799{-# INLINE intersperseM #-}

792intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a

800intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a

793−intersperseM a m = fromStream $ prependingStart (toStream m)

801+intersperseM a m = prependingStart m

794 where

802 where

795 prependingStart m1 = mkStream $ \st stp sng yld ->

803 prependingStart m1 = mkStream $ \st stp sng yld ->

796− let yieldk i x = unStreamShared (return i |: go x) st stp sng yld

804+ let yieldk i x = foldStreamShared st stp sng yld $ return i |: go x

797− in unStream m1 st stp sng yieldk

805+ in foldStream st stp sng yieldk m1

798− go m2 = fromStream $ mkStream $ \st stp sng yld ->

806+ go m2 = mkStream $ \st stp sng yld ->

799− let single i = unStreamShared (a |: yield i) st stp sng yld

807+ let single i = foldStreamShared st stp sng yld $ a |: yield i

800− yieldk i x = unStreamShared (a |: return i |: go x) st stp sng yld

808+ yieldk i x = foldStreamShared st stp sng yld $ a |: return i |: go x

801− in unStream m2 st stp single yieldk

809+ in foldStream st stp single yieldk m2

802

810

803{-# INLINE insertBy #-}

811{-# INLINE insertBy #-}

804insertBy :: IsStream t => (a -> a -> Ordering) -> a -> t m a -> t m a

812insertBy :: IsStream t => (a -> a -> Ordering) -> a -> t m a -> t m a

805−insertBy cmp x m = fromStream $ go (toStream m)

813+insertBy cmp x m = go m

806 where

814 where

807 go m1 = mkStream $ \st _ _ yld ->

815 go m1 = mkStream $ \st _ _ yld ->

808 let single a = case cmp x a of

816 let single a = case cmp x a of

@@ -812,7 +820,7 @@ insertBy cmp x m = fromStream $ go (toStream m)

812 yieldk a r = case cmp x a of

820 yieldk a r = case cmp x a of

813 GT -> yld a (go r)

821 GT -> yld a (go r)

814 _ -> yld x (a `cons` r)

822 _ -> yld x (a `cons` r)

815− in unStream m1 st stop single yieldk

823+ in foldStream st stop single yieldk m1

816

824

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

825------------------------------------------------------------------------------

818-- Deleting

826-- Deleting

@@ -820,14 +828,14 @@ insertBy cmp x m = fromStream $ go (toStream m)

820

828

821{-# INLINE deleteBy #-}

829{-# INLINE deleteBy #-}

822deleteBy :: IsStream t => (a -> a -> Bool) -> a -> t m a -> t m a

830deleteBy :: IsStream t => (a -> a -> Bool) -> a -> t m a -> t m a

823−deleteBy eq x m = fromStream $ go (toStream m)

831+deleteBy eq x m = go m

824 where

832 where

825 go m1 = mkStream $ \st stp sng yld ->

833 go m1 = mkStream $ \st stp sng yld ->

826 let single a = if eq x a then stp else sng a

834 let single a = if eq x a then stp else sng a

827 yieldk a r = if eq x a

835 yieldk a r = if eq x a

828− then unStream r st stp sng yld

836+ then foldStream st stp sng yld r

829 else yld a (go r)

837 else yld a (go r)

830− in unStream m1 st stp single yieldk

838+ in foldStream st stp single yieldk m1

831

839

832-------------------------------------------------------------------------------

840-------------------------------------------------------------------------------

833-- Map and Filter

841-- Map and Filter

@@ -835,70 +843,66 @@ deleteBy eq x m = fromStream $ go (toStream m)

835

843

836{-# INLINE mapMaybe #-}

844{-# INLINE mapMaybe #-}

837mapMaybe :: IsStream t => (a -> Maybe b) -> t m a -> t m b

845mapMaybe :: IsStream t => (a -> Maybe b) -> t m a -> t m b

838−mapMaybe f m = go (toStream m)

846+mapMaybe f m = go m

839 where

847 where

840− go m1 = fromStream $ mkStream $ \st stp sng yld ->

848+ go m1 = mkStream $ \st stp sng yld ->

841 let single a = case f a of

849 let single a = case f a of

842 Just b -> sng b

850 Just b -> sng b

843 Nothing -> stp

851 Nothing -> stp

844 yieldk a r = case f a of

852 yieldk a r = case f a of

845− Just b -> yld b (toStream $ go r)

853+ Just b -> yld b $ go r

846− Nothing -> unStream r st stp single yieldk

854+ Nothing -> foldStream (adaptState st) stp single yieldk r

847− in unStream m1 st stp single yieldk

855+ in foldStream (adaptState st) stp single yieldk m1

848

856

849------------------------------------------------------------------------------

857------------------------------------------------------------------------------

850-- Serial Zipping

858-- Serial Zipping

851------------------------------------------------------------------------------

859------------------------------------------------------------------------------

852

860

853−{-# INLINE zipWithS #-}

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

854−zipWithS :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c

862+--

855−zipWithS f = go

863+-- @since 0.1.0

864+{-# INLINABLE zipWith #-}

865+zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c

866+zipWith f = go

856 where

867 where

857 go mx my = mkStream $ \st stp sng yld -> do

868 go mx my = mkStream $ \st stp sng yld -> do

858 let merge a ra =

869 let merge a ra =

859 let single2 b = sng (f a b)

870 let single2 b = sng (f a b)

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

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

861− in unStream my st stp single2 yield2

872+ in foldStream (adaptState st) stp single2 yield2 my

862 let single1 a = merge a nil

873 let single1 a = merge a nil

863 yield1 = merge

874 yield1 = merge

864− unStream mx st stp single1 yield1

875+ foldStream (adaptState st) stp single1 yield1 mx

865−

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

867−--

868−-- @since 0.1.0

869−{-# INLINABLE zipWith #-}

870−zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c

871−zipWith f m1 m2 = fromStream $ zipWithS f (toStream m1) (toStream m2)

872

876

873-- | Zip two streams serially using a monadic zipping function.

877-- | Zip two streams serially using a monadic zipping function.

874--

878--

875-- @since 0.1.0

879-- @since 0.1.0

876zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c

880zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c

877−zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

881+zipWithM f m1 m2 = go m1 m2

878 where

882 where

879 go mx my = mkStream $ \st stp sng yld -> do

883 go mx my = mkStream $ \st stp sng yld -> do

880 let merge a ra =

884 let merge a ra =

881− let runIt x = unStream x st stp sng yld

885+ let runIt x = foldStream st stp sng yld x

882 single2 b = f a b >>= sng

886 single2 b = f a b >>= sng

883 yield2 b rb = f a b >>= \x -> runIt (x `cons` go ra rb)

887 yield2 b rb = f a b >>= \x -> runIt (x `cons` go ra rb)

884− in unStream my st stp single2 yield2

888+ in foldStream (adaptState st) stp single2 yield2 my

885 let single1 a = merge a nil

889 let single1 a = merge a nil

886 yield1 = merge

890 yield1 = merge

887− unStream mx st stp single1 yield1

891+ foldStream (adaptState st) stp single1 yield1 mx

888

892

889------------------------------------------------------------------------------

893------------------------------------------------------------------------------

890-- Merging

894-- Merging

891------------------------------------------------------------------------------

895------------------------------------------------------------------------------

892

896

893−{-# INLINE mergeByS #-}

897+{-# INLINE mergeByM #-}

894−mergeByS

898+mergeByM

895− :: Monad m

899+ :: (IsStream t, Monad m)

896− => (a -> a -> m Ordering) -> Stream m a -> Stream m a -> Stream m a

900+ => (a -> a -> m Ordering) -> t m a -> t m a -> t m a

897−mergeByS cmp = go

901+mergeByM cmp = go

898 where

902 where

899 go mx my = mkStream $ \st stp sng yld -> do

903 go mx my = mkStream $ \st stp sng yld -> do

900 let mergeWithY a ra =

904 let mergeWithY a ra =

901− let stop2 = unStream mx st stp sng yld

905+ let stop2 = foldStream st stp sng yld mx

902 single2 b = do

906 single2 b = do

903 r <- cmp a b

907 r <- cmp a b

904 case r of

908 case r of

@@ -909,17 +913,11 @@ mergeByS cmp = go

909 case r of

913 case r of

910 GT -> yld b (go (a `cons` ra) rb)

914 GT -> yld b (go (a `cons` ra) rb)

911 _ -> yld a (go ra (b `cons` rb))

915 _ -> yld a (go ra (b `cons` rb))

912− in unStream my st stop2 single2 yield2

916+ in foldStream st stop2 single2 yield2 my

913− let stopX = unStream my st stp sng yld

917+ let stopX = foldStream st stp sng yld my

914 singleX a = mergeWithY a nil

918 singleX a = mergeWithY a nil

915 yieldX = mergeWithY

919 yieldX = mergeWithY

916− unStream mx st stopX singleX yieldX

920+ foldStream st stopX singleX yieldX mx

917−

918−{-# INLINABLE mergeByM #-}

919−mergeByM

920− :: (IsStream t, Monad m)

921− => (a -> a -> m Ordering) -> t m a -> t m a -> t m a

922−mergeByM f m1 m2 = fromStream $ mergeByS f (toStream m1) (toStream m2)

923

921

924{-# INLINABLE mergeBy #-}

922{-# INLINABLE mergeBy #-}

925mergeBy

923mergeBy

@@ -937,14 +935,14 @@ the m = do

937 r <- uncons m

935 r <- uncons m

938 case r of

936 case r of

939 Nothing -> return Nothing

937 Nothing -> return Nothing

940− Just (h, t) -> go h (toStream t)

938+ Just (h, t) -> go h t

941 where

939 where

942 go h m1 =

940 go h m1 =

943 let single a | h == a = return $ Just h

941 let single a | h == a = return $ Just h

944 | otherwise = return Nothing

942 | otherwise = return Nothing

945 yieldk a r | h == a = go h r

943 yieldk a r | h == a = go h r

946 | otherwise = return Nothing

944 | otherwise = return Nothing

947− in unStream m1 defState (return $ Just h) single yieldk

945+ in foldStream defState (return $ Just h) single yieldk m1

948

946

949-------------------------------------------------------------------------------

947-------------------------------------------------------------------------------

950-- Bind utility

948-- Bind utility

@@ -952,20 +950,19 @@ the m = do

952

950

953{-# INLINE bindWith #-}

951{-# INLINE bindWith #-}

954bindWith

952bindWith

953+ :: IsStream t

955− :: (forall c. Stream m c -> Stream m c -> Stream m c)

954+ => (forall c. t m c -> t m c -> t m c)

956− -> Stream m a

955+ -> t m a

957− -> (a -> Stream m b)

956+ -> (a -> t m b)

958− -> Stream m b

957+ -> t m b

959bindWith par m1 f = go m1

958bindWith par m1 f = go m1

960 where

959 where

961 go m =

960 go m =

962 mkStream $ \st stp sng yld ->

961 mkStream $ \st stp sng yld ->

963− let runShared x = unStreamShared x st stp sng yld

962+ let foldShared = foldStreamShared st stp sng yld

964− runIsolated x = unStream x st stp sng yld

963+ single a = foldShared $ unShare (f a)

965−

966− single a = runIsolated $ f a

967− yieldk a r = runShared $ unShare (f a) `par` go r

964+ yieldk a r = foldShared $ unShare (f a) `par` go r

968− in unStream m st stp single yieldk

965+ in foldStream (adaptState st) stp single yieldk m

969

966

970------------------------------------------------------------------------------

967------------------------------------------------------------------------------

971-- Alternative & MonadPlus

968-- Alternative & MonadPlus

@@ -973,8 +970,8 @@ bindWith par m1 f = go m1

973

970

974_alt :: Stream m a -> Stream m a -> Stream m a

971_alt :: Stream m a -> Stream m a -> Stream m a

975_alt m1 m2 = mkStream $ \st stp sng yld ->

972_alt m1 m2 = mkStream $ \st stp sng yld ->

976− let stop = unStream m2 st stp sng yld

973+ let stop = foldStream st stp sng yld m2

977− in unStream m1 st stop sng yld

974+ in foldStream st stop sng yld m1

978

975

979------------------------------------------------------------------------------

976------------------------------------------------------------------------------

980-- MonadReader

977-- MonadReader

@@ -985,7 +982,7 @@ withLocal f m =

985 mkStream $ \st stp sng yld ->

982 mkStream $ \st stp sng yld ->

986 let single = local f . sng

983 let single = local f . sng

987 yieldk a r = local f $ yld a (withLocal f r)

984 yieldk a r = local f $ yld a (withLocal f r)

988− in unStream m st (local f stp) single yieldk

985+ in foldStream st (local f stp) single yieldk m

989

986

990------------------------------------------------------------------------------

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

991-- MonadError

988-- MonadError

@@ -1003,4 +1000,3 @@ withCatchError m h =

1003 yieldk a r = yld a (withCatchError r h)

1000 yieldk a r = yld a (withCatchError r h)

1004 in handle $ run m

1001 in handle $ run m

1005-}

1002-}

1006−

src/Streamly/Streams/StreamKType.hs+132−109

@@ -31,22 +31,23 @@ module Streamly.Streams.StreamKType

32 -- * The stream type

33 , Stream ()

34− , Streaming

36 -- * Construction

35 -- * Construction

37− , consMSerial

38 , mkStream

36 , mkStream

40 -- * Elimination

38 -- * Elimination

41− , unStream

42− , unStreamShared

43 , foldStream

39 , foldStream

44− , runStreamSVar

40+ , foldStreamShared

41+ , foldStreamSVar

43+ -- instances

44+ , consMSerial

46 , nil

45 , nil

47 , serial

46 , serial

48 , map

47 , map

49 , yieldM

48 , yieldM

49+

50+ , Streaming -- deprecated

50 )

51 )

51where

52where

@@ -58,36 +59,6 @@ import Prelude hiding (map)

59import Streamly.SVar

60import Streamly.SVar

61−------------------------------------------------------------------------------

62−-- The basic stream type

63−------------------------------------------------------------------------------

64−

65−-- | The type @Stream m a@ represents a monadic stream of values of type 'a'

66−-- constructed using actions in monad 'm'. It uses stop, singleton and yield

67−-- continuations equivalent to the following direct style type:

68−--

69−-- @

70−-- data Stream m a = Stop | Singleton a | Yield a (Stream m a)

71−-- @

72−--

73−-- To facilitate parallel composition we maintain a local state in an 'SVar'

74−-- that is shared across and is used for synchronization of the streams being

75−-- composed.

76−--

77−-- The singleton case can be expressed in terms of stop and yield but we have

78−-- it as a separate case to optimize composition operations for streams with

79−-- single element. We build singleton streams in the implementation of 'pure'

80−-- for Applicative and Monad, and in 'lift' for MonadTrans.

81−--

82−newtype Stream m a =

83− MkStream (forall r.

84− State Stream m a -- state

85− -> m r -- stop

86− -> (a -> m r) -- singleton

87− -> (a -> Stream m a -> m r) -- yield

88− -> m r

89− )

90−

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

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

92-- Types that can behave as a Stream

63-- Types that can behave as a Stream

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

64------------------------------------------------------------------------------

@@ -95,6 +66,9 @@ newtype Stream m a =

95infixr 5 `consM`

66infixr 5 `consM`

96infixr 5 |:

67infixr 5 |:

69+-- XXX Use a different SVar based on the stream type. But we need to make sure

70+-- that we do not lose performance due to polymorphism.

71+--

98-- | Class of types that can represent a stream of elements of some type 'a' in

72-- | Class of types that can represent a stream of elements of some type 'a' in

99-- some monad 'm'.

73-- some monad 'm'.

100--

74--

@@ -149,42 +123,64 @@ type Streaming = IsStream

149-- Type adapting combinators

123-- Type adapting combinators

150-------------------------------------------------------------------------------

124-------------------------------------------------------------------------------

151

125

126+-- XXX Move/reset the State here by reconstructing the stream with cleared

127+-- state. Can we make sure we do not do that when t1 = t2? If we do this then

128+-- we do not need to do that explicitly using svarStyle. It would act as

129+-- unShare when the stream type is the same.

130+--

152-- | Adapt any specific stream type to any other specific stream type.

131-- | Adapt any specific stream type to any other specific stream type.

153--

132--

154-- @since 0.1.0

133-- @since 0.1.0

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

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

156adapt = fromStream . toStream

135adapt = fromStream . toStream

157

136

158−-------------------------------------------------------------------------------

137+------------------------------------------------------------------------------

159−-- Instances

138+-- Basic stream type

160−-------------------------------------------------------------------------------

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

161−

162−{-# INLINE consMSerial #-}

163−consMSerial :: (Monad m) => m a -> Stream m a -> Stream m a

164−consMSerial m r = MkStream $ \_ _ _ yld -> m >>= \a -> yld a r

165−

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

167−-- IsStream Stream

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

169−

170−instance IsStream Stream where

171− toStream = id

172− fromStream = id

173−

174− {-# INLINE consM #-}

175− {-# SPECIALIZE consM :: IO a -> Stream IO a -> Stream IO a #-}

176− consM :: Monad m => m a -> Stream m a -> Stream m a

177− consM = consMSerial

178

140

179− {-# INLINE (|:) #-}

141+-- | The type @Stream m a@ represents a monadic stream of values of type 'a'

180− {-# SPECIALIZE (|:) :: IO a -> Stream IO a -> Stream IO a #-}

142+-- constructed using actions in monad 'm'. It uses stop, singleton and yield

143+-- continuations equivalent to the following direct style type:

144+--

145+-- @

146+-- data Stream m a = Stop | Singleton a | Yield a (Stream m a)

147+-- @

148+--

149+-- To facilitate parallel composition we maintain a local state in an 'SVar'

150+-- that is shared across and is used for synchronization of the streams being

151+-- composed.

152+--

153+-- The singleton case can be expressed in terms of stop and yield but we have

154+-- it as a separate case to optimize composition operations for streams with

155+-- single element. We build singleton streams in the implementation of 'pure'

156+-- for Applicative and Monad, and in 'lift' for MonadTrans.

157+--

158+-- XXX remove the Stream type parameter from State as it is always constant.

159+-- We can remove it from SVar as well

160+--

161+newtype Stream m a =

162+ MkStream (forall r.

163+ State Stream m a -- state

164+ -> m r -- stop

165+ -> (a -> m r) -- singleton

181− (|:) :: Monad m => m a -> Stream m a -> Stream m a

166+ -> (a -> Stream m a -> m r) -- yield

182− (|:) = consMSerial

167+ -> m r

168+ )

183

169

184------------------------------------------------------------------------------

170------------------------------------------------------------------------------

185-- Building a stream

171-- Building a stream

186------------------------------------------------------------------------------

172------------------------------------------------------------------------------

187

173

174+-- XXX The State is always parameterized by "Stream" which means State is not

175+-- different for different stream types. So we have to manually make sure that

176+-- when converting from one stream to another we migrate the state correctly.

177+-- This can be fixed if we use a different SVar type for different streams.

178+-- Currently we always use "SVar Stream" and therefore a different State type

179+-- parameterized by that stream.

180+--

181+-- XXX change the order of arguments to match the standard fold function

182+-- convention i.e. st yield single stop.

183+--

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

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

189-- continuation and a yield continuation.

185-- continuation and a yield continuation.

190mkStream:: IsStream t

186mkStream:: IsStream t

@@ -198,45 +194,31 @@ mkStream k = fromStream $ MkStream $ \st stp sng yld ->

198 let yieldk a r = yld a (toStream r)

194 let yieldk a r = yld a (toStream r)

199 in k st stp sng yieldk

195 in k st stp sng yieldk

200

196

201−-- | An empty stream.

202−--

203−-- @

204−-- > toList nil

205−-- []

206−-- @

207−--

208−-- @since 0.1.0

209−nil :: IsStream t => t m a

210−nil = fromStream $ mkStream $ \_ stp _ _ -> stp

211−

212------------------------------------------------------------------------------

197------------------------------------------------------------------------------

213-- Folding a stream

198-- Folding a stream

214------------------------------------------------------------------------------

199------------------------------------------------------------------------------

215

200

216−-- Run a stream in detached mode i.e. not joining an SVar

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

217−{-# INLINE unStream #-}

202+-- continuation and a yield continuation. The stream would share the current

218−unStream ::

203+-- SVar passed via the State.

219− Stream m a

204+{-# INLINE foldStreamShared #-}

205+foldStreamShared

206+ :: IsStream t

220− -> State Stream m b -- state

207+ => State Stream m a

221− -> m r -- stop

222− -> (a -> m r) -- singleton

223− -> (a -> Stream m a -> m r) -- yield

224 -> m r

208 -> m r

225−unStream (MkStream runner) st = runner (adaptState st)

226−

227−-- | Like unstream, but passes a shared SVar across continuations.

228−{-# INLINE unStreamShared #-}

229−unStreamShared ::

230− Stream m a

231− -> State Stream m a -- state

232− -> m r -- stop

233− -> (a -> m r) -- singleton

209+ -> (a -> m r)

234− -> (a -> Stream m a -> m r) -- yield

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

211+ -> t m a

235 -> m r

212 -> m r

236−unStreamShared (MkStream runner) st stp sng yld = runner st stp sng yld

213+foldStreamShared st stp sng yld m =

214+ let yieldk a x = yld a (fromStream x)

215+ MkStream k = toStream m

216+ in k st stp sng yieldk

237

217

238−-- | Fold a stream by providing an SVar, a stop continuation, a singleton

218+-- | Fold a stream by providing a State, stop continuation, a singleton

239−-- continuation and a yield continuation.

219+-- continuation and a yield continuation. The stream will not use the SVar

220+-- passed via State.

221+{-# INLINE foldStream #-}

240foldStream

222foldStream

241 :: IsStream t

223 :: IsStream t

242 => State Stream m a

224 => State Stream m a

@@ -245,41 +227,71 @@ foldStream

245 -> (a -> t m a -> m r)

227 -> (a -> t m a -> m r)

246 -> t m a

228 -> t m a

247 -> m r

229 -> m r

248−foldStream st blank single step m =

230+foldStream st stp sng yld m =

249− let yieldk a x = step a (fromStream x)

231+ let yieldk a x = yld a (fromStream x)

250− in unStreamShared (toStream m) st blank single yieldk

232+ MkStream k = toStream m

233+ in k (adaptState st) stp sng yieldk

251

234

252-- Run the stream using a run function associated with the SVar that runs the

235-- Run the stream using a run function associated with the SVar that runs the

253-- streams with a captured snapshot of the monadic state.

236-- streams with a captured snapshot of the monadic state.

254−{-# INLINE runStreamSVar #-}

237+{-# INLINE foldStreamSVar #-}

255−runStreamSVar

238+foldStreamSVar

256− :: MonadIO m

239+ :: (IsStream t, MonadIO m)

257 => SVar Stream m a

240 => SVar Stream m a

258− -> Stream m a

259 -> State Stream m a -- state

241 -> State Stream m a -- state

260 -> m r -- stop

242 -> m r -- stop

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

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

262− -> (a -> Stream m a -> m r) -- yield

244+ -> (a -> t m a -> m r) -- yield

245+ -> t m a

263 -> m ()

246 -> m ()

264−runStreamSVar sv m st stp sng yld =

247+foldStreamSVar sv st stp sng yld m =

265 let mrun = runInIO $ svarMrun sv

248 let mrun = runInIO $ svarMrun sv

266− in void $ liftIO $ mrun $ unStreamShared m st stp sng yld

249+ in void $ liftIO $ mrun $ foldStreamShared st stp sng yld m

250+

251+-------------------------------------------------------------------------------

252+-- Instances

253+-------------------------------------------------------------------------------

254+

255+{-# INLINE consMSerial #-}

256+consMSerial :: (IsStream t, Monad m) => m a -> t m a -> t m a

257+consMSerial m r = mkStream $ \_ _ _ yld -> m >>= \a -> yld a r

258+

259+-------------------------------------------------------------------------------

260+-- IsStream Stream

261+-------------------------------------------------------------------------------

262+

263+instance IsStream Stream where

264+ toStream = id

265+ fromStream = id

266+

267+ {-# INLINE consM #-}

268+ {-# SPECIALIZE consM :: IO a -> Stream IO a -> Stream IO a #-}

269+ consM :: Monad m => m a -> Stream m a -> Stream m a

270+ consM = consMSerial

271+

272+ {-# INLINE (|:) #-}

273+ {-# SPECIALIZE (|:) :: IO a -> Stream IO a -> Stream IO a #-}

274+ (|:) :: Monad m => m a -> Stream m a -> Stream m a

275+ (|:) = consMSerial

267

276

268------------------------------------------------------------------------------

277------------------------------------------------------------------------------

269-- Semigroup

278-- Semigroup

270------------------------------------------------------------------------------

279------------------------------------------------------------------------------

271

280

272−-- | Concatenates two streams sequentially i.e. the first stream is

281+-- | Polymorphic version of the 'Semigroup' operation '<>' of 'SerialT'.

273−-- exhausted completely before yielding any element from the second stream.

282+-- Appends two streams sequentially, yielding all elements from the first

283+-- stream, and then all elements from the second stream.

284+--

285+-- @since 0.2.0

274{-# INLINE serial #-}

286{-# INLINE serial #-}

275−serial :: Stream m a -> Stream m a -> Stream m a

287+serial :: IsStream t => t m a -> t m a -> t m a

276serial m1 m2 = go m1

288serial m1 m2 = go m1

277 where

289 where

278 go m = mkStream $ \st stp sng yld ->

290 go m = mkStream $ \st stp sng yld ->

279− let stop = unStream m2 st stp sng yld

291+ let stop = foldStream st stp sng yld m2

280 single a = yld a m2

292 single a = yld a m2

281 yieldk a r = yld a (go r)

293 yieldk a r = yld a (go r)

282− in unStream m st stop single yieldk

294+ in foldStream st stop single yieldk m

283

295

284instance Semigroup (Stream m a) where

296instance Semigroup (Stream m a) where

285 (<>) = serial

297 (<>) = serial

@@ -288,6 +300,17 @@ instance Semigroup (Stream m a) where

288-- Monoid

300-- Monoid

289------------------------------------------------------------------------------

301------------------------------------------------------------------------------

290

302

303+-- | An empty stream.

304+--

305+-- @

306+-- > toList nil

307+-- []

308+-- @

309+--

310+-- @since 0.1.0

311+nil :: IsStream t => t m a

312+nil = mkStream $ \_ stp _ _ -> stp

313+

291instance Monoid (Stream m a) where

314instance Monoid (Stream m a) where

292 mempty = nil

315 mempty = nil

293 mappend = (<>)

316 mappend = (<>)

@@ -299,10 +322,10 @@ instance Monoid (Stream m a) where

299

322

300{-# INLINE map #-}

323{-# INLINE map #-}

301map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b

324map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b

302−map f m = fromStream $ mkStream $ \st stp sng yld ->

325+map f m = mkStream $ \st stp sng yld ->

303 let single = sng . f

326 let single = sng . f

304− yieldk a r = yld (f a) (fmap f r)

327+ yieldk a r = yld (f a) (map f r)

305− in unStream (toStream m) st stp single yieldk

328+ in foldStream (adaptState st) stp single yieldk m

306

329

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

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

308 fmap = map

331 fmap = map

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

@@ -48,7 +48,7 @@ import Text.Read (Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec,

48 readListPrecDefault)

49import Prelude hiding (map, repeat, zipWith)

51−import Streamly.Streams.StreamK (IsStream(..), Stream, mkStream, unStream)

51+import Streamly.Streams.StreamK (IsStream(..), Stream, mkStream, foldStream)

52import Streamly.Streams.Async (mkAsync')

53import Streamly.Streams.Serial (map)

54import Streamly.SVar (MonadAsync, adaptState)

@@ -113,12 +113,12 @@ instance IsStream ZipSerialM where

113 {-# INLINE consM #-}

114 {-# SPECIALIZE consM :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}

115 consM :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a

116− consM m r = fromStream $ K.consMSerial m (toStream r)

116+ consM = K.consMSerial

117

118 {-# INLINE (|:) #-}

119 {-# SPECIALIZE (|:) :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}

120 (|:) :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a

121− m |: r = fromStream $ K.consMSerial m (toStream r)

121+ (|:) = K.consMSerial

122

123LIST_INSTANCES(ZipSerialM)

124

@@ -127,7 +127,7 @@ instance Monad m => Functor (ZipSerialM m) where

127

128instance Monad m => Applicative (ZipSerialM m) where

129 pure = ZipSerialM . K.repeat

130− m1 <*> m2 = fromStream $ K.zipWith id (toStream m1) (toStream m2)

130+ (<*>) = K.zipWith id

131

132FOLDABLE_INSTANCE(ZipSerialM)

133TRAVERSABLE_INSTANCE(ZipSerialM)

@@ -142,10 +142,10 @@ TRAVERSABLE_INSTANCE(ZipSerialM)

142-- @since 0.1.0

143zipAsyncWith :: (IsStream t, MonadAsync m)

144 => (a -> b -> c) -> t m a -> t m b -> t m c

145−zipAsyncWith f m1 m2 = fromStream $ mkStream $ \st stp sng yld -> do

145+zipAsyncWith f m1 m2 = mkStream $ \st stp sng yld -> do

146 ma <- mkAsync' (adaptState st) m1

147 mb <- mkAsync' (adaptState st) m2

148− unStream (toStream (K.zipWith f ma mb)) st stp sng yld

148+ foldStream st stp sng yld (K.zipWith f ma mb)

149

150-- | Like 'zipWithM' but zips concurrently i.e. both the streams being zipped

151-- are generated concurrently.

@@ -153,10 +153,10 @@ zipAsyncWith f m1 m2 = fromStream $ mkStream $ \st stp sng yld -> do

153-- @since 0.4.0

154zipAsyncWithM :: (IsStream t, MonadAsync m)

155 => (a -> b -> m c) -> t m a -> t m b -> t m c

156−zipAsyncWithM f m1 m2 = fromStream $ mkStream $ \st stp sng yld -> do

156+zipAsyncWithM f m1 m2 = mkStream $ \st stp sng yld -> do

157 ma <- mkAsync' (adaptState st) m1

158 mb <- mkAsync' (adaptState st) m2

159− unStream (toStream (K.zipWithM f ma mb)) st stp sng yld

159+ foldStream st stp sng yld (K.zipWithM f ma mb)

160

161------------------------------------------------------------------------------

162-- Parallely Zipping Streams

@@ -206,12 +206,12 @@ instance IsStream ZipAsyncM where

206 {-# INLINE consM #-}

207 {-# SPECIALIZE consM :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}

208 consM :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a

209− consM m r = fromStream $ K.consMSerial m (toStream r)

209+ consM = K.consMSerial

210

211 {-# INLINE (|:) #-}

212 {-# SPECIALIZE (|:) :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}

213 (|:) :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a

214− m |: r = fromStream $ K.consMSerial m (toStream r)

214+ (|:) = K.consMSerial

215

216instance Monad m => Functor (ZipAsyncM m) where

217 fmap = map