Working with `nim-brokers`

Drop-in CLAUDE.md addon for any project that depends on the brokers nimble package. Type-safe, decoupled messaging on top of chronos + results. All public APIs are exception-free: errors ride Result[T, string], never raises.

Mental model

Three macros, each declares a broker type and generates its full API. The type is the channel — you call class-method-style on the typedesc: T.emit, T.request, T.listen, T.setProvider. No instances, no singletons to wire.

Macro	Pattern	Producer side	Consumer side
`EventBroker`	pub/sub, many→many, fire-and-forget	`T.emit(...)`	`T.listen(handler)`
`RequestBroker`	request/response, single provider	`T.setProvider(handler)`	`T.request(...)`
`MultiRequestBroker`	request/response, many providers, fan-out	`T.setProvider(handler)` (N×)	`T.request(...)`
`SignalBroker`	one-way notification, single handler, no reply	`T.signal(...)`	`T.onSignal(handler)`

(mt) suffix → multi-thread variant (cross-thread dispatch). (sync) on RequestBroker → blocking, non-async. (API) → FFI shared-library surface.

Any provider/handler that is a class method (self.foo) can be installed with the bind* / rebind* sugar instead of a hand-written forwarding closure — see "Binding class-method providers" below.

Import only what you use:

import brokers/event_broker
import brokers/request_broker
import brokers/multi_request_broker
import brokers/signal_broker
import brokers/broker_context   # only if you need explicit contexts

EventBroker — pub/sub

import chronos, brokers/event_broker

EventBroker:
  type UserLoggedIn = object
    userId*: int
    name*: string

# listen returns Result[ListenerHandle, string]; keep the handle to drop later
let h = UserLoggedIn.listen(
  proc(evt: UserLoggedIn): Future[void] {.async: (raises: []).} =
    info "login", id = evt.userId
)

UserLoggedIn.emit(UserLoggedIn(userId: 7, name: "zoli"))   # by value
UserLoggedIn.emit(userId = 7, name = "zoli")               # by fields (inline-object only)

await UserLoggedIn.dropListener(h.get())   # drop one — cancels its in-flight work
await UserLoggedIn.dropAllListeners()      # drop all for this context

emit is sync void in every lane (single-thread, (mt), (API)): snapshots listeners, asyncSpawns each. It does not await delivery — await sleepAsync(0) or yield to flush in tests. Never await/waitFor an emit.
Handlers MUST be {.async: (raises: []).}. Swallow your own exceptions.
dropListener/dropAllListeners are async (Future[void]) in every lane — await them (or discard/waitFor in sync/{.thread.} contexts). Single-thread cancels in-flight handlers before returning; MT/API bodies are suspension-free.

Payload variants

EventBroker:
  type Tick = void              # payload-less signal: Tick.emit() / listen(proc(): Future[void]...)
EventBroker:
  type Score = int              # native/alias/external types auto-wrapped in distinct
EventBroker:
  type Blob = ref object        # ref payloads fine
    data*: seq[byte]

RequestBroker — single provider request/response

Two declaration styles. Coupled (named type + proc) and proc-sugar (payload decoupled, broker named after the Capitalized verb).

import chronos, brokers/request_broker

# Coupled: broker name == type name == request() return payload
RequestBroker:
  type FetchUser = object
    name*: string
  proc signature*(id: int): Future[Result[FetchUser, string]] {.async.}

FetchUser.setProvider(
  proc(id: int): Future[Result[FetchUser, string]] {.async.} =
    ok(FetchUser(name: "u" & $id))
).isOk()

let r = await FetchUser.request(42)        # Result[FetchUser, string]
FetchUser.clearProvider()

# Proc-sugar: broker = Capitalized verb, request() returns the RAW payload
RequestBroker:
  proc getVersion(): Future[Result[string, string]] {.async.}   # -> broker `GetVersion`

GetVersion.setProvider(
  proc(): Future[Result[string, string]] {.async.} = ok("1.2.3")).get()
let v = await GetVersion.request()         # r.value is plain string, no unwrap

Rules & behaviors:

One provider per signature. A second setProvider returns err(...) (no silent override). clearProvider() first to swap.
Two signature slots coexist: zero-arg and arg-based (overload by arity).
Provider exceptions are caught → err(<msg>). Unset provider → err(...).
isProvided() checks registration. T.request is async here.

Sync mode — no event loop needed

RequestBroker(sync):
  proc getId(): Result[int, string]              # note: no Future, no {.async.}
GetId.setProvider(proc(): Result[int, string] = ok(42)).isOk()
let id = GetId.request()                          # blocking, returns Result directly

void payload (action with no return value)

RequestBroker:
  proc doReset(force: bool): Future[Result[void, string]] {.async.}
DoReset.setProvider(proc(force: bool): Future[Result[void, string]] {.async.} =
  if force: ok() else: err("need force")).isOk()

MultiRequestBroker — fan-out to many providers

Async only. request() calls all providers via allFinished, returns Result[seq[Payload], string]. Any provider failing fails the whole request.

import chronos, brokers/multi_request_broker

MultiRequestBroker:
  type Quote = object
    price*: int
  proc signature*(sym: string): Future[Result[Quote, string]] {.async.}

discard Quote.setProvider(proc(sym: string): Future[Result[Quote, string]] {.async.} =
  ok(Quote(price: 100)))
discard Quote.setProvider(proc(sym: string): Future[Result[Quote, string]] {.async.} =
  ok(Quote(price: 101)))

let all = await Quote.request("BTC")    # all.get() is seq[Quote], len == 2
Quote.removeProvider(handle.get())      # remove one (handle from setProvider)
Quote.clearProviders()                  # remove all

No providers registered → ok(@[]) (empty, not error).
Identical handler refs deduplicated on registration.
setProvider returns Result[ProviderHandle, string]; capture it for removeProvider.

SignalBroker — one-way notification, single handler

Fire-and-forget into a module: an inverted EventBroker (single handler, no reply path). signal() is a plain (non-async) proc returning Result[void, string]; it does not tell you whether the handler succeeded — only whether it was accepted. Handler exceptions are swallowed (chronicles warn). For delivery confirmation, use RequestBroker with a void response.

import chronos, brokers/signal_broker

SignalBroker:
  type IngestSample = object
    deviceId*: string
    value*: float64

# ONE handler (a second onSignal returns err). Handler is async, raises: [].
discard IngestSample.onSignal(
  proc(s: IngestSample) {.async: (raises: []).} =
    info "sample", dev = s.deviceId, v = s.value)

let r = IngestSample.signal(IngestSample(deviceId: "d1", value: 0.5))  # by value
discard IngestSample.signal(deviceId = "d2", value = 1.25)             # by fields
# r: Result[void, string]
#   ok()  = ACCEPTED (a handler exists + queue had room) — NOT "handled"
#   err() = "no signal handler installed" | "queue full"

await IngestSample.dropSignalHandler()   # async Future[void] — await it

signal() is sync, never await it. dropSignalHandler() is async.
type Foo = void → payload-less pulse: Foo.signal() / onSignal(proc() ...).
Mock/replace trio (owning-thread only on (mt)): replaceSignalHandler, getCurrentSignalHandler, withMockSignalHandler(ctx, mock): body.
(mt) and (API) variants mirror the other brokers (one handler per context).

Binding class-method providers — `bind` / `rebind` (v3.1)

Nim has no bound-method values (self.send is not a closure), so installing a class method as a provider/handler normally needs a hand-written trampoline. The bind* sugar synthesises it — identical codegen, identical self capture. Passing a plain closure works too, so it is a strict superset of the typed verbs (setProvider / listen / onSignal stay untouched).

# before — hand-written forwarding closure
MessagingSend.setProvider(self.brokerCtx,
  proc(e: MessageEnvelope): Future[Result[RequestId, string]] {.async.} =
    await self.send(e))

# after — the sugar generates exactly that trampoline
MessagingSend.bindProvider(self.brokerCtx, self.send)

Broker	install sugar	replace sugar
`RequestBroker` / `(mt)`	`bindProvider`	`rebindProvider`
`MultiRequestBroker`	`bindProvider` (additive)	—
`EventBroker` / `(mt)`	`bindListener` (returns the listen handle)	—
`SignalBroker` / `(mt)`	`bindSignalHandler`	`rebindSignalHandler`

Each verb has a ctx-form (bindProvider(ctx, m)) and a no-ctx-form (bindProvider(m) → thread-global context).
Dual-slot RequestBrokers (zero-arg and arg signatures) disambiguate by arity automatically.
Works on the (API) lane too — usable inside setupProviders(ctx).

BrokerContext — scoping / multi-instance

Every API takes an optional first BrokerContext arg. Omit it → the thread-global context (DefaultBrokerContext). Use contexts to run independent broker instances (per component, per test, per thread).

import brokers/broker_context

let ctx = NewBrokerContext()                       # globally-unique id (atomic)

discard MyEvent.listen(ctx, handler)
MyEvent.emit(ctx, payload)
FetchUser.setProvider(ctx, provider)
let r = await FetchUser.request(ctx, 42)
await MyEvent.dropAllListeners(ctx)

Thread setup helpers (callable before the event loop starts):

Call	Use
`setThreadBrokerContext(ctx)`	adopt a context created elsewhere as this thread's global
`initThreadBrokerContext(): BrokerContext`	create + set as thread-global in one call
`threadGlobalBrokerContext()`	read current thread global (lock-free)

Async scoped swap (needs chronos loop): lockGlobalBrokerContext / lockNewGlobalBrokerContext templates.

Multi-thread variants `(mt)`

Add (mt). Identical call surface — emit stays sync void and drop* stay async (Future[void]), so the same source compiles with or without the tag. Cross-thread dispatch is handled under the hood. Build with --threads:on.

EventBroker(mt):
  type Job = object
    id*: int

# from any thread:
proc worker() {.thread.} =
  Job.emit(Job(id: 1))             # emit is sync void — same as single-thread

Same-thread calls take a direct fast path; cross-thread go through a per-bucket channel drained by one dispatch coroutine. fd cost is O(threads), not per-broker.
A thread that listens must keep its event loop alive (the broker dispatches on it).
MT brokers accept capacity kwargs: EventBroker(mt, queueDepth = ..., slabCapacity = ..., maxPayloadBytes = ..., preset = "..."). Omit for defaults.

Decision guide

You want…	Use
Notify N listeners, don't care about replies	`EventBroker`
Ask one authority for an answer	`RequestBroker`
Blocking call, no async context	`RequestBroker(sync)`
Ask everyone, aggregate replies	`MultiRequestBroker`
One-way notify a single handler, no reply	`SignalBroker`
Same pattern across OS threads	add `(mt)`, `--threads:on` (call surface unchanged)
Multiple isolated instances	pass a `BrokerContext` first arg
Install a class method (`self.foo`) as provider/handler	`bind` / `rebind` sugar
Expose to C/C++/Python/Rust/Go	`(API)` + `registerBrokerLibrary` (see AGENTS.md)

Gotchas

Handlers/providers are raises: [] — never let an exception escape; return err().
setProvider on a RequestBroker that already has one fails — clear first.
Single-thread emit returns immediately; await a yield before asserting in tests.
A non-object/ref object broker type is auto-wrapped in distinct; construct with T(value) and read with the base-type conversion.
Keep all interaction with one context on one thread (single-thread brokers are thread-local); cross-thread requires the (mt) variant.

FFI API `(API)` — expose brokers as a C/C++/Python/Rust/Go shared library

Add (API) to RequestBroker / EventBroker / SignalBroker. Same declaration syntax — it additionally generates a fixed C ABI and typed foreign wrappers. Wire format is CBOR; wrappers carry the typed surface. Build with -d:BrokerFfiApi --threads:on --app:lib. (SignalBroker(API) rides _call one-way — enqueue only, no response slot; _callAsync rejects it with ApiStatusOneWay.)

{.push raises: [].}
import brokers/[event_broker, request_broker, broker_context, api_library]

# Plain Nim object types used in signatures are AUTO-registered — no annotation.
type DeviceInfo* = object
  deviceId*: int64
  name*: string
  online*: bool

RequestBroker(API):
  type GetDevice = object        # broker name == type name == response payload
    deviceId*: int64
    name*: string
  proc signature*(deviceId: int64): Future[Result[GetDevice, string]] {.async.}

EventBroker(API):
  type DeviceStatusChanged = object
    deviceId*: int64
    online*: bool
    timestampMs*: int64

Providers + event emission live in one proc named setupProviders (the generated runtime calls it on the processing thread during createContext):

proc setupProviders(ctx: BrokerContext): Result[void, string] =
  let r = GetDevice.setProvider(ctx,        # always pass the ctx the runtime gives you
    proc(deviceId: int64): Future[Result[GetDevice, string]] {.closure, async.} =
      DeviceStatusChanged.emit(ctx,         # emit is sync void (all lanes)
        DeviceStatusChanged(deviceId: deviceId, online: true, timestampMs: 0))
      ok(GetDevice(deviceId: deviceId, name: "u")))
  if r.isErr(): return err("register GetDevice: " & r.error())
  ok()

# MUST be the last declaration in the module:
registerBrokerLibrary:
  name: "mylib"                  # MUST match --nimMainPrefix and the .so basename
  version: "1.0.0"              # baked into <lib>_version() static string
  initializeRequest: InitializeRequest   # post-create config broker (optional)
  shutdownRequest: ShutdownRequest        # orderly teardown broker (optional)
{.pop.}

Build (name / --nimMainPrefix / registerBrokerLibrary name must all match):

nim c -d:BrokerFfiApi --threads:on --app:lib --path:. \
  --outdir:build --nimMainPrefix:mylib mylib.nim

What you get — a fixed 12-function C ABI per library: _version, _initialize (once per process), _createContext (per instance), _shutdown(ctx), _allocBuffer, _freeBuffer, _call (sync round-trip), _callAsync (non-blocking, callback-completed), _subscribe, _unsubscribe, _listApis, _getSchema. <lib>.h (C) and <lib>.hpp (C++) are always emitted. Wire format is CBOR (the historical native C-ABI codegen was retired in 3.0.0).

Flag	Emits	Notes
(default)	`<lib>.h`, `<lib>.hpp`	C + C++ always
`-d:BrokerFfiApiGenPy`	`<lib>.py` (cbor2)	next to the `.so`
`-d:BrokerFfiApiGenRust`	`<lib>_rs/` Cargo crate	ciborium + serde
`-d:BrokerFfiApiGenGo`	`<lib>_go/` Go module	fxamacker/cbor

FFI rules:

registerBrokerLibrary is a no-op without -d:BrokerFfiApi — no when defined guard needed; the normal in-process broker API still works.
(API) brokers ride the MT lane, so they accept the same capacity kwargs as (mt): RequestBroker(API, queueDepth = .., slabCapacity = .., maxPayloadBytes = .., preset = "..").
_createContext() is readiness-synchronous: returns only after providers + listeners are installed and the event courier is live.
Inspect generated Nim with -d:brokerDebug → build/broker_debug/*.gen.nim.

BrokerInterface / BrokerImplement — hierarchical / OOP layer

An object-oriented facade over the brokers: an interface groups several brokers behind one abstract type; an implementation provides per-instance methods. Each instance gets its own BrokerContext, so two instances of the same impl are fully isolated. Direct instance.method() calls tunnel through broker dispatch (so provider mocks are honored — not a plain vtable call).

import brokers/broker_interface
import brokers/broker_implement

BrokerInterface(IGreeter):
  EventBroker:
    type Greeted = object
      who: string
  RequestBroker:
    proc greet(name: string): Future[Result[string, string]] {.async.}
  RequestBroker:
    proc version(): Future[Result[string, string]] {.async.}

type GreeterImpl = ref object of IGreeter   # MUST be `ref object of <Interface>`
  prefix: string

BrokerImplement GreeterImpl of IGreeter:
  proc new(T: typedesc[GreeterImpl], prefix: string): GreeterImpl =
    GreeterImpl(prefix: prefix)             # optional ctor; create() calls it
  method greet(self: GreeterImpl, name: string): Future[Result[string, string]] {.async.} =
    ok(self.prefix & name)
  method version(self: GreeterImpl): Future[Result[string, string]] {.async.} =
    ok("v2")

Use it:

let g = GreeterImpl.create(prefix = "hi ")   # new() + wires providers under g.brokerCtx
echo (waitFor g.greet("sue")).value          # "hi sue" — tunnels through Greet broker

let base: IGreeter = g                       # virtual dispatch via the interface type
echo (waitFor base.greet("x")).value         # resolves to the override

# Each instance is isolated by its own context:
let a = GreeterImpl.create(prefix = "a:")
let b = GreeterImpl.create(prefix = "b:")
# a.brokerCtx != b.brokerCtx

g.close()        # clears THIS instance's providers + listeners; idempotent

Event facade (instance-scoped listen/emit — context is injected for you):

discard g.listen(Greeted,
  proc(ev: Greeted): Future[void] {.async: (raises: []), gcsafe.} = …)
g.emit(Greeted, Greeted(who: "bob"))

Factory / dependency injection (resolve an impl behind the interface):

IGreeter.provideFactory(
  proc(cfg: string): Result[IGreeter, string] =
    ok(GreeterImpl.create(prefix = cfg)))
let d = IGreeter.create("cfg:")              # Result[IGreeter, string]; last factory wins

Key points:

The broker for proc greet is named Greet (Capitalized verb). Address it directly with the instance context: Greet.request(g.brokerCtx, "bob"), Greet.clearProvider(g.brokerCtx) (e.g. to install a mock).
Impl.create(args…) = fresh context + new + provider wiring. Impl.createUnderContext(ctx, args…) wires under an externally-supplied context (the path the FFI runtime drives).
BrokerInterface(API, IName) lowers the sub-brokers onto the MT/FFI lane so the whole interface can be exposed as a shared library; BrokerImplement is unchanged.
Sub-instances returned from a method (factory pattern) share the parent's classCtx (routing) but get a distinct instanceCtx — see classCtx() / instanceCtx() accessors.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Working with `nim-brokers`

Mental model

EventBroker — pub/sub

Payload variants

RequestBroker — single provider request/response

Sync mode — no event loop needed

void payload (action with no return value)

MultiRequestBroker — fan-out to many providers

SignalBroker — one-way notification, single handler

Binding class-method providers — `bind` / `rebind` (v3.1)

BrokerContext — scoping / multi-instance

Multi-thread variants `(mt)`

Decision guide

Gotchas

FFI API `(API)` — expose brokers as a C/C++/Python/Rust/Go shared library

BrokerInterface / BrokerImplement — hierarchical / OOP layer

FilesExpand file tree

CLAUDE_brokers_addon.md

Latest commit

History

CLAUDE_brokers_addon.md

File metadata and controls

Working with nim-brokers

Mental model

EventBroker — pub/sub

Payload variants

RequestBroker — single provider request/response

Sync mode — no event loop needed

void payload (action with no return value)

MultiRequestBroker — fan-out to many providers

SignalBroker — one-way notification, single handler

Binding class-method providers — bind* / rebind* (v3.1)

BrokerContext — scoping / multi-instance

Multi-thread variants (mt)

Decision guide

Gotchas

FFI API (API) — expose brokers as a C/C++/Python/Rust/Go shared library

BrokerInterface / BrokerImplement — hierarchical / OOP layer

Working with `nim-brokers`

Binding class-method providers — `bind` / `rebind` (v3.1)

Multi-thread variants `(mt)`

FFI API `(API)` — expose brokers as a C/C++/Python/Rust/Go shared library