diff --git a/Changelog.md b/Changelog.md --- 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 --- a/README.md +++ b/README.md @@ -14,15 +14,22 @@ 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 --- a/src/Streamly/Core.hs +++ b/src/Streamly/Core.hs @@ -25,6 +25,7 @@ -- * Construction , scons + , srepeat , snil -- * Composition @@ -216,6 +217,9 @@ 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 --- a/src/Streamly/Prelude.hs +++ b/src/Streamly/Prelude.hs @@ -24,6 +24,8 @@ , unfoldr , unfoldrM , each + , iterate + , iterateM -- * Elimination -- ** General Folds @@ -85,7 +87,8 @@ 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 @@ -119,6 +122,20 @@ {-# INLINE each #-} 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 diff --git a/src/Streamly/Streams.hs b/src/Streamly/Streams.hs --- a/src/Streamly/Streams.hs +++ b/src/Streamly/Streams.hs @@ -751,7 +751,7 @@ 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 @@ 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 --- 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 @@ ------------------------------------------------------------------------------- executable loops + default-language: Haskell2010 main-is: loops.hs hs-source-dirs: examples if flag(examples) @@ -209,6 +210,7 @@ buildable: False executable nested-loops + default-language: Haskell2010 main-is: nested-loops.hs hs-source-dirs: examples if flag(examples) @@ -221,6 +223,7 @@ 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 --- a/test/Main.hs +++ b/test/Main.hs @@ -422,8 +422,12 @@ 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 @@ -646,6 +650,16 @@ lst <- replicateM len x 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))