eventsourcing

package module
v0.1.8 Latest Latest
Warning

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

 Go to latest Published: Sep 27, 2026 License: MPL-2.0 Imports: 17 Imported by: 0

README

TerraSkye Eventsourcing

eventsourcing is a generic, type-safe event sourcing framework for Go.
It provides the building blocks for event-driven architectures, including command handling, query handling, event buses, envelopes, and a flexible iterator for read models.

This library focuses on simplicity, modern Go patterns, and full generics support, making it easy to build event-sourced systems with strong type guarantees.

Features

  • Commands & Command Handlers – Strongly typed command routing.
  • Events & Event Handlers – Publish and consume events safely.
  • Event Bus – Supports multiple subscribers with typed handlers.
  • Queries & Query Handlers – Request data through a type-safe query bus.
  • Query Gateway – Simple façade to dispatch typed queries.
  • Generic Iterator – Lazy, paginated, or buffered read model iteration.
  • Revision Management – Built-in support for aggregate and stream revisions.
  • Metadata & Envelopes – Rich event metadata included by default.
  • Type-Safe Generics Everywhere – Commands, events, queries, handlers, results.

Installation

go get github.com/terraskye/eventsourcing

OpenTelemetry Instrumentation

The otel subpackage provides built-in observability for your event-sourced application using OpenTelemetry standards.

Why Instrumentation?

Event-sourced systems are inherently distributed and asynchronous. Without proper observability:

  • Command failures are hard to diagnose
  • Event handler latency goes unnoticed
  • Concurrency conflicts are invisible
  • Performance bottlenecks in the event store remain hidden

The otel package wraps your handlers and stores with tracing spans and metrics, giving you full visibility into your system's behavior without modifying business logic.

What's Instrumented
Component Spans Metrics
Command Handlers command.handle <Type> duration, in-flight, handled, failed, conflicts
Event Handlers events.handle <Type> duration, handled
Event Store EventStore.Save, EventStore.LoadStream, etc. duration, saves, loads, errors, events appended/loaded
How to Use

Wrap your handlers and stores with the telemetry decorators:

import "github.com/terraskye/eventsourcing/otel"

// Wrap a command handler
handler := otel.WithCommandTelemetry(myCommandHandler)

// Wrap an event handler
eventHandler := otel.WithEventTelemetry(myEventHandler)

// Wrap an event store
store := otel.WithEventStoreTelemetry(myEventStore)
Configuration Options

Customize span names and attributes using options:

// Static operation name
handler := otel.WithCommandTelemetry(myHandler,
    otel.WithOperation("order.create"),
)

// Add static attributes to all spans
handler := otel.WithCommandTelemetry(myHandler,
    otel.WithAttributes(
        attribute.String("service.name", "orders"),
        attribute.String("service.version", "1.0.0"),
    ),
)

// Dynamic operation name based on context
handler := otel.WithCommandTelemetry(myHandler,
    otel.WithOperationGetter(func(ctx context.Context, defaultOp string) string {
        if tenant := TenantFromContext(ctx); tenant != "" {
            return fmt.Sprintf("%s [%s]", defaultOp, tenant)
        }
        return defaultOp
    }),
)

// Extract dynamic attributes from context
handler := otel.WithCommandTelemetry(myHandler,
    otel.WithAttributeGetter(func(ctx context.Context) []attribute.KeyValue {
        return []attribute.KeyValue{
            attribute.String("tenant.id", TenantFromContext(ctx)),
            attribute.String("user.id", UserFromContext(ctx)),
        }
    }),
)
Available Metrics

Commands:

  • eventsourcing.commands.handled - total successful commands
  • eventsourcing.commands.failed - total failed commands
  • eventsourcing.commands.duration - histogram of handling time (ms)
  • eventsourcing.commands.in_flight - currently processing
  • eventsourcing.concurrency.conflicts - optimistic locking conflicts

Events:

  • eventsourcing.eventbus.handled - events processed by handlers
  • eventsourcing.eventbus.duration - handler execution time (ms)
  • eventsourcing.events.appended - events written to store
  • eventsourcing.events.loaded - events read from store

Event Store:

  • eventsourcing.eventstore.saves - save operations
  • eventsourcing.eventstore.duration - operation time (ms)
  • eventsourcing.eventstore.errors - failed operations
Semantic Attributes

All spans include semantic attributes following OpenTelemetry conventions:

  • eventsourcing.command.type - the command type name
  • eventsourcing.aggregate.id - target aggregate ID
  • eventsourcing.stream.id - event stream identifier
  • eventsourcing.stream.version - stream version after operation
  • eventsourcing.event.type - event type name
  • eventsourcing.event.id - unique event ID
  • eventsourcing.event.global_position - global ordering position
  • eventsourcing.event.stream_position - position within stream

Documentation

Index

Constants

View Source 
const (
	InstrumentationVersion = "v0.1.7"
)

InstrumentationVersion is the module version reported as the instrumentation version of this package's OpenTelemetry meter and tracer.

Variables

View Source 
var (
	// ErrStreamNotFound is returned by an [EventStore] when [StreamExists] is
	// required for a stream that does not exist.
	ErrStreamNotFound = errors.New("stream not found")
	// ErrStreamExists is returned by an [EventStore] when [NoStream] is
	// required for a stream that already exists.
	ErrStreamExists = errors.New("stream already exists")
	// ErrInvalidEventBatch is returned by [EventStore.Save] when the given
	// envelopes do not all share the same stream ID.
	ErrInvalidEventBatch = errors.New("invalid event batch")
	// ErrHandlerNotFound is returned by a [QueryBus] when no [QueryHandler]
	// is registered for a query's type.
	ErrHandlerNotFound = errors.New("handler not registered")
	// ErrInvalidRevision is returned by an [EventStore] when given a
	// [StreamState] implementation it does not support.
	ErrInvalidRevision = errors.New("invalid revision")
	// ErrHandlerNotRegistered is returned by a [CommandBus] when no
	// [CommandHandler] is registered for a command's type.
	ErrHandlerNotRegistered = errors.New("no handler registered")
	// ErrDuplicateHandler is returned or panicked with when registering a
	// second handler for a command or event type that already has one, by
	// [CommandBus.Register], [EventBus.Subscribe] implementations, and
	// [QueryBus] registration.
	ErrDuplicateHandler = errors.New("duplicate handler registered")
	// ErrHandlerPanicked is joined into the error a [CommandBus] returns
	// when a [CommandHandler] panics instead of returning an error.
	ErrHandlerPanicked = errors.New("handler panicked when handling command")
	// ErrCommandBusClosed is wrapped into the error a [CommandBus] returns
	// when Dispatch is called after Stop.
	ErrCommandBusClosed = errors.New("command bus is closed")
	// ErrEventNotRegistered is returned by [NewEventByName] when no event
	// type is registered under the given name.
	ErrEventNotRegistered = errors.New("event not registered")
)
View Source 
var (

	// RegisterEventByType registers fn under the type name of the [Event] it
	// returns, i.e. fn().EventType(). Use this instead of [RegisterEvent]
	// when you need explicit control over the factory, for example to
	// inject constructor arguments. It panics if fn is nil, if fn() returns
	// nil, or if an event is already registered under that name.
	//
	// Example Usage:
	//   RegisterEventByType(func() Event { return &InventoryChanged{} })
	RegisterEventByType func(fn func() Event) = func(fn func() Event) {
		registerEventNameDefault(fn().EventType(), fn)
	}

	// RegisterEventByName registers fn under name, independently of the
	// event's own EventType() — useful when the name a store has events
	// persisted under no longer matches the current type name, for example
	// after a rename. As with [RegisterEventByType], it panics if fn is nil,
	// if fn() returns nil, or if name is already registered.
	//
	// Example Usage:
	//   RegisterEventByName("CustomEventName", func() Event { return &InventoryChanged{} })
	RegisterEventByName func(name string, fn func() Event) = func(name string, fn func() Event) {
		registerEventNameDefault(name, fn)
	}

	// NewEventByName returns a new instance of the [Event] registered under
	// name, or a non-nil [ErrEventNotRegistered] if no event is registered
	// under that name.
	//
	// Example Usage:
	//   ev, err := NewEventByName("InventoryChanged")
	NewEventByName func(name string) (Event, error) = newEventByNameDefault

	// EventNamesFor returns every name event's concrete type is registered
	// under, regardless of which [RegisterEvent]/[RegisterEventByType]/
	// [RegisterEventByName] call added each one.
	EventNamesFor func(event Event) []string = func(event Event) []string {
		registryMu.RLock()
		defer registryMu.RUnlock()

		return typeToNames[eventTypeKey(event)]
	}
)

Functions

func AggregateIDFromContext 

func AggregateIDFromContext(ctx context.Context) string

AggregateIDFromContext returns the aggregate ID set by WithEnvelope, or "" if ctx carries none.

func CausationFromContext 

func CausationFromContext(ctx context.Context) string

CausationFromContext returns the causation ID set by WithCausation, or "" if ctx carries none.

func EventIDFromContext 

func EventIDFromContext(ctx context.Context) uuid.UUID

EventIDFromContext returns the event ID set by WithEnvelope, or uuid.Nil if ctx carries none.

func GlobalVersionFromContext 

func GlobalVersionFromContext(ctx context.Context) uint64

GlobalVersionFromContext returns the global version set by WithEnvelope, or 0 if ctx carries none.

func MetadataFromContext 

func MetadataFromContext(ctx context.Context) map[string]any

MetadataFromContext returns the metadata set by WithEnvelope, or nil if ctx carries none.

func NewBusinessRuleViolation added in v0.1.7 

func NewBusinessRuleViolation(err error) error

NewBusinessRuleViolation wraps err as a BusinessRuleViolationError. If err is nil, it returns nil.

NewCommandHandler already wraps whatever error its Decider returns in a BusinessRuleViolationError, so a Decider used with it should just return the plain error — calling this in a Decider would nest one violation inside another. Use it when signaling a business rule violation from a hand-rolled CommandHandler that doesn't go through NewCommandHandler's decide step, so callers (e.g. an OpenTelemetry middleware) can still recognize it as an expected, recoverable rejection via errors.As.

func OccurredAtFromContext 

func OccurredAtFromContext(ctx context.Context) time.Time

OccurredAtFromContext returns the occurred-at time set by WithEnvelope, or the zero time.Time if ctx carries none.

func Register 

func Register[C Command](b *CommandBus, handler CommandHandler[C])

Register installs handler as the handler for command type C on b. The registration key is C's Command.CommandType, so there are no manual type strings to keep in sync. Register panics with ErrDuplicateHandler if a handler for C is already registered.

The middleware chain is applied here, at registration time, from the middlewares added via CommandBus.Use up to this point — so call Use first. Register is safe to call on a bus that is already dispatching, though wiring every handler during startup is the expected pattern.

Example: 

err := Register(bus, fooHandler)

func RegisterEvent 

func RegisterEvent[T any, PT eventPtr[T]](_ PT)

RegisterEvent registers the concrete event type T — inferred from the pointer passed in, whose value is otherwise discarded — under its default Event.EventType name. Each later NewEventByName call for that name returns a fresh new(T), so unlike a hand-written closure over a single instance, concurrent or repeated decodes never alias the same value. It panics if an event is already registered under that name.

Example Usage: 

RegisterEvent(&OrderCreated{})

func RegisterQueryHandler 

func RegisterQueryHandler[T Query, R any](bus *QueryBus, handler QueryHandler[T, R], opts ...HandlerOption)

RegisterQueryHandler registers a QueryHandler[T, R] on the bus. Use this when registering a type that explicitly implements the QueryHandler interface. For plain functions or method values, prefer RegisterQueryHandlerFunc.

Panics if a handler for the same query and result types is already registered.

Example Usage: 

RegisterQueryHandler(bus, myHandler)

func RegisterQueryHandlerFunc added in v0.1.6 

func RegisterQueryHandlerFunc[T Query, R any](bus *QueryBus, fn queryHandlerFunc[T, R], opts ...HandlerOption)

RegisterQueryHandlerFunc registers a plain function as a query handler. Type parameters are inferred from the function signature. Prefer this over RegisterQueryHandler when registering method values from a provider struct.

Panics if a handler for the same query and result types is already registered.

Example Usage: 

RegisterQueryHandlerFunc(bus, store.GetTask)
RegisterQueryHandlerFunc(bus, store.ListTasks)

func StreamIDFromContext 

func StreamIDFromContext(ctx context.Context) string

StreamIDFromContext returns the stream ID set by WithEnvelope, or "" if ctx carries none.

func VersionFromContext 

func VersionFromContext(ctx context.Context) uint64

VersionFromContext returns the stream version set by WithEnvelope, or 0 if ctx carries none.

func WithCausation 

func WithCausation(ctx context.Context, causation string) context.Context

WithCausation returns a copy of ctx carrying causation, the identifier of whatever caused the work now happening under ctx — for example, a command's type name — for handlers to attach to events or log entries.

func WithEnvelope 

func WithEnvelope(ctx context.Context, env *Envelope) context.Context

WithEnvelope returns a copy of ctx carrying env's stream ID, aggregate ID, event ID, version, global version, occurred-at time, and metadata, retrievable via the *FromContext functions below. It does not carry a causation ID; use WithCausation for that. The aggregate ID is "" if env.Event is nil.

Types

type Any 

type Any struct{}

Any expects nothing: append without checking the stream's current revision.

func (Any) String added in v0.1.8 

func (Any) String() string

func (Any) ToRawInt64 

func (Any) ToRawInt64() int64

type AppendResult 

type AppendResult struct {
	// Successful reports whether the events were persisted.
	Successful bool
	// StreamID is the stream the events were saved to.
	StreamID string
	// NextExpectedVersion is the version the stream's next [Revision] should
	// use.
	NextExpectedVersion uint64
}

AppendResult describes the outcome of an EventStore.Save call.

type BusinessRuleViolationError added in v0.1.8 

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

BusinessRuleViolationError wraps an error returned when a Command violates a business rule. Use it to signal expected, recoverable domain-level rejections, as opposed to infrastructure or persistence errors. Its cause is unexported — construct one with NewBusinessRuleViolation and read the cause back with BusinessRuleViolationError.Cause or errors.Unwrap.

func (*BusinessRuleViolationError) Cause added in v0.1.8 

func (e *BusinessRuleViolationError) Cause() error

func (*BusinessRuleViolationError) Error added in v0.1.8 

func (e *BusinessRuleViolationError) Error() string

func (*BusinessRuleViolationError) Unwrap added in v0.1.8 

func (e *BusinessRuleViolationError) Unwrap() error

type Command 

type Command interface {
	// AggregateID returns the ID of the entity this command targets, used to
	// locate that entity's state when handling the command.
	AggregateID() string
	// CommandType returns the name under which this command's concrete type is
	// registered; see [Register]. It must be stable across instances, since it
	// is derived from the zero value at registration time.
	CommandType() string
}

Command represents a user or system's intent to perform an action on a specific domain entity.

A Command models intent, not implementation: name it after what the user or system wants to achieve, in terms meaningful to the domain, rather than generic or mechanical terms like "Updated" or "Change". For example: 

Command Name      Intent
---------------- ----------------------------------------
ReserveSeat       Reserve a specific seat
CancelOrder       Cancel an existing order
ApproveUser       Approve a user registration
MarkInvoicePaid   Mark an invoice as paid
ShipOrder         Ship a customer order
CreateAccount     Create a new user account

Every Command must identify the entity it targets through Command.AggregateID, so a handler can locate that entity's state, and name itself through Command.CommandType, which a CommandBus uses to route it to the handler registered for it. A Command should be immutable after creation, representing a fixed intention at a point in time, and self-contained, carrying everything needed to handle it rather than relying on hidden state. For example: 

type ReserveSeat struct {
	ScreeningID string
	SeatNumber  string
	UserID      string
}

func (c ReserveSeat) AggregateID() string {
	return c.ScreeningID
}

func (c ReserveSeat) CommandType() string {
	return "ReserveSeat"
}

type CommandBus 

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

CommandBus is an in-memory, type-safe command dispatcher. It keeps a registry of command type names to handlers, sharded queues of incoming commands, and the synchronization needed for concurrent [Dispatch] and Register calls. Commands for a given aggregate are always routed to the same shard and processed one at a time, [Stop] waits for in-flight dispatches to finish, and a handler panic is recovered rather than crashing the bus.

func NewCommandBus 

func NewCommandBus(bufferSize int, shardCount int) *CommandBus

NewCommandBus returns a CommandBus with shardCount shards, each with a worker goroutine already running and a queue buffering up to bufferSize commands. shardCount is clamped to at least 1.

Example: 

bus := NewCommandBus(100, 4)

func (*CommandBus) Dispatch 

func (b *CommandBus) Dispatch(ctx context.Context, cmd Command) (AppendResult, error)

Dispatch enqueues cmd for the handler registered for its type and blocks until that handler returns. Commands for the same aggregate are routed to the same shard, and each shard is drained by a single worker, so they are never handled concurrently. Dispatch itself is safe to call concurrently.

The returned AppendResult carries the outcome of the append. The error is non-nil if ctx is cancelled before the command is enqueued or before the result arrives, if no handler is registered for the command's type, if the handler returns an error or panics, or if the bus has been stopped. Once the bus is stopped the error wraps ErrCommandBusClosed, including for a call already parked waiting to enqueue.

func (*CommandBus) Stop 

func (b *CommandBus) Stop()

Stop shuts down the bus. It stops accepting new commands, lets the workers finish whatever is already queued, and waits for every in-flight CommandBus.Dispatch to return before it does.

A Dispatch racing Stop either completes normally or fails with an error wrapping ErrCommandBusClosed; it never panics. Stop is idempotent and safe to call concurrently, but the bus cannot be restarted afterwards.

func (*CommandBus) Use added in v0.1.6 

func (b *CommandBus) Use(middlewares ...CommandHandlerMiddleware)

Use appends middlewares to the chain applied to command handlers. The first one appended is the outermost wrapper and runs first on each dispatch.

Use must be called before Register: the chain is baked into each handler at registration time, so handlers already registered are not re-wrapped and a later Use call has no effect on them. Configure the full chain during startup wiring.

Example Usage: 

bus.Use(
    logging.CommandLogging(logger),
    otel.CommandTelemetry(),
)

type CommandHandler 

type CommandHandler[C Command] func(ctx context.Context, command C) (AppendResult, error)

CommandHandler handles commands of the concrete type C, which must implement Command. A handler carries out the business logic for a command — validating it, deciding what should happen, and persisting any resulting events — and is typically registered with a CommandBus via Register, which dispatches each command to the handler registered for its type.

It returns an AppendResult describing the outcome and a non-nil error if handling failed, for example due to a validation error, a business-rule violation, or a persistence failure. Implementations should treat the command as immutable, express state changes as events persisted to an EventStore rather than by mutating in-memory state, and return errors rather than panicking.

Example Usage: 

func HandleReserveSeat(ctx context.Context, cmd ReserveSeat) (AppendResult, error) {
    if seatAlreadyReserved(cmd.SeatNumber) {
        return AppendResult{Successful: false}, fmt.Errorf("seat already reserved")
    }
    // persist a SeatReserved event via an EventStore, then:
    return AppendResult{Successful: true, StreamID: cmd.AggregateID()}, nil
}

func NewCommandHandler 

func NewCommandHandler[T any, C Command](
	store EventStore,
	initialState InitialState[T],
	evolve Evolver[T],
	decide Decider[T, C],
	opts ...CommandHandlerOption,
) CommandHandler[C]

NewCommandHandler returns a generic command handler for any aggregate type.

The returned handler, on each call: loads command's stream from store (named by DefaultStreamNamer, or a StreamNamer set via WithStreamNamer) using EventStore.LoadStreamFrom; folds every loaded event into initialState with evolve; passes the resulting state and the command to decide to get the events to persist; wraps them in [Envelope]s, stamping each with a new UUID, the next sequential version, the current time, and any metadata from functions added via WithMetadataExtractor; and saves them with EventStore.Save.

The StreamState configured via WithStreamState (default Any) determines how a save conflict is handled. Any and StreamExists don't pin the stream to an exact version — Any asserts nothing, and StreamExists only asserts non-emptiness, once loading has confirmed the stream exists — so the handler saves against the revision it just loaded and, on conflict, retries the whole load-evolve-decide-save cycle according to the backoff.BackOff set via WithRetryStrategy (default: no retries). NoStream and a specific Revision pin the stream to an exact point (version 0, or N) that the caller explicitly asserted; a conflict there means that expectation was violated, so it is returned immediately rather than retried. A StreamExists load failure (the stream doesn't exist yet) is likewise always returned immediately, never retried — it's a fail-fast precondition, not a save conflict to converge on. Any other Save or load error is also returned directly, without retrying.

If decide returns no events, the handler returns a successful AppendResult without calling Save. If decide returns a non-nil error, the handler returns it wrapped in a BusinessRuleViolationError.

Example Usage: 

handler := NewCommandHandler(store, initialStateFunc, evolveFunc, decideFunc, WithStreamState(Any{}))
result, err := handler(ctx, myCommand)

type CommandHandlerMiddleware added in v0.1.6 

type CommandHandlerMiddleware func(next CommandHandler[Command]) CommandHandler[Command]

CommandHandlerMiddleware decorates a command handler on a CommandBus with a cross-cutting concern such as logging, telemetry, or rate limiting. It receives the next handler in the chain and returns a handler that wraps it; call next to pass the command along, or return without calling it to short-circuit.

The chain is baked into a handler once, by Register, from the middlewares added via CommandBus.Use up to that moment. Add all middleware during startup wiring, before registering any handler. The first middleware passed to Use is the outermost wrapper and runs first on each dispatch.

Example Usage: 

var rateLimiter eventsourcing.CommandHandlerMiddleware = func(
    next eventsourcing.CommandHandler[eventsourcing.Command],
) eventsourcing.CommandHandler[eventsourcing.Command] {
    return func(ctx context.Context, cmd eventsourcing.Command) (eventsourcing.AppendResult, error) {
        if !myLimiter.Allow() {
            return eventsourcing.AppendResult{}, errors.New("rate limit exceeded")
        }
        return next(ctx, cmd)
    }
}
bus.Use(rateLimiter)

type CommandHandlerOption 

type CommandHandlerOption func(configuration *handlerOptions)

CommandHandlerOption configures a CommandHandler built by NewCommandHandler.

func WithMetadataExtractor 

func WithMetadataExtractor(fn func(ctx context.Context) map[string]any) CommandHandlerOption

WithMetadataExtractor adds fn to the metadata functions a NewCommandHandler calls for every command, merging their results into each resulting Envelope's Metadata. Multiple extractors can be combined; they run in the order they were added, later ones overwriting keys set by earlier ones.

Usage: 

handler := NewCommandHandler(store, initialState, evolve, decide, WithMetadataExtractor(myMetadataFunc))

func WithRetryStrategy 

func WithRetryStrategy(strategy backoff.BackOff) CommandHandlerOption

WithRetryStrategy sets the backoff.BackOff a NewCommandHandler uses to retry its load-evolve-decide-save cycle after a save conflict, when the handler is configured (via WithStreamState) with Any (the default) or StreamExists — neither pins the stream to an exact version, so a save conflict can be resolved by retrying. It does not apply to a StreamExists load failure (the stream not existing yet), which is always returned immediately. Without this option, no retries are performed and the handler returns the conflict directly. It has no effect when WithStreamState is set to NoStream or a specific Revision — those pin an exact point, so a conflict there is always returned immediately.

Usage: 

handler := NewCommandHandler(store, initialState, evolve, decide, WithRetryStrategy(myBackoff))

func WithStreamNamer 

func WithStreamNamer(namer StreamNamer) CommandHandlerOption

WithStreamNamer overrides the StreamNamer a NewCommandHandler uses to name the stream it loads and saves to, in place of DefaultStreamNamer.

Usage: 

handler := NewCommandHandler(store, initialState, evolve, decide, WithStreamNamer(myStreamNamer))

func WithStreamState 

func WithStreamState(state StreamState) CommandHandlerOption

WithStreamState sets the StreamState a NewCommandHandler expects the stream to be in when it saves events — Any (the default) to save against the revision the handler just loaded, or StreamExists to do the same after additionally requiring the stream to already exist; both retry on save conflict per WithRetryStrategy, since neither pins an exact version. (A StreamExists load failure — the stream not existing yet — is still a fail-fast precondition and is always returned immediately.) NoStream (stream must not exist) or a specific Revision (stream must be at exactly that version) pin the stream to an exact point the caller explicitly asserted; a conflict there is always returned immediately, never retried, since retrying would silently move past that point.

Usage: 

handler := NewCommandHandler(store, initialState, evolve, decide, WithStreamState(NoStream{}))

type Decider 

type Decider[T any, C Command] func(state T, cmd C) ([]Event, error)

Decider inspects state, as produced by an Evolver, against cmd and returns the events that should occur as a result, or a non-nil error if cmd violates a business rule or cannot be applied to state. It must not mutate state; state changes are expressed only through the returned events. An empty, nil slice means cmd produces no events, for example because it was idempotent or had no effect.

type Dispatcher 

type Dispatcher interface {
	Dispatch(ctx context.Context, cmd Command) (AppendResult, error)
}

Dispatcher has the same signature as CommandBus.Dispatch, for code that wants to depend on the dispatch behavior without depending on *CommandBus itself.

type Envelope 

type Envelope struct {
	// EventID uniquely identifies this occurrence of the event.
	EventID uuid.UUID
	// StreamID is the ID of the stream the event was appended to, typically
	// the aggregate's ID.
	StreamID string
	// Metadata carries caller-supplied, out-of-band information about the
	// event, such as correlation IDs or the acting user.
	Metadata map[string]any
	// Event is the wrapped domain event.
	Event Event
	// Version is the event's position within its own stream, starting at 1.
	Version uint64
	// GlobalVersion is the event's position across all streams, used to
	// resume a global subscription from a specific point.
	GlobalVersion uint64
	// OccurredAt is when the event was appended to the store.
	OccurredAt time.Time
}

Envelope wraps an Event with the metadata an EventStore or EventBus needs to persist, order, and route it.

type Event 

type Event interface {
	// AggregateID returns the ID of the aggregate this event happened to.
	AggregateID() string
	// EventType returns the name under which this event's concrete type is
	// registered; see [RegisterEvent].
	EventType() string
}

Event is a domain event describing a change that has already happened to an aggregate.

type EventBus 

type EventBus interface {
	// Subscribe registers handler under name to receive events, optionally
	// narrowed by options. It returns an error if handler is nil, name is
	// already registered, or the subscription otherwise could not be added
	// (for example, for a networked implementation).
	Subscribe(ctx context.Context, name string, handler EventHandler, options ...SubscriberOption) error

	// Use adds middlewares to be applied to every handler registered via
	// Subscribe afterward. It must be called before Subscribe, since
	// middleware is applied at subscribe time.
	Use(middlewares ...EventHandlerMiddleware)

	// Errors returns a channel on which asynchronous handling errors are
	// delivered.
	Errors() <-chan error

	// Close shuts down the EventBus and waits for all handlers to finish
	// processing events already delivered to them.
	Close() error
}

EventBus lets [EventHandler]s subscribe to receive events as they occur, fanning each event out to every matching subscriber. How events are fed into the bus (typically as a side effect of an EventStore save) is left to the implementation.

type EventGroupProcessor 

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

EventGroupProcessor routes each incoming event to the EventHandler registered for its concrete type, typically one built with OnEvent.

func NewEventGroupProcessor 

func NewEventGroupProcessor(handlers ...EventHandler) *EventGroupProcessor

NewEventGroupProcessor builds an EventGroupProcessor from handlers, which must each implement an internal EventName() string method — as the handlers returned by OnEvent do — used as the routing key. It panics if a handler doesn't implement that method, or if two handlers report the same EventName().

Example Usage: 

p := &Projector{}
group := NewEventGroupProcessor(
    OnEvent(p.OnCartCreated),
    OnEvent(p.OnItemAdded),
)
group.Handle(ctx, CartCreated{ID: "t1"})
group.Handle(ctx, ItemAdded{ID: "c1"})

func (*EventGroupProcessor) Handle 

func (p *EventGroupProcessor) Handle(ctx context.Context, ev Event) error

Handle routes ev to the handler registered for its concrete type, or returns SkippedEventError if none is registered.

func (*EventGroupProcessor) StreamFilter 

func (p *EventGroupProcessor) StreamFilter() []string

StreamFilter returns the sorted names of every registered event type this group has a handler for — useful, for example, as the filter passed to EventBus.Subscribe via a filtering SubscriberOption. For a handled type registered in the global event registry (see EventNamesFor) it includes every name that type is registered under, since one concrete event struct can be registered under several names (for example after a rename, via RegisterEventByName) and a subscriber needs to match all of them. A handled type that isn't registered contributes no name and is silently omitted from the result, rather than falling back to that type's own Event.EventType.

A handler that does not expose an EventInstance is resolved through its routing key instead, which is the same type key the registry files names under. NewEventGroupProcessor requires only EventName, so a hand-rolled handler that never implements EventInstance is a supported shape and is not dropped for it.

type EventHandler 

type EventHandler interface {
	// Handle processes event within ctx.
	Handle(ctx context.Context, event Event) error
}

EventHandler processes events delivered by an EventBus or EventGroupProcessor.

func NewEventHandlerFunc 

func NewEventHandlerFunc(fn func(ctx context.Context, event Event) error) EventHandler

NewEventHandlerFunc returns fn as an EventHandler, for quickly wrapping a function without defining a separate type. fn is called for every event it is invoked with, unfiltered by type, so this is the constructor to reach for when fn switches on the event type itself: 

switch ev := event.(type) {
case *CartCreated:
case *ItemAdded:
}

The returned handler cannot be registered with an EventGroupProcessor — that requires each handler to report the single event type name it handles, which a handler for every event type has no one value for. Use OnEvent for a handler usable there; it takes one concrete event type and will not compile against Event itself, since a handler for the interface could never be routed to.

Example Usage: 

handler := NewEventHandlerFunc(func(ctx context.Context, ev Event) error {
    fmt.Println("Received event:", TypeName(ev))
    return nil
})
err := bus.Subscribe(ctx, "logger", handler)

func OnEvent 

func OnEvent[T any, PT eventPtr[T]](fn func(ctx context.Context, ev PT) error) EventHandler

OnEvent returns fn as an EventHandler that only processes events of type *T, returning SkippedEventError for any other type. It is meant to be registered with an EventGroupProcessor, which uses the type's name to route only matching events to it.

fn must take a pointer, which is the form events arrive in: RegisterEvent registers a type by minting new(T), so NewEventByName — and therefore every EventStore and EventBus that rehydrates an event by name — hands out *T. A handler written over the value type would be routed under a different name than the events it is meant to receive, so it would compile, register, appear in EventGroupProcessor.StreamFilter, and silently never be called.

Example Usage: 

handler := OnEvent(func(ctx context.Context, ev *OrderCreated) error {
    fmt.Println("Order created:", ev.AggregateID())
    return nil
})
group := NewEventGroupProcessor(handler)
group.Handle(ctx, &OrderCreated{ID: "123"})

type EventHandlerMiddleware added in v0.1.6 

type EventHandlerMiddleware func(next EventHandler) EventHandler

EventHandlerMiddleware decorates an event handler on an EventBus with a cross-cutting concern such as logging, telemetry, or error handling. It receives the next handler in the chain and returns a handler that wraps it; call next to pass the event along, or return without calling it to short-circuit. Middleware is applied when a handler is passed to EventBus.Subscribe, and the first middleware passed to EventBus.Use is the outermost wrapper and runs first for each event.

Example Usage: 

var myMiddleware eventsourcing.EventHandlerMiddleware = func(next eventsourcing.EventHandler) eventsourcing.EventHandler {
    return eventsourcing.NewEventHandlerFunc(func(ctx context.Context, event eventsourcing.Event) error {
        // before
        err := next.Handle(ctx, event)
        // after
        return err
    })
}

type EventStore 

type EventStore interface {
	// Save appends events to the stream identified by their common StreamID,
	// which must be the same across every element of events. revision states
	// the caller's expectation of the stream's current state — [Any] to
	// append unconditionally, [NoStream] to require the stream not already
	// exist, [StreamExists] to require that it does, or a specific [Revision]
	// to require an exact version — and Save returns a
	// [StreamRevisionConflictError] if that expectation does not hold.
	Save(ctx context.Context, events []Envelope, revision StreamState) (AppendResult, error)

	// LoadStream returns an iterator over every event in id's stream, in
	// ascending version order.
	LoadStream(ctx context.Context, id string) (*Iterator[*Envelope], error)

	// LoadStreamFrom returns an iterator over id's stream starting at
	// version, in ascending version order. version is typically a
	// [Revision]; [Any] starts from the beginning of the stream.
	LoadStreamFrom(ctx context.Context, id string, version StreamState) (*Iterator[*Envelope], error)

	// LoadFromAll returns an iterator over events across every stream,
	// starting at version. Whether version and the iteration order it
	// produces are globally consistent, or only consistent within each
	// stream, is implementation-specific; consult the implementation before
	// relying on cross-stream ordering.
	LoadFromAll(ctx context.Context, version StreamState) (*Iterator[*Envelope], error)

	// Close releases any resources held by the EventStore, such as network
	// connections or file handles. Implementations should make Close
	// idempotent. The EventStore must not be used after Close is called.
	Close() error
}

EventStore is an append-only store of [Envelope]s, grouped into per- aggregate streams, that allows an aggregate's state to be reconstructed by replaying its stream.

Implementations must store events for a given stream in the order they were saved and yield them in that same order — oldest first — from every Load* method. The Iterator values Load* methods return are lazy and should be consumed promptly; implementations make no guarantee about their reusability or thread-safety once iteration ends.

A Load* iterator is bound to the ctx it was created with, and may hold resources — a database cursor, an open read stream — until iteration ends. Reading it to the end or failing releases them; a caller that may stop early must call Iterator.Close, which is safe to defer: 

it, err := store.LoadStream(ctx, id)
if err != nil {
	return err
}
defer it.Close()
for it.Next() {
	state = evolve(state, it.Value())
}
return it.Err()

type EventStoreMiddleware added in v0.1.6 

type EventStoreMiddleware func(next EventStore) EventStore

EventStoreMiddleware decorates an EventStore, typically to intercept one or more of its methods while delegating the rest to next. Apply one by wrapping a store directly: store = mw(store). A common implementation embeds EventStore for the pass-through methods and overrides only the ones of interest.

Example Usage: 

var metered eventsourcing.EventStoreMiddleware = func(next eventsourcing.EventStore) eventsourcing.EventStore {
    return &meteredStore{next: next, counter: myCounter}
}
store = metered(baseStore)

type Evolver 

type Evolver[T any] func(currentState T, envelope *Envelope) T

Evolver applies a single historical event to currentState and returns the resulting aggregate state of type T. It must not mutate currentState; NewCommandHandler calls it once per event, in stream order, folding the result of each call into the next.

type GenericQueryGateway deprecated 

type GenericQueryGateway[T Query, R any] = QueryGateway[T, R]

GenericQueryGateway is a backwards-compatible alias for QueryGateway.

Deprecated: use QueryGateway directly.

type HandlerOption 

type HandlerOption func(*handlerSettings)

HandlerOption represents an optional configuration function that can modify handler behavior or metadata. Currently reserved for future extensions such as worker pools, timeouts, or rate limiting.

type InitialState 

type InitialState[T any] func() T

InitialState returns the zero-event aggregate state of type T, before any events have been evolved into it — for example, an empty struct or one with its fields set to their defaults.

type IterFunc 

type IterFunc[T any] func(ctx context.Context) (T, error)

IterFunc produces the next item of an Iterator. It returns:

  • (item, nil) for each item, in order;
  • (zero, io.EOF), wrapped or not, after the last item;
  • (zero, err) for any other failure, which ends iteration.

An item returned together with a non-nil error is ignored, so the last item must be returned with a nil error. A function that stops partway through reading an item must return io.ErrUnexpectedEOF or another error rather than io.EOF; otherwise the partial read ends iteration as if the items were exhausted.

ctx is the context the Iterator was created with. A function that blocks must return promptly once ctx is done.

type Iterator 

type Iterator[T any] struct {
	// contains filtered or unexported fields
}

Iterator is a pull-based iterator over items of type T, such as the events returned by the Load methods of an EventStore.

Call Next to advance and Value to read the current item. When Next returns false, call Err to tell a clean end (nil) from a failure: 

it, err := store.Load(ctx, id)
if err != nil {
	return err
}
defer it.Close()
for it.Next() {
	apply(it.Value())
}
return it.Err()

An Iterator may hold resources, such as a database cursor, until iteration ends. Reaching the end or failing releases them automatically. A caller that may stop early must call Close; deferring Close is always safe. Alternatively, range over Iterator.Values or call Iterator.All, both of which close the iterator.

An Iterator is bound to the context it was created with. That context controls the entire iteration, not only the call that created it: keep it alive until iteration ends. Cancelling it ends iteration with the context's error.

An Iterator is single-use and is not safe for concurrent use by multiple goroutines. Create one with NewIteratorFunc or NewSliceIterator; the zero Iterator is not usable and advancing it panics.

func NewIteratorFunc 

func NewIteratorFunc[T any](ctx context.Context, next IterFunc[T], close func() error) *Iterator[T]

NewIteratorFunc returns an Iterator bound to ctx that draws its items from next. ctx must be non-nil; NewIteratorFunc panics if it is nil, since the context cannot be supplied later.

close releases the resources behind next, or is nil if there are none. The Iterator calls it exactly once: when Next first returns false, or when Close is called, whichever comes first. The close function

  • must release every resource before returning, including stopping and waiting for any goroutine next depends on;
  • must not depend on ctx, which is often already cancelled when close runs; derive a context with context.WithoutCancel and a timeout if cleanup needs one;
  • must refer to the resource that is current when it runs. A method value such as rows.Close binds its receiver when it is evaluated, so use a closure if the variable holding the resource is reassigned;
  • must not panic. If it does, the panic propagates and close is not called again, as with sync.Once. A caller that recovers from such a panic must not read a nil Err as a successful iteration: the iterator has no error to report because close never returned one.

func NewSliceIterator 

func NewSliceIterator[T any](ctx context.Context, s []T) *Iterator[T]

NewSliceIterator returns an Iterator bound to ctx that yields the elements of s in order. It holds no resources.

func Wrap added in v0.1.8 

func Wrap[T, U any](src *Iterator[T], next func(context.Context, *Iterator[T]) (U, error), done func(error)) *Iterator[U]

Wrap returns an iterator that draws its items from next, which reads from src. The returned iterator is bound to src's context, and closing it closes src exactly once, on every exit path.

next must return io.EOF when src ends, whatever the reason. It must not return src.Err(): the source's failure, and any error from closing it, reach the returned iterator through Close, and returning them from next as well would report the same failure twice. next may call src.Next any number of times per item, so it can drop items or buffer several.

done, if non-nil, is called exactly once when the returned iterator closes, with the error src ended with, or nil if it ended cleanly. It runs after src has been closed, so work ended there — a span, a metric, a leak check — covers the whole iteration.

func (*Iterator[T]) All 

func (it *Iterator[T]) All() ([]T, error)

All reads the remaining items, closes the iterator, and returns the items along with the result of Err. On failure it returns the items read before the error, as io.ReadAll does.

All returns only items not yet consumed by Next. On an iterator that is already exhausted or closed it returns no items, which cannot be distinguished from an empty sequence.

func (*Iterator[T]) Close added in v0.1.8 

func (it *Iterator[T]) Close() error

Close releases the iterator's resources and ends iteration. It may be called more than once, after iteration has ended, and on a nil Iterator; every call returns the result of the first. After Close, Next returns false.

Close on a nil Iterator returns nil rather than an error, unlike the nil-receiver checks in os.File, so that a deferred Close is harmless after a Load that returned no iterator.

Close must not be used to interrupt a Next running in another goroutine; cancel the iterator's context instead.

func (*Iterator[T]) Err 

func (it *Iterator[T]) Err() error

Err returns the error, if any, that ended iteration, joined with any error returned while closing the iterator. It returns nil if iteration ended because the items were exhausted. Err may be called after an explicit or implicit Close.

Err reports failures, not completeness: after a caller closes the iterator early, Err returns nil unless closing failed. Code that requires a whole stream must therefore read to exhaustion or use All; an early Close followed by an Err check cannot tell a complete stream from a truncated one.

func (*Iterator[T]) Next 

func (it *Iterator[T]) Next() bool

Next advances the iterator to the next item, which is then available through Value. It returns false when the items are exhausted, when the iterator's IterFunc fails, when the iterator's context is done, or after Close. Before returning false for the first time, Next closes the iterator, so that call includes the cost of releasing its resources.

func (*Iterator[T]) Value 

func (it *Iterator[T]) Value() T

Value returns the item produced by the most recent call to Next. It returns the zero value before the first call to Next and once Next has returned false. Items returned by earlier calls remain valid.

func (*Iterator[T]) Values added in v0.1.8 

func (it *Iterator[T]) Values() iter.Seq[T]

Values returns the remaining items as a sequence for use with range. The iterator is closed when the loop ends for any reason, including break, return, and panic. Errors are left to Err, as with a Next loop, so one rule covers both styles: 

for e := range it.Values() {
	apply(e)
}
return it.Err()

Values returns a single-use sequence: like Next, it continues from the current position, does not restart iteration, and yields nothing on a closed iterator.

type NoStream 

type NoStream struct{}

NoStream expects the stream not to exist yet.

func (NoStream) String added in v0.1.8 

func (NoStream) String() string

func (NoStream) ToRawInt64 

func (NoStream) ToRawInt64() int64

type Query 

type Query interface {
	// ID returns an identifier for this query occurrence, for example for
	// correlation in logs or traces.
	ID() []byte
}

Query is implemented by any type that can be handled by a QueryHandler.

type QueryBus 

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

QueryBus is a central registry of query handlers, keyed by their query and result types, so that multiple query types can be registered on a single bus. Handlers are executed through a typed QueryGateway created with NewQueryGateway.

Example Usage: 

bus := NewQueryBus()
RegisterQueryHandlerFunc(bus, store.GetTask)
RegisterQueryHandlerFunc(bus, store.ListTasks)

func NewQueryBus 

func NewQueryBus() *QueryBus

NewQueryBus creates a new, empty QueryBus ready for handler registration.

func (*QueryBus) Use added in v0.1.6 

func (q *QueryBus) Use(middlewares ...QueryHandlerMiddleware)

Use appends middlewares to the chain applied to query handlers. The first one appended is the outermost wrapper and runs first on each query.

Use must be called before RegisterQueryHandler or RegisterQueryHandlerFunc: the chain is baked into each handler at registration time, so handlers already registered are not re-wrapped and a later Use call has no effect on them. Configure the full chain during startup wiring.

Example Usage: 

bus.Use(
    logging.QueryLogging(logger),
    otel.QueryTelemetry(),
)

func (*QueryBus) Validate 

func (q *QueryBus) Validate() error

Validate reports an error listing every query/result type pair that a QueryGateway was created for via NewQueryGateway but that has no registered handler. Call it during startup, after all gateways and handlers are wired up, to catch a missing registration before it can surface as a runtime error.

type QueryGateway added in v0.1.2 

type QueryGateway[T Query, R any] func(ctx context.Context, qry T) (R, error)

QueryGateway is a typed, callable facade over a QueryBus. Call it directly like a function to execute the handler registered for query type T and result type R. It also implements QueryHandler, so it can be passed to decorators such as WithQueryTelemetry or WithQueryLogging.

Example Usage: 

gateway := NewQueryGateway[MyQuery, *MyResult](bus)
result, err := gateway(ctx, MyQuery{ID: "42"})

func NewQueryGateway 

func NewQueryGateway[T Query, R any](bus *QueryBus) QueryGateway[T, R]

NewQueryGateway returns a QueryGateway for query type T and result type R, backed by bus. It registers the (T, R) pair on bus as a requestee, so that a later call to QueryBus.Validate fails if no handler for that pair is ever registered.

Example Usage: 

listGateway := NewQueryGateway[ListTasks, *TaskList](bus)
findGateway := NewQueryGateway[ListTasks, *Task](bus)

func (QueryGateway[T, R]) HandleQuery added in v0.1.2 

func (g QueryGateway[T, R]) HandleQuery(ctx context.Context, qry T) (R, error)

HandleQuery implements QueryHandler by calling g.

type QueryHandler 

type QueryHandler[T Query, R any] interface {
	HandleQuery(ctx context.Context, qry T) (R, error)
}

QueryHandler handles queries of the concrete type T, which must implement Query, and produces a result of type R — typically a read model or an Iterator over one. It enables generic, type-safe registration and execution of query logic through a QueryBus.

Example Usage: 

type MyQuery struct { ID string }
type MyResult struct { Value int }

handler := NewQueryHandlerFunc(func(ctx context.Context, q MyQuery) (*MyResult, error) {
    return &MyResult{Value: 123}, nil
})

var _ QueryHandler[MyQuery, *MyResult] = handler

func NewQueryHandlerFunc 

func NewQueryHandlerFunc[T Query, R any](fn func(ctx context.Context, qry T) (R, error)) QueryHandler[T, R]

NewQueryHandlerFunc returns fn as a QueryHandler.

Example Usage: 

handler := NewQueryHandlerFunc(func(ctx context.Context, q MyQuery) (*MyResult, error) {
    return &MyResult{Value: 42}, nil
})

type QueryHandlerMiddleware added in v0.1.6 

type QueryHandlerMiddleware func(next QueryGateway[Query, any]) QueryGateway[Query, any]

QueryHandlerMiddleware decorates a query handler on a QueryBus. It receives the next handler in the chain and returns a handler that wraps it; call next to pass the query along, or return without calling it to short-circuit.

The query arrives as a Query, so qry.ID is available directly; use fmt.Sprintf("%T", qry) for the concrete type name. The result arrives as an any holding the concrete result value.

The chain is baked into a handler once, by RegisterQueryHandler, from the middlewares added via QueryBus.Use up to that moment. Add all middleware during startup wiring, before registering any handler. The first middleware passed to Use is the outermost wrapper and runs first on each query.

Example Usage: 

var myMiddleware eventsourcing.QueryHandlerMiddleware = func(
    next eventsourcing.QueryGateway[eventsourcing.Query, any],
) eventsourcing.QueryGateway[eventsourcing.Query, any] {
    return func(ctx context.Context, qry eventsourcing.Query) (any, error) {
        // before
        result, err := next(ctx, qry)
        // after
        return result, err
    }
}
bus.Use(myMiddleware)

type Revision 

type Revision uint64

Revision expects the stream to be at exactly this version.

func (Revision) String added in v0.1.8 

func (r Revision) String() string

func (Revision) ToRawInt64 

func (r Revision) ToRawInt64() int64

ToRawInt64 panics if r is above math.MaxInt64. The raw encoding reserves negative values for the markers Any and StreamExists returns, so a revision that large has no representation: int64(r) would sign-flip and be read as one of those markers. No stream reaches that many events, so a revision above it is a calculation that has gone wrong — a panic says so where a negative return would quietly change the caller's meaning.

type SkippedEventError added in v0.1.8 

type SkippedEventError struct {
	Event Event
}

SkippedEventError is returned when an EventHandler declines to process an Event because it is not the type the handler expects.

func (*SkippedEventError) Error added in v0.1.8 

func (e *SkippedEventError) Error() string

type StreamExists 

type StreamExists struct{}

StreamExists expects the stream to already exist.

func (StreamExists) String added in v0.1.8 

func (StreamExists) String() string

func (StreamExists) ToRawInt64 

func (StreamExists) ToRawInt64() int64

type StreamNamer 

type StreamNamer func(ctx context.Context, cmd Command) string

StreamNamer maps a Command to the name of the event stream it belongs to. 

var DefaultStreamNamer StreamNamer = func(ctx context.Context, cmd Command) string {
	return cmd.AggregateID()
}

DefaultStreamNamer is the StreamNamer used by command handlers when no custom namer is configured. By default it returns the command's Command.AggregateID as the stream name.

It can be overridden globally, for example to add tenant prefixes for multi-tenancy. Override it during program initialization, before any handler runs, to avoid data races on the global. 

// During startup:
DefaultStreamNamer = func(ctx context.Context, cmd Command) string {
	tenant, _ := ctx.Value(tenantKey).(string)
	return fmt.Sprintf("%s-orders-%s", tenant, cmd.AggregateID())
}

type StreamRevisionConflictError 

type StreamRevisionConflictError struct {
	Stream           string
	ExpectedRevision StreamState
	ActualRevision   StreamState
}

StreamRevisionConflictError is returned by an EventStore when a save is made against a Revision that no longer matches the stream's actual revision — an optimistic-concurrency conflict.

func (*StreamRevisionConflictError) Error 

func (s *StreamRevisionConflictError) Error() string

Error formats the conflict through each StreamState's own String method, so a store that reports a conflict without knowing the revisions involved — as the KurrentDB one does — renders them as <nil> instead of panicking.

type StreamState 

type StreamState interface {
	// ToRawInt64 encodes the expectation as a store-specific raw value: a
	// non-negative [Revision] number, or one of the special negative
	// markers used by [Any] and [StreamExists].
	ToRawInt64() int64
}

StreamState expresses a caller's expectation of a stream's revision, for example to an EventStore.Save call via WithStreamState. Any, NoStream, StreamExists, and Revision are the implementations recognized by this package's EventStore implementations.

type SubscriberOption 

type SubscriberOption func(cfg any)

SubscriberOption configures a subscription registered via EventBus.Subscribe. Available options are implementation-specific; see, for example, the memory package's WithFilterEvents.

Source Files

View all Source files

Directories

Path Synopsis
eventbus
file
kurrentdb
memory
postgres
eventstore
file
kurrentdb
memory
postgres

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.