BroadcastChan.Conduit

This module is identical to BroadcastChan, but replaces the parallel processing operations with functions for creating conduits and sinks that process in parallel.

Documentation

data Action

Action to take when an exception occurs while processing an element.

Constructors

Drop

Drop the current element and continue processing.

Retry

Retry by appending the current element to the queue of remaining elements.

Terminate

Stop all processing and reraise the exception.

Instances

Eq Action

Defined in BroadcastChan.Extra

Methods

(==) :: Action -> Action -> Bool

(/=) :: Action -> Action -> Bool

Show Action

Defined in BroadcastChan.Extra

Methods

showsPrec :: Int -> Action -> ShowS

show :: Action -> String

showList :: [Action] -> ShowS

data Handler (m :: Type -> Type) a

Exception handler for parallel processing.

Constructors

Simple Action

Always take the specified Action.

Handle (a -> SomeException -> m Action)

Allow inspection of the element, exception, and execution of monadic actions before deciding the Action to take.

parMapM

Arguments

:: (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 ()

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_

Arguments

:: (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 ()

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!

Re-exports from BroadcastChan

Datatypes

data BroadcastChan (dir :: Direction) a

The abstract type representing the read or write end of a BroadcastChan.

Instances

Eq (BroadcastChan dir a)

Defined in BroadcastChan.Internal

Methods

(==) :: BroadcastChan dir a -> BroadcastChan dir a -> Bool

(/=) :: BroadcastChan dir a -> BroadcastChan dir a -> Bool

data Direction

Used with DataKinds as phantom type indicating whether a BroadcastChan value is a read or write end.

Constructors

In

Indicates a write BroadcastChan

Out

Indicates a read BroadcastChan

type In = In

Alias for the Direction type from the Direction kind, allows users to write the BroadcastChan Direction a type without enabling DataKinds.

type Out = Out

Alias for the Direction type from the Direction kind, allows users to write the BroadcastChan Direction a type without enabling DataKinds.

Construction

newBroadcastChan :: MonadIO m => m (BroadcastChan In a)

Creates a new BroadcastChan write end.

newBChanListener :: MonadIO m => BroadcastChan dir a -> m (BroadcastChan Out a)

Create a new read end for a BroadcastChan.

BroadcastChan Direction:

Will receive all messages written to the channel after this read end is created.

BroadcastChan Direction:

Will receive all currently unread messages and all future messages.

Basic Operations

closeBChan :: MonadIO m => BroadcastChan In a -> m Bool

Close a BroadcastChan, disallowing further writes. Returns True if the BroadcastChan was closed. Returns False if the BroadcastChan was already closed.

isClosedBChan :: MonadIO m => BroadcastChan dir a -> m Bool

Check whether a BroadcastChan is closed. True meaning that future read/write operations on the channel will always fail.

BroadcastChan Direction:

True indicates the channel is closed and writes will always fail.

Beware of TOC-TOU races: It is possible for a BroadcastChan to be closed by another thread. If multiple threads use the same channel a closeBChan from another thread can result in the channel being closed right after isClosedBChan returns.

BroadcastChan Direction:

True indicates the channel is both closed and empty, meaning reads will always fail.

getBChanContents :: BroadcastChan dir a -> IO [a]

Return a lazy list representing the messages written to the channel.

Uses unsafeInterleaveIO to defer the IO operations.

BroadcastChan Direction:

The list contains every message written to the channel after this IO action completes.

BroadcastChan Direction:

The list contains every currently unread message and all future messages. It's safe to keep using the original channel in any thread.

Unlike getChanContents from Control.Concurrent, the list resulting from this function is not affected by reads on the input channel. Every message that is unread or written after the IO action completes will end up in the result list.

Foldl combinators

Combinators for use with Tekmo's foldl package.

foldBChan :: (MonadIO m, MonadIO n) => (x -> a -> x) -> x -> (x -> b) -> BroadcastChan d a -> n (m b)

Strict fold of the BroadcastChan​'s messages. Can be used with Control.Foldl from Tekmo's foldl package:

Control.Foldl.purely foldBChan :: (MonadIO m, MonadIO n) => Fold a b -> BroadcastChan d a -> n (m b)

The result of this function is a nested monadic value to give more fine-grained control/separation between the start of listening for messages and the start of processing. The inner action folds the actual messages and completes when the channel is closed and exhausted. The outer action controls from when on messages are received. Specifically:

BroadcastChan Direction:

Will process all messages sent after the outer action completes.

BroadcastChan Direction:

Will process all messages that are unread when the outer action completes, as well as all future messages.

After the outer action completes the fold is unaffected by other (concurrent) reads performed on the original channel. So it's safe to reuse the channel.

foldBChanM :: (MonadIO m, MonadIO n) => (x -> a -> m x) -> m x -> (x -> m b) -> BroadcastChan d a -> n (m b)

Strict, monadic fold of the BroadcastChan​'s messages. Can be used with Control.Foldl from Tekmo's foldl package:

Control.Foldl.impurely foldBChanM :: (MonadIO m, MonadIO n) => FoldM m a b -> BroadcastChan d a -> n (m b)

Has the same behaviour and guarantees as foldBChan.