9dfce42 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Merge pull request #57 from composewell/is-stream

9dfce42Harendra Kumar9 years ago

Changelog.md+2

compare with master

unifiedsplit

@@ -8,6 +8,8 @@

8* Deprecate `runStreamT`, rename to `runSerialT`

9* Deprecate `ZipStream`, rename to `ZipSerial`

10* Deprecate `runZipStream`, rename to `runZipSerial`

11+* Deprecate `Streaming`, rename to `IsStream`

12+* Deprecate `runStreaming`, rename to `runStream`

12## 0.1.2

14## 0.1.2

benchmark/Main.hs+1−1

@@ -107,7 +107,7 @@ afilter = Main.filter

107

108{-# INLINE streamly_basic #-}

109streamly_basic

110− :: (Alternative (t IO), Monad (t IO), A.Streaming t)

110+ :: (Alternative (t IO), Monad (t IO), A.IsStream t)

111 => (forall a. t IO a -> IO [a])

112 -> (t IO Int -> t IO Int -> t IO Int)

113 -> IO Int

src/Streamly.hs+6−4

@@ -16,7 +16,7 @@ module Streamly

16 -- $overview

18 MonadAsync

19− , Streaming

19+ , IsStream

21 -- * Product Style Composition

22 -- $product

@@ -49,7 +49,7 @@ module Streamly

49 , adapt

51 -- * Running Streams

52− , runStreaming

52+ , runStream

53 , runSerialT

54 , runInterleavedT

55 , runAsyncT

@@ -72,6 +72,8 @@ module Streamly

72 , MonadTrans (..)

74 -- * Deprecated

75+ , Streaming

76+ , runStreaming

75 , runStreamT

77 , runStreamT

76 , runZipStream

78 , runZipStream

77 , StreamT

79 , StreamT

@@ -124,7 +126,7 @@ import Control.Monad.Trans.Class (MonadTrans (..))

124-- 'AsyncT' and 'ParallelT', 'ZipStream' and 'ZipAsync', each representing a

126-- 'AsyncT' and 'ParallelT', 'ZipStream' and 'ZipAsync', each representing a

125-- stream of elements. All these types have the same underlying representation

127-- stream of elements. All these types have the same underlying representation

126-- and can be adapted from one to another using type adaptor combinators

128-- and can be adapted from one to another using type adaptor combinators

127−-- described later. Each of these types belongs to the 'Streaming' type class

129+-- described later. Each of these types belongs to the 'IsStream' type class

128-- which helps converting the specific type to and from the underlying generic

130-- which helps converting the specific type to and from the underlying generic

129-- stream type.

131-- stream type.

130--

132--

@@ -213,7 +215,7 @@ import Control.Monad.Trans.Class (MonadTrans (..))

213--

215--

214-- Code using streamly is usually written such that it is agnostic of any

216-- Code using streamly is usually written such that it is agnostic of any

215-- specific streaming type. We use a type variable (polymorphic type) with the

217-- specific streaming type. We use a type variable (polymorphic type) with the

216−-- 'Streaming' class constraint. Finally, when running the monad we can specify

218+-- 'IsStream' class constraint. Finally, when running the monad we can specify

217-- the actual type that we want to use to interpret the code. However, in

219-- the actual type that we want to use to interpret the code. However, in

218-- certain cases we may want to use a specific type to force a certain type of

220-- certain cases we may want to use a specific type to force a certain type of

219-- composition. These combinators can be used to convert the stream types from

221-- composition. These combinators can be used to convert the stream types from

src/Streamly/Prelude.hs+41−40

@@ -93,13 +93,14 @@ import qualified Prelude

93import qualified System.IO as IO

95import Streamly.Core

96−import Streamly.Streams

96+import Streamly.Streams hiding (runStream)

97+

97------------------------------------------------------------------------------

98------------------------------------------------------------------------------

98-- Construction

99-- Construction

99------------------------------------------------------------------------------

100------------------------------------------------------------------------------

100

101

101-- | Build a Stream by unfolding pure steps starting from a seed.

102-- | Build a Stream by unfolding pure steps starting from a seed.

102−unfoldr :: Streaming t => (b -> Maybe (a, b)) -> b -> t m a

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

103unfoldr step = fromStream . go

104unfoldr step = fromStream . go

104 where

105 where

105 go s = Stream $ \_ stp yld ->

106 go s = Stream $ \_ stp yld ->

@@ -108,7 +109,7 @@ unfoldr step = fromStream . go

108 Just (a, b) -> yld a (Just (go b))

109 Just (a, b) -> yld a (Just (go b))

109

110

110-- | Build a Stream by unfolding monadic steps starting from a seed.

111-- | Build a Stream by unfolding monadic steps starting from a seed.

111−unfoldrM :: (Streaming t, Monad m) => (b -> m (Maybe (a, b))) -> b -> t m a

112+unfoldrM :: (IsStream t, Monad m) => (b -> m (Maybe (a, b))) -> b -> t m a

112unfoldrM step = fromStream . go

113unfoldrM step = fromStream . go

113 where

114 where

114 go s = Stream $ \_ stp yld -> do

115 go s = Stream $ \_ stp yld -> do

@@ -120,17 +121,17 @@ unfoldrM step = fromStream . go

120-- XXX need eachInterleaved, eachAsync, eachParallel

121-- XXX need eachInterleaved, eachAsync, eachParallel

121-- | Same as @foldWith (<>)@ but more efficient.

122-- | Same as @foldWith (<>)@ but more efficient.

122{-# INLINE each #-}

123{-# INLINE each #-}

123−each :: (Streaming t, Foldable f) => f a -> t m a

124+each :: (IsStream t, Foldable f) => f a -> t m a

124each = Prelude.foldr cons nil

125each = Prelude.foldr cons nil

125

126

126-- | Iterate a pure function from a seed value, streaming the results forever

127-- | Iterate a pure function from a seed value, streaming the results forever

127−iterate :: Streaming t => (a -> a) -> a -> t m a

128+iterate :: IsStream t => (a -> a) -> a -> t m a

128iterate step = fromStream . go

129iterate step = fromStream . go

129 where

130 where

130 go s = scons s (Just (go (step s)))

131 go s = scons s (Just (go (step s)))

131

132

132-- | Iterate a monadic function from a seed value, streaming the results forever

133-- | Iterate a monadic function from a seed value, streaming the results forever

133−iterateM :: (Streaming t, Monad m) => (a -> m a) -> a -> t m a

134+iterateM :: (IsStream t, Monad m) => (a -> m a) -> a -> t m a

134iterateM step = fromStream . go

135iterateM step = fromStream . go

135 where

136 where

136 go s = Stream $ \_ _ yld -> do

137 go s = Stream $ \_ _ yld -> do

@@ -138,7 +139,7 @@ iterateM step = fromStream . go

138 yld s (Just (go a))

139 yld s (Just (go a))

139

140

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

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

141−fromHandle :: (Streaming t, MonadIO m) => IO.Handle -> t m String

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

142fromHandle h = fromStream go

143fromHandle h = fromStream go

143 where

144 where

144 go = Stream $ \_ stp yld -> do

145 go = Stream $ \_ stp yld -> do

@@ -156,7 +157,7 @@ fromHandle h = fromStream go

156-- Parallel variants of folds?

157-- Parallel variants of folds?

157

158

158-- | Right fold.

159-- | Right fold.

159−foldr :: (Streaming t, Monad m) => (a -> b -> b) -> b -> t m a -> m b

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

160foldr step acc m = go (toStream m)

161foldr step acc m = go (toStream m)

161 where

162 where

162 go m1 =

163 go m1 =

@@ -167,7 +168,7 @@ foldr step acc m = go (toStream m)

167

168

168-- | Right fold with a monadic step function. See 'toList' for an example use.

169-- | Right fold with a monadic step function. See 'toList' for an example use.

169{-# INLINE foldrM #-}

170{-# INLINE foldrM #-}

170−foldrM :: (Streaming t, Monad m) => (a -> b -> m b) -> b -> t m a -> m b

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

171foldrM step acc m = go (toStream m)

172foldrM step acc m = go (toStream m)

172 where

173 where

173 go m1 =

174 go m1 =

@@ -178,7 +179,7 @@ foldrM step acc m = go (toStream m)

178

179

179-- | Scan left. A strict left fold which accumulates the result of its reduction steps inside a stream, from left.

180-- | Scan left. A strict left fold which accumulates the result of its reduction steps inside a stream, from left.

180{-# INLINE scan #-}

181{-# INLINE scan #-}

181−scan :: Streaming t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b

182+scan :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b

182scan step begin done m = cons (done begin) $ fromStream $ go (toStream m) begin

183scan step begin done m = cons (done begin) $ fromStream $ go (toStream m) begin

183 where

184 where

184 go m1 !acc = Stream $ \_ stp yld ->

185 go m1 !acc = Stream $ \_ stp yld ->

@@ -192,7 +193,7 @@ scan step begin done m = cons (done begin) $ fromStream $ go (toStream m) begin

192-- | Strict left fold. This is typed to work with the foldl package. To use

193-- | Strict left fold. This is typed to work with the foldl package. To use

193-- it normally just pass 'id' as the third argument.

194-- it normally just pass 'id' as the third argument.

194{-# INLINE foldl #-}

195{-# INLINE foldl #-}

195−foldl :: (Streaming t, Monad m)

196+foldl :: (IsStream t, Monad m)

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

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

197foldl step begin done m = get $ go (toStream m) begin

198foldl step begin done m = get $ go (toStream m) begin

198 where

199 where

@@ -218,7 +219,7 @@ foldl step begin done m = get $ go (toStream m) begin

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

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

219-- | Strict left fold, with monadic step function. This is typed to work

220-- | Strict left fold, with monadic step function. This is typed to work

220-- with the foldl package. To use directly pass 'id' as the third argument.

221-- with the foldl package. To use directly pass 'id' as the third argument.

221−foldlM :: (Streaming t, Monad m)

222+foldlM :: (IsStream t, Monad m)

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

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

223foldlM step begin done m = go begin (toStream m)

224foldlM step begin done m = go begin (toStream m)

224 where

225 where

@@ -231,7 +232,7 @@ foldlM step begin done m = go begin (toStream m)

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

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

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

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

233-- the head of the stream and 'ma' its tail.

234-- the head of the stream and 'ma' its tail.

234−uncons :: (Streaming t, Monad m) => t m a -> m (Maybe (a, t m a))

235+uncons :: (IsStream t, Monad m) => t m a -> m (Maybe (a, t m a))

235uncons m =

236uncons m =

236 let stop = return Nothing

237 let stop = return Nothing

237 yield a Nothing = return (Just (a, nil))

238 yield a Nothing = return (Just (a, nil))

@@ -239,7 +240,7 @@ uncons m =

239 in (runStream (toStream m)) Nothing stop yield

240 in (runStream (toStream m)) Nothing stop yield

240

241

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

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

242−toHandle :: (Streaming t, MonadIO m) => IO.Handle -> t m String -> m ()

243+toHandle :: (IsStream t, MonadIO m) => IO.Handle -> t m String -> m ()

243toHandle h m = go (toStream m)

244toHandle h m = go (toStream m)

244 where

245 where

245 go m1 =

246 go m1 =

@@ -254,12 +255,12 @@ toHandle h m = go (toStream m)

254

255

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

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

256{-# INLINABLE toList #-}

257{-# INLINABLE toList #-}

257−toList :: (Streaming t, Monad m) => t m a -> m [a]

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

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

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

259

260

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

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

261{-# INLINE take #-}

262{-# INLINE take #-}

262−take :: Streaming t => Int -> t m a -> t m a

263+take :: IsStream t => Int -> t m a -> t m a

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

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

264 where

265 where

265 go n1 m1 = Stream $ \ctx stp yld ->

266 go n1 m1 = Stream $ \ctx stp yld ->

@@ -269,7 +270,7 @@ take n m = fromStream $ go n (toStream m)

269

270

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

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

271{-# INLINE filter #-}

272{-# INLINE filter #-}

272−filter :: Streaming t => (a -> Bool) -> t m a -> t m a

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

273filter p m = fromStream $ go (toStream m)

274filter p m = fromStream $ go (toStream m)

274 where

275 where

275 go m1 = Stream $ \ctx stp yld ->

276 go m1 = Stream $ \ctx stp yld ->

@@ -281,7 +282,7 @@ filter p m = fromStream $ go (toStream m)

281

282

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

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

283{-# INLINE takeWhile #-}

284{-# INLINE takeWhile #-}

284−takeWhile :: Streaming t => (a -> Bool) -> t m a -> t m a

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

285takeWhile p m = fromStream $ go (toStream m)

286takeWhile p m = fromStream $ go (toStream m)

286 where

287 where

287 go m1 = Stream $ \ctx stp yld ->

288 go m1 = Stream $ \ctx stp yld ->

@@ -292,7 +293,7 @@ takeWhile p m = fromStream $ go (toStream m)

292 in (runStream m1) ctx stp yield

293 in (runStream m1) ctx stp yield

293

294

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

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

295−drop :: Streaming t => Int -> t m a -> t m a

296+drop :: IsStream t => Int -> t m a -> t m a

296drop n m = fromStream $ go n (toStream m)

297drop n m = fromStream $ go n (toStream m)

297 where

298 where

298 go n1 m1 = Stream $ \ctx stp yld ->

299 go n1 m1 = Stream $ \ctx stp yld ->

@@ -306,7 +307,7 @@ drop n m = fromStream $ go n (toStream m)

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

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

307-- take the rest of the stream.

308-- take the rest of the stream.

308{-# INLINE dropWhile #-}

309{-# INLINE dropWhile #-}

309−dropWhile :: Streaming t => (a -> Bool) -> t m a -> t m a

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

310dropWhile p m = fromStream $ go (toStream m)

311dropWhile p m = fromStream $ go (toStream m)

311 where

312 where

312 go m1 = Stream $ \ctx stp yld ->

313 go m1 = Stream $ \ctx stp yld ->

@@ -317,7 +318,7 @@ dropWhile p m = fromStream $ go (toStream m)

317 in (runStream m1) ctx stp yield

318 in (runStream m1) ctx stp yield

318

319

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

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

320−all :: (Streaming t, Monad m) => (a -> Bool) -> t m a -> m Bool

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

321all p m = go (toStream m)

322all p m = go (toStream m)

322 where

323 where

323 go m1 =

324 go m1 =

@@ -328,7 +329,7 @@ all p m = go (toStream m)

328 in (runStream m1) Nothing (return True) yield

329 in (runStream m1) Nothing (return True) yield

329

330

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

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

331−any :: (Streaming t, Monad m) => (a -> Bool) -> t m a -> m Bool

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

332any p m = go (toStream m)

333any p m = go (toStream m)

333 where

334 where

334 go m1 =

335 go m1 =

@@ -339,22 +340,22 @@ any p m = go (toStream m)

339 in (runStream m1) Nothing (return False) yield

340 in (runStream m1) Nothing (return False) yield

340

341

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

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

342−sum :: (Streaming t, Monad m, Num a) => t m a -> m a

343+sum :: (IsStream t, Monad m, Num a) => t m a -> m a

343sum = foldl (+) 0 id

344sum = foldl (+) 0 id

344

345

345-- | Determine the product of all elements of a stream of numbers

346-- | Determine the product of all elements of a stream of numbers

346−product :: (Streaming t, Monad m, Num a) => t m a -> m a

347+product :: (IsStream t, Monad m, Num a) => t m a -> m a

347product = foldl (*) 1 id

348product = foldl (*) 1 id

348

349

349-- | Extract the first element of the stream, if any.

350-- | Extract the first element of the stream, if any.

350−head :: (Streaming t, Monad m) => t m a -> m (Maybe a)

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

351head m =

352head m =

352 let stop = return Nothing

353 let stop = return Nothing

353 yield a _ = return (Just a)

354 yield a _ = return (Just a)

354 in (runStream (toStream m)) Nothing stop yield

355 in (runStream (toStream m)) Nothing stop yield

355

356

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

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

357−tail :: (Streaming t, Monad m) => t m a -> m (Maybe (t m a))

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

358tail m =

359tail m =

359 let stop = return Nothing

360 let stop = return Nothing

360 yield _ Nothing = return $ Just nil

361 yield _ Nothing = return $ Just nil

@@ -363,18 +364,18 @@ tail m =

363

364

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

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

365{-# INLINE last #-}

366{-# INLINE last #-}

366−last :: (Streaming t, Monad m) => t m a -> m (Maybe a)

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

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

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

368

369

369-- | Determine whether the stream is empty.

370-- | Determine whether the stream is empty.

370−null :: (Streaming t, Monad m) => t m a -> m Bool

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

371null m =

372null m =

372 let stop = return True

373 let stop = return True

373 yield _ _ = return False

374 yield _ _ = return False

374 in (runStream (toStream m)) Nothing stop yield

375 in (runStream (toStream m)) Nothing stop yield

375

376

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

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

377−elem :: (Streaming t, Monad m, Eq a) => a -> t m a -> m Bool

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

378elem e m = go (toStream m)

379elem e m = go (toStream m)

379 where

380 where

380 go m1 =

381 go m1 =

@@ -384,7 +385,7 @@ elem e m = go (toStream m)

384 in (runStream m1) Nothing stop yield

385 in (runStream m1) Nothing stop yield

385

386

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

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

387−notElem :: (Streaming t, Monad m, Eq a) => a -> t m a -> m Bool

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

388notElem e m = go (toStream m)

389notElem e m = go (toStream m)

389 where

390 where

390 go m1 =

391 go m1 =

@@ -394,12 +395,12 @@ notElem e m = go (toStream m)

394 in (runStream m1) Nothing stop yield

395 in (runStream m1) Nothing stop yield

395

396

396-- | Determine the length of the stream.

397-- | Determine the length of the stream.

397−length :: (Streaming t, Monad m) => t m a -> m Int

398+length :: (IsStream t, Monad m) => t m a -> m Int

398length = foldl (\n _ -> n + 1) 0 id

399length = foldl (\n _ -> n + 1) 0 id

399

400

400-- | Returns the elements of the stream in reverse order.

401-- | Returns the elements of the stream in reverse order.

401-- The stream must be finite.

402-- The stream must be finite.

402−reverse :: (Streaming t) => t m a -> t m a

403+reverse :: (IsStream t) => t m a -> t m a

403reverse m = fromStream $ go Nothing (toStream m)

404reverse m = fromStream $ go Nothing (toStream m)

404 where

405 where

405 go rev rest = Stream $ \svr stp yld ->

406 go rev rest = Stream $ \svr stp yld ->

@@ -412,7 +413,7 @@ reverse m = fromStream $ go Nothing (toStream m)

412

413

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

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

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

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

415−minimum :: (Streaming t, Monad m, Ord a) => t m a -> m (Maybe a)

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

416minimum m = go Nothing (toStream m)

417minimum m = go Nothing (toStream m)

417 where

418 where

418 go r m1 =

419 go r m1 =

@@ -427,7 +428,7 @@ minimum m = go Nothing (toStream m)

427

428

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

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

429-- | Determine the maximum element in a stream.

430-- | Determine the maximum element in a stream.

430−maximum :: (Streaming t, Monad m, Ord a) => t m a -> m (Maybe a)

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

431maximum m = go Nothing (toStream m)

432maximum m = go Nothing (toStream m)

432 where

433 where

433 go r m1 =

434 go r m1 =

@@ -449,7 +450,7 @@ maximum m = go Nothing (toStream m)

449-- | Replace each element of the stream with the result of a monadic action

450-- | Replace each element of the stream with the result of a monadic action

450-- applied on the element.

451-- applied on the element.

451{-# INLINE mapM #-}

452{-# INLINE mapM #-}

452−mapM :: (Streaming t, Monad m) => (a -> m b) -> t m a -> t m b

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

453mapM f m = fromStream $ go (toStream m)

454mapM f m = fromStream $ go (toStream m)

454 where

455 where

455 go m1 = Stream $ \_ stp yld ->

456 go m1 = Stream $ \_ stp yld ->

@@ -460,7 +461,7 @@ mapM f m = fromStream $ go (toStream m)

460

461

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

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

462-- output of the action.

463-- output of the action.

463−mapM_ :: (Streaming t, Monad m) => (a -> m b) -> t m a -> m ()

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

464mapM_ f m = go (toStream m)

465mapM_ f m = go (toStream m)

465 where

466 where

466 go m1 =

467 go m1 =

@@ -471,7 +472,7 @@ mapM_ f m = go (toStream m)

471

472

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

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

473-- actions.

474-- actions.

474−sequence :: (Streaming t, Monad m) => t m (m a) -> t m a

475+sequence :: (IsStream t, Monad m) => t m (m a) -> t m a

475sequence m = fromStream $ go (toStream m)

476sequence m = fromStream $ go (toStream m)

476 where

477 where

477 go m1 = Stream $ \_ stp yld ->

478 go m1 = Stream $ \_ stp yld ->

@@ -481,7 +482,7 @@ sequence m = fromStream $ go (toStream m)

481 in (runStream m1) Nothing stop yield

482 in (runStream m1) Nothing stop yield

482

483

483-- | Generate a stream by performing an action @n@ times.

484-- | Generate a stream by performing an action @n@ times.

484−replicateM :: (Streaming t, Monad m) => Int -> m a -> t m a

485+replicateM :: (IsStream t, Monad m) => Int -> m a -> t m a

485replicateM n m = fromStream $ go n

486replicateM n m = fromStream $ go n

486 where

487 where

487 go cnt = Stream $ \_ stp yld ->

488 go cnt = Stream $ \_ stp yld ->

@@ -494,7 +495,7 @@ replicateM n m = fromStream $ go n

494------------------------------------------------------------------------------

495------------------------------------------------------------------------------

495

496

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

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

497−zipWithM :: Streaming t => (a -> b -> t m c) -> t m a -> t m b -> t m c

498+zipWithM :: IsStream t => (a -> b -> t m c) -> t m a -> t m b -> t m c

498zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

499zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

499 where

500 where

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

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

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

514

515

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

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

516-- concurrently) using a monadic zipping function.

517-- concurrently) using a monadic zipping function.

517−zipAsyncWithM :: (Streaming t, MonadAsync m)

518+zipAsyncWithM :: (IsStream t, MonadAsync m)

518 => (a -> b -> t m c) -> t m a -> t m b -> t m c

519 => (a -> b -> t m c) -> t m a -> t m b -> t m c

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

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

520 ma <- async m1

521 ma <- async m1

src/Streamly/Streams.hs+62−49

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

1+{-# LANGUAGE ConstraintKinds #-}

1{-# LANGUAGE FlexibleContexts #-}

2{-# LANGUAGE FlexibleContexts #-}

2{-# LANGUAGE FlexibleInstances #-}

3{-# LANGUAGE FlexibleInstances #-}

3{-# LANGUAGE GeneralizedNewtypeDeriving#-}

4{-# LANGUAGE GeneralizedNewtypeDeriving#-}

@@ -18,7 +19,8 @@

18--

19--

19module Streamly.Streams

20module Streamly.Streams

20 (

21 (

21− Streaming (..)

22+ IsStream (..)

23+ , Streaming -- deprecated

22 , MonadAsync

24 , MonadAsync

24 -- * SVars

26 -- * SVars

@@ -38,7 +40,8 @@ module Streamly.Streams

39 -- * Elimination

41 -- * Elimination

40 , streamFold

42 , streamFold

43+ , runStream

41− , runStreaming

44+ , runStreaming -- deprecated

42 , toSVar

45 , toSVar

44 -- * Transformation

47 -- * Transformation

@@ -100,7 +103,8 @@ import Control.Monad.State.Class (MonadState(..))

100import Control.Monad.Trans.Class (MonadTrans)

103import Control.Monad.Trans.Class (MonadTrans)

101import Data.Semigroup (Semigroup(..))

104import Data.Semigroup (Semigroup(..))

102import Prelude hiding (zipWith)

105import Prelude hiding (zipWith)

103−import Streamly.Core

106+import Streamly.Core hiding (runStream)

107+import qualified Streamly.Core as SC

104

108

105------------------------------------------------------------------------------

109------------------------------------------------------------------------------

106-- Types that can behave as a Stream

110-- Types that can behave as a Stream

@@ -108,16 +112,20 @@ import Streamly.Core

108

112

109-- | Class of types that can represent a stream of elements of some type 'a' in

113-- | Class of types that can represent a stream of elements of some type 'a' in

110-- some monad 'm'.

114-- some monad 'm'.

111−class Streaming t where

115+class IsStream t where

112 toStream :: t m a -> Stream m a

116 toStream :: t m a -> Stream m a

113 fromStream :: Stream m a -> t m a

117 fromStream :: Stream m a -> t m a

114

118

119+-- | Same as 'IsStream'.

120+{-# Deprecated Streaming "Please use IsStream instead." #-}

121+type Streaming = IsStream

122+

115------------------------------------------------------------------------------

123------------------------------------------------------------------------------

116-- Constructing a stream

124-- Constructing a stream

117------------------------------------------------------------------------------

125------------------------------------------------------------------------------

118

126

119-- | Represesnts an empty stream just like @[]@ represents an empty list.

127-- | Represesnts an empty stream just like @[]@ represents an empty list.

120−nil :: Streaming t => t m a

128+nil :: IsStream t => t m a

121nil = fromStream snil

129nil = fromStream snil

122

130

123infixr 5 `cons`

131infixr 5 `cons`

@@ -130,7 +138,7 @@ infixr 5 `cons`

130-- > (toList . serially) stream

138-- > (toList . serially) stream

131-- [1,2,3]

139-- [1,2,3]

132-- @

140-- @

133−cons :: (Streaming t) => a -> t m a -> t m a

141+cons :: IsStream t => a -> t m a -> t m a

134cons a r = fromStream $ scons a (Just (toStream r))

142cons a r = fromStream $ scons a (Just (toStream r))

135

143

136infixr 5 .:

144infixr 5 .:

@@ -161,7 +169,7 @@ infixr 5 .:

161-- composition of streams. The former method is a bit more efficient than the

169-- composition of streams. The former method is a bit more efficient than the

162-- latter.

170-- latter.

163--

171--

164−(.:) :: (Streaming t) => a -> t m a -> t m a

172+(.:) :: IsStream t => a -> t m a -> t m a

165(.:) = cons

173(.:) = cons

166

174

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

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

@@ -169,7 +177,7 @@ infixr 5 .:

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

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

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

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

171-- represent an "empty" stream.

179-- represent an "empty" stream.

172−streamBuild :: Streaming t

180+streamBuild :: IsStream t

173 => (forall r. Maybe (SVar m a)

181 => (forall r. Maybe (SVar m a)

174 -> (a -> Maybe (t m a) -> m r)

182 -> (a -> Maybe (t m a) -> m r)

175 -> m r

183 -> m r

@@ -181,11 +189,11 @@ streamBuild k = fromStream $ Stream $ \sv stp yld ->

181 in k sv yield stp

189 in k sv yield stp

182

190

183-- | Build a singleton stream from a callback function.

191-- | Build a singleton stream from a callback function.

184−fromCallback :: (Streaming t) => (forall r. (a -> m r) -> m r) -> t m a

192+fromCallback :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a

185fromCallback k = fromStream $ Stream $ \_ _ yld -> k (\a -> yld a Nothing)

193fromCallback k = fromStream $ Stream $ \_ _ yld -> k (\a -> yld a Nothing)

186

194

187-- | Read an SVar to get a stream.

195-- | Read an SVar to get a stream.

188−fromSVar :: (MonadAsync m, Streaming t) => SVar m a -> t m a

196+fromSVar :: (MonadAsync m, IsStream t) => SVar m a -> t m a

189fromSVar sv = fromStream $ fromStreamVar sv

197fromSVar sv = fromStream $ fromStreamVar sv

190

198

191------------------------------------------------------------------------------

199------------------------------------------------------------------------------

@@ -195,26 +203,31 @@ fromSVar sv = fromStream $ fromStreamVar sv

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

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

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

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

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

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

198−streamFold :: Streaming t

206+streamFold :: IsStream t

199 => Maybe (SVar m a) -> (a -> Maybe (t m a) -> m r) -> m r -> t m a -> m r

207 => Maybe (SVar m a) -> (a -> Maybe (t m a) -> m r) -> m r -> t m a -> m r

200streamFold sv step blank m =

208streamFold sv step blank m =

201 let yield a Nothing = step a Nothing

209 let yield a Nothing = step a Nothing

202 yield a (Just x) = step a (Just (fromStream x))

210 yield a (Just x) = step a (Just (fromStream x))

203− in (runStream (toStream m)) sv blank yield

211+ in (SC.runStream (toStream m)) sv blank yield

204

212

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

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

206−runStreaming :: (Monad m, Streaming t) => t m a -> m ()

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

207−runStreaming m = go (toStream m)

215+runStream m = go (toStream m)

208 where

216 where

209 go m1 =

217 go m1 =

210 let stop = return ()

218 let stop = return ()

211 yield _ Nothing = stop

219 yield _ Nothing = stop

212 yield _ (Just x) = go x

220 yield _ (Just x) = go x

213− in (runStream m1) Nothing stop yield

221+ in (SC.runStream m1) Nothing stop yield

222+

223+-- | Same as 'runStream'

224+{-# Deprecated runStreaming "Please use runStream instead." #-}

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

226+runStreaming = runStream

214

227

215-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then

228-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then

216-- be read back from the SVar using 'fromSVar'.

229-- be read back from the SVar using 'fromSVar'.

217−toSVar :: (Streaming t, MonadAsync m) => SVar m a -> t m a -> m ()

230+toSVar :: (IsStream t, MonadAsync m) => SVar m a -> t m a -> m ()

218toSVar sv m = toStreamVar sv (toStream m)

231toSVar sv m = toStreamVar sv (toStream m)

219

232

220------------------------------------------------------------------------------

233------------------------------------------------------------------------------

@@ -228,7 +241,7 @@ toSVar sv m = toStreamVar sv (toStream m)

228-- not drained fully we may have a thread blocked forever and once exhausted it

241-- not drained fully we may have a thread blocked forever and once exhausted it

229-- will always return 'empty'.

242-- will always return 'empty'.

230

243

231−async :: (Streaming t, MonadAsync m) => t m a -> m (t m a)

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

232async m = do

245async m = do

233 sv <- newStreamVar1 (SVarStyle Disjunction LIFO) (toStream m)

246 sv <- newStreamVar1 (SVarStyle Disjunction LIFO) (toStream m)

234 return $ fromSVar sv

247 return $ fromSVar sv

@@ -281,7 +294,7 @@ deriving instance MonadError e m => MonadError e (SerialT m)

281deriving instance MonadReader r m => MonadReader r (SerialT m)

294deriving instance MonadReader r m => MonadReader r (SerialT m)

282deriving instance MonadState s m => MonadState s (SerialT m)

295deriving instance MonadState s m => MonadState s (SerialT m)

283

296

284−instance Streaming SerialT where

297+instance IsStream SerialT where

285 toStream = getSerialT

298 toStream = getSerialT

286 fromStream = SerialT

299 fromStream = SerialT

287

300

@@ -301,7 +314,7 @@ type StreamT = SerialT

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

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

302 return = pure

315 return = pure

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

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

304− let run x = (runStream x) Nothing stp yld

317+ let run x = (SC.runStream x) Nothing stp yld

305 yield a Nothing = run $ getSerialT (f a)

318 yield a Nothing = run $ getSerialT (f a)

306 yield a (Just r) = run $ getSerialT (f a)

319 yield a (Just r) = run $ getSerialT (f a)

307 <> getSerialT (SerialT r >>= f)

320 <> getSerialT (SerialT r >>= f)

@@ -402,14 +415,14 @@ deriving instance MonadError e m => MonadError e (InterleavedT m)

402deriving instance MonadReader r m => MonadReader r (InterleavedT m)

415deriving instance MonadReader r m => MonadReader r (InterleavedT m)

403deriving instance MonadState s m => MonadState s (InterleavedT m)

416deriving instance MonadState s m => MonadState s (InterleavedT m)

404

417

405−instance Streaming InterleavedT where

418+instance IsStream InterleavedT where

406 toStream = getInterleavedT

419 toStream = getInterleavedT

407 fromStream = InterleavedT

420 fromStream = InterleavedT

408

421

409instance Monad m => Monad (InterleavedT m) where

422instance Monad m => Monad (InterleavedT m) where

410 return = pure

423 return = pure

411 (InterleavedT (Stream m)) >>= f = InterleavedT $ Stream $ \_ stp yld ->

424 (InterleavedT (Stream m)) >>= f = InterleavedT $ Stream $ \_ stp yld ->

412− let run x = (runStream x) Nothing stp yld

425+ let run x = (SC.runStream x) Nothing stp yld

413 yield a Nothing = run $ getInterleavedT (f a)

426 yield a Nothing = run $ getInterleavedT (f a)

414 yield a (Just r) = run $ getInterleavedT (f a)

427 yield a (Just r) = run $ getInterleavedT (f a)

415 `interleave`

428 `interleave`

@@ -516,7 +529,7 @@ deriving instance (MonadError e m, MonadAsync m) => MonadError e (AsyncT m)

516deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (AsyncT m)

529deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (AsyncT m)

517deriving instance (MonadState s m, MonadAsync m) => MonadState s (AsyncT m)

530deriving instance (MonadState s m, MonadAsync m) => MonadState s (AsyncT m)

518

531

519−instance Streaming AsyncT where

532+instance IsStream AsyncT where

520 toStream = getAsyncT

533 toStream = getAsyncT

521 fromStream = AsyncT

534 fromStream = AsyncT

522

535

@@ -530,7 +543,7 @@ parbind par m f = go m

530 where

543 where

531 go (Stream g) =

544 go (Stream g) =

532 Stream $ \ctx stp yld ->

545 Stream $ \ctx stp yld ->

533− let run x = (runStream x) ctx stp yld

546+ let run x = (SC.runStream x) ctx stp yld

534 yield a Nothing = run $ f a

547 yield a Nothing = run $ f a

535 yield a (Just r) = run $ f a `par` go r

548 yield a (Just r) = run $ f a `par` go r

536 in g Nothing stp yield

549 in g Nothing stp yield

@@ -640,7 +653,7 @@ deriving instance (MonadError e m, MonadAsync m) => MonadError e (ParallelT m)

640deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (ParallelT m)

653deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (ParallelT m)

641deriving instance (MonadState s m, MonadAsync m) => MonadState s (ParallelT m)

654deriving instance (MonadState s m, MonadAsync m) => MonadState s (ParallelT m)

642

655

643−instance Streaming ParallelT where

656+instance IsStream ParallelT where

644 toStream = getParallelT

657 toStream = getParallelT

645 fromStream = ParallelT

658 fromStream = ParallelT

646

659

@@ -718,17 +731,17 @@ instance (MonadAsync m, Floating a) => Floating (ParallelT m a) where

718------------------------------------------------------------------------------

731------------------------------------------------------------------------------

719

732

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

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

721−zipWith :: Streaming t => (a -> b -> c) -> t m a -> t m b -> t m c

734+zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c

722zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

735zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

723 where

736 where

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

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

725 let merge a ra =

738 let merge a ra =

726 let yield2 b Nothing = yld (f a b) Nothing

739 let yield2 b Nothing = yld (f a b) Nothing

727 yield2 b (Just rb) = yld (f a b) (Just (go ra rb))

740 yield2 b (Just rb) = yld (f a b) (Just (go ra rb))

728− in (runStream my) Nothing stp yield2

741+ in (SC.runStream my) Nothing stp yield2

729 let yield1 a Nothing = merge a snil

742 let yield1 a Nothing = merge a snil

730 yield1 a (Just ra) = merge a ra

743 yield1 a (Just ra) = merge a ra

731− (runStream mx) Nothing stp yield1

744+ (SC.runStream mx) Nothing stp yield1

732

745

733-- | The applicative instance of 'ZipSerial' zips a number of streams serially

746-- | The applicative instance of 'ZipSerial' zips a number of streams serially

734-- i.e. it produces one element from each stream serially and then zips all

747-- i.e. it produces one element from each stream serially and then zips all

@@ -767,7 +780,7 @@ instance Monad m => Applicative (ZipSerial m) where

767 pure = ZipSerial . srepeat

780 pure = ZipSerial . srepeat

768 (<*>) = zipWith id

781 (<*>) = zipWith id

769

782

770−instance Streaming ZipSerial where

783+instance IsStream ZipSerial where

771 toStream = getZipSerial

784 toStream = getZipSerial

772 fromStream = ZipSerial

785 fromStream = ZipSerial

773

786

@@ -817,12 +830,12 @@ instance (Monad m, Floating a) => Floating (ZipSerial m a) where

817

830

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

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

819-- concurrently) using a pure zipping function.

832-- concurrently) using a pure zipping function.

820−zipAsyncWith :: (Streaming t, MonadAsync m)

833+zipAsyncWith :: (IsStream t, MonadAsync m)

821 => (a -> b -> c) -> t m a -> t m b -> t m c

834 => (a -> b -> c) -> t m a -> t m b -> t m c

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

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

823 ma <- async m1

836 ma <- async m1

824 mb <- async m2

837 mb <- async m2

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

838+ (SC.runStream (toStream (zipWith f ma mb))) Nothing stp yld

826

839

827-- | Like 'ZipSerial' but zips in parallel, it generates all the elements to

840-- | Like 'ZipSerial' but zips in parallel, it generates all the elements to

828-- be zipped concurrently.

841-- be zipped concurrently.

@@ -855,7 +868,7 @@ instance MonadAsync m => Applicative (ZipAsync m) where

855 pure = ZipAsync . srepeat

868 pure = ZipAsync . srepeat

856 (<*>) = zipAsyncWith id

869 (<*>) = zipAsyncWith id

857

870

858−instance Streaming ZipAsync where

871+instance IsStream ZipAsync where

859 toStream = getZipAsync

872 toStream = getZipAsync

860 fromStream = ZipAsync

873 fromStream = ZipAsync

861

874

@@ -904,7 +917,7 @@ instance (MonadAsync m, Floating a) => Floating (ZipAsync m a) where

904-------------------------------------------------------------------------------

917-------------------------------------------------------------------------------

905

918

906-- | Adapt one streaming type to another.

919-- | Adapt one streaming type to another.

907−adapt :: (Streaming t1, Streaming t2) => t1 m a -> t2 m a

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

908adapt = fromStream . toStream

921adapt = fromStream . toStream

909

922

910-- | Interpret an ambiguously typed stream as 'SerialT'.

923-- | Interpret an ambiguously typed stream as 'SerialT'.

@@ -935,39 +948,39 @@ zippingAsync x = x

935-- Running Streams, convenience functions specialized to types

948-- Running Streams, convenience functions specialized to types

936-------------------------------------------------------------------------------

949-------------------------------------------------------------------------------

937

950

938−-- | Same as @runStreaming . serially@.

951+-- | Same as @runStream . serially@.

939runSerialT :: Monad m => SerialT m a -> m ()

952runSerialT :: Monad m => SerialT m a -> m ()

940−runSerialT = runStreaming

953+runSerialT = runStream

941

954

942-- | Same as @runSerialT@.

955-- | Same as @runSerialT@.

943{-# Deprecated runStreamT "Please use runSerialT instead." #-}

956{-# Deprecated runStreamT "Please use runSerialT instead." #-}

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

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

945runStreamT = runSerialT

958runStreamT = runSerialT

946

959

947−-- | Same as @runStreaming . interleaving@.

960+-- | Same as @runStream . interleaving@.

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

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

949−runInterleavedT = runStreaming

962+runInterleavedT = runStream

950

963

951−-- | Same as @runStreaming . asyncly@.

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

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

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

953−runAsyncT = runStreaming

966+runAsyncT = runStream

954

967

955−-- | Same as @runStreaming . parallely@.

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

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

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

957−runParallelT = runStreaming

970+runParallelT = runStream

958

971

959−-- | Same as @runStreaming . zipping@.

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

960runZipSerial :: Monad m => ZipSerial m a -> m ()

973runZipSerial :: Monad m => ZipSerial m a -> m ()

961−runZipSerial = runStreaming

974+runZipSerial = runStream

962

975

963{-# Deprecated runZipStream "Please use runZipSerial instead." #-}

976{-# Deprecated runZipStream "Please use runZipSerial instead." #-}

964-- | Same as ZipSerial.

977-- | Same as ZipSerial.

965runZipStream :: Monad m => ZipSerial m a -> m ()

978runZipStream :: Monad m => ZipSerial m a -> m ()

966runZipStream = runZipSerial

979runZipStream = runZipSerial

967

980

968−-- | Same as @runStreaming . zippingAsync@.

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

969runZipAsync :: Monad m => ZipAsync m a -> m ()

982runZipAsync :: Monad m => ZipAsync m a -> m ()

970−runZipAsync = runStreaming

983+runZipAsync = runStream

971

984

972------------------------------------------------------------------------------

985------------------------------------------------------------------------------

973-- Sum Style Composition

986-- Sum Style Composition

@@ -991,7 +1004,7 @@ infixr 5 <=>

991-- accumulate too many partially drained streams. To be clear, it can combine

1004-- accumulate too many partially drained streams. To be clear, it can combine

992-- infinite streams but not infinite number of streams.

1005-- infinite streams but not infinite number of streams.

993{-# INLINE (<=>) #-}

1006{-# INLINE (<=>) #-}

994−(<=>) :: Streaming t => t m a -> t m a -> t m a

1007+(<=>) :: IsStream t => t m a -> t m a -> t m a

995m1 <=> m2 = fromStream $ interleave (toStream m1) (toStream m2)

1008m1 <=> m2 = fromStream $ interleave (toStream m1) (toStream m2)

996

1009

997-- | Demand driven concurrent composition. In contrast to '<|>' this operator

1010-- | Demand driven concurrent composition. In contrast to '<|>' this operator

@@ -1011,7 +1024,7 @@ m1 <=> m2 = fromStream $ interleave (toStream m1) (toStream m2)

1011-- operator corresponds to the 'AsyncT' type for product style composition.

1024-- operator corresponds to the 'AsyncT' type for product style composition.

1012--

1025--

1013{-# INLINE (<|) #-}

1026{-# INLINE (<|) #-}

1014−(<|) :: (Streaming t, MonadAsync m) => t m a -> t m a -> t m a

1027+(<|) :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

1015m1 <| m2 = fromStream $ parLeft (toStream m1) (toStream m2)

1028m1 <| m2 = fromStream $ parLeft (toStream m1) (toStream m2)

1016

1029

1017------------------------------------------------------------------------------

1030------------------------------------------------------------------------------

@@ -1028,7 +1041,7 @@ m1 <| m2 = fromStream $ parLeft (toStream m1) (toStream m2)

1028--

1041--

1029-- @foldWith (<>) $ map return [1..3]@

1042-- @foldWith (<>) $ map return [1..3]@

1030{-# INLINABLE foldWith #-}

1043{-# INLINABLE foldWith #-}

1031−foldWith :: (Streaming t, Foldable f)

1044+foldWith :: (IsStream t, Foldable f)

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

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

1033foldWith f = foldr f nil

1046foldWith f = foldr f nil

1034

1047

@@ -1038,13 +1051,13 @@ foldWith f = foldr f nil

1038--

1051--

1039-- @foldMapWith (<>) return [1..3]@

1052-- @foldMapWith (<>) return [1..3]@

1040{-# INLINABLE foldMapWith #-}

1053{-# INLINABLE foldMapWith #-}

1041−foldMapWith :: (Streaming t, Foldable f)

1054+foldMapWith :: (IsStream t, Foldable f)

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

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

1043foldMapWith f g = foldr (f . g) nil

1056foldMapWith f g = foldr (f . g) nil

1044

1057

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

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

1046-- monadic streaming function is the last argument.

1059-- monadic streaming function is the last argument.

1047{-# INLINABLE forEachWith #-}

1060{-# INLINABLE forEachWith #-}

1048−forEachWith :: (Streaming t, Foldable f)

1061+forEachWith :: (IsStream t, Foldable f)

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

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

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

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

src/Streamly/Tutorial.hs+9−9

@@ -174,7 +174,7 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

174--

175-- 'toList' runs a stream computation and collects the result stream in a list

176-- in the underlying monad. 'toList' is a polymorphic function that works on

177−-- multiple stream types belonging to the class 'Streaming'. Therefore, before

177+-- multiple stream types belonging to the class 'IsStream'. Therefore, before

178-- you run a stream you need to tell how you want to interpret the stream by

179-- using one of the stream type combinators ('serially', 'asyncly', 'parallely'

180-- etc.). The combinator 'serially' is equivalent to annotating the type as @::

@@ -386,7 +386,7 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

386-- The previous section discussed ways to merge the elements of two streams

387-- without doing any transformation on them. In this section we will explore

388-- how to transform streams using 'Functor', 'Applicative' or 'Monad' style

389−-- compositions. The applicative and monad composition of all 'Streaming' types

389+-- compositions. The applicative and monad compositions of all 'IsStream' types

390-- behave exactly the same way as a list transformer. For simplicity of

391-- illustration we are using streams of pure values in the following examples.

392-- However, the real application of streams arises when these streams are

@@ -394,7 +394,7 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

394

395-- $monad

396--

397−-- In functional programmer's parlance the 'Monad' instance of 'Streaming'

397+-- In functional programmer's parlance the 'Monad' instance of 'IsStream'

398-- types implement non-determinism, exploring all possible combination of

399-- choices from both the streams. From an imperative programmer's point of view

400-- it behaves like nested loops i.e. for each element in the first stream and

@@ -573,7 +573,7 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

573--

574-- The streamly code is usually written in a way that is agnostic of the

575-- specific monadic composition type. We use a polymorphic type with a

576−-- 'Streaming' type class constraint. When running the stream we can choose the

576+-- 'IsStream' type class constraint. When running the stream we can choose the

577-- specific mode of composition. For example look at the following code.

578--

579-- @

@@ -581,7 +581,7 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

581-- import "Streamly.Prelude"

582--

583--

584−-- composed :: 'Streaming' t => t m a

584+-- composed :: 'IsStream' t => t m a

585-- composed = do

586-- sz <- sizes

587-- cl <- colors

@@ -608,10 +608,10 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

608-- this:

609--

610-- @

611−-- main = 'runStreaming' $ 'serially' $ composed

611+-- main = 'runStream' $ 'serially' $ composed

612−-- main = 'runStreaming' $ 'asyncly' $ composed

612+-- main = 'runStream' $ 'asyncly' $ composed

613−-- main = 'runStreaming' $ 'interleaving' $ composed

613+-- main = 'runStream' $ 'interleaving' $ composed

614−-- main = 'runStreaming' $ 'parallely' $ composed

614+-- main = 'runStream' $ 'parallely' $ composed

615-- @

616--

617-- As an exercise try to figure out the output of this code for each mode of

test/Main.hs+3−3

@@ -28,9 +28,9 @@ main = hspec $ do

28 describe "Runners" $ do

29 -- XXX move these to property tests

30 it "simple serially" $

31− (runStreaming . serially) (return (0 :: Int)) `shouldReturn` ()

31+ (runStream . serially) (return (0 :: Int)) `shouldReturn` ()

32 it "simple serially with IO" $

33− (runStreaming . serially) (liftIO $ putStrLn "hello") `shouldReturn` ()

33+ (runStream . serially) (liftIO $ putStrLn "hello") `shouldReturn` ()

35 describe "Empty" $ do

36 it "Monoid - mempty" $

@@ -445,7 +445,7 @@ loops f tsrt hsrt = do

445 return x `f` (if x < 3 then loopTail (x + 1) else empty)

446

447bindAndComposeSimple

448− :: (Streaming t, Alternative (t IO), Monad (t IO))

448+ :: (IsStream t, Alternative (t IO), Monad (t IO))

449 => (forall a. Ord a => t IO a -> IO [a])

450 -> (t IO Int -> t IO Int -> t IO Int)

451 -> Spec

test/Prop.hs+15−15

@@ -31,7 +31,7 @@ equals eq stream list = do

31 assert (stream `eq` list)

33constructWithReplicateM

34− :: (Streaming t)

34+ :: IsStream t

35 => (t IO Int -> t IO Int)

36 -> Word8

37 -> Property

@@ -43,7 +43,7 @@ constructWithReplicateM op len =

43 equals (==) stream list

45transformFromList

46− :: (Streaming t)

46+ :: IsStream t

47 => ([Int] -> t IO Int)

48 -> ([Int] -> [Int] -> Bool)

49 -> ([Int] -> [Int])

@@ -57,7 +57,7 @@ transformFromList constr eq listOp op a =

57 equals eq stream list

59foldFromList

60− :: (Streaming t)

60+ :: IsStream t

61 => ([Int] -> t IO Int)

62 -> (t IO Int -> t IO Int)

63 -> ([Int] -> [Int] -> Bool)

@@ -92,7 +92,7 @@ elemOp constr op streamOp listOp (x, xs) =

92 equals (==) stream list

94functorOps

95− :: (Streaming t, Functor (t IO))

95+ :: (IsStream t, Functor (t IO))

96 => ([Int] -> t IO Int)

97 -> String

98 -> (t IO Int -> t IO Int)

@@ -103,7 +103,7 @@ functorOps constr desc t eq = do

103 prop (desc ++ " fmap (+1)") $ transformFromList constr eq (fmap (+1)) $ t . (fmap (+1))

104

105transformOps

106− :: Streaming t

106+ :: IsStream t

107 => ([Int] -> t IO Int)

108 -> String

109 -> (t IO Int -> t IO Int)

@@ -155,7 +155,7 @@ wrapMaybe f =

155 else Just (f x)

156

157eliminationOps

158− :: Streaming t

158+ :: IsStream t

159 => ([Int] -> t IO Int)

160 -> String

161 -> (t IO Int -> t IO Int)

@@ -177,7 +177,7 @@ eliminationOps constr desc t = do

177-- head/tail/last may depend on the order in case of parallel streams

178-- so we test these only for serial streams.

179serialEliminationOps

180− :: Streaming t

180+ :: IsStream t

181 => ([Int] -> t IO Int)

182 -> String

183 -> (t IO Int -> t IO Int)

@@ -192,7 +192,7 @@ serialEliminationOps constr desc t = do

192 prop (desc ++ " last") $ eliminateOp constr (wrapMaybe last) $ A.last . t

193

194transformOpsWord8

195− :: Streaming t

195+ :: IsStream t

196 => ([Word8] -> t IO Word8)

197 -> String

198 -> (t IO Word8 -> t IO Word8)

@@ -203,7 +203,7 @@ transformOpsWord8 constr desc t = do

203

204-- XXX concatenate streams of multiple elements rather than single elements

205semigroupOps

206− :: ( Streaming t, MonadPlus (t IO)

206+ :: (IsStream t, MonadPlus (t IO)

207

208#if __GLASGOW_HASKELL__ < 804

209 , Semigroup (t IO Int)

@@ -219,7 +219,7 @@ semigroupOps desc t = do

219 prop (desc ++ " <|") $ foldFromList (foldMapWith (<|) return) t sortEq

220

221applicativeOps

222− :: (Streaming t, Applicative (t IO))

222+ :: (IsStream t, Applicative (t IO))

223 => ([Int] -> t IO Int)

224 -> (t IO (Int, Int) -> t IO (Int, Int))

225 -> ([(Int, Int)] -> [(Int, Int)] -> Bool)

@@ -231,7 +231,7 @@ applicativeOps constr t eq (a, b) = monadicIO $ do

231 equals eq stream list

232

233zipApplicative

234− :: (Streaming t, Applicative (t IO))

234+ :: (IsStream t, Applicative (t IO))

235 => ([Int] -> t IO Int)

236 -> (t IO (Int, Int) -> t IO (Int, Int))

237 -> ([(Int, Int)] -> [(Int, Int)] -> Bool)

@@ -247,7 +247,7 @@ zipApplicative constr t eq (a, b) = monadicIO $ do

247 equals eq stream3 list

248

249zipMonadic

250− :: (Streaming t, Monad (t IO))

250+ :: (IsStream t, Monad (t IO))

251 => ([Int] -> t IO Int)

252 -> (t IO (Int, Int) -> t IO (Int, Int))

253 -> ([(Int, Int)] -> [(Int, Int)] -> Bool)

@@ -268,7 +268,7 @@ zipMonadic constr t eq (a, b) =

268 equals eq stream2 list

269

270monadThen

271− :: (Streaming t, Monad (t IO))

271+ :: (IsStream t, Monad (t IO))

272 => ([Int] -> t IO Int)

273 -> (t IO Int -> t IO Int)

274 -> ([Int] -> [Int] -> Bool)

@@ -280,7 +280,7 @@ monadThen constr t eq (a, b) = monadicIO $ do

280 equals eq stream list

281

282monadBind

283− :: (Streaming t, Monad (t IO))

283+ :: (IsStream t, Monad (t IO))

284 => ([Int] -> t IO Int)

285 -> (t IO Int -> t IO Int)

286 -> ([Int] -> [Int] -> Bool)

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

309 A.toList . serially . (A.take 100) $ A.iterateM addM (0 :: Int)

310 `shouldReturn` (take 100 $ iterate (+ 1) 0)

311

312− let folded :: Streaming t => [a] -> t IO a

312+ let folded :: IsStream t => [a] -> t IO a

313 folded = adapt . serially . (\xs ->

314 case xs of

315 [x] -> return x -- singleton stream case