Documentation
¶
Overview ¶
Package bus is the agent's pub/sub messaging port: ambient, fire-and-forget signals that decouple producers from consumers. Proactive mode ("flynn watch"), cross-component notifications, and fleet/k8s event fan-out all publish to and subscribe on subjects rather than calling each other directly.
Like state, spine, and observe, messaging is a PORT with a zero-dependency in-process default (MemoryBus, Go channels) and heavy backends as opt-in adapters (NATS/JetStream for multi-process, fleet, and k8s fan-out). The subject grammar mirrors NATS so that adapter is a drop-in: dot-separated non-empty tokens, with `*` matching exactly one token and `>` (only as the final token) matching one or more trailing tokens.
The bus is deliberately distinct from its neighbours. The spine is the durable, ordered truth log; the jobs queue is durable, retried work. The bus is none of those: delivery is at-most-once, unordered across subscribers (but ordered within one), and not itself durable. When a signal must survive a restart it is recorded on the spine; the bus only moves signals between live components.
Index ¶
Constants ¶
const ( // TokenAny ("*") matches exactly one token at its position. TokenAny = "*" // TokenTail (">") matches one or more trailing tokens; valid only as the // final pattern token. TokenTail = ">" )
Wildcard tokens, mirroring NATS.
Variables ¶
var ( // ErrClosed is returned by Publish/Subscribe after the bus is closed. ErrClosed = fault.New(fault.Terminal, "bus_closed", "bus is closed") // ErrInvalidSubject is returned when a published subject is empty or contains // wildcards or malformed tokens. ErrInvalidSubject = fault.New(fault.Terminal, "bus_invalid_subject", "invalid subject") // ErrInvalidPattern is returned when a subscription pattern is malformed. ErrInvalidPattern = fault.New(fault.Terminal, "bus_invalid_pattern", "invalid subscription pattern") )
Sentinel errors. They are fault-classified so callers branch on the class (see fault.Classify) rather than on string matching.
Functions ¶
func Match ¶
Match reports whether subject matches pattern under the NATS-style grammar: tokens are separated by ".", "*" matches exactly one token, and ">" (only as the final token) matches one or more trailing tokens. It is total: any pair of strings yields a bool and never panics, so it is safe on untrusted input.
func ValidPattern ¶
ValidPattern reports whether s is a legal subscription pattern: one or more non-empty, whitespace-free tokens, where "*" may appear at any position and ">" only as the final token.
func ValidSubject ¶
ValidSubject reports whether s is a legal concrete subject to publish to: one or more non-empty, wildcard-free, whitespace-free tokens.
Types ¶
type Bus ¶
type Bus interface {
// Publish delivers m to all matching subscriptions. The subject must be a
// concrete subject (no wildcards); an invalid subject returns an error and
// delivers nothing.
Publish(ctx context.Context, m Message) error
// Subscribe registers h for every message whose subject matches pattern. The
// pattern may use the `*` and `>` wildcards; an invalid pattern returns an
// error.
Subscribe(ctx context.Context, pattern string, h Handler) (Subscription, error)
// Close stops the bus and all its subscriptions. Publish and Subscribe fail
// after Close.
Close() error
}
Bus is the pub/sub port. Publish delivers a message to every subscription whose pattern matches its subject; a message with no matching subscription is dropped. Implementations must be safe for concurrent use.
type Handler ¶
Handler processes a delivered message. It runs in the bus's delivery goroutine, not the publisher's, so it must not assume the publisher is still waiting. A returned error is observable (logged at the bus) but not retried: the bus is at-most-once, so durable retry belongs to the jobs queue, not here.
type MemoryBus ¶
type MemoryBus struct {
// contains filtered or unexported fields
}
MemoryBus is the zero-dependency, in-process Bus: pure Go channels, no broker, no network. It is the standalone default so the agent has working pub/sub with no setup, and the reference semantics the NATS adapter must match (held to bustest.RunSuite).
Delivery is asynchronous and ordered per subscription: each subscription owns a goroutine that runs its handler over a buffered mailbox in publish order. A handler that panics or errors cannot take down the bus or other subscribers; it is isolated and logged.
func (*MemoryBus) Close ¶
Close stops the bus: every subscription is unsubscribed and further Publish and Subscribe calls fail with ErrClosed.
func (*MemoryBus) Publish ¶
Publish delivers m to every matching subscription's mailbox, in the caller's goroutine up to the hand-off and then asynchronously in each subscription's goroutine. It blocks only while a matching mailbox is full, and unblocks early if ctx is cancelled or the subscription is removed.
type Message ¶
type Message struct {
// Subject is the concrete, wildcard-free address the message is published to.
Subject string
// Payload is the opaque message body.
Payload []byte
// Time is when the message was published; the bus stamps it from its clock
// when zero.
Time int64 // unix nanos
// Trace linkage carries the originating span/causation across the bus, so a
// signal published in one component stays correlated with the work it
// triggers. These mirror spine.Event and may be empty.
TraceID string
SpanID string
CausationID string
// OriginInstanceID is the instance that published the message, so fleet
// fan-out can attribute and de-loop it. May be empty for single-instance use.
OriginInstanceID string
}
Message is one published signal. Payload is opaque bytes so the wire format is the caller's choice and the NATS adapter passes it through unchanged.
type Option ¶
type Option func(*MemoryBus)
Option configures a MemoryBus.
func WithBuffer ¶
WithBuffer sets the per-subscription mailbox depth (default defaultBuffer). A larger buffer tolerates burstier publishers at the cost of more retained memory; values <= 0 fall back to the default.
func WithClock ¶
WithClock sets the time source used to stamp Message.Time when a publisher leaves it zero (default: clock.System).
func WithObservability ¶
func WithObservability(o *observe.Observability) Option
WithObservability sets the logger used to report handler errors and panics (default: observe.Default()).
type Subscription ¶
type Subscription interface {
// Subject returns the pattern the subscription was created with.
Subject() string
// Unsubscribe stops delivery to this subscription. It is safe to call more
// than once and from any goroutine.
Unsubscribe() error
}
Subscription is one active subscription. Unsubscribe stops delivery and is idempotent; after it returns no further handler calls begin.
Source Files
Directories