diff --git a/BroadcastChan/Pipes.hs b/BroadcastChan/Pipes.hs deleted file mode 100644 --- a/BroadcastChan/Pipes.hs +++ /dev/null @@ -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 --- 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 diff --git a/BroadcastChan/Pipes/Internal.hs b/BroadcastChan/Pipes/Internal.hs deleted file mode 100644 --- a/BroadcastChan/Pipes/Internal.hs +++ /dev/null @@ -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 diff --git a/BroadcastChan/Pipes/Throw.hs b/BroadcastChan/Pipes/Throw.hs deleted file mode 100644 --- a/BroadcastChan/Pipes/Throw.hs +++ /dev/null @@ -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 --- 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 diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -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. diff --git a/broadcast-chan-pipes.cabal b/broadcast-chan-pipes.cabal --- a/broadcast-chan-pipes.cabal +++ b/broadcast-chan-pipes.cabal @@ -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 -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 diff --git a/tests/PipeTest.hs b/tests/PipeTest.hs deleted file mode 100644 --- a/tests/PipeTest.hs +++ /dev/null @@ -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 - ]