diff --git a/BroadcastChan/Conduit.hs b/BroadcastChan/Conduit.hs deleted file mode 100644 --- a/BroadcastChan/Conduit.hs +++ /dev/null @@ -1,40 +0,0 @@ -{-# LANGUAGE Safe #-} -------------------------------------------------------------------------------- --- | --- Module : BroadcastChan.Conduit --- 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 conduits and sinks that --- process in parallel. -------------------------------------------------------------------------------- -module BroadcastChan.Conduit - ( 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.Conduit.Internal diff --git a/BroadcastChan/Conduit/Internal.hs b/BroadcastChan/Conduit/Internal.hs deleted file mode 100644 --- a/BroadcastChan/Conduit/Internal.hs +++ /dev/null @@ -1,100 +0,0 @@ -{-# LANGUAGE NamedFieldPuns #-} -{-# LANGUAGE Trustworthy #-} -{-# LANGUAGE ScopedTypeVariables #-} -module BroadcastChan.Conduit.Internal (parMapM, parMapM_) where - -import Control.Monad ((>=>)) -import Control.Monad.Trans.Resource (MonadResource) -import qualified Control.Monad.Trans.Resource as Resource -import qualified Control.Monad.Trans.Resource.Internal as ResourceI -import Control.Monad.Trans.Class (lift) -import Control.Monad.IO.Unlift (MonadUnliftIO, UnliftIO(..), askUnliftIO) -import Data.Acquire (ReleaseType(..), allocateAcquire, mkAcquireType) -import Data.Conduit (ConduitM, (.|), awaitForever, yield) -import qualified Data.Conduit.List as C -import Data.Foldable (traverse_) -import Data.Void (Void) - -import BroadcastChan.Extra (BracketOnError(..), Handler, ThreadBracket(..)) -import qualified BroadcastChan.Extra as Extra - -bracketOnError :: MonadResource m => IO a -> (a -> IO ()) -> m r -> m r -bracketOnError alloc clean work = - allocateAcquire (mkAcquireType alloc cleanup) >>= const work - where - cleanup x ReleaseException = clean x - cleanup _ _ = return () - --- | Create a conduit that processes inputs in parallel. --- --- This function does __NOT__ guarantee that input elements are processed or --- output in a deterministic order! -parMapM - :: (MonadResource m, MonadUnliftIO m) - => Handler m a - -- ^ Exception handler - -> Int - -- ^ Number of parallel threads to use - -> (a -> m b) - -- ^ Function to run in parallel - -> ConduitM a b m () -parMapM hnd threads workFun = do - UnliftIO runInIO <- lift askUnliftIO - - resourceState <- Resource.liftResourceT Resource.getInternalState - - let threadBracket = ThreadBracket - { setupFork = ResourceI.stateAlloc resourceState - , cleanupFork = ResourceI.stateCleanup ReleaseNormal resourceState - , cleanupForkError = - ResourceI.stateCleanup ReleaseException resourceState - } - - Bracket{allocate,cleanup,action} <- Extra.runParallelWith - threadBracket - (Left yield) - (Extra.mapHandler runInIO hnd) - threads - (runInIO . workFun) - body - - bracketOnError allocate cleanup action - where - body :: Monad m => (a -> m ()) -> (a -> m (Maybe b)) -> ConduitM a b m () - body buffer process = do - C.isolate threads .| C.mapM_ buffer - awaitForever $ lift . process >=> traverse_ yield - --- | Create a conduit sink that consumes inputs in parallel. --- --- This function does __NOT__ guarantee that input elements are processed or --- output in a deterministic order! -parMapM_ - :: (MonadResource m, MonadUnliftIO m) - => Handler m a - -- ^ Exception handler - -> Int - -- ^ Number of parallel threads to use - -> (a -> m ()) - -- ^ Function to run in parallel - -> ConduitM a Void m () -parMapM_ hnd threads workFun = do - UnliftIO runInIO <- lift askUnliftIO - - resourceState <- Resource.liftResourceT Resource.getInternalState - - let threadBracket = ThreadBracket - { setupFork = ResourceI.stateAlloc resourceState - , cleanupFork = ResourceI.stateCleanup ReleaseNormal resourceState - , cleanupForkError = - ResourceI.stateCleanup ReleaseException resourceState - } - - Bracket{allocate,cleanup,action} <- Extra.runParallelWith_ - threadBracket - (Extra.mapHandler runInIO hnd) - threads - (runInIO . workFun) - C.mapM_ - - bracketOnError allocate cleanup action diff --git a/BroadcastChan/Conduit/Throw.hs b/BroadcastChan/Conduit/Throw.hs deleted file mode 100644 --- a/BroadcastChan/Conduit/Throw.hs +++ /dev/null @@ -1,40 +0,0 @@ -{-# LANGUAGE Safe #-} -------------------------------------------------------------------------------- --- | --- Module : BroadcastChan.Conduit.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 conduits and sinks that --- process in parallel. -------------------------------------------------------------------------------- -module BroadcastChan.Conduit.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.Conduit.Internal diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,17 @@ +0.3.0 [2025.03.16] +------------------ +* Add missing reexports of `readBChan`, `writeBChan`, and `BChanError` to + `BroadcastChan.Conduit` and `BroadcastChan.Conduit.Throw`. +* Add reexports for the new `tryReadBChan` in `BroadcastChan.Conduit` and + `BroadcastChan.Conduit.Throw`. +* Turned into a trivial re-export of broadcast-chan:conduit. +* `broadcast-chan-conduit` is deprecated in favour of `broadcast-chan:conduit` + (sub-library of the `broadcast-chan` package). + +0.2.1.2 [2022.08.24] +-------------------- +* Revision updating bounds for GHC 9.4. + 0.2.1.2 [2021.12.01] -------------------- * Updated bounds for GHC 9.0 and 9.2. diff --git a/broadcast-chan-conduit.cabal b/broadcast-chan-conduit.cabal --- a/broadcast-chan-conduit.cabal +++ b/broadcast-chan-conduit.cabal @@ -1,13 +1,13 @@ -Cabal-Version: 2.2 +Cabal-Version: 3.4 Name: broadcast-chan-conduit -Version: 0.2.1.2 +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,48 +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-Source-Files: CHANGELOG.md +Extra-Doc-Files: CHANGELOG.md Synopsis: Conduit-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.Conduit + Reexported-Modules: BroadcastChan.Conduit, BroadcastChan.Conduit.Throw - Other-Modules: BroadcastChan.Conduit.Internal - Other-Extensions: NamedFieldPuns - Safe - ScopedTypeVariables - Trustworthy - - Build-Depends: base >= 4.8 && < 4.17 - , broadcast-chan == 0.2.1.2 - , conduit >= 1.2 && < 1.4 - , resourcet >= 1.1 && < 1.3 - , transformers >= 0.2 && < 0.7 - , unliftio-core >= 0.1 && < 0.3 - -Test-Suite conduit - Default-Language: Haskell2010 - Type: exitcode-stdio-1.0 - Main-Is: ConduitTest.hs - GHC-Options: -Wall -Wno-unused-do-bind -threaded -with-rtsopts=-qg - Hs-Source-Dirs: tests - - Build-Depends: base - , broadcast-chan-conduit - , broadcast-chan-tests - , containers >= 0.4 && < 0.7 - , conduit >= 1.2 && < 1.4 + Build-Depends: broadcast-chan:conduit == 0.3.0 Source-Repository head Type: git diff --git a/tests/ConduitTest.hs b/tests/ConduitTest.hs deleted file mode 100644 --- a/tests/ConduitTest.hs +++ /dev/null @@ -1,37 +0,0 @@ -import Control.Monad (void) -import Data.Set (Set) -import qualified Data.Set as S -import Data.Conduit -import qualified Data.Conduit.List as C - -import BroadcastChan.Conduit -import BroadcastChan.Test - -sequentialSink :: [a] -> (a -> IO b) -> IO () -sequentialSink inputs f = - runConduitRes $ C.sourceList inputs .| C.mapM_ (liftIO . void . f) - -parallelSink :: Handler IO a -> [a] -> (a -> IO b) -> Int -> IO () -parallelSink hnd inputs f n = runConduitRes $ - C.sourceList inputs .| parMapM_ handler n (liftIO . void . f) - where - handler = mapHandler liftIO hnd - -sequentialFold :: Ord b => [a] -> (a -> IO b) -> IO (Set b) -sequentialFold inputs f = runConduitRes $ - C.sourceList inputs .| C.mapM (liftIO . f) .| C.foldMap S.singleton - -parallelFold - :: Ord b => Handler IO a -> [a] -> (a -> IO b) -> Int -> IO (Set b) -parallelFold hnd inputs f n = runConduitRes $ - C.sourceList inputs - .| parMapM handler n (liftIO . f) - .| C.foldMap S.singleton - where - handler = mapHandler liftIO hnd - -main :: IO () -main = runTests "conduit" $ - [ genStreamTests "sink" sequentialSink parallelSink - , genStreamTests "fold" sequentialFold parallelFold - ]