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 ¶
Constants ¶
const DefaultSubjectPrefix = "iterion.events"
DefaultSubjectPrefix is the NATS subject namespace the bus publishes under. It is deliberately distinct from the run WORK queue's stream (ITERION_RUNS): the notification bus and the work queue have different delivery semantics, so they never share a subject tree. Override via NATSOptions.SubjectPrefix.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
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 ¶
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 ¶
NewInProcBus creates an empty in-process bus. logger may be nil.
func (*InProcBus) Drops ¶
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.
type NATSBus ¶ added in v0.50.0
type NATSBus struct {
// contains filtered or unexported fields
}
NATSBus is the cloud multi-replica Bus: a Core NATS (not JetStream) subject fan-out. Every trigger.Event is published to <prefix>.<source> and delivered to subscribers via a QUEUE GROUP keyed on the subscriber name, so exactly one replica's evaluator handles each event — the multi-host equivalent of InProcBus's single-worker-per-subscriber semantics. Whichever replica holds the (identical, store-backed) subscription set reacts; the others don't double-launch.
Core NATS, not JetStream, is deliberate: this is the LOSSY notification bus (bus.go), at-most-once with the producer's reconciliation path (dispatcher poll) as the backstop — not the exactly-once locked work queue (pkg/queue, which is the JetStream one). A dropped notification is recovered by the poll, so persistence + acks would be cost without benefit here.
Filtering mirrors InProcBus exactly: a subscriber consumes the whole event subject tree (<prefix>.>) and applies trigger.Matcher in-process, so the bus needs no per-field subject encoding and a subscription's filter can be arbitrarily rich.
func NewNATSBus ¶ added in v0.50.0
func NewNATSBus(nc *nats.Conn, opts NATSOptions) (*NATSBus, error)
NewNATSBus builds a NATSBus over an established NATS connection. The caller owns the connection lifecycle (the bus never closes it); callers typically pass the same low-level *nats.Conn the work queue uses (natsq.Conn.NATS()), since the bus and the queue address disjoint subject trees on one link.
func (*NATSBus) Publish ¶ added in v0.50.0
Publish encodes ev as JSON and fires it to <prefix>.<source>. Core NATS publish is non-blocking and never waits on subscribers — the lossy-fan-out contract. A publish to a subject with no subscribers is a no-op, not an error.
func (*NATSBus) Subscribe ¶ added in v0.50.0
Subscribe registers h under a NATS queue group named `name`, listening on the whole event subject tree (<prefix>.>). The queue group makes NATS deliver each event to exactly one member across all replicas sharing the name — so N server pods with the same evaluator subscription process each event once, not N times. Events are decoded and passed through filter before h, matching InProcBus. The returned cancel unsubscribes (idempotent).
type NATSOptions ¶ added in v0.50.0
type NATSOptions struct {
// SubjectPrefix overrides DefaultSubjectPrefix. Trailing dots are trimmed.
SubjectPrefix string
// Logger receives dropped-event / decode-error warnings. May be nil.
Logger *iterlog.Logger
}
NATSOptions configures a NATSBus.
Source Files