ad05792 · streamly · git.iko.soy

git.iko.soy/streamly

Deprecate Streamly module and move the exports to Streamly.Prelude.

- Deprecate and replace functions: foldWith => concatFoldableWith foldMapWith => concatMapFoldableWith forEachWith => concatForFoldableWith

commitad05792d0b361c4629fc3009f69e21988eeaae88 authorpranaysashank <pranaysashank@composewell.com> dateTue 8 Sep 2020, 12:55 (6 years ago) parente218fb9

73 files changed+819−385

compare with master

unifiedsplit

Changelog.md+9

@@ -28,6 +28,15 @@

2828 fully drained. Also, the resource acquisition and release is atomic with

2929 respect to async exceptions.

3030

31+### Deprecations

32+

33+* The `Streamly` module is now deprecated, its functionality is subsumed

34+ by `Streamly.Prelude`.

35+* Some functions from `Streamly` module have been renamed in `Streamly.Prelude` module:

36+ * `foldWith` has been replaced by `concatFoldableWith`

37+ * `foldMapWith` has been replaced by `concatMapFoldableWith`

38+ * `forEachWith` has been replaced by `concatForFoldableWith`

39+

3140## 0.7.2

3241

3342### Bug Fixes

README.md+14−20

@@ -106,9 +106,9 @@ Please see [INSTALL.md](./INSTALL.md) for instructions on how to use streamly

106106with your Haskell build tool or package manager. You may want to go through it

107107before jumping to run the examples below.

108108

109−The module `Streamly` provides just the core stream types, type casting and

110−concurrency control combinators. Stream construction, transformation, folding,

111−merging, zipping combinators are found in `Streamly.Prelude`.

109+The module `Streamly.Prelude` provides the core stream types and combinators

110+for type casting, controlling concurrency, stream construction, transformation,

111+folding, merging, and zipping.

112112

113113## Streaming Pipelines

114114

@@ -117,7 +117,6 @@ numbers from stdin, prints the squares of even numbers and exits if an even

117117number more than 9 is entered.

118118

119119``` haskell

120−import Streamly

121120import qualified Streamly.Prelude as S

122121import Data.Function ((&))

123122

@@ -150,7 +149,7 @@ each action is finished (asyncly):

150149

151150``` haskell

152151> let p n = threadDelay (n * 1000000) >> return n

153−> S.toList $ asyncly $ p 3 |: p 2 |: p 1 |: S.nil

152+> S.toList $ S.asyncly $ p 3 |: p 2 |: p 1 |: S.nil

154153[1,2,3]

155154```

156155

@@ -158,7 +157,7 @@ Use `aheadly` if you want speculative concurrency i.e. execute the actions in

158157the stream concurrently but consume the results in the specified order:

159158

160159``` haskell

161−> S.toList $ aheadly $ p 3 |: p 2 |: p 1 |: S.nil

160+> S.toList $ S.aheadly $ p 3 |: p 2 |: p 1 |: S.nil

162161[3,2,1]

163162```

164163

@@ -168,7 +167,7 @@ Monadic stream generation functions e.g. `unfoldrM`, `replicateM`, `repeatM`,

168167The following finishes in 10 seconds (100 seconds when serial):

169168

170169``` haskell

171−S.drain $ asyncly $ S.replicateM 10 $ p 10

170+S.drain $ S.asyncly $ S.replicateM 10 $ p 10

172171```

173172

174173## Concurrency Auto Scaling

@@ -177,7 +176,7 @@ Concurrency is auto-scaled i.e. more actions are executed concurrently if the

177176consumer is consuming the stream at a higher speed. How many tasks are executed

178177concurrently can be controlled by `maxThreads` and how many results are

179178buffered ahead of consumption can be controlled by `maxBuffer`. See the

180−documentation in the `Streamly` module.

179+documentation in the `Streamly.Prelude` module.

181180

182181## Concurrent Streaming Pipelines

183182

@@ -198,7 +197,7 @@ We can use `mapM` or `sequence` functions concurrently on a stream.

198197

199198``` haskell

200199> let p n = threadDelay (n * 1000000) >> return n

201−> S.drain $ aheadly $ S.mapM (\x -> p 1 >> print x) (serially $ repeatM (p 1))

200+> S.drain $ S.aheadly $ S.mapM (\x -> p 1 >> print x) (S.serially $ S.repeatM (p 1))

202201```

203202

204203## Serial and Concurrent Merging

@@ -209,11 +208,10 @@ stream, each with a delay of 1 to 10 seconds, respectively. Since all the

209208actions are concurrent we see one output printed every second:

210209

211210``` haskell

212−import Streamly

213211import qualified Streamly.Prelude as S

214212import Control.Concurrent (threadDelay)

215213

216−main = S.toList $ parallely $ foldMap delay [1..10]

214+main = S.toList $ S.parallely $ foldMap delay [1..10]

217215 where delay n = S.yieldM $ threadDelay (n * 1000000) >> print n

218216```

219217

@@ -222,7 +220,6 @@ below, see the tutorial for more ways. We use the following `delay`

222220function in the examples to demonstrate the concurrency aspects:

223221

224222``` haskell

225−import Streamly

226223import qualified Streamly.Prelude as S

227224import Control.Concurrent

228225

@@ -245,7 +242,7 @@ ThreadId 36: Delay 1

245242### Parallel

246243

247244``` haskell

248−main = S.drain . parallely $ delay 3 <> delay 2 <> delay 1

245+main = S.drain . S.parallely $ delay 3 <> delay 2 <> delay 1

249246```

250247```

251248ThreadId 42: Delay 1

@@ -258,7 +255,6 @@ ThreadId 40: Delay 3

258255The monad instance composes like a list monad.

259256

260257``` haskell

261−import Streamly

262258import qualified Streamly.Prelude as S

263259

264260loops = do

@@ -282,7 +278,7 @@ loop can run concurrently but the results are presented to the consumer of the

282278output in the same order as serial execution:

283279

284280``` haskell

285−main = S.drain $ aheadly $ loops

281+main = S.drain $ S.aheadly $ loops

286282```

287283

288284Different stream types execute the loop iterations in different ways. For

@@ -298,11 +294,10 @@ to concurrently generate squares of a stream of numbers and then concurrently

298294sum the square roots of all combinations of two streams:

299295

300296``` haskell

301−import Streamly

302297import qualified Streamly.Prelude as S

303298

304299main = do

305− s <- S.sum $ asyncly $ do

300+ s <- S.sum $ S.asyncly $ do

306301 -- Each square is performed concurrently, (<>) is concurrent

307302 x2 <- foldMap (\x -> return $ x * x) [1..100]

308303 y2 <- foldMap (\y -> return $ y * y) [1..100]

@@ -324,7 +319,7 @@ directories concurrently:

324319```haskell

325320import Control.Monad.IO.Class (liftIO)

326321import Path.IO (listDir, getCurrentDir) -- from path-io package

327−import Streamly (AsyncT, adapt)

322+import Streamly.Prelude (AsyncT, adapt)

328323import qualified Streamly.Prelude as S

329324

330325listDirRecursive :: AsyncT IO ()

@@ -350,10 +345,9 @@ For bounded concurrent streams, stream yield rate can be specified. For

350345example, to print hello once every second you can simply write this:

351346

352347``` haskell

353−import Streamly

354348import Streamly.Prelude as S

355349

356−main = S.drain $ asyncly $ avgRate 1 $ S.repeatM $ putStrLn "hello"

350+main = S.drain $ S.asyncly $ S.avgRate 1 $ S.repeatM $ putStrLn "hello"

357351```

358352

359353For some practical uses of rate control, see

benchmark/NanoBenchmarks.hs+1−1

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

66{-# LANGUAGE FlexibleContexts #-}

77{-# LANGUAGE ScopedTypeVariables #-}

9−import Streamly (SerialT)

9+import Streamly.Prelude (SerialT)

1010import Streamly.Internal.Data.SVar (MonadAsync)

1111

1212import qualified Streamly.Data.Array.Storable.Foreign as A

benchmark/Streamly/Benchmark/Data/ArrayOps.hs−1

@@ -21,7 +21,6 @@ import Control.Monad.IO.Class (MonadIO)

2121import Prelude (Bool, Int, Maybe(..), ($), (+), (.), (==), (>), undefined)

2222

2323import qualified Prelude as P

24−import qualified Streamly as S hiding (foldMapWith, runStream)

2524import qualified Streamly.Prelude as S

2625

2726#ifndef DATA_PRIM_ARRAY

benchmark/Streamly/Benchmark/Data/Fold.hs+1−3

@@ -17,7 +17,6 @@ import Prelude (IO, Int, Double, String, (>), (<*>), (<$>), (+), ($),

1717 (<=), Monad(..), (==), Maybe(..), (.), fromIntegral,

1818 compare, (>=), concat, seq)

1919

20−import qualified Streamly as S hiding (runStream)

2120import qualified Streamly.Prelude as S

2221import qualified Streamly.Internal.Data.Fold as FL

2322import qualified Streamly.Internal.Data.Pipe as Pipe

@@ -30,7 +29,6 @@ import qualified Streamly.Internal.Data.Fold as IFL

3029import qualified Streamly.Internal.Data.Stream.IsStream as IP

3130

3231import Gauge

33−import Streamly hiding (runStream)

3432import Streamly.Benchmark.Common

3533

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

@@ -52,7 +50,7 @@ source = sourceUnfoldrM

5250-- | Takes a fold method, and uses it with a default source.

5351{-# INLINE benchIOSink #-}

5452benchIOSink

55− :: (IsStream t, NFData b)

53+ :: (S.IsStream t, NFData b)

5654 => Int -> String -> (t IO Int -> IO b) -> Benchmark

5755benchIOSink value name f = bench name $ nfIO $ randomRIO (1,1) >>= f . source value

5856

benchmark/Streamly/Benchmark/Data/Parser.hs+2−3

@@ -24,14 +24,13 @@ import Prelude

2424import qualified Data.Traversable as TR

2525import qualified Data.Foldable as F

2626import qualified Control.Applicative as AP

27−import qualified Streamly as S hiding (runStream)

2827import qualified Streamly.Prelude as S

2928import qualified Streamly.Internal.Data.Fold as FL

3029import qualified Streamly.Internal.Data.Parser as PR

3130import qualified Streamly.Internal.Data.Stream.IsStream as IP

3231

3332import Gauge

34−import Streamly hiding (runStream)

33+import Streamly.Prelude (SerialT)

3534import Streamly.Benchmark.Common

3635

3736-------------------------------------------------------------------------------

@@ -52,7 +51,7 @@ sourceUnfoldrM value n = S.unfoldrM step n

5251-- | Takes a fold method, and uses it with a default source.

5352{-# INLINE benchIOSink #-}

5453benchIOSink

55− :: (IsStream t, NFData b)

54+ :: (S.IsStream t, NFData b)

5655 => Int -> String -> (t IO Int -> IO b) -> Benchmark

5756benchIOSink value name f =

5857 bench name $ nfIO $ randomRIO (1,1) >>= f . sourceUnfoldrM value

benchmark/Streamly/Benchmark/Data/Parser/ParserD.hs+2−3

@@ -22,14 +22,13 @@ import Prelude hiding (any, all, take, sequence, sequenceA, takeWhile)

2222import qualified Data.Traversable as TR

2323import qualified Data.Foldable as F

2424import qualified Control.Applicative as AP

25−import qualified Streamly as S hiding (runStream)

2625import qualified Streamly.Prelude as S

2726import qualified Streamly.Internal.Data.Fold as FL

2827import qualified Streamly.Internal.Data.Parser.ParserD as PR

2928import qualified Streamly.Internal.Data.Stream.IsStream as IP

3029

3130import Gauge

32−import Streamly hiding (runStream)

31+import Streamly.Prelude (SerialT, MonadAsync, IsStream)

3332import Streamly.Benchmark.Common

3433

3534-------------------------------------------------------------------------------

@@ -39,7 +38,7 @@ import Streamly.Benchmark.Common

3938-- XXX these can be moved to the common module

4039

4140{-# INLINE sourceUnfoldrM #-}

42−sourceUnfoldrM :: (S.IsStream t, S.MonadAsync m) => Int -> Int -> t m Int

41+sourceUnfoldrM :: (IsStream t, MonadAsync m) => Int -> Int -> t m Int

4342sourceUnfoldrM value n = S.unfoldrM step n

4443 where

4544 step cnt =

benchmark/Streamly/Benchmark/Data/Parser/ParserK.hs+2−3

@@ -23,7 +23,6 @@ import Prelude hiding (any, all, take, sequence, sequenceA, takeWhile)

2323import qualified Control.Applicative as AP

2424import qualified Data.Foldable as F

2525import qualified Data.Traversable as TR

26−import qualified Streamly as S hiding (runStream)

2726import qualified Streamly.Prelude as S

2827import qualified Streamly.Internal.Data.Fold as FL

2928import qualified Streamly.Internal.Data.Parser.ParserK.Types as PR

@@ -31,7 +30,7 @@ import qualified Streamly.Internal.Data.Parser.ParserD as PRD

3130import qualified Streamly.Internal.Data.Stream.IsStream as IP

3231

3332import Gauge

34−import Streamly hiding (runStream)

33+import Streamly.Prelude (SerialT, MonadAsync, IsStream)

3534import Streamly.Benchmark.Common

3635

3736-------------------------------------------------------------------------------

@@ -41,7 +40,7 @@ import Streamly.Benchmark.Common

4140-- XXX these can be moved to the common module

4241

4342{-# INLINE sourceUnfoldrM #-}

44−sourceUnfoldrM :: (S.IsStream t, S.MonadAsync m) => Int -> Int -> t m Int

43+sourceUnfoldrM :: (IsStream t, MonadAsync m) => Int -> Int -> t m Int

4544sourceUnfoldrM value n = S.unfoldrM step n

4645 where

4746 step cnt =

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

@@ -121,11 +121,11 @@ sourceFromFoldableM n =

121121

122122{-# INLINE sourceFoldMapWith #-}

123123sourceFoldMapWith :: Int -> Stream m Int

124−sourceFoldMapWith n = SP.foldMapWith S.serial S.yield [n..n+value]

124+sourceFoldMapWith n = SP.concatMapFoldableWith S.serial S.yield [n..n+value]

125125

126126{-# INLINE sourceFoldMapWithM #-}

127127sourceFoldMapWithM :: Monad m => Int -> Stream m Int

128−sourceFoldMapWithM n = SP.foldMapWith S.serial (S.yieldM . return) [n..n+value]

128+sourceFoldMapWithM n = SP.concatMapFoldableWith S.serial (S.yieldM . return) [n..n+value]

129129

130130-------------------------------------------------------------------------------

131131-- Elimination

@@ -484,8 +484,8 @@ o_1_space =

484484 , benchFold "fromFoldableM" toNull sourceFromFoldableM

485485

486486 -- appends

487− , benchFold "foldMapWith" toNull sourceFoldMapWith

488− , benchFold "foldMapWithM" toNull sourceFoldMapWithM

487+ , benchFold "concatMapFoldableWith" toNull sourceFoldMapWith

488+ , benchFold "concatMapFoldableWithM" toNull sourceFoldMapWithM

489489 ]

490490 , bgroup "elimination"

491491 [ benchFold "toNull" toNull sourceUnfoldrM

@@ -542,7 +542,7 @@ o_1_space =

542542 , benchIOSrc1 "concatMapRepl (sqrt n of sqrt n)"

543543 (concatMapRepl value2 value2)

544544

545− -- This is for comparison with foldMapWith

545+ -- This is for comparison with concatMapFoldableWith

546546 , benchIOSrc1 "concatMapWithId (n of 1) (fromFoldable)"

547547 (S.drain . S.concatMapBy S.serial id . sourceConcatMapId value)

548548

benchmark/Streamly/Benchmark/Prelude/Adaptive.hs−1

@@ -9,7 +9,6 @@ import Control.Concurrent (threadDelay)

99import Control.Monad (when)

1010import Control.Monad.IO.Class (liftIO)

1111import Gauge

12−import Streamly

1312import Streamly.Prelude as S

1413import System.Random (randomRIO)

1514

benchmark/Streamly/Benchmark/Prelude/Ahead.hs+1−1

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

88import Prelude hiding (mapM)

10−import Streamly (aheadly, serially, ahead, maxBuffer, maxThreads)

10+import Streamly.Prelude (aheadly, serially, ahead, maxBuffer, maxThreads)

1111import qualified Streamly.Prelude as S

1212

1313import Streamly.Benchmark.Common

benchmark/Streamly/Benchmark/Prelude/Async.hs+1−1

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

88import Prelude hiding (mapM)

10−import Streamly (asyncly, async, maxBuffer, maxThreads, serially)

10+import Streamly.Prelude (asyncly, async, maxBuffer, maxThreads, serially)

1111import qualified Streamly.Prelude as S

1212import qualified Streamly.Internal.Data.Stream.StreamK.Type as Internal

1313

benchmark/Streamly/Benchmark/Prelude/Concurrent.hs+1−1

@@ -10,7 +10,7 @@ import Control.Concurrent

1010import Control.Monad (when, replicateM)

1111

1212import Gauge

13−import Streamly

13+import Streamly.Prelude hiding (mapM_, replicateM)

1414import qualified Streamly.Prelude as S

1515

1616-------------------------------------------------------------------------------

benchmark/Streamly/Benchmark/Prelude/Parallel.hs+2−2

@@ -9,9 +9,9 @@

1010import Prelude hiding (mapM)

1111

12−import Streamly (SerialT, parallely, parallel, serially, maxBuffer, maxThreads)

12+import Streamly.Prelude

13+ ( SerialT, parallely, parallel, serially, maxBuffer, maxThreads)

1314

14−import qualified Streamly as S

1515import qualified Streamly.Prelude as S

1616import qualified Streamly.Internal.Data.Fold as FL

1717import qualified Streamly.Internal.Data.Stream.Parallel as Par

benchmark/Streamly/Benchmark/Prelude/Rate.hs+2−3

@@ -5,9 +5,8 @@

55-- License : BSD3

66-- Maintainer : streamly@composewell.com

8−import Streamly (asyncly, aheadly, maxThreads)

9−

10−import qualified Streamly as S

8+import Streamly.Prelude (asyncly, aheadly, maxThreads)

9+import qualified Streamly.Prelude as S

1110

1211import Streamly.Benchmark.Common

1312import Streamly.Benchmark.Prelude

benchmark/Streamly/Benchmark/Prelude/Serial.hs+4−2

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

2222{-# OPTIONS_GHC -fplugin Test.Inspection.Plugin #-}

2323#endif

2424

25+module Main where

26+

2527import Control.DeepSeq (NFData(..))

2628import Control.Monad (when)

2729import Control.Monad.IO.Class (MonadIO(..))

@@ -48,14 +50,14 @@ import Test.Inspection

4850import qualified Streamly.Internal.Data.Stream.StreamD as D

4951#endif

5052

51−import qualified Streamly as S

53+

5254import qualified Streamly.Prelude as S

5355import qualified Streamly.Internal.Data.Stream.IsStream as Internal

5456import qualified Streamly.Internal.Data.Fold as FL

5557import qualified Streamly.Internal.Data.Unfold as UF

5658

5759import Gauge

58−import Streamly

60+import Streamly.Prelude (SerialT, IsStream, serially, serial)

5961import Streamly.Benchmark.Common

6062import Streamly.Benchmark.Prelude

6163import Streamly.Internal.Data.Time.Units

benchmark/Streamly/Benchmark/Prelude/WAsync.hs+1−1

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

88import Prelude hiding (mapM)

10−import Streamly (wAsyncly, serially, wAsync, maxBuffer, maxThreads)

10+import Streamly.Prelude (wAsyncly, serially, wAsync, maxBuffer, maxThreads)

1111import qualified Streamly.Prelude as S

1212

1313import Streamly.Benchmark.Common

benchmark/Streamly/Benchmark/Prelude/WSerial.hs+1−1

@@ -19,7 +19,7 @@

1919{-# OPTIONS_GHC -fplugin Test.Inspection.Plugin #-}

2020#endif

2121

22−import Streamly

22+import Streamly.Prelude (wSerial, wSerially)

2323import qualified Streamly.Prelude as S

2424import qualified Streamly.Internal.Data.Stream.IsStream as Internal

2525import qualified Streamly.Internal.Data.Unfold as UF

benchmark/Streamly/Benchmark/Prelude/ZipAsync.hs+1−2

@@ -7,8 +7,7 @@

88{-# LANGUAGE FlexibleContexts #-}

10−import Streamly (serially)

11−import qualified Streamly as S

10+import Streamly.Prelude (serially)

1211import qualified Streamly.Prelude as S

1312

1413import Streamly.Benchmark.Common

benchmark/Streamly/Benchmark/Prelude/ZipSerial.hs+4−6

@@ -21,8 +21,7 @@

2121

2222import Prelude hiding (zip)

2323

24−import Streamly (zipSerially)

25−import qualified Streamly as S

24+import Streamly.Prelude (MonadAsync)

2625import qualified Streamly.Prelude as S

2726

2827import Streamly.Benchmark.Common

@@ -47,7 +46,7 @@ moduleName = "Prelude.ZipSerial"

4746-- XXX somehow copying this definition here instead of importing it performs

4847-- better. Need to investigate why.

4948{-# INLINE sourceUnfoldrM #-}

50−sourceUnfoldrM :: (S.IsStream t, S.MonadAsync m) => Int -> Int -> t m Int

49+sourceUnfoldrM :: (S.IsStream t, MonadAsync m) => Int -> Int -> t m Int

5150sourceUnfoldrM count start = S.unfoldrM step start

5251 where

5352 step cnt =

@@ -100,7 +99,7 @@ o_1_space_joining value =

10099o_1_space_mapping :: Int -> [Benchmark]

101100o_1_space_mapping value =

102101 [ bgroup "mapping"

103− [ benchIOSink value "fmap" $ fmapN zipSerially 1

102+ [ benchIOSink value "fmap" $ fmapN S.zipSerially 1

104103 ]

105104 ]

106105

@@ -113,7 +112,7 @@ o_1_space_outerProduct :: Int -> [Benchmark]

113112o_1_space_outerProduct value =

114113 [ bgroup "monad-outer-product"

115114 -- XXX needs fixing

116− [ benchIO "toNullAp" $ toNullAp value zipSerially

115+ [ benchIO "toNullAp" $ toNullAp value S.zipSerially

117116 ]

118117 ]

119118-}

@@ -140,4 +139,3 @@ main = do

140139 -- , o_1_space_outerProduct size

141140 ]

142141 ]

143−

benchmark/lib/Streamly/Benchmark/Common.hs+1−1

@@ -48,7 +48,7 @@ import System.Random (randomRIO)

4848

4949import qualified Streamly.Prelude as S

5050

51−import Streamly

51+import Streamly.Prelude (SerialT)

5252import Gauge

5353

5454-------------------------------------------------------------------------------

benchmark/lib/Streamly/Benchmark/Prelude.hs+9−7

@@ -16,11 +16,13 @@ import Control.Applicative (liftA2)

1616import Control.DeepSeq (NFData(..))

1717import Control.Exception (try)

1818import Data.Functor.Identity (Identity)

19+#if !(MIN_VERSION_base(4,11,0))

20+import Data.Semigroup (Semigroup((<>)))

21+#endif

1922import GHC.Exception (ErrorCall)

2023import System.Random (randomRIO)

2124

2225import qualified Data.Foldable as F

23−import qualified Streamly as S

2426import qualified Streamly.Prelude as S

2527import qualified Streamly.Internal.Data.Stream.IsStream as Internal

2628import qualified Streamly.Internal.Data.Pipe as Pipe

@@ -293,21 +295,21 @@ transformZipMapM t n =

293295-------------------------------------------------------------------------------

294296

295297{-# INLINE sourceFoldMapWith #-}

296−sourceFoldMapWith :: (S.IsStream t, S.Semigroup (t m Int))

298+sourceFoldMapWith :: (S.IsStream t, Semigroup (t m Int))

297299 => Int -> Int -> t m Int

298−sourceFoldMapWith value n = S.foldMapWith (S.<>) S.yield [n..n+value]

300+sourceFoldMapWith value n = S.concatMapFoldableWith (<>) S.yield [n..n+value]

299301

300302{-# INLINE sourceFoldMapWithStream #-}

301−sourceFoldMapWithStream :: (S.IsStream t, S.Semigroup (t m Int))

303+sourceFoldMapWithStream :: (S.IsStream t, Semigroup (t m Int))

302304 => Int -> Int -> t m Int

303−sourceFoldMapWithStream value n = S.foldMapWith (S.<>) S.yield

305+sourceFoldMapWithStream value n = S.concatMapFoldableWith (<>) S.yield

304306 $ (S.enumerateFromTo n (n + value) :: S.SerialT Identity Int)

305307

306308{-# INLINE sourceFoldMapWithM #-}

307−sourceFoldMapWithM :: (S.IsStream t, Monad m, S.Semigroup (t m Int))

309+sourceFoldMapWithM :: (S.IsStream t, Monad m, Semigroup (t m Int))

308310 => Int -> Int -> t m Int

309311sourceFoldMapWithM value n =

310− S.foldMapWith (S.<>) (S.yieldM . return) [n..n+value]

312+ S.concatMapFoldableWith (<>) (S.yieldM . return) [n..n+value]

311313

312314{-# INLINE sourceFoldMapM #-}

313315sourceFoldMapM :: (S.IsStream t, Monad m, Monoid (t m Int))

examples/AcidRain.hs+7−4

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

1+{-# LANGUAGE CPP #-}

12{-# LANGUAGE FlexibleContexts #-}

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

67-- This example is adapted from Gabriel Gonzalez's pipes-concurrency package.

78-- https://hackage.haskell.org/package/pipes-concurrency-2.0.8/docs/Pipes-Concurrent-Tutorial.html

9−import Streamly

10+#if !(MIN_VERSION_base(4,11,0))

11+import Data.Semigroup ((<>))

12+#endif

1013import Streamly.Prelude as S

1114import Control.Monad (void)

1215import Control.Monad.IO.Class (MonadIO(liftIO))

@@ -26,13 +29,13 @@ userAction = S.repeatM $ liftIO askUser

2629 _ -> putStrLn "Type potion or harm or quit" >> askUser

2730

2831acidRain :: MonadAsync m => SerialT m Event

29−acidRain = asyncly $ constRate 1 $ S.repeatM $ liftIO $ return $ Harm 1

32+acidRain = S.asyncly $ S.constRate 1 $ S.repeatM $ liftIO $ return $ Harm 1

3033

3134data Result = Check | Done

3235

3336runEvents :: (MonadAsync m, MonadState Int m) => SerialT m Result

3437runEvents = do

35− event <- userAction `parallel` acidRain

38+ event <- userAction `S.parallel` acidRain

3639 case event of

3740 Harm n -> modify (\h -> h - n) >> return Check

3841 Heal n -> modify (\h -> h + n) >> return Check

@@ -52,7 +55,7 @@ getStatus result =

5255

5356main :: IO ()

5457main = do

55− putStrLn "Your health is deteriorating due to acid rain,\

58+ putStrLn "Your health is deteriorating due to acid rain,\\

5659 \ type \"potion\" or \"quit\""

5760 let runGame = S.drainWhile (== Alive) $ S.mapM getStatus runEvents

5861 void $ runStateT runGame 60

examples/CirclingSquare.hs+2−3

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

99import Data.IORef

1010import Graphics.UI.SDL as SDL

11−import Streamly

1211import Streamly.Prelude as S

1312

1413------------------------------------------------------------------------------

@@ -78,6 +77,6 @@ main :: IO ()

7877main = do

7978 sdlInit

8079 cref <- newIORef (0,0)

81− S.drain $ asyncly $ constRate 40

80+ S.drain $ S.asyncly $ S.constRate 40

8281 $ S.repeatM (updateController cref)

83− `parallel` S.repeatM (updateDisplay cref)

82+ `S.parallel` S.repeatM (updateDisplay cref)

examples/ControlFlow.hs+7−2

@@ -1,4 +1,5 @@

1−{-# LANGUAGE FlexibleContexts #-}

1+{-# LANGUAGE CPP #-}

2+{-# LANGUAGE FlexibleContexts #-}

23{-# LANGUAGE ScopedTypeVariables #-}

45-------------------------------------------------------------------------------

@@ -14,8 +15,12 @@

1415-- This file provides an example where we enter a sequence of characters "x",

1516-- and "y" on separate lines, on the command line. When any other sequence is

1617-- entered the control flow short circuits at the first non-matching char and

18+

1719-- exits.

1820

21+#if !(MIN_VERSION_base(4,11,0))

22+import Data.Semigroup (Semigroup(..))

23+#endif

1924import Control.Concurrent (threadDelay)

2025import Control.Exception (catch, SomeException)

2126import Control.Monad

@@ -25,7 +30,7 @@ import Control.Monad.Trans.Class

2530import Control.Monad.Trans.Maybe

2631import Control.Monad.Trans.Except

2732import Control.Monad.Trans.Cont

28−import Streamly

33+import Streamly.Prelude

2934import qualified Streamly.Prelude as S

3035

3136-------------------------------------------------------------------------------

examples/EchoServer.hs+2−3

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

33import Data.Function ((&))

5−import Streamly

65import Streamly.Internal.Network.Socket (handleWithM)

76import Streamly.Network.Socket

@@ -11,8 +10,8 @@ import qualified Streamly.Prelude as S

1110

1211main :: IO ()

1312main =

14− serially (S.unfold TCP.acceptOnPort 8091)

15− & parallely . S.mapM (handleWithM echo)

13+ S.serially (S.unfold TCP.acceptOnPort 8091)

14+ & S.parallely . S.mapM (handleWithM echo)

1615 & S.drain

1716

1817 where

examples/FileSinkServer.hs+1−2

@@ -9,7 +9,6 @@ import Control.Monad.IO.Class (liftIO)

99import Network.Socket (close)

1010import System.Environment (getArgs)

1111

12−import Streamly

1312import Streamly.Unicode.Stream

1413import qualified Streamly.FileSystem.Handle as FH

1514import qualified Streamly.Data.Array.Storable.Foreign as A

@@ -26,7 +25,7 @@ main = do

2625 (\src -> S.fold (FH.write src)

2726 $ encodeLatin1Lax

2827 $ S.concatUnfold A.read

29− $ S.concatMapWith parallel use

28+ $ S.concatMapWith S.parallel use

3029 $ S.unfold TCP.acceptOnPort 8090)

3130

3231 where

examples/FromFileClient.hs+1−2

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

66import System.Environment (getArgs)

8−import Streamly

98import qualified Streamly.Prelude as S

109import qualified Streamly.Internal.FileSystem.Handle as IFH

1110import qualified Streamly.Internal.Network.Inet.TCP as TCP

@@ -18,4 +17,4 @@ main =

1817 withFile file ReadMode $ \src ->

1918 S.fold (TCP.writeChunks (127, 0, 0, 1) 8090)

2019 $ IFH.toChunks src

21− in getArgs >>= S.drain . parallely . S.mapM sendFile . S.fromList

20+ in getArgs >>= S.drain . S.parallely . S.mapM sendFile . S.fromList

examples/ListDir.hs+1−2

@@ -3,7 +3,6 @@ module Main (main) where

33import Data.Bifunctor (bimap)

44import Data.Function ((&))

55import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))

6−import Streamly (ahead)

87import qualified Streamly.Prelude as S

98import qualified Streamly.Internal.Data.Stream.IsStream as S

@@ -14,7 +13,7 @@ import qualified Streamly.Internal.FileSystem.Dir as Dir

1413main :: IO ()

1514main = do

1615 hSetBuffering stdout LineBuffering

17− S.mapM_ print $ S.iterateMapLeftsWith ahead listDir (S.yield $ (Left "."))

16+ S.mapM_ print $ S.iterateMapLeftsWith S.ahead listDir (S.yield $ (Left "."))

1817

1918 where

2019

examples/MergeSort.hs+1−1

@@ -10,7 +10,7 @@ import Data.Word

1010import System.Random (getStdGen, randoms)

1111import Data.List (sort)

1212

13−import Streamly

13+import Streamly.Prelude (Serial)

1414import qualified Streamly.Prelude as S

1515

1616getSorted :: Serial Word16

examples/Rate.hs+2−3

@@ -1,11 +1,10 @@

1−import Streamly

21import qualified Streamly.Prelude as S

32import qualified Streamly.Internal.Data.Stream.IsStream as Internal

54main :: IO ()

65main =

76 S.mapM_ print

8− $ asyncly

9− $ avgRate 1

7+ $ S.asyncly

8+ $ S.avgRate 1

109 $ Internal.timestamped

1110 $ S.repeatM (pure "tick")

examples/SearchQuery.hs+10−4

@@ -1,7 +1,13 @@

1−import Streamly

1+{-# LANGUAGE CPP #-}

3+#if !(MIN_VERSION_base(4,11,0))

4+import Data.Semigroup ((<>))

5+#endif

26import Streamly.Prelude (drain, nil, yieldM, (|:))

37import Network.HTTP.Simple

9+import qualified Streamly.Prelude as S

10+

511-- | Runs three search engine queries in parallel and prints the search engine

612-- names in the fastest first order.

713--

@@ -10,13 +16,13 @@ import Network.HTTP.Simple

1016main :: IO ()

1117main = do

1218 putStrLn "Using parallel stream construction"

13− drain . parallely $ google |: bing |: duckduckgo |: nil

19+ drain . S.parallely $ google |: bing |: duckduckgo |: nil

1420

1521 putStrLn "\nUsing parallel semigroup composition"

16− drain . parallely $ yieldM google <> yieldM bing <> yieldM duckduckgo

22+ drain . S.parallely $ yieldM google <> yieldM bing <> yieldM duckduckgo

1723

1824 putStrLn "\nUsing parallel applicative zip"

19− drain . zipAsyncly $

25+ drain . S.zipAsyncly $

2026 (,,) <$> yieldM google <*> yieldM bing <*> yieldM duckduckgo

2127

2228 where

examples/WordCount.hs+1−2

@@ -40,11 +40,10 @@ import Streamly.Internal.Unicode.Stream

4040 (DecodeState, DecodeError(..), CodePoint, decodeUtf8Either,

4141 resumeDecodeUtf8Either)

4242

43−import qualified Streamly as S

43+import qualified Streamly.Internal.Data.Stream.IsStream as S

4444import qualified Streamly.Unicode.Stream as S

4545import qualified Streamly.FileSystem.Handle as FH

4646import qualified Streamly.Data.Array.Storable.Foreign as A

47−import qualified Streamly.Prelude as S

4847import qualified Data.Vector.Storable.Mutable as V

4948

5049-------------------------------------------------------------------------------

src/Streamly.hs+29−13

@@ -51,11 +51,12 @@

5151-- please see the "Streamly.Tutorial" module and the examples directory in this

5252-- package.

5353

54+{-# LANGUAGE NoMonomorphismRestriction #-}

5455{-# OPTIONS_GHC -Wno-orphans #-}

5556

5657#include "inline.hs"

5758

58−module Streamly

59+module Streamly {-# DEPRECATED "Please use \"Streamly.Prelude\" instead." #-}

5960 (

6061 -- -- * Concepts Overview

6162 -- -- ** Streams

@@ -181,13 +182,12 @@ module Streamly

181182 -- finite container of streams. Note that these are just special cases of

182183 -- the more general 'concatMapWith' operation.

183184 --

184− , IP.foldWith

185− , IP.foldMapWith

186− , IP.forEachWith

185+ , foldWith

186+ , foldMapWith

187+ , forEachWith

187188

188189 -- * Re-exports

189190 , Semigroup (..)

190−

191191 -- * Deprecated

192192 , Streaming

193193 , runStream

@@ -206,14 +206,6 @@ module Streamly

206206 , zippingAsync

207207 , (<=>)

208208 , (<|)

209−

210− {-

211− -- * Deprecated/Moved

212− -- | These APIs have been moved to other modules

213− , foldWith

214− , foldMapWith

215− , forEachWith

216− -}

217209 )

218210where

219211

@@ -640,3 +632,27 @@ mkAsync = return . Async.mkAsync

640632-- which specific type you are converting from or to. If you see a an

641633-- @ambiguous type variable@ error then most likely you are using 'adapt'

642634-- unnecessarily on polymorphic code.

635+

636+-- | Same as 'Streamly.Prelude.concatFoldableWith'

637+--

638+-- @since 0.1.0

639+{-# DEPRECATED foldWith "Please use 'Streamly.Prelude.concatFoldableWith' instead." #-}

640+{-# INLINEABLE foldWith #-}

641+foldWith :: (IsStream t, Foldable f) => (t m a -> t m a -> t m a) -> f (t m a) -> t m a

642+foldWith = P.concatFoldableWith

643+

644+-- | Same as 'Streamly.Prelude.concatMapFoldableWith'

645+--

646+-- @since 0.1.0

647+{-# DEPRECATED foldMapWith "Please use 'Streamly.Prelude.concatMapFoldableWith' instead." #-}

648+{-# INLINEABLE foldMapWith #-}

649+foldMapWith :: (IsStream t, Foldable f) => (t m b -> t m b -> t m b) -> (a -> t m b) -> f a -> t m b

650+foldMapWith = P.concatMapFoldableWith

651+

652+-- | Same as 'Streamly.Prelude.concatForFoldableWith'

653+--

654+-- @since 0.1.0

655+{-# DEPRECATED forEachWith "Please use 'Streamly.Prelude.concatForFoldableWith' instead." #-}

656+{-# INLINEABLE forEachWith #-}

657+forEachWith :: (IsStream t, Foldable f) => (t m b -> t m b -> t m b) -> f a -> (a -> t m b) -> t m b

658+forEachWith = P.concatForFoldableWith

src/Streamly/Internal/Data/Pipe.hs+1−1

@@ -242,7 +242,7 @@ import Prelude

242242-- import qualified Data.Map.Strict as Map

243243-- import qualified Prelude

244244

245−-- import Streamly (MonadAsync, parallel)

245+-- import Streamly.Prelude (MonadAsync, parallel)

246246-- import Streamly.Data.Fold.Types (Fold(..))

247247import Streamly.Internal.Data.Pipe.Types

248248 (Pipe(..), PipeState(..), Step(..), zipWith, tee, map, compose)

src/Streamly/Internal/Data/SVar.hs+8−3

@@ -228,7 +228,6 @@ data WorkerInfo = WorkerInfo

228228 , workerLatencyStart :: IORef (Count, AbsTime)

229229 }

230230

231−

232231-- | Specifies the stream yield rate in yields per second (@Hertz@).

233232-- We keep accumulating yield credits at 'rateGoal'. At any point of time we

234233-- allow only as many yields as we have accumulated as per 'rateGoal' since the

@@ -248,7 +247,9 @@ data WorkerInfo = WorkerInfo

248247-- If the 'rateGoal' is 0 or negative the stream never yields a value.

249248-- If the 'rateBuffer' is 0 or negative we do not attempt to recover.

250249--

251−-- @since 0.5.0

250+-- /Since: 0.5.0 ("Streamly")/

251+--

252+-- @since 0.8.0

252253data Rate = Rate

253254 { rateLow :: Double -- ^ The lower rate limit

254255 , rateGoal :: Double -- ^ The target rate we want to achieve

@@ -897,11 +898,15 @@ withDiagMVar sv label action =

897898-- Spawning threads

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

899900

901+-- /Since: 0.8.0 ("Streamly.Prelude")/

902+--

900903-- | A monad that can perform concurrent or parallel IO operations. Streams

901904-- that can be composed concurrently require the underlying monad to be

902905-- 'MonadAsync'.

903906--

904−-- @since 0.1.0

907+-- /Since: 0.1.0 ("Streamly")/

908+--

909+-- @since 0.8.0

905910type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)

906911

907912-- When we run computations concurrently, we completely isolate the state of

src/Streamly/Internal/Data/Stream/Ahead.hs+15−7

@@ -596,7 +596,9 @@ infixr 6 `ahead`

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

597597-- Merges two streams sequentially but with concurrent lookahead.

598598--

599−-- @since 0.3.0

599+-- /Since: 0.3.0 ("Streamly")/

600+--

601+-- @since 0.8.0

600602{-# INLINE ahead #-}

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

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

@@ -628,12 +630,12 @@ consMAhead m r = fromStream $ K.yieldM m `ahead` (toStream r)

628630-- container of streams.

629631--

630632-- @

631−-- import "Streamly"

633+-- import "Streamly.Prelude"

632634-- import qualified "Streamly.Prelude" as S

633635-- import Control.Concurrent

634636--

635637-- main = do

636−-- xs \<- S.'toList' . 'aheadly' $ (p 1 |: p 2 |: nil) <> (p 3 |: p 4 |: nil)

638+-- xs \<- S.'toList' . S.'aheadly' $ (p 1 |: p 2 |: nil) <> (p 3 |: p 4 |: nil)

637639-- print xs

638640-- where p n = threadDelay 1000000 >> return n

639641-- @

@@ -650,7 +652,7 @@ consMAhead m r = fromStream $ K.yieldM m `ahead` (toStream r)

650652-- 'SerialT'.

651653--

652654-- @

653−-- main = S.drain . 'aheadly' $ do

655+-- main = S.drain . S.'aheadly' $ do

654656-- n <- return 3 \<\> return 2 \<\> return 1

655657-- S.yieldM $ do

656658-- threadDelay (n * 1000000)

@@ -662,19 +664,25 @@ consMAhead m r = fromStream $ K.yieldM m `ahead` (toStream r)

662664-- ThreadId 38: Delay 3

663665-- @

664666--

665−-- @since 0.3.0

667+-- /Since: 0.3.0 ("Streamly")/

668+--

669+-- @since 0.8.0

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

667671 deriving (MonadTrans)

668672

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

670674-- 'AheadT' documentation for more details.

671675--

672−-- @since 0.3.0

676+-- /Since: 0.3.0 ("Streamly")/

677+--

678+-- @since 0.8.0

673679type Ahead = AheadT IO

674680

675681-- | Fix the type of a polymorphic stream as 'AheadT'.

676682--

677−-- @since 0.3.0

683+-- /Since: 0.3.0 ("Streamly")/

684+--

685+-- @since 0.8.0

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

679687aheadly = K.adapt

680688

src/Streamly/Internal/Data/Stream/Async.hs+31−15

@@ -513,6 +513,8 @@ mkAsyncD m = D.Stream step Nothing

513513 D.Skip s -> D.Skip (Just $ D.Stream step1 s)

514514 D.Stop -> D.Stop

515515

516+-- /Since: 0.8.0 ("Streamly.Prelude")/

517+--

516518-- This is slightly faster than the CPS version above

517519--

518520-- | Make the stream producer and consumer run concurrently by introducing a

@@ -635,7 +637,9 @@ infixr 6 `async`

635637-- Merges two streams possibly concurrently, preferring the

636638-- elements from the left one when available.

637639--

638−-- @since 0.2.0

640+-- /Since: 0.2.0 ("Streamly")/

641+--

642+-- @since 0.8.0

639643{-# INLINE async #-}

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

641645async = joinStreamVarAsync AsyncVar

@@ -675,11 +679,10 @@ consMAsync m r = fromStream $ K.yieldM m `async` (toStream r)

675679-- operation can be used to fold an infinite lazy container of streams.

676680--

677681-- @

678−-- import "Streamly"

679682-- import qualified "Streamly.Prelude" as S

680683-- import Control.Concurrent

681684--

682−-- main = (S.toList . 'asyncly' $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print

685+-- main = (S.toList . S.'asyncly' $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print

683686-- @

684687-- @

685688-- [1,2,3,4]

@@ -698,7 +701,7 @@ consMAsync m r = fromStream $ K.yieldM m `async` (toStream r)

698701-- consumer.

699702--

700703-- @

701−-- main = 'drain' . 'asyncly' $ do

704+-- main = S.'drain' . S.'asyncly' $ do

702705-- n <- return 3 \<\> return 2 \<\> return 1

703706-- S.yieldM $ do

704707-- threadDelay (n * 1000000)

@@ -710,19 +713,25 @@ consMAsync m r = fromStream $ K.yieldM m `async` (toStream r)

710713-- ThreadId 38: Delay 3

711714-- @

712715--

713−-- @since 0.1.0

716+-- /Since: 0.1.0 ("Streamly")/

717+--

718+-- @since 0.8.0

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

715720 deriving (MonadTrans)

716721

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

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

719724--

720−-- @since 0.2.0

725+-- /Since: 0.2.0 ("Streamly")/

726+--

727+-- @since 0.8.0

721728type Async = AsyncT IO

722729

723730-- | Fix the type of a polymorphic stream as 'AsyncT'.

724731--

725−-- @since 0.1.0

732+-- /Since: 0.1.0 ("Streamly")/

733+--

734+-- @since 0.8.0

726735asyncly :: IsStream t => AsyncT m a -> t m a

727736asyncly = adapt

728737

@@ -810,7 +819,9 @@ infixr 6 `wAsync`

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

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

812821--

813−-- @since 0.2.0

822+-- @since 0.8.0

823+--

824+-- /Since: 0.2.0 ("Streamly")/

814825{-# INLINE wAsync #-}

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

816827wAsync = joinStreamVarAsync WAsyncVar

@@ -828,11 +839,10 @@ wAsync = joinStreamVarAsync WAsyncVar

828839-- contrast to the depth wise scheduling behavior of 'AsyncT'.

829840--

830841-- @

831−-- import "Streamly"

832842-- import qualified "Streamly.Prelude" as S

833843-- import Control.Concurrent

834844--

835−-- main = (S.toList . 'wAsyncly' . maxThreads 1 $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print

845+-- main = (S.toList . S.'wAsyncly' . S.maxThreads 1 $ (S.fromList [1,2]) \<> (S.fromList [3,4])) >>= print

836846-- @

837847-- @

838848-- [1,3,2,4]

@@ -843,7 +853,7 @@ wAsync = joinStreamVarAsync WAsyncVar

843853-- now take a more general example:

844854--

845855-- @

846−-- main = (S.toList . 'wAsyncly' . maxThreads 1 $ (S.fromList [1,2,3]) \<> (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])) >>= print

856+-- main = (S.toList . S.'wAsyncly' . S.maxThreads 1 $ (S.fromList [1,2,3]) \<> (S.fromList [4,5,6]) \<> (S.fromList [7,8,9])) >>= print

847857-- @

848858-- @

849859-- [1,4,2,7,5,3,8,6,9]

@@ -899,7 +909,7 @@ wAsync = joinStreamVarAsync WAsyncVar

899909-- concurrently using a round robin scheduling.

900910--

901911-- @

902−-- main = 'drain' . 'wAsyncly' $ do

912+-- main = S.'drain' . S.'wAsyncly' $ do

903913-- n <- return 3 \<\> return 2 \<\> return 1

904914-- S.yieldM $ do

905915-- threadDelay (n * 1000000)

@@ -911,19 +921,25 @@ wAsync = joinStreamVarAsync WAsyncVar

911921-- ThreadId 38: Delay 3

912922-- @

913923--

914−-- @since 0.2.0

924+-- /Since: 0.2.0 ("Streamly")/

925+--

926+-- @since 0.8.0

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

916928 deriving (MonadTrans)

917929

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

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

920932--

921−-- @since 0.2.0

933+-- /Since: 0.2.0 ("Streamly")/

934+--

935+-- @since 0.8.0

922936type WAsync = WAsyncT IO

923937

924938-- | Fix the type of a polymorphic stream as 'WAsyncT'.

925939--

926−-- @since 0.2.0

940+-- /Since: 0.2.0 ("Streamly")/

941+--

942+-- @since 0.8.0

927943wAsyncly :: IsStream t => WAsyncT m a -> t m a

928944wAsyncly = adapt

929945

src/Streamly/Internal/Data/Stream/Combinators.hs+21−7

@@ -48,7 +48,9 @@ import Streamly.Internal.Data.Stream.Serial (SerialT)

4848-- When the actions are CPU bound this option can be used to

4949-- control the amount of CPU used by the stream.

5050--

51−-- @since 0.4.0

51+-- /Since: 0.4.0 ("Streamly")/

52+--

53+-- @since 0.8.0

5254{-# INLINE_NORMAL maxThreads #-}

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

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

@@ -73,7 +75,9 @@ maxThreadsSerial _ = id

7375-- applicative zip streams generates an infinite stream causing unbounded

7476-- concurrent generation with no limit on the buffer or threads.

7577--

76−-- @since 0.4.0

78+-- /Since: 0.4.0 ("Streamly")/

79+--

80+-- @since 0.8.0

7781{-# INLINE_NORMAL maxBuffer #-}

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

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

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

97101-- * The maximum rate that the stream producer can achieve

98102-- * The maximum rate that the stream consumer can achieve

99103--

100−-- @since 0.5.0

104+-- /Since: 0.5.0 ("Streamly")/

105+--

106+-- @since 0.8.0

101107{-# INLINE_NORMAL rate #-}

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

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

@@ -126,7 +132,9 @@ yieldRateSerial _ = id

126132-- go down to half of the specified rate on the lower side and double of

127133-- the specified rate on the higher side.

128134--

129−-- @since 0.5.0

135+-- /Since: 0.5.0 ("Streamly")/

136+--

137+-- @since 0.8.0

130138avgRate :: IsStream t => Double -> t m a -> t m a

131139avgRate r = rate (Just $ Rate (r/2) r (2*r) maxBound)

132140

@@ -137,7 +145,9 @@ avgRate r = rate (Just $ Rate (r/2) r (2*r) maxBound)

137145-- specified rate, even though it may possibly go above it at times, the

138146-- upper limit is double of the specified rate.

139147--

140−-- @since 0.5.0

148+-- /Since: 0.5.0 ("Streamly")/

149+--

150+-- @since 0.8.0

141151minRate :: IsStream t => Double -> t m a -> t m a

142152minRate r = rate (Just $ Rate r r (2*r) maxBound)

143153

@@ -150,7 +160,9 @@ minRate r = rate (Just $ Rate r r (2*r) maxBound)

150160-- applications where certain resource usage must not be allowed to go

151161-- beyond certain limits.

152162--

153−-- @since 0.5.0

163+-- /Since: 0.5.0 ("Streamly")/

164+--

165+-- @since 0.8.0

154166maxRate :: IsStream t => Double -> t m a -> t m a

155167maxRate r = rate (Just $ Rate (r/2) r r maxBound)

156168

@@ -162,7 +174,9 @@ maxRate r = rate (Just $ Rate (r/2) r r maxBound)

162174-- applications like graphics frame refresh where we need to maintain a

163175-- constant refresh rate.

164176--

165−-- @since 0.5.0

177+-- /Since: 0.5.0 ("Streamly")/

178+--

179+-- @since 0.8.0

166180constRate :: IsStream t => Double -> t m a -> t m a

167181constRate r = rate (Just $ Rate r r r 0)

168182

src/Streamly/Internal/Data/Stream/IsStream.hs+98−19

@@ -19,9 +19,51 @@

1919

2020module Streamly.Internal.Data.Stream.IsStream

2121 (

22+ -- * Stream Types

23+ -- ** Serial Streams

24+ SerialT

25+ , Serial

26+ , WSerialT

27+ , WSerial

28+

29+ -- ** Speculative Streams

30+ , AheadT

31+ , Ahead

32+

33+ -- ** Asynchronous Streams

34+ , AsyncT

35+ , Async

36+ , WAsyncT

37+ , WAsync

38+ , ParallelT

39+ , Parallel

40+ , mkAsync

41+

42+ -- ** Zipping Streams

43+ , ZipSerialM

44+ , ZipSerial

45+ , ZipAsyncM

46+ , ZipAsync

47+

48+ -- * Stream Type Adapters

49+ , IsStream ()

50+

51+ , serially

52+ , wSerially

53+ , asyncly

54+ , aheadly

55+ , wAsyncly

56+ , parallely

57+ , zipSerially

58+ , zipAsyncly

59+ , adapt

60+

61+ -- * Type Synonyms

62+ , MonadAsync

63+

2264 -- * Construction

2365 -- ** Primitives

24− K.nil

66+ , K.nil

2567 , K.nilM

2668 , K.cons

2769 , (K..:)

@@ -415,6 +457,7 @@ module Streamly.Internal.Data.Stream.IsStream

415457 -- * Combining Streams

416458

417459 -- ** Appending

460+ , serial

418461 , append

419462

420463 -- ** Interleaving

@@ -423,18 +466,22 @@ module Streamly.Internal.Data.Stream.IsStream

423466 , interleaveSuffix

424467 , interleaveInfix

425468

469+ , wSerial

426470 , Serial.wSerialFst

427471 , Serial.wSerialMin

428472

429473 -- ** Scheduling

474+ , ahead

475+ , async

476+ , wAsync

430477 , roundrobin

431478

432479 -- ** Parallel

480+ , parallel

433481 , Par.parallelFst

434482 , Par.parallelMin

435483

436484 -- ** Merging

437−

438485 -- , merge

439486 , mergeBy

440487 , mergeByM

@@ -448,9 +495,9 @@ module Streamly.Internal.Data.Stream.IsStream

448495 , Z.zipAsyncWithM

449496

450497 -- ** Flattening a Container of Streams

451− , foldWith

452− , foldMapWith

453− , forEachWith

498+ , concatFoldableWith

499+ , concatMapFoldableWith

500+ , concatForFoldableWith

454501

455502 -- ** Flattening a Stream of Streams

456503 , concat

@@ -514,6 +561,18 @@ module Streamly.Internal.Data.Stream.IsStream

514561 , rights

515562 , iterateMapLeftsWith

516563

564+ -- * Concurrency Control

565+ , maxThreads

566+ , maxBuffer

567+

568+ -- * Rate Limiting

569+ , Rate (..)

570+ , rate

571+ , avgRate

572+ , minRate

573+ , maxRate

574+ , constRate

575+

517576 -- * Diagnostics

518577 , inspectMode

519578

@@ -555,14 +614,26 @@ import Streamly.Internal.Data.Fold.Types (Fold (..), Fold2 (..))

555614import Streamly.Internal.Data.Parser (Parser (..))

556615import Streamly.Internal.Data.Unfold.Types (Unfold)

557616import Streamly.Internal.Data.Array.Storable.Foreign.Types (Array, writeNUnsafe)

558−import Streamly.Internal.Data.SVar (MonadAsync, defState, Rate)

559−import Streamly.Internal.Data.Stream.Combinators (inspectMode, maxYields)

617+import Streamly.Internal.Data.SVar (MonadAsync, defState, Rate (..))

618+import Streamly.Internal.Data.Stream.Ahead (AheadT, Ahead, ahead, aheadly)

619+import Streamly.Internal.Data.Stream.Async

620+ ( AsyncT, Async, WAsyncT, WAsync, mkAsync, async, asyncly, wAsync

621+ , wAsyncly)

622+import Streamly.Internal.Data.Stream.Combinators

623+ ( inspectMode, maxBuffer, maxThreads, maxYields, rate, avgRate, minRate

624+ , maxRate, constRate)

625+import Streamly.Internal.Data.Stream.Parallel

626+ ( ParallelT, Parallel, parallel, parallely)

560627import Streamly.Internal.Data.Stream.Prelude

561− (fromStreamS, toStreamS, foldWith, foldMapWith, forEachWith)

628+ (fromStreamS, toStreamS, concatFoldableWith, concatMapFoldableWith

629+ , concatForFoldableWith)

562630import Streamly.Internal.Data.Stream.StreamD (fromStreamD, toStreamD)

563−import Streamly.Internal.Data.Stream.StreamK (IsStream((|:), consM))

564−import Streamly.Internal.Data.Stream.Serial (SerialT, WSerialT)

565−import Streamly.Internal.Data.Stream.Zip (ZipSerialM)

631+import Streamly.Internal.Data.Stream.StreamK (IsStream((|:), consM), adapt)

632+import Streamly.Internal.Data.Stream.Serial

633+ ( SerialT, WSerialT, Serial, WSerial, serial, wSerial, serially

634+ , wSerially)

635+import Streamly.Internal.Data.Stream.Zip

636+ ( ZipSerialM, ZipSerial, ZipAsyncM, ZipAsync, zipSerially, zipAsyncly)

566637import Streamly.Internal.Data.Pipe.Types (Pipe (..))

567638import Streamly.Internal.Data.Time.Units

568639 ( AbsTime, MilliSecond64(..), addToAbsTime, toRelTime

@@ -918,9 +989,9 @@ iterate step = fromStreamS . S.iterate step

918989--

919990-- /Concurrent/

920991--

921−-- /Since: 0.7.0 (signature change)/

922−--

923992-- /Since: 0.1.2/

993+--

994+-- /Since: 0.7.0 (signature change)/

924995{-# INLINE_EARLY iterateM #-}

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

926997iterateM = K.iterateM

@@ -1997,7 +2068,9 @@ infixl 1 |&.

19972068--

19982069-- /Concurrent/

19992070--

2000−-- @since 0.3.0

2071+-- /Since: 0.3.0 ("Streamly")/

2072+--

2073+-- @since 0.8.0

20012074{-# INLINE (|$) #-}

20022075(|$) :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> (t m a -> t m b)

20032076-- (|$) f = f . Async.mkAsync

@@ -2024,7 +2097,9 @@ applyAsync = (|$)

20242097--

20252098-- /Concurrent/

20262099--

2027−-- @since 0.3.0

2100+-- /Since: 0.3.0 ("Streamly")/

2101+--

2102+-- @since 0.8.0

20282103{-# INLINE (|&) #-}

20292104(|&) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> t m b) -> t m b

20302105x |& f = f |$ x

@@ -2049,7 +2124,9 @@ x |& f = f |$ x

20492124--

20502125-- /Concurrent/

20512126--

2052−-- @since 0.3.0

2127+-- /Since: 0.3.0 ("Streamly")/

2128+--

2129+-- @since 0.8.0

20532130{-# INLINE (|$.) #-}

20542131(|$.) :: (IsStream t, MonadAsync m) => (t m a -> m b) -> (t m a -> m b)

20552132-- (|$.) f = f . Async.mkAsync

@@ -2073,7 +2150,9 @@ foldAsync = (|$.)

20732150--

20742151-- /Concurrent/

20752152--

2076−-- @since 0.3.0

2153+-- /Since: 0.3.0 ("Streamly")/

2154+--

2155+-- @since 0.8.0

20772156{-# INLINE (|&.) #-}

20782157(|&.) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> m b) -> m b

20792158x |&. f = f |$. x

@@ -2109,9 +2188,9 @@ transform pipe xs = fromStreamD $ D.transform pipe (toStreamD xs)

21092188-- designed to work with the @foldl@ library. The suffix @x@ is a mnemonic for

21102189-- extraction.

21112190--

2112−-- /Since: 0.7.0 (Monad m constraint)/

2113−--

21142191-- /Since 0.2.0/

2192+--

2193+-- /Since: 0.7.0 (Monad m constraint)/

21152194{-# DEPRECATED scanx "Please use scanl followed by map instead." #-}

21162195{-# INLINE scanx #-}

21172196scanx :: (IsStream t, Monad m) => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b

src/Streamly/Internal/Data/Stream/Parallel.hs+14−7

@@ -249,7 +249,9 @@ infixr 6 `parallel`

249249--

250250-- `nilM` is currently @Internal@.

251251--

252−-- @since 0.2.0

252+-- /Since: 0.2.0 ("Streamly")/

253+--

254+-- @since 0.8.0

253255{-# INLINE parallel #-}

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

255257parallel = joinStreamVarPar ParallelVar StopNone

@@ -451,7 +453,7 @@ distributeAsync_ = flip (foldr tapAsync)

451453-- efficient than 'ParallelT'.

452454--

453455-- @

454−-- main = ('toList' . 'parallely' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print

456+-- main = (S.'toList' . S.'parallely' $ (S.fromFoldable [1,2]) \<> (S.fromFoldable [3,4])) >>= print

455457-- @

456458-- @

457459-- [1,3,2,4]

@@ -471,11 +473,10 @@ distributeAsync_ = flip (foldr tapAsync)

471473-- of the loop concurrently.

472474--

473475-- @

474−-- import "Streamly"

475476-- import qualified "Streamly.Prelude" as S

476477-- import Control.Concurrent

477478--

478−-- main = 'drain' . 'parallely' $ do

479+-- main = S.'drain' . S.'parallely' $ do

479480-- n <- return 3 \<\> return 2 \<\> return 1

480481-- S.yieldM $ do

481482-- threadDelay (n * 1000000)

@@ -490,21 +491,27 @@ distributeAsync_ = flip (foldr tapAsync)

490491-- Note that parallel composition can only combine a finite number of

491492-- streams as it needs to retain state for each unfinished stream.

492493--

494+-- /Since: 0.1.0 ("Streamly")/

495+--

493496-- /Since: 0.7.0 (maxBuffer applies to ParallelT streams)/

494497--

495−-- /Since: 0.1.0/

498+-- @since 0.8.0

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

497500 deriving (MonadTrans)

498501

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

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

501504--

502−-- @since 0.2.0

505+-- /Since: 0.2.0 ("Streamly")/

506+--

507+-- @since 0.8.0

503508type Parallel = ParallelT IO

504509

505510-- | Fix the type of a polymorphic stream as 'ParallelT'.

506511--

507−-- @since 0.1.0

512+-- /Since: 0.1.0 ("Streamly")/

513+--

514+-- @since 0.8.0

508515parallely :: IsStream t => ParallelT m a -> t m a

509516parallely = adapt

510517

src/Streamly/Internal/Data/Stream/Prelude.hs+25−24

@@ -58,9 +58,9 @@ module Streamly.Internal.Data.Stream.Prelude

5858 , K.concatMap

5959

6060 -- * Fold Utilities

61− , foldWith

62− , foldMapWith

63− , forEachWith

61+ , concatFoldableWith

62+ , concatMapFoldableWith

63+ , concatForFoldableWith

6464 )

6565where

6666

@@ -292,58 +292,59 @@ foldbWith :: IsStream t

292292foldbWith f = K.foldb f K.nil

293293-}

294294

295−-- /Since: 0.7.0 ("Streamly.Prelude")/

296−--

295+

297296-- | A variant of 'Data.Foldable.fold' that allows you to fold a 'Foldable'

298297-- container of streams using the specified stream sum operation.

299298--

300−-- @foldWith 'async' $ map return [1..3]@

299+-- @concatFoldableWith 'async' $ map return [1..3]@

301300--

302301-- Equivalent to:

303302--

304303-- @

305−-- foldWith f = S.foldMapWith f id

304+-- concatFoldableWith f = S.concatMapFoldableWith f id

306305-- @

307306--

307+-- /Since: 0.8.0 (Renamed foldWith to concatFoldableWith)/

308+--

308309-- /Since: 0.1.0 ("Streamly")/

309−{-# INLINABLE foldWith #-}

310−foldWith :: (IsStream t, Foldable f)

310+{-# INLINABLE concatFoldableWith #-}

311+concatFoldableWith :: (IsStream t, Foldable f)

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

312−foldWith f = Prelude.foldr f K.nil

313+concatFoldableWith f = Prelude.foldr f K.nil

313314

314−-- /Since: 0.7.0 ("Streamly.Prelude")/

315−--

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

317316-- on a 'Foldable' container and then fold it using the specified stream merge

318317-- operation.

319318--

320−-- @foldMapWith 'async' return [1..3]@

319+-- @concatMapFoldableWith 'async' return [1..3]@

321320--

322321-- Equivalent to:

323322--

324323-- @

325−-- foldMapWith f g xs = S.concatMapWith f g (S.fromFoldable xs)

324+-- concatMapFoldableWith f g xs = S.concatMapWith f g (S.fromFoldable xs)

326325-- @

327326--

327+-- /Since: 0.8.0 (Renamed foldMapWith to concatMapFoldableWith)/

328+--

328329-- /Since: 0.1.0 ("Streamly")/

329−{-# INLINABLE foldMapWith #-}

330−foldMapWith :: (IsStream t, Foldable f)

330+{-# INLINABLE concatMapFoldableWith #-}

331+concatMapFoldableWith :: (IsStream t, Foldable f)

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

332−foldMapWith f g = Prelude.foldr (f . g) K.nil

333+concatMapFoldableWith f g = Prelude.foldr (f . g) K.nil

333334

334−-- /Since: 0.7.0 ("Streamly.Prelude")/

335−--

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

335+-- | Like 'concatMapFoldableWith' but with the last two arguments reversed i.e. the

337336-- monadic streaming function is the last argument.

338337--

339338-- Equivalent to:

340339--

341340-- @

342−-- forEachWith = flip S.foldMapWith

341+-- concatForFoldableWith = flip S.concatMapFoldableWith

343342-- @

344343--

344+-- /Since: 0.8.0 (Renamed forEachWith to concatForFoldableWith)/

345+--

345346-- /Since: 0.1.0 ("Streamly")/

346−{-# INLINABLE forEachWith #-}

347−forEachWith :: (IsStream t, Foldable f)

347+{-# INLINABLE concatForFoldableWith #-}

348+concatForFoldableWith :: (IsStream t, Foldable f)

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

349−forEachWith f xs g = Prelude.foldr (f . g) K.nil xs

350+concatForFoldableWith f xs g = Prelude.foldr (f . g) K.nil xs

src/Streamly/Internal/Data/Stream/Serial.hs+21−7

@@ -137,7 +137,9 @@ import Prelude hiding (map, mapM, errorWithoutStackTrace)

137137-- In the code above, the 'serially' combinator can be omitted as the default

138138-- stream type is 'SerialT'.

139139--

140−-- @since 0.2.0

140+-- /Since: 0.2.0 ("Streamly")/

141+--

142+-- @since 0.8.0

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

142144 -- XXX when deriving do we inherit an INLINE?

143145 deriving (Semigroup, Monoid, MonadTrans)

@@ -145,7 +147,9 @@ newtype SerialT m a = SerialT {getSerialT :: Stream m a}

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

146148-- for more details.

147149--

148−-- @since 0.2.0

150+-- /Since: 0.2.0 ("Streamly")/

151+--

152+-- @since 0.8.0

149153type Serial = SerialT IO

150154

151155-- |

@@ -155,7 +159,9 @@ type StreamT = SerialT

155159

156160-- | Fix the type of a polymorphic stream as 'SerialT'.

157161--

158−-- @since 0.1.0

162+-- /Since: 0.1.0 ("Streamly")/

163+--

164+-- @since 0.8.0

159165serially :: IsStream t => SerialT m a -> t m a

160166serially = adapt

161167

@@ -309,14 +315,18 @@ TRAVERSABLE_INSTANCE(SerialT)

309315-- (2,4)

310316-- @

311317--

312−-- @since 0.2.0

318+-- /Since: 0.2.0 ("Streamly")/

319+--

320+-- @since 0.8.0

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

314322 deriving (MonadTrans)

315323

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

317325-- documentation for more details.

318326--

319−-- @since 0.2.0

327+-- /Since: 0.2.0 ("Streamly")/

328+--

329+-- @since 0.8.0

320330type WSerial = WSerialT IO

321331

322332-- |

@@ -326,7 +336,9 @@ type InterleavedT = WSerialT

326336

327337-- | Fix the type of a polymorphic stream as 'WSerialT'.

328338--

329−-- @since 0.2.0

339+-- /Since: 0.2.0 ("Streamly")/

340+--

341+-- @since 0.8.0

330342wSerially :: IsStream t => WSerialT m a -> t m a

331343wSerially = adapt

332344

@@ -372,7 +384,9 @@ infixr 6 `wSerial`

372384-- When one stream stops the rest of the other stream is used in the output

373385-- stream.

374386--

375−-- @since 0.2.0

387+-- @since 0.8.0

388+--

389+-- /Since: 0.2.0 ("Streamly")/

376390{-# INLINE wSerial #-}

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

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

src/Streamly/Internal/Data/Stream/StreamK/Type.hs+9−3

@@ -132,7 +132,9 @@ infixr 5 |:

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

133133-- some monad 'm'.

134134--

135−-- @since 0.2.0

135+-- /Since: 0.2.0 ("Streamly")/

136+--

137+-- @since 0.8.0

136138class

137139#if __GLASGOW_HASKELL__ >= 806

138140 ( forall m a. MonadAsync m => Semigroup (t m a)

@@ -198,7 +200,9 @@ type Streaming = IsStream

198200--

199201-- | Adapt any specific stream type to any other specific stream type.

200202--

201−-- @since 0.1.0

203+-- /Since: 0.1.0 ("Streamly")/

204+--

205+-- @since 0.8.0

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

203207adapt = fromStream . toStream

204208

@@ -761,7 +765,9 @@ infixr 6 `serial`

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

762766-- stream, and then all elements from the second stream.

763767--

764−-- @since 0.2.0

768+-- /Since: 0.2.0 ("Streamly")/

769+--

770+-- @since 0.8.0

765771{-# INLINE serial #-}

766772serial :: IsStream t => t m a -> t m a -> t m a

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

src/Streamly/Internal/Data/Stream/Zip.hs+18−6

@@ -130,7 +130,9 @@ zipAsyncWith f = zipAsyncWithM (\a b -> return (f a b))

130130-- The 'Semigroup' instance of this type works the same way as that of

131131-- 'SerialT'.

132132--

133−-- @since 0.2.0

133+-- /Since: 0.2.0 ("Streamly")/

134+--

135+-- @since 0.8.0

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

135137 deriving (Semigroup, Monoid)

136138

@@ -141,12 +143,16 @@ type ZipStream = ZipSerialM

141143

142144-- | An IO stream whose applicative instance zips streams serially.

143145--

144−-- @since 0.2.0

146+-- /Since: 0.2.0 ("Streamly")/

147+--

148+-- @since 0.8.0

145149type ZipSerial = ZipSerialM IO

146150

147151-- | Fix the type of a polymorphic stream as 'ZipSerialM'.

148152--

149−-- @since 0.2.0

153+-- /Since: 0.2.0 ("Streamly")/

154+--

155+-- @since 0.8.0

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

151157zipSerially = K.adapt

152158

@@ -209,18 +215,24 @@ TRAVERSABLE_INSTANCE(ZipSerialM)

209215-- The 'Semigroup' instance of this type works the same way as that of

210216-- 'SerialT'.

211217--

212−-- @since 0.2.0

218+-- /Since: 0.2.0 ("Streamly")/

219+--

220+-- @since 0.8.0

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

214222 deriving (Semigroup, Monoid)

215223

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

217225--

218−-- @since 0.2.0

226+-- /Since: 0.2.0 ("Streamly")/

227+--

228+-- @since 0.8.0

219229type ZipAsync = ZipAsyncM IO

220230

221231-- | Fix the type of a polymorphic stream as 'ZipAsyncM'.

222232--

223−-- @since 0.2.0

233+-- /Since: 0.2.0 ("Streamly")/

234+--

235+-- @since 0.8.0

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

225237zipAsyncly = K.adapt

226238

src/Streamly/Internal/FileSystem/Event/Darwin.hs+1−1

@@ -162,7 +162,7 @@ import Foreign.Marshal.Array (withArray)

162162import Foreign.Ptr (Ptr, castPtr)

163163import Foreign.Storable (Storable(..))

164164import GHC.IO.Handle.FD (fdToHandle)

165−import Streamly (SerialT)

165+import Streamly.Prelude (SerialT)

166166import Streamly.Internal.Data.Cont (contListMap)

167167import Streamly.Internal.Data.Parser (Parser)

168168import Streamly.Internal.Data.Array.Storable.Foreign.Types (Array(..))

src/Streamly/Internal/FileSystem/Event/Linux.hs+1−1

@@ -159,7 +159,7 @@ import Foreign.Storable (peek, peekByteOff, sizeOf)

159159import GHC.IO.Device (IODeviceType(Stream))

160160import GHC.IO.FD (fdFD, mkFD)

161161import GHC.IO.Handle.FD (mkHandleFromFD)

162−import Streamly (SerialT)

162+import Streamly.Prelude (SerialT)

163163import Streamly.Internal.Data.Parser (Parser)

164164import Streamly.Internal.Data.Array.Storable.Foreign.Types (Array(..))

165165import System.IO (Handle, hClose, IOMode(ReadMode))

src/Streamly/Internal/FileSystem/Handle.hs+1−1

@@ -109,7 +109,7 @@ import GHC.ForeignPtr (mallocPlainForeignPtrBytes)

109109import System.IO (Handle, hGetBufSome, hPutBuf, stdin, stdout)

110110import Prelude hiding (read)

111111

112−import Streamly (MonadAsync)

112+import Streamly.Prelude (MonadAsync)

113113import Streamly.Data.Fold (Fold)

114114import Streamly.Internal.Data.Fold.Types (Fold2(..))

115115import Streamly.Internal.Data.Unfold.Types (Unfold(..))

src/Streamly/Internal/Network/Inet/TCP.hs+1−1

@@ -103,7 +103,7 @@ import Network.Socket

103103 socket)

104104import Prelude hiding (read)

105105

106−import Streamly (MonadAsync)

106+import Streamly.Prelude (MonadAsync)

107107import Streamly.Internal.Data.Fold.Types (Fold(..))

108108import Streamly.Internal.Data.SVar (fork)

109109import Streamly.Internal.Data.Unfold.Types (Unfold(..))

src/Streamly/Internal/Network/Socket.hs+1−1

@@ -87,7 +87,7 @@ import Prelude hiding (read)

8787

8888import qualified Network.Socket as Net

8989

90−import Streamly (MonadAsync)

90+import Streamly.Prelude (MonadAsync)

9191import Streamly.Internal.Data.Unfold.Types (Unfold(..))

9292import Streamly.Internal.Data.Array.Storable.Foreign.Types (Array(..), lpackArraysChunksOf)

9393import Streamly.Internal.Data.Array.Storable.Foreign.Mut.Types (mutableArray)

src/Streamly/Internal/Unicode/Array/Char.hs+2−1

@@ -18,9 +18,10 @@ module Streamly.Internal.Unicode.Array.Char

1818where

1919

2020import Control.Monad.IO.Class (MonadIO)

21−import Streamly (IsStream, MonadAsync)

21+import Streamly.Prelude (MonadAsync)

2222import Prelude hiding (String, lines, words, unlines, unwords)

2323import Streamly.Data.Array.Storable.Foreign (Array)

24+import Streamly.Internal.Data.Stream.IsStream (IsStream)

2425

2526import qualified Streamly.Internal.Unicode.Stream as S

2627import qualified Streamly.Data.Array.Storable.Foreign as A

src/Streamly/Internal/Unicode/Array/Prim/Pinned.hs+2−1

@@ -18,8 +18,9 @@ module Streamly.Internal.Unicode.Array.Prim.Pinned

1818where

1919

2020import Control.Monad.IO.Class (MonadIO(..))

21−import Streamly (IsStream, MonadAsync)

21+import Streamly.Prelude (MonadAsync)

2222import Prelude hiding (String, lines, words, unlines, unwords)

23+import Streamly.Internal.Data.Stream.IsStream (IsStream)

2324import Streamly.Internal.Data.Array.Prim.Pinned (Array)

2425

2526import qualified Streamly.Internal.Unicode.Stream as S

src/Streamly/Internal/Unicode/Char.hs+1−1

@@ -24,7 +24,7 @@ where

2424

2525import Data.Char (isAsciiUpper, isAsciiLower)

2626

27−-- import Streamly (IsStream)

27+-- import Streamly.Prelude (IsStream)

2828

2929-------------------------------------------------------------------------------

3030-- Unicode aware operations on strings

src/Streamly/Internal/Unicode/Stream.hs+1−1

@@ -66,11 +66,11 @@ import GHC.ForeignPtr (ForeignPtr (..))

6666import GHC.IO.Encoding.Failure (isSurrogate)

6767import GHC.Ptr (Ptr (..), plusPtr)

6868import System.IO.Unsafe (unsafePerformIO)

69−import Streamly (IsStream)

7069import Streamly.Data.Fold (Fold)

7170import Streamly.Data.Array.Storable.Foreign (Array)

7271import Streamly.Internal.Data.Unfold (Unfold)

7372import Streamly.Internal.Data.SVar (adaptState)

73+import Streamly.Internal.Data.Stream.IsStream (IsStream)

7474import Streamly.Internal.Data.Stream.StreamD (Stream(..), Step (..))

7575import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))

7676

src/Streamly/Prelude.hs+274−16

@@ -34,11 +34,62 @@

3434-- Deconstruction and folds accept a 'SerialT' type instead of a polymorphic

3535-- type to ensure that streams always have a concrete monomorphic type by

3636-- default, reducing type errors. In case you want to use any other type of

37−-- stream you can use one of the type combinators provided in the "Streamly"

38−-- module to convert the stream type.

37+-- stream you can use one of the type combinators provided to convert the

38+-- stream type.

3939

4040module Streamly.Prelude

4141 (

42+ -- * Stream Types

43+ -- $streamtypes

44+

45+ -- ** Serial Streams

46+ -- $serial

47+ SerialT

48+ , WSerialT

49+

50+ -- ** Speculative Streams

51+ -- $ahead

52+ , AheadT

53+

54+ -- ** Asynchronous Streams

55+ -- $async

56+ , AsyncT

57+ , WAsyncT

58+ , ParallelT

59+ , mkAsync

60+

61+ -- ** Zipping Streams

62+ -- $zipping

63+ , ZipSerialM

64+ , ZipAsyncM

65+

66+ -- * IO Streams

67+ , Serial

68+ , WSerial

69+ , Ahead

70+ , Async

71+ , WAsync

72+ , Parallel

73+ , ZipSerial

74+ , ZipAsync

75+

76+ -- * Type Synonyms

77+ , MonadAsync

78+

79+ -- * Stream Type Adapters

80+ -- $adapters

81+ , IsStream ()

82+

83+ , serially

84+ , wSerially

85+ , asyncly

86+ , aheadly

87+ , wAsyncly

88+ , parallely

89+ , zipSerially

90+ , zipAsyncly

91+ , adapt

92+

4293 -- * Construction

4394 -- ** Primitives

4495 -- | Primitives to construct a stream from pure values or monadic actions.

@@ -47,7 +98,7 @@ module Streamly.Prelude

4798 -- versions provided in this module can be much more efficient in most

4899 -- cases. Users can create custom combinators using these primitives.

49100

50− nil

101+ , nil

51102 , cons

52103 , (.:)

53104

@@ -263,6 +314,13 @@ module Streamly.Prelude

263314 -- trimming sequences

264315 , stripPrefix

265316

317+ -- * Parallel Function Application

318+ -- $application

319+ , (|$)

320+ , (|&)

321+ , (|$.)

322+ , (|&.)

323+

266324 -- * Transformation

267325

268326 -- ** Mapping

@@ -464,6 +522,16 @@ module Streamly.Prelude

464522 -- >> S.toList $ fold $ [S.fromList [1,2], S.fromList [3,4]]

465523 -- [1,2,3,4]

466524 -- @

525+ , serial

526+

527+ -- ** Interleaving

528+ , wSerial

529+

530+ -- ** Scheduling

531+ , ahead

532+ , async

533+ , wAsync

534+ , parallel

467535

468536 -- ** Merging

469537 -- | Streams form a commutative semigroup under the merge

@@ -498,18 +566,6 @@ module Streamly.Prelude

498566 , zipAsyncWith

499567 , zipAsyncWithM

500568

501− {-

502− -- ** Folding Containers of Streams

503− -- | These are variants of standard 'Foldable' fold functions that use a

504− -- polymorphic stream sum operation (e.g. 'async' or 'wSerial') to fold a

505− -- finite container of streams. Note that these are just special cases of

506− -- the more general 'concatMapWith' operation.

507− --

508− , foldMapWith

509− , forEachWith

510− , foldWith

511− -}

512−

513569 -- ** Folding Streams of Streams

514570 -- | Stream operations like map and filter represent loop processing in

515571 -- imperative programming terms. Similarly, the imperative concept of

@@ -539,6 +595,16 @@ module Streamly.Prelude

539595 , concatMapM

540596 , concatUnfold

541597

598+ -- ** Folding Containers of Streams

599+ -- | These are variants of standard 'Foldable' fold functions that use a

600+ -- polymorphic stream sum operation (e.g. 'async' or 'wSerial') to fold a

601+ -- finite container of streams. Note that these are just special cases of

602+ -- the more general 'concatMapWith' operation.

603+ --

604+ , concatFoldableWith

605+ , concatMapFoldableWith

606+ , concatForFoldableWith

607+

542608 -- * Exceptions

543609 , before

544610 , after

@@ -547,6 +613,19 @@ module Streamly.Prelude

547613 , finally

548614 , handle

549615

616+ -- * Concurrency Control

617+ -- $concurrency

618+ , maxThreads

619+ , maxBuffer

620+

621+ -- * Rate Limiting

622+ , Rate (..)

623+ , rate

624+ , avgRate

625+ , minRate

626+ , maxRate

627+ , constRate

628+

550629 -- * Deprecated

551630 , once

552631 , each

@@ -571,9 +650,124 @@ import Prelude

571650

572651import Streamly.Internal.Data.Stream.IsStream

573652

653+

654+-- $streamtypes

655+-- The basic stream type is 'Serial', it represents a sequence of IO actions,

656+-- and is a 'Monad'. The type 'SerialT' is a monad transformer that can

657+-- represent a sequence of actions in an arbitrary monad. The type 'Serial' is

658+-- in fact a synonym for @SerialT IO@. There are a few more types similar to

659+-- 'SerialT', all of them represent a stream and differ only in the

660+-- 'Semigroup', 'Applicative' and 'Monad' compositions of the stream. 'Serial'

661+-- and 'WSerial' types compose serially whereas 'Async' and 'WAsync'

662+-- types compose concurrently. All these types can be freely inter-converted

663+-- using type combinators without any cost. You can freely switch to any type

664+-- of composition at any point in the program. When no type annotation or

665+-- explicit stream type combinators are used, the default stream type is

666+-- inferred as 'Serial'.

667+

668+-- $serial

669+--

670+-- When a stream consumer demands an element from a serial stream constructed

671+-- as @a \`consM` b \`consM` ... nil@, the action @a@ at the head of the stream

672+-- sequence is executed and the result is supplied to the consumer. When the

673+-- next element is demanded, the action @b@ is executed and its result is

674+-- supplied. Thus, the effects are performed and results are consumed strictly

675+-- in a serial order. Serial streams can be considered as /spatially ordered/

676+-- streams as the order of execution and consumption is the same as the spatial

677+-- order in which the actions are composed by the programmer.

678+--

679+-- Serial streams enforce the side effects as well as the results of the

680+-- actions to be in the same order in which the actions are added to the

681+-- stream. Therefore, the semigroup operation for serial streams is not

682+-- commutative:

683+--

684+-- @

685+-- a <> b is not the same as b <> a

686+-- @

687+--

688+-- There are two serial stream types 'SerialT' and 'WSerialT'. The stream

689+-- evaluation of both the variants works in the same way as described above,

690+-- they differ only in the 'Semigroup' and 'Monad' implementaitons.

691+

692+-- $ahead

693+--

694+-- When a stream consumer demands an element from a speculative stream

695+-- constructed as @a \`consM` b \`consM` ... nil@, the action @a@ at the head

696+-- of the stream is executed and the output of the action is supplied to the

697+-- consumer. However, in addition to the action at the head multiple actions

698+-- following it may also be executed concurrently and the results buffered.

699+-- When the next element is demanded it may be served from the buffer and we

700+-- may execute the next action in the sequence to keep the buffer adequately

701+-- filled. Thus, the actions are executed concurrently but results consumed in

702+-- serial order just like serial streams. `consM` can be used to fold an

703+-- infinite lazy container of effects, as the number of concurrent executions

704+-- is limited.

705+--

706+-- Similar to 'consM', the monadic stream generation (e.g. replicateM) and

707+-- transformation operations (e.g. mapM) on speculative streams can execute

708+-- multiple effects concurrently in a speculative manner.

709+--

710+-- How many effects can be executed concurrently and how many results can be

711+-- buffered are controlled by 'maxThreads' and 'maxBuffer' combinators

712+-- respectively. The actual number of concurrent threads is adjusted according

713+-- to the rate at which the consumer is consuming the stream. It may even

714+-- execute actions serially in a single thread if that is enough to match the

715+-- consumer's speed.

716+--

717+-- Speculative streams enforce ordering of the results of actions in the stream

718+-- but the side effects are only partially ordered. Therefore, the semigroup

719+-- operation for speculative streams is not commutative from the pure outputs

720+-- perspective but commutative from side effects perspective.

721+

722+-- $async

723+--

724+-- /Scheduling and execution:/ In an asynchronous stream @a \`consM` b \`consM`

725+-- c ...@, the actions @a@, @b@, and @c@ are executed concurrently with the

726+-- consumer of the stream. The actions are /scheduled/ for execution in the

727+-- same order as they are specified in the stream. Multiple scheduled actions

728+-- may be /executed/ concurrently in parallel threads of execution. The

729+-- actions may be executed out of order and they may complete at arbitrary

730+-- times. Therefore, the /effects/ of the actions may be observed out of

731+-- order.

732+--

733+-- /Buffering:/ The /results/ from multiple threads of execution are queued in

734+-- a buffer as soon as they become available. The consumer of the stream is

735+-- served from this buffer. Therefore, the consumer may observe the results to

736+-- be out of order. In other words, an asynchronous stream is an unordered

737+-- stream i.e. order does not matter.

738+--

739+-- /Concurrency control:/ Threads are suspended if the `maxBuffer` limit is

740+-- reached, and resumed when the consumer makes space in the buffer. The

741+-- maximum number of concurrent threads depends on `maxThreads`. Number of

742+-- threads is increased or decreased based on the speed of the consumer.

743+--

744+-- /Generation operations:/ Concurrent stream generation operations e.g.

745+-- 'Streamly.Prelude.replicateM' when used in async style schedule and execute

746+-- the stream generating actions in the manner described above. The generation

747+-- actions run concurrently, effects and results of the actions as observed by

748+-- the consumer of the stream may be out of order.

749+--

750+-- /Transformation operations:/ Concurrent stream transformation operations

751+-- e.g. 'Streamly.Prelude.mapM', when used in async style, schedule and

752+-- execute transformation actions in the manner described above. Transformation

753+-- actions run concurrently, effects and results of the actions may be

754+-- observed by the consumer out of order.

755+--

756+-- /Variants:/ There are two asynchronous stream types 'AsyncT' and 'WAsyncT'.

757+-- They are identical with respect to single stream evaluation behavior. Their

758+-- behaviors differ in how they combine multiple streams using 'Semigroup' or

759+-- 'Monad' composition. Since the order of elements does not matter in

760+-- asynchronous streams the 'Semigroup' operation is effectively commutative.

761+

762+-- $zipping

763+--

764+-- 'ZipSerialM' and 'ZipAsyncM', provide 'Applicative' instances for zipping the

765+-- corresponding elements of two streams together. Note that these types are

766+-- not monads.

767+

574768-- $rightfolds

575769--

576−-- Let's take a closer look at the @foldr@ definition for lists, as given

770+-- Let's take a closer look at the @foldr@ definition for lists, as given3

577771-- earlier:

578772--

579773-- @

@@ -802,3 +996,67 @@ import Streamly.Internal.Data.Stream.IsStream

802996-- @

803997-- [1.0,2.0,3.0,4.0,5.0,6.0,7.0,8.0,9.0,10.0,11.0,12.0,13.0,14.0,15.0,16.0,17.0,18.0,19.0]

804998-- @

999+

1000+-- $application

1001+--

1002+-- Stream processing functions can be composed in a chain using function

1003+-- application with or without the '$' operator, or with reverse function

1004+-- application operator '&'. Streamly provides concurrent versions of these

1005+-- operators applying stream processing functions such that each stage of the

1006+-- stream can run in parallel. The operators start with a @|@; we can read '|$'

1007+-- as "@parallel dollar@" to remember that @|@ comes before '$'.

1008+--

1009+-- Imports for the code snippets below:

1010+--

1011+-- @

1012+-- import qualified Streamly.Prelude as S

1013+-- import Control.Concurrent

1014+-- @

1015+

1016+-- $concurrency

1017+--

1018+-- These combinators can be used at any point in a stream composition to set

1019+-- parameters to control the concurrency of the /argument stream/. A control

1020+-- parameter set at any point remains effective for any concurrent combinators

1021+-- used in the argument stream until it is reset by using the combinator again.

1022+-- These control parameters have no effect on non-concurrent combinators in the

1023+-- stream, or on non-concurrent streams.

1024+--

1025+-- /Pitfall:/ Remember that 'maxBuffer' in the following example applies to

1026+-- 'mapM' and any other combinators that may follow it, and it does not apply

1027+-- to the combinators before it:

1028+--

1029+-- @

1030+-- ...

1031+-- $ S.maxBuffer 10

1032+-- $ S.mapM ...

1033+-- ...

1034+-- @

1035+--

1036+-- If we use '&' instead of '$' the situation will reverse, in the following

1037+-- example, 'maxBuffer' does not apply to 'mapM', it applies to combinators

1038+-- that come before it, because those are the arguments to 'maxBuffer':

1039+--

1040+-- @

1041+-- ...

1042+-- & S.maxBuffer 10

1043+-- & S.mapM ...

1044+-- ...

1045+-- @

1046+

1047+-- $adapters

1048+--

1049+-- You may want to use different stream composition styles at different points

1050+-- in your program. Stream types can be freely converted or adapted from one

1051+-- type to another. The 'IsStream' type class facilitates type conversion of

1052+-- one stream type to another. It is not used directly, instead the type

1053+-- combinators provided below are used for conversions.

1054+--

1055+-- To adapt from one monomorphic type (e.g. 'AsyncT') to another monomorphic

1056+-- type (e.g. 'SerialT') use the 'adapt' combinator. To give a polymorphic code

1057+-- a specific interpretation or to adapt a specific type to a polymorphic type

1058+-- use the type specific combinators e.g. 'asyncly' or 'wSerially'. You

1059+-- cannot adapt polymorphic code to polymorphic code, as the compiler would not know

1060+-- which specific type you are converting from or to. If you see a an

1061+-- @ambiguous type variable@ error then most likely you are using 'adapt'

1062+-- unnecessarily on polymorphic code.

src/Streamly/Tutorial.hs+74−91

@@ -170,7 +170,6 @@ module Streamly.Tutorial

170170 )

171171where

172172

173−import Streamly hiding (foldWith, foldMapWith, forEachWith)

174173import Streamly.Prelude

175174import Data.Semigroup

176175import Control.Applicative

@@ -345,8 +344,7 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

345344-- GHCi session and import these lines to start playing.

346345--

347346-- @

348−-- > import "Streamly"

349−-- > import "Streamly.Prelude" ((|:))

347+-- > import "Streamly.Prelude" ((|:), (|&))

350348-- > import qualified "Streamly.Prelude" as S

351349--

352350-- > import Control.Concurrent

@@ -412,15 +410,19 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

412410--

413411-- @

414412-- > let p n = threadDelay (n * 1000000) >> return n

415−-- > S.'toList' $ 'parallely' $ p 3 |: p 2 |: p 1 |: S.'nil'

413+-- > :set +s

414+-- > S.'toList' $ S.'parallely' $ p 3 |: p 2 |: p 1 |: S.'nil'

416415-- [1,2,3]

417−-- > S.'toList' $ 'aheadly' $ p 3 |: p 2 |: p 1 |: S.'nil'

416+-- (3.01 secs, 2,018,432 bytes)

417+-- > S.'toList' $ S.'aheadly' $ p 3 |: p 2 |: p 1 |: S.'nil'

418418-- [3,2,1]

419+-- (3.01 secs, 2,055,880 bytes)

419420-- @

420421-- The following finishes in 10 seconds (100 seconds when serial):

421422--

422423-- @

423−-- > S.drain $ 'asyncly' $ S.'replicateM' 10 $ p 10

424+-- > S.'drain' $ S.'asyncly' $ S.'replicateM' 10 $ p 10

425+-- (10.01 secs, 2,080,320 bytes)

424426-- @

425427--

426428

@@ -456,7 +458,6 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

456458-- numbers them and prints them:

457459--

458460-- @

459−-- import "Streamly"

460461-- import qualified "Streamly.Prelude" as S

461462-- import Data.Char (toUpper)

462463-- import Data.Function ((&))

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

482483--

483484-- @

484485-- > let p n = threadDelay (n * 1000000) >> return n

485−-- > S.'drain' $ S.aheadly $ S.'mapM' (\\x -> p 1 >> print x) (serially $ S.repeatM (p 1))

486+-- > S.'drain' $ S.'aheadly' $ S.'mapM' (\\x -> p 1 >> print x) (S.'serially' $ S.'repeatM' (p 1))

486487-- @

487488--

488489

@@ -515,7 +516,6 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

515516-- not explicitly specified use the imports shown below.

516517--

517518-- @

518−-- import "Streamly"

519519-- import "Streamly.Prelude" ((|:), nil)

520520-- import qualified "Streamly.Prelude" as S

521521--

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

553553-- 3, 4:

554554--

555555-- @

556−-- main = S.'drain' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)

556+-- main = S.'drain' $ (print 1 |: print 2 |: S.nil) <> (print 3 |: print 4 |: S.nil)

557557-- @

558558-- @

559559-- 1

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

580580-- irrespective of the type of stream:

581581--

582582-- @

583−-- main = S.'drain' $ (print 1 |: print 2 |: nil) \`serial` (print 3 |: print 4 |: nil)

583+-- main = S.'drain' $ (print 1 |: print 2 |: S.nil) \`S.serial` (print 3 |: print 4 |: S.nil)

584584-- @

585585

586586-- $interleaved

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

597597-- The following example prints the sequence 1, 3, 2, 4

598598--

599599-- @

600−-- main = S.'drain' . 'wSerially' $ (print 1 |: print 2 |: nil) <> (print 3 |: print 4 |: nil)

600+-- main = S.'drain' . S.'wSerially' $ (print 1 |: print 2 |: S.'nil') <> (print 3 |: print 4 |: S.'nil')

601601-- @

602602-- @

603603-- 1

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

612612-- thread and take a combined total of @3 + 2 + 1 = 6@ seconds:

613613--

614614-- @

615−-- main = S.'drain' . 'wSerially' $ delay 3 <> delay 2 <> delay 1

615+-- main = S.'drain' . S.'wSerially' $ delay 3 <> delay 2 <> delay 1

616616-- @

617617-- @

618618-- ThreadId 36: Delay 3

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

626626-- combinator in the following example:

627627--

628628-- @

629−-- main = S.'drain' $ (print 1 |: print 2 |: nil) \`wSerial` (print 3 |: print 4 |: nil)

629+-- main = S.'drain' $ (print 1 |: print 2 |: S.'nil') \`S.wSerial` (print 3 |: print 4 |: S.'nil')

630630-- @

631631-- @

632632-- 1

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

656656--

657657-- @

658658-- main = do

659−-- xs \<- S.'toList' . 'aheadly' $ (p 1 |: p 2 |: nil) <> (p 3 |: p 4 |: nil)

659+-- xs \<- S.'toList' . S.'aheadly' $ (p 1 |: p 2 |: S.nil) <> (p 3 |: p 4 |: S.nil)

660660-- print xs

661661-- where p n = threadDelay 1000000 >> return n

662662-- @

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

693693--

694694-- @

695695-- main = do

696−-- xs \<- S.'toList' . 'asyncly' $ (p 1 |: p 2 |: nil) <> (p 3 |: p 4 |: nil)

696+-- xs \<- S.'toList' . S.'asyncly' $ (p 1 |: p 2 |: S.'nil') <> (p 3 |: p 4 |: S.'nil')

697697-- print xs

698698-- where p n = threadDelay 1000000 >> return n

699699-- @

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

708708--

709709-- @

710710-- main = do

711−-- xs \<- S.'toList' . 'asyncly' $ (serially $ p 1 |: p 2 |: nil) <> (serially $ p 3 |: p 4 |: nil)

711+-- xs \<- S.'toList' . S.'asyncly' $ (S.'serially' $ p 1 |: p 2 |: S.'nil') <> (serially $ p 3 |: p 4 |: S.'nil')

712712-- print xs

713713-- where p n = threadDelay 1000000 >> return n

714714-- @

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

722722-- (3, 2, 1) = 3@ seconds:

723723--

724724-- @

725−-- main = S.'drain' . 'asyncly' $ delay 3 '<>' delay 2 '<>' delay 1

725+-- main = S.'drain' . S.'asyncly' $ delay 3 '<>' delay 2 '<>' delay 1

726726-- @

727727-- @

728728-- ThreadId 42: Delay 1

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

741741-- second as all of the actions are concurrent.

742742--

743743-- @

744−-- main = S.'drain' . 'asyncly' $ (delay 1 <> delay 2) <> (delay 3 <> delay 4)

744+-- main = S.'drain' . S.'asyncly' $ (delay 1 <> delay 2) <> (delay 3 <> delay 4)

745745-- @

746746-- @

747747-- 1

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

758758-- even if none of them blocks:

759759--

760760-- @

761−-- main = S.'drain' . 'asyncly' $ traced (sqrt 9) '<>' traced (sqrt 16) '<>' traced (sqrt 25)

761+-- main = S.'drain' . S.'asyncly' $ traced (sqrt 9) '<>' traced (sqrt 16) '<>' traced (sqrt 25)

762762-- where traced m = S.'yieldM' (myThreadId >>= print) >> return m

763763-- @

764764-- @

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

776776-- not used the 'asyncly' combinator in the following example:

777777--

778778-- @

779−-- main = S.'drain' $ delay 3 \`async` delay 2 \`async` delay 1

779+-- main = S.'drain' $ delay 3 \`S.'async'` delay 2 \`S.'async'` delay 1

780780-- @

781781-- @

782782-- ThreadId 42: Delay 1

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

819819-- first traversal order but this is not guaranteed.

820820--

821821-- @

822−-- main = S.'drain' . 'wAsyncly' $ (serially $ print 1 |: print 2 |: nil) <> (serially $ print 3 |: print 4 |: nil)

822+-- main = S.'drain' . S.'wAsyncly' $ (S.'serially' $ print 1 |: print 2 |: S.'nil') <> (S.'serially' $ print 3 |: print 4 |: S.'nil')

823823-- @

824824-- @

825825-- 1

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

834834-- the 'wAsyncly' combinator in the following example:

835835--

836836-- @

837−-- main = S.'drain' $ delay 3 \`wAsync` delay 2 \`wAsync` delay 1

837+-- main = S.'drain' $ delay 3 \S.`wAsync` delay 2 \S.`wAsync` delay 1

838838-- @

839839-- @

840840-- ThreadId 42: Delay 1

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

873873-- package to run this example:

874874--

875875-- @

876−-- import "Streamly"

877876-- import qualified Streamly.Prelude as S

878877-- import Network.HTTP.Simple

879878--

880−-- main = S.'drain' . 'parallely' $ google \<> bing \<> duckduckgo

879+-- main = S.'drain' . S.'parallely' $ google \<> bing \<> duckduckgo

881880-- where

882881-- google = get "https://www.google.com/search?q=haskell"

883882-- bing = get "https://www.bing.com/search?q=haskell"

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

891890-- 'parallely' combinator in the following example:

892891--

893892-- @

894−-- main = S.'drain' $ delay 3 \`parallel` delay 2 \`wAsync` delay 1

893+-- main = S.'drain' $ delay 3 \`S.'parallel'` delay 2 \`S.'wAsync'` delay 1

895894-- @

896895-- @

897896-- ThreadId 42: Delay 1

@@ -921,27 +920,26 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

921920-- style using 'fold' or 'foldMap'. We have also provided some fold utilities

922921-- to fold streams using the polymorphic combine operations:

923922--

924−-- * 'foldWith' is like 'fold', it folds a 'Foldable' container of streams

925−-- using the given composition operator.

926−-- * 'foldMapWith' is like 'foldMap', it folds like @foldWith@ but also maps a

927−-- function before folding.

928−-- * 'forEachWith' is like @foldMapwith@ but the container argument comes before

929−-- the function argument.

923+-- * 'concatFoldableWith' is like 'fold', it folds a 'Foldable' container of

924+-- streams using the given composition operator.

925+-- * 'concatMapFoldableWith' is like 'foldMap', it folds like

926+-- @concatFoldableWith@ but also maps a function before folding.

927+-- * 'concatForFoldableWith' is like @concatMapFoldableWith@ but the container

928+-- argument comes before the function argument.

930929--

931930-- All of the following are equivalent and start ten concurrent tasks each with

932931-- a delay from 1 to 10 seconds, resulting in the printing of each number every

933932-- second:

934933--

935934-- @

936−-- import "Streamly"

937935-- import qualified "Streamly.Prelude" as S

938936-- import Control.Concurrent

939937--

940938-- main = do

941−-- S.'drain' $ 'asyncly' $ foldMap delay [1..10]

942−-- S.'drain' $ S.'foldWith' 'async' (map delay [1..10])

943−-- S.'drain' $ S.'foldMapWith' 'async' delay [1..10]

944−-- S.'drain' $ S.'forEachWith' 'async' [1..10] delay

939+-- S.'drain' $ S.'asyncly' $ foldMap delay [1..10]

940+-- S.'drain' $ S.'concatFoldableWith' S.'async' (map delay [1..10])

941+-- S.'drain' $ S.'concatMapFoldableWith' S.'async' delay [1..10]

942+-- S.'drain' $ S.'concatForFoldableWith' S.'async' [1..10] delay

945943-- where delay n = S.'yieldM' $ threadDelay (n * 1000000) >> print n

946944-- @

947945

@@ -987,7 +985,6 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

987985-- instead.

988986--

989987-- @

990−-- import "Streamly"

991988-- import qualified "Streamly.Prelude" as S

992989--

993990-- main = S.'drain' $ do

@@ -1009,12 +1006,11 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

10091006-- written like this:

10101007--

10111008-- @

1012−-- import "Streamly"

10131009-- import qualified "Streamly.Prelude" as S

10141010--

10151011-- import Control.Monad (forever)

10161012--

1017−-- main = S.'drain' $ forever $ S.yieldM getLine >>= S.yieldM . putStrLn

1013+-- main = S.'drain' $ forever $ S.'yieldM' getLine >>= S.'yieldM' . putStrLn

10181014-- @

10191015--

10201016-- When multiple streams are composed using this style they nest in a DFS

@@ -1023,7 +1019,6 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

10231019-- loops in imperative programming.

10241020--

10251021-- @

1026−-- import "Streamly"

10271022-- import qualified "Streamly.Prelude" as S

10281023--

10291024-- main = S.'drain' $ do

@@ -1052,10 +1047,9 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

10521047-- type of the stream as 'Ahead'.

10531048--

10541049-- @

1055−-- import "Streamly"

10561050-- import qualified "Streamly.Prelude" as S

10571051--

1058−-- comp = S.'toList' . 'aheadly' $ do

1052+-- comp = S.'toList' . S.'aheadly' $ do

10591053-- x <- S.'fromFoldable' [3,2,1]

10601054-- delay x >> return x

10611055--

@@ -1093,10 +1087,9 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

10931087-- type of the stream as 'Async'.

10941088--

10951089-- @

1096−-- import "Streamly"

10971090-- import qualified "Streamly.Prelude" as S

10981091--

1099−-- main = S.'drain' . 'asyncly' $ do

1092+-- main = S.'drain' . S.'asyncly' $ do

11001093-- x <- S.'fromFoldable' [3,2,1]

11011094-- delay x

11021095-- @

@@ -1120,10 +1113,9 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

11201113-- on the demand from the consumer of the stream.

11211114--

11221115-- @

1123−-- import "Streamly"

11241116-- import qualified "Streamly.Prelude" as S

11251117--

1126−-- main = S.'drain' . 'asyncly' $ do

1118+-- main = S.'drain' . S.'asyncly' $ do

11271119-- x <- S.'fromFoldable' [1,2]

11281120-- y <- S.'fromFoldable' [3,4]

11291121-- S.'yieldM' $ putStrLn $ show (x, y)

@@ -1145,10 +1137,9 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

11451137-- type of the stream as 'WSerial'.

11461138--

11471139-- @

1148−-- import "Streamly"

11491140-- import qualified "Streamly.Prelude" as S

11501141--

1151−-- main = S.'drain' . 'wSerially' $ do

1142+-- main = S.'drain' . S.'wSerially' $ do

11521143-- x <- S.'fromFoldable' [1,2]

11531144-- y <- S.'fromFoldable' [3,4]

11541145-- S.yieldM $ putStrLn $ show (x, y)

@@ -1173,10 +1164,9 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

11731164-- specify the type of the stream as 'WAsync'.

11741165--

11751166-- @

1176−-- import "Streamly"

11771167-- import qualified "Streamly.Prelude" as S

11781168--

1179−-- main = S.'drain' . 'wAsyncly' $ do

1169+-- main = S.'drain' . S.'wAsyncly' $ do

11801170-- x <- S.'fromFoldable' [1,2]

11811171-- y <- S.'fromFoldable' [3,4]

11821172-- S.'yieldM' $ putStrLn $ show (x, y)

@@ -1199,10 +1189,9 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

11991189-- annotation can be used to specify the type of the stream as 'Parallel'.

12001190--

12011191-- @

1202−-- import "Streamly"

12031192-- import qualified "Streamly.Prelude" as S

12041193--

1205−-- main = S.'drain' . 'parallely' $ do

1194+-- main = S.'drain' . S.'parallely' $ do

12061195-- x <- S.'fromFoldable' [3,2,1]

12071196-- delay x

12081197-- @

@@ -1220,10 +1209,9 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

12201209-- specific mode of composition. For example take a look at the following code.

12211210--

12221211-- @

1223−-- import "Streamly"

12241212-- import qualified "Streamly.Prelude" as S

12251213--

1226−-- composed :: (IsStream t, Monad (t IO)) => t IO ()

1214+-- composed :: (S.IsStream t, Monad (t IO)) => t IO ()

12271215-- composed = do

12281216-- sz <- sizes

12291217-- cl <- colors

@@ -1240,11 +1228,11 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

12401228-- Now we can interpret this in whatever way we want:

12411229--

12421230-- @

1243−-- main = S.'drain' . 'serially' $ composed

1244−-- main = S.'drain' . 'wSerially' $ composed

1245−-- main = S.'drain' . 'asyncly' $ composed

1246−-- main = S.'drain' . 'wAsyncly' $ composed

1247−-- main = S.'drain' . 'parallely' $ composed

1231+-- main = S.'drain' . S.'serially' $ composed

1232+-- main = S.'drain' . S.'wSerially' $ composed

1233+-- main = S.'drain' . S.'asyncly' $ composed

1234+-- main = S.'drain' . S.'wAsyncly' $ composed

1235+-- main = S.'drain' . S.'parallely' $ composed

12481236-- @

12491237--

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

@@ -1257,7 +1245,6 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

12571245-- serial.

12581246--

12591247-- @

1260−-- import "Streamly"

12611248-- import qualified "Streamly.Prelude" as S

12621249--

12631250-- main = (S.'toList' $ fmap show $ S.'fromFoldable' [1..10]) >>= print

@@ -1276,7 +1263,6 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

12761263-- serially and takes a total 17 seconds (1 + 3 + 4 + 2 + 3 + 4):

12771264--

12781265-- @

1279−-- import "Streamly"

12801266-- import qualified "Streamly.Prelude" as S

12811267-- import Control.Concurrent

12821268--

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

12841270-- s2 = d 3 <> d 4

12851271-- d n = delay n >> return n

12861272--

1287−-- main = (S.'toList' . 'serially' $ (,) \<$> s1 \<*> s2) >>= print

1273+-- main = (S.'toList' . S.'serially' $ (,) \<$> s1 \<*> s2) >>= print

12881274-- @

12891275-- @

12901276-- ThreadId 36: Delay 1

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

13001286-- order but since it is serial it takes a total of 17 seconds:

13011287--

13021288-- @

1303−-- main = (S.'toList' . 'wSerially' $ (,) \<$> s1 \<*> s2) >>= print

1289+-- main = (S.'toList' . S.'wSerially' $ (,) \<$> s1 \<*> s2) >>= print

13041290-- @

13051291-- @

13061292-- ThreadId 36: Delay 1

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

13161302-- of 6 seconds which is max (1, 2) + max (3, 4):

13171303--

13181304-- @

1319−-- main = (S.'toList' . 'asyncly' $ (,) \<$> s1 \<*> s2) >>= print

1305+-- main = (S.'toList' . S.'asyncly' $ (,) \<$> s1 \<*> s2) >>= print

13201306-- @

13211307-- @

13221308-- ThreadId 34: Delay 1

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

13321318-- therefore takes a total of 6 seconds (2 + 4):

13331319--

13341320-- @

1335−-- main = (S.'toList' . 'wAsyncly' $ (,) \<$> s1 \<*> s2) >>= print

1321+-- main = (S.'toList' . S.'wAsyncly' $ (,) \<$> s1 \<*> s2) >>= print

13361322-- @

13371323-- @

13381324-- ThreadId 34: Delay 1

@@ -1360,15 +1346,14 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

13601346-- zip composition:

13611347--

13621348-- @

1363−-- import "Streamly"

13641349-- import qualified "Streamly.Prelude" as S

13651350-- import Control.Concurrent

13661351--

13671352-- d n = delay n >> return n

1368−-- s1 = 'serially' $ d 1 <> d 2

1369−-- s2 = 'serially' $ d 3 <> d 4

1353+-- s1 = S.'serially' $ d 1 <> d 2

1354+-- s2 = S.'serially' $ d 3 <> d 4

13701355--

1371−-- main = (S.'toList' . 'zipSerially' $ (,) \<$> s1 \<*> s2) >>= print

1356+-- main = (S.'toList' . S.'zipSerially' $ (,) \<$> s1 \<*> s2) >>= print

13721357-- @

13731358--

13741359-- This takes total 10 seconds to zip, which is (1 + 2 + 3 + 4) since

@@ -1390,18 +1375,17 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

13901375--

13911376--

13921377-- @

1393−-- import "Streamly"

13941378-- import qualified "Streamly.Prelude" as S

13951379-- import Control.Concurrent

13961380-- import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))

13971381--

13981382-- d n = delay n >> return n

1399−-- s1 = 'serially' $ d 1 <> d 2

1400−-- s2 = 'serially' $ d 3 <> d 4

1383+-- s1 = S.'serially' $ d 1 <> d 2

1384+-- s2 = S.'serially' $ d 3 <> d 4

14011385--

14021386-- main = do

14031387-- hSetBuffering stdout LineBuffering

1404−-- (S.'toList' . 'zipAsyncly' $ (,) \<$> s1 \<*> s2) >>= print

1388+-- (S.'toList' . S.'zipAsyncly' $ (,) \<$> s1 \<*> s2) >>= print

14051389-- @

14061390--

14071391-- This takes 7 seconds to zip, which is max (1,3) + max (2,4) because 1 and 3

@@ -1432,17 +1416,16 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

14321416-- combinator instead of explicitly folding with 'async'.

14331417--

14341418-- @

1435−-- import "Streamly"

14361419-- import qualified "Streamly.Prelude" as S

14371420-- import Data.List (sum)

14381421--

14391422-- main = do

14401423-- z \<- S.'toList'

1441−-- $ 'serially' -- Serial monadic processing (sqrt below)

1424+-- $ S.'serially' -- Serial monadic processing (sqrt below)

14421425-- $ do

1443−-- x2 \<- 'forEachWith' 'async' [1..100] $ -- Concurrent @"for"@ loop

1426+-- x2 \<- S.'concatForFoldableWith' S.'async' [1..100] $ -- Concurrent @"for"@ loop

14441427-- \\x -> return $ x * x -- body of the loop

1445−-- y2 \<- 'forEachWith' 'async' [1..100] $

1428+-- y2 \<- S.'concatForFoldableWith' S.'async' [1..100] $

14461429-- \\y -> return $ y * y

14471430-- return $ sqrt (x2 + y2)

14481431-- print $ sum z

@@ -1477,11 +1460,11 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

14771460-- @

14781461-- {-\# LANGUAGE FlexibleContexts #-}

14791462--

1480−-- import "Streamly"

1463+-- import "Streamly.Prelude" (MonadAsync, SerialT)

14811464-- import "Streamly.Prelude" as S

1482−-- import Control.Monad (void, when)

1465+-- import Control.Monad (void)

14831466-- import Control.Monad.IO.Class (MonadIO(liftIO))

1484−-- import Control.Monad.State (MonadState, get, modify, runStateT, put)

1467+-- import Control.Monad.State (MonadState, get, modify, runStateT)

14851468--

14861469-- data Event = Quit | Harm Int | Heal Int deriving (Show)

14871470--

@@ -1497,13 +1480,13 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

14971480-- _ -> putStrLn "Type potion or harm or quit" >> askUser

14981481--

14991482-- acidRain :: MonadAsync m => 'SerialT' m Event

1500−-- acidRain = 'asyncly' $ 'constRate' 1 $ S.'repeatM' $ liftIO $ return $ Harm 1

1483+-- acidRain = S.'asyncly' $ S.'constRate' 1 $ S.'repeatM' $ liftIO $ return $ Harm 1

15011484--

15021485-- data Result = Check | Done

15031486--

15041487-- runEvents :: (MonadAsync m, MonadState Int m) => 'SerialT' m Result

15051488-- runEvents = do

1506−-- event \<- userAction \`parallel` acidRain

1489+-- event \<- userAction \`S.'parallel'` acidRain

15071490-- case event of

15081491-- Harm n -> modify (\\h -> h - n) >> return Check

15091492-- Heal n -> modify (\\h -> h + n) >> return Check

@@ -1556,12 +1539,12 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

15561539-- Interop with @vector@:

15571540--

15581541-- @

1559−-- import Streamly

1560−-- import qualified Streamly.Prelude as S

1542+-- import "Streamly.Prelude" (SerialT)

1543+-- import qualified "Streamly.Prelude" as S

15611544-- import qualified Data.Vector.Fusion.Stream.Monadic as V

15621545--

15631546-- -- | vector to streamly

1564−-- fromVector :: (IsStream t, Monad m) => V.Stream m a -> t m a

1547+-- fromVector :: (S.IsStream t, Monad m) => V.Stream m a -> t m a

15651548-- fromVector = S.unfoldrM unconsV

15661549-- where

15671550-- unconsV v = do

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

15741557--

15751558-- -- | streamly to vector

15761559-- toVector :: Monad m => SerialT m a -> V.Stream m a

1577−-- toVector = V.unfoldrM (S.uncons . adapt)

1560+-- toVector = V.unfoldrM (S.uncons . S.adapt)

15781561--

15791562-- main = do

15801563-- S.toList (fromVector (V.fromList [1..3])) >>= print

@@ -1584,13 +1567,13 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

15841567-- Interop with @pipes@:

15851568--

15861569-- @

1587−-- import "Streamly"

1570+-- import "Streamly.Prelude" (SerialT)

15881571-- import qualified "Streamly.Prelude" as S

15891572-- import qualified Pipes as P

15901573-- import qualified Pipes.Prelude as P

15911574--

15921575-- -- | pipes to streamly

1593−-- fromPipes :: (IsStream t, Monad m) => P.Producer a m r -> t m a

1576+-- fromPipes :: (S.IsStream t, Monad m) => P.Producer a m r -> t m a

15941577-- fromPipes = S.'unfoldrM' unconsP

15951578-- where

15961579-- -- Adapt P.next to return a Maybe instead of Either

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

16111594-- Interop with @streaming@:

16121595--

16131596-- @

1614−-- import "Streamly"

1597+-- import "Streamly.Prelude" (SerialT, MonadAsync)

16151598-- import qualified "Streamly.Prelude" as S

16161599-- import qualified Streaming as SG

16171600-- import qualified Streaming.Prelude as SG

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

16351618-- Interop with @conduit@:

16361619--

16371620-- @

1638−-- import "Streamly"

1621+-- import "Streamly.Prelude" (SerialT)

16391622-- import qualified "Streamly.Prelude" as S

16401623-- import qualified Data.Conduit as C

16411624-- import qualified Data.Conduit.List as C

test/Main.hs+38−34

@@ -13,14 +13,18 @@ import Data.Foldable (forM_, fold)

1313import Data.Function ((&))

1414import Data.List (sort)

1515import Data.Maybe (fromJust, isJust)

16+#if !(MIN_VERSION_base(4,11,0))

17+import Data.Semigroup (Semigroup(..))

18+#endif

1619import System.Mem (performMajorGC)

1720

1821import Data.IORef

1922import Test.Hspec as H

2023

21−import Streamly

22−import Streamly.Prelude ((.:), nil)

23−import qualified Streamly as S

24+import Streamly.Prelude

25+ ( AsyncT, SerialT, WSerialT, MonadAsync, IsStream, (.:), nil, adapt

26+ , ahead, async, aheadly, asyncly, serially, parallely, wAsync, wAsyncly

27+ , wSerially, maxBuffer)

2428import qualified Streamly.Prelude as S

2529import qualified Streamly.Internal.Data.Stream.IsStream as IP

2630

@@ -478,7 +482,7 @@ parallelTests = H.parallel $ do

478482 it "wAsync maintains independent states in concurrent tasks"

479483 (monadicStateSnapshotOp wAsync)

480484 it "parallel maintains independent states in concurrent tasks"

481− (monadicStateSnapshotOp Streamly.parallel)

485+ (monadicStateSnapshotOp S.parallel)

482486

483487 ---------------------------------------------------------------------------

484488 -- Slower tests are at the end

@@ -757,8 +761,8 @@ composeWithMonadThrow t = do

757761

758762 oneLevelNestedProduct desc t1 =

759763 it ("One level nested product" <> desc) $ do

760− let s1 = t $ S.foldMapWith (<>) return [1..4]

761− s2 = t1 $ S.foldMapWith (<>) return [5..8]

764+ let s1 = t $ S.concatMapFoldableWith (<>) return [1..4]

765+ s2 = t1 $ S.concatMapFoldableWith (<>) return [5..8]

762766 try $ tl (do

763767 x <- adapt s1

764768 y <- s2

@@ -783,8 +787,8 @@ _composeWithMonadError t = do

783787

784788nestTwoSerial :: Expectation

785789nestTwoSerial =

786− let s1 = S.foldMapWith (<>) return [1..4]

787− s2 = S.foldMapWith (<>) return [5..8]

790+ let s1 = S.concatMapFoldableWith (<>) return [1..4]

791+ s2 = S.concatMapFoldableWith (<>) return [5..8]

788792 in toListSerial (do

789793 x <- s1

790794 y <- s2

@@ -793,8 +797,8 @@ nestTwoSerial =

793797

794798nestTwoAhead :: Expectation

795799nestTwoAhead =

796− let s1 = S.foldMapWith (<>) return [1..4]

797− s2 = S.foldMapWith (<>) return [5..8]

800+ let s1 = S.concatMapFoldableWith (<>) return [1..4]

801+ s2 = S.concatMapFoldableWith (<>) return [5..8]

798802 in (S.toList . aheadly) (do

799803 x <- s1

800804 y <- s2

@@ -803,22 +807,22 @@ nestTwoAhead =

803807

804808nestTwoSerialApp :: Expectation

805809nestTwoSerialApp =

806− let s1 = S.foldMapWith (<>) return [1..4]

807− s2 = S.foldMapWith (<>) return [5..8]

810+ let s1 = S.concatMapFoldableWith (<>) return [1..4]

811+ s2 = S.concatMapFoldableWith (<>) return [5..8]

808812 in toListSerial ((+) <$> s1 <*> s2)

809813 `shouldReturn` ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])

810814

811815nestTwoAheadApp :: Expectation

812816nestTwoAheadApp =

813− let s1 = S.foldMapWith (<>) return [1..4]

814− s2 = S.foldMapWith (<>) return [5..8]

817+ let s1 = S.concatMapFoldableWith (<>) return [1..4]

818+ s2 = S.concatMapFoldableWith (<>) return [5..8]

815819 in (S.toList . aheadly) ((+) <$> s1 <*> s2)

816820 `shouldReturn` ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])

817821

818822nestTwoInterleaved :: Expectation

819823nestTwoInterleaved =

820− let s1 = S.foldMapWith (<>) return [1..4]

821− s2 = S.foldMapWith (<>) return [5..8]

824+ let s1 = S.concatMapFoldableWith (<>) return [1..4]

825+ s2 = S.concatMapFoldableWith (<>) return [5..8]

822826 in toListInterleaved (do

823827 x <- s1

824828 y <- s2

@@ -827,15 +831,15 @@ nestTwoInterleaved =

827831

828832nestTwoInterleavedApp :: Expectation

829833nestTwoInterleavedApp =

830− let s1 = S.foldMapWith (<>) return [1..4]

831− s2 = S.foldMapWith (<>) return [5..8]

834+ let s1 = S.concatMapFoldableWith (<>) return [1..4]

835+ s2 = S.concatMapFoldableWith (<>) return [5..8]

832836 in toListInterleaved ((+) <$> s1 <*> s2)

833837 `shouldReturn` ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])

834838

835839nestTwoAsync :: Expectation

836840nestTwoAsync =

837− let s1 = S.foldMapWith (<>) return [1..4]

838− s2 = S.foldMapWith (<>) return [5..8]

841+ let s1 = S.concatMapFoldableWith (<>) return [1..4]

842+ s2 = S.concatMapFoldableWith (<>) return [5..8]

839843 in sort <$> toListAsync (do

840844 x <- s1

841845 y <- s2

@@ -844,15 +848,15 @@ nestTwoAsync =

844848

845849nestTwoAsyncApp :: Expectation

846850nestTwoAsyncApp =

847− let s1 = S.foldMapWith (<>) return [1..4]

848− s2 = S.foldMapWith (<>) return [5..8]

851+ let s1 = S.concatMapFoldableWith (<>) return [1..4]

852+ s2 = S.concatMapFoldableWith (<>) return [5..8]

849853 in sort <$> toListAsync ((+) <$> s1 <*> s2)

850854 `shouldReturn` sort ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])

851855

852856nestTwoWAsync :: Expectation

853857nestTwoWAsync =

854− let s1 = S.foldMapWith (<>) return [1..4]

855− s2 = S.foldMapWith (<>) return [5..8]

858+ let s1 = S.concatMapFoldableWith (<>) return [1..4]

859+ s2 = S.concatMapFoldableWith (<>) return [5..8]

856860 in sort <$> (S.toList . wAsyncly) (do

857861 x <- s1

858862 y <- s2

@@ -861,8 +865,8 @@ nestTwoWAsync =

861865

862866nestTwoParallel :: Expectation

863867nestTwoParallel =

864− let s1 = S.foldMapWith (<>) return [1..4]

865− s2 = S.foldMapWith (<>) return [5..8]

868+ let s1 = S.concatMapFoldableWith (<>) return [1..4]

869+ s2 = S.concatMapFoldableWith (<>) return [5..8]

866870 in sort <$> (S.toList . parallely) (do

867871 x <- s1

868872 y <- s2

@@ -871,15 +875,15 @@ nestTwoParallel =

871875

872876nestTwoWAsyncApp :: Expectation

873877nestTwoWAsyncApp =

874− let s1 = S.foldMapWith (<>) return [1..4]

875− s2 = S.foldMapWith (<>) return [5..8]

878+ let s1 = S.concatMapFoldableWith (<>) return [1..4]

879+ s2 = S.concatMapFoldableWith (<>) return [5..8]

876880 in sort <$> (S.toList . wAsyncly) ((+) <$> s1 <*> s2)

877881 `shouldReturn` sort ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])

878882

879883nestTwoParallelApp :: Expectation

880884nestTwoParallelApp =

881− let s1 = S.foldMapWith (<>) return [1..4]

882− s2 = S.foldMapWith (<>) return [5..8]

885+ let s1 = S.concatMapFoldableWith (<>) return [1..4]

886+ s2 = S.concatMapFoldableWith (<>) return [5..8]

883887 in sort <$> (S.toList . parallely) ((+) <$> s1 <*> s2)

884888 `shouldReturn` sort ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])

885889

@@ -918,7 +922,7 @@ composeAndComposeSimple

918922 -> (t2 IO Int -> t2 IO Int)

919923 -> [[Int]] -> Spec

920924composeAndComposeSimple t1 t2 answer = do

921− let rfold = adapt . t2 . S.foldMapWith (<>) return

925+ let rfold = adapt . t2 . S.concatMapFoldableWith (<>) return

922926 it "Compose right associated outer expr, right folded inner" $

923927 (S.toList . t1) (rfold [1,2,3] <> (rfold [4,5,6] <> rfold [7,8,9]))

924928 `shouldReturn` head answer

@@ -966,10 +970,10 @@ bindAndComposeSimple

966970 -> (t2 IO Int -> t2 IO Int)

967971 -> Spec

968972bindAndComposeSimple t1 t2 = do

969− -- XXX need a bind in the body of forEachWith instead of a simple return

973+ -- XXX need a bind in the body of concatForFoldableWith instead of a simple return

970974 it "Compose many (right fold) with bind" $

971975 (sort <$> (S.toList . t1)

972− (adapt . t2 $ S.forEachWith (<>) [1..10 :: Int] return))

976+ (adapt . t2 $ S.concatForFoldableWith (<>) [1..10 :: Int] return))

973977 `shouldReturn` [1..10]

974978

975979 it "Compose many (left fold) with bind" $

@@ -1079,7 +1083,7 @@ testFromCallback :: IO Int

10791083testFromCallback = do

10801084 ref <- newIORef Nothing

10811085 let stream = S.map Just (IP.fromCallback (setCallback ref))

1082− `Streamly.parallel` runCallback ref

1086+ `S.parallel` runCallback ref

10831087 S.sum $ S.map fromJust $ S.takeWhile isJust stream

10841088

10851089 where

test/PureStreams.hs+4−2

@@ -4,15 +4,17 @@

55module Main (main) where

7+#if !(MIN_VERSION_base(4,11,0))

8+import Data.Semigroup ((<>))

9+#endif

710import Test.Hspec

811import qualified GHC.Exts as GHC

912

10−import Streamly

11−

1213#ifdef USE_STREAMLY_LIST

1314import Data.Functor.Identity

1415import Streamly.Internal.Data.List (List(..), pattern Cons, pattern Nil, ZipList(..),

1516 fromZipList, toZipList)

17+import Streamly.Prelude (SerialT)

1618import qualified Streamly.Prelude as S

1719#else

1820import Prelude -- to suppress compiler warning

test/Streamly/Test/Array.hs+1−2

@@ -17,8 +17,7 @@ import Test.QuickCheck.Monadic (monadicIO, assert, run)

1717

1818import Test.Hspec as H

1919

20−import Streamly (SerialT)

21−

20+import Streamly.Prelude (SerialT)

2221import qualified Streamly.Prelude as S

2322

2423#ifdef TEST_SMALL_ARRAY

test/Streamly/Test/FileSystem/Event.hs+1−1

@@ -17,7 +17,7 @@ import Data.List.NonEmpty (NonEmpty)

1717import Data.Word (Word8)

1818import Streamly.Internal.Data.Array.Storable.Foreign (Array)

1919import System.Environment (getArgs)

20−import Streamly (SerialT)

20+import Streamly.Prelude (SerialT)

2121

2222import qualified Data.List.NonEmpty as NonEmpty

2323import qualified Streamly.Unicode.Stream as Unicode

test/Streamly/Test/Prelude.hs+8−8

@@ -80,7 +80,7 @@ import Data.List

8080 )

8181import Data.Maybe (mapMaybe)

8282#if __GLASGOW_HASKELL__ < 808

83−import Data.Semigroup ((<>))

83+import Data.Semigroup (Semigroup, (<>))

8484#endif

8585import GHC.Word (Word8)

8686import Test.Hspec.QuickCheck

@@ -88,9 +88,9 @@ import Test.Hspec

8888import Test.QuickCheck (Property, choose, forAll, withMaxSuccess)

8989import Test.QuickCheck.Monadic (assert, monadicIO, run)

9090

91−import Streamly

92−import Streamly.Prelude ((.:), nil)

93−import Streamly as S

91+import Streamly.Prelude

92+ ( SerialT, IsStream, (.:), nil, (|&), serially, avgRate, rate, maxBuffer

93+ , maxThreads, maxBuffer)

9494import qualified Streamly.Prelude as S

9595import qualified Streamly.Data.Fold as FL

9696import qualified Streamly.Internal.Data.Fold as FL

@@ -502,7 +502,7 @@ monoidOps desc z eq t = do

502502 prop (desc <> " Compose empty at the end") $ spec (singleton 1 <> z) [1]

503503 prop (desc <> " Compose two") $ spec (singleton 0 <> singleton 1) [0, 1]

504504 prop (desc <> " Compose many") $

505− spec (S.forEachWith (<>) [1 .. 100] singleton) [1 .. 100]

505+ spec (S.concatForFoldableWith (<>) [1 .. 100] singleton) [1 .. 100]

506506

507507 -- These are not covered by the property tests

508508 prop (desc <> " Compose three - empty in the middle") $

@@ -553,8 +553,8 @@ semigroupOps

553553 -> (t IO Int -> SerialT IO Int)

554554 -> Spec

555555semigroupOps desc eq t = do

556− prop (desc <> " <>") $ foldFromList (S.foldMapWith (<>) singleton) t eq

557− prop (desc <> " mappend") $ foldFromList (S.foldMapWith mappend singleton) t eq

556+ prop (desc <> " <>") $ foldFromList (S.concatMapFoldableWith (<>) singleton) t eq

557+ prop (desc <> " mappend") $ foldFromList (S.concatMapFoldableWith mappend singleton) t eq

558558

559559-------------------------------------------------------------------------------

560560-- Transformation operations

@@ -850,7 +850,7 @@ folded =

850850 (\xs ->

851851 case xs of

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

853− _ -> S.foldMapWith (<>) return xs)

853+ _ -> S.concatMapFoldableWith (<>) return xs)

854854

855855makeCommonOps :: IsStream t => (t m a -> c) -> [(String, t m a -> c)]

856856makeCommonOps t =

test/Streamly/Test/Prelude/Ahead.hs+1−1

@@ -17,7 +17,7 @@ import Test.Hspec.QuickCheck

1717import Test.QuickCheck.Monadic (monadicIO, run)

1818import Test.Hspec as H

1919

20−import Streamly

20+import Streamly.Prelude

2121import qualified Streamly.Prelude as S

2222

2323import Streamly.Test.Common

test/Streamly/Test/Prelude/Async.hs+1−1

@@ -15,7 +15,7 @@ import Data.Semigroup ((<>))

1515import Test.Hspec.QuickCheck

1616import Test.Hspec as H

1717

18−import Streamly

18+import Streamly.Prelude

1919import qualified Streamly.Prelude as S

2020

2121import Streamly.Test.Prelude

test/Streamly/Test/Prelude/Concurrent.hs+1−1

@@ -27,7 +27,7 @@ import Test.QuickCheck

2727 (Property, withMaxSuccess)

2828import Test.QuickCheck.Monadic (monadicIO, run)

2929

30−import Streamly

30+import Streamly.Prelude hiding (replicateM, reverse)

3131import qualified Streamly.Prelude as S

3232

3333import Streamly.Test.Common

test/Streamly/Test/Prelude/Parallel.hs+1−1

@@ -15,7 +15,7 @@ import Data.Semigroup ((<>))

1515import Test.Hspec.QuickCheck

1616import Test.Hspec as H

1717

18−import Streamly

18+import Streamly.Prelude

1919import qualified Streamly.Prelude as S

2020

2121import Streamly.Test.Prelude

test/Streamly/Test/Prelude/Serial.hs+2−1

@@ -29,7 +29,8 @@ import Test.QuickCheck

2929import Test.QuickCheck.Monadic (assert, monadicIO, run)

3030import Test.Hspec as H

3131

32−import Streamly

32+import Streamly.Prelude

33+ ( SerialT, IsStream, avgRate, maxBuffer, serial, serially)

3334import qualified Streamly.Prelude as S

3435import qualified Streamly.Data.Fold as FL

3536

test/Streamly/Test/Prelude/WAsync.hs+1−1

@@ -15,7 +15,7 @@ import Data.Semigroup ((<>))

1515import Test.Hspec.QuickCheck

1616import Test.Hspec as H

1717

18−import Streamly

18+import Streamly.Prelude

1919import qualified Streamly.Prelude as S

2020

2121import Streamly.Test.Prelude

test/Streamly/Test/Prelude/WSerial.hs+1−1

@@ -17,7 +17,7 @@ import Test.Hspec.QuickCheck

1717import Test.QuickCheck.Monadic (monadicIO, run)

1818import Test.Hspec as H

1919

20−import Streamly

20+import Streamly.Prelude

2121import qualified Streamly.Prelude as S

2222

2323import Streamly.Test.Common

test/Streamly/Test/Prelude/ZipAsync.hs+1−1

@@ -12,7 +12,7 @@ module Streamly.Test.Prelude.ZipAsync where

1212import Test.Hspec.QuickCheck

1313import Test.Hspec as H

1414

15−import Streamly

15+import Streamly.Prelude

1616import qualified Streamly.Prelude as S

1717

1818import Streamly.Test.Prelude

test/Streamly/Test/Prelude/ZipSerial.hs+1−1

@@ -15,7 +15,7 @@ import Data.Semigroup ((<>))

1515import Test.Hspec.QuickCheck

1616import Test.Hspec as H

1717

18−import Streamly

18+import Streamly.Prelude

1919import qualified Streamly.Prelude as S

2020

2121import Streamly.Test.Prelude

test/loops.hs+4−2

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

1−import Streamly

1+#if !(MIN_VERSION_base(4,11,0))

2+import Data.Semigroup ((<>))

3+#endif

24import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))

3−import Streamly.Prelude (nil, yieldM, drain)

5+import Streamly.Prelude

57main :: IO ()

68main = do

test/nested-loops.hs+4−2

@@ -1,8 +1,10 @@

1+#if !(MIN_VERSION_base(4,11,0))

2+import Data.Semigroup ((<>))

3+#endif

14import Control.Concurrent (myThreadId)

25import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))

36import System.Random (randomIO)

4−import Streamly

5−import Streamly.Prelude (drain, nil, yieldM)

7+import Streamly.Prelude

79main :: IO ()

810main = drain $ do

test/parallel-loops.hs+1−1

@@ -1,7 +1,7 @@

11import Control.Concurrent (myThreadId, threadDelay)

22import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))

33import System.Random (randomIO)

4−import Streamly

4+import Streamly.Prelude

55import qualified Streamly.Prelude as S

77main :: IO ()