nim-brokers - Tutorial

Every project hand-rolls the same plumbing

RequestBroker:        # in-proc: direct dispatch
RequestBroker(mt):    # cross-thread: lock-free ring
RequestBroker(API):   # cross-language: CBOR shared lib

Mental model — 4 shapes × 3 lanes

Use case: the producer must not import its consumers

Running example — a telemetry pipeline: a sensor driver produces readings; a dashboard, a logger, later a network relay all react.

• Adding a consumer touches only the consumer's module.
• No init-time wiring, no shared registry object, no ownership question.
• Everything runs on the one chronos loop you already have.

From this code… to that code

# driver.nim owns its consumers' hooks
type SensorDriver = ref object
  onReading: seq[proc(id: string,
                      c: float) {.gcsafe.}]

# main.nim — wiring, ordering, removal:
# all manual, all coupled
driver.onReading.add(dashboard.update)
driver.onReading.add(logger.record)

# telemetry_events.nim — the contract
EventBroker:
  type SensorReading = object
    sensorId*: string
    valueC*: float

# dashboard.nim — subscribes itself
discard SensorReading.listenIt:
  updateGauge(it.sensorId, it.valueC)

# driver.nim — imports nobody
SensorReading.emit(sensorId = "t1",
                   valueC = 21.5)

EventBroker — the rules

• emit is sync void: it snapshots the listener set and asyncSpawns each — never await it.
• Listener signature: proc(evt: T): Future[void] {.async: (raises: []).} — handle your own errors.
• listen / listenIt return Result[ListenerHandle, string] — keep the handle to drop later.
• drop* is async in every lane — await it; dropping cancels the listener's in-flight work.

let h = SensorReading.listenIt:
  await store(it)                     # bodies may await

SensorReading.emit(sensorId = "t1", valueC = 21.5)
await sleepAsync(0)                   # test idiom: yield once so handlers run

await SensorReading.dropListener(h.get())
await SensorReading.dropAllListeners()

Full reference: USAGEGUIDE.md § EventBroker · COOKBOOK.md § EventBroker

One provider, typed, swappable — the DI seed

# contract — visible to callers and provider alike
RequestBroker:
  type Calibration = object
    offset*: float
  proc signature*(sensorId: string): Future[Result[Calibration, string]] {.async.}

# storage.nim — the provider (exactly one per signature)
discard Calibration.provideIt:                # declared arg names injected
  ok(Calibration(offset: lookupOffset(sensorId)))

# driver.nim — the consumer; no import of storage
let cal = await Calibration.request("t1")

# swap in place — tests, hot reconfig — no clearProvider dance
discard Calibration.reprovideIt:
  ok(Calibration(offset: 0.0))

Sync mode & fool-proof provider bodies

RequestBroker(sync):
  proc plusOne(x: int): Result[int, string]
  # proc-sugar: broker = PlusOne,
  # request() returns the RAW payload

discard PlusOne.setProvider(
  proc(x: int): Result[int, string] =
    ok(x + 1))

echo PlusOne.request(41).get()  # 42, blocks

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

# a body that could fall through would
# silently answer err("") — so it's a
# COMPILE error instead:

FetchUser.provideIt:
  echo "side effect only"   # void trailing call

FetchUser.provideIt:
  if id > 0:                # if without else
    return ok(FetchUser(name: "u" & $id))

Fan-out: health probes, plugin queries, capability discovery

MultiRequestBroker:
  type Health = object
    subsystem*: string
    healthy*: bool
  proc signature*(deep: bool): Future[Result[Health, string]] {.async.}

# every provideIt ADDS a provider (no replace verb here)
let h1 = Health.provideIt: ok(Health(subsystem: "net",  healthy: netUp()))
let h2 = Health.provideIt: ok(Health(subsystem: "disk", healthy: diskOk(deep)))

let all = await Health.request(true)    # Result[seq[Health], string] — len == 2
Health.removeProvider(h1.get())         # drop one by handle

Single-thread lane only — no (mt)/(API) variant.

"Was it accepted?" — not "was it handled?"

SignalBroker:
  type FlushHint = object
    reason*: string

discard FlushHint.onSignalIt:          # ONE handler; a second onSignal => err
  await persister.flush()              # exceptions swallowed (warn) — no reply path

let r = FlushHint.signal(reason = "buffer 90%")   # plain proc — NEVER await it
if r.isErr:
  metrics.inc("flush_dropped")         # backpressure is a visible branch

Which shape do I reach for?

A distinct uint32 that multiplexes broker instances

let ctxA = NewBrokerContext()   # globally unique (atomic counter, thread-safe)
let ctxB = NewBrokerContext()

discard SensorReading.listenIt(ctxA): recordTenantA(it)
discard SensorReading.listenIt(ctxB): recordTenantB(it)

SensorReading.emit(ctxA, SensorReading(sensorId: "t1", valueC: 20.0))
# only ctxA's listener fires — ctxB is fully isolated

await SensorReading.dropAllListeners(ctxA)

• Every broker API takes an optional first context arg; omitted → thread-global DefaultBrokerContext.
• Per-tenant, per-component, per-test isolation — same broker type, zero extra declarations.
• Foundation for what comes later: BrokerInterface gives every instance its own context.

From hand-rolled vtable… to a declared contract

type Greeter = ref object of RootObj
method greet(g: Greeter, name: string):
    Future[Result[string, string]] {.base.} =
  doAssert false, "override me"

type GreeterImpl = ref object of Greeter
# ...plus: factory, registry lookup,
# and mock plumbing rebuilt per test

BrokerInterface(IGreeter):
  RequestBroker:
    proc greet(name: string):
      Future[Result[string, string]] {.async.}

Implement, use — and mock without a framework

type GreeterImpl = ref object of IGreeter
  prefix: string

BrokerImplement GreeterImpl of IGreeter:
  proc new(T: typedesc[GreeterImpl], prefix: string): GreeterImpl =
    GreeterImpl(prefix: prefix)
  method greet(self: GreeterImpl, name: string): Future[Result[string, string]] {.async.} =
    ok(self.prefix & name)

let g = GreeterImpl.create(prefix = "hello ")
assert (waitFor g.greet("alice")).value == "hello alice"

# the call tunnels through the broker — so a scoped mock is honored
# even for the DIRECT method call:
Greet.withMockProvider(g.brokerCtx,
  proc(name: string): Future[Result[string, string]] {.async.} =
    ok("MOCK<" & name & ">")):
  assert (waitFor g.greet("alice")).value == "MOCK<alice>"

The transition is one token

RequestBroker:          # single-thread: direct dispatch on one chronos loop
RequestBroker(mt):      # cross-thread: lock-free ring — CALL SITES UNCHANGED

• Compile with --threads:on and --mm:orc or --mm:refc.
• When you want it: provider lives on a dedicated thread (main/UI loop) and workers query it · typed decoupled interface across threads without channel plumbing · sandboxed BrokerContexts served by different threads.
• The invariant holds: emit/signal sync, drop* async — the same source compiles in both lanes.

Under the hood (v2): ring + slab + pool

• No Channel[T] on the hot path — pre-allocated payload slab + (for requests) a response-slot pool; zero per-call shared-heap allocation.
• One shared ThreadSignalPtr per thread — all broker types on a thread share it; zero OS fds per broker type.
• Same-thread calls bypass the ring entirely — near-zero overhead.
• Everything is sized at compile time (type-driven defaults, overridable).

RequestBroker(mt) — provider on main, workers ask

RequestBroker(mt):
  type Weather = object
    city*: string
    tempC*: float
  proc signature*(city: string): Future[Result[Weather, string]] {.async.}

proc worker() {.thread.} =
  let res = waitFor Weather.request("Berlin")   # ANY thread may ask
  doAssert res.isOk()

proc main() {.async.} =
  doAssert Weather.setProvider(                 # provider runs on THIS thread
    proc(city: string): Future[Result[Weather, string]] {.async.} =
      ok(Weather(city: city, tempC: 21.5))
  ).isOk()
  var t: Thread[void]
  t.createThread(worker)
  # ... join, then: Weather.clearProvider()

Exactly the section-03 shape — the request line did not change.

RequestBroker(mt) — operational rules

• The provider runs on the thread that called setProvider and serves requests sequentially on its event loop — throughput is bounded by handler time.
• Cross-thread timeout — default 5 s; unresponsive provider ⇒ err, not a hang:

Weather.setRequestTimeout(chronos.seconds(2))
echo Weather.requestTimeout()          # 2 seconds — same-thread calls unaffected

• Same-thread path = direct provider call through threadvar state — zero ring/slab traffic.

EventBroker(mt) — broadcast without copies

• Listeners register on any thread; emitters emit from any thread. Same-thread delivery skips the ring (asyncSpawn).
• Cross-thread fan-out: the event is encoded once into a slab cell, refcounted to N_listeners, then one ring enqueue + one signal per listener thread — no per-listener payload copy.
• emit() stays sync void — marshal + enqueue happen before it returns.

Tuning: size it at the declaration

# Wide ring + slab for bursty broadcast — built-in preset:
EventBroker(mt, preset = fastBurst, maxPayloadBytes = 1024):
  type WireEvent = object
    payload*: seq[byte]

# Memory-constrained, fully manual:
RequestBroker(mt, queueDepth = 16, responseSlots = 8, maxResponseBytes = 512):
  type LedState = object
  proc query*(id: uint8): Future[Result[LedState, string]] {.async.}

• Defaults are type-driven — sensible sizes derived from the payload shape.
• Every MT callsite emits a compile-time hint: resolved capacities, their origin (default / kwarg / preset:<n> / auto:<reason>) and an idle-RAM estimate — you see at build time what the broker reserves.

Full knob/preset/failure-mode reference: doc/MT_BROKER_CONFIG.md

MT checklist — the six things people get wrong

• --threads:on + --mm:orc|refc; prefer -d:release + orc for throughput.
• emit/signal stay sync — never await; drop* stays async — always await (or waitFor from {.thread.} procs).
• Overflow of a bounded ring/pool is an err — branch on it; don't discard blindly.
• Keep the provider thread's loop responsive — it serves sequentially.
• dropListener from the registering thread; dropAllListeners from anywhere.
• Bursty or large payloads? Set preset/queueDepth/maxPayloadBytes — read the compile-time hint line.

(API) = the MT broker + a generated language boundary

RequestBroker(API):     # on the Nim side: just a normal multi-thread broker
  type Weather = object ...

• registerBrokerLibrary generates the lifecycle exports, context registry, startup threads — and the wrapper artifacts.
• You get: libmylib + C header over a fixed 12-function CBOR C ABI + idiomatic C++ / Python / Rust / Go wrappers — typed Result/error surfaces, no hand-written FFI plumbing.
• Requests and events tunnel over extensible CBOR — the same contract you declared in Nim.
• MultiRequestBroker has no API lane; SignalBroker(API) is one-way (see call modes).

Author once — BrokerInterface(API, IWeather)

type Forecast* = object       # plain type — auto-registered on reference
  day*: string
  tempC*: float64
  outlook*: string

BrokerInterface(API, IWeather):   # (API) propagates to every inner broker
  EventBroker:
    type StormWarning = object
      region*: string
      severity*: int32

  RequestBroker:
    proc getDayForecast(day: string):
      Future[Result[Forecast, string]] {.async.}

  RequestBroker:
    proc getCurrentTempC(city: string):
      Future[Result[float64, string]] {.async.}

# + WeatherImpl = ref object of IWeather & BrokerImplement (section 07),
#   then — last in the module:
registerBrokerLibrary:
  name: "weather"
  version: "1.0.0"
  ...

…generated four times — same contract, native dialects

class Weather {
  Result<Forecast> getDayForecast(std::string_view day);
  Result<double>   getCurrentTempC(std::string_view city);
  // + getDayForecastAsync(day, cb) / getDayForecastFuture(day)

  EventHandle onStormWarning(std::function<void(
    Weather& owner, std::string_view region, int32_t severity)> cb);
  void offStormWarning(EventHandle h = 0);   // 0 = remove all
};

class Weather:
    def get_day_forecast(self, day) -> Result
    def get_current_temp_c(self, city) -> Result
    # + get_day_forecast_async(day, cb)

    def on_storm_warning(self, cb) -> int   # cb(owner, region, severity)
    def off_storm_warning(self, handle=0)   # 0 = remove all

impl Weather {
    pub fn get_day_forecast(&self, day: &str) -> Result<Forecast>;
    pub fn get_current_temp_c(&self, city: &str) -> Result<f64>;
    // + get_day_forecast_async(day, timeout).await

    pub fn on_storm_warning(&self,
        cb: impl Fn(String, i32) + Send + 'static) -> EventHandle;
    pub fn off_storm_warning(&self, h: EventHandle);  // 0 = remove all
}

func (w *Weather) GetDayForecast(day string) (Forecast, error)
func (w *Weather) GetCurrentTempC(city string) (float64, error)
// + GetDayForecastContext(ctx, day)

func (w *Weather) OnStormWarning(
    cb func(region string, severity int32)) EventHandle
func (w *Weather) OffStormWarning(h EventHandle)  // 0 = remove all

Runtime model & call modes

Where to go from here

• examples/ffiapi — minimal end-to-end library with C++ / Python / Rust / Go clients.
• examples/torpedo — a full game engine as a Nim library, four language frontends.
• examples/persistence — a persistence service across the boundary.
• Reference: doc/FFI_API.md · type support matrix: doc/TYPESUPPORT.md.

Take-away — declare once, use anywhere, tag the reach

Everything else is documented — go deep on demand: USAGEGUIDE.md · doc/COOKBOOK.md · doc/MT_BROKER_CONFIG.md · doc/FFI_API.md · DeepWiki · debug dumps: -d:brokerDebug (see USAGEGUIDE § Debug)

λ Thank you — questions?