dispatcher

Dispatcher actor for observability events.

The dispatcher is an OTP actor that:

Receives Event(SessionEvent) messages
Calls emit_telemetry(event) — projects lightweight metrics to :telemetry (ALWAYS, by construction)
Fans out the full event to registered consumers (fire-and-forget process.send)
Accepts RegisterConsumer(StartedConsumer) to dynamically add consumers

Types

DispatcherMessage

Messages that the dispatcher actor can receive.

pub type DispatcherMessage {
  Event(events.SessionEvent)
  RegisterConsumer(consumer_spec.StartedConsumer)
  RegisterConsumerSync(
    consumer_spec.StartedConsumer,
    process.Subject(Nil),
  )
  Flush(reply_to: process.Subject(Nil))
  Shutdown(reply_to: process.Subject(Result(Nil, ShutdownError)))
  Stop
}

Constructors

```
Event(events.SessionEvent)
```
A session event to dispatch to consumers and telemetry.
```
RegisterConsumer(consumer_spec.StartedConsumer)
```
Register a new consumer to receive session events.

RegisterConsumerSync(
  consumer_spec.StartedConsumer,
  process.Subject(Nil),
)

```
Flush(reply_to: process.Subject(Nil))
```
Acknowledge after all earlier events have been dispatched.
```
Shutdown(reply_to: process.Subject(Result(Nil, ShutdownError)))
```
Drain events and ask every registered consumer to acknowledge shutdown.
```
Stop
```
Stop the dispatcher actor (for testing/cleanup).

FlushError

</>

Errors returned when flushing the dispatcher.

pub type FlushError {
  FlushTimeout
}

Constructors

```
FlushTimeout
```
The dispatcher did not acknowledge the barrier before the timeout.

RegistrationError

</>

Errors returned when registering a dispatcher consumer.

pub type RegistrationError {
  RegistrationTimeout
}

Constructors

```
RegistrationTimeout
```
The dispatcher did not acknowledge registration before the timeout.

ShutdownError

</>

Errors returned when draining consumers through the dispatcher.

pub type ShutdownError {
  ShutdownTimeout
  ConsumerStop(error: consumer_spec.StopError)
}

Constructors

```
ShutdownTimeout
```
The dispatcher did not acknowledge the shutdown request before the timeout.
```
ConsumerStop(error: consumer_spec.StopError)
```
A consumer did not acknowledge its graceful stop.

Values

flush

</>

pub fn flush(
  dispatcher: process.Subject(DispatcherMessage),
) -> Nil

Wait until every event sent before this call has been dispatched.

flush_with_timeout

</>

pub fn flush_with_timeout(
  dispatcher: process.Subject(DispatcherMessage),
  timeout_ms: Int,
) -> Result(Nil, FlushError)

Wait for every event sent before this call to be dispatched.

register_consumer

</>

pub fn register_consumer(
  dispatcher: process.Subject(DispatcherMessage),
  consumer: consumer_spec.StartedConsumer,
) -> Result(Nil, RegistrationError)

The barrier makes it safe to emit an event immediately after registration. Returns a typed error if the dispatcher does not acknowledge registration.

register_consumer_sync

</>

pub fn register_consumer_sync(
  dispatcher: process.Subject(DispatcherMessage),
  consumer: consumer_spec.StartedConsumer,
) -> Result(Nil, RegistrationError)

Synchronous alias for register_consumer.

register_consumer_with_timeout

</>

pub fn register_consumer_with_timeout(
  dispatcher: process.Subject(DispatcherMessage),
  consumer: consumer_spec.StartedConsumer,
  timeout_ms: Int,
) -> Result(Nil, RegistrationError)

This is useful for callers that need a bounded startup failure and for deterministic tests; successful registration remains synchronous.

shutdown

</>

pub fn shutdown(
  dispatcher: process.Subject(DispatcherMessage),
) -> Result(Nil, ShutdownError)

Drain the dispatcher and request a graceful stop from every consumer.

The dispatcher handles this message after all earlier events in its own mailbox. Consumer stop messages are sent by that same process, preserving event-before-stop ordering for each consumer.

shutdown_with_timeout

</>

pub fn shutdown_with_timeout(
  dispatcher: process.Subject(DispatcherMessage),
  timeout_ms: Int,
) -> Result(Nil, ShutdownError)

Drain the dispatcher with an explicit shutdown acknowledgement timeout.

start

</>

pub fn start() -> Result(
  process.Subject(DispatcherMessage),
  actor.StartError,
)

Start a new dispatcher actor.

supervised

</>

pub fn supervised(
  name: process.Name(DispatcherMessage),
) -> supervision.ChildSpecification(Nil)

Create a supervised dispatcher actor for use in a supervision tree.

supervised_with_consumers

</>

pub fn supervised_with_consumers(
  name: process.Name(DispatcherMessage),
  consumers: List(consumer_spec.StartedConsumer),
) -> supervision.ChildSpecification(Nil)

Create a supervised dispatcher with its named consumers configured at start.

The subjects are retained by every dispatcher reconstruction, so a OneForAll restart cannot lose the consumer registrations.