eventbus

Documentation

Overview

Package eventbus provides a generic, synchronous in-memory event bus. It carries no persistence concern: libraries that need delivery guarantees, such as the outbox package, build on top of it.

A Bus is safe for concurrent use. No lock is held while a handler runs, so handlers may subscribe or publish reentrantly without deadlocking.

Index

• Variables
• func Consume[T any](ctx context.Context, broker Broker, id ListenerID, ...) error
• func SubscribeTo[T any](bus *Bus, id ListenerID, handle func(ctx context.Context, event T) error) error
• type Broker
• type Bus
  • func NewBus() *Bus
  • func (b *Bus) Deliver(ctx context.Context, id ListenerID, event any) error
  • func (b *Bus) ListenersFor(event any) []ListenerID
  • func (b *Bus) Publish(ctx context.Context, event any) error
  • func (b *Bus) Subscribe(id ListenerID, matches func(event any) bool, handle Handler) error
• type ConsumerConfiguration
  • func DefaultConsumerConfiguration() ConsumerConfiguration
• type ConsumerOption
  • func WithBroadcastDelivery() ConsumerOption
  • func WithMaximumAttempts(attempts int) ConsumerOption
  • func WithRetryDelay(delay func(attempt int) time.Duration) ConsumerOption
  • func WithUnorderedDelivery() ConsumerOption
  • func WithWorkers(workers int) ConsumerOption
• type ConsumerRegistration
• type DeadLetter
• type Handler
• type ListenerID
• type MemoryBroker
  • func NewMemoryBroker() *MemoryBroker
  • func (b *MemoryBroker) FindExhausted(_ context.Context, limit int) ([]DeadLetter, error)
  • func (b *MemoryBroker) Publish(_ context.Context, event any) error
  • func (b *MemoryBroker) Resubmit(_ context.Context, reference string) (bool, error)
  • func (b *MemoryBroker) Stop()
  • func (b *MemoryBroker) Subscribe(_ context.Context, registration ConsumerRegistration) error
• type Ordering

Variables

var (
	// ErrDuplicateListener reports a subscription under an identifier that
	// is already taken.
	ErrDuplicateListener = errors.New("listener identifier is already subscribed")

	// ErrUnknownListener reports a delivery towards an identifier with no
	// subscription behind it.
	ErrUnknownListener = errors.New("listener is not subscribed")

	// ErrListenerPanic reports a handler that panicked while processing an
	// event; the panic is recovered and surfaced through this error.
	ErrListenerPanic = errors.New("listener panicked")
)

var ErrBrokerStopped = errors.New("memory broker is stopped")

ErrBrokerStopped reports an operation against a memory broker whose Stop was already called.

Functions

func Consume

func Consume[T any](
	ctx context.Context,
	broker Broker,
	id ListenerID,
	handle func(ctx context.Context, event T) error,
	options ...ConsumerOption,
) error

Consume registers a consumer of every event of type T on the broker. It is the typed front door of the Broker contract: the untyped registration is assembled from the type parameter, with FIFO ordering and the default retry budget unless options say otherwise.

func SubscribeTo

func SubscribeTo[T any](bus *Bus, id ListenerID, handle func(ctx context.Context, event T) error) error

SubscribeTo registers a listener that receives every event of type T.

Types

type Broker

type Broker interface {
	// Publish hands the event to every matching consumer's queue.
	Publish(ctx context.Context, event any) error

	// Subscribe registers a consumer. Engines with durable registrations use
	// the context for their writes.
	Subscribe(ctx context.Context, registration ConsumerRegistration) error

	// FindExhausted returns the dead letters, oldest first, up to the limit.
	FindExhausted(ctx context.Context, limit int) ([]DeadLetter, error)

	// Resubmit gives the referenced dead letter a fresh attempt budget. It
	// reports false when the reference is unknown or was already resubmitted.
	Resubmit(ctx context.Context, reference string) (bool, error)
}

Broker delivers each published event to every matching consumer exactly once per consumer, retrying independently per consumer with a bounded attempt budget. Consumers process FIFO by default and may opt out for concurrency. The contract is engine-independent: the in-memory engine of this package keeps everything in the process, while a persistent engine (such as the sqlbus module) adds durability and cross-node consumption behind the same interface.

Publish reports only validation and persistence failures: handler outcomes are settled asynchronously through the per-consumer retry machinery and surface as dead letters once the attempt budget is consumed. In-process engines deliver exactly once per consumer for the lifetime of the process; persistent engines settle each delivery exactly once but may execute a handler again after a crash or an expired claim, so handlers must be idempotent when durability is in play.

type Bus

type Bus struct {
	// contains filtered or unexported fields
}

Bus delivers events synchronously to subscribed listeners. Listeners are identified by a unique ListenerID, so callers can address one listener individually or multicast an event to every matching listener.

func NewBus

func NewBus() *Bus

NewBus constructs an empty Bus.

func (*Bus) Deliver

func (b *Bus) Deliver(ctx context.Context, id ListenerID, event any) error

Deliver hands the event to the identified listener. A panicking handler is recovered and reported as an error wrapping ErrListenerPanic, so one misbehaving listener cannot take down the publishing caller.

func (*Bus) ListenersFor

func (b *Bus) ListenersFor(event any) []ListenerID

ListenersFor returns the identifiers of every listener subscribed to the event, in subscription order.

func (*Bus) Publish

func (b *Bus) Publish(ctx context.Context, event any) error

Publish multicasts the event to every matching listener, in subscription order. The returned error joins the failures of every listener that rejected the event; the remaining listeners are still invoked.

func (*Bus) Subscribe

func (b *Bus) Subscribe(id ListenerID, matches func(event any) bool, handle Handler) error

Subscribe registers a listener under a unique identifier. The matches predicate decides which events the listener receives.

type ConsumerConfiguration

type ConsumerConfiguration struct {
	// Ordering selects FIFO (the default) or unordered processing.
	Ordering Ordering

	// MaximumAttempts bounds how many delivery attempts one event may consume
	// before it is parked as a dead letter (default 5).
	MaximumAttempts int

	// RetryDelay returns how long to wait before the next attempt, given the
	// attempt count just spent (1 for the first attempt). The default doubles
	// from 5 seconds up to 5 minutes.
	RetryDelay func(attempt int) time.Duration

	// Workers bounds how many events an unordered consumer processes at once
	// (default 8). A FIFO consumer always processes one event at a time.
	Workers int

	// Broadcast requests one delivery per application node instead of one
	// delivery per consumer. Engines confined to a single process treat it as
	// regular consumption, because the node and the consumer coincide.
	Broadcast bool
}

ConsumerConfiguration is the resolved behavior of one consumer. Construct it through DefaultConsumerConfiguration and the ConsumerOption functions.

func DefaultConsumerConfiguration

func DefaultConsumerConfiguration() ConsumerConfiguration

DefaultConsumerConfiguration returns the configuration every consumer starts from: FIFO ordering, five attempts, and a doubling backoff.

type ConsumerOption

type ConsumerOption func(*ConsumerConfiguration)

ConsumerOption customizes one consumer at subscription time.

func WithBroadcastDelivery

func WithBroadcastDelivery() ConsumerOption

WithBroadcastDelivery requests one delivery per application node instead of one delivery per consumer, for node-local concerns such as invalidating an in-memory cache. Engines confined to a single process treat it as regular consumption.

func WithMaximumAttempts

func WithMaximumAttempts(attempts int) ConsumerOption

WithMaximumAttempts overrides how many delivery attempts one event may consume before it is parked as a dead letter (default 5).

func WithRetryDelay

func WithRetryDelay(delay func(attempt int) time.Duration) ConsumerOption

WithRetryDelay overrides the backoff schedule of failed deliveries: the function receives the attempt count just spent (1 for the first attempt) and returns how long to wait before the next one.

func WithUnorderedDelivery

func WithUnorderedDelivery() ConsumerOption

WithUnorderedDelivery opts the consumer out of FIFO ordering: events are processed concurrently and a failing event retries without delaying the others.

func WithWorkers

func WithWorkers(workers int) ConsumerOption

WithWorkers overrides how many events an unordered consumer processes at once (default 8). A FIFO consumer ignores it and processes one at a time.

type ConsumerRegistration

type ConsumerRegistration struct {
	// ID identifies the consumer; it must be unique within the broker.
	ID ListenerID

	// Matches decides which events the consumer receives.
	Matches func(event any) bool

	// Probe carries the zero value of the consumed event type, so an engine
	// that persists events can resolve the durable type name without invoking
	// the handler. In-memory engines ignore it.
	Probe any

	// Handle processes one delivered event.
	Handle Handler

	// Configuration is the resolved consumer behavior.
	Configuration ConsumerConfiguration
}

ConsumerRegistration describes one consumer to a Broker engine.

type DeadLetter

type DeadLetter struct {
	// Reference identifies the dead letter towards Resubmit. Its format is
	// engine-specific and opaque to the caller.
	Reference string

	// ListenerID names the consumer whose delivery exhausted its budget.
	ListenerID ListenerID

	// Event holds the parked event when the engine can restore it.
	Event any

	// Attempts counts the delivery attempts consumed.
	Attempts int

	// LastError describes the failure of the final attempt.
	LastError string

	// PublicationDate records when the event was published.
	PublicationDate time.Time
}

DeadLetter describes an event whose delivery consumed its attempt budget for one consumer. It stays inspectable until an operator resubmits it.

type Handler

type Handler func(ctx context.Context, event any) error

Handler processes one event delivered to a listener.

type ListenerID

type ListenerID string

ListenerID identifies a subscribed listener.

type MemoryBroker

type MemoryBroker struct {
	// contains filtered or unexported fields
}

MemoryBroker is the in-process Broker engine: every consumer owns a queue in memory and nothing survives the process. Delivery is exactly once per consumer for the lifetime of the process; there is no crash recovery, which is the concern the persistent engines add behind the same contract.

The broker composes the synchronous Bus of this package as its delivery fabric, so handler panics are recovered per delivery and cannot take a queue worker down. A MemoryBroker is safe for concurrent use.

func NewMemoryBroker

func NewMemoryBroker() *MemoryBroker

NewMemoryBroker constructs an empty MemoryBroker.

func (*MemoryBroker) FindExhausted

func (b *MemoryBroker) FindExhausted(_ context.Context, limit int) ([]DeadLetter, error)

FindExhausted returns the dead letters, oldest first, up to the limit.

func (*MemoryBroker) Publish

func (b *MemoryBroker) Publish(_ context.Context, event any) error

Publish hands the event to the queue of every matching consumer. It never waits for handlers: outcomes settle asynchronously through the per-consumer retry machinery.

func (*MemoryBroker) Resubmit

func (b *MemoryBroker) Resubmit(_ context.Context, reference string) (bool, error)

Resubmit removes the referenced dead letter and re-enqueues its event with a fresh attempt budget at the position its original publication order dictates, matching how the persistent engines order resubmissions. It reports false when the reference is unknown.

func (*MemoryBroker) Stop

func (b *MemoryBroker) Stop()

Stop cancels the workers, wakes every one blocked on an empty queue, and waits for the in-flight deliveries to return. Queued events and pending retries are dropped: the memory engine carries no durability by design.

func (*MemoryBroker) Subscribe

func (b *MemoryBroker) Subscribe(_ context.Context, registration ConsumerRegistration) error

Subscribe registers the consumer and starts its workers: one for a FIFO consumer, Configuration.Workers for an unordered one.

type Ordering

type Ordering string

Ordering selects how a consumer works through its queue.

const (
	// OrderingFIFO processes the events of the consumer strictly in
	// publication order, one at a time. A failing event blocks its successors
	// while it retries; once it exhausts its attempt budget it is parked as a
	// dead letter and the queue continues.
	OrderingFIFO Ordering = "FIFO"

	// OrderingUnordered processes the events of the consumer concurrently,
	// with no ordering guarantee. A failing event retries on its own schedule
	// without delaying the others.
	OrderingUnordered Ordering = "UNORDERED"
)