125 lines
5.0 KiB
Go
125 lines
5.0 KiB
Go
// Package bus provides an in-process publish/subscribe event bus. It is the
|
|
// asynchronous communication channel mandated by the Cyrene architecture:
|
|
// synchronous calls go through interfaces and dependency injection, while
|
|
// asynchronous events and agent-to-agent messages flow through this bus. The
|
|
// API is deliberately NATS-like so the bus can later be swapped for an
|
|
// out-of-process NATS deployment (or another transport) without changing call
|
|
// sites; see DevDocs/00-overview.md and DevDocs/02-development-constraints.md.
|
|
//
|
|
// # Matching
|
|
//
|
|
// Both subjects and subscription patterns are dot-separated token strings.
|
|
// Two wildcards are supported:
|
|
//
|
|
// - "*" matches exactly one non-empty token. For example "runtime.*.start"
|
|
// matches "runtime.loop.start" but not "runtime.loop".
|
|
// - ">" matches one or more trailing tokens and may only appear as the last
|
|
// token (NATS semantics). For example "runtime.>" matches
|
|
// "runtime.loop.start" but not "runtime".
|
|
//
|
|
// A pattern with no wildcard matches only an identical subject.
|
|
//
|
|
// # Delivery semantics
|
|
//
|
|
// Publish fans out asynchronously and never waits for a handler to finish.
|
|
// Each subscription is served by a single worker goroutine reading from a
|
|
// bounded per-subscription channel, which serializes handler execution for
|
|
// that subscription and bounds its memory. When the buffer is full the event
|
|
// is dropped and counted (see busImpl.Dropped) instead of blocking the
|
|
// publisher; that trade-off is what keeps Publish non-blocking and is the
|
|
// documented cost of an unbounded producer.
|
|
//
|
|
// Handler errors are ignored by this M0 core; consumers that need them should
|
|
// capture and log them at the call site. A panic raised by a handler is
|
|
// recovered so a misbehaving subscriber cannot take down the bus or starve
|
|
// other subscribers.
|
|
package bus
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
)
|
|
|
|
// Event is a single published message.
|
|
type Event struct {
|
|
// Subject is the token string the message was published to.
|
|
Subject string
|
|
// Payload is the arbitrary value associated with the event.
|
|
Payload any
|
|
}
|
|
|
|
// Handler processes a delivered Event. It runs on the subscription's own
|
|
// worker goroutine, so invocations for the same subscription are serialized.
|
|
// A non-nil error is ignored by the M0 core (see the package comment).
|
|
type Handler func(ctx context.Context, e Event) error
|
|
|
|
// Bus is the in-process publish/subscribe contract. Keep this interface
|
|
// narrow so an alternate transport (for example NATS) can satisfy it without
|
|
// changing consumers, and so it is trivial to mock in tests.
|
|
type Bus interface {
|
|
// Publish asynchronously fans an event out to every subscription whose
|
|
// pattern matches subject. It never blocks waiting for a handler to run;
|
|
// when a subscription's buffer is full the event is dropped for that
|
|
// subscription and counted. It returns ErrClosed if the bus is closed and
|
|
// a wrapped error if subject is invalid or ctx is canceled.
|
|
Publish(ctx context.Context, subject string, payload any) error
|
|
// Subscribe registers h for events whose subject matches pattern and
|
|
// returns an unsubscribe function that removes the registration. When ctx
|
|
// is canceled the registration is removed automatically. It returns
|
|
// ErrClosed if the bus is already closed.
|
|
Subscribe(ctx context.Context, pattern string, h Handler) (unsubscribe func(), err error)
|
|
// Close is idempotent and safe to call concurrently. It prevents further
|
|
// Publish/Subscribe calls, unsubscribes every registration, cancels the
|
|
// handlers' contexts, and stops every worker. It returns nil.
|
|
Close() error
|
|
}
|
|
|
|
// Sentinel errors returned by the bus. Use errors.Is to test for them.
|
|
var (
|
|
// ErrClosed is returned when publishing to or subscribing on a closed bus.
|
|
ErrClosed = errors.New("bus: closed")
|
|
// ErrInvalidArgument is returned when a subject is empty, a pattern is
|
|
// malformed, or a handler is nil.
|
|
ErrInvalidArgument = errors.New("bus: invalid argument")
|
|
)
|
|
|
|
// defaultQueueSize is the per-subscription event buffer used when no option
|
|
// overrides it.
|
|
const defaultQueueSize = 64
|
|
|
|
// config carries the resolved construction options.
|
|
type config struct {
|
|
queueSize int
|
|
}
|
|
|
|
// Option customizes a bus built by New. Options are applied in order before
|
|
// the bus starts accepting traffic.
|
|
type Option func(*config)
|
|
|
|
// WithQueueSize sets the per-subscription event buffer size. Buffers bound how
|
|
// many undelivered events a subscription may hold before events begin to be
|
|
// dropped. Non-positive values are ignored and the default is kept.
|
|
func WithQueueSize(n int) Option {
|
|
return func(c *config) {
|
|
if n > 0 {
|
|
c.queueSize = n
|
|
}
|
|
}
|
|
}
|
|
|
|
// New builds a bus with the supplied options. The returned *busImpl
|
|
// implements Bus and also exposes busImpl.Dropped for observability.
|
|
func New(opts ...Option) *busImpl {
|
|
cfg := config{queueSize: defaultQueueSize}
|
|
for _, opt := range opts {
|
|
opt(&cfg)
|
|
}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
return &busImpl{
|
|
queueSize: cfg.queueSize,
|
|
subs: make(map[uint64]*subscription),
|
|
ctx: ctx,
|
|
cancel: cancel,
|
|
}
|
|
}
|