weft/event_manager

A typed gen_event: one process holding an ordered list of handlers, each carrying its own private state.

The problem, and the encoding that answers it

gleam_otp has no gen_event binding and probably never will, because gen_event’s central structure is a list in which every element has a different state type. A token counter holds an Int, a transcript holds a file handle, and gen_event keeps both in one list. Gleam has no direct representation for that list, which is exactly why the binding is missing upstream.

The way out is to stop asking the handler to expose its state. A handler is a function from an event to its own successor, and the state lives in the successor’s closure where it never reaches the type:

Handler(step: fn(event) -> Outcome(event))

List(Handler(event)) is now homogeneous while every handler in it is still as differently-stated as gen_event’s. handler writes the recursion for you from ordinary state-threading code, so the closure trick is an implementation detail of this module rather than something callers perform by hand.

The encoding buys one thing Erlang’s does not have. A handler that is broken has to say why, because Result(state, String) is the return type its author is given; in Erlang a handler simply crashes and is silently swapped out.

What the manager guarantees

Handlers run in the order they were added, and a successor takes its predecessor’s place in the list, so the order is stable across events. add_handler at runtime appends, so a late handler runs last.
A handler returning Failed is removed and the reason is logged. Its siblings do not see the removal at all: they were run before it or are run after it, from the same fan-out.
sync_notify returns only once every handler has finished with the event. It is a call whose reply the manager sends after the fan-out, so a caller that needs “the effects are visible now” — a test, a shutdown ordering, backpressure — gets it without sleeping.

Isolation, and its honest limit

Handlers run in the manager’s process, as gen_event’s do. Two things follow, and only one of them is shared with OTP.

A slow handler stalls the manager and everything behind it. gen_event has the same property, and it is the price of the ordering above.

A handler that raises — a panic, an assertion that does not hold, a badmatch inside the closure — takes the manager down with it. Here weft is weaker than Erlang, which wraps every callback in a try/catch, removes the offender and carries on. gen_event catches because its callbacks are dynamically typed and it has no other way to be told about a broken one; weft asks for the failure in the type instead, and Failed is isolated exactly as gen_event’s removal is. What weft does not do is rescue a handler that does not know it is broken, because Gleam has no exception handling and the rescue would have to be FFI. So: a raising handler is a manager crash, and the answer to it is a supervisor, not a Failed. Better to say that out loud than to let a reader infer an isolation that is not there.

What is deliberately not here

Per-handler call. gen_event:call(Manager, HandlerId, Request) has no honest type, because the request and reply types differ per handler. The answer is that a handler which needs to be queried is not a handler, it is an actor that happens to subscribe. Hand it a Subject of its own and let callers talk to it directly. This is not a gap to be filed; it is a thing the type system is right to refuse.

Handler supervision. add_sup_handler and the gen_event_EXIT message are out for v1. The Failed-drop-and-log path covers what the removal notification is usually used for; an on_handler_exit builder option waits for a real consumer rather than being guessed at now.

Handler identity. add_handler returns Nil, so there is no HandlerRef and no delete_handler(ref). RemoveSelf covers removal from the inside, which is every case seen so far. External removal, and the reference type it would need, are deferred until something actually wants them.

Example

A session bus where one handler counts tokens and another accumulates a transcript. Neither state type is visible outside its constructor, and both are in the same list.

let assert Ok(started) =
  event_manager.new()
  |> event_manager.add(event_manager.handler(0, on_event: count_tokens))
  |> event_manager.add(event_manager.handler(transcript, on_event: append))
  |> event_manager.start

event_manager.notify(started.data, Token("hello"))

// A full disk removes the transcript with `Failed`; the counter keeps
// counting, and this returns 1.
event_manager.count_handlers(started.data, waiting: 1000)

Types

Builder

opaque </>

A description of a manager, ready to start or to hand to a supervisor.

Built with new and refined with add and named.

pub opaque type Builder(event)

Handler

opaque </>

One handler in a manager’s list, with its state sealed inside a closure.

Only event appears in the type, which is what lets handlers with unrelated state types share a List(Handler(event)). Build one with handler, or with handler_with_outcome when the handler needs to remove itself.

pub opaque type Handler(event)

Message

opaque </>

What a manager accepts.

Opaque, because every constructor has a function in this module that sends it correctly — notify, sync_notify, add_handler, count_handlers — and a hand-built SyncNotify with the wrong reply subject is a hang rather than a type error. The type is public so that a caller can name process.Name(Message(event)) for named, and Subject(Message(event)) for whatever it stores the manager in.

pub opaque type Message(event)

Outcome

</>

What a handler has decided, having seen one event.

The manager acts on this and nothing else: a handler cannot reach the manager’s list, cannot see its siblings, and cannot stop the manager.

pub type Outcome(event) {
  Keep(handler: Handler(event))
  RemoveSelf
  Failed(reason: String)
}

Constructors

```
Keep(handler: Handler(event))
```
Handled. Use this handler for the next event, in the same position.
```
RemoveSelf
```
Handled. Remove me from the manager; my siblings carry on unchanged.
```
Failed(reason: String)
```
This handler is broken and knows it. The manager drops it and logs the reason. Siblings and the manager itself are unaffected, and that isolation is gen_event’s whole value.

Values

add

</>

pub fn add(
  builder: Builder(event),
  handler: Handler(event),
) -> Builder(event)

Add a handler to the manager being described.

Handlers run in the order they were added, so this is also the order the fan-out will visit them in.

Examples

event_manager.new()
|> event_manager.add(token_counter)
|> event_manager.add(transcript)

add_handler

</>

pub fn add_handler(
  manager: process.Subject(Message(event)),
  handler: Handler(event),
) -> Nil

Add a handler to a running manager.

The handler is appended, so it runs last in the fan-out, and it sees only events sent after this call — messages to one process from one process arrive in order, so an event sent after this one cannot be handled before the handler is in the list.

There is no handle to remove it by; see this module’s header on handler identity, and RemoveSelf for removal from the inside.

Examples

event_manager.add_handler(bus, audit_log)
// The audit log sees this event, and none before it.
event_manager.notify(bus, Token("hello"))

count_handlers

</>

pub fn count_handlers(
  manager: process.Subject(Message(event)),
  waiting timeout: Int,
) -> Int

How many handlers the manager currently holds.

The count falls as handlers remove themselves and as broken ones are dropped, so it is the observable side of RemoveSelf and Failed.

waiting is milliseconds, and the caller crashes on timeout for the reason given on sync_notify.

Examples

event_manager.count_handlers(bus, waiting: 1000)
// -> 2

handler

</>

pub fn handler(
  state: state,
  on_event handle: fn(state, event) -> Result(state, String),
) -> Handler(event)

Build a handler from ordinary state-threading code.

on_event is written as if it owned a plain piece of state: take the state and the event, return the next state, or an Error saying what went wrong. The recursion that carries the state forward is written here, in the successor’s closure, so the state never escapes into the type.

An Error becomes Failed: the handler is removed and the reason logged. A handler that wants to retire without having failed wants handler_with_outcome and RemoveSelf.

Examples

// State is an `Int` and is invisible from outside.
let tokens =
  event_manager.handler(0, on_event: fn(count, event) {
    case event {
      Token(_) -> Ok(count + 1)
      Report(reply:) -> {
        process.send(reply, count)
        Ok(count)
      }
    }
  })

// The same shape, failing out of the list when the disk fills. The type
// forces the handler to say why it is going.
let transcript =
  event_manager.handler(file, on_event: fn(file, event) {
    case write(file, event) {
      Ok(file) -> Ok(file)
      Error(reason) -> Error("transcript write failed: " <> reason)
    }
  })

handler_with_outcome

</>

pub fn handler_with_outcome(
  on_event step: fn(event) -> Outcome(event),
) -> Handler(event)

Build a handler that returns its own Outcome.

This is handler without the state-threading convenience, and it exists for the outcomes that convenience cannot express — RemoveSelf above all, since a handler that has finished its work is not a handler that has Failed. The caller writes the successor recursion by hand and gets RemoveSelf in exchange.

Examples

// A handler that fires once and then retires.
let once =
  event_manager.handler_with_outcome(fn(event) {
    announce(event)
    event_manager.RemoveSelf
  })

// Hand-written state threading, for a handler that also removes itself.
fn countdown(remaining: Int) -> event_manager.Handler(event) {
  use _event <- event_manager.handler_with_outcome
  case remaining {
    0 -> event_manager.RemoveSelf
    n -> event_manager.Keep(countdown(n - 1))
  }
}

named

</>

pub fn named(
  builder: Builder(event),
  name: process.Name(Message(event)),
) -> Builder(event)

Register the manager under name when it starts, so it can be reached by a named subject rather than by passing a subject around.

If the name is already registered the manager fails to start. This is how a supervised bus is reached: the supervisor holds the child, and everyone else sends to process.named_subject(name).

Examples

let name = process.new_name("session_bus")
let assert Ok(_) =
  event_manager.new() |> event_manager.named(name) |> event_manager.start
event_manager.notify(process.named_subject(name), Token("hello"))

new

</>

pub fn new() -> Builder(event)

Describe a manager with no handlers.

A manager with an empty list is useful rather than degenerate: every notify is a no-op until something subscribes with add_handler, which is the usual shape for a bus started by a supervisor before its consumers exist.

Examples

let assert Ok(started) = event_manager.new() |> event_manager.start
event_manager.add_handler(started.data, audit_log)

notify

</>

pub fn notify(
  manager: process.Subject(Message(event)),
  event: event,
) -> Nil

Send an event to every handler, without waiting.

Returns as soon as the event is in the manager’s mailbox, which is before any handler has seen it. Use sync_notify when the caller needs the handlers’ effects to have happened.

Examples

event_manager.notify(bus, Token("hello"))

start

</>

pub fn start(
  builder: Builder(event),
) -> Result(
  actor.Started(process.Subject(Message(event))),
  actor.StartError,
)

Start a manager from a builder.

The manager is a weft/actor whose state is the handler list, so the process is linked to the caller and started.data is the subject to send on, exactly as any other weft actor’s is.

Examples

let assert Ok(started) =
  event_manager.new()
  |> event_manager.add(token_counter)
  |> event_manager.start
event_manager.notify(started.data, Token("hello"))

supervised

</>

pub fn supervised(
  builder: Builder(event),
) -> supervision.ChildSpecification(
  process.Subject(Message(event)),
)

Describe this manager as a supervisor’s child.

Returns gleam/otp/supervision’s own ChildSpecification, by way of weft/actor, so a manager is added to a gleam_otp supervisor exactly as any other actor is. A supervised manager is nearly always a named one, since a restart replaces the subject the caller was holding.

Examples

supervisor.new(supervisor.OneForOne)
|> supervisor.add(event_manager.supervised(bus_builder))
|> supervisor.start

sync_notify

</>

pub fn sync_notify(
  manager: process.Subject(Message(event)),
  event: event,
  waiting timeout: Int,
) -> Nil

Send an event to every handler and wait until all of them have handled it.

The manager replies after the fan-out, never before, so when this returns every handler has run and every effect a handler performed is visible. That makes it the backpressure variant — a producer that outruns the bus blocks here — and the shutdown-ordering one, and it is why a test can assert on a handler’s effects with a zero timeout instead of a sleep.

waiting is milliseconds. As with weft/actor’s call, which this is, the caller crashes if the manager does not reply in time rather than carrying on against a process in an unknown state.

Examples

event_manager.sync_notify(bus, Token("hello"), waiting: 1000)
// -> every handler has now finished with the event