packages feed

broadcast-chan-conduit 0.2.1.2 → 0.3.0

raw patch · 6 files changed

+25/−252 lines, 6 filesdep −basedep −broadcast-chan-conduitdep −broadcast-chan-testsdep ~broadcast-chan

Dependencies removed: base, broadcast-chan-conduit, broadcast-chan-tests, conduit, containers, resourcet, transformers, unliftio-core

Dependency ranges changed: broadcast-chan

Files

− BroadcastChan/Conduit.hs 
@@ -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 <merijn@inconsistent.nl>--- 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
− BroadcastChan/Conduit/Internal.hs 
@@ -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
− BroadcastChan/Conduit/Throw.hs 
@@ -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 <merijn@inconsistent.nl>--- 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
CHANGELOG.md view
@@ -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.
broadcast-chan-conduit.cabal view
@@ -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 <merijn@inconsistent.nl>-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
− tests/ConduitTest.hs 
@@ -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-    ]