summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorharendra <>2018-03-25 02:13:00 (GMT)
committerhdiff <hdiff@hdiff.luite.com>2018-03-25 02:13:00 (GMT)
commit8ced2082b2adef42a267918db581fabb304f3112 (patch)
tree1f851d1528f11e6208965da122eb9f0383d70df5
parent83183a57cd05584bb23cbe70c01fb81b3b2b1a09 (diff)
version 0.1.20.1.2
-rw-r--r--Changelog.md9
-rw-r--r--README.md15
-rw-r--r--src/Streamly/Core.hs4
-rw-r--r--src/Streamly/Prelude.hs19
-rw-r--r--src/Streamly/Streams.hs4
-rw-r--r--streamly.cabal5
-rw-r--r--test/Main.hs18
7 files changed, 64 insertions, 10 deletions
diff --git a/Changelog.md b/Changelog.md
index b8c9865..3d562c7 100644
--- a/Changelog.md
+++ b/Changelog.md
@@ -1,3 +1,12 @@
+## 0.1.2
+
+### Enhancements
+* Add `iterate`, `iterateM` stream operations
+
+### Bug Fixes
+* Fixed a bug that casued unexpected behavior when `pure` was used to inject
+ values in Applicative composition of `ZipStream` and `ZipAsync` types.
+
## 0.1.1
### Enhancements
diff --git a/README.md b/README.md
index 61453cb..d135b53 100644
--- a/README.md
+++ b/README.md
@@ -14,15 +14,22 @@ whether the code runs serially or concurrently. It naturally integrates
concurrency with streaming rather than adding it as an afterthought.
Moreover, it interworks with the popular streaming libraries.
-See the haddock documentation for full reference. It is recommended to read
-the comprehensive tutorial module `Streamly.Tutorial` first. Also see
-`Streamly.Examples` for some working examples.
+See the haddock documentation for full reference. It is recommended that you
+read `Streamly.Tutorial` first. Also see `Streamly.Examples` for some working
+examples.
`Streamly` has best in class performance even though it generalizes streaming
to concurrent composition that does not mean it sacrifices non-concurrent
performance. See
[streaming-benchmarks](https://github.com/composewell/streaming-benchmarks) for
-detailed performance comparison with regular streaming libraries.
+detailed performance comparison with regular streaming libraries and the
+explanation of the benchmarks. The following graphs show a summary, the first
+one measures how four pipeline stages in a series perform, the second one
+measures the performance of individual stream operations; in both cases the
+stream processes a million elements:
+
+![Composing Pipeline Stages](charts/ComposingPipelineStages.svg)
+![All Operations at a Glance](charts/AllOperationsataGlance.svg)
## Non-determinism
diff --git a/src/Streamly/Core.hs b/src/Streamly/Core.hs
index 7ed7362..8a76311 100644
--- a/src/Streamly/Core.hs
+++ b/src/Streamly/Core.hs
@@ -25,6 +25,7 @@ module Streamly.Core
-- * Construction
, scons
+ , srepeat
, snil
-- * Composition
@@ -217,6 +218,9 @@ type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
scons :: a -> Maybe (Stream m a) -> Stream m a
scons a r = Stream $ \_ _ yld -> yld a r
+srepeat :: a -> Stream m a
+srepeat a = let x = scons a (Just x) in x
+
snil :: Stream m a
snil = Stream $ \_ stp _ -> stp
diff --git a/src/Streamly/Prelude.hs b/src/Streamly/Prelude.hs
index 57a727c..12f73c4 100644
--- a/src/Streamly/Prelude.hs
+++ b/src/Streamly/Prelude.hs
@@ -24,6 +24,8 @@ module Streamly.Prelude
, unfoldr
, unfoldrM
, each
+ , iterate
+ , iterateM
-- * Elimination
-- ** General Folds
@@ -85,7 +87,8 @@ import Prelude hiding (filter, drop, dropWhile, take,
mapM, mapM_, sequence, all, any,
sum, product, elem, notElem,
maximum, minimum, head, last,
- tail, length, null, reverse)
+ tail, length, null, reverse,
+ iterate)
import qualified Prelude
import qualified System.IO as IO
@@ -120,6 +123,20 @@ unfoldrM step = fromStream . go
each :: (Streaming t, Foldable f) => f a -> t m a
each = Prelude.foldr cons nil
+-- | Iterate a pure function from a seed value, streaming the results forever
+iterate :: Streaming t => (a -> a) -> a -> t m a
+iterate step = fromStream . go
+ where
+ go s = scons s (Just (go (step s)))
+
+-- | Iterate a monadic function from a seed value, streaming the results forever
+iterateM :: (Streaming t, Monad m) => (a -> m a) -> a -> t m a
+iterateM step = fromStream . go
+ where
+ go s = Stream $ \_ _ yld -> do
+ a <- step s
+ yld s (Just (go a))
+
-- | Read lines from an IO Handle into a stream of Strings.
fromHandle :: (Streaming t, MonadIO m) => IO.Handle -> t m String
fromHandle h = fromStream go
diff --git a/src/Streamly/Streams.hs b/src/Streamly/Streams.hs
index fd3a220..0e41076 100644
--- a/src/Streamly/Streams.hs
+++ b/src/Streamly/Streams.hs
@@ -751,7 +751,7 @@ instance Monad m => Functor (ZipStream m) where
in m Nothing stp yield
instance Monad m => Applicative (ZipStream m) where
- pure a = ZipStream $ scons a Nothing
+ pure = ZipStream . srepeat
(<*>) = zipWith id
instance Streaming ZipStream where
@@ -838,7 +838,7 @@ instance Monad m => Functor (ZipAsync m) where
in m Nothing stp yield
instance MonadAsync m => Applicative (ZipAsync m) where
- pure a = ZipAsync $ scons a Nothing
+ pure = ZipAsync . srepeat
(<*>) = zipAsyncWith id
instance Streaming ZipAsync where
diff --git a/streamly.cabal b/streamly.cabal
index a6e5fef..dd2e7f6 100644
--- a/streamly.cabal
+++ b/streamly.cabal
@@ -1,5 +1,5 @@
name: streamly
-version: 0.1.1
+version: 0.1.2
synopsis: Beautiful Streaming, Concurrent and Reactive Composition
description:
Streamly is a monad transformer unifying non-determinism
@@ -198,6 +198,7 @@ benchmark bench
-------------------------------------------------------------------------------
executable loops
+ default-language: Haskell2010
main-is: loops.hs
hs-source-dirs: examples
if flag(examples)
@@ -209,6 +210,7 @@ executable loops
buildable: False
executable nested-loops
+ default-language: Haskell2010
main-is: nested-loops.hs
hs-source-dirs: examples
if flag(examples)
@@ -221,6 +223,7 @@ executable nested-loops
buildable: False
executable parallel-loops
+ default-language: Haskell2010
main-is: parallel-loops.hs
hs-source-dirs: examples
if flag(examples)
diff --git a/test/Main.hs b/test/Main.hs
index e740b87..1907faa 100644
--- a/test/Main.hs
+++ b/test/Main.hs
@@ -422,8 +422,12 @@ zipOps z zM app = do
it "Applicative zip" $
let s1 = adapt $ serially $ foldMapWith (<>) return [1..10]
s2 = adapt $ serially $ foldMapWith (<>) return [1..]
- in (A.toList . app) ((+) <$> s1 <*> s2)
- `shouldReturn` ([2,4..20] :: [Int])
+ f = A.toList . app
+ functorial = f $ (+) <$> s1 <*> s2
+ applicative = f $ pure (+) <*> s1 <*> s2
+ expected = ([2,4..20] :: [Int])
+ in (,) <$> functorial <*> applicative
+ `shouldReturn` (expected, expected)
timed :: Int -> StreamT IO Int
timed x = liftIO (threadDelay (x * 100000)) >> return x
@@ -647,6 +651,16 @@ streamOperations (stream, list, len) = do
return $ str == lst
`shouldReturn` True
+ it "iterate" $
+ (A.toList . serially . (A.take len) $ (A.iterate (+ 1) (0 :: Int)))
+ `shouldReturn` (take len $ iterate (+ 1) 0)
+
+ it "iterateM" $ do
+ let addM = (\ y -> return (y + 1))
+ A.toList . serially . (A.take len) $ A.iterateM addM (0 :: Int)
+ `shouldReturn` (take len $ iterate (+ 1) 0)
+
+
-- Filtering
it "filter all out" $ transform (A.filter (> len)) (filter (> len))
it "filter all in" $ transform (A.filter (<= len)) (filter (<= len))