broadcast-chan-pipes 0.2.1.1 → 0.3.0
raw patch · 6 files changed
+47/−214 lines, 6 filesdep −basedep −broadcast-chan-pipesdep −broadcast-chan-testsdep ~broadcast-chan
Dependencies removed: base, broadcast-chan-pipes, broadcast-chan-tests, containers, foldl, pipes, pipes-safe
Dependency ranges changed: broadcast-chan
Files
- BroadcastChan/Pipes.hs +0/−40
- BroadcastChan/Pipes/Internal.hs +0/−67
- BroadcastChan/Pipes/Throw.hs +0/−40
- CHANGELOG.md +35/−0
- broadcast-chan-pipes.cabal +12/−32
- tests/PipeTest.hs +0/−35
− BroadcastChan/Pipes.hs
@@ -1,40 +0,0 @@-{-# LANGUAGE Safe #-}----------------------------------------------------------------------------------- |--- Module : BroadcastChan.Pipes--- Copyright : (C) 2014-2021 Merijn Verstraaten--- License : BSD-style (see the file LICENSE)--- Maintainer : Merijn Verstraaten <merijn@inconsistent.nl>--- Stability : experimental--- Portability : haha------ This module is identical to "BroadcastChan", but replaces the parallel--- processing operations with functions for creating producers and effects that--- process in parallel.---------------------------------------------------------------------------------module BroadcastChan.Pipes- ( Action(..)- , Handler(..)- , parMapM- , parMapM_- -- * Re-exports from "BroadcastChan"- -- ** Datatypes- , BroadcastChan- , Direction(..)- , In- , Out- -- ** Construction- , newBroadcastChan- , newBChanListener- -- ** Basic Operations- , closeBChan- , isClosedBChan- , getBChanContents- -- ** Foldl combinators- -- | Combinators for use with Tekmo's @foldl@ package.- , foldBChan- , foldBChanM- ) where--import BroadcastChan hiding (parMapM_)-import BroadcastChan.Pipes.Internal
− BroadcastChan/Pipes/Internal.hs
@@ -1,67 +0,0 @@-{-# LANGUAGE NamedFieldPuns #-}-{-# LANGUAGE Safe #-}-{-# LANGUAGE ScopedTypeVariables #-}-module BroadcastChan.Pipes.Internal (parMapM, parMapM_) where--import Control.Monad ((>=>), replicateM)-import Data.Foldable (traverse_)-import Pipes-import qualified Pipes.Prelude as P-import Pipes.Safe (MonadSafe)-import qualified Pipes.Safe as Safe--import BroadcastChan.Extra- (BracketOnError(..), Handler, runParallel, runParallel_)--bracketOnError :: MonadSafe m => IO a -> (a -> IO b) -> m c -> m c-bracketOnError alloc clean =- Safe.bracketOnError (liftIO alloc) (liftIO . clean) . const---- | Create a producer that processes its inputs in parallel.------ This function does __NOT__ guarantee that input elements are processed or--- output in a deterministic order!-parMapM- :: forall a b m- . MonadSafe m- => Handler IO a- -- ^ Exception handler- -> Int- -- ^ Number of parallel threads to use- -> (a -> IO b)- -- ^ Function to run in parallel- -> Producer a m ()- -- ^ Input producer- -> Producer b m ()-parMapM hndl i f prod = do- Bracket{allocate,cleanup,action} <- runParallel (Left yield) hndl i f body- bracketOnError allocate cleanup action- where- body :: (a -> m ()) -> (a -> m (Maybe b)) -> Producer b m ()- body buffer process = prod >-> work- where- work :: Pipe a b m ()- work = do- replicateM i (await >>= lift . buffer)- for cat $ lift . process >=> traverse_ yield---- | Create an Effect that processes its inputs in parallel.------ This function does __NOT__ guarantee that input elements are processed or--- output in a deterministic order!-parMapM_- :: MonadSafe m- => Handler IO a- -- ^ Exception handler- -> Int- -- ^ Number of parallel threads to use- -> (a -> IO ())- -- ^ Function to run in parallel- -> Producer a m r- -- ^ Input producer- -> Effect m r-parMapM_ hndl i f prod = do- Bracket{allocate,cleanup,action} <- runParallel_ hndl i f workProd- bracketOnError allocate cleanup action- where- workProd buffer = prod >-> P.mapM_ buffer
− BroadcastChan/Pipes/Throw.hs
@@ -1,40 +0,0 @@-{-# LANGUAGE Safe #-}----------------------------------------------------------------------------------- |--- Module : BroadcastChan.Pipes.Throw--- Copyright : (C) 2014-2021 Merijn Verstraaten--- License : BSD-style (see the file LICENSE)--- Maintainer : Merijn Verstraaten <merijn@inconsistent.nl>--- Stability : experimental--- Portability : haha------ This module is identical to "BroadcastChan.Throw", but replaces the parallel--- processing operations with functions for creating producers and effects that--- process in parallel.---------------------------------------------------------------------------------module BroadcastChan.Pipes.Throw- ( Action(..)- , Handler(..)- , parMapM- , parMapM_- -- * Re-exports from "BroadcastChan.Throw"- -- ** Datatypes- , BroadcastChan- , Direction(..)- , In- , Out- -- ** Construction- , newBroadcastChan- , newBChanListener- -- ** Basic Operations- , closeBChan- , isClosedBChan- , getBChanContents- -- ** Foldl combinators- -- | Combinators for use with Tekmo's @foldl@ package.- , foldBChan- , foldBChanM- ) where--import BroadcastChan.Throw hiding (parMapM_)-import BroadcastChan.Pipes.Internal
+ CHANGELOG.md view
@@ -0,0 +1,35 @@+0.3.0 [2025.03.16]+------------------+* Add missing reexports of `readBChan`, `writeBChan`, and `BChanError` to+ `BroadcastChan.Pipes` and `BroadcastChan.Pipes.Throw`.+* Add reexports for the new `tryReadBChan` in `BroadcastChan.Pipes` and+ `BroadcastChan.Pipes.Throw`.+* Turned into a trivial re-export of broadcast-chan:pipes.+* `broadcast-chan-pipes` is deprecated in favour of `broadcast-chan:pipes`+ (sub-library of the `broadcast-chan` package).++0.2.1.2 [2022.08.24]+--------------------+* Revision updating bounds for GHC 9.4.++0.2.1.1 [2021.12.01]+--------------------+* Updated bounds for GHC 9.0 and 9.2.+* Tighten bound on broadcast-chan to only use version with fixed race+ condition.++0.2.1 [2019.11.17]+------------------+* Bump for new broadcast-chan release.++0.2.0.2 [2019.03.30]+--------------------+* Update bounds for GHC 8.6++0.2.0.1 [2018.09.24]+--------------------+* Ditch GHC 7.6.3 support.++0.2.0 [2018.09.20]+------------------+* Initial release.
broadcast-chan-pipes.cabal view
@@ -1,13 +1,13 @@-Cabal-Version: 2.2+Cabal-Version: 3.4 Name: broadcast-chan-pipes-Version: 0.2.1.1+Version: 0.3.0 Homepage: https://github.com/merijn/broadcast-chan Bug-Reports: https://github.com/merijn/broadcast-chan/issues Author: Merijn Verstraaten Maintainer: Merijn Verstraaten <merijn@inconsistent.nl>-Copyright: Copyright © 2014-2021 Merijn Verstraaten+Copyright: Copyright © 2014-2025 Merijn Verstraaten License: BSD-3-Clause License-File: LICENSE@@ -15,44 +15,24 @@ Category: System Build-Type: Simple Tested-With: GHC == 8.0.2, GHC == 8.2.2, GHC == 8.4.4, GHC == 8.6.5,- GHC == 8.8.4, GHC == 8.10.7, GHC == 9.0.1, GHC == 9.2.1+ GHC == 8.8.4, GHC == 8.10.7, GHC == 9.0.2, GHC == 9.2.7,+ GHC == 9.4.8, GHC == 9.6.6, GHC == 9.8.4, GHC == 9.10.1,+ GHC == 9.12.1 +Extra-Doc-Files: CHANGELOG.md+ Synopsis: Pipes-based parallel streaming code for broadcast-chan Description:- __WARNING:__ While the code in this library should be fairly stable and- production, the API is something I'm still working on. API changes will- follow the PVP, but __expect__ breaking API changes in future versions!+ __WARNING: As of 0.3.0 this package is deprecated, use the sublibrary of+ broadcast-chan directly!__ Library Default-Language: Haskell2010- GHC-Options: -Wall -O2 -Wno-unused-do-bind- Exposed-Modules: BroadcastChan.Pipes+ Reexported-Modules: BroadcastChan.Pipes, BroadcastChan.Pipes.Throw- Other-Modules: BroadcastChan.Pipes.Internal - Other-Extensions: NamedFieldPuns- Safe- ScopedTypeVariables-- Build-Depends: base >= 4.8 && < 4.17- , broadcast-chan == 0.2.1.2- , pipes >= 4.1.6 && < 4.4- , pipes-safe >= 2.2 && < 2.4--Test-Suite pipes- Default-Language: Haskell2010- Type: exitcode-stdio-1.0- Main-Is: PipeTest.hs- GHC-Options: -Wall -Wno-unused-do-bind -threaded -with-rtsopts=-qg- Hs-Source-Dirs: tests- Build-Depends: base- , broadcast-chan-pipes- , broadcast-chan-tests- , containers >= 0.4 && < 0.7- , foldl >= 1.0.4 && < 1.5- , pipes >= 4.1.6 && < 4.4- , pipes-safe >= 2.2 && < 2.4+ Build-Depends: broadcast-chan:pipes == 0.3.0 Source-Repository head Type: git
− tests/PipeTest.hs
@@ -1,35 +0,0 @@-import Control.Foldl (purely, set)-import Data.Set (Set)-import Pipes-import qualified Pipes.Prelude as P-import Pipes.Safe (runSafeT)--import BroadcastChan.Pipes-import BroadcastChan.Test--sequentialSink :: [a] -> (a -> IO b) -> IO ()-sequentialSink inputs f = runSafeT . runEffect $- each inputs >-> P.mapM_ (liftIO . void. f)--parallelSink :: Handler IO a -> [a] -> (a -> IO b) -> Int -> IO ()-parallelSink hnd inputs f n =- runSafeT . runEffect $ parMapM_ handler n (void . f) $ each inputs- where- handler = mapHandler liftIO hnd--sequentialFold :: Ord b => [a] -> (a -> IO b) -> IO (Set b)-sequentialFold inputs f = runSafeT $ purely P.fold set $- each inputs >-> P.mapM (liftIO . f)--parallelFold- :: Ord b => Handler IO a -> [a] -> (a -> IO b) -> Int -> IO (Set b)-parallelFold hnd inputs f n =- runSafeT . purely P.fold set . parMapM handler n f $ each inputs- where- handler = mapHandler liftIO hnd--main :: IO ()-main = runTests "pipes" $- [ genStreamTests "sink" sequentialSink parallelSink- , genStreamTests "fold" sequentialFold parallelFold- ]