RIO.ProcessPool.Broker

A broker extracts a key value from incoming messages and creates, keeps and destroys a resource for each key.

The demultiplexed messages and their resources are passed to a custom MessageHandler/

The user provides a Demultiplexer is a pure function that returns a key for the resource associated to the message and potientially changes the message.

The demultiplexer may also return a value indicating that a new resource must be created, or that a message shall be ignored.

The broker is run in a seperate process using async. The usual way to stop a broker is to cancel it.

When cancelling a broker, the resource cleanup actions for all resources will be called with async exceptions masked.

In order to prevent the resource map filling up with dead resources, the user of this module has to ensure that whenever a resource is not required anymore, a message will be sent to the broker, that will cause the MessageHandler to be executed for the resource, which will in turn return, return RemoveResource.

Documentation

spawnBroker :: forall brokerBoxArg k w' w a m. (HasLogFunc m, Ord k, Display k, IsMessageBoxArg brokerBoxArg) => brokerBoxArg -> BrokerConfig k w' w a m -> RIO m (Either SomeException (Input (MessageBox brokerBoxArg) w', Async BrokerResult))

Spawn a broker with a new MessageBox, and return its message Input channel as well as the Async handle of the spawned process, needed to stop the broker process.

• k is the key for the resource associated to an incoming message
• w' is the type of incoming messages.
• w is the type of the demultiplexed messages.
• a specifies the resource type.
• m is the base monad

data BrokerConfig k w' w a m

The broker configuration, used by spawnBroker.

• k is the key for the resource associated to an incoming message
• w' is the type of incoming messages.
• w is the type of the demultiplexed messages.
• a specifies the resource type.
• m is the base monad

Constructors

MkBrokerConfig

Fields

• demultiplexer :: !(Demultiplexer w' k w)
• messageDispatcher :: !(MessageHandler k w a m)
• resourceCreator :: !(ResourceCreator k w a m)
• resourceCleaner :: !(ResourceCleaner k a m)

data BrokerResult

This is just what the Async returned from spawnBroker returns, it's current purpose is to make code easier to read.

Instead of some Async () that could be anything, there is Async BrokerResult.

Constructors

MkBrokerResult

Instances

Eq BrokerResult

Defined in RIO.ProcessPool.Broker

Methods

(==) :: BrokerResult -> BrokerResult -> Bool

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

Show BrokerResult

Defined in RIO.ProcessPool.Broker

Methods

showsPrec :: Int -> BrokerResult -> ShowS

show :: BrokerResult -> String

showList :: [BrokerResult] -> ShowS

type ResourceCreator k w a m = k -> Maybe w -> RIO m a

User supplied callback to create and initialize a resource. (Sync-) Exceptions thrown from this function are caught, and the broker continues.

• k is the key for the resource associated to an incoming message
• w is the type of the demultiplexed messages.
• a specifies the resource type.
• m is the monad of the returned action.

type Demultiplexer w' k w = w' -> Multiplexed k w

User supplied callback to extract the key and the Multiplexed from a message. (Sync-) Exceptions thrown from this function are caught and lead to dropping of the incoming message, while the broker continues.

• k is the key for the resource associated to an incoming message
• w' is the type of incoming messages.
• w is the type of the demultiplexed messages.

type ResourceCleaner k a m = k -> a -> RIO m ()

User supplied callback called _with exceptions masked_ when the MessageHandler returns RemoveResource (Sync-) Exceptions thrown from this function are caught, and do not prevent the removal of the resource, also the broker continues.

• k is the key for the resource associated to an incoming message
• a specifies the resource type.
• m is the monad of the returned action.

type MessageHandler k w a m = k -> w -> a -> RIO m (ResourceUpdate a)

User supplied callback to use the Multiplexed message and the associated resource. (Sync-) Exceptions thrown from this function are caught and lead to immediate cleanup of the resource but the broker continues.

• Type k is the key for the resource associated to an incoming message
• Type w is the type of incoming, demultiplexed, messages.
• Type a specifies the resource type.
• Type m is the base monad

data Multiplexed k w

The action that the broker has to take for in incoming message.

• k is the key for the resource associated to an incoming message
• w is the type of the demultiplexed messages.

Constructors

Initialize k !(Maybe w)

The message is an initialization message, that requires the creation of a new resouce for the given key. When the resource is created, then maybe additionally a message will also be dispatched.

Dispatch k w

Dispatch a message using an existing resource. Silently ignore if no resource for the key exists.

data ResourceUpdate a

This value indicates in what state a worker is in after the MessageHandler action was executed.

Constructors

KeepResource

The resources is still required.

UpdateResource a

The resource is still required but must be updated.

RemoveResource !(Maybe a)

The resource is obsolete and can be removed from the broker. The broker will call ResourceCleaner either on the current, or an updated resource value.