9dfce42 · streamly · git.iko.soy

git.iko.soy/streamly

Merge pull request #57 from composewell/is-stream

rename Streaming to IsStream and runStreaming to runStream

commit9dfce42e7593ad0026947b236fea4112dcceec17 authorHarendra Kumar <harendra.kumar@gmail.com> dateSun 15 Apr 2018, 17:54 (9 years ago) parent4d99ede 408fef1

8 files changed+139−121

compare with master

unifiedsplit

Changelog.md+2 benchmark/Main.hs+1−1 src/Streamly.hs+6−4 src/Streamly/Prelude.hs+41−40 src/Streamly/Streams.hs+62−49 src/Streamly/Tutorial.hs+9−9 test/Main.hs+3−3 test/Prop.hs+15−15

Changelog.md+2

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

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

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

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

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

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

1113

1214## 0.1.2

1315

benchmark/Main.hs+1−1

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

107107

108108{-# INLINE streamly_basic #-}

109109streamly_basic

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

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

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

112112 -> (t IO Int -> t IO Int -> t IO Int)

113113 -> IO Int

src/Streamly.hs+6−4

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

1616 -- $overview

1717

1818 MonadAsync

19− , Streaming

19+ , IsStream

2020

2121 -- * Product Style Composition

2222 -- $product

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

4949 , adapt

5050

5151 -- * Running Streams

52− , runStreaming

52+ , runStream

5353 , runSerialT

5454 , runInterleavedT

5555 , runAsyncT

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

7272 , MonadTrans (..)

7373

7474 -- * Deprecated

75+ , Streaming

76+ , runStreaming

7577 , runStreamT

7678 , runZipStream

7779 , StreamT

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

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

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

126128-- 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

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

129131-- stream type.

130132--

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

213215--

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

215217-- 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

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

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

219221-- 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

9393import qualified System.IO as IO

9494

9595import Streamly.Core

96−import Streamly.Streams

96+import Streamly.Streams hiding (runStream)

97+

9798------------------------------------------------------------------------------

9899-- Construction

99100------------------------------------------------------------------------------

100101

101102-- | 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

103104unfoldr step = fromStream . go

104105 where

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

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

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

109110

110111-- | 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

112113unfoldrM step = fromStream . go

113114 where

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

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

120121-- XXX need eachInterleaved, eachAsync, eachParallel

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

122123{-# INLINE each #-}

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

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

124125each = Prelude.foldr cons nil

125126

126127-- | 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

128129iterate step = fromStream . go

129130 where

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

131132

132133-- | 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

134135iterateM step = fromStream . go

135136 where

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

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

138139 yld s (Just (go a))

139140

140141-- | 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

142143fromHandle h = fromStream go

143144 where

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

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

156157-- Parallel variants of folds?

157158

158159-- | 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

160161foldr step acc m = go (toStream m)

161162 where

162163 go m1 =

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

167168

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

169170{-# 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

171172foldrM step acc m = go (toStream m)

172173 where

173174 go m1 =

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

178179

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

180181{-# 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

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

183184 where

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

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

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

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

194195{-# INLINE foldl #-}

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

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

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

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

198199 where

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

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

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

220221-- 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)

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

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

224225 where

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

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

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

233234-- 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))

235236uncons m =

236237 let stop = return Nothing

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

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

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

240241

241242-- | 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 ()

243244toHandle h m = go (toStream m)

244245 where

245246 go m1 =

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

254255

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

256257{-# INLINABLE toList #-}

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

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

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

259260

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

261262{-# INLINE take #-}

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

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

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

264265 where

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

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

269270

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

271272{-# 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

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

274275 where

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

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

281282

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

283284{-# 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

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

286287 where

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

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

292293 in (runStream m1) ctx stp yield

293294

294295-- | 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

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

297298 where

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

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

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

307308-- take the rest of the stream.

308309{-# 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

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

311312 where

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

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

317318 in (runStream m1) ctx stp yield

318319

319320-- | 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

321322all p m = go (toStream m)

322323 where

323324 go m1 =

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

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

329330

330331-- | 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

332333any p m = go (toStream m)

333334 where

334335 go m1 =

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

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

340341

341342-- | 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

343344sum = foldl (+) 0 id

344345

345346-- | 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

347348product = foldl (*) 1 id

348349

349350-- | 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)

351352head m =

352353 let stop = return Nothing

353354 yield a _ = return (Just a)

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

355356

356357-- | 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))

358359tail m =

359360 let stop = return Nothing

360361 yield _ Nothing = return $ Just nil

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

363364

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

365366{-# 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)

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

368369

369370-- | 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

371372null m =

372373 let stop = return True

373374 yield _ _ = return False

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

375376

376377-- | 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

378379elem e m = go (toStream m)

379380 where

380381 go m1 =

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

384385 in (runStream m1) Nothing stop yield

385386

386387-- | 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

388389notElem e m = go (toStream m)

389390 where

390391 go m1 =

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

394395 in (runStream m1) Nothing stop yield

395396

396397-- | 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

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

399400

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

401402-- 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

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

404405 where

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

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

412413

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

414415-- | 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)

416417minimum m = go Nothing (toStream m)

417418 where

418419 go r m1 =

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

427428

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

429430-- | 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)

431432maximum m = go Nothing (toStream m)

432433 where

433434 go r m1 =

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

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

450451-- applied on the element.

451452{-# 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

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

454455 where

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

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

460461

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

462463-- 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 ()

464465mapM_ f m = go (toStream m)

465466 where

466467 go m1 =

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

471472

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

473474-- 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

475476sequence m = fromStream $ go (toStream m)

476477 where

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

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

481482 in (runStream m1) Nothing stop yield

482483

483484-- | 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

485486replicateM n m = fromStream $ go n

486487 where

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

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

494495------------------------------------------------------------------------------

495496

496497-- | 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

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

499500 where

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

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

514515

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

516517-- concurrently) using a monadic zipping function.

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

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

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

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

520521 ma <- async m1

src/Streamly/Streams.hs+62−49

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

1+{-# LANGUAGE ConstraintKinds #-}

12{-# LANGUAGE FlexibleContexts #-}

23{-# LANGUAGE FlexibleInstances #-}

34{-# LANGUAGE GeneralizedNewtypeDeriving#-}

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

1819--

1920module Streamly.Streams

2021 (

21− Streaming (..)

22+ IsStream (..)

23+ , Streaming -- deprecated

2224 , MonadAsync

2325

2426 -- * SVars

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

3840

3941 -- * Elimination

4042 , streamFold

41− , runStreaming

43+ , runStream

44+ , runStreaming -- deprecated

4245 , toSVar

4346

4447 -- * Transformation

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

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

101104import Data.Semigroup (Semigroup(..))

102105import Prelude hiding (zipWith)

103−import Streamly.Core

106+import Streamly.Core hiding (runStream)

107+import qualified Streamly.Core as SC

104108

105109------------------------------------------------------------------------------

106110-- Types that can behave as a Stream

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

108112

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

110114-- some monad 'm'.

111−class Streaming t where

115+class IsStream t where

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

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

114118

119+-- | Same as 'IsStream'.

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

121+type Streaming = IsStream

122+

115123------------------------------------------------------------------------------

116124-- Constructing a stream

117125------------------------------------------------------------------------------

118126

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

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

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

121129nil = fromStream snil

122130

123131infixr 5 `cons`

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

130138-- > (toList . serially) stream

131139-- [1,2,3]

132140-- @

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

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

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

135143

136144infixr 5 .:

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

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

162170-- latter.

163171--

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

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

165173(.:) = cons

166174

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

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

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

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

171179-- represent an "empty" stream.

172−streamBuild :: Streaming t

180+streamBuild :: IsStream t

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

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

175183 -> m r

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

181189 in k sv yield stp

182190

183191-- | 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

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

186194

187195-- | 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

189197fromSVar sv = fromStream $ fromStreamVar sv

190198

191199------------------------------------------------------------------------------

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

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

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

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

198−streamFold :: Streaming t

206+streamFold :: IsStream t

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

200208streamFold sv step blank m =

201209 let yield a Nothing = step a Nothing

202210 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

204212

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

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

207−runStreaming m = go (toStream m)

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

215+runStream m = go (toStream m)

208216 where

209217 go m1 =

210218 let stop = return ()

211219 yield _ Nothing = stop

212220 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

214227

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

216229-- 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 ()

218231toSVar sv m = toStreamVar sv (toStream m)

219232

220233------------------------------------------------------------------------------

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

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

229242-- will always return 'empty'.

230243

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)

232245async m = do

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

234247 return $ fromSVar sv

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

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

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

283296

284−instance Streaming SerialT where

297+instance IsStream SerialT where

285298 toStream = getSerialT

286299 fromStream = SerialT

287300

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

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

302315 return = pure

303316 (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

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

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

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

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

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

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

404417

405−instance Streaming InterleavedT where

418+instance IsStream InterleavedT where

406419 toStream = getInterleavedT

407420 fromStream = InterleavedT

408421

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

410423 return = pure

411424 (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

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

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

415428 `interleave`

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

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

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

518531

519−instance Streaming AsyncT where

532+instance IsStream AsyncT where

520533 toStream = getAsyncT

521534 fromStream = AsyncT

522535

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

530543 where

531544 go (Stream g) =

532545 Stream $ \ctx stp yld ->

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

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

534547 yield a Nothing = run $ f a

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

536549 in g Nothing stp yield

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

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

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

642655

643−instance Streaming ParallelT where

656+instance IsStream ParallelT where

644657 toStream = getParallelT

645658 fromStream = ParallelT

646659

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

718731------------------------------------------------------------------------------

719732

720733-- | 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

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

723736 where

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

725738 let merge a ra =

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

727740 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

729742 let yield1 a Nothing = merge a snil

730743 yield1 a (Just ra) = merge a ra

731− (runStream mx) Nothing stp yield1

744+ (SC.runStream mx) Nothing stp yield1

732745

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

734747-- 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

767780 pure = ZipSerial . srepeat

768781 (<*>) = zipWith id

769782

770−instance Streaming ZipSerial where

783+instance IsStream ZipSerial where

771784 toStream = getZipSerial

772785 fromStream = ZipSerial

773786

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

817830

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

819832-- concurrently) using a pure zipping function.

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

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

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

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

823836 ma <- async m1

824837 mb <- async m2

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

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

826839

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

828841-- be zipped concurrently.

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

855868 pure = ZipAsync . srepeat

856869 (<*>) = zipAsyncWith id

857870

858−instance Streaming ZipAsync where

871+instance IsStream ZipAsync where

859872 toStream = getZipAsync

860873 fromStream = ZipAsync

861874

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

904917-------------------------------------------------------------------------------

905918

906919-- | 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

908921adapt = fromStream . toStream

909922

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

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

935948-- Running Streams, convenience functions specialized to types

936949-------------------------------------------------------------------------------

937950

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

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

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

940−runSerialT = runStreaming

953+runSerialT = runStream

941954

942955-- | Same as @runSerialT@.

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

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

945958runStreamT = runSerialT

946959

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

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

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

949−runInterleavedT = runStreaming

962+runInterleavedT = runStream

950963

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

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

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

953−runAsyncT = runStreaming

966+runAsyncT = runStream

954967

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

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

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

957−runParallelT = runStreaming

970+runParallelT = runStream

958971

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

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

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

961−runZipSerial = runStreaming

974+runZipSerial = runStream

962975

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

964977-- | Same as ZipSerial.

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

966979runZipStream = runZipSerial

967980

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

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

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

970−runZipAsync = runStreaming

983+runZipAsync = runStream

971984

972985------------------------------------------------------------------------------

973986-- Sum Style Composition

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

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

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

9931006{-# 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

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

9961009

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

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

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

10121025--

10131026{-# 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

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

10161029

10171030------------------------------------------------------------------------------

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

10281041--

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

10301043{-# INLINABLE foldWith #-}

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

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

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

10331046foldWith f = foldr f nil

10341047

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

10381051--

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

10401053{-# INLINABLE foldMapWith #-}

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

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

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

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

10441057

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

10461059-- monadic streaming function is the last argument.

10471060{-# INLINABLE forEachWith #-}

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

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

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

10501063forEachWith 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))

174174--

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

176176-- 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

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

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

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

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

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

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

388388-- 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

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

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

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

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

394394

395395-- $monad

396396--

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

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

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

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

400400-- 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))

573573--

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

575575-- 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

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

578578--

579579-- @

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

581581-- import "Streamly.Prelude"

582582--

583583--

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

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

585585-- composed = do

586586-- sz <- sizes

587587-- cl <- colors

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

608608-- this:

609609--

610610-- @

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

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

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

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

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

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

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

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

615615-- @

616616--

617617-- 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

2828 describe "Runners" $ do

2929 -- XXX move these to property tests

3030 it "simple serially" $

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

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

3232 it "simple serially with IO" $

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

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

3434

3535 describe "Empty" $ do

3636 it "Monoid - mempty" $

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

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

446446

447447bindAndComposeSimple

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

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

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

450450 -> (t IO Int -> t IO Int -> t IO Int)

451451 -> Spec

test/Prop.hs+15−15

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

3131 assert (stream `eq` list)

3232

3333constructWithReplicateM

34− :: (Streaming t)

34+ :: IsStream t

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

3636 -> Word8

3737 -> Property

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

4343 equals (==) stream list

4444

4545transformFromList

46− :: (Streaming t)

46+ :: IsStream t

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

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

4949 -> ([Int] -> [Int])

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

5757 equals eq stream list

5858

5959foldFromList

60− :: (Streaming t)

60+ :: IsStream t

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

6262 -> (t IO Int -> t IO Int)

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

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

9292 equals (==) stream list

9393

9494functorOps

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

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

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

9797 -> String

9898 -> (t IO Int -> t IO Int)

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

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

104104

105105transformOps

106− :: Streaming t

106+ :: IsStream t

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

108108 -> String

109109 -> (t IO Int -> t IO Int)

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

155155 else Just (f x)

156156

157157eliminationOps

158− :: Streaming t

158+ :: IsStream t

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

160160 -> String

161161 -> (t IO Int -> t IO Int)

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

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

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

179179serialEliminationOps

180− :: Streaming t

180+ :: IsStream t

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

182182 -> String

183183 -> (t IO Int -> t IO Int)

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

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

193193

194194transformOpsWord8

195− :: Streaming t

195+ :: IsStream t

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

197197 -> String

198198 -> (t IO Word8 -> t IO Word8)

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

203203

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

205205semigroupOps

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

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

207207

208208#if __GLASGOW_HASKELL__ < 804

209209 , Semigroup (t IO Int)

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

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

220220

221221applicativeOps

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

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

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

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

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

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

231231 equals eq stream list

232232

233233zipApplicative

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

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

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

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

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

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

247247 equals eq stream3 list

248248

249249zipMonadic

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

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

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

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

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

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

268268 equals eq stream2 list

269269

270270monadThen

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

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

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

273273 -> (t IO Int -> t IO Int)

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

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

280280 equals eq stream list

281281

282282monadBind

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

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

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

285285 -> (t IO Int -> t IO Int)

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

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

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

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

311311

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

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

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

314314 case xs of

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