README
¶
event (package)
Import: github.com/kausys/azync/event
User guide: ../event.md · GoDoc: package docs via go doc / pkg.go.dev.
Role
CQRS event bus: insert-only ledger + fan-out delivery jobs (source=event) on a shared Core.
Source layout
| File / area | Responsibility |
|---|---|
event.go, open.go |
New / Open |
publisher.go, tx.go |
Publish, transactional publish |
worker.go, register.go |
Subscriber handlers |
manager.go |
Admin + Replay |
context.go, options.go |
Delivery metadata, knobs |
Driver surface
Requires Core + event ledger + delivery jobs. No DAG/WorkflowStore.
Public surface (summary)
New/Open→Publisher,Worker,ManagerRegister/RegisterFuncTxPublisher[T]Manager.Replay
Boundaries
- Deliveries rehydrate from ledger; Replay does not need the original publish payload in-process.
- Prefer
Worker.Ready()before first publish so new subscriptions are included in fan-out. - Does not import
queue/dag/workflow.
Tests
go test ./event/... · driver conformance via driver/drivertest.
Documentation
¶
Overview ¶
Package event is a durable CQRS event bus over an azync Core.
Publish atomically appends an event to the ledger and fans out one delivery job per subscriber registered for that event type at that moment (the matching snapshot is taken inside the backend's Publish, in one transaction; callers must not pre-select subscribers). Deliveries are ordinary jobs of the event source: a Worker leases, executes and settles them on the shared engine, one job kind per subscriber. Delivery is at-least-once and deliberately unordered, so handlers must deduplicate their effects by (event id, subscriber). Aggregate id and version travel in the envelope for consumers that need to reject stale work.
A subscriber is registered on the Worker with Register (an implementer of the Subscriber interface plus one or more typed On bindings) or the RegisterFunc shorthand. Handlers receive the decoded domain event directly; all delivery metadata — the subscriber name, the attempt, the ledger identifiers and the publish-time annotations — travels on the context and is read through the package accessors (Attempt, IsRetry, EventID, Metadata, ...). Start upserts each subscriber's durable subscriptions before the engine runs, so a subscription is born on the first Start: events published earlier created no deliveries for it, and Manager.Replay is the way to feed a new subscriber historical events (flagged Replay, without overwriting history).
The delivery error taxonomy is deliberately minimal: a plain handler error retries with the engine backoff, and Permanent dead-letters immediately — there is no RetryAfter or Reportable. A payload that fails to decode is treated as Permanent, since it will never decode on retry. A handler panic is recovered and settles as an ordinary failure, never crashing the worker process. Because the Worker registers one engine kind per subscriber, an event whose subscriber has no live registration is simply never leased; the "missing handler" case disappears by design.
Compose a Runtime over a shared Core with New, or standalone with Open; neither migrates automatically (call Runtime.Migrate first). The Manager exposes the admin surface: stats, retry, replay, retention and the event and delivery listings.
Index ¶
- Constants
- func AggregateID(ctx context.Context) string
- func AggregateType(ctx context.Context) string
- func Attempt(ctx context.Context) int
- func EventID(ctx context.Context) uuid.UUID
- func IsNotFound(err error) bool
- func IsReplay(ctx context.Context) bool
- func IsRetry(ctx context.Context) bool
- func MaxAttempts(ctx context.Context) int
- func Metadata(ctx context.Context) map[string]string
- func NewContext(parent context.Context, d Delivery) context.Context
- func OccurredAt(ctx context.Context) time.Time
- func Permanent(err error) error
- func RegisterFunc[T EventArgs](w *Worker, name string, fn func(ctx context.Context, evt T) error, ...) error
- func SubscriberName(ctx context.Context) string
- func Type(ctx context.Context) string
- func Version(ctx context.Context) int64
- type AttemptError
- type Binding
- type DeadFilter
- type Delivery
- type DeliveryFilter
- type DeliveryListPage
- type DeliveryView
- type EventArgs
- type EventFilter
- type EventListPage
- type EventView
- type JobState
- type Manager
- func (m *Manager) DeleteSubscription(ctx context.Context, name, eventType string) (int64, error)
- func (m *Manager) DeliveryAttempts(ctx context.Context, deliveryID uuid.UUID) ([]AttemptError, error)
- func (m *Manager) DrainSubscriber(ctx context.Context, name string) (int64, error)
- func (m *Manager) Get(ctx context.Context, id uuid.UUID) (*EventView, error)
- func (m *Manager) List(ctx context.Context, filter EventFilter, page, size int) (EventListPage, error)
- func (m *Manager) ListDeliveries(ctx context.Context, filter DeliveryFilter, page, size int) (DeliveryListPage, error)
- func (m *Manager) ListSubscribers(ctx context.Context, eventType string) ([]SubscriberView, error)
- func (m *Manager) OpsStats(ctx context.Context) (OpsStats, error)
- func (m *Manager) Replay(ctx context.Context, filter ReplayFilter) (ReplayReport, error)
- func (m *Manager) Retain(ctx context.Context, before time.Time, limit int) (int64, error)
- func (m *Manager) Retry(ctx context.Context, deliveryID uuid.UUID) error
- func (m *Manager) RetryDead(ctx context.Context, filter DeadFilter) (int64, error)
- func (m *Manager) Stats(ctx context.Context) (Stats, error)
- type OpsStats
- type Option
- func WithCompletedRetention(d time.Duration) Option
- func WithCoreOptions(opts ...azync.Option) Option
- func WithDeadRetention(d time.Duration) Option
- func WithDefaultConcurrency(n int) Option
- func WithDefaultMaxAttempts(n int) Option
- func WithFetchBatchSize(n int) Option
- func WithFetchCooldown(d time.Duration) Option
- func WithFetchPollInterval(d time.Duration) Option
- func WithHandlerTimeout(d time.Duration) Option
- func WithIdleBackoffMax(d time.Duration) Option
- func WithLeaseTTL(d time.Duration) Option
- func WithLedgerRetention(d time.Duration) Option
- func WithMaxConcurrency(n int) Option
- func WithMaxReaps(n int) Option
- func WithShutdownDrain(d time.Duration) Option
- func WithStatsRetention(d time.Duration) Option
- type PublishOption
- type Publisher
- type RegisterOption
- type ReplayFilter
- type ReplayReport
- type Runtime
- type Stats
- type Subscriber
- type SubscriberView
- type Subscription
- type TxPublisherClient
- type Worker
Constants ¶
const ( StatePending = driver.StatePending StateScheduled = driver.StateScheduled StateActive = driver.StateActive StateDead = driver.StateDead StatePaused = driver.StatePaused StateSucceeded = driver.StateSucceeded )
Delivery lifecycle states, re-exported from the driver contract.
Variables ¶
This section is empty.
Functions ¶
func AggregateID ¶
AggregateID is the source aggregate id, if any. Empty outside a delivery.
func AggregateType ¶
AggregateType is the source aggregate type, if any. Empty outside a delivery.
func Attempt ¶
Attempt is the 1-based delivery attempt; the first delivery is attempt 1. Zero outside a delivery.
func IsNotFound ¶
IsNotFound reports whether err is the driver's not-found / wrong-state error, returned by admin operations whose target delivery was absent or in an unexpected state.
func IsReplay ¶
IsReplay reports whether the delivery was created by Manager.Replay rather than the original publish fan-out. False outside a delivery.
func IsRetry ¶
IsRetry reports whether this is a re-delivery (Attempt > 1). False outside a delivery.
func MaxAttempts ¶
MaxAttempts is the resolved retry budget for the delivery. Zero outside a delivery.
func Metadata ¶
Metadata returns the string-valued annotations attached at publish time. Nil outside a delivery.
func NewContext ¶
NewContext returns a copy of parent carrying d, so a handler can be exercised in isolation in a test without a running worker: build a Delivery, attach it, and the accessors below read from it exactly as they do in production.
func OccurredAt ¶
OccurredAt is the domain time the event happened. Zero outside a delivery.
func Permanent ¶
Permanent marks a handler error as non-retryable: the delivery goes straight to the dead letter instead of consuming its remaining retry budget.
func RegisterFunc ¶
func RegisterFunc[T EventArgs](w *Worker, name string, fn func(ctx context.Context, evt T) error, opts ...RegisterOption) error
RegisterFunc registers a single-type subscriber under an explicit name: sugar over Worker.Register for the common one-event-per-consumer shape, with T inferred from fn. It fails on an empty name, a duplicate subscriber, or a call after Start. Like Register, it upserts the durable (name, T.EventType()) subscription in Start (see Worker.Register for the durability caveat).
func SubscriberName ¶
SubscriberName is the name of the subscriber processing the delivery. Empty outside a delivery.
Types ¶
type AttemptError ¶
type AttemptError = driver.AttemptError
AttemptError is one recorded failure in a delivery's retry history.
type Binding ¶
type Binding struct {
// contains filtered or unexported fields
}
Binding pairs one event type with the typed handler that consumes it. It is opaque: build one with On, which infers the event type and the payload decoding from the handler signature, and hand the results to Worker.Register.
func On ¶
On builds a typed Binding for the event type T identifies. T is inferred from fn, so the caller never writes it: On(func(ctx, OrderCreated) error) binds the "orders.created.v1" type its EventType reports. The handler receives the decoded domain event; delivery metadata travels on ctx (Attempt, IsRetry, EventID, ...).
type DeadFilter ¶
type DeadFilter struct {
Subscriber string
}
DeadFilter scopes a bulk dead-delivery retry to a single subscriber. An empty Subscriber targets every subscriber.
type Delivery ¶
type Delivery struct {
// ID is the ledger event id.
ID uuid.UUID
// Type is the event type.
Type string
// AggregateType and AggregateID identify the source aggregate, if any.
AggregateType string
AggregateID string
// Version is the aggregate version this event advanced to.
Version int64
// OccurredAt is the domain time the event happened.
OccurredAt time.Time
// Meta carries the string-valued annotations attached at publish time
// (including any app-specific tenant / trace keys).
Meta map[string]string
// Subscriber is the name of the consumer this delivery targets.
Subscriber string
// Attempt is the 1-based delivery attempt (first delivery is attempt 1).
Attempt int
// MaxAttempts is the resolved retry budget for this delivery.
MaxAttempts int
// Replay is true for deliveries created by Manager.Replay rather than the
// original publish fan-out.
Replay bool
}
Delivery is the cross-cutting metadata of one event delivery. Handlers receive the decoded domain event as their argument; everything about the delivery itself — which subscriber it targets, which attempt this is, the ledger identifiers and the publish-time annotations — travels on the context and is read through the package accessors (Attempt, IsRetry, EventID, Metadata, ...).
It deliberately carries no payload: the payload is already decoded into the handler's typed argument. Delivery is at-least-once and deliberately unordered, so handlers must deduplicate their effects by (EventID, Subscriber). AggregateID and Version are exposed for consumers that reject stale work.
type DeliveryFilter ¶
DeliveryFilter selects delivery jobs for the admin list. Zero fields are unbounded.
type DeliveryListPage ¶
type DeliveryListPage struct {
Items []DeliveryView
Page int
Size int
Total int64
}
DeliveryListPage is one page of deliveries for the admin list.
type DeliveryView ¶
type DeliveryView struct {
ID uuid.UUID
EventID uuid.UUID
Subscriber string
State JobState
Attempt int
MaxAttempts int
Replay bool
AvailableAt time.Time
LastError string
CreatedAt time.Time
FailedAt time.Time
CompletedAt time.Time
}
DeliveryView is the admin projection of one delivery job. It deliberately carries no event type: a listed delivery exposes only the job row, and the event type lives in the ledger, joined only at dequeue — look the event up by EventID when the type is needed.
type EventArgs ¶
type EventArgs interface {
EventType() string
}
EventArgs is a JSON-serializable event whose EventType is wire-stable (decoupled from the Go type path), e.g. "orders.created.v1".
type EventFilter ¶
type EventFilter = driver.EventFilter
EventFilter selects ledger events for the admin list.
type EventListPage ¶
EventListPage is one page of events for the admin list.
type EventView ¶
type EventView struct {
ID uuid.UUID
Type string
AggregateType string
AggregateID string
Version int64
OccurredAt time.Time
// DispatchedAt is zero when the event has no deliveries. Otherwise it equals
// OccurredAt — publish creates deliveries atomically, so "dispatched" means
// "has at least one delivery snapshot".
DispatchedAt time.Time
Meta map[string]string
Payload json.RawMessage
}
EventView is the admin projection of one ledger event.
type JobState ¶
JobState is the wire state of a delivery — the same values the driver persists for every job source.
type Manager ¶
type Manager struct {
// contains filtered or unexported fields
}
Manager is the event administration surface: inspection, retry, replay, retention and ops projections. Pure library — no auth, no HTTP; embed it behind your own ops endpoints. It operates the event delivery source only and deliberately has no pause or purge operations beyond DeleteSubscription and DrainSubscriber.
func (*Manager) DeleteSubscription ¶ added in v0.0.4
DeleteSubscription removes the (name, eventType) subscriber registration, or every registration of name when eventType is empty, and returns the number removed. Existing delivery jobs already fanned out are untouched — they keep pinning the ledger events they belong to until they either settle naturally or DrainSubscriber clears them. Deleting a subscription that Worker.Start would otherwise re-create on the next deploy (because a handler for it is still registered) only wins until that restart; remove the registration in code too.
func (*Manager) DeliveryAttempts ¶
func (m *Manager) DeliveryAttempts(ctx context.Context, deliveryID uuid.UUID) ([]AttemptError, error)
DeliveryAttempts returns one delivery's failure history, oldest attempt first.
func (*Manager) DrainSubscriber ¶ added in v0.0.4
DrainSubscriber deletes every pending, scheduled and paused delivery job for name and returns the count removed. This discards undelivered work — call it only after DeleteSubscription, when you have decided those deliveries will never be handled, typically to unblock Retain from a retired subscriber's abandoned backlog (Retain skips any event with a non-terminal delivery, so an undrained retired subscriber can pin its events in the ledger forever).
func (*Manager) List ¶
func (m *Manager) List(ctx context.Context, filter EventFilter, page, size int) (EventListPage, error)
List returns one page of events (page is 0-based, default size 50).
func (*Manager) ListDeliveries ¶
func (m *Manager) ListDeliveries(ctx context.Context, filter DeliveryFilter, page, size int) (DeliveryListPage, error)
ListDeliveries returns one page of deliveries (page is 0-based, default size 50).
The unified JobFilter has no EventID predicate, so a filter that scopes by EventID loads the full (subscriber, state) match set from the driver and filters, then paginates, in memory — the documented cost of scoping by event. Without an EventID, pagination and totals are delegated straight to the driver.
func (*Manager) ListSubscribers ¶
ListSubscribers returns the subscriber catalog, optionally filtered by event type.
func (*Manager) Replay ¶
func (m *Manager) Replay(ctx context.Context, filter ReplayFilter) (ReplayReport, error)
Replay re-fans-out ledger events matching the filter into fresh deliveries flagged Replay, without overwriting the original delivery history.
func (*Manager) Retain ¶
Retain deletes ledger events occurring before the cutoff whose deliveries have all reached a terminal state (succeeded or dead), cascading to those deliveries, and returns the number of events removed. Events with any in-flight delivery (pending, scheduled, active or paused) are skipped.
func (*Manager) RetryDead ¶
RetryDead re-enqueues every dead delivery matching the filter and returns the count. An empty Subscriber targets every subscriber.
func (*Manager) Stats ¶
Stats returns the ledger event count, the delivery depths summed across every subscriber, and the subscriber registration count.
The result is not an atomic snapshot: it is assembled from three separate reads (delivery depths, ops counts, and the ledger event total), so a publish or delivery that lands between them can leave the returned counters mutually inconsistent by a small margin. It is intended for dashboards and ops views, not for exact accounting.
type Option ¶
type Option func(*config) error
Option configures an event Runtime. Options compose; later options win.
func WithCompletedRetention ¶
WithCompletedRetention overrides how long succeeded deliveries are kept. A negative value is rejected; zero means retain forever.
func WithCoreOptions ¶
WithCoreOptions forwards options to the Core that Open builds internally (schema, logger, notify channel, shared defaults...). Valid only with Open; New rejects it because the Core is already constructed.
func WithDeadRetention ¶ added in v0.0.4
WithDeadRetention overrides how long dead (exhausted-retry) deliveries are kept. A negative value is rejected; zero (the default) means retain forever.
func WithDefaultConcurrency ¶
WithDefaultConcurrency overrides the per-subscriber concurrency (each subscriber is one engine fetch partition). Must be positive.
func WithDefaultMaxAttempts ¶
WithDefaultMaxAttempts overrides the retry budget applied to subscribers registered without their own MaxAttempts. Must be positive.
func WithFetchBatchSize ¶
WithFetchBatchSize overrides how many deliveries one dequeue leases. Must be positive.
func WithFetchCooldown ¶
WithFetchCooldown overrides the pause after a productive fetch. Must be positive.
func WithFetchPollInterval ¶
WithFetchPollInterval overrides the idle polling period. Must be positive.
func WithHandlerTimeout ¶
WithHandlerTimeout overrides the per-delivery wall clock applied to every handler (default 5m; analogous to the queue's job timeout). Must be positive.
func WithIdleBackoffMax ¶
WithIdleBackoffMax overrides the idle backoff cap of the fetch loops. Must be positive.
func WithLeaseTTL ¶
WithLeaseTTL overrides the shared lease duration for this runtime. Must be positive.
func WithLedgerRetention ¶ added in v0.0.4
WithLedgerRetention enables the ledger retention loop, sweeping ledger events older than d whose deliveries have all reached a terminal state (see driver.Store.Retain) roughly once an hour. A negative value is rejected; zero (the default) disables the loop, matching prior behavior — the ledger is retained forever unless an operator opts in.
func WithMaxConcurrency ¶
WithMaxConcurrency overrides the total concurrent-handler cap across every subscriber. Must be positive.
func WithMaxReaps ¶
WithMaxReaps overrides how many lease expirations a delivery survives before the reaper kills it. Must be positive.
func WithShutdownDrain ¶
WithShutdownDrain overrides how long Start waits for in-flight handlers on shutdown. Must be positive.
func WithStatsRetention ¶
WithStatsRetention overrides how long daily stat counters are kept. A negative value is rejected; zero means retain forever.
type PublishOption ¶
type PublishOption func(*publishOptions)
PublishOption customizes one Publish.
func WithAggregate ¶
func WithAggregate(aggregateType, id string) PublishOption
WithAggregate stamps the source aggregate type and id on the event.
func WithMeta ¶
func WithMeta(key, value string) PublishOption
WithMeta attaches one metadata entry (repeatable). Use meta for app-specific fields such as tenant id or trace identifiers — azync does not model those as first-class columns.
func WithVersion ¶
func WithVersion(version int64) PublishOption
WithVersion stamps the aggregate version this event advances to.
type Publisher ¶
type Publisher struct {
// contains filtered or unexported fields
}
Publisher appends events to the durable ledger and registers subscribers. Publish atomically snapshots the subscribers matching an event and fans out one delivery job per match.
func (*Publisher) Publish ¶
func (p *Publisher) Publish(ctx context.Context, args EventArgs, opts ...PublishOption) (uuid.UUID, error)
Publish appends an event to the ledger. The backend selects the matching subscriber snapshot and creates the initial deliveries in one transaction; it returns the new event's id. A producer span wraps the call (SpanKindProducer), and ctx's trace context (if any) travels with the event via Meta so every fanned-out delivery's consumer span becomes its child (see engine.ExtractTraceContext).
func (*Publisher) Register ¶
func (p *Publisher) Register(ctx context.Context, subscription Subscription) error
Register adds or updates a durable subscription. A subscription registered with MaxAttempts <= 0 inherits the runtime's default retry budget (floored at 1); registration is an upsert keyed by (Name, EventType). Worker.Register and RegisterFunc perform this upsert automatically in Start; call this directly only for administrative registration (migrations, or subscribers consumed by external processes).
type RegisterOption ¶
type RegisterOption func(*registerOptions)
RegisterOption customizes RegisterFunc.
func WithMaxAttempts ¶
func WithMaxAttempts(n int) RegisterOption
WithMaxAttempts pins the subscriber's retry budget, overriding the runtime DefaultMaxAttempts. A non-positive value is ignored (the default applies).
type ReplayFilter ¶
type ReplayFilter = driver.ReplayFilter
ReplayFilter selects ledger events to re-fan-out into fresh deliveries.
type ReplayReport ¶
type ReplayReport struct {
Created int64
}
ReplayReport summarizes a Replay: how many fresh deliveries were created.
type Runtime ¶
type Runtime struct {
// contains filtered or unexported fields
}
Runtime is the event bus over one azync Core: the Publisher, the Worker and the Manager, all operating the event delivery job source only. Publishing appends to the durable ledger and fans out one delivery job per matching subscriber; consuming runs the shared engine with one job kind per subscriber.
func New ¶
New composes an event runtime over a shared Core. Settings start from the Core's defaults and event options override them per runtime.
func Open ¶
Open builds a standalone event runtime that owns a private Core opened from dsn (pass Core options through WithCoreOptions). Close closes the owned Core. Open never migrates; call Migrate before using a fresh schema.
func (*Runtime) Close ¶
Close releases the runtime's resources: the private Core when the runtime was built with Open, nothing when it composes over a shared Core. When the runtime owns its Core, Close first waits (bounded by ctx) for a running Worker to finish draining, so the store is not closed out from under in-flight settlements; on timeout it logs a warning and closes anyway rather than hanging indefinitely.
func (*Runtime) Migrate ¶
Migrate brings the backend schema up to date (requires a driver.Migrator). Open and New never migrate automatically.
type Stats ¶
type Stats struct {
Events int64
Pending int64
Scheduled int64
Active int64
Paused int64
Succeeded int64
Dead int64
Subscribers int64
}
Stats summarizes the durable event and delivery ledger: the ledger event count, the instantaneous delivery depths summed across every subscriber, and the registration count.
type Subscriber ¶
type Subscriber interface {
SubscriberName() string
}
Subscriber identifies a durable event consumer by name. Implement it on the value you pass to Worker.Register; the type stays non-generic (Go interfaces cannot be generic) because the event types it consumes are supplied separately, as typed bindings built with On.
A subscriber may additionally implement interface{ MaxAttempts() int } to pin its retry budget; without it the runtime's DefaultMaxAttempts applies (floored at 1).
type SubscriberView ¶
type SubscriberView = driver.SubscriberView
SubscriberView is one subscriber registration projected for the ops surface.
type Subscription ¶
Subscription is a durable registration binding a named consumer to one event type with its own retry budget. A newly registered subscription receives future publishes only; use Manager.Replay for historical events. Registration is an upsert keyed by (Name, EventType).
Worker.Register and RegisterFunc upsert their subscriptions automatically in Start; construct a Subscription by hand only for the administrative Publisher.Register path (migrations, or subscribers consumed by external processes).
type TxPublisherClient ¶
type TxPublisherClient[TTx any] struct { // contains filtered or unexported fields }
TxPublisherClient publishes events inside the caller's own backend transaction, so the append and its delivery fan-out commit atomically with the caller's writes (outbox pattern). Build one with TxPublisher.
func TxPublisher ¶
func TxPublisher[TTx any](r *Runtime) (*TxPublisherClient[TTx], error)
TxPublisher builds the transactional publish client for the driver's transaction handle type TTx (e.g. pgx.Tx for the pg driver). It fails immediately when the runtime's driver does not support transactional publishes for that type.
func (*TxPublisherClient[TTx]) PublishTx ¶
func (c *TxPublisherClient[TTx]) PublishTx(ctx context.Context, tx TTx, args EventArgs, opts ...PublishOption) (uuid.UUID, error)
PublishTx performs Publish within tx, letting the caller atomically commit application writes and the event fan-out. It returns the new event's id.
type Worker ¶
type Worker struct {
// contains filtered or unexported fields
}
Worker leases and executes event deliveries on the shared engine. Subscribers register via Register or RegisterFunc before Start; Start upserts each subscription durably and builds one engine job kind per subscriber, so a subscriber name maps to exactly one fetch partition even when it consumes several event types (the adapter routes each delivery to the binding for its type). A subscriber without a live registration is simply never fetched — the "missing handler" case disappears by design.
func (*Worker) Ready ¶
func (w *Worker) Ready() <-chan struct{}
Ready closes after wakeup setup succeeds and the polling loops are running. Polling-only workers become ready immediately after Start.
func (*Worker) Register ¶
func (w *Worker) Register(s Subscriber, bindings ...Binding) error
Register binds a subscriber to one or more typed event handlers. Each Binding (built with On) contributes the event type it consumes; the subscriber becomes one engine kind that routes each delivery to the binding matching the event's type. It fails on an empty subscriber name, no bindings, two bindings for the same event type, a duplicate subscriber, or a call after Start.
The subscriber's retry budget is the value it reports through the optional interface{ MaxAttempts() int }; without it the runtime's DefaultMaxAttempts applies (floored at 1).
Durability caveat: the (name, event type) subscriptions are upserted in Start, so a subscription is born on the first Start — events published before it created no deliveries for this subscriber. Adding bindings across restarts adds subscriptions but never removes stale ones; deleting a subscription is a deliberate administrative act, not a side effect of dropping a binding.
func (*Worker) Start ¶
Start upserts every subscriber's durable subscriptions, registers one engine kind per subscriber, and runs the engine (fetch, execute, settle, maintenance) until ctx is cancelled. The durable upsert runs before the engine starts, so once Ready closes every subscription exists and future publishes fan out to it. On cancellation in-flight handlers drain for up to the shutdown drain budget. Start fails if called twice; a Start that failed during setup (before the engine ran) may be retried.
func (*Worker) Wait ¶ added in v0.0.4
Wait blocks until a Start call has returned, or ctx ends first, whichever comes first. If Start was never called, Wait returns immediately (there is nothing to wait for). Close uses Wait to avoid closing a shared store out from under an in-flight drain; callers coordinating their own shutdown (stop ctx, then Wait, then release other resources) should do the same.
Source Files
¶
Directories
¶
| Path | Synopsis |
|---|---|
|
Package eventtest provides an in-memory publisher seam for tests: a Recorder that satisfies the same Publish signature as the real event.Publisher, so application code under test can publish without a running runtime and the test can assert on what was published.
|
Package eventtest provides an in-memory publisher seam for tests: a Recorder that satisfies the same Publish signature as the real event.Publisher, so application code under test can publish without a running runtime and the test can assert on what was published. |