Event Bus

The event bus processes queued pub/sub actions on one dispatcher goroutine and delivers matching events to subscriber channels.

The Go snippets are implementation and extension fragments. They assume an existing component context, logger, handlers, and application event types.

Event Structure

type Event struct {
    System string  // Component/module (e.g., "registry", "process")
    Kind   string  // Event type (e.g., "create", "update", "exit")
    Path   string  // Entity identifier
    Data   any     // Payload
    Aux    any     // In-process dispatcher context; not propagated to processes
}

Bus Architecture

flowchart LR
    subgraph Publishers
        P1[Component]
        P2[Component]
    end

    subgraph Bus
        Q[actionQueue]
        D[dispatcher goroutine]
        S[subscribers map]
    end

    subgraph Subscribers
        S1[chan Event]
        S2[chan Event]
    end

    P1 & P2 -->|enqueue| Q
    Q -->|signal| D
    D -->|match & deliver| S1 & S2
    D <-->|manage| S

The bus stores state in a simple structure:

type Bus struct {
    subscribers       map[SubscriberID]sub
    subscriberCounter uint64
    maxSubscribers    int

    actionQueue []action
    spareQueue  []action
    actionMu    sync.Mutex
    actionReady chan struct{}  // buffered=1

    closed atomic.Bool
}

All mutations go through the dispatcher goroutine, eliminating race conditions without complex locking.

Actions

Four action types flow through the queue:

Subscribe and Unsubscribe block until the dispatcher confirms. Send is fire-and-forget. The bus accepts at most DefaultMaxSubscribers subscriptions (4096 by default); subscriptions beyond the cap fail with ErrSubscribersCapReached.

Subscribe is rejected with ErrSubscribersCapReached once the bus holds DefaultMaxSubscribers (4096) active subscriptions.

Subscribe fails immediately when the subscription context is already canceled, and again at the dispatcher if it is canceled before the ownership decision is made — the bus never takes a channel it did not install.

Unsubscribe is an ownership barrier, not a best-effort hint. It returns only after the dispatcher acknowledges, so the caller can release the channel knowing the bus holds no in-flight send reference. When it arrives after Stop, the acknowledgement waits for the dispatcher to finish delivering the batch it already drained.

Stop is likewise terminal: a second concurrent Stop does not return early on the already-closed flag but waits for the dispatcher to drain and exit.

Queue Swapping

The dispatcher uses slice swapping to avoid allocations in steady state:

func (b *Bus) processActions() bool {
    b.actionMu.Lock()
    actions := b.actionQueue
    b.actionQueue = b.spareQueue[:0]
    b.spareQueue = nil
    b.actionMu.Unlock()

    for i := range actions {
        // process action
    }

    clear(actions)
    b.actionMu.Lock()
    b.spareQueue = actions[:0]
    b.actionMu.Unlock()
    return true
}

Two slices alternate: one for processing, one for new arrivals. The actionReady channel is buffered to 1, so signaling never blocks and multiple enqueues coalesce into one wakeup.

Pattern Matching

Subscriptions compile patterns once at subscribe time:

type sub struct {
    subID   SubscriberID
    ctx     context.Context
    system  *wildcard.Wildcard
    kind    *wildcard.Wildcard
    eventCh chan<- Event
}

The wildcard package supports four pattern types:

Patterns split on . so registry.* matches registry.create but not registry.entry.create. The pattern registry.** matches all three of registry, registry.create, and registry.entry.create.

Event Delivery

During Send processing, the dispatcher iterates subscribers:

for id, s := range b.subscribers {
    if s.system != nil && !s.system.Match(a.event.System) {
        continue
    }
    if s.kind != nil && !s.kind.Match(a.event.Kind) {
        continue
    }

    select {
    case <-a.ctx.Done():
        goto cleanup
    case <-s.ctx.Done():
        expiredSubs = append(expiredSubs, id)
    case s.eventCh <- a.event:
    }
}

If a subscriber's context is canceled, it's marked for removal during that delivery pass. The event context can also cancel delivery mid-iteration.

Lua Process Bridge

The events dispatcher bridges Go events to Lua processes. It subscribes once to all events ("**") and routes internally based on process subscriptions:

type Dispatcher struct {
    bus    event.Bus
    node   relay.Node
    subID  SubscriberID
    eventC chan event.Event

    mu   sync.RWMutex
    subs map[string]*subscription  // topic -> subscription
}

When a Lua process subscribes via events.subscribe(), the dispatcher stores the pattern and target PID. Matching events are packaged and sent via relay:

func (d *Dispatcher) routeEvent(evt event.Event) {
    d.mu.RLock()
    defer d.mu.RUnlock()

    for _, sub := range d.subs {
        if !matchPattern(sub.system, evt.System) {
            continue
        }
        if sub.kind != "" && sub.kind != "*" && !matchPattern(sub.kind, evt.Kind) {
            continue
        }

        data := map[string]any{
            "system": evt.System,
            "kind":   evt.Kind,
            "path":   evt.Path,
        }
        if evt.Data != nil {
            data["data"] = evt.Data
        }

        pkg := relay.NewPackage(pid.PID{}, sub.pid, sub.topic, payload.New(data))
        d.node.Send(pkg)
    }
}

Helper Types

Subscriber

Wraps channel subscription with a callback:

handler, err := eventbus.NewSubscriber(ctx, bus, "registry", "entry.*",
    func(evt Event) {
        // handle
    })
if err != nil {
    return err
}
defer handler.Close()

Spawns two goroutines: one reads events and calls the handler, another waits for context cancellation to unsubscribe.

EventRouter

Manages multiple handlers with centralized lifecycle:

router, err := eventbus.StartRouter(ctx, bus,
    WithHandlers(handler1, handler2),
    WithLogger(log))
if err != nil {
    return err
}
defer router.Stop()

Each handler implements Pattern() and Handle(). The router creates a Subscriber for each and closes all on Stop.

AwaitService

Request-response over pub/sub. It keeps a single subscription per (system, kind) pair and routes events to waiters by Path:

svc := eventbus.NewAwaitService(bus)
if err := svc.Start(ctx); err != nil {
    return err
}
defer svc.Stop()

waiter, err := svc.Prepare(ctx, "test", "response.(accept|reject)", "test/path", 5*time.Second)
if err != nil {
    return err
}
defer waiter.Close()

bus.Send(ctx, triggeringEvent)

result := waiter.Wait()  // returns AwaitResult{Event, Accepted, Error}

Prepare registers the waiter before the triggering event is sent, avoiding the race where the response arrives before the wait is registered. Wait blocks until a matching Path event arrives or the timeout (default DefaultAwaitTimeout, 30s, when non-positive) expires. Accepted is true when the event kind is accept, *.accept, or *.accepted; otherwise the kind is treated as a rejection and any error in Data surfaces as Error. The convenience Await(ctx, system, kind, path, timeout) combines Prepare and Wait. The boot infrastructure registers an AwaitService on the context (event.GetAwaitService).

Shutdown

1. Stop() atomically sets closed flag and enqueues Stop action
2. Dispatcher clears subscriber map
3. Remaining queued actions are drained:
   • Subscribe requests get "bus is closed" error
   • Unsubscribe requests complete immediately
   • Send events are dropped
4. WaitGroup completes

See Also

• Registry - Primary event producer
• Command Dispatch - Process-to-handler routing