3ae9530 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

use folds and map from direct style stream

3ae9530Harendra Kumar8 years ago

Changelog.md+1

compare with master

unifiedsplit

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

10* Add `repeat` to generate an infinite stream by repeating a pure value

11* Add `fromList` and `fromListM` to generate streams from lists, faster than

12 `fromFoldable` and `fromFoldableM`

13+* Add `map` as a synonym of fmap

13* Significant performance improvements

14* Significant performance improvements

15## 0.3.0

16## 0.3.0

benchmark/Linear.hs+8−2

@@ -15,6 +15,7 @@ import Gauge

16-- We need a monadic bind here to make sure that the function f does not get

17-- completely optimized out by the compiler in some cases.

18+{-# INLINE benchIO #-}

18benchIO :: (IsStream t, NFData b) => String -> (t IO Int -> IO b) -> Benchmark

19benchIO :: (IsStream t, NFData b) => String -> (t IO Int -> IO b) -> Benchmark

19benchIO name f = bench name $ nfIO $ randomRIO (1,1000) >>= f . Ops.source

20benchIO name f = bench name $ nfIO $ randomRIO (1,1000) >>= f . Ops.source

@@ -50,13 +51,18 @@ main = do

50 , benchSrcIO serially "foldMapWithM" Ops.sourceFoldMapWithM

51 , benchSrcIO serially "foldMapWithM" Ops.sourceFoldMapWithM

51 ]

52 ]

52 , bgroup "elimination"

53 , bgroup "elimination"

54+ [ benchIO "toNull" $ Ops.toNull serially

55+ , benchIO "mapM_" Ops.mapM_

53− [ benchIO "toList" Ops.toList

56+ , benchIO "toList" Ops.toList

57+ , benchIO "foldr" Ops.foldr

58+ , benchIO "foldrM" Ops.foldrM

54− , benchIO "fold" Ops.foldl

59+ , benchIO "foldl'" Ops.foldl

55 , benchIO "last" Ops.last

60 , benchIO "last" Ops.last

56 ]

61 ]

57 , bgroup "transformation"

62 , bgroup "transformation"

58 [ benchIO "scan" Ops.scan

63 [ benchIO "scan" Ops.scan

59 , benchIO "map" Ops.map

64 , benchIO "map" Ops.map

65+ , benchIO "fmap" Ops.fmap

60 , benchIO "mapM" (Ops.mapM serially)

66 , benchIO "mapM" (Ops.mapM serially)

61 , benchIO "concat" Ops.concat

67 , benchIO "concat" Ops.concat

62 ]

68 ]

benchmark/LinearOps.hs+15−4

@@ -26,6 +26,7 @@ maxValue = value + 1000

27{-# INLINE scan #-}

28{-# INLINE map #-}

29+{-# INLINE fmap #-}

29{-# INLINE filterEven #-}

30{-# INLINE filterEven #-}

30{-# INLINE filterAllOut #-}

31{-# INLINE filterAllOut #-}

31{-# INLINE filterAllIn #-}

32{-# INLINE filterAllIn #-}

@@ -39,7 +40,7 @@ maxValue = value + 1000

39{-# INLINE composeAllInFilters #-}

40{-# INLINE composeAllInFilters #-}

40{-# INLINE composeAllOutFilters #-}

41{-# INLINE composeAllOutFilters #-}

41{-# INLINE composeMapAllInFilter #-}

42{-# INLINE composeMapAllInFilter #-}

42−scan, map, filterEven, filterAllOut,

43+scan, map, fmap, filterEven, filterAllOut,

43 filterAllIn, takeOne, takeAll, takeWhileTrue, dropAll, dropWhileTrue, zip,

44 filterAllIn, takeOne, takeAll, takeWhileTrue, dropAll, dropWhileTrue, zip,

44 concat, composeAllInFilters, composeAllOutFilters,

45 concat, composeAllInFilters, composeAllOutFilters,

45 composeMapAllInFilter

46 composeMapAllInFilter

@@ -50,13 +51,17 @@ scan, map, filterEven, filterAllOut,

50composeMapM :: S.MonadAsync m => Stream m Int -> m ()

51composeMapM :: S.MonadAsync m => Stream m Int -> m ()

52{-# INLINE toList #-}

53{-# INLINE toList #-}

54+{-# INLINE foldr #-}

55+{-# INLINE foldrM #-}

53−toList :: Monad m => Stream m Int -> m [Int]

56+toList, foldr, foldrM :: Monad m => Stream m Int -> m [Int]

54{-# INLINE foldl #-}

57{-# INLINE foldl #-}

55foldl :: Monad m => Stream m Int -> m Int

58foldl :: Monad m => Stream m Int -> m Int

56{-# INLINE last #-}

59{-# INLINE last #-}

57last :: Monad m => Stream m Int -> m (Maybe Int)

60last :: Monad m => Stream m Int -> m (Maybe Int)

58{-# INLINE toNull #-}

61{-# INLINE toNull #-}

59toNull :: Monad m => (t m Int -> S.SerialT m Int) -> t m Int -> m ()

62toNull :: Monad m => (t m Int -> S.SerialT m Int) -> t m Int -> m ()

63+{-# INLINE mapM_ #-}

64+mapM_ :: Monad m => Stream m a -> m ()

60{-# INLINE mapM #-}

65{-# INLINE mapM #-}

61mapM :: (S.IsStream t, S.MonadAsync m)

66mapM :: (S.IsStream t, S.MonadAsync m)

62 => (t m Int -> S.SerialT m Int) -> t m Int -> m ()

67 => (t m Int -> S.SerialT m Int) -> t m Int -> m ()

@@ -72,6 +77,7 @@ type Stream m a = S.SerialT m a

72{-# INLINE source #-}

77{-# INLINE source #-}

73source :: (S.MonadAsync m, S.IsStream t) => Int -> t m Int

78source :: (S.MonadAsync m, S.IsStream t) => Int -> t m Int

74source n = S.serially $ sourceUnfoldrM n

79source n = S.serially $ sourceUnfoldrM n

80+-- source n = S.serially $ sourceFromList n

76{-# INLINE sourceFromList #-}

82{-# INLINE sourceFromList #-}

77sourceFromList :: (Monad m, S.IsStream t) => Int -> t m Int

83sourceFromList :: (Monad m, S.IsStream t) => Int -> t m Int

@@ -126,7 +132,10 @@ runStream = S.runStream

126-------------------------------------------------------------------------------

132-------------------------------------------------------------------------------

127

133

128toNull t = runStream . t

134toNull t = runStream . t

135+mapM_ = S.mapM_ (\_ -> return ())

129toList = S.toList

136toList = S.toList

137+foldr = S.foldr (:) []

138+foldrM = S.foldrM (\a xs -> return (a : xs)) []

130foldl = S.foldl' (+) 0

139foldl = S.foldl' (+) 0

131last = S.last

140last = S.last

132

141

@@ -139,7 +148,8 @@ transform :: Monad m => Stream m a -> m ()

139transform = runStream

148transform = runStream

140

149

141scan = transform . S.scanl' (+) 0

150scan = transform . S.scanl' (+) 0

151+fmap = transform . Prelude.fmap (+1)

142−map = transform . fmap (+1)

152+map = transform . S.map (+1)

143mapM t = transform . t . S.mapM return

153mapM t = transform . t . S.mapM return

144filterEven = transform . S.filter even

154filterEven = transform . S.filter even

145filterAllOut = transform . S.filter (> maxValue)

155filterAllOut = transform . S.filter (> maxValue)

@@ -169,7 +179,8 @@ compose f = transform . f . f . f . f

169composeMapM = compose (S.mapM return)

179composeMapM = compose (S.mapM return)

170composeAllInFilters = compose (S.filter (<= maxValue))

180composeAllInFilters = compose (S.filter (<= maxValue))

171composeAllOutFilters = compose (S.filter (> maxValue))

181composeAllOutFilters = compose (S.filter (> maxValue))

182+composeMapAllInFilter =

172−composeMapAllInFilter = compose (S.filter (<= maxValue) . fmap (subtract 1))

183+ compose (S.filter (<= maxValue) . Prelude.fmap (subtract 1))

173

184

174{-# INLINABLE composeScaling #-}

185{-# INLINABLE composeScaling #-}

175composeScaling :: Monad m => Int -> Stream m Int -> m ()

186composeScaling :: Monad m => Int -> Stream m Int -> m ()

src/Streamly/Prelude.hs+43−80

@@ -47,11 +47,11 @@ module Streamly.Prelude

47 (

48 -- * Construction

49 -- | Primitives to construct or inspect a stream.

50− nil

50+ K.nil

51 , K.yield

52 , K.yieldM

53− , cons

53+ , K.cons

54− , (.:)

54+ , (K..:)

55 , consM

56 , (|:)

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

116 , reverse

117

118 -- * Mapping

119+ , Serial.map

119 , mapM

120 , mapM

120 , mapMaybe

121 , mapMaybe

121 , mapMaybeM

122 , mapMaybeM

@@ -140,25 +141,24 @@ module Streamly.Prelude

140 )

141 )

141where

142where

142

143

143−import Control.Monad (void)

144−import Control.Monad.IO.Class (MonadIO(..))

144+import Control.Monad.IO.Class (MonadIO(..))

145−import Data.Maybe (isJust, fromJust)

145+import Data.Maybe (isJust, fromJust)

146+import Prelude

146−import Prelude hiding (filter, drop, dropWhile, take,

147+ hiding (filter, drop, dropWhile, take, takeWhile, zipWith, foldr,

147− takeWhile, zipWith, foldr, foldl,

148− mapM, mapM_, sequence, all, any,

148+ foldl, map, mapM, mapM_, sequence, all, any, sum, product, elem,

149− sum, product, elem, notElem,

150− maximum, minimum, head, last,

149+ notElem, maximum, minimum, head, last, tail, length, null,

151− tail, length, null, reverse,

152− iterate)

150+ reverse, iterate)

153import qualified Prelude

151import qualified Prelude

154import qualified System.IO as IO

152import qualified System.IO as IO

155

153

154+import Streamly.SVar (MonadAsync)

156−import Streamly.Streams.StreamK (IsStream(..), cons, nil, (.:))

155+import Streamly.Streams.StreamK (IsStream(..))

156+import Streamly.Streams.Serial (SerialT)

157+import Streamly.Streams.Zip (zipWith, zipWithM, zipAsyncWith, zipAsyncWithM)

158+

157import qualified Streamly.Streams.StreamK as K

159import qualified Streamly.Streams.StreamK as K

158import qualified Streamly.Streams.StreamD as D

160import qualified Streamly.Streams.StreamD as D

159−import Streamly.Streams.Zip

160−import Streamly.Streams.Serial

161+import qualified Streamly.Streams.Serial as Serial

161−import Streamly.SVar (MonadAsync)

162

163------------------------------------------------------------------------------

164-- Construction

@@ -260,7 +260,7 @@ fromListM = fromStream . D.toStreamK . D.fromListM

260-- @since 0.3.0

261{-# INLINE fromFoldableM #-}

262fromFoldableM :: (IsStream t, MonadAsync m, Foldable f) => f (m a) -> t m a

263−fromFoldableM = Prelude.foldr consM nil

263+fromFoldableM = Prelude.foldr consM K.nil

264

265-- | Same as 'fromFoldable'.

266--

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

284replicateM :: (IsStream t, MonadAsync m) => Int -> m a -> t m a

285replicateM n m = go n

286 where

287− go cnt = if cnt <= 0 then nil else m |: go (cnt - 1)

287+ go cnt = if cnt <= 0 then K.nil else m |: go (cnt - 1)

288

289-- | Generate a stream by repeatedly executing a monadic action forever.

290--

@@ -360,14 +360,11 @@ fromHandle h = fromStream go

360-- @

361--

362-- @since 0.1.0

363+{-# INLINE foldr #-}

363foldr :: Monad m => (a -> b -> b) -> b -> SerialT m a -> m b

364foldr :: Monad m => (a -> b -> b) -> b -> SerialT m a -> m b

365+-- XXX somehow this definition does not perform well, need to investigate

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

366+-- foldr step acc m = D.foldr step acc $ D.fromStreamK (toStream m)

365− where

367+foldr f = foldrM (\a b -> return (f a b))

366− go m1 =

367− let stop = return acc

368− single a = return (step a acc)

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

370− in (K.unStream m1) Nothing stop single yieldk

371

368

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

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

373-- stream into a list:

370-- stream into a list:

@@ -380,13 +377,7 @@ foldr step acc m = go (toStream m)

380-- @since 0.2.0

377-- @since 0.2.0

381{-# INLINE foldrM #-}

378{-# INLINE foldrM #-}

382foldrM :: Monad m => (a -> b -> m b) -> b -> SerialT m a -> m b

379foldrM :: Monad m => (a -> b -> m b) -> b -> SerialT m a -> m b

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

380+foldrM step acc m = D.foldrM step acc $ D.fromStreamK (toStream m)

384− where

385− go m1 =

386− let stop = return acc

387− single a = step a acc

388− yieldk a r = go r >>= step a

389− in (K.unStream m1) Nothing stop single yieldk

390

381

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

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

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

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

@@ -397,7 +388,7 @@ foldrM step acc m = go (toStream m)

397{-# INLINE scanx #-}

388{-# INLINE scanx #-}

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

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

399scanx step begin done m =

390scanx step begin done m =

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

391+ K.cons (done begin) $ fromStream $ go (toStream m) begin

401 where

392 where

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

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

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

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

@@ -430,24 +421,7 @@ scanl' step begin m = scanx step begin id m

430-- @since 0.2.0

421-- @since 0.2.0

431{-# INLINE foldx #-}

422{-# INLINE foldx #-}

432foldx :: Monad m => (x -> a -> x) -> x -> (x -> b) -> SerialT m a -> m b

423foldx :: Monad m => (x -> a -> x) -> x -> (x -> b) -> SerialT m a -> m b

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

424+foldx = K.foldx

434− where

435− {-# NOINLINE get #-}

436− get m1 =

437− let single = return . done

438− in (K.unStream m1) Nothing undefined single undefined

439−

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

441− -- however that is more expensive because of unnecessary recursion

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

443− -- continuations is much more efficient.

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

445− let stop = sng acc

446− single a = sng $ step acc a

447− yieldk a r =

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

449− in (K.unStream stream) Nothing undefined sng yld

450− in (K.unStream m1) Nothing stop single yieldk

451

425

452-- |

426-- |

453-- @since 0.1.0

427-- @since 0.1.0

@@ -460,20 +434,14 @@ foldl = foldx

460-- @since 0.2.0

434-- @since 0.2.0

461{-# INLINE foldl' #-}

435{-# INLINE foldl' #-}

462foldl' :: Monad m => (b -> a -> b) -> b -> SerialT m a -> m b

436foldl' :: Monad m => (b -> a -> b) -> b -> SerialT m a -> m b

463−foldl' step begin m = foldx step begin id m

437+foldl' step begin m = D.foldl' step begin $ D.fromStreamK (toStream m)

464

438

465-- XXX replace the recursive "go" with explicit continuations.

439-- XXX replace the recursive "go" with explicit continuations.

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

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

467--

441--

468-- @since 0.2.0

442-- @since 0.2.0

469foldxM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> SerialT m a -> m b

443foldxM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> SerialT m a -> m b

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

444+foldxM = K.foldxM

471− where

472− go !acc m1 =

473− let stop = acc >>= done

474− single a = acc >>= \b -> step b a >>= done

475− yieldk a r = acc >>= \b -> go (step b a) r

476− in (K.unStream m1) Nothing stop single yieldk

477

445

478-- |

446-- |

479-- @since 0.1.0

447-- @since 0.1.0

@@ -485,7 +453,7 @@ foldlM = foldxM

485--

453--

486-- @since 0.2.0

454-- @since 0.2.0

487foldlM' :: Monad m => (b -> a -> m b) -> b -> SerialT m a -> m b

455foldlM' :: Monad m => (b -> a -> m b) -> b -> SerialT m a -> m b

488−foldlM' step begin m = foldxM step (return begin) return m

456+foldlM' step begin m = D.foldlM' step begin $ D.fromStreamK (toStream m)

489

457

490-- | Decompose a stream into its head and tail. If the stream is empty, returns

458-- | Decompose a stream into its head and tail. If the stream is empty, returns

491-- 'Nothing'. If the stream is non-empty, returns @Just (a, ma)@, where @a@ is

459-- 'Nothing'. If the stream is non-empty, returns @Just (a, ma)@, where @a@ is

@@ -495,7 +463,7 @@ foldlM' step begin m = foldxM step (return begin) return m

495uncons :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (a, t m a))

463uncons :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (a, t m a))

496uncons m =

464uncons m =

497 let stop = return Nothing

465 let stop = return Nothing

498− single a = return (Just (a, nil))

466+ single a = return (Just (a, K.nil))

499 yieldk a r = return (Just (a, fromStream r))

467 yieldk a r = return (Just (a, fromStream r))

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

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

501

469

@@ -518,9 +486,9 @@ toHandle h m = go (toStream m)

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

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

519--

487--

520-- @since 0.1.0

488-- @since 0.1.0

521−{-# INLINABLE toList #-}

489+{-# INLINE toList #-}

522toList :: Monad m => SerialT m a -> m [a]

490toList :: Monad m => SerialT m a -> m [a]

523−toList = foldrM (\a xs -> return (a : xs)) []

491+toList m = D.toList $ D.fromStreamK (toStream m)

524

492

525-- | Take first 'n' elements from the stream and discard the rest.

493-- | Take first 'n' elements from the stream and discard the rest.

526--

494--

@@ -644,7 +612,7 @@ head m =

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

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

645tail m =

613tail m =

646 let stop = return Nothing

614 let stop = return Nothing

647− single _ = return $ Just nil

615+ single _ = return $ Just K.nil

648 yieldk _ r = return $ Just $ fromStream r

616 yieldk _ r = return $ Just $ fromStream r

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

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

650

618

@@ -653,7 +621,7 @@ tail m =

653-- @since 0.1.1

621-- @since 0.1.1

654{-# INLINE last #-}

622{-# INLINE last #-}

655last :: Monad m => SerialT m a -> m (Maybe a)

623last :: Monad m => SerialT m a -> m (Maybe a)

656−last = foldl (\_ y -> Just y) Nothing id

624+last m = D.last $ D.fromStreamK (toStream m)

657

625

658-- | Determine whether the stream is empty.

626-- | Determine whether the stream is empty.

659--

627--

@@ -761,14 +729,14 @@ maximum m = go Nothing (toStream m)

761-- /Concurrent (do not use with 'parallely' on infinite streams)/

729-- /Concurrent (do not use with 'parallely' on infinite streams)/

762--

730--

763-- @since 0.1.0

731-- @since 0.1.0

764−{-# INLINE mapM #-}

732+{-# INLINE_EARLY mapM #-}

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

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

766−mapM f m = go (toStream m)

734+mapM = K.mapM

767− where

735+

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

736+{-# RULES "mapM serial" mapM = mapMSerial #-}

769− let single a = f a >>= sng

737+{-# INLINE mapMSerial #-}

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

738+mapMSerial :: Monad m => (a -> m b) -> SerialT m a -> SerialT m b

771− in (K.unStream m1) Nothing stp single yieldk

739+mapMSerial = Serial.mapM

772

740

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

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

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

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

@@ -802,14 +770,9 @@ mapMaybeM f = fmap fromJust . filter isJust . mapM f

802-- output of the action.

770-- output of the action.

803--

771--

804-- @since 0.1.0

772-- @since 0.1.0

773+{-# INLINE mapM_ #-}

805mapM_ :: Monad m => (a -> m b) -> SerialT m a -> m ()

774mapM_ :: Monad m => (a -> m b) -> SerialT m a -> m ()

806−mapM_ f m = go (toStream m)

775+mapM_ f m = D.mapM_ f $ D.fromStreamK (toStream m)

807− where

808− go m1 =

809− let stop = return ()

810− single a = void (f a)

811− yieldk a r = f a >> go r

812− in (K.unStream m1) Nothing stop single yieldk

813

776

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

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

815-- actions.

778-- actions.

src/Streamly/Streams/Ahead.hs+9−6

@@ -44,13 +44,16 @@ import Data.Semigroup (Semigroup(..))

44import qualified Data.Heap as H

46import Streamly.Streams.SVar (fromSVar)

47−import Streamly.Streams.StreamK

47+import Streamly.Streams.Serial (map)

48import Streamly.SVar

49+import Streamly.Streams.StreamK (IsStream(..), Stream(..))

50+import qualified Streamly.Streams.StreamK as K

50#ifdef DIAGNOSTICS

52#ifdef DIAGNOSTICS

51import Control.Monad (when)

53import Control.Monad (when)

52import Data.IORef (writeIORef, readIORef)

54import Data.IORef (writeIORef, readIORef)

53#endif

55#endif

56+import Prelude hiding (map)

55#include "Instances.hs"

58#include "Instances.hs"

@@ -128,7 +131,7 @@ workLoopAhead sv q heap = runHeap

128 else liftIO $ sendStop sv

131 else liftIO $ sendStop sv

129

132

130 singleToHeap seqNo a = toHeap seqNo (AheadEntryPure a)

133 singleToHeap seqNo a = toHeap seqNo (AheadEntryPure a)

131− yieldToHeap seqNo a r = toHeap seqNo (AheadEntryStream (a `cons` r))

134+ yieldToHeap seqNo a r = toHeap seqNo (AheadEntryStream (a `K.cons` r))

132

135

133 singleOutput seqNo a = do

136 singleOutput seqNo a = do

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

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

@@ -242,7 +245,7 @@ aheadS m1 m2 = Stream $ \svr stp sng yld -> do

242-- of combining streams using ahead.

245-- of combining streams using ahead.

243{-# INLINE consMAhead #-}

246{-# INLINE consMAhead #-}

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

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

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

248+consMAhead m r = K.yieldM m `aheadS` r

246

249

247------------------------------------------------------------------------------

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

248-- AheadT

251-- AheadT

@@ -293,7 +296,7 @@ consMAhead m r = yieldM m `aheadS` r

293--

296--

294-- @since 0.3.0

297-- @since 0.3.0

295newtype AheadT m a = AheadT {getAheadT :: Stream m a}

298newtype AheadT m a = AheadT {getAheadT :: Stream m a}

296− deriving (Functor, MonadTrans)

299+ deriving (MonadTrans)

297

300

298-- | A serial IO stream of elements of type @a@ with concurrent lookahead. See

301-- | A serial IO stream of elements of type @a@ with concurrent lookahead. See

299-- 'AheadT' documentation for more details.

302-- 'AheadT' documentation for more details.

@@ -305,7 +308,7 @@ type Ahead a = AheadT IO a

305--

308--

306-- @since 0.3.0

309-- @since 0.3.0

307aheadly :: IsStream t => AheadT m a -> t m a

310aheadly :: IsStream t => AheadT m a -> t m a

308−aheadly = adapt

311+aheadly = K.adapt

309

312

310instance IsStream AheadT where

313instance IsStream AheadT where

311 toStream = getAheadT

314 toStream = getAheadT

@@ -339,7 +342,7 @@ instance MonadAsync m => Semigroup (AheadT m a) where

339------------------------------------------------------------------------------

342------------------------------------------------------------------------------

340

343

341instance MonadAsync m => Monoid (AheadT m a) where

344instance MonadAsync m => Monoid (AheadT m a) where

342− mempty = nil

345+ mempty = K.nil

343 mappend = (<>)

346 mappend = (<>)

344

347

345------------------------------------------------------------------------------

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

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

@@ -31,7 +31,6 @@ module Streamly.Streams.Async

31 , WAsync

32 , wAsyncly

33 , wAsync

34− , runAsyncT -- deprecated

35 )

34 )

36where

35where

@@ -49,11 +48,14 @@ import Data.IORef (IORef, newIORef, readIORef)

49import Data.Maybe (fromJust)

48import Data.Maybe (fromJust)

50import Data.Semigroup (Semigroup(..))

49import Data.Semigroup (Semigroup(..))

51+import Prelude hiding (map)

52import qualified Data.Set as S

54import Streamly.Streams.SVar (fromSVar)

55−import Streamly.Streams.StreamK

55+import Streamly.Streams.Serial (map)

56import Streamly.SVar

57+import Streamly.Streams.StreamK (IsStream(..), Stream(..), adapt)

58+import qualified Streamly.Streams.StreamK as K

58#include "Instances.hs"

60#include "Instances.hs"

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

323-- of combining streams using async.

325-- of combining streams using async.

324{-# INLINE consMAsync #-}

326{-# INLINE consMAsync #-}

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

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

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

328+consMAsync m r = K.yieldM m `asyncS` r

327

329

328------------------------------------------------------------------------------

330------------------------------------------------------------------------------

329-- AsyncT

331-- AsyncT

@@ -379,7 +381,7 @@ consMAsync m r = yieldM m `asyncS` r

379--

381--

380-- @since 0.1.0

382-- @since 0.1.0

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

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

382− deriving (Functor, MonadTrans)

384+ deriving (MonadTrans)

383

385

384-- | A demand driven left biased parallely composing IO stream of elements of

386-- | A demand driven left biased parallely composing IO stream of elements of

385-- type @a@. See 'AsyncT' documentation for more details.

387-- type @a@. See 'AsyncT' documentation for more details.

@@ -417,7 +419,7 @@ instance MonadAsync m => Semigroup (AsyncT m a) where

417------------------------------------------------------------------------------

419------------------------------------------------------------------------------

418

420

419instance MonadAsync m => Monoid (AsyncT m a) where

421instance MonadAsync m => Monoid (AsyncT m a) where

420− mempty = nil

422+ mempty = K.nil

421 mappend = (<>)

423 mappend = (<>)

422

424

423------------------------------------------------------------------------------

425------------------------------------------------------------------------------

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

426

428

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

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

428 return = pure

430 return = pure

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

431+ (AsyncT m) >>= f = AsyncT $ K.bindWith asyncS m (getAsyncT . f)

430

432

431------------------------------------------------------------------------------

433------------------------------------------------------------------------------

432-- Other instances

434-- Other instances

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

447-- of combining streams using wAsync.

449-- of combining streams using wAsync.

448{-# INLINE consMWAsync #-}

450{-# INLINE consMWAsync #-}

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

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

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

452+consMWAsync m r = K.yieldM m `wAsyncS` r

451

453

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

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

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

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

@@ -505,7 +507,7 @@ wAsync m1 m2 = fromStream $ wAsyncS (toStream m1) (toStream m2)

505--

507--

506-- @since 0.2.0

508-- @since 0.2.0

507newtype WAsyncT m a = WAsyncT {getWAsyncT :: Stream m a}

509newtype WAsyncT m a = WAsyncT {getWAsyncT :: Stream m a}

508− deriving (Functor, MonadTrans)

510+ deriving (MonadTrans)

509

511

510-- | A round robin parallely composing IO stream of elements of type @a@.

512-- | A round robin parallely composing IO stream of elements of type @a@.

511-- See 'WAsyncT' documentation for more details.

513-- See 'WAsyncT' documentation for more details.

@@ -543,7 +545,7 @@ instance MonadAsync m => Semigroup (WAsyncT m a) where

543------------------------------------------------------------------------------

545------------------------------------------------------------------------------

544

546

545instance MonadAsync m => Monoid (WAsyncT m a) where

547instance MonadAsync m => Monoid (WAsyncT m a) where

546− mempty = nil

548+ mempty = K.nil

547 mappend = (<>)

549 mappend = (<>)

548

550

549------------------------------------------------------------------------------

551------------------------------------------------------------------------------

@@ -553,7 +555,7 @@ instance MonadAsync m => Monoid (WAsyncT m a) where

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

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

554 return = pure

556 return = pure

555 (WAsyncT m) >>= f =

557 (WAsyncT m) >>= f =

556− WAsyncT $ bindWith wAsyncS m (getWAsyncT . f)

558+ WAsyncT $ K.bindWith wAsyncS m (getWAsyncT . f)

557

559

558------------------------------------------------------------------------------

560------------------------------------------------------------------------------

559-- Other instances

561-- Other instances

@@ -561,10 +563,3 @@ instance MonadAsync m => Monad (WAsyncT m) where

561

563

562MONAD_APPLICATIVE_INSTANCE(WAsyncT,MONADPARALLEL)

564MONAD_APPLICATIVE_INSTANCE(WAsyncT,MONADPARALLEL)

563MONAD_COMMON_INSTANCES(WAsyncT, MONADPARALLEL)

565MONAD_COMMON_INSTANCES(WAsyncT, MONADPARALLEL)

564−

565−-- | Same as @runStream . asyncly@.

566−--

567−-- @since 0.1.0

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

569−runAsyncT :: Monad m => AsyncT m a -> m ()

570−runAsyncT = runStream

src/Streamly/Streams/Instances.hs+5−2

@@ -9,10 +9,13 @@

10#define MONAD_APPLICATIVE_INSTANCE(STREAM,CONSTRAINT) \

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

12− pure = STREAM . yield; \

12+ pure = STREAM . K.yield; \

13 (<*>) = ap }

15#define MONAD_COMMON_INSTANCES(STREAM,CONSTRAINT) \

16+instance Monad m => Functor (STREAM m) where { \

17+ fmap = map }; \

18+ \

16instance (MonadBase b m, Monad m CONSTRAINT) => MonadBase b (STREAM m) where {\

19instance (MonadBase b m, Monad m CONSTRAINT) => MonadBase b (STREAM m) where {\

17 liftBase = liftBaseDefault }; \

20 liftBase = liftBaseDefault }; \

18 \

21 \

@@ -31,7 +34,7 @@ instance (MonadError e m CONSTRAINT) => MonadError e (STREAM m) where { \

31 \

34 \

32instance (MonadReader r m CONSTRAINT) => MonadReader r (STREAM m) where { \

35instance (MonadReader r m CONSTRAINT) => MonadReader r (STREAM m) where { \

33 ask = lift ask; \

36 ask = lift ask; \

34− local f m = fromStream $ withLocal f (toStream m) }; \

37+ local f m = fromStream $ K.withLocal f (toStream m) }; \

35 \

38 \

36instance (MonadState s m CONSTRAINT) => MonadState s (STREAM m) where { \

39instance (MonadState s m CONSTRAINT) => MonadState s (STREAM m) where { \

37 get = lift get; \

40 get = lift get; \

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

@@ -24,7 +24,6 @@ module Streamly.Streams.Parallel

24 , Parallel

25 , parallely

26 , parallel

27− , runParallelT -- deprecated

29 -- * Function application

28 -- * Function application

30 , mkParallel

29 , mkParallel

@@ -45,10 +44,13 @@ import Control.Monad.State.Class (MonadState(..))

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

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

46import Data.Functor (void)

45import Data.Functor (void)

47import Data.Semigroup (Semigroup(..))

46import Data.Semigroup (Semigroup(..))

47+import Prelude hiding (map)

49import Streamly.Streams.SVar (fromSVar)

50−import Streamly.Streams.StreamK

50+import Streamly.Streams.Serial (map)

51import Streamly.SVar

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

53+import qualified Streamly.Streams.StreamK as K

53#include "Instances.hs"

55#include "Instances.hs"

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

96-- of combining streams using parallel.

98-- of combining streams using parallel.

97{-# INLINE consMParallel #-}

99{-# INLINE consMParallel #-}

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

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

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

101+consMParallel m r = K.yieldM m `parallelStream` r

100

102

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

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

102-- Merges two streams concurrently.

104-- Merges two streams concurrently.

@@ -307,7 +309,7 @@ x |&. f = f |$. x

307--

309--

308-- @since 0.1.0

310-- @since 0.1.0

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

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

310− deriving (Functor, MonadTrans)

312+ deriving (MonadTrans)

311

313

312-- | A parallely composing IO stream of elements of type @a@.

314-- | A parallely composing IO stream of elements of type @a@.

313-- See 'ParallelT' documentation for more details.

315-- See 'ParallelT' documentation for more details.

@@ -345,7 +347,7 @@ instance MonadAsync m => Semigroup (ParallelT m a) where

345------------------------------------------------------------------------------

347------------------------------------------------------------------------------

346

348

347instance MonadAsync m => Monoid (ParallelT m a) where

349instance MonadAsync m => Monoid (ParallelT m a) where

348− mempty = nil

350+ mempty = K.nil

349 mappend = (<>)

351 mappend = (<>)

350

352

351------------------------------------------------------------------------------

353------------------------------------------------------------------------------

@@ -355,7 +357,7 @@ instance MonadAsync m => Monoid (ParallelT m a) where

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

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

356 return = pure

358 return = pure

357 (ParallelT m) >>= f

359 (ParallelT m) >>= f

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

360+ = ParallelT $ K.bindWith parallelStream m (getParallelT . f)

359

361

360------------------------------------------------------------------------------

362------------------------------------------------------------------------------

361-- Other instances

363-- Other instances

@@ -363,10 +365,3 @@ instance MonadAsync m => Monad (ParallelT m) where

363

365

364MONAD_APPLICATIVE_INSTANCE(ParallelT,MONADPARALLEL)

366MONAD_APPLICATIVE_INSTANCE(ParallelT,MONADPARALLEL)

365MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)

367MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)

366−

367−-- | Same as @runStream . parallely@.

368−--

369−-- @since 0.1.0

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

371−runParallelT :: Monad m => ParallelT m a -> m ()

372−runParallelT = runStream

src/Streamly/Streams/Prelude.hs+59−7

@@ -29,7 +29,13 @@ module Streamly.Streams.Prelude

29 (

30 -- * Elimination

31 runStream

32− , runStreaming -- deprecated

32+ , runStreaming -- deprecated

33+ , runStreamT -- deprecated

34+ , runInterleavedT -- deprecated

35+ , runParallelT -- deprecated

36+ , runAsyncT -- deprecated

37+ , runZipStream -- deprecated

38+ , runZipAsync -- deprecated

34 -- * Fold Utilities

40 -- * Fold Utilities

35 , foldWith

41 , foldWith

@@ -38,8 +44,12 @@ module Streamly.Streams.Prelude

38 )

44 )

39where

45where

41−import Streamly.Streams.Serial (SerialT)

47+import Streamly.Streams.StreamK (IsStream(..))

48+import Streamly.Streams.Serial (SerialT, WSerialT)

42−import Streamly.Streams.StreamK hiding (runStream)

49+import Streamly.Streams.Parallel (ParallelT)

50+import Streamly.Streams.Async (AsyncT)

51+import Streamly.Streams.Zip (ZipSerialM, ZipAsyncM)

52+

43import qualified Streamly.Streams.StreamD as D

53import qualified Streamly.Streams.StreamD as D

44import qualified Streamly.Streams.StreamK as K

54import qualified Streamly.Streams.StreamK as K

@@ -63,7 +73,49 @@ runStream m = D.runStream $ D.fromStreamK (toStream m)

63-- @since 0.1.0

73-- @since 0.1.0

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

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

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

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

66−runStreaming = runStream . adapt

76+runStreaming = runStream . K.adapt

77+

78+-- | Same as @runStream@.

79+--

80+-- @since 0.1.0

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

82+runStreamT :: Monad m => SerialT m a -> m ()

83+runStreamT = runStream

84+

85+-- | Same as @runStream . wSerially@.

86+--

87+-- @since 0.1.0

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

89+runInterleavedT :: Monad m => WSerialT m a -> m ()

90+runInterleavedT = runStream . K.adapt

91+

92+-- | Same as @runStream . parallely@.

93+--

94+-- @since 0.1.0

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

96+runParallelT :: Monad m => ParallelT m a -> m ()

97+runParallelT = runStream . K.adapt

98+

99+-- | Same as @runStream . asyncly@.

100+--

101+-- @since 0.1.0

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

103+runAsyncT :: Monad m => AsyncT m a -> m ()

104+runAsyncT = runStream . K.adapt

105+

106+-- | Same as @runStream . zipping@.

107+--

108+-- @since 0.1.0

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

110+runZipStream :: Monad m => ZipSerialM m a -> m ()

111+runZipStream = runStream . K.adapt

112+

113+-- | Same as @runStream . zippingAsync@.

114+--

115+-- @since 0.1.0

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

117+runZipAsync :: Monad m => ZipAsyncM m a -> m ()

118+runZipAsync = runStream . K.adapt

119

68------------------------------------------------------------------------------

120------------------------------------------------------------------------------

69-- Fold Utilities

121-- Fold Utilities

@@ -78,7 +130,7 @@ runStreaming = runStream . adapt

78{-# INLINABLE foldWith #-}

130{-# INLINABLE foldWith #-}

79foldWith :: (IsStream t, Foldable f)

131foldWith :: (IsStream t, Foldable f)

80 => (t m a -> t m a -> t m a) -> f (t m a) -> t m a

132 => (t m a -> t m a -> t m a) -> f (t m a) -> t m a

81−foldWith f = foldr f nil

133+foldWith f = foldr f K.nil

134

83-- | A variant of 'foldMap' that allows you to map a monadic streaming action

135-- | A variant of 'foldMap' that allows you to map a monadic streaming action

84-- on a 'Foldable' container and then fold it using the specified stream sum

136-- on a 'Foldable' container and then fold it using the specified stream sum

@@ -90,7 +142,7 @@ foldWith f = foldr f nil

90{-# INLINABLE foldMapWith #-}

142{-# INLINABLE foldMapWith #-}

91foldMapWith :: (IsStream t, Foldable f)

143foldMapWith :: (IsStream t, Foldable f)

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

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

93−foldMapWith f g = foldr (f . g) nil

145+foldMapWith f g = foldr (f . g) K.nil

146

95-- | Like 'foldMapWith' but with the last two arguments reversed i.e. the

147-- | Like 'foldMapWith' but with the last two arguments reversed i.e. the

96-- monadic streaming function is the last argument.

148-- monadic streaming function is the last argument.

@@ -99,4 +151,4 @@ foldMapWith f g = foldr (f . g) nil

99{-# INLINABLE forEachWith #-}

151{-# INLINABLE forEachWith #-}

100forEachWith :: (IsStream t, Foldable f)

152forEachWith :: (IsStream t, Foldable f)

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

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

102−forEachWith f xs g = foldr (f . g) nil xs

154+forEachWith f xs g = foldr (f . g) K.nil xs

src/Streamly/Streams/Serial.hs+22−29

@@ -36,9 +36,9 @@ module Streamly.Streams.Serial

36 , wSerially

37 , interleaving -- deprecated

39− -- * Running Streams

39+ -- * Transformation

40− , runStreamT -- deprecated

40+ , map

41− , runInterleavedT -- deprecated

41+ , mapM

42 )

43where

@@ -51,11 +51,14 @@ import Control.Monad.Reader.Class (MonadReader(..))

51import Control.Monad.State.Class (MonadState(..))

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

53import Data.Semigroup (Semigroup(..))

54+import Prelude hiding (map, mapM)

55−import Streamly.Streams.StreamK hiding (serial)

56+import Streamly.Streams.StreamK (IsStream(..), adapt, Stream(..))

56import qualified Streamly.Streams.StreamK as K

57import qualified Streamly.Streams.StreamK as K

58+import qualified Streamly.Streams.StreamD as D

58#include "Instances.hs"

60#include "Instances.hs"

61+#include "inline.h"

60------------------------------------------------------------------------------

63------------------------------------------------------------------------------

61-- SerialT

64-- SerialT

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

118--

121--

119-- @since 0.2.0

122-- @since 0.2.0

120newtype SerialT m a = SerialT {getSerialT :: Stream m a}

123newtype SerialT m a = SerialT {getSerialT :: Stream m a}

121− deriving (Semigroup, Monoid, Functor, MonadTrans)

124+ deriving (Semigroup, Monoid, MonadTrans)

122

125

123-- | A serial IO stream of elements of type @a@. See 'SerialT' documentation

126-- | A serial IO stream of elements of type @a@. See 'SerialT' documentation

124-- for more details.

127-- for more details.

@@ -144,12 +147,12 @@ instance IsStream SerialT where

144 {-# INLINE consM #-}

147 {-# INLINE consM #-}

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

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

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

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

147− consM m r = fromStream $ consMSerial m (toStream r)

150+ consM m r = fromStream $ K.consMSerial m (toStream r)

148

151

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

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

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

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

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

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

152− m |: r = fromStream $ consMSerial m (toStream r)

155+ m |: r = fromStream $ K.consMSerial m (toStream r)

153

156

154------------------------------------------------------------------------------

157------------------------------------------------------------------------------

155-- Semigroup

158-- Semigroup

@@ -180,6 +183,14 @@ instance Monad m => Monad (SerialT m) where

180-- Other instances

183-- Other instances

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

184------------------------------------------------------------------------------

182

185

186+{-# INLINE_EARLY mapM #-}

187+mapM :: (IsStream t, Monad m) => (a -> m b) -> t m a -> t m b

188+mapM f m = fromStream $ D.toStreamK $ D.mapM f $ D.fromStreamK (toStream m)

189+

190+{-# INLINE map #-}

191+map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b

192+map f = mapM (return . f)

193+

183MONAD_APPLICATIVE_INSTANCE(SerialT,)

194MONAD_APPLICATIVE_INSTANCE(SerialT,)

184MONAD_COMMON_INSTANCES(SerialT,)

195MONAD_COMMON_INSTANCES(SerialT,)

185

196

@@ -225,7 +236,7 @@ MONAD_COMMON_INSTANCES(SerialT,)

225--

236--

226-- @since 0.2.0

237-- @since 0.2.0

227newtype WSerialT m a = WSerialT {getWSerialT :: Stream m a}

238newtype WSerialT m a = WSerialT {getWSerialT :: Stream m a}

228− deriving (Functor, MonadTrans)

239+ deriving (MonadTrans)

229

240

230-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'

241-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'

231-- documentation for more details.

242-- documentation for more details.

@@ -258,12 +269,12 @@ instance IsStream WSerialT where

258 {-# INLINE consM #-}

269 {-# INLINE consM #-}

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

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

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

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

261− consM m r = fromStream $ consMSerial m (toStream r)

272+ consM m r = fromStream $ K.consMSerial m (toStream r)

262

273

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

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

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

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

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

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

266− m |: r = fromStream $ consMSerial m (toStream r)

277+ m |: r = fromStream $ K.consMSerial m (toStream r)

267

278

268------------------------------------------------------------------------------

279------------------------------------------------------------------------------

269-- Semigroup

280-- Semigroup

@@ -303,7 +314,7 @@ infixr 5 <=>

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

314------------------------------------------------------------------------------

304

315

305instance Monoid (WSerialT m a) where

316instance Monoid (WSerialT m a) where

306− mempty = nil

317+ mempty = K.nil

307 mappend = (<>)

318 mappend = (<>)

308

319

309------------------------------------------------------------------------------

320------------------------------------------------------------------------------

@@ -324,21 +335,3 @@ instance Monad m => Monad (WSerialT m) where

324

335

325MONAD_APPLICATIVE_INSTANCE(WSerialT,)

336MONAD_APPLICATIVE_INSTANCE(WSerialT,)

326MONAD_COMMON_INSTANCES(WSerialT,)

337MONAD_COMMON_INSTANCES(WSerialT,)

327−

328−-------------------------------------------------------------------------------

329−-- Running Streams

330−-------------------------------------------------------------------------------

331−

332−-- | Same as @runStream@.

333−--

334−-- @since 0.1.0

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

336−runStreamT :: Monad m => SerialT m a -> m ()

337−runStreamT = runStream

338−

339−-- | Same as @runStream . wSerially@.

340−--

341−-- @since 0.1.0

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

343−runInterleavedT :: Monad m => InterleavedT m a -> m ()

344−runInterleavedT = runStream

src/Streamly/Streams/StreamD.hs+87−11

@@ -34,24 +34,40 @@ module Streamly.Streams.StreamD

34 Step (..)

35 , Stream (..)

37− -- * Elimination

37+ -- * Construction

38− , runStream

39− , uncons

40−

41− -- * Generation

42 , nil

38 , nil

43 , yield

39 , yield

44 , yieldM

40 , yieldM

41+

42+ -- * Generation by Unfolding

45 , unfoldr

43 , unfoldr

46 , unfoldrM

44 , unfoldrM

45+

46+ -- * Special Generation

47 , repeat

48 , enumFromStepN

49 , fromList

50 , fromListM

52− -- * Transformation

52+ -- * Deconstruction

53− , mapM

53+ , uncons

54+

55+ -- * Elimination by Folding

56+ -- ** General Folds

57+ , foldr

58+ , foldrM

59+ , foldl'

60+ , foldlM'

61+

62+ -- ** Special Folds

63+ , runStream

54 , mapM_

64 , mapM_

65+ , toList

66+ , last

67+

68+ -- * Mapping

69+ , map

70+ , mapM

56 -- * Conversion

72 -- * Conversion

57 , toStreamK

73 , toStreamK

@@ -59,9 +75,11 @@ module Streamly.Streams.StreamD

59 )

75 )

60where

76where

78+import GHC.Types ( SPEC(..) )

79+import Prelude hiding (map, mapM, mapM_, repeat, foldr, last)

80+

62import Streamly.SVar (MonadAsync)

81import Streamly.SVar (MonadAsync)

63import qualified Streamly.Streams.StreamK as K

82import qualified Streamly.Streams.StreamK as K

64−import Prelude hiding (mapM, mapM_, repeat)

66------------------------------------------------------------------------------

84------------------------------------------------------------------------------

67-- The direct style stream type

85-- The direct style stream type

@@ -208,6 +226,14 @@ mapM f (Stream step state) = Stream step' state

208 Yield x s -> f x >>= \a -> return $ Yield a s

226 Yield x s -> f x >>= \a -> return $ Yield a s

209 Stop -> return Stop

227 Stop -> return Stop

210

228

229+{-# INLINE map #-}

230+map :: Monad m => (a -> b) -> Stream m a -> Stream m b

231+map f = mapM (return . f)

232+

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

234+ {-# INLINE fmap #-}

235+ fmap = map

236+

211-------------------------------------------------------------------------------

237-------------------------------------------------------------------------------

212-- Elimination

238-- Elimination

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

239-------------------------------------------------------------------------------

@@ -225,15 +251,65 @@ uncons (Stream step state) = go state

225-- | Run a streaming composition, discard the results.

251-- | Run a streaming composition, discard the results.

226{-# INLINE_LATE runStream #-}

252{-# INLINE_LATE runStream #-}

227runStream :: Monad m => Stream m a -> m ()

253runStream :: Monad m => Stream m a -> m ()

228−runStream (Stream step state) = go state

254+runStream (Stream step state) = go SPEC state

229 where

255 where

230− go st = do

256+ go !_ st = do

231 r <- step st

257 r <- step st

232 case r of

258 case r of

233− Yield _ s -> go s

259+ Yield _ s -> go SPEC s

234 Stop -> return ()

260 Stop -> return ()

235

261

236-- | Execute a monadic action for each element of the 'Stream'

262-- | Execute a monadic action for each element of the 'Stream'

237{-# INLINE_NORMAL mapM_ #-}

263{-# INLINE_NORMAL mapM_ #-}

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

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

239mapM_ m = runStream . mapM m

265mapM_ m = runStream . mapM m

266+

267+{-# INLINE_NORMAL foldrM #-}

268+foldrM :: Monad m => (a -> b -> m b) -> b -> Stream m a -> m b

269+foldrM f z (Stream step state) = go SPEC state

270+ where

271+ go !_ st = do

272+ r <- step st

273+ case r of

274+ Yield x s -> go SPEC s >>= f x

275+ Stop -> return z

276+

277+{-# INLINE_NORMAL foldr #-}

278+foldr :: Monad m => (a -> b -> b) -> b -> Stream m a -> m b

279+foldr f = foldrM (\a b -> return (f a b))

280+

281+{-# INLINE_NORMAL foldlM' #-}

282+foldlM' :: Monad m => (b -> a -> m b) -> b -> Stream m a -> m b

283+foldlM' fstep begin (Stream step state) = go SPEC begin state

284+ where

285+ go !_ acc st = acc `seq` do

286+ r <- step st

287+ case r of

288+ Yield x s -> do

289+ acc' <- fstep acc x

290+ go SPEC acc' s

291+ Stop -> return acc

292+

293+{-# INLINE foldl' #-}

294+foldl' :: Monad m => (b -> a -> b) -> b -> Stream m a -> m b

295+foldl' fstep = foldlM' (\b a -> return (fstep b a))

296+

297+{-# INLINE toList #-}

298+toList :: Monad m => Stream m a -> m [a]

299+toList = foldr (:) []

300+

301+{-# INLINE_NORMAL last #-}

302+last :: Monad m => Stream m a -> m (Maybe a)

303+last (Stream step state) = go0 SPEC state

304+ where

305+ go0 !_ st = do

306+ r <- step st

307+ case r of

308+ Yield x s -> go1 SPEC x s

309+ Stop -> return Nothing

310+

311+ go1 !_ x st = do

312+ r <- step st

313+ case r of

314+ Yield y s -> go1 SPEC y s

315+ Stop -> return (Just x)

src/Streamly/Streams/StreamK.hs+127−9

@@ -1,3 +1,4 @@

1+{-# LANGUAGE BangPatterns #-}

1{-# LANGUAGE CPP #-}

2{-# LANGUAGE CPP #-}

2{-# LANGUAGE ConstraintKinds #-}

3{-# LANGUAGE ConstraintKinds #-}

3{-# LANGUAGE FlexibleContexts #-}

4{-# LANGUAGE FlexibleContexts #-}

@@ -34,10 +35,6 @@ module Streamly.Streams.StreamK

34 -- * The stream type

35 -- * The stream type

35 , Stream (..)

36 , Stream (..)

37− -- * Elimination

38− , foldStream

39− , runStream

40−

41 -- * Construction

38 -- * Construction

42 , mkStream

39 , mkStream

43 , nil

40 , nil

@@ -57,6 +54,24 @@ module Streamly.Streams.StreamK

57 , repeat

54 , repeat

58 , fromFoldable

55 , fromFoldable

57+ -- * Elimination

58+ , foldStream

59+ , foldr

60+ , foldrM

61+ , foldx

62+ , foldl'

63+ , foldxM

64+ , foldlM'

65+

66+ , runStream

67+ , mapM_

68+ , toList

69+ , last

70+

71+ -- * Transformation

72+ , map

73+ , mapM

74+

60 -- * Semigroup Style Composition

75 -- * Semigroup Style Composition

61 , serial

76 , serial

@@ -71,10 +86,12 @@ module Streamly.Streams.StreamK

71 )

86 )

72where

87where

89+import Control.Monad (void)

74import Control.Monad.Reader.Class (MonadReader(..))

90import Control.Monad.Reader.Class (MonadReader(..))

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

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

76import Data.Semigroup (Semigroup(..))

92import Data.Semigroup (Semigroup(..))

77−import Prelude hiding (repeat)

93+import Prelude hiding (foldl, foldr, last, map, mapM, mapM_, repeat)

94+import qualified Prelude

79import Streamly.SVar

96import Streamly.SVar

@@ -328,6 +345,74 @@ foldStream svr blank single step m =

328 let yieldk a x = step a (fromStream x)

345 let yieldk a x = step a (fromStream x)

329 in (unStream (toStream m)) svr blank single yieldk

346 in (unStream (toStream m)) svr blank single yieldk

330

347

348+-- | Lazy right associative fold.

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

350+foldr step acc m = go (toStream m)

351+ where

352+ go m1 =

353+ let stop = return acc

354+ single a = return (step a acc)

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

356+ in (unStream m1) Nothing stop single yieldk

357+

358+-- | Lazy right fold with a monadic step function.

359+{-# INLINE foldrM #-}

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

361+foldrM step acc m = go (toStream m)

362+ where

363+ go m1 =

364+ let stop = return acc

365+ single a = step a acc

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

367+ in (unStream m1) Nothing stop single yieldk

368+

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

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

371+-- argument) to the folded value at the end. This is designed to work with the

372+-- @foldl@ library. The suffix @x@ is a mnemonic for extraction.

373+{-# INLINE foldx #-}

374+foldx :: (IsStream t, Monad m)

375+ => (x -> a -> x) -> x -> (x -> b) -> t m a -> m b

376+foldx step begin done m = get $ go (toStream m) begin

377+ where

378+ {-# NOINLINE get #-}

379+ get m1 =

380+ let single = return . done

381+ in (unStream m1) Nothing undefined single undefined

382+

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

384+ -- however that is more expensive because of unnecessary recursion

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

386+ -- continuations is much more efficient.

387+ go m1 !acc = Stream $ \_ _ sng yld ->

388+ let stop = sng acc

389+ single a = sng $ step acc a

390+ yieldk a r =

391+ let stream = go r (step acc a)

392+ in (unStream stream) Nothing undefined sng yld

393+ in (unStream m1) Nothing stop single yieldk

394+

395+-- | Strict left associative fold.

396+{-# INLINE foldl' #-}

397+foldl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> m b

398+foldl' step begin m = foldx step begin id m

399+

400+-- XXX replace the recursive "go" with explicit continuations.

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

402+foldxM :: (IsStream t, Monad m)

403+ => (x -> a -> m x) -> m x -> (x -> m b) -> t m a -> m b

404+foldxM step begin done m = go begin (toStream m)

405+ where

406+ go !acc m1 =

407+ let stop = acc >>= done

408+ single a = acc >>= \b -> step b a >>= done

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

410+ in (unStream m1) Nothing stop single yieldk

411+

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

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

414+foldlM' step begin m = foldxM step (return begin) return m

415+

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

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

332runStream m = go (toStream m)

417runStream m = go (toStream m)

333 where

418 where

@@ -337,6 +422,26 @@ runStream m = go (toStream m)

337 yieldk _ r = go (toStream r)

422 yieldk _ r = go (toStream r)

338 in (unStream m1) Nothing stop single yieldk

423 in (unStream m1) Nothing stop single yieldk

339

424

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

426+-- output of the action.

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

428+mapM_ f m = go (toStream m)

429+ where

430+ go m1 =

431+ let stop = return ()

432+ single a = void (f a)

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

434+ in (unStream m1) Nothing stop single yieldk

435+

436+{-# INLINABLE toList #-}

437+toList :: (IsStream t, Monad m) => t m a -> m [a]

438+toList = foldr (:) []

439+

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

441+{-# INLINE last #-}

442+last :: (IsStream t, Monad m) => t m a -> m (Maybe a)

443+last = foldx (\_ y -> Just y) Nothing id

444+

340-------------------------------------------------------------------------------

445-------------------------------------------------------------------------------

341-- Generation

446-- Generation

342-------------------------------------------------------------------------------

447-------------------------------------------------------------------------------

@@ -409,11 +514,24 @@ instance Monoid (Stream m a) where

409-- Functor

514-- Functor

410-------------------------------------------------------------------------------

515-------------------------------------------------------------------------------

411

516

517+{-# INLINE map #-}

518+map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b

519+map f m = fromStream $ Stream $ \_ stp sng yld ->

520+ let single = sng . f

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

522+ in unStream (toStream m) Nothing stp single yieldk

523+

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

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

525+ fmap = map

526+

527+{-# INLINE mapM #-}

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

529+mapM f m = go (toStream m)

530+ where

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

531+ go m1 = fromStream $ Stream $ \svr stp sng yld ->

414− let single = sng . f

532+ let single a = f a >>= sng

415− yieldk a r = yld (f a) (fmap f r)

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

416− in (unStream m) Nothing stp single yieldk

534+ in (unStream m1) Nothing stp single yieldk

417

535

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

536-------------------------------------------------------------------------------

419-- Bind utility

537-- Bind utility

src/Streamly/Streams/Zip.hs+23−30

@@ -30,23 +30,24 @@ module Streamly.Streams.Zip

30 , ZipStream -- deprecated

31 , zipSerially

32 , zipping -- deprecated

33− , runZipStream -- deprecated

35 , ZipAsyncM

34 , ZipAsyncM

36 , ZipAsync

35 , ZipAsync

37 , zipAsyncly

36 , zipAsyncly

38 , zippingAsync -- deprecated

37 , zippingAsync -- deprecated

39− , runZipAsync -- deprecated

40 )

38 )

41where

39where

43import Data.Semigroup (Semigroup(..))

41import Data.Semigroup (Semigroup(..))

44−import Prelude hiding (repeat, zipWith)

42+import Prelude hiding (map, repeat, zipWith)

44+import Streamly.Streams.StreamK (IsStream(..), Stream(..))

46import Streamly.Streams.Async (mkAsync)

45import Streamly.Streams.Async (mkAsync)

47−import Streamly.Streams.StreamK

46+import Streamly.Streams.Serial (map)

48import Streamly.SVar (MonadAsync)

47import Streamly.SVar (MonadAsync)

49+import qualified Streamly.Streams.StreamK as K

50+

50#include "Instances.hs"

51#include "Instances.hs"

52------------------------------------------------------------------------------

53------------------------------------------------------------------------------

@@ -62,7 +63,7 @@ zipWithS f m1 m2 = go m1 m2

62 let single2 b = sng (f a b)

63 let single2 b = sng (f a b)

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

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

64 in unStream my Nothing stp single2 yield2

65 in unStream my Nothing stp single2 yield2

65− let single1 a = merge a nil

66+ let single1 a = merge a K.nil

66 yield1 a ra = merge a ra

67 yield1 a ra = merge a ra

67 unStream mx Nothing stp single1 yield1

68 unStream mx Nothing stp single1 yield1

@@ -85,7 +86,7 @@ zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

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

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

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

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

87 in unStream my Nothing stp single2 yield2

88 in unStream my Nothing stp single2 yield2

88− let single1 a = merge a nil

89+ let single1 a = merge a K.nil

89 yield1 a ra = merge a ra

90 yield1 a ra = merge a ra

90 unStream mx Nothing stp single1 yield1

91 unStream mx Nothing stp single1 yield1

@@ -112,7 +113,7 @@ zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

112--

113--

113-- @since 0.2.0

114-- @since 0.2.0

114newtype ZipSerialM m a = ZipSerialM {getZipSerialM :: Stream m a}

115newtype ZipSerialM m a = ZipSerialM {getZipSerialM :: Stream m a}

115− deriving (Functor, Semigroup, Monoid)

116+ deriving (Semigroup, Monoid)

116

117

117-- |

118-- |

118-- @since 0.1.0

119-- @since 0.1.0

@@ -128,7 +129,7 @@ type ZipSerial a = ZipSerialM IO a

128--

129--

129-- @since 0.2.0

130-- @since 0.2.0

130zipSerially :: IsStream t => ZipSerialM m a -> t m a

131zipSerially :: IsStream t => ZipSerialM m a -> t m a

131−zipSerially = adapt

132+zipSerially = K.adapt

132

133

133-- | Same as 'zipSerially'.

134-- | Same as 'zipSerially'.

134--

135--

@@ -144,15 +145,18 @@ instance IsStream ZipSerialM where

144 {-# INLINE consM #-}

145 {-# INLINE consM #-}

145 {-# SPECIALIZE consM :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}

146 {-# SPECIALIZE consM :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}

146 consM :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a

147 consM :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a

147− consM m r = fromStream $ consMSerial m (toStream r)

148+ consM m r = fromStream $ K.consMSerial m (toStream r)

148

149

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

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

150 {-# SPECIALIZE (|:) :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}

151 {-# SPECIALIZE (|:) :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}

151 (|:) :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a

152 (|:) :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a

152− m |: r = fromStream $ consMSerial m (toStream r)

153+ m |: r = fromStream $ K.consMSerial m (toStream r)

154+

155+instance Monad m => Functor (ZipSerialM m) where

156+ fmap = map

153

157

154instance Monad m => Applicative (ZipSerialM m) where

158instance Monad m => Applicative (ZipSerialM m) where

155− pure = ZipSerialM . repeat

159+ pure = ZipSerialM . K.repeat

156 m1 <*> m2 = fromStream $ zipWith id (toStream m1) (toStream m2)

160 m1 <*> m2 = fromStream $ zipWith id (toStream m1) (toStream m2)

157

161

158------------------------------------------------------------------------------

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

@@ -203,7 +207,7 @@ zipAsyncWithM f m1 m2 = fromStream $ Stream $ \_ stp sng yld -> do

203--

207--

204-- @since 0.2.0

208-- @since 0.2.0

205newtype ZipAsyncM m a = ZipAsyncM {getZipAsyncM :: Stream m a}

209newtype ZipAsyncM m a = ZipAsyncM {getZipAsyncM :: Stream m a}

206− deriving (Functor, Semigroup, Monoid)

210+ deriving (Semigroup, Monoid)

207

211

208-- | An IO stream whose applicative instance zips streams wAsyncly.

212-- | An IO stream whose applicative instance zips streams wAsyncly.

209--

213--

@@ -214,7 +218,7 @@ type ZipAsync a = ZipAsyncM IO a

214--

218--

215-- @since 0.2.0

219-- @since 0.2.0

216zipAsyncly :: IsStream t => ZipAsyncM m a -> t m a

220zipAsyncly :: IsStream t => ZipAsyncM m a -> t m a

217−zipAsyncly = adapt

221+zipAsyncly = K.adapt

218

222

219-- | Same as 'zipAsyncly'.

223-- | Same as 'zipAsyncly'.

220--

224--

@@ -229,27 +233,16 @@ instance IsStream ZipAsyncM where

229 {-# INLINE consM #-}

233 {-# INLINE consM #-}

230 {-# SPECIALIZE consM :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}

234 {-# SPECIALIZE consM :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}

231 consM :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a

235 consM :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a

232− consM m r = fromStream $ consMSerial m (toStream r)

236+ consM m r = fromStream $ K.consMSerial m (toStream r)

233

237

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

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

235 {-# SPECIALIZE (|:) :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}

239 {-# SPECIALIZE (|:) :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}

236 (|:) :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a

240 (|:) :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a

237− m |: r = fromStream $ consMSerial m (toStream r)

241+ m |: r = fromStream $ K.consMSerial m (toStream r)

242+

243+instance Monad m => Functor (ZipAsyncM m) where

244+ fmap = map

238

245

239instance MonadAsync m => Applicative (ZipAsyncM m) where

246instance MonadAsync m => Applicative (ZipAsyncM m) where

240− pure = ZipAsyncM . repeat

247+ pure = ZipAsyncM . K.repeat

241 m1 <*> m2 = zipAsyncWith id m1 m2

248 m1 <*> m2 = zipAsyncWith id m1 m2

242−

243−-- | Same as @runStream . zipping@.

244−--

245−-- @since 0.1.0

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

247−runZipStream :: Monad m => ZipSerialM m a -> m ()

248−runZipStream = runStream

249−

250−-- | Same as @runStream . zippingAsync@.

251−--

252−-- @since 0.1.0

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

254−runZipAsync :: Monad m => ZipAsyncM m a -> m ()

255−runZipAsync = runStream

streamly.cabal+1

@@ -160,6 +160,7 @@ library

160 -Wnoncanonical-monadfail-instances

161

162 build-depends: base >= 4.8 && < 5

163+ , ghc-prim >= 0.2 && < 0.6

163 , containers >= 0.5 && < 0.6

164 , containers >= 0.5 && < 0.6

164 , heaps >= 0.3 && < 0.4

165 , heaps >= 0.3 && < 0.4

165

166