7252323 · streamly · git.iko.soy

git.iko.soy/streamly

Rename the CPS stream type to StreamK (#2269)

commit7252323b76a291aca9f512d39f0d464d992f0f00 authorRanjeet Ranjan <ranjeetr@composewell.com> dateThu 2 Mar 2023, 12:03 (4 years ago) parent34b2b49

28 files changed+403−401

compare with master

unifiedsplit

benchmark/Streamly/Benchmark/Data/Stream/Common.hs+2−2

@@ -128,10 +128,10 @@ unCross = id

128128type MonadAsync = Monad

129129

130130#ifdef USE_STREAMK

131−mkCross :: StreamK.Stream m a -> StreamK.CrossStreamK m a

131+mkCross :: StreamK m a -> StreamK.CrossStreamK m a

132132mkCross = StreamK.mkCross

133133

134−unCross :: StreamK.CrossStreamK m a -> StreamK.Stream m a

134+unCross :: StreamK.CrossStreamK m a -> StreamK m a

135135unCross = StreamK.unCross

136136#else

137137mkCross :: Stream m a -> Stream.CrossStream m a

benchmark/Streamly/Benchmark/Data/Stream/StreamK.hs+3−3

@@ -116,7 +116,7 @@ Benchmarks that need to be added

116116-- Stream generation and elimination

117117-------------------------------------------------------------------------------

118118

119−type Stream m a = S.Stream m a

119+type Stream m a = S.StreamK m a

120120

121121{-# INLINE unfoldr #-}

122122unfoldr :: Int -> Int -> Stream m Int

@@ -501,7 +501,7 @@ concatMapBySerial outer inner n =

501501-- Nested Composition

502502-------------------------------------------------------------------------------

503503

504−instance Monad m => Applicative (S.Stream m) where

504+instance Monad m => Applicative (S.StreamK m) where

505505 {-# INLINE pure #-}

506506 pure = S.fromPure

507507

@@ -520,7 +520,7 @@ instance Monad m => Applicative (S.Stream m) where

520520-- NOTE: even though concatMap for StreamD is 3x faster compared to StreamK,

521521-- the monad instance of StreamD is slower than StreamK after foldr/build

522522-- fusion.

523−instance Monad m => Monad (S.Stream m) where

523+instance Monad m => Monad (S.StreamK m) where

524524 {-# INLINE return #-}

525525 return = pure

526526

core/src/Streamly/Internal/Data/Array/Generic/Mut/Type.hs+1−1

@@ -498,7 +498,7 @@ toStreamD arr@Array{..} =

498498 D.mapM (`getIndexUnsafe` arr) $ D.enumerateFromToIntegral 0 (arrLen - 1)

499499

500500{-# INLINE toStreamK #-}

501−toStreamK :: MonadIO m => Array a -> K.Stream m a

501+toStreamK :: MonadIO m => Array a -> K.StreamK m a

502502toStreamK arr@Array{..} = K.unfoldrM step 0

503503

504504 where

core/src/Streamly/Internal/Data/Array/Mut/Type.hs+9−8

@@ -244,6 +244,7 @@ import GHC.Ptr (Ptr(..))

244244import Streamly.Internal.Data.Fold.Type (Fold(..))

245245import Streamly.Internal.Data.Producer.Type (Producer (..))

246246import Streamly.Internal.Data.Stream.StreamD.Type (Stream)

247+import Streamly.Internal.Data.Stream.StreamK.Type (StreamK)

247248import Streamly.Internal.Data.SVar.Type (adaptState, defState)

248249import Streamly.Internal.Data.Unfold.Type (Unfold(..))

249250import Streamly.Internal.System.IO (arrayPayloadSize, defaultChunkSize)

@@ -1339,7 +1340,7 @@ arraysOf n (D.Stream step state) =

13391340-- | Buffer the stream into arrays in memory.

13401341{-# INLINE arrayStreamKFromStreamD #-}

13411342arrayStreamKFromStreamD :: forall m a. (MonadIO m, Unbox a) =>

1342− D.Stream m a -> m (K.Stream m (MutArray a))

1343+ D.Stream m a -> m (StreamK m (MutArray a))

13431344arrayStreamKFromStreamD =

13441345 let n = allocBytesToElemCount (undefined :: a) defaultChunkSize

13451346 in D.foldr K.cons K.nil . arraysOf n

@@ -1554,7 +1555,7 @@ toStreamD = toStreamDWith liftIO

15541555{-# INLINE toStreamKWith #-}

15551556toStreamKWith ::

15561557 forall m a. (Monad m, Unbox a)

1557− => (forall b. IO b -> m b) -> MutArray a -> K.Stream m a

1558+ => (forall b. IO b -> m b) -> MutArray a -> StreamK m a

15581559toStreamKWith liftio MutArray{..} = go arrStart

15591560

15601561 where

@@ -1565,7 +1566,7 @@ toStreamKWith liftio MutArray{..} = go arrStart

15651566 in liftio elemM `K.consM` go (INDEX_NEXT(p,a))

15661567

15671568{-# INLINE toStreamK #-}

1568−toStreamK :: forall m a. (MonadIO m, Unbox a) => MutArray a -> K.Stream m a

1569+toStreamK :: forall m a. (MonadIO m, Unbox a) => MutArray a -> StreamK m a

15691570toStreamK = toStreamKWith liftIO

15701571

15711572{-# INLINE_NORMAL toStreamDRevWith #-}

@@ -1596,7 +1597,7 @@ toStreamDRev = toStreamDRevWith liftIO

15961597{-# INLINE toStreamKRevWith #-}

15971598toStreamKRevWith ::

15981599 forall m a. (Monad m, Unbox a)

1599− => (forall b. IO b -> m b) -> MutArray a -> K.Stream m a

1600+ => (forall b. IO b -> m b) -> MutArray a -> StreamK m a

16001601toStreamKRevWith liftio MutArray {..} =

16011602 let p = INDEX_PREV(arrEnd,a)

16021603 in go p

@@ -1609,7 +1610,7 @@ toStreamKRevWith liftio MutArray {..} =

16091610 in liftio elemM `K.consM` go (INDEX_PREV(p,a))

16101611

16111612{-# INLINE toStreamKRev #-}

1612−toStreamKRev :: forall m a. (MonadIO m, Unbox a) => MutArray a -> K.Stream m a

1613+toStreamKRev :: forall m a. (MonadIO m, Unbox a) => MutArray a -> StreamK m a

16131614toStreamKRev = toStreamKRevWith liftIO

16141615

16151616-------------------------------------------------------------------------------

@@ -1836,7 +1837,7 @@ writeNAligned align = writeNWith (newAlignedPinned align)

18361837--

18371838{-# INLINE_NORMAL writeChunks #-}

18381839writeChunks :: (MonadIO m, Unbox a) =>

1839− Int -> Fold m a (K.Stream n (MutArray a))

1840+ Int -> Fold m a (StreamK n (MutArray a))

18401841writeChunks n = FL.many (writeN n) FL.toStreamK

18411842

18421843-- XXX Compare writeWith with fromStreamD which uses an array of streams

@@ -1946,7 +1947,7 @@ fromListRevN n xs = D.fold (writeRevN n) $ D.fromList xs

19461947-------------------------------------------------------------------------------

19471948

19481949{-# INLINE arrayStreamKLength #-}

1949−arrayStreamKLength :: (Monad m, Unbox a) => K.Stream m (MutArray a) -> m Int

1950+arrayStreamKLength :: (Monad m, Unbox a) => StreamK m (MutArray a) -> m Int

19501951arrayStreamKLength as = K.foldl' (+) 0 (K.map length as)

19511952

19521953-- | Convert an array stream to an array. Note that this requires peak memory

@@ -1954,7 +1955,7 @@ arrayStreamKLength as = K.foldl' (+) 0 (K.map length as)

19541955--

19551956{-# INLINE fromArrayStreamK #-}

19561957fromArrayStreamK :: (Unbox a, MonadIO m) =>

1957− K.Stream m (MutArray a) -> m (MutArray a)

1958+ StreamK m (MutArray a) -> m (MutArray a)

19581959fromArrayStreamK as = do

19591960 len <- arrayStreamKLength as

19601961 fromStreamDN len $ D.unfoldMany reader $ D.fromStreamK as

core/src/Streamly/Internal/Data/Array/Type.hs+3−3

@@ -262,7 +262,7 @@ fromStreamD str = unsafeFreeze <$> MA.fromStreamD str

262262

263263{-# INLINE bufferChunks #-}

264264bufferChunks :: (MonadIO m, Unbox a) =>

265− D.Stream m a -> m (K.Stream m (Array a))

265+ D.Stream m a -> m (K.StreamK m (Array a))

266266bufferChunks m = D.foldr K.cons K.nil $ arraysOf defaultChunkSize m

267267

268268-- | @arraysOf n stream@ groups the elements in the input stream into arrays of

@@ -348,7 +348,7 @@ toStreamD :: forall m a. (Monad m, Unbox a) => Array a -> D.Stream m a

348348toStreamD arr = MA.toStreamDWith (return . unsafeInlineIO) (unsafeThaw arr)

349349

350350{-# INLINE toStreamK #-}

351−toStreamK :: forall m a. (Monad m, Unbox a) => Array a -> K.Stream m a

351+toStreamK :: forall m a. (Monad m, Unbox a) => Array a -> K.StreamK m a

352352toStreamK arr = MA.toStreamKWith (return . unsafeInlineIO) (unsafeThaw arr)

353353

354354{-# INLINE_NORMAL toStreamDRev #-}

@@ -357,7 +357,7 @@ toStreamDRev arr =

357357 MA.toStreamDRevWith (return . unsafeInlineIO) (unsafeThaw arr)

358358

359359{-# INLINE toStreamKRev #-}

360−toStreamKRev :: forall m a. (Monad m, Unbox a) => Array a -> K.Stream m a

360+toStreamKRev :: forall m a. (Monad m, Unbox a) => Array a -> K.StreamK m a

361361toStreamKRev arr =

362362 MA.toStreamKRevWith (return . unsafeInlineIO) (unsafeThaw arr)

363363

core/src/Streamly/Internal/Data/Fold/Type.hs+2−2

@@ -688,7 +688,7 @@ toList = foldr' (:) []

688688

689689-- xn : ... : x2 : x1 : []

690690{-# INLINE toStreamKRev #-}

691−toStreamKRev :: Monad m => Fold m a (K.Stream n a)

691+toStreamKRev :: Monad m => Fold m a (K.StreamK n a)

692692toStreamKRev = foldl' (flip K.cons) K.nil

693693

694694-- | A fold that buffers its input to a pure stream.

@@ -698,7 +698,7 @@ toStreamKRev = foldl' (flip K.cons) K.nil

698698--

699699-- /Internal/

700700{-# INLINE toStreamK #-}

701−toStreamK :: Monad m => Fold m a (K.Stream n a)

701+toStreamK :: Monad m => Fold m a (K.StreamK n a)

702702toStreamK = foldr K.cons K.nil

703703

704704------------------------------------------------------------------------------

core/src/Streamly/Internal/Data/Stream/Chunked.hs+3−3

@@ -367,7 +367,7 @@ foldBreakD (FL.Fold fstep initial extract) stream@(D.Stream step state) = do

367367

368368{-# INLINE_NORMAL foldBreakK #-}

369369foldBreakK :: forall m a b. (MonadIO m, Unbox a) =>

370− Fold m a b -> K.Stream m (Array a) -> m (b, K.Stream m (Array a))

370+ Fold m a b -> K.StreamK m (Array a) -> m (b, K.StreamK m (Array a))

371371foldBreakK (FL.Fold fstep initial extract) stream = do

372372 res <- initial

373373 case res of

@@ -644,8 +644,8 @@ parseBreakD

644644parseBreakK ::

645645 forall m a b. (MonadIO m, Unbox a)

646646 => PRD.Parser a m b

647− -> K.Stream m (Array.Array a)

648− -> m (Either ParseError b, K.Stream m (Array.Array a))

647+ -> K.StreamK m (Array.Array a)

648+ -> m (Either ParseError b, K.StreamK m (Array.Array a))

649649parseBreakK (PRD.Parser pstep initial extract) stream = do

650650 res <- initial

651651 case res of

core/src/Streamly/Internal/Data/Stream/Common.hs+8−8

@@ -50,7 +50,7 @@ import Prelude hiding (foldr, repeat)

5050-- 'K.fromFoldable' for serial streams.

5151--

5252{-# INLINE_EARLY fromList #-}

53−fromList :: Monad m => [a] -> K.Stream m a

53+fromList :: Monad m => [a] -> K.StreamK m a

5454fromList = D.toStreamK . D.fromList

5555{-# RULES "fromList fallback to StreamK" [1]

5656 forall a. D.toStreamK (D.fromList a) = K.fromFoldable a #-}

@@ -58,7 +58,7 @@ fromList = D.toStreamK . D.fromList

5858-- | Convert a stream into a list in the underlying monad.

5959--

6060{-# INLINE toList #-}

61−toList :: Monad m => K.Stream m a -> m [a]

61+toList :: Monad m => K.StreamK m a -> m [a]

6262toList m = D.toList $ D.fromStreamK m

6363

6464------------------------------------------------------------------------------

@@ -66,23 +66,23 @@ toList m = D.toList $ D.fromStreamK m

6666------------------------------------------------------------------------------

6767

6868{-# INLINE foldrM #-}

69−foldrM :: Monad m => (a -> m b -> m b) -> m b -> K.Stream m a -> m b

69+foldrM :: Monad m => (a -> m b -> m b) -> m b -> K.StreamK m a -> m b

7070foldrM step acc m = D.foldrM step acc $ D.fromStreamK m

7171

7272{-# INLINE foldr #-}

73−foldr :: Monad m => (a -> b -> b) -> b -> K.Stream m a -> m b

73+foldr :: Monad m => (a -> b -> b) -> b -> K.StreamK m a -> m b

7474foldr f z = foldrM (\a b -> f a <$> b) (return z)

7575

7676-- | Strict left associative fold.

7777--

7878{-# INLINE foldl' #-}

7979foldl' ::

80− Monad m => (b -> a -> b) -> b -> K.Stream m a -> m b

80+ Monad m => (b -> a -> b) -> b -> K.StreamK m a -> m b

8181foldl' step begin m = D.foldl' step begin $ D.fromStreamK m

8282

8383

8484{-# INLINE fold #-}

85−fold :: Monad m => Fold m a b -> K.Stream m a -> m b

85+fold :: Monad m => Fold m a b -> K.StreamK m a -> m b

8686fold fld m = D.fold fld $ D.fromStreamK m

8787

8888------------------------------------------------------------------------------

@@ -93,7 +93,7 @@ fold fld m = D.fold fld $ D.fromStreamK m

9393--

9494{-# INLINE eqBy #-}

9595eqBy :: Monad m =>

96− (a -> b -> Bool) -> K.Stream m a -> K.Stream m b -> m Bool

96+ (a -> b -> Bool) -> K.StreamK m a -> K.StreamK m b -> m Bool

9797eqBy f m1 m2 = D.eqBy f (D.fromStreamK m1) (D.fromStreamK m2)

9898

9999-- | Compare two streams

@@ -101,5 +101,5 @@ eqBy f m1 m2 = D.eqBy f (D.fromStreamK m1) (D.fromStreamK m2)

101101{-# INLINE cmpBy #-}

102102cmpBy

103103 :: Monad m

104− => (a -> b -> Ordering) -> K.Stream m a -> K.Stream m b -> m Ordering

104+ => (a -> b -> Ordering) -> K.StreamK m a -> K.StreamK m b -> m Ordering

105105cmpBy f m1 m2 = D.cmpBy f (D.fromStreamK m1) (D.fromStreamK m2)

core/src/Streamly/Internal/Data/Stream/StreamD/Type.hs+4−4

@@ -186,7 +186,7 @@ import qualified Streamly.Internal.Data.Unfold.Type as Unfold

186186-- | A stream consists of a step function that generates the next step given a

187187-- current state, and the current state.

188188data Stream m a =

189− forall s. UnStream (State K.Stream m a -> s -> m (Step s a)) s

189+ forall s. UnStream (State K.StreamK m a -> s -> m (Step s a)) s

190190

191191-- XXX This causes perf trouble when pattern matching with "Stream" in a

192192-- recursive way, e.g. in uncons, foldBreak, concatMap. We need to get rid of

@@ -195,7 +195,7 @@ unShare :: Stream m a -> Stream m a

195195unShare (UnStream step state) = UnStream step' state

196196 where step' gst = step (adaptState gst)

197197

198−pattern Stream :: (State K.Stream m a -> s -> m (Step s a)) -> s -> Stream m a

198+pattern Stream :: (State K.StreamK m a -> s -> m (Step s a)) -> s -> Stream m a

199199pattern Stream step state <- (unShare -> UnStream step state)

200200 where Stream = UnStream

201201

@@ -361,7 +361,7 @@ fromList = Stream step

361361

362362-- | Convert a CPS encoded StreamK to direct style step encoded StreamD

363363{-# INLINE_LATE fromStreamK #-}

364−fromStreamK :: Applicative m => K.Stream m a -> Stream m a

364+fromStreamK :: Applicative m => K.StreamK m a -> Stream m a

365365fromStreamK = Stream step

366366 where

367367 step gst m1 =

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

372372

373373-- | Convert a direct style step encoded StreamD to a CPS encoded StreamK

374374{-# INLINE_LATE toStreamK #-}

375−toStreamK :: Monad m => Stream m a -> K.Stream m a

375+toStreamK :: Monad m => Stream m a -> K.StreamK m a

376376toStreamK (Stream step state) = go state

377377 where

378378 go st = K.MkStream $ \gst yld _ stp ->

core/src/Streamly/Internal/Data/Stream/StreamK.hs+61−61

@@ -49,8 +49,8 @@

4949module Streamly.Internal.Data.Stream.StreamK

5050 (

5151 -- * The stream type

52− Stream(..) -- XXX stop exporting this

53− , StreamK

52+ Stream

53+ , StreamK(..)

5454 , fromStream

5555 , toStream

5656

@@ -285,7 +285,7 @@ toStream = Stream.fromStreamK

285285-- concurrent version could be async or ahead etc. Depending on how we queue

286286-- back the feedback portion b, it could be DFS or BFS style.

287287--

288−unfoldrA :: (b -> Maybe (m a, b)) -> b -> Stream m a

288+unfoldrA :: (b -> Maybe (m a, b)) -> b -> StreamK m a

289289unfoldrA = undefined

290290-}

291291

@@ -293,35 +293,35 @@ unfoldrA = undefined

293293-- Special generation

294294-------------------------------------------------------------------------------

295295

296−repeatM :: Monad m => m a -> Stream m a

296+repeatM :: Monad m => m a -> StreamK m a

297297repeatM = repeatMWith consM

298298

299299{-# INLINE replicateM #-}

300−replicateM :: Monad m => Int -> m a -> Stream m a

300+replicateM :: Monad m => Int -> m a -> StreamK m a

301301replicateM = replicateMWith consM

302302{-# INLINE replicate #-}

303−replicate :: Int -> a -> Stream m a

303+replicate :: Int -> a -> StreamK m a

304304replicate n a = go n

305305 where

306306 go cnt = if cnt <= 0 then nil else a `cons` go (cnt - 1)

307307

308308{-# INLINE fromIndicesM #-}

309−fromIndicesM :: Monad m => (Int -> m a) -> Stream m a

309+fromIndicesM :: Monad m => (Int -> m a) -> StreamK m a

310310fromIndicesM = fromIndicesMWith consM

311311{-# INLINE fromIndices #-}

312−fromIndices :: (Int -> a) -> Stream m a

312+fromIndices :: (Int -> a) -> StreamK m a

313313fromIndices gen = go 0

314314 where

315315 go n = gen n `cons` go (n + 1)

316316

317317{-# INLINE iterate #-}

318−iterate :: (a -> a) -> a -> Stream m a

318+iterate :: (a -> a) -> a -> StreamK m a

319319iterate step = go

320320 where

321321 go !s = cons s (go (step s))

322322

323323{-# INLINE iterateM #-}

324−iterateM :: Monad m => (a -> m a) -> m a -> Stream m a

324+iterateM :: Monad m => (a -> m a) -> m a -> StreamK m a

325325iterateM = iterateMWith consM

326326

327327-------------------------------------------------------------------------------

@@ -329,7 +329,7 @@ iterateM = iterateMWith consM

329329-------------------------------------------------------------------------------

330330

331331{-# INLINE fromList #-}

332−fromList :: [a] -> Stream m a

332+fromList :: [a] -> StreamK m a

333333fromList = fromFoldable

334334

335335-------------------------------------------------------------------------------

@@ -337,7 +337,7 @@ fromList = fromFoldable

337337-------------------------------------------------------------------------------

338338

339339{-# INLINE foldr1 #-}

340−foldr1 :: Monad m => (a -> a -> a) -> Stream m a -> m (Maybe a)

340+foldr1 :: Monad m => (a -> a -> a) -> StreamK m a -> m (Maybe a)

341341foldr1 step m = do

342342 r <- uncons m

343343 case r of

@@ -354,7 +354,7 @@ foldr1 step m = do

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

355355{-# INLINABLE foldlMx' #-}

356356foldlMx' :: Monad m

357− => (x -> a -> m x) -> m x -> (x -> m b) -> Stream m a -> m b

357+ => (x -> a -> m x) -> m x -> (x -> m b) -> StreamK m a -> m b

358358foldlMx' step begin done = go begin

359359 where

360360 go !acc m1 =

@@ -379,7 +379,7 @@ foldlMx' step begin done = go begin

379379-- 5050

380380--

381381{-# INLINABLE fold #-}

382−fold :: Monad m => FL.Fold m a b -> Stream m a -> m b

382+fold :: Monad m => FL.Fold m a b -> StreamK m a -> m b

383383fold (FL.Fold step begin done) m = do

384384 res <- begin

385385 case res of

@@ -408,7 +408,7 @@ fold (FL.Fold step begin done) m = do

408408-- /Internal/

409409{-# INLINE foldEither #-}

410410foldEither :: Monad m =>

411− Fold m a b -> Stream m a -> m (Either (Fold m a b) (b, Stream m a))

411+ Fold m a b -> StreamK m a -> m (Either (Fold m a b) (b, StreamK m a))

412412foldEither (FL.Fold step begin done) m = do

413413 res <- begin

414414 case res of

@@ -436,7 +436,7 @@ foldEither (FL.Fold step begin done) m = do

436436-- would be 'StreamK.nil' if the stream finished before the fold.

437437--

438438{-# INLINE foldBreak #-}

439−foldBreak :: Monad m => Fold m a b -> Stream m a -> m (b, Stream m a)

439+foldBreak :: Monad m => Fold m a b -> StreamK m a -> m (b, StreamK m a)

440440foldBreak fld strm = do

441441 r <- foldEither fld strm

442442 case r of

@@ -513,7 +513,7 @@ foldConcat

513513

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

515515{-# INLINE foldlM' #-}

516−foldlM' :: Monad m => (b -> a -> m b) -> m b -> Stream m a -> m b

516+foldlM' :: Monad m => (b -> a -> m b) -> m b -> StreamK m a -> m b

517517foldlM' step begin = foldlMx' step begin return

518518

519519------------------------------------------------------------------------------

@@ -521,7 +521,7 @@ foldlM' step begin = foldlMx' step begin return

521521------------------------------------------------------------------------------

522522

523523{-# INLINE head #-}

524−head :: Monad m => Stream m a -> m (Maybe a)

524+head :: Monad m => StreamK m a -> m (Maybe a)

525525-- head = foldrM (\x _ -> return $ Just x) (return Nothing)

526526head m =

527527 let stop = return Nothing

@@ -530,7 +530,7 @@ head m =

530530 in foldStream defState yieldk single stop m

531531

532532{-# INLINE elem #-}

533−elem :: (Monad m, Eq a) => a -> Stream m a -> m Bool

533+elem :: (Monad m, Eq a) => a -> StreamK m a -> m Bool

534534elem e = go

535535 where

536536 go m1 =

@@ -540,7 +540,7 @@ elem e = go

540540 in foldStream defState yieldk single stop m1

541541

542542{-# INLINE notElem #-}

543−notElem :: (Monad m, Eq a) => a -> Stream m a -> m Bool

543+notElem :: (Monad m, Eq a) => a -> StreamK m a -> m Bool

544544notElem e = go

545545 where

546546 go m1 =

@@ -550,7 +550,7 @@ notElem e = go

550550 in foldStream defState yieldk single stop m1

551551

552552{-# INLINABLE all #-}

553−all :: Monad m => (a -> Bool) -> Stream m a -> m Bool

553+all :: Monad m => (a -> Bool) -> StreamK m a -> m Bool

554554all p = go

555555 where

556556 go m1 =

@@ -561,7 +561,7 @@ all p = go

561561 in foldStream defState yieldk single (return True) m1

562562

563563{-# INLINABLE any #-}

564−any :: Monad m => (a -> Bool) -> Stream m a -> m Bool

564+any :: Monad m => (a -> Bool) -> StreamK m a -> m Bool

565565any p = go

566566 where

567567 go m1 =

@@ -573,11 +573,11 @@ any p = go

573573

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

575575{-# INLINE last #-}

576−last :: Monad m => Stream m a -> m (Maybe a)

576+last :: Monad m => StreamK m a -> m (Maybe a)

577577last = foldlx' (\_ y -> Just y) Nothing id

578578

579579{-# INLINE minimum #-}

580−minimum :: (Monad m, Ord a) => Stream m a -> m (Maybe a)

580+minimum :: (Monad m, Ord a) => StreamK m a -> m (Maybe a)

581581minimum = go Nothing

582582 where

583583 go Nothing m1 =

@@ -601,7 +601,7 @@ minimum = go Nothing

601601{-# INLINE minimumBy #-}

602602minimumBy

603603 :: (Monad m)

604− => (a -> a -> Ordering) -> Stream m a -> m (Maybe a)

604+ => (a -> a -> Ordering) -> StreamK m a -> m (Maybe a)

605605minimumBy cmp = go Nothing

606606 where

607607 go Nothing m1 =

@@ -621,7 +621,7 @@ minimumBy cmp = go Nothing

621621 in foldStream defState yieldk single stop m1

622622

623623{-# INLINE maximum #-}

624−maximum :: (Monad m, Ord a) => Stream m a -> m (Maybe a)

624+maximum :: (Monad m, Ord a) => StreamK m a -> m (Maybe a)

625625maximum = go Nothing

626626 where

627627 go Nothing m1 =

@@ -643,7 +643,7 @@ maximum = go Nothing

643643 in foldStream defState yieldk single stop m1

644644

645645{-# INLINE maximumBy #-}

646−maximumBy :: Monad m => (a -> a -> Ordering) -> Stream m a -> m (Maybe a)

646+maximumBy :: Monad m => (a -> a -> Ordering) -> StreamK m a -> m (Maybe a)

647647maximumBy cmp = go Nothing

648648 where

649649 go Nothing m1 =

@@ -663,7 +663,7 @@ maximumBy cmp = go Nothing

663663 in foldStream defState yieldk single stop m1

664664

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

666−(!!) :: Monad m => Stream m a -> Int -> m (Maybe a)

666+(!!) :: Monad m => StreamK m a -> Int -> m (Maybe a)

667667m !! i = go i m

668668 where

669669 go n m1 =

@@ -675,7 +675,7 @@ m !! i = go i m

675675 in foldStream defState yieldk single (return Nothing) m1

676676

677677{-# INLINE lookup #-}

678−lookup :: (Monad m, Eq a) => a -> Stream m (a, b) -> m (Maybe b)

678+lookup :: (Monad m, Eq a) => a -> StreamK m (a, b) -> m (Maybe b)

679679lookup e = go

680680 where

681681 go m1 =

@@ -686,7 +686,7 @@ lookup e = go

686686 in foldStream defState yieldk single (return Nothing) m1

687687

688688{-# INLINE findM #-}

689−findM :: Monad m => (a -> m Bool) -> Stream m a -> m (Maybe a)

689+findM :: Monad m => (a -> m Bool) -> StreamK m a -> m (Maybe a)

690690findM p = go

691691 where

692692 go m1 =

@@ -699,11 +699,11 @@ findM p = go

699699 in foldStream defState yieldk single (return Nothing) m1

700700

701701{-# INLINE find #-}

702−find :: Monad m => (a -> Bool) -> Stream m a -> m (Maybe a)

702+find :: Monad m => (a -> Bool) -> StreamK m a -> m (Maybe a)

703703find p = findM (return . p)

704704

705705{-# INLINE findIndices #-}

706−findIndices :: (a -> Bool) -> Stream m a -> Stream m Int

706+findIndices :: (a -> Bool) -> StreamK m a -> StreamK m Int

707707findIndices p = go 0

708708 where

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

@@ -721,7 +721,7 @@ findIndices p = go 0

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

722722-- output of the action.

723723{-# INLINE mapM_ #-}

724−mapM_ :: Monad m => (a -> m b) -> Stream m a -> m ()

724+mapM_ :: Monad m => (a -> m b) -> StreamK m a -> m ()

725725mapM_ f = go

726726 where

727727 go m1 =

@@ -731,7 +731,7 @@ mapM_ f = go

731731 in foldStream defState yieldk single stop m1

732732

733733{-# INLINE mapM #-}

734−mapM :: Monad m => (a -> m b) -> Stream m a -> Stream m b

734+mapM :: Monad m => (a -> m b) -> StreamK m a -> StreamK m b

735735mapM = mapMWith consM

736736

737737------------------------------------------------------------------------------

@@ -739,13 +739,13 @@ mapM = mapMWith consM

739739------------------------------------------------------------------------------

740740

741741{-# INLINABLE toList #-}

742−toList :: Monad m => Stream m a -> m [a]

742+toList :: Monad m => StreamK m a -> m [a]

743743toList = foldr (:) []

744744

745745-- Based on suggestions by David Feuer and Pranay Sashank

746746{-# INLINE hoist #-}

747747hoist :: (Monad m, Monad n)

748− => (forall x. m x -> n x) -> Stream m a -> Stream n a

748+ => (forall x. m x -> n x) -> StreamK m a -> StreamK n a

749749hoist f str =

750750 mkStream $ \st yld sng stp ->

751751 let single = return . sng

@@ -759,7 +759,7 @@ hoist f str =

759759-------------------------------------------------------------------------------

760760

761761{-# INLINE scanlx' #-}

762−scanlx' :: (x -> a -> x) -> x -> (x -> b) -> Stream m a -> Stream m b

762+scanlx' :: (x -> a -> x) -> x -> (x -> b) -> StreamK m a -> StreamK m b

763763scanlx' step begin done m =

764764 cons (done begin) $ go m begin

765765 where

@@ -771,7 +771,7 @@ scanlx' step begin done m =

771771 in foldStream (adaptState st) yieldk single stp m1

772772

773773{-# INLINE scanl' #-}

774−scanl' :: (b -> a -> b) -> b -> Stream m a -> Stream m b

774+scanl' :: (b -> a -> b) -> b -> StreamK m a -> StreamK m b

775775scanl' step begin = scanlx' step begin id

776776

777777-------------------------------------------------------------------------------

@@ -779,7 +779,7 @@ scanl' step begin = scanlx' step begin id

779779-------------------------------------------------------------------------------

780780

781781{-# INLINE filter #-}

782−filter :: (a -> Bool) -> Stream m a -> Stream m a

782+filter :: (a -> Bool) -> StreamK m a -> StreamK m a

783783filter p = go

784784 where

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

@@ -790,7 +790,7 @@ filter p = go

790790 in foldStream st yieldk single stp m1

791791

792792{-# INLINE take #-}

793−take :: Int -> Stream m a -> Stream m a

793+take :: Int -> StreamK m a -> StreamK m a

794794take = go

795795 where

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

@@ -800,7 +800,7 @@ take = go

800800 else foldStream st yieldk sng stp m1

801801

802802{-# INLINE takeWhile #-}

803−takeWhile :: (a -> Bool) -> Stream m a -> Stream m a

803+takeWhile :: (a -> Bool) -> StreamK m a -> StreamK m a

804804takeWhile p = go

805805 where

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

@@ -811,7 +811,7 @@ takeWhile p = go

811811 in foldStream st yieldk single stp m1

812812

813813{-# INLINE drop #-}

814−drop :: Int -> Stream m a -> Stream m a

814+drop :: Int -> StreamK m a -> StreamK m a

815815drop n m = unShare (go n m)

816816 where

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

@@ -823,7 +823,7 @@ drop n m = unShare (go n m)

823823 else foldStreamShared st yieldk single stp m1

824824

825825{-# INLINE dropWhile #-}

826−dropWhile :: (a -> Bool) -> Stream m a -> Stream m a

826+dropWhile :: (a -> Bool) -> StreamK m a -> StreamK m a

827827dropWhile p = go

828828 where

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

@@ -840,7 +840,7 @@ dropWhile p = go

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

841841-- other stream types to overload it.

842842{-# INLINE sequence #-}

843−sequence :: Monad m => Stream m (m a) -> Stream m a

843+sequence :: Monad m => StreamK m (m a) -> StreamK m a

844844sequence = go

845845 where

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

@@ -853,7 +853,7 @@ sequence = go

853853-------------------------------------------------------------------------------

854854

855855{-# INLINE intersperseM #-}

856−intersperseM :: Monad m => m a -> Stream m a -> Stream m a

856+intersperseM :: Monad m => m a -> StreamK m a -> StreamK m a

857857intersperseM a = prependingStart

858858 where

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

@@ -868,11 +868,11 @@ intersperseM a = prependingStart

868868 in foldStream st yieldk single stp m2

869869

870870{-# INLINE intersperse #-}

871−intersperse :: Monad m => a -> Stream m a -> Stream m a

871+intersperse :: Monad m => a -> StreamK m a -> StreamK m a

872872intersperse a = intersperseM (return a)

873873

874874{-# INLINE insertBy #-}

875−insertBy :: (a -> a -> Ordering) -> a -> Stream m a -> Stream m a

875+insertBy :: (a -> a -> Ordering) -> a -> StreamK m a -> StreamK m a

876876insertBy cmp x = go

877877 where

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

@@ -890,7 +890,7 @@ insertBy cmp x = go

890890------------------------------------------------------------------------------

891891

892892{-# INLINE deleteBy #-}

893−deleteBy :: (a -> a -> Bool) -> a -> Stream m a -> Stream m a

893+deleteBy :: (a -> a -> Bool) -> a -> StreamK m a -> StreamK m a

894894deleteBy eq x = go

895895 where

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

@@ -905,7 +905,7 @@ deleteBy eq x = go

905905-------------------------------------------------------------------------------

906906

907907{-# INLINE mapMaybe #-}

908−mapMaybe :: (a -> Maybe b) -> Stream m a -> Stream m b

908+mapMaybe :: (a -> Maybe b) -> StreamK m a -> StreamK m b

909909mapMaybe f = go

910910 where

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

@@ -923,7 +923,7 @@ mapMaybe f = go

923923--

924924-- @since 0.1.0

925925{-# INLINE zipWith #-}

926−zipWith :: Monad m => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c

926+zipWith :: Monad m => (a -> b -> c) -> StreamK m a -> StreamK m b -> StreamK m c

927927zipWith f = zipWithM (\a b -> return (f a b))

928928

929929-- | Zip two streams serially using a monadic zipping function.

@@ -931,7 +931,7 @@ zipWith f = zipWithM (\a b -> return (f a b))

931931-- @since 0.1.0

932932{-# INLINE zipWithM #-}

933933zipWithM :: Monad m =>

934− (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c

934+ (a -> b -> m c) -> StreamK m a -> StreamK m b -> StreamK m c

935935zipWithM f = go

936936

937937 where

@@ -951,7 +951,7 @@ zipWithM f = go

951951

952952{-# INLINE mergeByM #-}

953953mergeByM :: Monad m =>

954− (a -> a -> m Ordering) -> Stream m a -> Stream m a -> Stream m a

954+ (a -> a -> m Ordering) -> StreamK m a -> StreamK m a -> StreamK m a

955955mergeByM cmp = go

956956

957957 where

@@ -1019,7 +1019,7 @@ mergeByM cmp = go

10191019 in foldStream st yield single stop mx

10201020

10211021{-# INLINE mergeBy #-}

1022−mergeBy :: (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a

1022+mergeBy :: (a -> a -> Ordering) -> StreamK m a -> StreamK m a -> StreamK m a

10231023-- XXX GHC: This has slightly worse performance than replacing "r <- cmp x y"

10241024-- with "let r = cmp x y" in the monadic version. The definition below is

10251025-- exactly the same as mergeByM except this change.

@@ -1087,7 +1087,7 @@ mergeBy cmp = go

10871087------------------------------------------------------------------------------

10881088

10891089{-# INLINE the #-}

1090−the :: (Eq a, Monad m) => Stream m a -> m (Maybe a)

1090+the :: (Eq a, Monad m) => StreamK m a -> m (Maybe a)

10911091the m = do

10921092 r <- uncons m

10931093 case r of

@@ -1105,7 +1105,7 @@ the m = do

11051105-- Alternative & MonadPlus

11061106------------------------------------------------------------------------------

11071107

1108−_alt :: Stream m a -> Stream m a -> Stream m a

1108+_alt :: StreamK m a -> StreamK m a -> StreamK m a

11091109_alt m1 m2 = mkStream $ \st yld sng stp ->

11101110 let stop = foldStream st yld sng stp m2

11111111 in foldStream st yld sng stop m1

@@ -1118,7 +1118,7 @@ _alt m1 m2 = mkStream $ \st yld sng stp ->

11181118-- XXX handle and test cross thread state transfer

11191119withCatchError

11201120 :: MonadError e m

1121− => Stream m a -> (e -> Stream m a) -> Stream m a

1121+ => StreamK m a -> (e -> StreamK m a) -> StreamK m a

11221122withCatchError m h =

11231123 mkStream $ \_ stp sng yld ->

11241124 let run x = unStream x Nothing stp sng yieldk

@@ -1150,8 +1150,8 @@ splitAt n ls

11501150parseDBreak

11511151 :: Monad m

11521152 => PR.Parser a m b

1153− -> Stream m a

1154− -> m (Either ParseError b, Stream m a)

1153+ -> StreamK m a

1154+ -> m (Either ParseError b, StreamK m a)

11551155parseDBreak (PR.Parser pstep initial extract) stream = do

11561156 res <- initial

11571157 case res of

@@ -1238,7 +1238,7 @@ parseDBreak (PR.Parser pstep initial extract) stream = do

12381238-- and convert ParserD to ParserK for element parsing using StreamK.

12391239{-# INLINE parseD #-}

12401240parseD :: Monad m =>

1241− Parser.Parser a m b -> Stream m a -> m (Either ParseError b)

1241+ Parser.Parser a m b -> StreamK m a -> m (Either ParseError b)

12421242parseD f = fmap fst . parseDBreak f

12431243

12441244-------------------------------------------------------------------------------

@@ -1373,7 +1373,7 @@ parseBreakChunks parser input = do

13731373

13741374{-# INLINE parseChunks #-}

13751375parseChunks :: (Monad m, Unbox a) =>

1376− ParserK.Parser a m b -> Stream m (Array a) -> m (Either ParseError b)

1376+ ParserK.Parser a m b -> StreamK m (Array a) -> m (Either ParseError b)

13771377parseChunks f = fmap fst . parseBreakChunks f

13781378

13791379-------------------------------------------------------------------------------

@@ -1393,7 +1393,7 @@ parseChunks f = fmap fst . parseBreakChunks f

13931393-- /O(n) space/

13941394--

13951395{-# INLINE sortBy #-}

1396−sortBy :: Monad m => (a -> a -> Ordering) -> Stream m a -> Stream m a

1396+sortBy :: Monad m => (a -> a -> Ordering) -> StreamK m a -> StreamK m a

13971397-- sortBy f = Stream.concatPairsWith (Stream.mergeBy f) Stream.fromPure

13981398sortBy cmp =

13991399 let p =

core/src/Streamly/Internal/Data/Stream/StreamK/Transformer.hs+5−5

@@ -19,14 +19,14 @@ where

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

2020import Control.Monad.Trans.State.Strict (StateT)

2121import Streamly.Internal.Data.Stream.StreamK

22− (Stream, nil, cons, uncons, concatEffect)

22+ (StreamK, nil, cons, uncons, concatEffect)

2323

2424import qualified Control.Monad.Trans.State.Strict as State

2525

2626-- | Lazy left fold to an arbitrary transformer monad.

2727{-# INLINE foldlT #-}

2828foldlT :: (Monad m, Monad (s m), MonadTrans s)

29− => (s m b -> a -> s m b) -> s m b -> Stream m a -> s m b

29+ => (s m b -> a -> s m b) -> s m b -> StreamK m a -> s m b

3030foldlT step = go

3131 where

3232 go acc m1 = do

@@ -38,7 +38,7 @@ foldlT step = go

3838-- | Right associative fold to an arbitrary transformer monad.

3939{-# INLINE foldrT #-}

4040foldrT :: (Monad m, Monad (s m), MonadTrans s)

41− => (a -> s m b -> s m b) -> s m b -> Stream m a -> s m b

41+ => (a -> s m b -> s m b) -> s m b -> StreamK m a -> s m b

4242foldrT step final = go

4343 where

4444 go m1 = do

@@ -52,7 +52,7 @@ foldrT step final = go

5252------------------------------------------------------------------------------

5353

5454{-# INLINE evalStateT #-}

55−evalStateT :: Monad m => m s -> Stream (StateT s m) a -> Stream m a

55+evalStateT :: Monad m => m s -> StreamK (StateT s m) a -> StreamK m a

5656evalStateT = go

5757

5858 where

@@ -66,7 +66,7 @@ evalStateT = go

6666

6767{-# INLINE liftInner #-}

6868liftInner :: (Monad m, MonadTrans t, Monad (t m)) =>

69− Stream m a -> Stream (t m) a

69+ StreamK m a -> StreamK (t m) a

7070liftInner = go

7171

7272 where

core/src/Streamly/Internal/Data/Stream/StreamK/Type.hs+201−200

@@ -15,8 +15,8 @@

1515module Streamly.Internal.Data.Stream.StreamK.Type

1616 (

1717 -- * StreamK type

18− Stream (..) -- XXX stop exporting this

19− , StreamK

18+ Stream

19+ , StreamK (..)

2020

2121 -- * CrossStreamK type wrapper

2222 , CrossStreamK

@@ -181,7 +181,7 @@ import Prelude hiding

181181-- direct style type:

182182--

183183-- @

184−-- data Stream m a = Stop | Singleton a | Yield a (Stream m a)

184+-- data StreamK m a = Stop | Singleton a | Yield a (StreamK m a)

185185-- @

186186--

187187-- To facilitate parallel composition we maintain a local state in an 'SVar'

@@ -203,24 +203,25 @@ import Prelude hiding

203203--

204204-- >>> (<>) = Stream.append

205205--

206−type StreamK = Stream

206+{-# DEPRECATED Stream "Please use StreamK instead." #-}

207+type Stream = StreamK

207208

208−newtype Stream m a =

209+newtype StreamK m a =

209210 MkStream (forall r.

210− State Stream m a -- state

211− -> (a -> Stream m a -> m r) -- yield

211+ State StreamK m a -- state

212+ -> (a -> StreamK m a -> m r) -- yield

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

213214 -> m r -- stop

214215 -> m r

215216 )

216217

217218mkStream

218− :: (forall r. State Stream m a

219− -> (a -> Stream m a -> m r)

219+ :: (forall r. State StreamK m a

220+ -> (a -> StreamK m a -> m r)

220221 -> (a -> m r)

221222 -> m r

222223 -> m r)

223− -> Stream m a

224+ -> StreamK m a

224225mkStream = MkStream

225226

226227-- | A terminal function that has no continuation to follow.

@@ -236,16 +237,16 @@ _wrapM :: Monad m => m a -> YieldK m a

236237_wrapM m = (m >>=)

237238

238239-- | Make an empty stream from a stop function.

239−fromStopK :: StopK m -> Stream m a

240+fromStopK :: StopK m -> StreamK m a

240241fromStopK k = mkStream $ \_ _ _ stp -> k stp

241242

242243-- | Make a singleton stream from a callback function. The callback function

243244-- calls the one-shot yield continuation to yield an element.

244−fromYieldK :: YieldK m a -> Stream m a

245+fromYieldK :: YieldK m a -> StreamK m a

245246fromYieldK k = mkStream $ \_ _ sng _ -> k sng

246247

247248-- | Add a yield function at the head of the stream.

248−consK :: YieldK m a -> Stream m a -> Stream m a

249+consK :: YieldK m a -> StreamK m a -> StreamK m a

249250consK k r = mkStream $ \_ yld _ _ -> k (`yld` r)

250251

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

@@ -274,7 +275,7 @@ infixr 5 `cons`

274275-- >>> cons x xs = return x `StreamK.consM` xs

275276--

276277{-# INLINE_NORMAL cons #-}

277−cons :: a -> Stream m a -> Stream m a

278+cons :: a -> StreamK m a -> StreamK m a

278279cons a r = mkStream $ \_ yield _ _ -> yield a r

279280

280281infixr 5 .:

@@ -288,7 +289,7 @@ infixr 5 .:

288289--

289290-- @since 0.1.1

290291{-# INLINE (.:) #-}

291−(.:) :: a -> Stream m a -> Stream m a

292+(.:) :: a -> StreamK m a -> StreamK m a

292293(.:) = cons

293294

294295-- | A stream that terminates without producing any output or side effect.

@@ -297,7 +298,7 @@ infixr 5 .:

297298-- []

298299--

299300{-# INLINE_NORMAL nil #-}

300−nil :: Stream m a

301+nil :: StreamK m a

301302nil = mkStream $ \_ _ _ stp -> stp

302303

303304-- | A stream that terminates without producing any output, but produces a side

@@ -309,7 +310,7 @@ nil = mkStream $ \_ _ _ stp -> stp

309310--

310311-- /Pre-release/

311312{-# INLINE_NORMAL nilM #-}

312−nilM :: Applicative m => m b -> Stream m a

313+nilM :: Applicative m => m b -> StreamK m a

313314nilM m = mkStream $ \_ _ _ stp -> m *> stp

314315

315316-- | Create a singleton stream from a pure value.

@@ -319,7 +320,7 @@ nilM m = mkStream $ \_ _ _ stp -> m *> stp

319320-- >>> fromPure = StreamK.fromEffect . pure

320321--

321322{-# INLINE_NORMAL fromPure #-}

322−fromPure :: a -> Stream m a

323+fromPure :: a -> StreamK m a

323324fromPure a = mkStream $ \_ _ single _ -> single a

324325

325326-- | Create a singleton stream from a monadic action.

@@ -330,7 +331,7 @@ fromPure a = mkStream $ \_ _ single _ -> single a

330331-- hello

331332--

332333{-# INLINE_NORMAL fromEffect #-}

333−fromEffect :: Monad m => m a -> Stream m a

334+fromEffect :: Monad m => m a -> StreamK m a

334335fromEffect m = mkStream $ \_ _ single _ -> m >>= single

335336

336337infixr 5 `consM`

@@ -356,14 +357,14 @@ infixr 5 `consM`

356357-- >>> consM x xs = StreamK.fromEffect x `StreamK.append` xs

357358--

358359{-# INLINE consM #-}

359−{-# SPECIALIZE consM :: IO a -> Stream IO a -> Stream IO a #-}

360−consM :: Monad m => m a -> Stream m a -> Stream m a

360+{-# SPECIALIZE consM :: IO a -> StreamK IO a -> StreamK IO a #-}

361+consM :: Monad m => m a -> StreamK m a -> StreamK m a

361362consM m r = MkStream $ \_ yld _ _ -> m >>= (`yld` r)

362363

363364-- XXX specialize to IO?

364365{-# INLINE consMBy #-}

365366consMBy :: Monad m =>

366− (Stream m a -> Stream m a -> Stream m a) -> m a -> Stream m a -> Stream m a

367+ (StreamK m a -> StreamK m a -> StreamK m a) -> m a -> StreamK m a -> StreamK m a

367368consMBy f m r = fromEffect m `f` r

368369

369370------------------------------------------------------------------------------

@@ -375,11 +376,11 @@ consMBy f m r = fromEffect m `f` r

375376-- SVar passed via the State.

376377{-# INLINE_EARLY foldStreamShared #-}

377378foldStreamShared

378− :: State Stream m a

379− -> (a -> Stream m a -> m r)

379+ :: State StreamK m a

380+ -> (a -> StreamK m a -> m r)

380381 -> (a -> m r)

381382 -> m r

382− -> Stream m a

383+ -> StreamK m a

383384 -> m r

384385foldStreamShared s yield single stop (MkStream k) = k s yield single stop

385386

@@ -388,11 +389,11 @@ foldStreamShared s yield single stop (MkStream k) = k s yield single stop

388389-- passed via State.

389390{-# INLINE foldStream #-}

390391foldStream

391− :: State Stream m a

392− -> (a -> Stream m a -> m r)

392+ :: State StreamK m a

393+ -> (a -> StreamK m a -> m r)

393394 -> (a -> m r)

394395 -> m r

395− -> Stream m a

396+ -> StreamK m a

396397 -> m r

397398foldStream s yield single stop (MkStream k) =

398399 k (adaptState s) yield single stop

@@ -409,16 +410,16 @@ foldStream s yield single stop (MkStream k) =

409410--

410411{-# INLINE foldrSWith #-}

411412foldrSWith ::

412− (forall r. State Stream m b

413− -> (b -> Stream m b -> m r)

413+ (forall r. State StreamK m b

414+ -> (b -> StreamK m b -> m r)

414415 -> (b -> m r)

415416 -> m r

416− -> Stream m b

417+ -> StreamK m b

417418 -> m r)

418− -> (a -> Stream m b -> Stream m b)

419− -> Stream m b

420− -> Stream m a

421− -> Stream m b

419+ -> (a -> StreamK m b -> StreamK m b)

420+ -> StreamK m b

421+ -> StreamK m a

422+ -> StreamK m b

422423foldrSWith f step final m = go m

423424 where

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

@@ -438,10 +439,10 @@ foldrSWith f step final m = go m

438439-- | Fold sharing the SVar state within the reconstructed stream

439440{-# INLINE_NORMAL foldrSShared #-}

440441foldrSShared ::

441− (a -> Stream m b -> Stream m b)

442− -> Stream m b

443− -> Stream m a

444− -> Stream m b

442+ (a -> StreamK m b -> StreamK m b)

443+ -> StreamK m b

444+ -> StreamK m a

445+ -> StreamK m b

445446foldrSShared = foldrSWith foldStreamShared

446447

447448-- XXX consM is a typeclass method, therefore rewritten already. Instead maybe

@@ -477,17 +478,17 @@ foldrSShared = foldrSWith foldStreamShared

477478-- Map (+2) on odd elements and filter out the even elements:

478479--

479480-- >>> step x xs = if odd x then (x + 2) `StreamK.cons` xs else xs

480−-- >>> input = StreamK.fromStream (Stream.fromList [1..5]) :: Stream IO Int

481+-- >>> input = StreamK.fromStream (Stream.fromList [1..5]) :: StreamK IO Int

481482-- >>> Stream.fold Fold.toList $ StreamK.toStream $ StreamK.foldrS step StreamK.nil input

482483-- [3,5,7]

483484--

484485-- /Pre-release/

485486{-# INLINE_NORMAL foldrS #-}

486487foldrS ::

487− (a -> Stream m b -> Stream m b)

488− -> Stream m b

489− -> Stream m a

490− -> Stream m b

488+ (a -> StreamK m b -> StreamK m b)

489+ -> StreamK m b

490+ -> StreamK m a

491+ -> StreamK m b

491492foldrS = foldrSWith foldStream

492493

493494{-# RULES "foldrS/id" foldrS cons nil = \x -> x #-}

@@ -505,16 +506,16 @@ foldrS = foldrSWith foldStream

505506

506507{-# INLINE foldrSMWith #-}

507508foldrSMWith :: Monad m

508− => (forall r. State Stream m b

509− -> (b -> Stream m b -> m r)

509+ => (forall r. State StreamK m b

510+ -> (b -> StreamK m b -> m r)

510511 -> (b -> m r)

511512 -> m r

512− -> Stream m b

513+ -> StreamK m b

513514 -> m r)

514− -> (m a -> Stream m b -> Stream m b)

515− -> Stream m b

516− -> Stream m a

517− -> Stream m b

515+ -> (m a -> StreamK m b -> StreamK m b)

516+ -> StreamK m b

517+ -> StreamK m a

518+ -> StreamK m b

518519foldrSMWith f step final m = go m

519520 where

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

@@ -526,10 +527,10 @@ foldrSMWith f step final m = go m

526527

527528{-# INLINE_NORMAL foldrSM #-}

528529foldrSM :: Monad m

529− => (m a -> Stream m b -> Stream m b)

530− -> Stream m b

531− -> Stream m a

532− -> Stream m b

530+ => (m a -> StreamK m b -> StreamK m b)

531+ -> StreamK m b

532+ -> StreamK m a

533+ -> StreamK m b

533534foldrSM = foldrSMWith foldStream

534535

535536-- {-# RULES "foldrSM/id" foldrSM consM nil = \x -> x #-}

@@ -541,10 +542,10 @@ foldrSM = foldrSMWith foldStream

541542-- Like foldrSM but sharing the SVar state within the recostructed stream.

542543{-# INLINE_NORMAL foldrSMShared #-}

543544foldrSMShared :: Monad m

544− => (m a -> Stream m b -> Stream m b)

545− -> Stream m b

546− -> Stream m a

547− -> Stream m b

545+ => (m a -> StreamK m b -> StreamK m b)

546+ -> StreamK m b

547+ -> StreamK m a

548+ -> StreamK m b

548549foldrSMShared = foldrSMWith foldStreamShared

549550

550551-- {-# RULES "foldrSM/id" foldrSM consM nil = \x -> x #-}

@@ -560,7 +561,7 @@ foldrSMShared = foldrSMWith foldStreamShared

560561-------------------------------------------------------------------------------

561562

562563{-# INLINE_NORMAL build #-}

563−build :: forall m a. (forall b. (a -> b -> b) -> b -> b) -> Stream m a

564+build :: forall m a. (forall b. (a -> b -> b) -> b -> b) -> StreamK m a

564565build g = g cons nil

565566

566567{-# RULES "foldrM/build"

@@ -586,45 +587,45 @@ build g = g cons nil

586587-- build a stream by applying cons and nil to a build function

587588{-# INLINE_NORMAL buildS #-}

588589buildS ::

589− ((a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a)

590− -> Stream m a

590+ ((a -> StreamK m a -> StreamK m a) -> StreamK m a -> StreamK m a)

591+ -> StreamK m a

591592buildS g = g cons nil

592593

593594{-# RULES "foldrS/buildS"

594595 forall k z

595− (g :: (a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a).

596+ (g :: (a -> StreamK m a -> StreamK m a) -> StreamK m a -> StreamK m a).

596597 foldrS k z (buildS g) = g k z #-}

597598

598599{-# RULES "foldrS/cons/buildS"

599600 forall k z x

600− (g :: (a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a).

601+ (g :: (a -> StreamK m a -> StreamK m a) -> StreamK m a -> StreamK m a).

601602 foldrS k z (x `cons` buildS g) = k x (g k z) #-}

602603

603604{-# RULES "foldrSShared/buildS"

604605 forall k z

605− (g :: (a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a).

606+ (g :: (a -> StreamK m a -> StreamK m a) -> StreamK m a -> StreamK m a).

606607 foldrSShared k z (buildS g) = g k z #-}

607608

608609{-# RULES "foldrSShared/cons/buildS"

609610 forall k z x

610− (g :: (a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a).

611+ (g :: (a -> StreamK m a -> StreamK m a) -> StreamK m a -> StreamK m a).

611612 foldrSShared k z (x `cons` buildS g) = k x (g k z) #-}

612613

613614-- build a stream by applying consM and nil to a build function

614615{-# INLINE_NORMAL buildSM #-}

615616buildSM :: Monad m

616− => ((m a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a)

617− -> Stream m a

617+ => ((m a -> StreamK m a -> StreamK m a) -> StreamK m a -> StreamK m a)

618+ -> StreamK m a

618619buildSM g = g consM nil

619620

620621{-# RULES "foldrSM/buildSM"

621622 forall k z

622− (g :: (m a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a).

623+ (g :: (m a -> StreamK m a -> StreamK m a) -> StreamK m a -> StreamK m a).

623624 foldrSM k z (buildSM g) = g k z #-}

624625

625626{-# RULES "foldrSMShared/buildSM"

626627 forall k z

627− (g :: (m a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a).

628+ (g :: (m a -> StreamK m a -> StreamK m a) -> StreamK m a -> StreamK m a).

628629 foldrSMShared k z (buildSM g) = g k z #-}

629630

630631-- Disabled because this may not fire as consM is a class Op

@@ -640,25 +641,25 @@ buildSM g = g consM nil

640641-- reconstructing a stream.

641642{-# INLINE_NORMAL buildM #-}

642643buildM :: Monad m

643− => (forall r. (a -> Stream m a -> m r)

644+ => (forall r. (a -> StreamK m a -> m r)

644645 -> (a -> m r)

645646 -> m r

646647 -> m r

647648 )

648− -> Stream m a

649+ -> StreamK m a

649650buildM g = mkStream $ \st yld sng stp ->

650651 g (\a r -> foldStream st yld sng stp (return a `consM` r)) sng stp

651652

652653-- | Like 'buildM' but shares the SVar state across computations.

653654{-# INLINE_NORMAL sharedMWith #-}

654655sharedMWith :: Monad m

655− => (m a -> Stream m a -> Stream m a)

656− -> (forall r. (a -> Stream m a -> m r)

656+ => (m a -> StreamK m a -> StreamK m a)

657+ -> (forall r. (a -> StreamK m a -> m r)

657658 -> (a -> m r)

658659 -> m r

659660 -> m r

660661 )

661− -> Stream m a

662+ -> StreamK m a

662663sharedMWith cns g = mkStream $ \st yld sng stp ->

663664 g (\a r -> foldStreamShared st yld sng stp (return a `cns` r)) sng stp

664665

@@ -668,50 +669,50 @@ sharedMWith cns g = mkStream $ \st yld sng stp ->

668669

669670{-# INLINE_NORMAL augmentS #-}

670671augmentS ::

671− ((a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a)

672− -> Stream m a

673− -> Stream m a

672+ ((a -> StreamK m a -> StreamK m a) -> StreamK m a -> StreamK m a)

673+ -> StreamK m a

674+ -> StreamK m a

674675augmentS g xs = g cons xs

675676

676677{-# RULES "augmentS/nil"

677− forall (g :: (a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a).

678+ forall (g :: (a -> StreamK m a -> StreamK m a) -> StreamK m a -> StreamK m a).

678679 augmentS g nil = buildS g

679680 #-}

680681

681682{-# RULES "foldrS/augmentS"

682683 forall k z xs

683− (g :: (a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a).

684+ (g :: (a -> StreamK m a -> StreamK m a) -> StreamK m a -> StreamK m a).

684685 foldrS k z (augmentS g xs) = g k (foldrS k z xs)

685686 #-}

686687

687688{-# RULES "augmentS/buildS"

688− forall (g :: (a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a)

689− (h :: (a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a).

689+ forall (g :: (a -> StreamK m a -> StreamK m a) -> StreamK m a -> StreamK m a)

690+ (h :: (a -> StreamK m a -> StreamK m a) -> StreamK m a -> StreamK m a).

690691 augmentS g (buildS h) = buildS (\c n -> g c (h c n))

691692 #-}

692693

693694{-# INLINE_NORMAL augmentSM #-}

694695augmentSM :: Monad m =>

695− ((m a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a)

696− -> Stream m a -> Stream m a

696+ ((m a -> StreamK m a -> StreamK m a) -> StreamK m a -> StreamK m a)

697+ -> StreamK m a -> StreamK m a

697698augmentSM g xs = g consM xs

698699

699700{-# RULES "augmentSM/nil"

700701 forall

701− (g :: (m a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a).

702+ (g :: (m a -> StreamK m a -> StreamK m a) -> StreamK m a -> StreamK m a).

702703 augmentSM g nil = buildSM g

703704 #-}

704705

705706{-# RULES "foldrSM/augmentSM"

706707 forall k z xs

707− (g :: (m a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a).

708+ (g :: (m a -> StreamK m a -> StreamK m a) -> StreamK m a -> StreamK m a).

708709 foldrSM k z (augmentSM g xs) = g k (foldrSM k z xs)

709710 #-}

710711

711712{-# RULES "augmentSM/buildSM"

712713 forall

713− (g :: (m a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a)

714− (h :: (m a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a).

714+ (g :: (m a -> StreamK m a -> StreamK m a) -> StreamK m a -> StreamK m a)

715+ (h :: (m a -> StreamK m a -> StreamK m a) -> StreamK m a -> StreamK m a).

715716 augmentSM g (buildSM h) = buildSM (\c n -> g c (h c n))

716717 #-}

717718

@@ -721,7 +722,7 @@ augmentSM g xs = g consM xs

721722

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

723724{-# INLINE_NORMAL foldrM #-}

724−foldrM :: (a -> m b -> m b) -> m b -> Stream m a -> m b

725+foldrM :: (a -> m b -> m b) -> m b -> StreamK m a -> m b

725726foldrM step acc m = go m

726727 where

727728 go m1 =

@@ -732,15 +733,15 @@ foldrM step acc m = go m

732733

733734{-# INLINE_NORMAL foldrMKWith #-}

734735foldrMKWith

735− :: (State Stream m a

736− -> (a -> Stream m a -> m b)

736+ :: (State StreamK m a

737+ -> (a -> StreamK m a -> m b)

737738 -> (a -> m b)

738739 -> m b

739− -> Stream m a

740+ -> StreamK m a

740741 -> m b)

741742 -> (a -> m b -> m b)

742743 -> m b

743− -> ((a -> Stream m a -> m b) -> (a -> m b) -> m b -> m b)

744+ -> ((a -> StreamK m a -> m b) -> (a -> m b) -> m b -> m b)

744745 -> m b

745746foldrMKWith f step acc = go

746747 where

@@ -760,7 +761,7 @@ foldrMKWith f step acc = go

760761-- XXX in which case will foldrM/buildM fusion be useful?

761762{-# RULES "foldrM/buildM"

762763 forall step acc (g :: (forall r.

763− (a -> Stream m a -> m r)

764+ (a -> StreamK m a -> m r)

764765 -> (a -> m r)

765766 -> m r

766767 -> m r

@@ -771,7 +772,7 @@ foldrMKWith f step acc = go

771772{-

772773{-# RULES "foldrM/sharedM"

773774 forall step acc (g :: (forall r.

774− (a -> Stream m a -> m r)

775+ (a -> StreamK m a -> m r)

775776 -> (a -> m r)

776777 -> m r

777778 -> m r

@@ -792,11 +793,11 @@ foldrMKWith f step acc = go

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

793794{-# INLINE foldlx' #-}

794795foldlx' :: forall m a b x. Monad m

795− => (x -> a -> x) -> x -> (x -> b) -> Stream m a -> m b

796+ => (x -> a -> x) -> x -> (x -> b) -> StreamK m a -> m b

796797foldlx' step begin done m = get $ go m begin

797798 where

798799 {-# NOINLINE get #-}

799− get :: Stream m x -> m b

800+ get :: StreamK m x -> m b

800801 get m1 =

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

802803 -- okay?

@@ -808,7 +809,7 @@ foldlx' step begin done m = get $ go m begin

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

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

810811 -- continuations is much more efficient.

811− go :: Stream m a -> x -> Stream m x

812+ go :: StreamK m a -> x -> StreamK m x

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

813814 let stop = sng acc

814815 single a = sng $ step acc a

@@ -819,7 +820,7 @@ foldlx' step begin done m = get $ go m begin

819820

820821-- | Strict left associative fold.

821822{-# INLINE foldl' #-}

822−foldl' :: Monad m => (b -> a -> b) -> b -> Stream m a -> m b

823+foldl' :: Monad m => (b -> a -> b) -> b -> StreamK m a -> m b

823824foldl' step begin = foldlx' step begin id

824825

825826------------------------------------------------------------------------------

@@ -836,7 +837,7 @@ foldl' step begin = foldlx' step begin id

836837-- > drain = foldl' (\_ _ -> ()) ()

837838-- > drain = mapM_ (\_ -> return ())

838839{-# INLINE drain #-}

839−drain :: Monad m => Stream m a -> m ()

840+drain :: Monad m => StreamK m a -> m ()

840841drain = foldrM (\_ xs -> xs) (return ())

841842{-

842843drain = go

@@ -849,7 +850,7 @@ drain = go

849850-}

850851

851852{-# INLINE null #-}

852−null :: Monad m => Stream m a -> m Bool

853+null :: Monad m => StreamK m a -> m Bool

853854-- null = foldrM (\_ _ -> return True) (return False)

854855null m =

855856 let stop = return True

@@ -876,7 +877,7 @@ infixr 6 `append`

876877-- 'concatMapWith' et. al.

877878--

878879{-# INLINE append #-}

879−append :: Stream m a -> Stream m a -> Stream m a

880+append :: StreamK m a -> StreamK m a -> StreamK m a

880881-- XXX This doubles the time of toNullAp benchmark, may not be fusing properly

881882-- serial xs ys = augmentS (\c n -> foldrS c n xs) ys

882883append m1 m2 = go m1

@@ -889,17 +890,17 @@ append m1 m2 = go m1

889890

890891-- join/merge/append streams depending on consM

891892{-# INLINE conjoin #-}

892−conjoin :: Monad m => Stream m a -> Stream m a -> Stream m a

893+conjoin :: Monad m => StreamK m a -> StreamK m a -> StreamK m a

893894conjoin xs = augmentSM (\c n -> foldrSM c n xs)

894895

895−instance Semigroup (Stream m a) where

896+instance Semigroup (StreamK m a) where

896897 (<>) = append

897898

898899------------------------------------------------------------------------------

899900-- Monoid

900901------------------------------------------------------------------------------

901902

902−instance Monoid (Stream m a) where

903+instance Monoid (StreamK m a) where

903904 mempty = nil

904905 mappend = (<>)

905906

@@ -912,11 +913,11 @@ instance Monoid (Stream m a) where

912913-- rewrite rules. Ideally don't run hlint on this file.

913914{-# INLINE_LATE mapFB #-}

914915mapFB :: forall b m a.

915− (b -> Stream m b -> Stream m b)

916+ (b -> StreamK m b -> StreamK m b)

916917 -> (a -> b)

917918 -> a

918− -> Stream m b

919− -> Stream m b

919+ -> StreamK m b

920+ -> StreamK m b

920921mapFB c f = \x ys -> c (f x) ys

921922

922923{-# RULES

@@ -925,7 +926,7 @@ mapFB c f = \x ys -> c (f x) ys

925926 #-}

926927

927928{-# INLINE map #-}

928−map :: (a -> b) -> Stream m a -> Stream m b

929+map :: (a -> b) -> StreamK m a -> StreamK m b

929930map f xs = buildS (\c n -> foldrS (mapFB c f) n xs)

930931

931932-- XXX This definition might potentially be more efficient, but the cost in the

@@ -964,15 +965,15 @@ mapMFB c f x = c (x >>= f)

964965-- XXX fusion would be easier for monomoprhic stream types.

965966-- {-# RULES "mapM serial" mapM = mapMSerial #-}

966967{-# INLINE mapMSerial #-}

967−mapMSerial :: Monad m => (a -> m b) -> Stream m a -> Stream m b

968+mapMSerial :: Monad m => (a -> m b) -> StreamK m a -> StreamK m b

968969mapMSerial f xs = buildSM (\c n -> foldrSMShared (mapMFB c f) n xs)

969970

970971{-# INLINE mapMWith #-}

971972mapMWith ::

972− (m b -> Stream m b -> Stream m b)

973+ (m b -> StreamK m b -> StreamK m b)

973974 -> (a -> m b)

974− -> Stream m a

975− -> Stream m b

975+ -> StreamK m a

976+ -> StreamK m b

976977mapMWith cns f = foldrSShared (\x xs -> f x `cns` xs) nil

977978

978979{-

@@ -987,7 +988,7 @@ mapMWith cns f = go

987988

988989-- XXX in fact use the Stream type everywhere and only use polymorphism in the

989990-- high level modules/prelude.

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

991+instance Monad m => Functor (StreamK m) where

991992 fmap = map

992993

993994------------------------------------------------------------------------------

@@ -1003,8 +1004,8 @@ instance Monad m => Functor (Stream m) where

10031004-- well, because they do not get inlined. Need to add INLINE in Ord class in

10041005-- base?

10051006

1006−instance IsList (Stream Identity a) where

1007− type (Item (Stream Identity a)) = a

1007+instance IsList (StreamK Identity a) where

1008+ type (Item (StreamK Identity a)) = a

10081009

10091010 {-# INLINE fromList #-}

10101011 fromList = fromFoldable

@@ -1014,11 +1015,11 @@ instance IsList (Stream Identity a) where

10141015

10151016-- XXX Fix these

10161017{-

1017−instance Eq a => Eq (Stream Identity a) where

1018+instance Eq a => Eq (StreamK Identity a) where

10181019 {-# INLINE (==) #-}

10191020 (==) xs ys = runIdentity $ eqBy (==) xs ys

10201021

1021−instance Ord a => Ord (Stream Identity a) where

1022+instance Ord a => Ord (StreamK Identity a) where

10221023 {-# INLINE compare #-}

10231024 compare xs ys = runIdentity $ cmpBy compare xs ys

10241025

@@ -1053,18 +1054,18 @@ instance Ord a => Ord (Stream Identity a) where

10531054 min x y = if x <= y then x else y

10541055-}

10551056

1056−instance Show a => Show (Stream Identity a) where

1057+instance Show a => Show (StreamK Identity a) where

10571058 showsPrec p dl = showParen (p > 10) $

10581059 showString "fromList " . shows (toList dl)

10591060

1060−instance Read a => Read (Stream Identity a) where

1061+instance Read a => Read (StreamK Identity a) where

10611062 readPrec = parens $ prec 10 $ do

10621063 Ident "fromList" <- lexP

10631064 fromList <$> readPrec

10641065

10651066 readListPrec = readListPrecDefault

10661067

1067−instance (a ~ Char) => IsString (Stream Identity a) where

1068+instance (a ~ Char) => IsString (StreamK Identity a) where

10681069 {-# INLINE fromString #-}

10691070 fromString = fromList

10701071

@@ -1074,7 +1075,7 @@ instance (a ~ Char) => IsString (Stream Identity a) where

10741075

10751076-- | Lazy right associative fold.

10761077{-# INLINE foldr #-}

1077−foldr :: Monad m => (a -> b -> b) -> b -> Stream m a -> m b

1078+foldr :: Monad m => (a -> b -> b) -> b -> StreamK m a -> m b

10781079foldr step acc = foldrM (\x xs -> xs >>= \b -> return (step x b)) (return acc)

10791080

10801081-- The default Foldable instance has several issues:

@@ -1084,7 +1085,7 @@ foldr step acc = foldrM (\x xs -> xs >>= \b -> return (step x b)) (return acc)

10841085-- use right folds, they cannot run in constant memory. We provide

10851086-- implementations using strict left folds here.

10861087

1087−instance (Foldable m, Monad m) => Foldable (Stream m) where

1088+instance (Foldable m, Monad m) => Foldable (StreamK m) where

10881089

10891090 {-# INLINE foldMap #-}

10901091 foldMap f =

@@ -1136,7 +1137,7 @@ instance (Foldable m, Monad m) => Foldable (Stream m) where

11361137-- Traversable

11371138-------------------------------------------------------------------------------

11381139

1139−instance Traversable (Stream Identity) where

1140+instance Traversable (StreamK Identity) where

11401141 {-# INLINE traverse #-}

11411142 traverse f xs =

11421143 runIdentity

@@ -1153,17 +1154,17 @@ instance Traversable (Stream Identity) where

11531154

11541155-- | Detach a stream from an SVar

11551156{-# INLINE unShare #-}

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

1157+unShare :: StreamK m a -> StreamK m a

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

11581159 foldStream st yld sng stp x

11591160

11601161-- XXX the function stream and value stream can run in parallel

11611162{-# INLINE crossApplyWith #-}

11621163crossApplyWith ::

1163− (Stream m b -> Stream m b -> Stream m b)

1164− -> Stream m (a -> b)

1165− -> Stream m a

1166− -> Stream m b

1164+ (StreamK m b -> StreamK m b -> StreamK m b)

1165+ -> StreamK m (a -> b)

1166+ -> StreamK m a

1167+ -> StreamK m b

11671168crossApplyWith par fstream stream = go1 fstream

11681169

11691170 where

@@ -1192,9 +1193,9 @@ crossApplyWith par fstream stream = go1 fstream

11921193--

11931194{-# INLINE crossApply #-}

11941195crossApply ::

1195− Stream m (a -> b)

1196− -> Stream m a

1197− -> Stream m b

1196+ StreamK m (a -> b)

1197+ -> StreamK m a

1198+ -> StreamK m b

11981199crossApply fstream stream = go1 fstream

11991200

12001201 where

@@ -1222,9 +1223,9 @@ crossApply fstream stream = go1 fstream

12221223

12231224{-# INLINE crossApplySnd #-}

12241225crossApplySnd ::

1225− Stream m a

1226− -> Stream m b

1227− -> Stream m b

1226+ StreamK m a

1227+ -> StreamK m b

1228+ -> StreamK m b

12281229crossApplySnd fstream stream = go1 fstream

12291230

12301231 where

@@ -1246,9 +1247,9 @@ crossApplySnd fstream stream = go1 fstream

12461247

12471248{-# INLINE crossApplyFst #-}

12481249crossApplyFst ::

1249− Stream m a

1250− -> Stream m b

1251− -> Stream m a

1250+ StreamK m a

1251+ -> StreamK m b

1252+ -> StreamK m a

12521253crossApplyFst fstream stream = go1 fstream

12531254

12541255 where

@@ -1282,10 +1283,10 @@ crossApplyFst fstream stream = go1 fstream

12821283-- Note that the second stream is evaluated multiple times.

12831284--

12841285{-# INLINE crossWith #-}

1285−crossWith :: Monad m => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c

1286+crossWith :: Monad m => (a -> b -> c) -> StreamK m a -> StreamK m b -> StreamK m c

12861287crossWith f m1 m2 = fmap f m1 `crossApply` m2

12871288

1288−-- | Given a @Stream m a@ and @Stream m b@ generate a stream with all possible

1289+-- | Given a @StreamK m a@ and @StreamK m b@ generate a stream with all possible

12891290-- combinations of the tuple @(a, b)@.

12901291--

12911292-- Definition:

@@ -1304,7 +1305,7 @@ crossWith f m1 m2 = fmap f m1 `crossApply` m2

13041305--

13051306-- /Pre-release/

13061307{-# INLINE cross #-}

1307−cross :: Monad m => Stream m a -> Stream m b -> Stream m (a, b)

1308+cross :: Monad m => StreamK m a -> StreamK m b -> StreamK m (a, b)

13081309cross = crossWith (,)

13091310

13101311-- XXX This is just concatMapWith with arguments flipped. We need to keep this

@@ -1313,10 +1314,10 @@ cross = crossWith (,)

13131314-- if the argument order is changed.

13141315{-# INLINE bindWith #-}

13151316bindWith ::

1316− (Stream m b -> Stream m b -> Stream m b)

1317− -> Stream m a

1318− -> (a -> Stream m b)

1319− -> Stream m b

1317+ (StreamK m b -> StreamK m b -> StreamK m b)

1318+ -> StreamK m a

1319+ -> (a -> StreamK m b)

1320+ -> StreamK m b

13201321bindWith par m1 f = go m1

13211322 where

13221323 go m =

@@ -1341,14 +1342,14 @@ bindWith par m1 f = go m1

13411342{-# INLINE concatMapWith #-}

13421343concatMapWith

13431344 ::

1344− (Stream m b -> Stream m b -> Stream m b)

1345− -> (a -> Stream m b)

1346− -> Stream m a

1347− -> Stream m b

1345+ (StreamK m b -> StreamK m b -> StreamK m b)

1346+ -> (a -> StreamK m b)

1347+ -> StreamK m a

1348+ -> StreamK m b

13481349concatMapWith par f xs = bindWith par xs f

13491350

13501351{-# INLINE concatMap #-}

1351−concatMap :: (a -> Stream m b) -> Stream m a -> Stream m b

1352+concatMap :: (a -> StreamK m b) -> StreamK m a -> StreamK m b

13521353concatMap = concatMapWith append

13531354

13541355{-

@@ -1392,10 +1393,10 @@ concatMap_ f xs = buildS

13921393{-# INLINE mergeMapWith #-}

13931394mergeMapWith

13941395 ::

1395− (Stream m b -> Stream m b -> Stream m b)

1396− -> (a -> Stream m b)

1397− -> Stream m a

1398− -> Stream m b

1396+ (StreamK m b -> StreamK m b -> StreamK m b)

1397+ -> (a -> StreamK m b)

1398+ -> StreamK m a

1399+ -> StreamK m b

13991400mergeMapWith combine f str = go (leafPairs str)

14001401

14011402 where

@@ -1448,7 +1449,7 @@ mergeMapWith combine f str = go (leafPairs str)

14481449 in foldStream (adaptState st) yieldk single stop stream

14491450

14501451{-

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

1452+instance Monad m => Applicative (StreamK m) where

14521453 {-# INLINE pure #-}

14531454 pure = fromPure

14541455

@@ -1467,7 +1468,7 @@ instance Monad m => Applicative (Stream m) where

14671468-- NOTE: even though concatMap for StreamD is 3x faster compared to StreamK,

14681469-- the monad instance of StreamD is slower than StreamK after foldr/build

14691470-- fusion.

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

1471+instance Monad m => Monad (StreamK m) where

14711472 {-# INLINE return #-}

14721473 return = pure

14731474

@@ -1506,10 +1507,10 @@ concatUnfoldr = undefined

15061507--

15071508{-# INLINE concatIterateWith #-}

15081509concatIterateWith ::

1509− (Stream m a -> Stream m a -> Stream m a)

1510− -> (a -> Stream m a)

1511− -> Stream m a

1512− -> Stream m a

1510+ (StreamK m a -> StreamK m a -> StreamK m a)

1511+ -> (a -> StreamK m a)

1512+ -> StreamK m a

1513+ -> StreamK m a

15131514concatIterateWith combine f = iterateStream

15141515

15151516 where

@@ -1532,10 +1533,10 @@ concatIterateWith combine f = iterateStream

15321533--

15331534{-# INLINE mergeIterateWith #-}

15341535mergeIterateWith ::

1535− (Stream m a -> Stream m a -> Stream m a)

1536− -> (a -> Stream m a)

1537− -> Stream m a

1538− -> Stream m a

1536+ (StreamK m a -> StreamK m a -> StreamK m a)

1537+ -> (a -> StreamK m a)

1538+ -> StreamK m a

1539+ -> StreamK m a

15391540mergeIterateWith combine f = iterateStream

15401541

15411542 where

@@ -1566,11 +1567,11 @@ mergeIterateWith combine f = iterateStream

15661567{-# INLINE concatIterateScanWith #-}

15671568concatIterateScanWith

15681569 :: Monad m

1569− => (Stream m a -> Stream m a -> Stream m a)

1570− -> (b -> a -> m (b, Stream m a))

1570+ => (StreamK m a -> StreamK m a -> StreamK m a)

1571+ -> (b -> a -> m (b, StreamK m a))

15711572 -> m b

1572− -> Stream m a

1573− -> Stream m a

1573+ -> StreamK m a

1574+ -> StreamK m a

15741575concatIterateScanWith combine f initial stream =

15751576 concatEffect $ do

15761577 b <- initial

@@ -1598,9 +1599,9 @@ concatIterateScanWith combine f initial stream =

15981599{-

15991600concatMapEitherWith

16001601 :: (forall x. t m x -> t m x -> t m x)

1601− -> (a -> t m (Either (Stream m b) b))

1602− -> Stream m a

1603− -> Stream m b

1602+ -> (a -> t m (Either (StreamK m b) b))

1603+ -> StreamK m a

1604+ -> StreamK m b

16041605concatMapEitherWith = undefined

16051606-}

16061607

@@ -1622,10 +1623,10 @@ concatMapEitherWith = undefined

16221623{-# INLINE concatIterateLeftsWith #-}

16231624concatIterateLeftsWith

16241625 :: (b ~ Either a c)

1625− => (Stream m b -> Stream m b -> Stream m b)

1626− -> (a -> Stream m b)

1627− -> Stream m b

1628− -> Stream m b

1626+ => (StreamK m b -> StreamK m b -> StreamK m b)

1627+ -> (a -> StreamK m b)

1628+ -> StreamK m b

1629+ -> StreamK m b

16291630concatIterateLeftsWith combine f =

16301631 concatIterateWith combine (either f (const nil))

16311632

@@ -1650,7 +1651,7 @@ infixr 6 `interleave`

16501651-- streams.

16511652--

16521653{-# INLINE interleave #-}

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

1654+interleave :: StreamK m a -> StreamK m a -> StreamK m a

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

16551656 let stop = foldStream st yld sng stp m2

16561657 single a = yld a m2

@@ -1662,7 +1663,7 @@ infixr 6 `interleaveFst`

16621663-- | Like `interleave` but stops interleaving as soon as the first stream stops.

16631664--

16641665{-# INLINE interleaveFst #-}

1665−interleaveFst :: Stream m a -> Stream m a -> Stream m a

1666+interleaveFst :: StreamK m a -> StreamK m a -> StreamK m a

16661667interleaveFst m1 m2 = mkStream $ \st yld sng stp -> do

16671668 let yieldFirst a r = yld a (yieldSecond r m2)

16681669 in foldStream st yieldFirst sng stp m1

@@ -1681,7 +1682,7 @@ infixr 6 `interleaveMin`

16811682-- stops.

16821683--

16831684{-# INLINE interleaveMin #-}

1684−interleaveMin :: Stream m a -> Stream m a -> Stream m a

1685+interleaveMin :: StreamK m a -> StreamK m a -> StreamK m a

16851686interleaveMin m1 m2 = mkStream $ \st yld _ stp -> do

16861687 let stop = stp

16871688 -- "single a" is defined as "yld a (interleaveMin m2 nil)" instead of

@@ -1696,7 +1697,7 @@ interleaveMin m1 m2 = mkStream $ \st yld _ stp -> do

16961697-------------------------------------------------------------------------------

16971698

16981699{-# INLINE unfoldr #-}

1699−unfoldr :: (b -> Maybe (a, b)) -> b -> Stream m a

1700+unfoldr :: (b -> Maybe (a, b)) -> b -> StreamK m a

17001701unfoldr next s0 = build $ \yld stp ->

17011702 let go s =

17021703 case next s of

@@ -1706,10 +1707,10 @@ unfoldr next s0 = build $ \yld stp ->

17061707

17071708{-# INLINE unfoldrMWith #-}

17081709unfoldrMWith :: Monad m =>

1709− (m a -> Stream m a -> Stream m a)

1710+ (m a -> StreamK m a -> StreamK m a)

17101711 -> (b -> m (Maybe (a, b)))

17111712 -> b

1712− -> Stream m a

1713+ -> StreamK m a

17131714unfoldrMWith cns step = go

17141715

17151716 where

@@ -1721,14 +1722,14 @@ unfoldrMWith cns step = go

17211722 Nothing -> stp

17221723

17231724{-# INLINE unfoldrM #-}

1724−unfoldrM :: Monad m => (b -> m (Maybe (a, b))) -> b -> Stream m a

1725+unfoldrM :: Monad m => (b -> m (Maybe (a, b))) -> b -> StreamK m a

17251726unfoldrM = unfoldrMWith consM

17261727

17271728-- | Generate an infinite stream by repeating a pure value.

17281729--

17291730-- /Pre-release/

17301731{-# INLINE repeat #-}

1731−repeat :: a -> Stream m a

1732+repeat :: a -> StreamK m a

17321733repeat a = let x = cons a x in x

17331734

17341735-- | Like 'repeatM' but takes a stream 'cons' operation to combine the actions

@@ -1744,7 +1745,7 @@ repeatMWith cns = go

17441745 go m = m `cns` go m

17451746

17461747{-# INLINE replicateMWith #-}

1747−replicateMWith :: (m a -> Stream m a -> Stream m a) -> Int -> m a -> Stream m a

1748+replicateMWith :: (m a -> StreamK m a -> StreamK m a) -> Int -> m a -> StreamK m a

17481749replicateMWith cns n m = go n

17491750

17501751 where

@@ -1753,7 +1754,7 @@ replicateMWith cns n m = go n

17531754

17541755{-# INLINE fromIndicesMWith #-}

17551756fromIndicesMWith ::

1756− (m a -> Stream m a -> Stream m a) -> (Int -> m a) -> Stream m a

1757+ (m a -> StreamK m a -> StreamK m a) -> (Int -> m a) -> StreamK m a

17571758fromIndicesMWith cns gen = go 0

17581759

17591760 where

@@ -1763,7 +1764,7 @@ fromIndicesMWith cns gen = go 0

17631764

17641765{-# INLINE iterateMWith #-}

17651766iterateMWith :: Monad m =>

1766− (m a -> Stream m a -> Stream m a) -> (a -> m a) -> m a -> Stream m a

1767+ (m a -> StreamK m a -> StreamK m a) -> (a -> m a) -> m a -> StreamK m a

17671768iterateMWith cns step = go

17681769

17691770 where

@@ -1773,11 +1774,11 @@ iterateMWith cns step = go

17731774 foldStreamShared st stp sng yld (return next `cns` go (step next))

17741775

17751776{-# INLINE headPartial #-}

1776−headPartial :: Monad m => Stream m a -> m a

1777+headPartial :: Monad m => StreamK m a -> m a

17771778headPartial = foldrM (\x _ -> return x) (error "head of nil")

17781779

17791780{-# INLINE tailPartial #-}

1780−tailPartial :: Stream m a -> Stream m a

1781+tailPartial :: StreamK m a -> StreamK m a

17811782tailPartial m = mkStream $ \st yld sng stp ->

17821783 let stop = error "tail of nil"

17831784 single _ = stp

@@ -1835,7 +1836,7 @@ tailPartial m = mkStream $ \st yld sng stp ->

18351836--

18361837-- /Pre-release/

18371838{-# INLINE mfix #-}

1838−mfix :: Monad m => (m a -> Stream m a) -> Stream m a

1839+mfix :: Monad m => (m a -> StreamK m a) -> StreamK m a

18391840mfix f = mkStream $ \st yld sng stp ->

18401841 let single a = foldStream st yld sng stp $ a `cons` ys

18411842 yieldk a _ = foldStream st yld sng stp $ a `cons` ys

@@ -1862,11 +1863,11 @@ mfix f = mkStream $ \st yld sng stp ->

18621863--

18631864-- @since 0.2.0

18641865{-# INLINE fromFoldable #-}

1865−fromFoldable :: Foldable f => f a -> Stream m a

1866+fromFoldable :: Foldable f => f a -> StreamK m a

18661867fromFoldable = Prelude.foldr cons nil

18671868

18681869{-# INLINE fromFoldableM #-}

1869−fromFoldableM :: (Foldable f, Monad m) => f (m a) -> Stream m a

1870+fromFoldableM :: (Foldable f, Monad m) => f (m a) -> StreamK m a

18701871fromFoldableM = Prelude.foldr consM nil

18711872

18721873-------------------------------------------------------------------------------

@@ -1874,7 +1875,7 @@ fromFoldableM = Prelude.foldr consM nil

18741875-------------------------------------------------------------------------------

18751876

18761877{-# INLINE uncons #-}

1877−uncons :: Applicative m => Stream m a -> m (Maybe (a, Stream m a))

1878+uncons :: Applicative m => StreamK m a -> m (Maybe (a, StreamK m a))

18781879uncons m =

18791880 let stop = pure Nothing

18801881 single a = pure (Just (a, nil))

@@ -1882,7 +1883,7 @@ uncons m =

18821883 in foldStream defState yieldk single stop m

18831884

18841885{-# INLINE tail #-}

1885−tail :: Applicative m => Stream m a -> m (Maybe (Stream m a))

1886+tail :: Applicative m => StreamK m a -> m (Maybe (StreamK m a))

18861887tail =

18871888 let stop = pure Nothing

18881889 single _ = pure $ Just nil

@@ -1895,7 +1896,7 @@ tail =

18951896--

18961897-- /Pre-release/

18971898{-# INLINE init #-}

1898−init :: Applicative m => Stream m a -> m (Maybe (Stream m a))

1899+init :: Applicative m => StreamK m a -> m (Maybe (StreamK m a))

18991900init = go1

19001901 where

19011902 go1 m1 = do

@@ -1914,7 +1915,7 @@ init = go1

19141915-- | Lazy left fold to a stream.

19151916{-# INLINE foldlS #-}

19161917foldlS ::

1917− (Stream m b -> a -> Stream m b) -> Stream m b -> Stream m a -> Stream m b

1918+ (StreamK m b -> a -> StreamK m b) -> StreamK m b -> StreamK m a -> StreamK m b

19181919foldlS step = go

19191920 where

19201921 go acc rest = mkStream $ \st yld sng stp ->

@@ -1925,7 +1926,7 @@ foldlS step = go

19251926 in foldStream (adaptState st) yieldk single stop rest

19261927

19271928{-# INLINE reverse #-}

1928−reverse :: Stream m a -> Stream m a

1929+reverse :: StreamK m a -> StreamK m a

19291930reverse = foldlS (flip cons) nil

19301931

19311932------------------------------------------------------------------------------

@@ -1934,20 +1935,20 @@ reverse = foldlS (flip cons) nil

19341935

19351936-- | Run an action before evaluating the stream.

19361937{-# INLINE before #-}

1937−before :: Monad m => m b -> Stream m a -> Stream m a

1938+before :: Monad m => m b -> StreamK m a -> StreamK m a

19381939before action stream =

19391940 mkStream $ \st yld sng stp ->

19401941 action >> foldStreamShared st yld sng stp stream

19411942

19421943-- | concat . fromEffect

19431944{-# INLINE concatEffect #-}

1944−concatEffect :: Monad m => m (Stream m a) -> Stream m a

1945+concatEffect :: Monad m => m (StreamK m a) -> StreamK m a

19451946concatEffect action =

19461947 mkStream $ \st yld sng stp ->

19471948 action >>= foldStreamShared st yld sng stp

19481949

19491950{-# INLINE concatMapEffect #-}

1950−concatMapEffect :: Monad m => (b -> Stream m a) -> m b -> Stream m a

1951+concatMapEffect :: Monad m => (b -> StreamK m a) -> m b -> StreamK m a

19511952concatMapEffect f action =

19521953 mkStream $ \st yld sng stp ->

19531954 action >>= foldStreamShared st yld sng stp . f

@@ -1980,7 +1981,7 @@ concatMapEffect f action =

19801981-- :}

19811982-- [(1,3),(1,4),(2,3),(2,4)]

19821983--

1983−newtype CrossStreamK m a = CrossStreamK {unCrossStreamK :: Stream m a}

1984+newtype CrossStreamK m a = CrossStreamK {unCrossStreamK :: StreamK m a}

19841985 deriving (Functor, Semigroup, Monoid, Foldable)

19851986

19861987-- | Wrap the 'StreamK' type in a 'CrossStreamK' newtype to enable cross

core/src/Streamly/Internal/Data/Stream/Type.hs+3−3

@@ -79,7 +79,7 @@ import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

7979--

8080-- >>> (<>) = Stream.append

8181--

82−newtype StreamK m a = StreamK (K.Stream m a)

82+newtype StreamK m a = StreamK (K.StreamK m a)

8383 -- XXX when deriving do we inherit an INLINE?

8484 deriving (Semigroup, Monoid)

8585

@@ -90,11 +90,11 @@ type Stream = StreamK

9090------------------------------------------------------------------------------

9191

9292{-# INLINE_EARLY fromStreamK #-}

93−fromStreamK :: K.Stream m a -> Stream m a

93+fromStreamK :: K.StreamK m a -> Stream m a

9494fromStreamK = StreamK

9595

9696{-# INLINE_EARLY toStreamK #-}

97−toStreamK :: Stream m a -> K.Stream m a

97+toStreamK :: Stream m a -> K.StreamK m a

9898toStreamK (StreamK k) = k

9999

100100{-# INLINE_EARLY fromStreamD #-}

core/src/Streamly/Internal/Data/Unfold.hs+1−1

@@ -591,7 +591,7 @@ fromStreamD = Unfold step pure

591591 Stop -> Stop) <$> step1 defState state1

592592

593593{-# INLINE_NORMAL fromStreamK #-}

594−fromStreamK :: Applicative m => Unfold m (K.Stream m a) a

594+fromStreamK :: Applicative m => Unfold m (K.StreamK m a) a

595595fromStreamK = Unfold step pure

596596

597597 where

core/src/inline.hs+1−1

@@ -16,7 +16,7 @@

1616-- Then, fromStreamK/toStreamK are inlined in the last phase:

1717--

1818-- {-# INLINE_LATE toStreamK #-}

19−-- toStreamK :: Monad m => Stream m a -> K.Stream m a```

19+-- toStreamK :: Monad m => Stream m a -> K.StreamK m a```

2020--

2121-- The fallback rules make sure that if we could not fuse the direct style

2222-- operations then better use the CPS style operation, because unfused direct

src/Streamly/Internal/Data/Stream/Channel/Types.hs+4−4

@@ -815,10 +815,10 @@ cleanupSVar workerSet = do

815815-- Used for concurrent evaluation of streams using a Channel.

816816{-# INLINE concatMapDivK #-}

817817concatMapDivK :: Monad m =>

818− (K.Stream m a -> m ())

819− -> (a -> K.Stream m b)

820− -> K.Stream m a

821− -> K.Stream m b

818+ (K.StreamK m a -> m ())

819+ -> (a -> K.StreamK m b)

820+ -> K.StreamK m a

821+ -> K.StreamK m b

822822concatMapDivK useTail useHead stream =

823823 K.mkStream $ \st yld sng stp -> do

824824 let foldShared = K.foldStreamShared st yld sng stp

src/Streamly/Internal/Data/Stream/Concurrent.hs+14−14

@@ -189,9 +189,9 @@ parEval modifier input = withChannel modifier input (const id)

189189_appendGeneric :: MonadAsync m =>

190190 ((Config -> Config) -> m (Channel m a))

191191 -> (Config -> Config)

192− -> K.Stream m a

193− -> K.Stream m a

194− -> K.Stream m a

192+ -> K.StreamK m a

193+ -> K.StreamK m a

194+ -> K.StreamK m a

195195_appendGeneric newChan modifier stream1 stream2 = K.concatEffect action

196196

197197 where

@@ -216,7 +216,7 @@ _appendGeneric newChan modifier stream1 stream2 = K.concatEffect action

216216-- The output stream is the result of the evaluation.

217217{-# INLINE appendWithK #-}

218218appendWithK :: MonadAsync m =>

219− (Config -> Config) -> K.Stream m a -> K.Stream m a -> K.Stream m a

219+ (Config -> Config) -> K.StreamK m a -> K.StreamK m a -> K.StreamK m a

220220appendWithK modifier stream1 stream2 =

221221{-

222222 if getOrdered (modifier defaultConfig)

@@ -236,7 +236,7 @@ appendWithK modifier stream1 stream2 =

236236--

237237{-# INLINE _appendWithChanK #-}

238238_appendWithChanK :: MonadAsync m =>

239− Channel m a -> K.Stream m a -> K.Stream m a -> K.Stream m a

239+ Channel m a -> K.StreamK m a -> K.StreamK m a -> K.StreamK m a

240240_appendWithChanK chan stream1 stream2 =

241241 K.before (toChannelK chan stream2) stream1

242242

@@ -284,8 +284,8 @@ parTwo modifier stream1 stream2 =

284284{-# INLINE mkEnqueue #-}

285285mkEnqueue :: MonadAsync m =>

286286 Channel m b

287− -> ((K.Stream m a -> m ()) -> K.Stream m a -> K.Stream m b)

288− -> m (K.Stream m a -> m ())

287+ -> ((K.StreamK m a -> m ()) -> K.StreamK m a -> K.StreamK m b)

288+ -> m (K.StreamK m a -> m ())

289289mkEnqueue chan runner = do

290290 runInIO <- askRunInIO

291291 return

@@ -308,7 +308,7 @@ mkEnqueue chan runner = do

308308-- concurrently map and evaluate a stream.

309309{-# INLINE parConcatMapChanK #-}

310310parConcatMapChanK :: MonadAsync m =>

311− Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b

311+ Channel m b -> (a -> K.StreamK m b) -> K.StreamK m a -> K.StreamK m b

312312parConcatMapChanK chan f stream =

313313 let run q = concatMapDivK q f

314314 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)

@@ -316,7 +316,7 @@ parConcatMapChanK chan f stream =

316316

317317{-# INLINE parConcatMapChanKAny #-}

318318parConcatMapChanKAny :: MonadAsync m =>

319− Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b

319+ Channel m b -> (a -> K.StreamK m b) -> K.StreamK m a -> K.StreamK m b

320320parConcatMapChanKAny chan f stream =

321321 let done = K.nilM (stopChannel chan)

322322 run q = concatMapDivK q (\x -> K.append (f x) done)

@@ -324,7 +324,7 @@ parConcatMapChanKAny chan f stream =

324324

325325{-# INLINE parConcatMapChanKFirst #-}

326326parConcatMapChanKFirst :: MonadAsync m =>

327− Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b

327+ Channel m b -> (a -> K.StreamK m b) -> K.StreamK m a -> K.StreamK m b

328328parConcatMapChanKFirst chan f stream =

329329 let done = K.nilM (stopChannel chan)

330330 run q = concatMapDivK q f

@@ -341,9 +341,9 @@ parConcatMapChanKFirst chan f stream =

341341parConcatMapChanKGeneric :: MonadAsync m =>

342342 (Config -> Config)

343343 -> Channel m b

344− -> (a -> K.Stream m b)

345− -> K.Stream m a

346− -> K.Stream m b

344+ -> (a -> K.StreamK m b)

345+ -> K.StreamK m a

346+ -> K.StreamK m b

347347parConcatMapChanKGeneric modifier chan f stream = do

348348 let cfg = modifier defaultConfig

349349 case getStopWhen cfg of

@@ -359,7 +359,7 @@ parConcatMapChanKGeneric modifier chan f stream = do

359359--

360360{-# INLINE parConcatMapK #-}

361361parConcatMapK :: MonadAsync m =>

362− (Config -> Config) -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b

362+ (Config -> Config) -> (a -> K.StreamK m b) -> K.StreamK m a -> K.StreamK m b

363363parConcatMapK modifier f input =

364364 let g = parConcatMapChanKGeneric modifier

365365 in withChannelK modifier input (`g` f)

src/Streamly/Internal/Data/Stream/Concurrent/Channel.hs+3−3

@@ -81,9 +81,9 @@ newChannel modifier =

8181{-# INLINE withChannelK #-}

8282withChannelK :: MonadAsync m =>

8383 (Config -> Config)

84− -> K.Stream m a

85− -> (Channel m b -> K.Stream m a -> K.Stream m b)

86− -> K.Stream m b

84+ -> K.StreamK m a

85+ -> (Channel m b -> K.StreamK m a -> K.StreamK m b)

86+ -> K.StreamK m b

8787withChannelK modifier input evaluator = K.concatEffect action

8888

8989 where

src/Streamly/Internal/Data/Stream/Concurrent/Channel/Append.hs+26−26

@@ -55,9 +55,9 @@ import Streamly.Internal.Data.Stream.Channel.Worker

5555{-# INLINE enqueueLIFO #-}

5656enqueueLIFO ::

5757 Channel m a

58− -> IORef ([(RunInIO m, K.Stream m a)], [(RunInIO m, K.Stream m a)])

58+ -> IORef ([(RunInIO m, K.StreamK m a)], [(RunInIO m, K.StreamK m a)])

5959 -> Bool

60− -> (RunInIO m, K.Stream m a)

60+ -> (RunInIO m, K.StreamK m a)

6161 -> IO ()

6262enqueueLIFO sv q inner m = do

6363 atomicModifyIORefCAS_ q $ \(xs, ys) ->

@@ -68,8 +68,8 @@ data QResult a = QEmpty | QOuter a | QInner a

6868

6969{-# INLINE dequeue #-}

7070dequeue :: MonadIO m =>

71− IORef ([(RunInIO m, K.Stream m a)], [(RunInIO m, K.Stream m a)])

72− -> m (QResult (RunInIO m, K.Stream m a))

71+ IORef ([(RunInIO m, K.StreamK m a)], [(RunInIO m, K.StreamK m a)])

72+ -> m (QResult (RunInIO m, K.StreamK m a))

7373dequeue qref =

7474 liftIO

7575 $ atomicModifyIORefCAS qref

@@ -83,7 +83,7 @@ data WorkerStatus = Continue | Suspend

8383{-# INLINE workLoopLIFO #-}

8484workLoopLIFO

8585 :: MonadRunInIO m

86− => IORef ([(RunInIO m, K.Stream m a)], [(RunInIO m, K.Stream m a)])

86+ => IORef ([(RunInIO m, K.StreamK m a)], [(RunInIO m, K.StreamK m a)])

8787 -> Channel m a

8888 -> Maybe WorkerInfo

8989 -> m ()

@@ -144,7 +144,7 @@ workLoopLIFO qref sv winfo = run

144144{-# INLINE workLoopLIFOLimited #-}

145145workLoopLIFOLimited

146146 :: forall m a. MonadRunInIO m

147− => IORef ([(RunInIO m, K.Stream m a)], [(RunInIO m, K.Stream m a)])

147+ => IORef ([(RunInIO m, K.StreamK m a)], [(RunInIO m, K.StreamK m a)])

148148 -> Channel m a

149149 -> Maybe WorkerInfo

150150 -> m ()

@@ -234,8 +234,8 @@ workLoopLIFOLimited qref sv winfo = run

234234{-# INLINE enqueueAhead #-}

235235enqueueAhead ::

236236 Channel m a

237− -> IORef ([K.Stream m a], Int)

238− -> (RunInIO m, K.Stream m a)

237+ -> IORef ([K.StreamK m a], Int)

238+ -> (RunInIO m, K.StreamK m a)

239239 -> IO ()

240240enqueueAhead sv q m = do

241241 -- XXX The queue is LIFO. When parConcatIterate queues more than one items

@@ -444,7 +444,7 @@ updateHeapSeq hpVar seqNo =

444444{-# INLINE underMaxHeap #-}

445445underMaxHeap ::

446446 Channel m a

447− -> Heap (Entry Int (AheadHeapEntry K.Stream m a))

447+ -> Heap (Entry Int (AheadHeapEntry K.StreamK m a))

448448 -> IO Bool

449449underMaxHeap sv hp = do

450450 (_, len) <- readIORef (outputQueue sv)

@@ -466,7 +466,7 @@ underMaxHeap sv hp = do

466466-- False => continue

467467preStopCheck ::

468468 Channel m a

469− -> IORef (Heap (Entry Int (AheadHeapEntry K.Stream m a)) , Maybe Int)

469+ -> IORef (Heap (Entry Int (AheadHeapEntry K.StreamK m a)) , Maybe Int)

470470 -> IO Bool

471471preStopCheck sv heap =

472472 -- check the stop condition under a lock before actually

@@ -518,11 +518,11 @@ abortExecution sv winfo = do

518518--

519519processHeap

520520 :: MonadRunInIO m

521− => IORef ([K.Stream m a], Int)

522− -> IORef (Heap (Entry Int (AheadHeapEntry K.Stream m a)), Maybe Int)

521+ => IORef ([K.StreamK m a], Int)

522+ -> IORef (Heap (Entry Int (AheadHeapEntry K.StreamK m a)), Maybe Int)

523523 -> Channel m a

524524 -> Maybe WorkerInfo

525− -> AheadHeapEntry K.Stream m a

525+ -> AheadHeapEntry K.StreamK m a

526526 -> Int

527527 -> Bool -- we are draining the heap before we stop

528528 -> m ()

@@ -629,8 +629,8 @@ processHeap q heap sv winfo entry sno stopping = loopHeap sno entry

629629{-# NOINLINE drainHeap #-}

630630drainHeap

631631 :: MonadRunInIO m

632− => IORef ([K.Stream m a], Int)

633− -> IORef (Heap (Entry Int (AheadHeapEntry K.Stream m a)), Maybe Int)

632+ => IORef ([K.StreamK m a], Int)

633+ -> IORef (Heap (Entry Int (AheadHeapEntry K.StreamK m a)), Maybe Int)

634634 -> Channel m a

635635 -> Maybe WorkerInfo

636636 -> m ()

@@ -645,11 +645,11 @@ data HeapStatus = HContinue | HStop

645645

646646processWithoutToken

647647 :: MonadRunInIO m

648− => IORef ([K.Stream m a], Int)

649− -> IORef (Heap (Entry Int (AheadHeapEntry K.Stream m a)), Maybe Int)

648+ => IORef ([K.StreamK m a], Int)

649+ -> IORef (Heap (Entry Int (AheadHeapEntry K.StreamK m a)), Maybe Int)

650650 -> Channel m a

651651 -> Maybe WorkerInfo

652− -> K.Stream m a

652+ -> K.StreamK m a

653653 -> Int

654654 -> m ()

655655processWithoutToken q heap sv winfo m seqNo = do

@@ -725,11 +725,11 @@ data TokenWorkerStatus = TokenContinue Int | TokenSuspend

725725

726726processWithToken

727727 :: MonadRunInIO m

728− => IORef ([K.Stream m a], Int)

729− -> IORef (Heap (Entry Int (AheadHeapEntry K.Stream m a)), Maybe Int)

728+ => IORef ([K.StreamK m a], Int)

729+ -> IORef (Heap (Entry Int (AheadHeapEntry K.StreamK m a)), Maybe Int)

730730 -> Channel m a

731731 -> Maybe WorkerInfo

732− -> K.Stream m a

732+ -> K.StreamK m a

733733 -> Int

734734 -> m ()

735735processWithToken q heap sv winfo action sno = do

@@ -834,8 +834,8 @@ processWithToken q heap sv winfo action sno = do

834834

835835workLoopAhead

836836 :: MonadRunInIO m

837− => IORef ([K.Stream m a], Int)

838− -> IORef (Heap (Entry Int (AheadHeapEntry K.Stream m a)), Maybe Int)

837+ => IORef ([K.StreamK m a], Int)

838+ -> IORef (Heap (Entry Int (AheadHeapEntry K.StreamK m a)), Maybe Int)

839839 -> Channel m a

840840 -> Maybe WorkerInfo

841841 -> m ()

@@ -897,8 +897,8 @@ getLifoSVar mrun cfg = do

897897 wfw <- newIORef False

898898 running <- newIORef Set.empty

899899 q <- newIORef

900− ( [] :: [(RunInIO m, K.Stream m a)]

901− , [] :: [(RunInIO m, K.Stream m a)]

900+ ( [] :: [(RunInIO m, K.StreamK m a)]

901+ , [] :: [(RunInIO m, K.StreamK m a)]

902902 )

903903 -- Sequence number is incremented whenever something is de-queued,

904904 -- therefore, first sequence number would be 0

@@ -936,7 +936,7 @@ getLifoSVar mrun cfg = do

936936 -> (Channel m a -> m [ChildEvent a])

937937 -> (Channel m a -> m Bool)

938938 -> (Channel m a -> IO Bool)

939− -> (IORef ([(RunInIO m, K.Stream m a)], [(RunInIO m, K.Stream m a)])

939+ -> (IORef ([(RunInIO m, K.StreamK m a)], [(RunInIO m, K.StreamK m a)])

940940 -> Channel m a

941941 -> Maybe WorkerInfo

942942 -> m())

src/Streamly/Internal/Data/Stream/Concurrent/Channel/Interleave.hs+5−5

@@ -44,8 +44,8 @@ data WorkerStatus = Continue | Suspend

4444{-# INLINE enqueueFIFO #-}

4545enqueueFIFO ::

4646 Channel m a

47− -> LinkedQueue (RunInIO m, K.Stream m a)

48− -> (RunInIO m, K.Stream m a)

47+ -> LinkedQueue (RunInIO m, K.StreamK m a)

48+ -> (RunInIO m, K.StreamK m a)

4949 -> IO ()

5050enqueueFIFO sv q m = do

5151 pushL q m

@@ -54,7 +54,7 @@ enqueueFIFO sv q m = do

5454{-# INLINE workLoopFIFO #-}

5555workLoopFIFO

5656 :: MonadRunInIO m

57− => LinkedQueue (RunInIO m, K.Stream m a)

57+ => LinkedQueue (RunInIO m, K.StreamK m a)

5858 -> Channel m a

5959 -> Maybe WorkerInfo

6060 -> m ()

@@ -93,7 +93,7 @@ workLoopFIFO q sv winfo = run

9393{-# INLINE workLoopFIFOLimited #-}

9494workLoopFIFOLimited

9595 :: forall m a. MonadRunInIO m

96− => LinkedQueue (RunInIO m, K.Stream m a)

96+ => LinkedQueue (RunInIO m, K.StreamK m a)

9797 -> Channel m a

9898 -> Maybe WorkerInfo

9999 -> m ()

@@ -180,7 +180,7 @@ getFifoSVar mrun cfg = do

180180 -> (Channel m a -> m [ChildEvent a])

181181 -> (Channel m a -> m Bool)

182182 -> (Channel m a -> IO Bool)

183− -> (LinkedQueue (RunInIO m, K.Stream m a)

183+ -> (LinkedQueue (RunInIO m, K.StreamK m a)

184184 -> Channel m a

185185 -> Maybe WorkerInfo

186186 -> m())

src/Streamly/Internal/Data/Stream/Concurrent/Channel/Operations.hs+3−3

@@ -89,7 +89,7 @@ import Test.Inspection (inspect, hasNoTypeClassesExcept)

8989-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then

9090-- be read back from the SVar using 'fromSVar'.

9191{-# INLINE toChannelK #-}

92−toChannelK :: MonadRunInIO m => Channel m a -> K.Stream m a -> m ()

92+toChannelK :: MonadRunInIO m => Channel m a -> K.StreamK m a -> m ()

9393toChannelK sv m = do

9494 runIn <- askRunInIO

9595 liftIO $ enqueue sv False (runIn, m)

@@ -114,7 +114,7 @@ joinChannel = undefined

114114

115115-- | Pull a stream from an SVar.

116116{-# NOINLINE fromChannelRaw #-}

117−fromChannelRaw :: (MonadIO m, MonadThrow m) => Channel m a -> K.Stream m a

117+fromChannelRaw :: (MonadIO m, MonadThrow m) => Channel m a -> K.StreamK m a

118118fromChannelRaw sv = K.MkStream $ \st yld sng stp -> do

119119 list <- readOutputQ sv

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

@@ -185,7 +185,7 @@ inspect $ hasNoTypeClassesExcept 'fromChannelRaw

185185-- it to multiple consumers. or should we use an explicit dupChannel for that?

186186

187187{-# INLINE fromChannelK #-}

188−fromChannelK :: MonadAsync m => Channel m a -> K.Stream m a

188+fromChannelK :: MonadAsync m => Channel m a -> K.StreamK m a

189189fromChannelK sv =

190190 K.mkStream $ \st yld sng stp -> do

191191 ref <- liftIO $ newIORef ()

src/Streamly/Internal/Data/Stream/Concurrent/Channel/Type.hs+2−2

@@ -28,7 +28,7 @@ import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS_)

2828import Streamly.Internal.Data.Stream.Channel.Dispatcher (dumpSVarStats)

2929import Streamly.Internal.Data.Stream.Channel.Worker

3030 (sendYield, sendStop, sendWithDoorBell)

31−import Streamly.Internal.Data.Stream.StreamK.Type (Stream)

31+import Streamly.Internal.Data.Stream.StreamK.Type (StreamK)

3232

3333import Streamly.Internal.Data.Stream.Channel.Types

3434

@@ -88,7 +88,7 @@ data Channel m a = Channel

8888 , remainingWork :: Maybe (IORef Count)

8989 , yieldRateInfo :: Maybe YieldRateInfo

9090

91− , enqueue :: Bool -> (RunInIO m, Stream m a) -> IO ()

91+ , enqueue :: Bool -> (RunInIO m, StreamK m a) -> IO ()

9292 , eagerDispatch :: m ()

9393 , isWorkDone :: IO Bool

9494 , isQueueDone :: IO Bool

src/Streamly/Internal/Data/Stream/IsStream/Type.hs+16−16

@@ -15,7 +15,7 @@ module Streamly.Internal.Data.Stream.IsStream.Type {-# DEPRECATED "Please use \"

1515 (

1616 -- * IsStream Type Class

1717 IsStream (..)

18− , K.Stream (..)

18+ , K.StreamK (..)

1919

2020 -- * Type Conversion

2121 , fromStreamD

@@ -124,7 +124,7 @@ import qualified Streamly.Internal.Data.Stream.StreamD.Type as D

124124 , drain, eqBy, cmpBy, fromList, toList, foldrMx, foldlMx'

125125 , foldlx', foldl', fold)

126126import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

127− (Stream(..), cons, fromEffect

127+ (StreamK(..), cons, fromEffect

128128 , nil, fromPure, bindWith, drain

129129 , fromFoldable, nilM, repeat)

130130import qualified Streamly.Internal.Data.Stream.StreamK as StreamK

@@ -160,8 +160,8 @@ class

160160 , forall m. MonadAsync m => Applicative (t m)

161161 ) =>

162162 IsStream t where

163− toStream :: t m a -> K.Stream m a

164− fromStream :: K.Stream m a -> t m a

163+ toStream :: t m a -> K.StreamK m a

164+ fromStream :: K.StreamK m a -> t m a

165165 -- | Constructs a stream by adding a monadic action at the head of an

166166 -- existing stream. For example:

167167 --

@@ -231,7 +231,7 @@ fromStreamD = fromStream . D.toStreamK

231231-- | Adapt a polymorphic consM operation to a StreamK cons operation

232232{-# INLINE toConsK #-}

233233toConsK :: IsStream t =>

234− (m a -> t m a -> t m a) -> m a -> K.Stream m a -> K.Stream m a

234+ (m a -> t m a -> t m a) -> m a -> K.StreamK m a -> K.StreamK m a

235235toConsK cns x xs = toStream $ x `cns` fromStream xs

236236

237237------------------------------------------------------------------------------

@@ -317,7 +317,7 @@ toList m = D.toList $ toStreamD m

317317-- continuation and a yield continuation.

318318{-# INLINE_EARLY mkStream #-}

319319mkStream :: IsStream t

320− => (forall r. State K.Stream m a

320+ => (forall r. State K.StreamK m a

321321 -> (a -> t m a -> m r)

322322 -> (a -> m r)

323323 -> m r

@@ -329,8 +329,8 @@ mkStream k = fromStream $ K.MkStream $ \st yld sng stp ->

329329

330330{-# RULES "mkStream from stream" mkStream = mkStreamFromStream #-}

331331mkStreamFromStream :: IsStream t

332− => (forall r. State K.Stream m a

333− -> (a -> K.Stream m a -> m r)

332+ => (forall r. State K.StreamK m a

333+ -> (a -> K.StreamK m a -> m r)

334334 -> (a -> m r)

335335 -> m r

336336 -> m r)

@@ -339,12 +339,12 @@ mkStreamFromStream k = fromStream $ K.MkStream k

339339

340340{-# RULES "mkStream stream" mkStream = mkStreamStream #-}

341341mkStreamStream

342− :: (forall r. State K.Stream m a

343− -> (a -> K.Stream m a -> m r)

342+ :: (forall r. State K.StreamK m a

343+ -> (a -> K.StreamK m a -> m r)

344344 -> (a -> m r)

345345 -> m r

346346 -> m r)

347− -> K.Stream m a

347+ -> K.StreamK m a

348348mkStreamStream = K.MkStream

349349

350350------------------------------------------------------------------------------

@@ -399,7 +399,7 @@ fold fld m = D.fold fld $ toStreamD m

399399{-# INLINE_EARLY foldStreamShared #-}

400400foldStreamShared

401401 :: IsStream t

402− => State K.Stream m a

402+ => State K.StreamK m a

403403 -> (a -> t m a -> m r)

404404 -> (a -> m r)

405405 -> m r

@@ -414,11 +414,11 @@ foldStreamShared st yld sng stp m =

414414{-# RULES "foldStreamShared from stream"

415415 foldStreamShared = foldStreamSharedStream #-}

416416foldStreamSharedStream

417− :: State K.Stream m a

418− -> (a -> K.Stream m a -> m r)

417+ :: State K.StreamK m a

418+ -> (a -> K.StreamK m a -> m r)

419419 -> (a -> m r)

420420 -> m r

421− -> K.Stream m a

421+ -> K.StreamK m a

422422 -> m r

423423foldStreamSharedStream st yld sng stp m =

424424 let K.MkStream k = m

@@ -430,7 +430,7 @@ foldStreamSharedStream st yld sng stp m =

430430{-# INLINE foldStream #-}

431431foldStream

432432 :: IsStream t

433− => State K.Stream m a

433+ => State K.StreamK m a

434434 -> (a -> t m a -> m r)

435435 -> (a -> m r)

436436 -> m r

src/Streamly/Internal/Data/Stream/Parallel.hs+16−16

@@ -65,7 +65,7 @@ import Streamly.Internal.Data.Stream.StreamD.Type (Step(..))

6565

6666import qualified Data.Set as Set

6767import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

68− (Stream, foldStreamShared, mkStream, foldStream, fromEffect

68+ (StreamK, foldStreamShared, mkStream, foldStream, fromEffect

6969 , nil, concatMapWith, fromPure, bindWith)

7070import qualified Streamly.Internal.Data.Stream.StreamD.Type as D

7171 (Stream(..), mapM, toStreamK, fromStreamK)

@@ -91,7 +91,7 @@ import Prelude hiding (map)

9191-- :}

9292

9393{-# INLINABLE withLocal #-}

94−withLocal :: MonadReader r m => (r -> r) -> K.Stream m a -> K.Stream m a

94+withLocal :: MonadReader r m => (r -> r) -> K.StreamK m a -> K.StreamK m a

9595withLocal f m =

9696 K.mkStream $ \st yld sng stp ->

9797 let single = local f . sng

@@ -109,7 +109,7 @@ withLocal f m =

109109{-# NOINLINE runOne #-}

110110runOne

111111 :: MonadIO m

112− => State K.Stream m a -> K.Stream m a -> Maybe WorkerInfo -> m ()

112+ => State K.StreamK m a -> K.StreamK m a -> Maybe WorkerInfo -> m ()

113113runOne st m0 winfo =

114114 case getYieldLimit st of

115115 Nothing -> go m0

@@ -132,7 +132,7 @@ runOne st m0 winfo =

132132

133133runOneLimited

134134 :: MonadIO m

135− => State K.Stream m a -> K.Stream m a -> Maybe WorkerInfo -> m ()

135+ => State K.StreamK m a -> K.StreamK m a -> Maybe WorkerInfo -> m ()

136136runOneLimited st m0 winfo = go m0

137137

138138 where

@@ -169,7 +169,7 @@ runOneLimited st m0 winfo = go m0

169169

170170{-# NOINLINE forkSVarPar #-}

171171forkSVarPar :: MonadAsync m

172− => SVarStopStyle -> K.Stream m a -> K.Stream m a -> K.Stream m a

172+ => SVarStopStyle -> K.StreamK m a -> K.StreamK m a -> K.StreamK m a

173173forkSVarPar ss m r = K.mkStream $ \st yld sng stp -> do

174174 sv <- newParallelVar ss st

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

@@ -186,9 +186,9 @@ joinStreamVarPar ::

186186 MonadAsync m

187187 => SVarStyle

188188 -> SVarStopStyle

189− -> K.Stream m a

190− -> K.Stream m a

191− -> K.Stream m a

189+ -> K.StreamK m a

190+ -> K.StreamK m a

191+ -> K.StreamK m a

192192joinStreamVarPar style ss m1 m2 = K.mkStream $ \st yld sng stp ->

193193 case streamVar st of

194194 Just sv | svarStyle sv == style && svarStopStyle sv == ss -> do

@@ -238,7 +238,7 @@ joinStreamVarPar style ss m1 m2 = K.mkStream $ \st yld sng stp ->

238238-------------------------------------------------------------------------------

239239

240240{-# INLINE parallelK #-}

241−parallelK :: MonadAsync m => K.Stream m a -> K.Stream m a -> K.Stream m a

241+parallelK :: MonadAsync m => K.StreamK m a -> K.StreamK m a -> K.StreamK m a

242242parallelK = joinStreamVarPar ParallelVar StopNone

243243

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

@@ -256,7 +256,7 @@ consM m (ParallelT r) = ParallelT $ parallelK (K.fromEffect m) r

256256--

257257-- /Pre-release/

258258{-# INLINE parallelFstK #-}

259−parallelFstK :: MonadAsync m => K.Stream m a -> K.Stream m a -> K.Stream m a

259+parallelFstK :: MonadAsync m => K.StreamK m a -> K.StreamK m a -> K.StreamK m a

260260parallelFstK = joinStreamVarPar ParallelVar StopBy

261261

262262-- This is a race like combinator for streams.

@@ -266,7 +266,7 @@ parallelFstK = joinStreamVarPar ParallelVar StopBy

266266--

267267-- /Pre-release/

268268{-# INLINE parallelMinK #-}

269−parallelMinK :: MonadAsync m => K.Stream m a -> K.Stream m a -> K.Stream m a

269+parallelMinK :: MonadAsync m => K.StreamK m a -> K.StreamK m a -> K.StreamK m a

270270parallelMinK = joinStreamVarPar ParallelVar StopAny

271271

272272------------------------------------------------------------------------------

@@ -277,7 +277,7 @@ parallelMinK = joinStreamVarPar ParallelVar StopAny

277277--

278278-- /Pre-release/

279279--

280−mkParallelK :: MonadAsync m => K.Stream m a -> K.Stream m a

280+mkParallelK :: MonadAsync m => K.StreamK m a -> K.StreamK m a

281281mkParallelK m = K.mkStream $ \st yld sng stp -> do

282282 sv <- newParallelVar StopNone (adaptState st)

283283 -- pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream m)

@@ -327,7 +327,7 @@ mkParallelD m = D.Stream step Nothing

327327-- is determined by the prevailing 'Streamly.Prelude.maxBuffer' setting.

328328--

329329-- @

330−-- Stream m a -> m b

330+-- StreamK m a -> m b

331331-- |

332332-- -----stream m a ---------------stream m a-----

333333--

@@ -351,7 +351,7 @@ mkParallelD m = D.Stream step Nothing

351351-- /Pre-release/

352352{-# INLINE tapAsyncK #-}

353353tapAsyncK ::

354− MonadAsync m => (K.Stream m a -> m b) -> K.Stream m a -> K.Stream m a

354+ MonadAsync m => (K.StreamK m a -> m b) -> K.StreamK m a -> K.StreamK m a

355355tapAsyncK f m = K.mkStream $ \st yld sng stp -> do

356356 sv <- SVar.newFoldSVar st (f . Stream.toStreamK)

357357 K.foldStreamShared st yld sng stp

@@ -429,7 +429,7 @@ tapAsyncF f (D.Stream step1 state1) = D.Stream step TapInit

429429-- /Since: 0.7.0 (maxBuffer applies to ParallelT streams)/

430430--

431431-- @since 0.8.0

432−newtype ParallelT m a = ParallelT {getParallelT :: K.Stream m a}

432+newtype ParallelT m a = ParallelT {getParallelT :: K.StreamK m a}

433433

434434instance MonadTrans ParallelT where

435435 {-# INLINE lift #-}

@@ -519,7 +519,7 @@ MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)

519519-- /Pre-release/

520520--

521521{-# INLINE_NORMAL newCallbackStream #-}

522−newCallbackStream :: MonadAsync m => m (a -> m (), K.Stream m a)

522+newCallbackStream :: MonadAsync m => m (a -> m (), K.StreamK m a)

523523newCallbackStream = do

524524 sv <- newParallelVar StopNone defState

525525

src/Streamly/Internal/Data/Stream/SVar/Eliminate.hs+1−1

@@ -43,7 +43,7 @@ import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)

4343import qualified Streamly.Internal.Data.Stream.StreamD.Type as D

4444 (Stream(..), Step(..), fold)

4545import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

46− (Stream(..), mkStream, foldStream, foldStreamShared, nilM)

46+ (Stream, mkStream, foldStream, foldStreamShared, nilM)

4747import qualified Streamly.Internal.Data.Stream.Serial as Stream

4848 (fromStreamK, toStreamK)

4949

src/Streamly/Internal/Data/Stream/SVar/Generate.hs+4−4

@@ -46,7 +46,7 @@ import System.Mem (performMajorGC)

4646import qualified Streamly.Internal.Data.Stream.StreamD.Type as D

4747 (Stream(..), Step(..))

4848import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

49− (Stream(..), foldStreamShared, mkStream, foldStream)

49+ (Stream, foldStreamShared, mkStream, foldStream)

5050import qualified Streamly.Internal.Data.Stream.Serial as Stream (fromStreamK)

5151

5252import Streamly.Internal.Data.SVar

@@ -113,7 +113,7 @@ toSVar sv m = do

113113-- | Pull a stream from an SVar.

114114{-# NOINLINE fromStreamVar #-}

115115fromStreamVar :: MonadAsync m => SVar K.Stream m a -> K.Stream m a

116−fromStreamVar sv = K.MkStream $ \st yld sng stp -> do

116+fromStreamVar sv = K.mkStream $ \st yld sng stp -> do

117117 list <- readOutputQ sv

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

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

@@ -130,13 +130,13 @@ fromStreamVar sv = K.MkStream $ \st yld sng stp -> do

130130 stp

131131

132132 {-# INLINE processEvents #-}

133− processEvents [] = K.MkStream $ \st yld sng stp -> do

133+ processEvents [] = K.mkStream $ \st yld sng stp -> do

134134 done <- postProcess sv

135135 if done

136136 then allDone stp

137137 else K.foldStream st yld sng stp $ fromStreamVar sv

138138

139− processEvents (ev : es) = K.MkStream $ \st yld sng stp -> do

139+ processEvents (ev : es) = K.mkStream $ \st yld sng stp -> do

140140 let rest = processEvents es

141141 case ev of

142142 ChildYield a -> yld a rest

src/Streamly/Internal/Data/Stream/Serial.hs+1−1

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

11{-# LANGUAGE UndecidableInstances #-}

2−{-# OPTIONS_GHC -Wno-orphans #-}

2+{-# OPTIONS_GHC -Wno-deprecations #-}

44-- |

55-- Module : Streamly.Internal.Data.Stream.Serial

src/inline.hs+1−1