Safe Haskell	None
Language	Haskell2010

RIO.ProcessPool.Broker

Description

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.

Synopsis

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))
data BrokerConfig k w' w a m = MkBrokerConfig {
- demultiplexer :: !(Demultiplexer w' k w)
- messageDispatcher :: !(MessageHandler k w a m)
- resourceCreator :: !(ResourceCreator k w a m)
- resourceCleaner :: !(ResourceCleaner k a m)
}
data BrokerResult = MkBrokerResult
type ResourceCreator k w a m = k -> Maybe w -> RIO m a
type Demultiplexer w' k w = w' -> Multiplexed k w
type ResourceCleaner k a m = k -> a -> RIO m ()
type MessageHandler k w a m = k -> w -> a -> RIO m (ResourceUpdate a)
data Multiplexed k w
- = Initialize k !(Maybe w)
- | Dispatch k w
data ResourceUpdate a
- = KeepResource
- | UpdateResource a
- | RemoveResource !(Maybe a)

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)) Source #

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 Source #

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 Source #

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

Instances details

Eq BrokerResult Source #
Instance details Defined in RIO.ProcessPool.Broker Methods (==) :: BrokerResult -> BrokerResult -> Bool # (/=) :: BrokerResult -> BrokerResult -> Bool #
Show BrokerResult Source #
Instance details 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 Source #

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 Source #

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 () Source #

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) Source #

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 Source #

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 Source #

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.