Documentation
¶
Overview ¶
Package eventrouter provides a Router that separates event subscription declaration from execution. Cells declare handlers via AddContractHandler; the Router owns goroutine lifecycle, setup-error detection, and graceful shutdown.
ref: ThreeDotsLabs/watermill message/router.go — AddContractHandler/Run/Close pattern. Adopted: declaration-then-run split, Running() readiness signal, WaitGroup goroutine tracking. Deviated: no publish-side in AddContractHandler (GoCell publishes via outbox.Writer, not Router); startup readiness uses an explicit Ready signal from the Subscriber interface rather than a timeout heuristic — aligned with Uber fx OnStart synchronous return semantics.
Run lifecycle (4 phases):
Phase 1: serially call Subscriber.Setup(sub) for each handler — blocking topology declaration. Phase 2: launch one Subscribe goroutine per handler concurrently. Phase 3: wait on Subscriber.Ready(sub) for every handler (explicit signal, no timeout). Phase 4: block until ctx cancel or a runtime subscription error.
Index ¶
- Constants
- func NewContractTracingSubscriber(inner outbox.Subscriber, tr wrapper.Tracer, opts ...TracingSubscriberOption) outbox.Subscriber
- type EventCollector
- type NopEventCollector
- func (NopEventCollector) DecSubscriptionActive(_ context.Context, _ string)
- func (NopEventCollector) IncSubscriptionActive(_ context.Context, _ string)
- func (NopEventCollector) ObserveReadyWait(_ context.Context, _ string, _ time.Duration)
- func (NopEventCollector) RecordRuntimeError(_ context.Context, _, _ string, _ RuntimeErrorReason)
- func (NopEventCollector) RecordSetupError(_ context.Context, _, _, _ string)
- type Option
- type Router
- func (r *Router) AddContractHandler(spec contractspec.ContractSpec, handler outbox.EntryHandler, ...) error
- func (r *Router) AddSubscriptionValidator(v cell.SubscriptionValidator)
- func (r *Router) Close(ctx context.Context) error
- func (r *Router) HandlerCount() int
- func (r *Router) Health() error
- func (r *Router) Run(ctx context.Context) error
- func (r *Router) Running() <-chan struct{}
- type RuntimeErrorReason
- type SubscriberWrapperFunc
- type TracingSubscriberOption
Constants ¶
const ( // SetupErrorReasonSetupError is emitted when Subscriber.Setup returns an // error (Phase 1 failure). SetupErrorReasonSetupError = "setup_error" // SetupErrorReasonReadyTimeout is emitted when one or more subscriptions // fail to signal Ready within the configured readyTimeout (Phase 3). SetupErrorReasonReadyTimeout = "ready_timeout" // SetupErrorReasonPanic is emitted when a subscription goroutine panics // inside runSubscribe (Phase 2 unrecoverable failure). SetupErrorReasonPanic = "panic" // SetupErrorReasonSubscribeFailure is emitted when a subscription // goroutine's SubscribeEntry returns an error before the router reaches // Running() — i.e. the failure surfaces during Phase 3 await-ready, so // it is a setup-phase fault, not a runtime fault. // // Wave 2 (PR #593 review fix-up): introduced to anchor the // phase-classified reason at the Phase 3 consumer point of setupErr, // removing the producer-side `select <-r.running` flicker that mixed // this failure with RuntimeErrorReasonRuntimeFault. SetupErrorReasonSubscribeFailure = "subscribe_failure" )
SetupErrorReason* are the closed-set values passed to EventCollector.RecordSetupError. The ADR for the metrics-gaugevec-funnel (docs/architecture/202605191500-adr-metrics-gaugevec-funnel.md) enumerates all three reasons; adding a new reason requires updating that ADR and every collector implementation.
const DefaultReadyTimeout = 30 * time.Second
DefaultReadyTimeout bounds Phase 3 of Run() so a subscriber that never signals Ready (broker reconnect storm, mis-configured topology) does not block bootstrap indefinitely. 30s aligns with Uber fx StartTimeout default. Set to a non-positive value via WithReadyTimeout to disable the bound.
Variables ¶
This section is empty.
Functions ¶
func NewContractTracingSubscriber ¶
func NewContractTracingSubscriber(inner outbox.Subscriber, tr wrapper.Tracer, opts ...TracingSubscriberOption) outbox.Subscriber
NewContractTracingSubscriber decorates an outbox.Subscriber so every contract-bound delivery attempt gets one span that ends after final broker settlement. It delegates lifecycle methods unchanged and wraps Subscribe handlers with wrapper.WrapSubscriber at subscription registration time.
Subscribe-time invalid Subscription metadata surfaces as an error from the returned Subscriber.Subscribe call (single-source validation via outbox.Subscription.Validate); previous behavior was a panic from the removed wrapper.MustWrapSubscriber helper.
Setup AND Subscribe both gate on Subscription.Validate so the lifecycle boundary is consistent: a malformed Subscription cannot pre-create topology only to fail later at Subscribe. Ready returns a chan and cannot signal validation errors; callers that drive lifecycle directly must call Setup first to surface validation failures (Watermill / Kratos pattern: registration-time validation owned by the decorator).
Types ¶
type EventCollector ¶
type EventCollector interface {
IncSubscriptionActive(ctx context.Context, cellID string)
DecSubscriptionActive(ctx context.Context, cellID string)
RecordSetupError(ctx context.Context, cellID, topic, reason string)
ObserveReadyWait(ctx context.Context, cellID string, d time.Duration)
// RecordRuntimeError records a runtime-phase fault for the given cell+topic.
// reason is a closed-set value drawn from RuntimeErrorReason* constants;
// the typed parameter prevents callers from drifting to free-form strings.
// Phase 4 owner — any failure detected after Running() closes is recorded
// here as runtime_fault. Phase 1–3 failures live on the setup metric via
// RecordSetupError (PR #593 review fix-up P2#3).
RecordRuntimeError(ctx context.Context, cellID, topic string, reason RuntimeErrorReason)
}
EventCollector receives EventRouter lifecycle observations. The concrete production impl is *runtime/observability/metrics.EventRouterCollector. Defined here as an interface so router_test.go can use a spy implementation without depending on Prometheus.
ref: ThreeDotsLabs/watermill router metrics middleware — subscription lifecycle counters and gauges matching router_messages_processed_total / router_handler_active.
type NopEventCollector ¶
type NopEventCollector struct{}
NopEventCollector is the default when no collector is wired. All methods are no-ops so production code paths are always safe to call without a nil check.
func (NopEventCollector) DecSubscriptionActive ¶
func (NopEventCollector) DecSubscriptionActive(_ context.Context, _ string)
DecSubscriptionActive is a no-op for NopEventCollector (default when no metrics backend is wired).
func (NopEventCollector) IncSubscriptionActive ¶
func (NopEventCollector) IncSubscriptionActive(_ context.Context, _ string)
IncSubscriptionActive is a no-op for NopEventCollector (default when no metrics backend is wired).
func (NopEventCollector) ObserveReadyWait ¶
ObserveReadyWait is a no-op for NopEventCollector (default when no metrics backend is wired).
func (NopEventCollector) RecordRuntimeError ¶
func (NopEventCollector) RecordRuntimeError(_ context.Context, _, _ string, _ RuntimeErrorReason)
RecordRuntimeError is a no-op for NopEventCollector (default when no metrics backend is wired).
func (NopEventCollector) RecordSetupError ¶
func (NopEventCollector) RecordSetupError(_ context.Context, _, _, _ string)
RecordSetupError is a no-op for NopEventCollector (default when no metrics backend is wired).
type Option ¶
type Option func(*Router)
Option configures a Router.
func WithEventRouterCollector ¶
func WithEventRouterCollector(c EventCollector) Option
WithEventRouterCollector wires an EventCollector into the Router so that subscription lifecycle events (setup errors, ready-wait durations, active counts) are recorded. Typed-nil inputs are silently ignored; the final fallback to NopEventCollector{} occurs at New() after all options are applied.
Follows the cumulative builder noop pattern (runtime-api.md §Option 范式分层): nil input ≠ "remove the collector", it means "no change this call".
func WithReadyTimeout ¶
WithReadyTimeout overrides the default ready-wait budget. A value <= 0 disables the timeout (waits indefinitely on Ready channels and ctx).
type Router ¶
type Router struct {
// contains filtered or unexported fields
}
Router manages event subscription lifecycle. It is populated from RegistrySnapshot.Subscriptions drained by bootstrap phase6, and provides Run/Close for the execution phase.
Run MUST be called at most once. Calling Run a second time returns an error.
The subscriber field is *outbox.SubscriberWithMiddleware so that runSubscribe can call SubscribeEntry (EntryHandler → business middleware chain → ConsumerBase.Wrap → Inner.Subscribe). The EntryToSubscriberHandler lift function has been deleted; SubscribeEntry is the only public entry point, which is the structural guarantee that ConsumerBase is the explicit conversion boundary between business (EntryHandler) and broker (SubscriberHandler) layers.
ref: ThreeDotsLabs/watermill message/router.go — router holds *Router.handlers, not a generic middleware list: the router owns the composition.
func New ¶
New creates a Router that will use the given SubscriberWithMiddleware for all subscriptions. The subscriber field is typed as *outbox.SubscriberWithMiddleware so that runSubscribe can call SubscribeEntry and get the full business pipeline: business middleware chain → ConsumerBase.Wrap → Inner.Subscribe.
clk is required; pass clock.Real() at the composition root or clockmock.New(...) in tests. Panics on nil or typed-nil clock.
func (*Router) AddContractHandler ¶
func (r *Router) AddContractHandler( spec contractspec.ContractSpec, handler outbox.EntryHandler, consumerGroup string, ownerCellID string, opts ...cell.SubscriptionOption, ) error
AddContractHandler registers a contract-first subscription intent. The Router stores the contract metadata on the Subscription; bootstrap decorates the concrete Subscriber with NewContractTracingSubscriber so delivery spans close after final broker settlement.
spec.Kind MUST be "event" and spec.Topic MUST be set. topic used for the Subscriber.Setup / Subscribe lifecycle is derived from spec.Topic — callers do not pass a separate topic string.
ownerCellID is the cell that owns this subscription — distinct from consumerGroup (which may include a role suffix like "accesscore-rbac-session-sync"). It is mandatory and populates Subscription.CellID directly; there is no fallback to consumerGroup because Registry.Subscribe forces cellID to flow positionally from cell metadata at codegen time (HARD contract).
Returns a non-nil error when handler is nil, consumerGroup is empty, ownerCellID is empty, or the spec is malformed; callers should propagate the error to the bootstrap phase6 subscription walker.
ref: ThreeDotsLabs/watermill router.AddHandler handlerName / NATS subscription metadata. ref: ADR docs/architecture/202605111000-adr-subscription-cellid-mandatory.md
func (*Router) AddSubscriptionValidator ¶
func (r *Router) AddSubscriptionValidator(v cell.SubscriptionValidator)
AddSubscriptionValidator registers a validator that is invoked for every subsequent AddContractHandler call. Validators run outside the Router lock so a slow or panicking validator does not hold the subscription store. A nil validator is silently ignored.
func (*Router) Close ¶
Close shuts down the Router in three phases:
StopIntake (optional): if the subscriber implements outbox.SubscriberIntakeStopper, StopIntake(ctx) is called first. This signals the broker to stop delivering new messages (basic.cancel) while in-flight handlers continue to run. Errors from StopIntake are logged as warnings and do not abort the shutdown sequence.
Cancel: the internal runCtx subtree is canceled, unblocking Subscribe goroutines that are waiting on context cancellation.
Drain: wg.Wait() blocks until all goroutines exit. If ctx expires before drain completes, Close returns ctx.Err().
ref: ThreeDotsLabs/watermill message/router.go — closingInProgressCh two-phase barrier. ref: uber-go/fx app.go — run ctx vs stop ctx separation.
func (*Router) HandlerCount ¶
HandlerCount returns the number of registered handlers.
func (*Router) Health ¶
Health reports whether the router is ready to serve subscriptions. It returns nil only after startup completes successfully and before a setup or runtime failure has been recorded. After graceful shutdown it returns a distinguishable "shutting down" error.
func (*Router) Run ¶
Run starts all registered subscriptions and blocks until ctx is canceled or an unrecoverable subscription error occurs.
Run MUST be called at most once; a second call returns errAlreadyRunning.
Lifecycle (4 phases):
Phase 1: serially call Subscriber.Setup(sub) — blocking topology declaration.
Any Setup error causes Run to return immediately without starting subscriptions.
Phase 2: launch one Subscribe goroutine per handler concurrently.
Phase 3: wait on Subscriber.Ready(sub) for every handler.
Running() is closed only after ALL Ready channels close.
Phase 4: block until ctx cancel or a runtime subscription error.
On context cancellation, Run waits for all goroutines to finish before returning.
func (*Router) Running ¶
func (r *Router) Running() <-chan struct{}
Running returns a channel that is closed when all subscriptions have successfully started consuming. Callers can use this to wait for the Router to be ready (e.g., in bootstrap).
Note: if Run returns a setup error, Running() is never closed. Callers should also monitor the error from Run.
type RuntimeErrorReason ¶
type RuntimeErrorReason string
RuntimeErrorReason is the type-safe closed-set for RecordRuntimeError's reason argument. Using a named type prevents callers from passing arbitrary strings and documents the full value set at the type level.
const ( // RuntimeErrorReasonRuntimeFault is emitted for any runtime fault // detected in Phase 4 (after Running() closes). RuntimeErrorReasonRuntimeFault RuntimeErrorReason = "runtime_fault" )
RuntimeErrorReason* are the closed-set values passed to EventCollector.RecordRuntimeError. Runtime errors occur after startup (Phase 4) and are semantically distinct from setup errors (Phase 1-3).
PR #593 review fix-up (P1#1 + P2#3): the reason set is intentionally single-valued. Phase 3 setup-phase failures (SubscribeEntry returning an error before Running(), or ready timeout) are owned by the setup metric only; the previous RuntimeErrorReasonSubscribeFailure / ReadyWaitTimeout reasons were a phase-boundary violation that produced double-counting and reason flicker. The Phase 4 consumer always records runtime_fault — the router is running, so the underlying cause is by definition a runtime fault.
type SubscriberWrapperFunc ¶
type SubscriberWrapperFunc func(sub outbox.Subscription, next outbox.SubscriberHandler) outbox.SubscriberHandler
SubscriberWrapperFunc is applied at Subscribe time after wrapper.WrapSubscriber. It allows composition-root layers to inject per-subscription instrumentation (e.g. settlement observers) at the SubscriberHandler layer without requiring eventrouter to import observability packages.
type TracingSubscriberOption ¶
type TracingSubscriberOption func(*contractTracingSubscriber)
TracingSubscriberOption configures a contractTracingSubscriber.
func WithSubscriberWrapper ¶
func WithSubscriberWrapper(fn SubscriberWrapperFunc) TracingSubscriberOption
WithSubscriberWrapper adds a SubscriberHandler-layer wrapper applied after wrapper.WrapSubscriber inside Subscribe. The wrapper receives the validated Subscription and the tracing-wrapped handler; it may append DeliveryOutcome.SettlementObservers or perform other SubscriberHandler-layer instrumentation.
Source Files