Documentation
¶
Overview ¶
Package events provides concrete payload descriptors and ordered in-process dispatch. Durable enqueue is a separate transactional outbox boundary.
Index ¶
- func ListenerDefinition(policy jobs.Policy) jobs.Definition[queuedListener]
- func Module(name foundation.ProviderID, key foundation.Key[*Bus], config Config) foundation.Module
- func QueuedOutboxDefinition(policy jobs.Policy) jobs.Definition[queuedEvent]
- func RegisterListener[E any](r *foundation.Registrar, busKey foundation.Key[*Bus], topic Topic[E], ...) error
- func RegisterOutbox(r *foundation.Registrar, key foundation.Key[*Outbox], ...) error
- func RegisterQueuedListener[E any](r *foundation.Registrar, busKey foundation.Key[*Bus], topic Topic[E], ...) error
- func RegisterSubscriber(r *foundation.Registrar, busKey foundation.Key[*Bus], id string, ...) error
- func RegisterTopic[E any](r *foundation.Registrar, busKey foundation.Key[*Bus], topic Topic[E]) error
- type Bus
- type Config
- type Declaration
- type Handler
- type Interceptor
- type Listener
- type ListenerID
- type ListenerQueue
- type Name
- type Outbox
- type QueuedOutbox
- type Stored
- type Subscriber
- type Topic
- func (t Topic[E]) AfterCommit(ctx context.Context, tx *database.Tx, bus *Bus, input E) error
- func (t Topic[E]) Declare(listeners ...Listener[E]) (Declaration, error)
- func (t Topic[E]) Dispatch(ctx context.Context, bus *Bus, input E) error
- func (t Topic[E]) Enqueue(ctx context.Context, tx *database.Tx, producer *Outbox, input E) (outbox.ID[E], error)
- func (t Topic[E]) Find(ctx context.Context, executor database.Executor, producer *Outbox, ...) (value.Optional[Stored[E]], error)
- func (t Topic[E]) Name() Name
- func (t Topic[E]) OutboxID(ctx context.Context) (outbox.ID[E], bool)
- func (t Topic[E]) Validate() error
- func (t Topic[E]) Version() Version
- type Version
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func ListenerDefinition ¶
func ListenerDefinition(policy jobs.Policy) jobs.Definition[queuedListener]
ListenerDefinition is the job schema queued listeners use. Register it (via ListenerQueue.Declaration) with the dispatcher the queue binds to.
func Module ¶
func Module(name foundation.ProviderID, key foundation.Key[*Bus], config Config) foundation.Module
Module prepares one bus per application, binds all provider contributions before boot, and retains listener work until shutdown has actually completed.
func QueuedOutboxDefinition ¶
func QueuedOutboxDefinition(policy jobs.Policy) jobs.Definition[queuedEvent]
QueuedOutboxDefinition is the schema used by QueuedOutbox.Declaration. The private payload stays framework-owned; callers can pass this definition to jobs.RegisterJob using type inference, just like ListenerDefinition.
func RegisterListener ¶
func RegisterListener[E any](r *foundation.Registrar, busKey foundation.Key[*Bus], topic Topic[E], name ListenerID, construct func(foundation.Resolver) (Handler[E], error)) error
RegisterListener constructs a typed handler once through the application's existing service graph. Constructors are pure and may resolve their own bus; binding is a separate construction step, avoiding a bus/listener cycle.
func RegisterOutbox ¶
func RegisterOutbox(r *foundation.Registrar, key foundation.Key[*Outbox], busKey foundation.Key[*Bus], destination outbox.Destination) error
RegisterOutbox binds a typed producer through ordinary application assembly. A durable destination may be registered only once per application, even when different public service keys or buses are supplied. Duplicate detection and constructor lifetime remain owned by foundation's existing service graph.
func RegisterQueuedListener ¶
func RegisterQueuedListener[E any](r *foundation.Registrar, busKey foundation.Key[*Bus], topic Topic[E], name ListenerID, construct func(foundation.Resolver) (Handler[E], error)) error
RegisterQueuedListener is RegisterListener for a listener that runs as a job through the bus's ListenerQueue (see Listener.Queued).
func RegisterSubscriber ¶
func RegisterSubscriber(r *foundation.Registrar, busKey foundation.Key[*Bus], id string, construct func(foundation.Resolver) (Subscriber, error)) error
RegisterSubscriber constructs a subscriber once through the application's service graph and contributes all of its listeners to the bus. id names the subscriber uniquely within the bus.
func RegisterTopic ¶
func RegisterTopic[E any](r *foundation.Registrar, busKey foundation.Key[*Bus], topic Topic[E]) error
RegisterTopic declares an intentionally listener-free topic, such as a schema used only for durable enqueue. Listener registration also declares its topic; callers do not need both. Registration is explicit and local to the selected bus.
Types ¶
type Bus ¶
type Bus struct {
// contains filtered or unexported fields
}
Bus owns an immutable registry and synchronous dispatch lifetimes. Prepare is pure construction; Start connects its application lifetime. Close cancels work and waits for actual handler exit, even when a handler ignores cancellation. A Bus must not be copied. Its zero value is invalid; use Prepare or Module.
func (*Bus) Close ¶
Close starts shutdown even if the supplied wait context has expired. A timed out caller can wait again; callbacks remain owned until they actually return. Closing from this bus's active handler context fails instead of waiting on itself.
func (*Bus) Done ¶
func (b *Bus) Done() <-chan struct{}
Done closes only after admission has stopped and every owned callback exited.
func (*Bus) Intercept ¶
func (b *Bus) Intercept(token eventseam.Token, interceptor Interceptor) (func(), error)
Intercept installs interceptor on this bus until the returned restore runs. It is a framework test seam: token comes from the module-internal eventseam package, so only framework test helpers (testkit/events.NewFake) can install an interceptor; production code cannot silently suppress listeners.
type Config ¶
Config bounds concurrent captures/deliveries and the active dispatch chain. A top-level burst beyond MaxInFlight queues briefly (bounded by the caller's context and the shared admission wait) and then fails with fault.Overloaded. A nested dispatch from an active listener never waits for capacity held by its own ancestors; it fails immediately with fault.Overloaded instead.
func DefaultConfig ¶
func DefaultConfig() Config
type Declaration ¶
type Declaration struct {
// contains filtered or unexported fields
}
Declaration is the explicit heterogeneous registration boundary. Its schema and handlers are constructed through Topic[E], retaining concrete type checks. It contains no captured event payload and performs no I/O during construction.
func Subscribe ¶
func Subscribe(subscribers ...Subscriber) ([]Declaration, error)
Subscribe collects subscribers' declarations for Prepare.
type Handler ¶
Handler receives an independent concrete payload and the dispatch context. It must respect cancellation and return before its owning dispatch can finish.
type Interceptor ¶
Interceptor observes a dispatch before its listeners run; returning true suppresses them. testkit/events.NewFake installs one. It runs as an owned callback, must return promptly and must not retain the payload text.
type Listener ¶
type Listener[E any] struct { // contains filtered or unexported fields }
Listener pairs a semantic name with a concrete handler. A declaration takes an owned copy; later slice edits cannot alter a prepared listener registry.
func (Listener[E]) Queued ¶
Queued returns this listener marked to run as a job on the bus's ListenerQueue instead of inline. Sync and queued listeners can be mixed per event; each queued listener is enqueued at its position in declaration order with a fresh payload snapshot, and runs later with its own retries.
type ListenerID ¶
type ListenerID string
ListenerID identifies one ordered listener within an event name/version.
type ListenerQueue ¶
type ListenerQueue struct {
// contains filtered or unexported fields
}
ListenerQueue runs one bus's queued listeners through the ordinary job worker. Create it once per bus, register Declaration in the job registry, and Bind the dispatcher before dispatching events with queued listeners. A queued listener receives a freshly decoded payload with the job's attribution and retries under the job policy; earlier deliveries are not undone on failure.
func NewListenerQueue ¶
func NewListenerQueue(bus *Bus, policy jobs.Policy) (*ListenerQueue, error)
func (*ListenerQueue) Bind ¶
func (q *ListenerQueue) Bind(dispatcher *jobs.Dispatcher) error
Bind selects the dispatcher queued listeners are enqueued through. It can be set once, after the dispatcher's registry includes Declaration.
func (*ListenerQueue) Declaration ¶
func (q *ListenerQueue) Declaration() (jobs.Declaration, error)
Declaration is the job handler that delivers one queued listener.
type Name ¶
type Name string
Name is a stable semantic event name, shared by registration and transport.
type Outbox ¶
type Outbox struct {
// contains filtered or unexported fields
}
Outbox binds a stable durable destination to this application's typed event registry. Construct it once and inject it into domain code. It does not poll or dispatch persisted messages, and it never owns a second bus lifetime.
func PrepareOutbox ¶
func PrepareOutbox(destination outbox.Destination, bus *Bus) (*Outbox, error)
PrepareOutbox performs no I/O or schema migration. Its bus may be prepared during application construction; enqueue/reload require that bus to be running.
func (*Outbox) Destination ¶
func (o *Outbox) Destination() outbox.Destination
type QueuedOutbox ¶
type QueuedOutbox struct {
// contains filtered or unexported fields
}
QueuedOutbox delivers durable event rows through the ordinary job worker. One declaration routes explicit destinations to their prepared buses. Earlier listeners may run again after failure/crash; use Topic.OutboxID for idempotency. Buses and queue infrastructure remain owned by their application providers.
func NewQueuedOutbox ¶
func NewQueuedOutbox(policy jobs.Policy, producers ...*Outbox) (*QueuedOutbox, error)
func (*QueuedOutbox) Declaration ¶
func (q *QueuedOutbox) Declaration() (jobs.Declaration, error)
func (*QueuedOutbox) PublicationRoute ¶
func (q *QueuedOutbox) PublicationRoute(destination outbox.Destination, dispatcher *jobs.Dispatcher) (publisher.Route, error)
type Stored ¶
type Stored[E any] struct { // contains filtered or unexported fields }
Stored preserves a persisted event's immutable payload and attribution. It is not a delivery receipt. Payload returns a fresh concrete DTO on each call.
func (Stored[E]) Origin ¶
func (s Stored[E]) Origin() attribution.Origin
type Subscriber ¶
type Subscriber interface {
Declarations() ([]Declaration, error)
}
Subscriber groups related listeners in one type, typically constructed with shared dependencies. Declarations returns one Topic.Declare per event it handles; each may mix sync and queued listeners.
type Topic ¶
type Topic[E any] struct { // contains filtered or unexported fields }
Topic preserves the concrete payload type through registration and dispatch. Define it once and reuse the descriptor rather than repeating names in calls.
func (Topic[E]) AfterCommit ¶
AfterCommit snapshots payload and attribution now, then schedules process-local dispatch through the actual transaction. Rollback suppresses dispatch. Execution uses the outer commit callback context, not a retained per-write context. Listener failure is reported as committed AfterCommitFailed by the database. This has no crash durability; durable publication requires an outbox row.
func (Topic[E]) Declare ¶
func (t Topic[E]) Declare(listeners ...Listener[E]) (Declaration, error)
Declare contributes this schema and ordered listeners. A declaration with no listeners is valid and permits intentional no-op dispatch. Undeclared topics are rejected. Payload JSON validation occurs when a concrete payload is captured.
func (Topic[E]) Dispatch ¶
Dispatch captures a typed payload and runs registered listeners in order. Each listener receives a fresh decoded value. Failure stops later listeners; successful earlier side effects are not automatically reversed. Use the supplied transaction explicitly when listener work needs rollback semantics.
func (Topic[E]) Enqueue ¶
func (t Topic[E]) Enqueue(ctx context.Context, tx *database.Tx, producer *Outbox, input E) (outbox.ID[E], error)
Enqueue snapshots the concrete payload and origin and inserts their outbox row through the supplied business transaction. Returning an ID does not establish outer commit or delivery. Rollback removes the row; no listeners run here. Delivery belongs to the worker runtime and must tolerate at-least-once attempts.
func (Topic[E]) Find ¶
func (t Topic[E]) Find(ctx context.Context, executor database.Executor, producer *Outbox, id outbox.ID[E]) (value.Optional[Stored[E]], error)
Find reloads a payload-owned ID only within this producer and topic version. A record from another destination/schema is absent. The registered concrete JSON schema is checked again; malformed persisted data fails without dispatch.
Source Files