streaming-concurrency 0.1.0.0 → 0.2.0.0

raw patch · 5 files changed

+274/−51 lines, 5 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• ChangeLog.md +19/−0
• src/Streaming/Concurrent.hs +126/−23
• src/Streaming/Concurrent/Lifted.hs +107/−19
• streaming-concurrency.cabal +2/−2
• test/merging.hs +20/−7

@@ -1,5 +1,24 @@ # Revision history for streaming-concurrency +## 0.2.0.0  -- 2017-07-05++* Rename functions to match the `with...` naming scheme:++    - `merge{Streams,ByteStrings}` to `withMerged{Streams,ByteStrings}`+    - `read{Stream,ByteString}Basket` to `with{Stream,ByteString}Basket`++* Add the ability to use buffers to concurrently transform the stream;+  following functions added:++    - `withBufferedTransform`+    - `withStreamMap`+    - `withStreamMapM`+    - `withStreamTransform`++    (Note: it is not sound in the general case to transform a+    streaming `ByteString`; as such, functions for this are not+    implemented though it is possible to do yourself.)+ ## 0.1.0.0  -- 2017-07-04  * First version.

@@ -28,16 +28,22 @@   , newest     -- * Using a buffer   , withBuffer+  , withBufferedTransform   , InBasket(..)   , OutBasket(..)     -- * Stream support   , writeStreamBasket-  , readStreamBasket-  , mergeStreams+  , withStreamBasket+  , withMergedStreams+    -- ** Mapping+  , withStreamMap+  , withStreamMapM+  , withStreamTransform     -- * ByteString support   , writeByteStringBasket-  , readByteStringBasket-  , mergeByteStrings+  , withByteStringBasket+  , withMergedByteStrings+    -- $bytestringtransform   ) where  import           Data.ByteString.Streaming (ByteString, reread, unconsChunk)@@ -46,7 +52,8 @@  import           Control.Applicative             ((<|>)) import           Control.Concurrent.Async.Lifted (concurrently,-                                                  forConcurrently_)+                                                  forConcurrently_,+                                                  replicateConcurrently_) import qualified Control.Concurrent.STM          as STM import           Control.Monad                   (when) import           Control.Monad.Base              (MonadBase, liftBase)@@ -63,20 +70,24 @@ -- --   Note that the monad of the resultant Stream can be different from --   the final result.-mergeStreams :: (MonadMask m, MonadBaseControl IO m, MonadBase IO n, Foldable t)-                => Buffer a -> t (Stream (Of a) m v)-                -> (Stream (Of a) n () -> m r) -> m r-mergeStreams buff strs f = withBuffer buff-                                      (forConcurrently_ strs . flip writeStreamBasket)-                                      (`readStreamBasket` f)+--+--   @since 0.2.0.0+withMergedStreams :: (MonadMask m, MonadBaseControl IO m, MonadBase IO n, Foldable t)+                     => Buffer a -> t (Stream (Of a) m v)+                     -> (Stream (Of a) n () -> m r) -> m r+withMergedStreams buff strs f = withBuffer buff+                                           (forConcurrently_ strs . flip writeStreamBasket)+                                           (`withStreamBasket` f) --- | A streaming 'ByteString' variant of 'mergeStreams'.-mergeByteStrings :: (MonadMask m, MonadBaseControl IO m, MonadBase IO n, Foldable t)-                    => Buffer B.ByteString -> t (ByteString m v)-                    -> (ByteString n () -> m r) -> m r-mergeByteStrings buff bss f = withBuffer buff-                                         (forConcurrently_ bss . flip writeByteStringBasket)-                                         (`readByteStringBasket` f)+-- | A streaming 'ByteString' variant of 'withMergedStreams'.+--+--   @since 0.2.0.0+withMergedByteStrings :: (MonadMask m, MonadBaseControl IO m, MonadBase IO n, Foldable t)+                         => Buffer B.ByteString -> t (ByteString m v)+                         -> (ByteString n () -> m r) -> m r+withMergedByteStrings buff bss f = withBuffer buff+                                              (forConcurrently_ bss . flip writeByteStringBasket)+                                              (`withByteStringBasket` f)  -- | Write a single stream to a buffer. --@@ -100,19 +111,111 @@                  when continue (go bs')  -- | Read the output of a buffer into a stream.-readStreamBasket :: (MonadBase IO m) => OutBasket a+--+--   @since 0.2.0.0+withStreamBasket :: (MonadBase IO m) => OutBasket a                     -> (Stream (Of a) m () -> r)                     -> r-readStreamBasket (OutBasket receive) f = f (S.untilRight getNext)+withStreamBasket (OutBasket receive) f = f (S.untilRight getNext)   where     getNext = maybe (Right ()) Left <$> liftBase (STM.atomically receive) --- | A streaming 'ByteString' variant of 'readStreamBasket'.-readByteStringBasket :: (MonadBase IO m) => OutBasket B.ByteString+-- | A streaming 'ByteString' variant of 'withStreamBasket'.+--+--   @since 0.2.0.0+withByteStringBasket :: (MonadBase IO m) => OutBasket B.ByteString                         -> (ByteString m () -> r)                         -> r-readByteStringBasket (OutBasket receive) f =+withByteStringBasket (OutBasket receive) f =   f (reread (liftBase . STM.atomically) receive)++--------------------------------------------------------------------------------++-- | Use buffers to concurrently transform the provided data.+--+--   In essence, this is a @demultiplexer -> multiplexer@+--   transformation: the incoming data is split into @n@ individual+--   segments, the results of which are then merged back together+--   again.+--+--   Note: ordering of elements in the output is undeterministic.+--+--   @since 0.2.0.0+withBufferedTransform :: (MonadMask m, MonadBaseControl IO m)+                         => Int+                            -- ^ How many concurrent computations to run.+                         -> (OutBasket a -> InBasket b -> m ab)+                            -- ^ What to do with each individual concurrent+                            --   computation; result is ignored.+                         -> (InBasket a -> m i)+                            -- ^ Provide initial data; result is ignored.+                         -> (OutBasket b -> m r) -> m r+withBufferedTransform n transform feed consume =+  withBuffer buff feed $ \obA ->+    withBuffer buff (replicateConcurrently_ n . transform obA)+      consume+  where+    buff :: Buffer v+    buff = bounded n++-- | Concurrently map a function over all elements of a 'Stream'.+--+--   Note: ordering of elements in the output is undeterministic.+--+--   @since 0.2.0.0+withStreamMap :: (MonadMask m, MonadBaseControl IO m, MonadBase IO n)+                 => Int -- ^ How many concurrent computations to run.+                 -> (a -> b)+                 -> Stream (Of a) m i+                 -> (Stream (Of b) n () -> m r) -> m r+withStreamMap n = withStreamTransform n . S.map++-- | Concurrently map a monadic function over all elements of a+--   'Stream'.+--+--   Note: ordering of elements in the output is undeterministic.+--+--   @since 0.2.0.0+withStreamMapM :: (MonadMask m, MonadBaseControl IO m, MonadBase IO n)+                  => Int -- ^ How many concurrent computations to run.+                  -> (a -> m b)+                  -> Stream (Of a) m i+                  -> (Stream (Of b) n () -> m r) -> m r+withStreamMapM n = withStreamTransform n . S.mapM++-- | Concurrently split the provided stream into @n@ streams and+--   transform them all using the provided function.+--+--   Note: ordering of elements in the output is undeterministic.+--+--   @since 0.2.0.0+withStreamTransform :: (MonadMask m, MonadBaseControl IO m, MonadBase IO n)+                       => Int -- ^ How many concurrent computations to run.+                       -> (Stream (Of a) m () -> Stream (Of b) m t)+                       -> Stream (Of a) m i+                       -> (Stream (Of b) n () -> m r) -> m r+withStreamTransform n f inp cont =+  withBufferedTransform n transform feed consume+  where+    feed = writeStreamBasket inp++    transform obA ibB = withStreamBasket obA+                          (flip writeStreamBasket ibB . f)++    consume = flip withStreamBasket cont++{- $bytestringtransform++No 'ByteString' equivalents of 'withStreamMap', etc. are provided as+it is very rare for individual chunks of a 'ByteString' - the sizes of+which can vary - to be independent of their position within the+overall stream.++If you can make such guarantees (e.g. you know that each chunk is a+distinct line and the ordering of these doesn't matter) then you can+use 'withBufferedTransform' to write your own.++-}  -------------------------------------------------------------------------------- -- This entire section is almost completely taken from

@@ -1,4 +1,4 @@-{-# LANGUAGE FlexibleContexts, MultiParamTypeClasses #-}+{-# LANGUAGE FlexibleContexts, MultiParamTypeClasses, TypeFamilies #-}  {- |    Module      : Streaming.Concurrent.Lifted@@ -21,16 +21,22 @@   , newest     -- * Using a buffer   , withBuffer+  , withBufferedTransform   , InBasket(..)   , OutBasket(..)     -- * Stream support   , writeStreamBasket-  , readStreamBasket-  , mergeStreams+  , withStreamBasket+  , withMergedStreams+    -- ** Mapping+  , withStreamMap+  , withStreamMapM+  , withStreamTransform     -- * ByteString support   , writeByteStringBasket-  , readByteStringBasket-  , mergeByteStrings+  , withByteStringBasket+  , withMergedByteStrings+    -- $bytestringtransform   ) where  import           Data.ByteString.Streaming (ByteString)@@ -50,16 +56,20 @@ -- | Concurrently merge multiple streams together. -- --   The resulting order is unspecified.-mergeStreams :: (Withable w, MonadBaseControl IO (WithMonad w), MonadBase IO m, Foldable t)-                => Buffer a -> t (Stream (Of a) (WithMonad w) v)-                -> w (Stream (Of a) m ())-mergeStreams buff strs = liftWith (SC.mergeStreams buff strs)+--+--   @since 0.2.0.0+withMergedStreams :: (Withable w, MonadBaseControl IO (WithMonad w), MonadBase IO m, Foldable t)+                     => Buffer a -> t (Stream (Of a) (WithMonad w) v)+                     -> w (Stream (Of a) m ())+withMergedStreams buff strs = liftWith (SC.withMergedStreams buff strs) --- | A streaming 'ByteString' variant of 'mergeStreams'.-mergeByteStrings :: (Withable w, MonadBaseControl IO (WithMonad w), MonadBase IO n, Foldable t)-                    => Buffer B.ByteString -> t (ByteString (WithMonad w) v)-                    -> w (ByteString n ())-mergeByteStrings buff bss = liftWith (SC.mergeByteStrings buff bss)+-- | A streaming 'ByteString' variant of 'withMergedStreams'.+--+--   @since 0.2.0.0+withMergedByteStrings :: (Withable w, MonadBaseControl IO (WithMonad w), MonadBase IO n, Foldable t)+                         => Buffer B.ByteString -> t (ByteString (WithMonad w) v)+                         -> w (ByteString n ())+withMergedByteStrings buff bss = liftWith (SC.withMergedByteStrings buff bss)  -- | Write a single stream to a buffer. --@@ -77,13 +87,91 @@ -- | Read the output of a buffer into a stream. -- --   Note that there is no requirement that @m ~ WithMonad w@.-readStreamBasket :: (Withable w, MonadBase IO m) => OutBasket a -> w (Stream (Of a) m ())-readStreamBasket ob = liftWith (SC.readStreamBasket ob)+--+--   @since 0.2.0.0+withStreamBasket :: (Withable w, MonadBase IO m) => OutBasket a -> w (Stream (Of a) m ())+withStreamBasket ob = liftWith (SC.withStreamBasket ob) --- | A streaming 'ByteString' variant of 'readStreamBasket'.-readByteStringBasket :: (Withable w, MonadBase IO m)+-- | A streaming 'ByteString' variant of 'withStreamBasket'.+--+--   @since 0.2.0.0+withByteStringBasket :: (Withable w, MonadBase IO m)                         => OutBasket B.ByteString -> w (ByteString m ())-readByteStringBasket ob = liftWith (SC.readByteStringBasket ob)+withByteStringBasket ob = liftWith (SC.withByteStringBasket ob)++-- | Use buffers to concurrently transform the provided data.+--+--   In essence, this is a @demultiplexer -> multiplexer@+--   transformation: the incoming data is split into @n@ individual+--   segments, the results of which are then merged back together+--   again.+--+--   Note: ordering of elements in the output is undeterministic.+--+--   @since 0.2.0.0+withBufferedTransform :: (Withable w, MonadBaseControl IO (WithMonad w))+                         => Int+                            -- ^ How many concurrent computations to run.+                         -> (OutBasket a -> InBasket b -> WithMonad w ab)+                            -- ^ What to do with each individual concurrent+                            --   computation; result is ignored.+                         -> (InBasket a -> WithMonad w i)+                            -- ^ Provide initial data; result is ignored.+                         -> w (OutBasket b)+withBufferedTransform n transform feed =+  liftWith (SC.withBufferedTransform n transform feed)++-- | Concurrently map a function over all elements of a 'Stream'.+--+--   Note: ordering of elements in the output is undeterministic.+--+--   @since 0.2.0.0+withStreamMap :: (Withable w, MonadBaseControl IO (WithMonad w), MonadBase IO n)+                 => Int -- ^ How many concurrent computations to run.+                 -> (a -> b)+                 -> Stream (Of a) (WithMonad w) i+                 -> w (Stream (Of b) n ())+withStreamMap n f inp = liftWith (SC.withStreamMap n f inp)++-- | Concurrently map a monadic function over all elements of a+--   'Stream'.+--+--   Note: ordering of elements in the output is undeterministic.+--+--   @since 0.2.0.0+withStreamMapM :: (Withable w, MonadBaseControl IO (WithMonad w), MonadBase IO n)+                  => Int -- ^ How many concurrent computations to run.+                  -> (a -> WithMonad w b)+                  -> Stream (Of a) (WithMonad w) i+                  -> w (Stream (Of b) n ())+withStreamMapM n f inp = liftWith (SC.withStreamMapM n f inp)++-- | Concurrently split the provided stream into @n@ streams and+--   transform them all using the provided function.+--+--   Note: ordering of elements in the output is undeterministic.+--+--   @since 0.2.0.0+withStreamTransform :: ( Withable w, m ~ WithMonad w, MonadBaseControl IO m+                       , MonadBase IO n)+                       => Int -- ^ How many concurrent computations to run.+                       -> (Stream (Of a) m () -> Stream (Of b) m t)+                       -> Stream (Of a) m i+                       -> w (Stream (Of b) n ())+withStreamTransform n f inp = liftWith (SC.withStreamTransform n f inp)++{- $bytestringtransform++No 'ByteString' equivalents of 'withStreamMap', etc. are provided as+it is very rare for individual chunks of a 'ByteString' - the sizes of+which can vary - to be independent of their position within the+overall stream.++If you can make such guarantees (e.g. you know that each chunk is a+distinct line and the ordering of these doesn't matter) then you can+use 'withBufferedTransform' to write your own.++-}  -- | Use a buffer to asynchronously communicate. --

@@ -1,5 +1,5 @@ name:                streaming-concurrency-version:             0.1.0.0+version:             0.2.0.0 synopsis:            Concurrency support for the streaming ecosystem description:   The primary purpose for this library is to be able to merge multiple@@ -50,4 +50,4 @@                      , streaming-bytestring   hs-source-dirs:      test   default-language:    Haskell2010-  ghc-options:         -Wall+  ghc-options:         -Wall -threaded

@@ -22,7 +22,7 @@ import Data.Monoid               (mconcat) import Test.Hspec                (describe, hspec) import Test.Hspec.QuickCheck     (prop)-import Test.QuickCheck           (Property, ioProperty)+import Test.QuickCheck           (Positive(..), Property, ioProperty) import Test.QuickCheck.Instances ()  --------------------------------------------------------------------------------@@ -32,20 +32,33 @@   describe "Merging retains all elements" $ do     prop "Stream" (mergeCheck :: [[Int]] -> Property)     prop "ByteString" mergeBSCheck+  describe "Stream transformation" $ do+    prop "map id" (streamMapCheckId :: Positive Int -> [Int] -> Property)+    prop "map show" (streamMapCheck show :: Positive Int -> [Int] -> Property)  mergeCheck :: (Ord a) => [[a]] -> Property-mergeCheck ass = ioProperty (mergeStreams unbounded-                                          (map each ass)-                                          (eqOn sort as . toList_))+mergeCheck ass = ioProperty (withMergedStreams unbounded+                                               (map each ass)+                                               (eqOn sort as . toList_))   where     as = concat ass  mergeBSCheck :: [B.ByteString] -> Property-mergeBSCheck bss = ioProperty (mergeByteStrings unbounded-                                                (map fromStrict bss)-                                                (eqOn B.sort bs . toStrict_))+mergeBSCheck bss = ioProperty (withMergedByteStrings unbounded+                                                     (map fromStrict bss)+                                                     (eqOn B.sort bs . toStrict_))   where     bs = mconcat bss++streamMapCheckId :: (Ord a) => Positive Int -> [a] -> Property+streamMapCheckId (Positive n) as =+  ioProperty (withStreamMap n id (each as) (eqOn sort as . toList_))++streamMapCheck :: (Ord b) => (a -> b) -> Positive Int -> [a] -> Property+streamMapCheck f (Positive n) as =+  ioProperty (withStreamMap n f (each as) (eqOn sort bs . toList_))+  where+    bs = map f as  eqOn :: (Eq b, Functor f) => (a -> b) -> a -> f a -> f Bool eqOn f a = fmap (((==)`on`f) a)