messaging

package module
v0.2.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 14, 2026 License: Apache-2.0 Imports: 16 Imported by: 0

README

go-messaging

The messaging infrastructure library of Go Elemental: one broker-agnostic contract for a service to emit events and run reactors on. Events are go-core's CloudEvents event type, emitted through a transactional outbox and consumed idempotently through an inbox, and every consumer is a go-core reactor on a subscription source.

Modules

Module Import path Tags
Base github.com/standards-lab/go-messaging v*
PostgreSQL engine github.com/standards-lab/go-messaging/postgres postgres/v*
NATS JetStream provider github.com/standards-lab/go-messaging/nats nats/v*

The base module depends on the standard library, go-core, and sqlate with its query package, the adjacent library the outbox's and inbox's engine contract builds on, and nothing else: no driver, no dialect, no broker client. depguard holds the base to that line in mise run check. An engine's driver and dialect enter only through its sub-module, and a broker's client only through its provider's.

Packages

  • messaging (the root) is the broker contract and the runtime: Broker and Publisher, Subscription and its Start, the delivery Attempt, the source's Report, Config, the Runtime with its Relay and Consume reactors, the error hook, Failure, Hook, and OnFailure, and the quiesce gate's Gate and Gated.
  • outbox writes a command's events in the command's own transaction and relays them to the broker, quarantining a row that can never be published.
  • inbox is a consumer's claim on an event, so a redelivery is handled once.
  • memory is the in-memory provider, for a service's tests and the conformance double.
  • messagingtest is the conformance suite, Run, every provider passes.
  • postgres (sub-module) is the PostgreSQL engine: the outbox's and inbox's statements, the messaging migration set, Verify, and Pending. Its pgtest package gives each test a throwaway database.
  • nats (sub-module) is the NATS JetStream provider: the Broker on one stream, its Config block with the stream's owner switch, and the durable pull consumers its sources bind. Its natstest package gives each test its own stream.

The package documentation, go doc github.com/standards-lab/go-messaging/..., is the reference for each exported name; each package's doc.go inventories its exports.

Usage

Wiring the runtime

A service builds one Runtime from its messaging Config, a broker, and its engine's statements. The base module imports neither: the composition root hands them over. Declare the engine's migration set ahead of the service's own, and call postgres.Verify from the verify stage.

cfg := messaging.Config{Source: "/billing"}
err := cfg.Finalize("BILLING") // defaults relay_poll and relay_timeout, reads BILLING_MESSAGING_*

set, err := postgres.Migrations() // the messaging set: messaging_outbox and messaging_inbox

rt, err := messaging.New(cfg, broker, postgres.Outbox(), postgres.Inbox(), shutdown, logger)

broker is any messaging.Broker: memory.New() in a test, and in production the NATS JetStream provider, below. shutdown is the process's drain timeout, within which every reactor the runtime builds drains. Error handling is elided throughout.

A domain emits through rt.Recorder, in its command's own transaction, so an event is written exactly when the state it reports commits:

_, err := sqlate.Transact(ctx, db, rt.Recorder.Emit(ctx, func(tx *sqlate.Tx, q *event.Queue) (struct{}, error) {
	// ... the command's writes on tx ...
	InvoiceIssued.Raise(q, invoiceID, InvoiceIssuedData{Amount: amount})
	return struct{}{}, nil
}))
The NATS broker

The nats sub-module's Broker publishes to one JetStream stream and subscribes durable pull consumers on it. Its Config is the service's nats block, finalized as the messaging block is:

owner := true // the one service that owns the stream; every other leaves Owner nil
ncfg := nats.Config{Stream: "billing", Prefix: "billing", Owner: &owner}
err := ncfg.Finalize("BILLING") // defaults url, reads BILLING_NATS_*, and validates

broker, err := nats.New(ncfg) // no I/O: Start connects, at the lowest stage

The broker publishes each event with its type as the subject, unchanged, so prefix is the namespace of the service's event types: with the prefix billing, it publishes billing.invoice.issued, and a consumer subscribes to that type. Publish refuses a type outside the prefix as permanent, so the relay quarantines its row, and Subscribe refuses a subscription naming one.

One service owns a stream: it sets owner true, and its Start creates the stream or updates it to its config, with the subjects <prefix>.>, the deduplication window duplicates, and the retention max_age. Every other service on the stream leaves owner unset or false: its broker binds the stream as it finds it and writes nothing, and its Start fails when the stream is missing or does not capture its prefix. Only the owner sets duplicates and max_age; a non-owner that sets either fails validation, naming the field. In JSON:

{"nats": {"url": "nats://nats:4222", "stream": "billing", "prefix": "billing", "owner": true, "max_age": "168h"}}

{"nats": {"url": "nats://nats:4222", "stream": "billing", "prefix": "billing"}}

The broker is a lifecycle node of the graph, started before every reactor on it; a call that reaches it before Start fails with nats.ErrNotStarted. broker.Conn() is the native connection, for request and reply in the composition root.

The relay and consumers as graph nodes

The runtime registers nothing. The composition root makes the broker and each reactor a node of its go-core graph, and the coordinator starts, drains, and watches them from their methods. In the sketch below, runtime and db are the service's nodes for the *messaging.Runtime and its *sqlate.DB.

rt.Relay(db) is the reactor that publishes the committed outbox rows. Every node that commits events uses the relay's node, so it drains before the relay, whose last pass publishes what was committed during the drain.

rt.Consume builds a consuming reactor. It decodes each event's data into the consumer's own type and hands its handler a Claim over the inbox, bound to the event and to the subscription's Name. The handler runs the claim first in its own transaction and commits nothing more when it is not the first:

relay := g.Define("relay", func(s *graph.Scope) (*reactor.Reactor[event.Event], error) {
	return s.Use(runtime).Relay(s.Use(db)), nil
})

ledger := g.Define("ledger", func(s *graph.Scope) (*reactor.Reactor[event.Event], error) {
	pool := s.Use(db) // a Use runs in the constructor, never in the handler
	sub := messaging.Subscription{Name: "ledger", Types: []string{"billing.invoice.issued"}, MaxDeliver: 5}
	return s.Use(runtime).Consume(sub, func(ctx context.Context, d InvoiceIssuedData, claim messaging.Claim) error {
		_, err := sqlate.Transact(ctx, pool, func(tx *sqlate.Tx) (struct{}, error) {
			if first, err := claim(ctx, tx); err != nil || !first {
				return struct{}{}, err
			}
			// ... the consumer's work on tx ...
			return struct{}{}, nil
		})
		return err
	})
})

The handler's return is the delivery's outcome: nil acknowledges it, an error redelivers it after RetryDelay, and an error wrapped with event.Permanent terminates it. Data that does not decode is refused permanently.

The error hook

The failures the relay and the consumers survive, which no caller sees as an error, go to the runtime's hook, each as a messaging.Failure whose Kind names what failed: relay (a pass ended by a transient publish failure or a database error), quarantine, source (an error a consumer's source survived), or exhausted. Without a hook, the runtime logs each failure at warn as messaging failure. The default logs every failure it is given: during an outage, about four a second from the relay at the default poll, and from each consumer member on the NATS broker one failed pull about every 1.1 seconds, so a service that wants failures throttled, counted, or alerted on sets its own:

rt, err := messaging.New(cfg, broker, postgres.Outbox(), postgres.Inbox(), shutdown, logger,
	messaging.OnFailure(func(ctx context.Context, f messaging.Failure) {
		failures.WithLabelValues(string(f.Kind)).Inc()
	}))

A hook must not block for long: the relay and the consumer wait on it.

Quarantine

An outbox row that can never be published, one whose event does not decode or whose publish failed with an event.Permanent error, is quarantined in its own transaction, so it does not hold the outbox back. The relay reports it as a quarantine failure carrying an *outbox.QuarantineError. The row stays in messaging_outbox with quarantined_at and quarantine_reason set; the relay never claims it again and postgres.Pending does not count it.

To inspect the quarantined rows:

SELECT seq, source, id, header, quarantined_at, quarantine_reason
FROM messaging_outbox
WHERE quarantined_at IS NOT NULL
ORDER BY seq;

To release a row once its cause is fixed, clear both columns together, which the table's check constraint requires; the relay claims it on its next pass:

UPDATE messaging_outbox
SET quarantined_at = NULL, quarantine_reason = NULL
WHERE seq = $1;
MaxDeliver exhaustion

A subscription's MaxDeliver bounds the deliveries of one event; 0 is unlimited. A handler reads which delivery it is on with messaging.Attempt(ctx). When the handler fails, with an error not marked permanent, on the delivery whose attempt is MaxDeliver, the broker gives the event up and the runtime reports an exhausted failure with the event's identity and the attempt.

Exhaustion is seen only from the handler's error. An event whose last delivery overruns AckWait, with the handler returning nil or returning late, is given up unreported. Size AckWait above the handler's worst case where an unreported loss matters.

The quiesce gate

messaging.Gated gives the runtime a gate its reactors hold shared, such as go-core's reactor.Gate, so an operation that holds it exclusively, a restore or a migration, runs with the messaging paused:

var gate reactor.Gate
rt, err := messaging.New(cfg, broker, postgres.Outbox(), postgres.Inbox(), shutdown, logger,
	messaging.Gated(&gate))

The relay holds the gate for each outbox row, from before the row's transaction begins until it commits or rolls back, so while the gate is held exclusively it opens no transaction. Each delivery a Consume reactor handles holds it for the handler's whole call. A wait the drain ends reports nothing: the relay claims no row, and the consumer hands its handler nothing and leaves the event for redelivery. A delivery's wait counts against its AckWait, so one that waits past it is redelivered and counts against MaxDeliver. Without Gated, nothing holds a gate.

Development

The repository is a Go workspace of three modules, ., postgres, and nats. Tasks run through mise:

mise run check                 # build, vet, format, fix, tidy, test, and lint every module, then sqlint; writes nothing
mise run integration:postgres  # the postgres module's integration suite against a fresh compose PostgreSQL
mise run integration:nats      # the nats module's integration suite against a fresh compose NATS
mise run integration           # both suites, each against its own fresh service
mise run currency              # report requirements, Go, tools, actions, and images behind their latest
mise run upgrade               # upgrade every module's go directive and requirements, and the tools, to their latest
mise run up                    # start every compose service and wait until each is healthy
mise run down                  # stop them, discarding their data

The check builds each module with the workspace off, so each resolves its own pins. compose.yml runs PostgreSQL on port 5435 and NATS with JetStream on port 4225; POSTGRES_PORT and NATS_PORT move them, and MESSAGING_DSN and MESSAGING_NATS_URL in mise.toml follow. Each integration task starts only its own service and stops it whatever the result. Both services keep their data in tmpfs; the postgres suite gives each test its own database, and the nats suite its own stream. CI runs the check on every pull request and push to main, and both integration suites on push to main and on manual dispatch.

Releases

Each module is versioned and released on its own tags: v* for the base module, postgres/v* for the engine, and nats/v* for the provider. A pushed tag creates the module's GitHub release from the matching section of its changelog: CHANGELOG.md, postgres/CHANGELOG.md, and nats/CHANGELOG.md.

License

Apache License 2.0.

Documentation

Overview

Package messaging is the standard tier's broker operations: publishing an event and subscribing to events as a reactor source.

The package exports:

Brokers and subscriptions

A Broker is what a composition root builds reactors from. A Subscription names a durable consumer: every source subscribed under one Name shares that consumer's position and splits its work, so replicas of a service form a delivery group, and each distinct Name receives every event. The handler's return is the delivery's outcome, so a reactor acknowledges, redelivers, or terminates without a broker type in sight, and Attempt reads which delivery of the event the handler is on. A source passes the errors it survives to the Report given to Subscribe.

The runtime

A Runtime is one service's messaging: the broker, the outbox and inbox over its database engine's statements, and the recorder its domains emit through. Its Runtime.Relay publishes the outbox, and its Runtime.Consume builds a consuming reactor that decodes each event into the consumer's own type and binds its inbox claim. The failures both survive reach the hook OnFailure sets, each as a Failure, and are logged at warn without one. Under Gated, the relay holds a quiesce gate for each outbox row and a consumer for each delivery, so an operation that holds the gate exclusively runs with both paused. Config is the service's messaging configuration block, beside its provider's. The runtime registers nothing: the composition root makes the broker and each reactor a node of its graph, and the coordinator infers from their methods how to start, drain, and watch them.

Providers and engines

The package imports no broker and no engine: the composition root hands them to New. A provider, such as the memory package, implements Broker, and the messagingtest package holds the conformance suite every provider passes. The outbox and inbox packages run their tables on any engine that supplies the statements; the postgres sub-module is that engine for PostgreSQL, and the nats sub-module is the NATS JetStream provider.

The base module depends on the standard library, go-core, and sqlate and its query package, the adjacent library the outbox's and inbox's engine contract builds on.

Index

Constants

View Source 
const (
	// DefaultAckWait is the AckWait of a subscription that sets none.
	DefaultAckWait = 30 * time.Second
	// DefaultDuplicates is a broker's deduplication window when it sets
	// none: JetStream's own default.
	DefaultDuplicates = 2 * time.Minute
)
View Source 
const (
	// DefaultRelayPoll is how long the relay waits between passes once the
	// outbox is empty, when a configuration sets none.
	DefaultRelayPoll = 250 * time.Millisecond
	// DefaultRelayTimeout bounds the relay's publish of one event, when a
	// configuration sets none.
	DefaultRelayTimeout = 10 * time.Second
)

Variables

This section is empty.

Functions

func Attempt 

func Attempt(ctx context.Context) int

Attempt returns the attempt of the delivery whose handler context is ctx: 1 on the first delivery of an event, and one more on each redelivery, so a handler on attempt MaxDeliver is the event's last. A context no provider gave has no attempt, and Attempt returns 0.

func CheckName 

func CheckName(s string) error

CheckName reports whether s can name a durable consumer or a stream: it is non-empty, with no whitespace, control character, '.', '*', '>', '/', or '\', the characters a JetStream name may not hold.

func CheckType 

func CheckType(t string) error

CheckType reports whether t is an event type a broker can route: one or more tokens separated by '.', none of them empty, with no whitespace, control character, '*', or '>'. It is event.CheckType, the rule a domain's event kinds are declared under, so a type a domain can define is a type a broker routes.

func Encode 

func Encode(e event.Event) (event.Header, []byte, error)

Encode is the check and encoding every provider's Publish makes: e must pass event.Event.Validate and its type CheckType, and it is returned in binary content mode, as event.Encode gives it.

func WithAttempt 

func WithAttempt(ctx context.Context, attempt int) context.Context

WithAttempt returns ctx carrying attempt, the number of the delivery its handler is given: 1 for the first. A provider calls it on each delivery's handler context.

Types

type Broker 

type Broker interface {
	Publisher
	// Subscribe validates sub and returns a source of its events: a member
	// of the durable consumer sub.Name names.
	//
	// The first subscription under a Name creates its consumer, which starts
	// where sub.Start says: under [StartAll], at the beginning of the stream,
	// so it receives every event, including those published before it
	// subscribed; under [StartNew], at its own creation. A later subscription
	// under the same Name binds that consumer and must match its
	// configuration; a mismatch fails Subscribe, or the source's Receive when
	// the provider binds there.
	//
	// A member claims one delivery at a time, and the next only once the
	// handler's outcome is settled, so members of one Name share the work
	// rather than one member taking a batch. The outcome is the handler's
	// return: nil acknowledges the delivery; an error redelivers it after
	// sub.RetryDelay, until sub.MaxDeliver deliveries have been made; an
	// error marked by [event.Permanent] terminates it, so it is never
	// delivered again. A handler error never ends the source.
	//
	// Each delivery's handler context carries its attempt, which [Attempt]
	// reads: 1 on the first delivery of an event, and one more on each
	// redelivery, whether the previous delivery's handler returned an error
	// or overran AckWait.
	//
	// The source passes report each error it survives, one that does not
	// end its Receive, such as a message it cannot decode, with the context
	// it was receiving under. A nil report discards them.
	Subscribe(sub Subscription, report Report) (reactor.Source[event.Event], error)
}

Broker publishes events and subscribes reactors to them.

type Claim 

type Claim = func(ctx context.Context, tx *sqlate.Tx) (first bool, err error)

Claim records, in a command's transaction, that the consumer is handling the event it was bound to, and reports whether this is the first time. A command runs it first and, on false, commits nothing more. It is an alias, so a domain's own claim type of the same shape accepts it.

type Config 

type Config struct {
	Source       string             `json:"source"`
	RelayPoll    libconfig.Duration `json:"relay_poll"`
	RelayTimeout libconfig.Duration `json:"relay_timeout"`
}

Config is a service's messaging block, the part of its messaging that names no broker: the provider's own block sits beside it. Source is the service's CloudEvents source, one value for every replica. RelayPoll is how long the relay waits between passes once the outbox is empty. RelayTimeout bounds the relay's publish of one event, and so how long a hung publish holds the event's outbox row; it is the relay's outbox.Timeout.

func (*Config) Finalize 

func (c *Config) Finalize(envPrefix string) error

Finalize defaults RelayPoll to DefaultRelayPoll and RelayTimeout to DefaultRelayTimeout, reads the block's environment overrides when a prefix is given (<PREFIX>_MESSAGING_SOURCE, …_RELAY_POLL, and …_RELAY_TIMEOUT), and validates the result.

func (*Config) Merge 

func (c *Config) Merge(src *Config)

Merge overlays src's set fields onto the receiver.

func (Config) Validate 

func (c Config) Validate() error

Validate reports every way c is unusable: the source must be set, and the poll and the timeout positive.

type Failure 

type Failure struct {
	// Kind is what failed.
	Kind FailureKind
	// Consumer is the subscription's Name, for a source or exhausted
	// failure; it is empty for the relay's.
	Consumer string
	// EventSource, EventID, and EventType identify the event, when it is
	// known: always for an exhausted failure, and for a relay or
	// quarantine failure whose row decoded. They are empty otherwise.
	EventSource string
	EventID     string
	EventType   string
	// Attempt is the delivery the handler failed on, for an exhausted
	// failure; it is 0 for the others, which no delivery attempt names.
	Attempt int
	// Err is the failure itself: the handler's error, the error the source
	// reported, or the relay's report, an [*outbox.QuarantineError], an
	// [*outbox.PublishError], a database error, or a gate's error.
	Err error
}

Failure is a messaging failure the runtime reports to its Hook: one that no caller sees as an error, because the relay or the consumer survives it.

type FailureKind 

type FailureKind string

FailureKind names what failed in a Failure. 

const (
	// FailureRelay is a relay pass that ended on an error: a transient
	// publish failure, after which the next pass retries the event, a
	// database error, or, under [Gated], the gate's failure other than its
	// wait ending.
	FailureRelay FailureKind = "relay"
	// FailureQuarantine is an outbox row the relay quarantined, committed
	// by the time it is reported: the row does not decode, or its publish
	// failed with an error marked [event.Permanent]. Its event is never
	// published.
	FailureQuarantine FailureKind = "quarantine"
	// FailureSource is an error a consumer's source survived, such as a
	// message it could not decode: the [Report] it was subscribed with.
	FailureSource FailureKind = "source"
	// FailureExhausted is an event whose handler failed, with an error not
	// marked [event.Permanent], on the subscription's last delivery, its
	// MaxDeliver: the broker gives the event up, and nothing redelivers it.
	FailureExhausted FailureKind = "exhausted"
)

type Gate added in v0.2.0 

type Gate interface {
	Shared(ctx context.Context) (release func(), err error)
}

Gate is a quiesce gate the runtime's reactors hold shared while they work, so an operation that holds it exclusively runs with them paused: go-core's reactor.Gate is one. Shared waits until the gate admits a shared holder and returns its release, or fails, holding nothing; a wait ctx ends returns ctx's error.

type Hook 

type Hook func(ctx context.Context, f Failure)

Hook receives each Failure the runtime reports, with the context it was reported under. It must not block for long: the relay and the consumer wait on it before they go on.

type Option 

type Option func(*Runtime)

Option configures a Runtime.

func Gated added in v0.2.0 

func Gated(g Gate) Option

Gated sets the gate the runtime's reactors hold shared. Without it, or with a nil gate, they hold none.

The relay holds g for each outbox row, from before the row's transaction begins until the transaction commits or rolls back, so a relay waiting on g opens no transaction and claims no row. Its wait ends with the relay's stop, and its last pass's with that pass's bound, either of which claims nothing and reports nothing. Any other failure of g ends the pass and is reported as a FailureRelay, leaving the relay ready, since it is not the database's.

Each delivery a Runtime.Consume reactor handles holds g for the handler's whole call, and the handler is not called until it does. A wait that the drain ends hands the handler nothing and leaves the event unacknowledged, for the broker to redeliver. The wait is bounded by the delivery's AckWait like the handler is: a delivery that waits on g past AckWait is redelivered, and counts against MaxDeliver, so an event whose last delivery waits that long is given up unreported. Any other failure of g is the delivery's outcome, as a transient error of the handler's would be: the event is redelivered, and on its last delivery reported as a FailureExhausted.

func OnFailure 

func OnFailure(h Hook) Option

OnFailure sets the hook the runtime reports each Failure to. Without it, or with a nil hook, the runtime logs each failure at warn on its logger. The default logs every failure it is given, so during an outage it logs each failed relay pass, about four a second at the default poll, and each failed pull of the nats provider's sources, one per consumer member about every 1.1 seconds, a pull's one-second bound plus its 100ms wait before the next; a service that wants failures throttled, counted, or alerted on sets a hook that does so.

type Publisher 

type Publisher interface {
	Publish(ctx context.Context, e event.Event) error
}

Publisher publishes an event. Publish returns once the broker holds the event, so a delivery can outlive the publisher.

Publish rejects an event that Encode rejects. The broker deduplicates on the event's source and id, which together identify a CloudEvents event: an event whose source and id the broker has seen within its deduplication window is accepted and dropped, so a publisher that retries, such as the outbox's relay, delivers the event once, while another source's event that reuses the id still delivers.

type Report 

type Report func(ctx context.Context, err error)

Report receives an error a source survives: one it handled itself and kept receiving after, which would otherwise go unseen. It must not block for long, since the source waits on it before its next delivery.

type Runtime 

type Runtime struct {
	Recorder *event.Recorder[*sqlate.Tx]
	// contains filtered or unexported fields
}

Runtime is one service's messaging, built from its Config, the broker its provider supplies, and its database engine's outbox and inbox statements. The broker carries events to and from the service. The outbox holds the events a command emits until the relay publishes them, and the inbox records the events a consumer has handled. A domain takes Recorder, its one value for emitting events, and records each event in its command's transaction.

The failures the relay and the consumers survive, which no caller sees as an error, go to the runtime's Hook, set with OnFailure: each as a Failure naming what failed.

A Runtime registers nothing and knows nothing of the coordinator: the composition root makes the broker and each reactor the runtime builds a node of its graph, which the coordinator starts, drains, and watches.

func New 

func New(cfg Config, broker Broker, outboxEngine outbox.Engine, inboxEngine inbox.Engine, shutdown time.Duration, logger *slog.Logger, opts ...Option) (*Runtime, error)

New builds a service's messaging from cfg, over broker and the engine's statements, logging to logger. shutdown is the process's drain timeout, which every reactor the runtime builds drains within. New checks cfg, which is finalized, and shutdown, which must be positive. It does no I/O.

func (*Runtime) Consume 

func (r *Runtime) Consume[T any](sub Subscription, fn func(ctx context.Context, data T, claim Claim) error) (*reactor.Reactor[event.Event], error)

Consume returns the reactor that consumes sub's events, with the bound reactor.CancelWithin gives the drain timeout. The reactor decodes each event's data into T, the consumer's own type for the payload, and hands it to fn with a Claim over the inbox under sub.Name, bound to the event, so fn's domain never sees the event and a redelivery changes nothing. Data that does not decode is refused with event.Permanent, because no redelivery can fix it.

The reactor logs each delivery with the consumer, the event's type, id, and subject, and its outcome, so the service's log shows the traffic it receives. The outcome is "event consumed" at info when handled; repeat, when the claim found the event handled already; or retried, with the error, when the broker will redeliver it. A permanent refusal, from the decode or from fn, is "event refused" at warn with the error, because the broker terminates the delivery and nothing else reports it.

The reactor reports to the hook each error the source survives, as a FailureSource, and each event fn fails, with an error not marked permanent, on the delivery whose Attempt is sub.MaxDeliver, as a FailureExhausted: the broker gives that event up. An unlimited subscription, a MaxDeliver of 0, exhausts nothing. Exhaustion is seen only from fn's error, so an event whose last delivery overruns AckWait, with fn returning nil or late, is given up unreported.

Under Gated, each delivery holds the gate shared for fn's whole call, and fn is not called until it does. A wait that the drain or the delivery's deadline ends hands fn nothing, logs and reports nothing, and leaves the event to be redelivered.

func (*Runtime) Relay 

func (r *Runtime) Relay(db sqlate.Beginner) *reactor.Reactor[event.Event]

Relay returns the reactor that publishes the outbox's committed events on db to the broker, with the bound reactor.CancelWithin gives the drain timeout. Each publish is bounded by the configuration's RelayTimeout. The relay makes a last pass, bounded at a quarter of the drain timeout and so below the reactor's bound, that publishes what the producers committed while they drained. The root therefore makes every node that commits events use the relay's node, so they drain before it. The relay logs each event it publishes at info with its type, id, and subject, so the service's log shows the traffic it sends.

The relay reports each row it quarantines to the hook as a FailureQuarantine, and each pass a publish failure or a database error ends as a FailureRelay, with the event's identity when the relay knew it.

Under Gated, the relay holds the gate shared for each row, from before the row's transaction begins until it settles.

type Start 

type Start int

Start is where a new durable consumer begins in the stream. 

const (
	// StartAll begins at the stream's beginning, so the consumer receives
	// every event the stream retains, including those published before it
	// was created.
	StartAll Start = iota
	// StartNew begins at the consumer's creation, so the consumer receives
	// only the events published after it. A provider may create the consumer
	// as late as the source's first Receive, so an event published between
	// Subscribe and that Receive may be skipped; a caller that must not miss
	// one uses StartAll.
	StartNew
)

type Subscription 

type Subscription struct {
	// Name is the durable consumer, and so its delivery group: sources
	// subscribed under the same Name share one position and split the work.
	// It is a token: no whitespace, '.', '*', '>', '/', or '\'.
	Name string
	// Types filters on the event's type; empty matches every type. Each
	// entry must pass [CheckType].
	Types []string
	// MaxDeliver bounds the deliveries of one event; 0 is unlimited.
	MaxDeliver int
	// AckWait is the handler's deadline for each delivery, and so the
	// deadline on its context. When it passes, the event is redelivered and
	// the late outcome is ignored. 0 is [DefaultAckWait].
	AckWait time.Duration
	// RetryDelay is how long an event whose handler returned an error waits
	// before it is redelivered; 0 redelivers it at once.
	RetryDelay time.Duration
	// Start is where the Name's consumer begins when this subscription
	// creates it; the zero value is [StartAll]. It is part of the consumer's
	// configuration, so a later subscription under the Name must match it,
	// even though it moves nothing once the consumer exists.
	Start Start
}

Subscription describes a durable consumer of events.

func (Subscription) Matches 

func (sub Subscription) Matches(t string) bool

Matches reports whether sub's type filter admits an event of type t.

func (Subscription) Normalize 

func (sub Subscription) Normalize() Subscription

Normalize returns sub in the form a provider configures its consumer from: AckWait defaulted to DefaultAckWait, and Types copied, sorted, and without repeats, nil when empty. Subscriptions that mean the same thing normalize equal, so they bind the same consumer, and the caller cannot change the filter afterward.

func (Subscription) Validate 

func (sub Subscription) Validate() error

Validate reports every way sub is unusable.

Source Files

View all Source files

Directories

Path Synopsis
Package inbox is a consumer's guard against handling a redelivered event twice, on any SQL engine that supplies its statement.
 Package inbox is a consumer's guard against handling a redelivered event twice, on any SQL engine that supplies its statement.
internal
engine
Package engine checks the statements an engine's module supplies to the outbox and the inbox, which hold no SQL of their own.
 Package engine checks the statements an engine's module supplies to the outbox and the inbox, which hold no SQL of their own.
Package memory is the in-memory messaging provider: the conformance double for the standard tier and a test double for a service.
 Package memory is the in-memory messaging provider: the conformance double for the standard tier and a test double for a service.
Package messagingtest is the conformance suite every messaging provider passes.
 Package messagingtest is the conformance suite every messaging provider passes.
nats module
Package outbox writes events through a transactional outbox and publishes them with a relay, on any SQL engine that supplies the statements.
 Package outbox writes events through a transactional outbox and publishes them with a relay, on any SQL engine that supplies the statements.
postgres module

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.