streaming-concurrency 0.2.0.0 → 0.3.0.0

raw patch · 6 files changed

+84/−127 lines, 6 filesdep −bytestringdep −streaming-bytestringdep ~lifted-asyncPVP ok

version bump matches the API change (PVP)

Dependencies removed: bytestring, streaming-bytestring

Dependency ranges changed: lifted-async

API changes (from Hackage documentation)

Files

• ChangeLog.md +33/−0
• README.md +26/−4
• src/Streaming/Concurrent.hs +3/−51
• src/Streaming/Concurrent/Lifted.hs +5/−46
• streaming-concurrency.cabal +14/−11
• test/merging.hs +3/−15

@@ -1,5 +1,38 @@ # Revision history for streaming-concurrency +## 0.3.0.0  -- 2017-07-05++* Removed support for streaming `ByteString`s.++    The ByteString-related functions which were previously implemented+    are broken conceptually and should not be used; they will likely+    be removed in a future release.++    A ByteString consists of a stream of bytes; the meaning of each byte+    is dependent upon its position in the overall stream.++    In terms of implementation, these bytes are accumulated into+    _chunks_, each of which may very well be of a different size.++    As such, if such a chunk is fed into a buffer with multiple readers,+    then there is no guarantee which reader will actually receive it or if+    it makes sense about what it is in isolation (e.g. it could be split+    mid-word, or possibly even in the middle of a Unicode character).++    If, however, multiple ByteStrings are fed into a buffer with a single+    reader, the order it has when coming out is similarly undeterministic:+    it isn't possible to coherently ensure the sanity of the resulting+    ByteString.++    As such, unless you really want to consider it as a stream of+    raw 8-bit numbers, trying to do any concurrency with a ByteString+    will only lead to trouble.  If you do need such functionality, you+    can implement it yourself using buffers containing `Word8` values+    (in which case you can use `Data.ByteString.Streaming.unpack`).++* Fix lower bound of `lifted-async` (`replicateConcurrently_` was+  added in 0.9.3).+ ## 0.2.0.0  -- 2017-07-05  * Rename functions to match the `with...` naming scheme:

@@ -5,12 +5,28 @@  > Concurrency for the [streaming] ecosystem -The primary purpose for this library is to be able to merge multiple-`Stream`s together.  However, it is possible to build higher-abstractions on top of this to be able to also feed multiple streams.- [streaming]: http://hackage.haskell.org/package/streaming +There are two primary higher-level use-cases for this library:++1. Merge multiple `Stream`s together.++2. A conceptual `Stream`-based equivalent to [`parMap`] (albeit+   utilising concurrency rather than true parallelism).++    [`parMap`]: http://hackage.haskell.org/package/parallel/docs/Control-Parallel-Strategies.html#v:parMap++However, low-level functions are also exposed so you can construct+your own methods of concurrently using `Stream`s (and there are also+non-`Stream`-specific functions if you wish to use it with other data+types).++Conceptually, the approach taken is to consider a typical+correspondence system with an in-basket/tray for receiving messages+for others, and an out-basket/tray to be later dealt with.  Inputs are+thus provided into the `InBasket` and removed once available from the+`OutBasket`.+ Thanks and recognition ---------------------- @@ -21,3 +37,9 @@ collection.  [pipes-concurrency]: http://hackage.haskell.org/package/pipes-concurrency++Another main difference is that the naming of the `input` and `output`+types has been switched around: [pipes-concurrency] seems to consider+them from the point of view of the supplying/consuming `Pipe`s,+whereas here they are considered from the point of view of the+`Buffer` itself.

@@ -39,16 +39,10 @@   , withStreamMap   , withStreamMapM   , withStreamTransform-    -- * ByteString support-  , writeByteStringBasket-  , withByteStringBasket-  , withMergedByteStrings-    -- $bytestringtransform   ) where -import           Data.ByteString.Streaming (ByteString, reread, unconsChunk)-import           Streaming                 (Of, Stream)-import qualified Streaming.Prelude         as S+import           Streaming         (Of, Stream)+import qualified Streaming.Prelude as S  import           Control.Applicative             ((<|>)) import           Control.Concurrent.Async.Lifted (concurrently,@@ -59,7 +53,6 @@ import           Control.Monad.Base              (MonadBase, liftBase) import           Control.Monad.Catch             (MonadMask, bracket, bracket_) import           Control.Monad.Trans.Control     (MonadBaseControl)-import qualified Data.ByteString                 as B import           Data.Foldable                   (forM_)  --------------------------------------------------------------------------------@@ -79,16 +72,6 @@                                            (forConcurrently_ strs . flip writeStreamBasket)                                            (`withStreamBasket` 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. -- --   Type written to make it easier if this is the only stream being@@ -101,15 +84,6 @@                   continue <- liftBase (STM.atomically (send a))                   when continue (go str') --- | A streaming 'ByteString' variant of 'writeStreamBasket'.-writeByteStringBasket :: (MonadBase IO m) => ByteString m r -> InBasket B.ByteString -> m ()-writeByteStringBasket bstring (InBasket send) = go bstring-  where-    go bs = do chNxt <- unconsChunk bs-               forM_ chNxt $ \(chnk, bs') -> do-                 continue <- liftBase (STM.atomically (send chnk))-                 when continue (go bs')- -- | Read the output of a buffer into a stream. -- --   @since 0.2.0.0@@ -120,15 +94,6 @@   where     getNext = maybe (Right ()) Left <$> liftBase (STM.atomically receive) --- | A streaming 'ByteString' variant of 'withStreamBasket'.------   @since 0.2.0.0-withByteStringBasket :: (MonadBase IO m) => OutBasket B.ByteString-                        -> (ByteString m () -> r)-                        -> r-withByteStringBasket (OutBasket receive) f =-  f (reread (liftBase . STM.atomically) receive)- --------------------------------------------------------------------------------  -- | Use buffers to concurrently transform the provided data.@@ -204,19 +169,6 @@      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 -- pipes-concurrent by Gabriel Gonzalez:@@ -240,7 +192,7 @@ --   argument. -- --   A buffer size @<= 0@ will result in a permanently empty buffer,---   which could result in a system that hang.+--   which could result in a system that hangs. bounded :: Int -> Buffer a bounded 1 = Single bounded n = Bounded n

@@ -32,25 +32,17 @@   , withStreamMap   , withStreamMapM   , withStreamTransform-    -- * ByteString support-  , writeByteStringBasket-  , withByteStringBasket-  , withMergedByteStrings-    -- $bytestringtransform   ) where -import           Data.ByteString.Streaming (ByteString)-import           Streaming                 (Of, Stream)-import           Streaming.Concurrent      (Buffer, InBasket(..), OutBasket(..),-                                            bounded, latest, newest, unbounded)-import qualified Streaming.Concurrent      as SC-import           Streaming.With.Lifted     (Withable(..))+import           Streaming             (Of, Stream)+import           Streaming.Concurrent  (Buffer, InBasket(..), OutBasket(..),+                                        bounded, latest, newest, unbounded)+import qualified Streaming.Concurrent  as SC+import           Streaming.With.Lifted (Withable(..))  import Control.Monad.Base          (MonadBase) import Control.Monad.Trans.Control (MonadBaseControl) -import qualified Data.ByteString as B- --------------------------------------------------------------------------------  -- | Concurrently merge multiple streams together.@@ -63,14 +55,6 @@                      -> w (Stream (Of a) m ()) withMergedStreams buff strs = liftWith (SC.withMergedStreams buff strs) --- | 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. -- --   Type written to make it easier if this is the only stream being@@ -79,11 +63,6 @@                      => Stream (Of a) (WithMonad w) r -> InBasket a -> w () writeStreamBasket stream ib = liftAction (SC.writeStreamBasket stream ib) --- | A streaming 'ByteString' variant of 'writeStreamBasket'.-writeByteStringBasket :: (Withable w, MonadBase IO (WithMonad w))-                         => ByteString (WithMonad w) r -> InBasket B.ByteString -> w ()-writeByteStringBasket bs ib = liftAction (SC.writeByteStringBasket bs ib)- -- | Read the output of a buffer into a stream. -- --   Note that there is no requirement that @m ~ WithMonad w@.@@ -92,13 +71,6 @@ withStreamBasket :: (Withable w, MonadBase IO m) => OutBasket a -> w (Stream (Of a) m ()) withStreamBasket ob = liftWith (SC.withStreamBasket ob) --- | A streaming 'ByteString' variant of 'withStreamBasket'.------   @since 0.2.0.0-withByteStringBasket :: (Withable w, MonadBase IO m)-                        => OutBasket B.ByteString -> w (ByteString m ())-withByteStringBasket ob = liftWith (SC.withByteStringBasket ob)- -- | Use buffers to concurrently transform the provided data. -- --   In essence, this is a @demultiplexer -> multiplexer@@@ -159,19 +131,6 @@                        -> 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,17 +1,24 @@ name:                streaming-concurrency-version:             0.2.0.0+version:             0.3.0.0 synopsis:            Concurrency support for the streaming ecosystem description:-  The primary purpose for this library is to be able to merge multiple-  @Stream@s together.  However, it is possible to build higher-  abstractions on top of this to be able to also feed multiple-  streams.+  There are two primary higher-level use-cases for this library:+  .+  1. Merge multiple @Stream@s together.+  .+  2. A conceptual @Stream@-based equivalent to @parMap@ (albeit+     utilising concurrency rather than true parallelism).+  .+  However, low-level functions are also exposed so you can construct+  your own methods of concurrently using @Stream@s (and there are also+  non-@Stream@-specific functions if you wish to use it with other data+  types). license:             MIT license-file:        LICENSE author:              Ivan Lazar Miljenovic maintainer:          Ivan.Miljenovic@gmail.com copyright:           Ivan Lazar Miljenovic-category:            Data+category:            Data, Streaming build-type:          Simple extra-source-files:  ChangeLog.md, README.md cabal-version:       >=1.10@@ -25,13 +32,11 @@   exposed-modules:     Streaming.Concurrent                      , Streaming.Concurrent.Lifted   build-depends:       base ==4.*-                     , bytestring                      , exceptions >= 0.6 && < 0.9-                     , lifted-async >= 0.9.1 && < 0.10+                     , lifted-async >= 0.9.3 && < 0.10                      , monad-control == 1.*                      , stm >= 2.4 && < 3                      , streaming >= 0.1.4.0 && < 0.2-                     , streaming-bytestring >= 0.1.4.5 && < 0.2                      , streaming-with == 0.1.*                      , transformers-base   hs-source-dirs:      src@@ -42,12 +47,10 @@   main-is:             merging.hs   build-depends:       streaming-concurrency                      , base-                     , bytestring                      , hspec == 2.4.*                      , QuickCheck == 2.*                      , quickcheck-instances                      , streaming-                     , streaming-bytestring   hs-source-dirs:      test   default-language:    Haskell2010   ghc-options:         -Wall -threaded

@@ -12,14 +12,10 @@  import Streaming.Concurrent -import Data.ByteString.Streaming (fromStrict, toStrict_)-import Streaming.Prelude         (each, toList_)--import qualified Data.ByteString as B+import Streaming.Prelude (each, toList_)  import Data.Function             (on) import Data.List                 (concat, sort)-import Data.Monoid               (mconcat) import Test.Hspec                (describe, hspec) import Test.Hspec.QuickCheck     (prop) import Test.QuickCheck           (Positive(..), Property, ioProperty)@@ -29,9 +25,8 @@  main :: IO () main = hspec $ do-  describe "Merging retains all elements" $ do-    prop "Stream" (mergeCheck :: [[Int]] -> Property)-    prop "ByteString" mergeBSCheck+  describe "Merging retains all elements" $+    prop "Int Stream" (mergeCheck :: [[Int]] -> Property)   describe "Stream transformation" $ do     prop "map id" (streamMapCheckId :: Positive Int -> [Int] -> Property)     prop "map show" (streamMapCheck show :: Positive Int -> [Int] -> Property)@@ -42,13 +37,6 @@                                                (eqOn sort as . toList_))   where     as = concat ass--mergeBSCheck :: [B.ByteString] -> Property-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 =