eventbus

Documentation

Overview

Package eventbus is the internal publish/subscribe spine that carries trigger.Event values from producers (native board, run completion, forge webhooks, schedule ticks, custom ingress) to consumers (the trigger Evaluator). It has two interchangeable implementations selected at wiring time — InProcBus for local single-host (CLI/studio) and NATSBus for cloud multi-tenant fan-out — so the same trigger.Evaluator consumes events identically in both modes.

The bus is a fan-out NOTIFICATION channel, deliberately separate from the run WORK queue (pkg/queue, iterion.queue.runs): events are at-least-once and lossy under back-pressure, runs are exactly-once and locked. They have different delivery semantics, so they get different transports.

Index

• type Bus
• type Handler
• type InProcBus
  • func NewInProcBus(logger *iterlog.Logger) *InProcBus
  • func (b *InProcBus) Drops(name string) int64
  • func (b *InProcBus) Publish(_ context.Context, ev trigger.Event) error
  • func (b *InProcBus) Subscribe(name string, filter trigger.Matcher, h Handler) (func(), error)

Types

type Bus

type Bus interface {
	Publish(ctx context.Context, ev trigger.Event) error
	// Subscribe delivers events matching filter to h. name identifies the
	// subscriber (used as the durable consumer name by NATSBus; informational
	// for InProcBus). An empty Matcher matches every event.
	Subscribe(name string, filter trigger.Matcher, h Handler) (cancel func(), err error)
}

Bus is the publish/subscribe contract. Publish never blocks on a slow subscriber (lossy fan-out); Subscribe registers a durable-named handler pre-filtered by a Matcher and returns a cancel func.

type Handler

type Handler func(ctx context.Context, ev trigger.Event) error

Handler processes one event. It runs on a per-subscriber worker goroutine, so it may do store I/O without stalling the publisher. A returned error is logged and otherwise ignored — the bus does not retry (the producer's own reconciliation path, e.g. the dispatcher poll, is the safety net).

type InProcBus

type InProcBus struct {
	// contains filtered or unexported fields
}

InProcBus is the local single-host Bus: an in-process fan-out with one buffered channel + worker goroutine per subscriber. It mirrors the watch_coordinator lifecycle (buffered chan → single worker → drop-on-full) and runview.EventBroker's lossy semantics. Zero external dependencies.

func NewInProcBus

func NewInProcBus(logger *iterlog.Logger) *InProcBus

NewInProcBus creates an empty in-process bus. logger may be nil.

func (*InProcBus) Drops

func (b *InProcBus) Drops(name string) int64

Drops reports how many events were dropped for the named subscriber because its buffer was full (test/observability helper). Returns -1 for an unknown name.

func (*InProcBus) Publish

func (b *InProcBus) Publish(_ context.Context, ev trigger.Event) error

Publish fans ev out to every subscriber whose filter matches. Non-blocking: a full subscriber buffer drops the event and bumps its drop counter.

func (*InProcBus) Subscribe

func (b *InProcBus) Subscribe(name string, filter trigger.Matcher, h Handler) (func(), error)

Subscribe registers h under name, pre-filtered by filter, and starts its worker goroutine. The returned cancel stops the worker and unregisters the subscriber (idempotent).