event

package module
v0.0.0-...-0cb72f2 Latest Latest
Warning

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

 Go to latest Published: May 29, 2026 License: MIT Imports: 13 Imported by: 0

README

go-event

go-event is a small Go library for message-oriented services and event sourcing. It defines stable domain contracts for commands, queries, events, event-sourced aggregates, and transport-neutral subscriptions, with a NATS JetStream backend for production.

The library keeps the domain space intentionally small, but each concept has a specific role:

  • Command is a synchronous request to make a service do something. It is expected to complete quickly, usually within a request timeout configured by the message bus implementation.
  • Query[R] is like a command, but returns a typed result. Query handlers should read state and report an answer without committing domain changes.
  • Event records a committed aggregate state change. Events can also be published as integration events so other services can react asynchronously.
  • ESAggregate rebuilds state by applying the event types it explicitly declares.
  • Store appends and loads aggregate event streams with optimistic concurrency. For background on event sourcing, see Martin Fowler's Event Sourcing and Greg Young's Building an Event Storage.
  • Reader streams decoded aggregate history for audit and read-side inspection.
  • Bus lets services communicate by dispatching commands, invoking queries, and subscribing to commands, queries, and events. Event subscriptions have two common models: an event handler builds local projections from an ordered event stream, while a queue event handler behaves like an asynchronous command distributed among workers.

Install

go get github.com/dpotapov/go-event

Quick Tutorial: Order Processing Example

import (
    ...
	event "github.com/dpotapov/go-event"
	natsevt "github.com/dpotapov/go-event/nats"
	"github.com/nats-io/nats.go"
)

Start with small aggregate identity types. Embedding one in each message promotes the AggregateRef methods and keeps message definitions focused on payload fields.

type OrderID string

// Scope names the service or bounded context that owns this aggregate.
func (OrderID) AggregateScope() string { return "sales" }
func (OrderID) AggregateType() string  { return "order" }
func (id OrderID) AggregateID() string { return string(id) }

type WorkerID int

func (WorkerID) AggregateScope() string { return "ops" }
func (WorkerID) AggregateType() string  { return "worker" }
func (id WorkerID) AggregateID() string { return strconv.Itoa(int(id)) }

Define the command, domain event, and an event-like log message:

type OrderCreateCommand struct {
	OrderID    `json:"order_id"`
	CustomerID string `json:"customer_id"`
}

func (*OrderCreateCommand) CommandName() string { return "create" }

type OrderCreatedEvent struct {
	OrderID    `json:"order_id"`
	CustomerID string `json:"customer_id"`
}

func (*OrderCreatedEvent) EventName() string { return "created" }

type WorkerLogMessage struct {
	WorkerID `json:"worker_id"`

	Level   string    `json:"level"`
	Message string    `json:"message"`
	Time    time.Time `json:"time"`
}

// MessageKind is optional. Events default to "event", commands to "command",
// and queries to "query". Logs are event-like: they happened and should be
// recorded, not handled as calls to do something.
func (*WorkerLogMessage) MessageKind() string { return "log" }
func (e *WorkerLogMessage) EventName() string { return e.Level }

The worker subscribes to OrderCreateCommand and stores the resulting OrderCreatedEvent with optimistic concurrency through event.Execute.

func startOrderWorker(ctx context.Context, nc *nats.Conn) (*event.Bus, error) {
	es, err := natsevt.NewEventStore(nc, natsevt.EventStoreConfig{})
	if err != nil {
		return nil, fmt.Errorf("create event store: %w", err)
	}

	bus, err := natsevt.NewBus(nc, natsevt.BusConfig{
		Queue: "order-workers",
	})
	if err != nil {
		return nil, fmt.Errorf("create bus: %w", err)
	}
	event.HandleCommand(bus, orderCreateHandler(es, bus, WorkerID(1)))
	if err := bus.Connect(ctx); err != nil {
		return nil, fmt.Errorf("connect bus: %w", err)
	}
	return bus, nil
}

There are two setup rules in the example above:

  • Register handlers before calling Connect. The typed helper event.HandleCommand tells the bus both which subject to subscribe to and which Go type to decode when a message arrives.
  • Teach each event-sourced aggregate how to replay its history. event.Execute loads the current aggregate state before running the command handler, so the aggregate lists the event types it can apply with EventTypes.

Most services can start with natsevt.NewBus. Use event.NewBus only when you need lower-level transport configuration, such as custom JetStream consumer options.

The command handler mutates an event-sourced aggregate. After the order is created, it publishes a log message directly through the bus because logs do not need aggregate replay or optimistic concurrency checks. The newOrder factory gives event.Execute a fresh aggregate when it retries after a concurrent write.

type Order struct {
	OrderID
	Created bool
}

func (*Order) EventTypes() ([]event.Event, event.SnapshotEvent) {
	return []event.Event{&OrderCreatedEvent{}}, nil
}

func (o *Order) Apply(evt event.Event) {
	switch e := evt.(type) {
	case *OrderCreatedEvent:
		o.OrderID = e.OrderID
		o.Created = true
	}
}

func orderCreateHandler(store event.Store, publisher event.Publisher, workerID WorkerID) func(context.Context, *OrderCreateCommand) error {
	return func(ctx context.Context, cmd *OrderCreateCommand) error {
		newOrder := func() *Order {
			return &Order{OrderID: cmd.OrderID}
		}

		err := event.Execute(ctx, store,
			newOrder,
			func(ctx context.Context, order *Order, version, attempt uint64) ([]event.Event, error) {
				if order.Created {
					return nil, nil
				}
				return []event.Event{&OrderCreatedEvent{
					OrderID:    cmd.OrderID,
					CustomerID: cmd.CustomerID,
				}}, nil
			},
		)
		if err != nil {
			return fmt.Errorf("create order: %w", err)
		}
		return publisher.Publish(ctx, &WorkerLogMessage{
			WorkerID: workerID,
			Level:    "info",
			Message:  "order created",
			Time:     time.Now(),
		})
	}
}

Use Store.Save for event-sourced aggregate state. Use Bus.Publish for simple messages such as logs where there is no aggregate version to check.

NATS Subject Convention

The NATS backend uses one five-token subject shape:

<scope>.<kind>.<aggregate-type>.<aggregate-id>.<message-name>

Kinds are:

event
command
query

Examples:

billing.event.account.acct-123.created
billing.command.account.acct-123.rename
billing.query.account.acct-123.get

Concrete tokens must not contain dots, whitespace, *, >, or path separators. Stream subjects use one trailing wildcard:

<scope>.event.<aggregate-type>.>

The default stream name is:

<SCOPE>_<AGG>_EVENTS

Event Store CAS

The NATS event store uses JetStream per-subject optimistic concurrency. For an aggregate save it sets:

Nats-Expected-Last-Subject-Sequence
Nats-Expected-Last-Subject-Sequence-Subject

The expected subject is the aggregate filter:

<scope>.event.<aggregate-type>.<aggregate-id>.>

That means the returned aggregate version is the last JetStream stream sequence for that aggregate, not a simple count of aggregate events. This is deliberate: it lets different event names for the same aggregate share one CAS boundary.

When saving multiple events, the NATS backend uses Nats-Batch-Id, Nats-Batch-Sequence, and Nats-Batch-Commit so the batch commits atomically.

Event Store Snapshots

The NATS event store can write snapshot events to the same aggregate stream. A snapshot uses the normal event kind and defaults to this subject:

<scope>.event.<aggregate-type>.<aggregate-id>.snapshot

Enable automatic snapshot writes with EventStoreConfig.SnapshotEvery. When the number of aggregate events since the latest snapshot reaches SnapshotEvery, the store writes a new snapshot after a successful save. Snapshot write failures are logged because the domain events were already committed.

On load, the store reads the latest snapshot with JetStream direct get-last and then replays only events after that snapshot sequence. SnapshotEvery only controls writes; existing snapshots are used on load even when automatic writes are disabled.

There are two snapshot styles:

  • Implicit snapshots: aggregates do not implement Snapshottable; the store stores the aggregate value itself. On replay, snapshot bytes are decoded directly into the aggregate and Apply is skipped for that snapshot message. This works best with exported JSON fields or an aggregate-level NATS codec.
  • Explicit snapshots: aggregates implement Snapshottable and return a domain snapshot event from TakeSnapshot. EventTypes returns that snapshot prototype as its second return value, and Apply should restore state idempotently from it.
type OrderSnapshot struct {
	event.SnapshotEventBase

	OrderID string `json:"order_id"`
	Created bool   `json:"created"`
}

func (*Order) EventTypes() ([]event.Event, event.SnapshotEvent) {
	return []event.Event{&OrderCreatedEvent{}}, &OrderSnapshot{}
}

func (o *Order) TakeSnapshot() (event.SnapshotEvent, error) {
	return &OrderSnapshot{OrderID: o.OrderID, Created: o.Created}, nil
}

Embed event.SnapshotEventBase to use the default snapshot event name, or override EventName on the snapshot type to choose a different final subject token.

Reading Aggregate History

Use Store.Load when you need to rebuild aggregate state for command handling. Use Reader.Stream when you need decoded events for audit, debug, or read-side inspection without mutating an aggregate.

The NATS event store implements both event.Store and event.Reader:

for rec, err := range store.Stream(ctx, &Order{OrderID: id}, event.ReadOptions{}) {
	if err != nil {
		return fmt.Errorf("stream order history: %w", err)
	}
	// rec.Event, rec.Subject, rec.Data, rec.Version, rec.Timestamp
}

When the reference passed to Stream also implements ESAggregate, events are decoded into the concrete types declared by EventTypes, matching the load replay contract. When the reference is only an AggregateRef, the stream yields all matching aggregate messages as event.GenericEvent values and leaves the stored payload in StoredEvent.Data as opaque bytes:

ref := event.AggregateKey{Scope: "sales", Type: "order", ID: string(id)}
for rec, err := range store.Stream(ctx, ref, event.ReadOptions{}) {
	if err != nil {
		return fmt.Errorf("stream raw order history: %w", err)
	}
	// rec.Event.EventName() names the event; rec.Data contains transport bytes.
}

ReadOptions controls replay:

  • SkipSnapshotFastPath — when false (default), streaming starts after the latest snapshot for event-sourced aggregates, matching the replay window Load uses.
  • IncludeSnapshots — when false (default), snapshot subject messages are omitted for event-sourced aggregates. Set true to include explicit snapshot events.

Helpers:

events, err := event.Collect(ctx, store, agg, event.ReadOptions{})
rec, ok, err := event.FindFirst(ctx, store, agg, event.ReadOptions{}, func(evt event.Event) bool {
	return evt.EventName() == "created"
})

By default, the read path uses an ephemeral JetStream ordered consumer filtered to one aggregate. When running with scoped NATS permissions, configure EventStoreConfig.ConsumerName to return a deterministic consumer name for the aggregate:

store, err := natsevt.NewEventStore(nc, natsevt.EventStoreConfig{
	ConsumerName: func(ref event.AggregateRef) string {
		return "orders_" + ref.AggregateID()
	},
})

The same consumer name is used for Load, snapshot counting during Save, and Stream. Calls for the same aggregate/name should be sequential because the store deletes any stale named consumer before use and deletes it again after use.

Stream Creation

The library owns the subject convention, but your application owns stream creation. Use SetStreamSubjects to put the right subject filter on any JetStream stream config:

cfg := jetstream.StreamConfig{
	Name:               "BILLING_ACCOUNT_EVENTS",
	Storage:            jetstream.FileStorage,
	AllowAtomicPublish: true,
}
natsevt.SetStreamSubjects(&cfg, OrderID(""))
stream, err := js.CreateOrUpdateStream(ctx, cfg)

Handlers are registered before Connect. Ordered event handlers are grouped by aggregate type behind handler-derived ordered subscription filters and catch up before command and queue handlers start; queue handlers use precise durable consumers.

Queue Consumer Strategy

In the NATS backend, queue handlers on the same event stream share one durable consumer named after the queue. For queue projection on the test/account stream, the durable is projection with filters such as test.event.account.*.created and test.log.account.acct-1.info.

Documentation

Overview

Package event provides small CQRS, event sourcing, and message-bus abstractions for Go services.

Index

Constants

View Source 
const DefaultSnapshotEventName = "snapshot"

Variables

View Source 
var (
	ErrConflict          = errors.New("event version conflict")
	ErrNoResponders      = errors.New("no responders available")
	ErrInvalidResultType = errors.New("invalid query result type")
	ErrClosed            = errors.New("bus is closed")
)
View Source 
var ErrInvalidName = errors.New("invalid message name")

Functions

func Ask 

func Ask[R any](ctx context.Context, dispatcher Dispatcher, query Query[R]) (R, error)

func Execute 

func Execute[A ESAggregate](
	ctx context.Context,
	store Store,
	newAggregate func() A,
	mutate func(context.Context, A, uint64, uint64) ([]Event, error),
	opts ...ExecuteOption,
) error

Execute loads a fresh aggregate, runs mutate, and saves the produced events with optimistic concurrency. newAggregate is called for every retry so a failed attempt never leaks partially applied state into the next attempt.

func HandleCommand 

func HandleCommand[T Command](b *Bus, handler func(context.Context, T) error, opts ...HandlerOption)

HandleCommand registers a command handler for T. It must be called before Connect; it panics if the bus is already connected or closed, or if T cannot be registered as a message type.

func HandleEvent 

func HandleEvent[T Event](b *Bus, handler func(context.Context, T) error, opts ...HandlerOption)

HandleEvent registers an ordered event handler for T. It must be called before Connect; it panics if the bus is already connected or closed, or if T cannot be registered as a message type.

func HandleQuery 

func HandleQuery[T Command, R any](b *Bus, handler func(context.Context, T) (R, error))

HandleQuery registers a query handler for T. It must be called before Connect; it panics if the bus is already connected or closed, or if T cannot be registered as a message type.

func HandleQueueEvent 

func HandleQueueEvent[T Event](b *Bus, handler func(context.Context, T) error, opts ...HandlerOption)

HandleQueueEvent registers a queue event handler for T. It must be called before Connect; it panics if the bus is already connected or closed, if T cannot be registered as a message type, or if the bus has no queue configured.

func MustRegisterMessageType 

func MustRegisterMessageType[T any](target TypeRegistry)

func RegisterMessageType 

func RegisterMessageType[T any](target TypeRegistry) error

func ValidateToken 

func ValidateToken(label, token string) error

ValidateToken checks one concrete subject token. Wildcards are intentionally rejected here; subscription filters use separate helpers in transport packages.

func WithMessageMetadata 

func WithMessageMetadata(ctx context.Context, metadata MessageMetadata) context.Context

Types

type AggregateKey 

type AggregateKey struct {
	Scope string
	Type  string
	ID    string
}

AggregateKey is a concrete aggregate reference for callers that only need to identify an aggregate, not replay its state.

func (AggregateKey) AggregateID 

func (k AggregateKey) AggregateID() string

func (AggregateKey) AggregateScope 

func (k AggregateKey) AggregateScope() string

func (AggregateKey) AggregateType 

func (k AggregateKey) AggregateType() string

type AggregateRef 

type AggregateRef interface {
	AggregateScope() string
	AggregateType() string
	AggregateID() string
}

AggregateRef identifies one aggregate instance. "Ref" means reference: these methods name the aggregate without carrying its event-sourced state or behavior.

type Bus 

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

func NewBus 

func NewBus(dispatcher Dispatcher, publisher Publisher, commands CommandSubscriber, events EventSubscriber, opts BusOptions) *Bus

func (*Bus) Connect 

func (b *Bus) Connect(ctx context.Context) error

func (*Bus) Disconnect 

func (b *Bus) Disconnect() error

func (*Bus) DispatchCommand 

func (b *Bus) DispatchCommand(ctx context.Context, cmd Command) error

func (*Bus) DispatchQuery 

func (b *Bus) DispatchQuery(ctx context.Context, query Command, result any) error

func (*Bus) Done 

func (b *Bus) Done() <-chan error

func (*Bus) Publish 

func (b *Bus) Publish(ctx context.Context, events ...Event) error

func (*Bus) RegisterMessageType 

func (b *Bus) RegisterMessageType(prototype any) error

type BusOptions 

type BusOptions struct {
	Queue                    string
	QueueMaxRetries          int
	Logger                   *slog.Logger
	CursorStore              CursorStore
	CursorBootPolicy         CursorBootPolicy
	OnSubscriptionError      func(*SubscriptionError) bool
	CommandConflictRetries   int
	ExcludeCommandDebugNames map[string]bool
}

type Command 

type Command interface {
	AggregateRef
	CommandName() string
}

Command describes an action requested against one aggregate instance.

type CommandHandler 

type CommandHandler interface {
	HandleCommand(context.Context, Command) error
}

type CommandHandlerFunc 

type CommandHandlerFunc func(context.Context, Command) error

func (CommandHandlerFunc) HandleCommand 

func (f CommandHandlerFunc) HandleCommand(ctx context.Context, cmd Command) error

type CommandSubscriber 

type CommandSubscriber interface {
	SubscribeCommand(context.Context, CommandHandler, CommandSubscriptionConfig) (Subscription, error)
	SubscribeQuery(context.Context, QueryHandler, CommandSubscriptionConfig) (Subscription, error)
}

type CommandSubscriptionConfig 

type CommandSubscriptionConfig struct {
	AggregateScope string
	Kind           MessageKind
	AggregateTypes []string
	AggregateIDs   []string
	CommandNames   []string
	Queue          string
}

func (CommandSubscriptionConfig) Validate 

func (c CommandSubscriptionConfig) Validate() error

type ConflictError 

type ConflictError struct {
	Expected uint64
	Actual   uint64
}

func (*ConflictError) Error 

func (e *ConflictError) Error() string

func (*ConflictError) Is 

func (e *ConflictError) Is(target error) bool

type CursorBootPolicy 

type CursorBootPolicy int
const (
	CursorBootNew CursorBootPolicy = iota
	CursorBootAll
)

type CursorStore 

type CursorStore interface {
	LoadCursor(context.Context, string) ([]byte, error)
	SaveCursor(context.Context, string, []byte) error
}

type DecoderRegistry 

type DecoderRegistry interface {
	TypeRegistry
	NewEvent(kind MessageKind, scope, aggregateType, name string) (Event, error)
	DecodeEvent(Subject, []byte) (Event, error)
	NewCommand(kind MessageKind, scope, aggregateType, name string) (Command, error)
	DecodeCommand(Subject, []byte) (Command, error)
}

func NewTypeRegistry 

func NewTypeRegistry() DecoderRegistry

type Dispatcher 

type Dispatcher interface {
	DispatchCommand(context.Context, Command) error
	DispatchQuery(context.Context, Command, any) error
}

type ESAggregate 

type ESAggregate interface {
	AggregateRef
	EventTypes() ([]Event, SnapshotEvent)
	Apply(Event)
}

ESAggregate is an event-sourced entity that can rebuild its state from events.

type Event 

type Event interface {
	AggregateRef
	EventName() string
}

Event describes a committed state change for one aggregate instance.

type EventHandler 

type EventHandler interface {
	HandleEvent(context.Context, Event, []byte) error
}

type EventHandlerFunc 

type EventHandlerFunc func(context.Context, Event, []byte) error

func (EventHandlerFunc) HandleEvent 

func (f EventHandlerFunc) HandleEvent(ctx context.Context, evt Event, cursor []byte) error

type EventSubscriber 

type EventSubscriber interface {
	SubscribeEvents(context.Context, EventHandler, EventSubscriptionConfig) (Subscription, error)
}

type EventSubscriptionConfig 

type EventSubscriptionConfig struct {
	AggregateScope string
	Kind           MessageKind
	AggregateTypes []string
	AggregateIDs   []string
	EventNames     []string
	EventFilters   []EventSubscriptionFilter

	Queue        string
	ConsumerName string
	MaxRetries   int

	Cursor           []byte
	CursorBootPolicy CursorBootPolicy
	CaughtUp         func()
	OnError          func(*SubscriptionError) bool
}

func (EventSubscriptionConfig) Validate 

func (c EventSubscriptionConfig) Validate() error

type EventSubscriptionFilter 

type EventSubscriptionFilter struct {
	Kind         MessageKind
	AggregateIDs []string
	EventNames   []string
}

type ExecuteConfig 

type ExecuteConfig struct {
	MaxAttempts int
	BaseDelay   time.Duration
	MaxDelay    time.Duration
}

type ExecuteOption 

type ExecuteOption func(*ExecuteConfig)

func WithExecuteBackoff 

func WithExecuteBackoff(baseDelay, maxDelay time.Duration) ExecuteOption

func WithExecuteMaxAttempts 

func WithExecuteMaxAttempts(n int) ExecuteOption

type GenericEvent 

type GenericEvent struct {
	Subject Subject
}

GenericEvent identifies a stored event when the caller did not provide an event-sourced aggregate with concrete event types.

func (GenericEvent) AggregateID 

func (e GenericEvent) AggregateID() string

func (GenericEvent) AggregateScope 

func (e GenericEvent) AggregateScope() string

func (GenericEvent) AggregateType 

func (e GenericEvent) AggregateType() string

func (GenericEvent) EventName 

func (e GenericEvent) EventName() string

type HandlerOption 

type HandlerOption func(*handlerOptions)

func WithAggregateIDs 

func WithAggregateIDs(ids ...string) HandlerOption

type MemoryCursorStore 

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

func NewMemoryCursorStore 

func NewMemoryCursorStore() *MemoryCursorStore

func (*MemoryCursorStore) LoadCursor 

func (s *MemoryCursorStore) LoadCursor(ctx context.Context, stream string) ([]byte, error)

func (*MemoryCursorStore) SaveCursor 

func (s *MemoryCursorStore) SaveCursor(ctx context.Context, stream string, cursor []byte) error

type MessageKind 

type MessageKind string
const (
	KindEvent   MessageKind = "event"
	KindCommand MessageKind = "command"
	KindQuery   MessageKind = "query"
)

type MessageKindOverride 

type MessageKindOverride interface {
	MessageKind() string
}

MessageKindOverride optionally overrides the default kind token used for a message. Empty values are ignored and the call site default is used.

type MessageMetadata 

type MessageMetadata struct {
	NumDelivered uint64
	Timestamp    time.Time
}

func MessageMetadataFromContext 

func MessageMetadataFromContext(ctx context.Context) (MessageMetadata, bool)

type Publisher 

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

type Query 

type Query[R any] interface {
	Command
	ResultType() R
}

Query describes a read-only request. ResultType is only used for generic type inference and should return the zero value of R.

type QueryHandler 

type QueryHandler interface {
	HandleQuery(context.Context, Command) (any, error)
}

type QueryHandlerFunc 

type QueryHandlerFunc func(context.Context, Command) (any, error)

func (QueryHandlerFunc) HandleQuery 

func (f QueryHandlerFunc) HandleQuery(ctx context.Context, query Command) (any, error)

type ReadOptions 

type ReadOptions struct {
	// SkipSnapshotFastPath disables Load-style "start after latest snapshot"
	// optimization for event-sourced aggregates. When false (default), typed
	// Stream calls start at snapshotSeq+1.
	SkipSnapshotFastPath bool

	// IncludeSnapshots controls whether snapshot subject messages are yielded
	// for event-sourced aggregates. Default false: domain events only.
	IncludeSnapshots bool
}

ReadOptions configures aggregate event streaming.

type Reader 

type Reader interface {
	Stream(ctx context.Context, ref AggregateRef, opts ReadOptions) iter.Seq2[StoredEvent, error]
}

Reader streams aggregate history without mutating aggregate state. If ref also implements ESAggregate, events are decoded into the concrete EventTypes. Plain AggregateRef values stream generic events with opaque Data bytes.

type ResponseError 

type ResponseError json.RawMessage

ResponseError represents a structured error received in a CQRS response. It contains the raw JSON payload which can be interpreted by higher layers.

func (ResponseError) Error 

func (e ResponseError) Error() string

Error implements the error interface.

func (ResponseError) MarshalJSON 

func (e ResponseError) MarshalJSON() ([]byte, error)

MarshalJSON implements json.Marshaler, allowing the error to be re-serialized.

func (ResponseError) UnmarshalInto 

func (e ResponseError) UnmarshalInto(v any) error

UnmarshalInto allows callers to extract the underlying JSON into a typed struct.

type SnapshotEvent 

type SnapshotEvent interface {
	Event
	SnapshotEvent()
}

type SnapshotEventBase 

type SnapshotEventBase struct{}

func (SnapshotEventBase) EventName 

func (SnapshotEventBase) EventName() string

func (SnapshotEventBase) SnapshotEvent 

func (SnapshotEventBase) SnapshotEvent()

type Snapshottable 

type Snapshottable interface {
	ESAggregate
	TakeSnapshot() (SnapshotEvent, error)
}

Snapshottable aggregates can provide an explicit snapshot event.

type Store 

type Store interface {
	Save(ctx context.Context, ref AggregateRef, expectedVersion uint64, events ...Event) error
	Load(ctx context.Context, agg ESAggregate) (version uint64, err error)
}

Store appends and loads aggregate event streams.

type StoredEvent 

type StoredEvent struct {
	Event     Event
	Subject   Subject
	Data      []byte
	Version   uint64 // store sequence for this message
	Timestamp time.Time
}

StoredEvent is one aggregate message read from the event store.

func Collect 

func Collect(ctx context.Context, r Reader, ref AggregateRef, opts ReadOptions) ([]StoredEvent, error)

Collect reads all events from a stream into a slice.

func FindFirst 

func FindFirst(ctx context.Context, r Reader, ref AggregateRef, opts ReadOptions, pred func(Event) bool) (StoredEvent, bool, error)

FindFirst returns the first stored event matching pred.

type Subject 

type Subject struct {
	Scope         string
	Kind          MessageKind
	AggregateType string
	AggregateID   string
	Name          string
}

Subject is the transport-neutral delivery subject of a command, query, or event.

func CommandSubject 

func CommandSubject(cmd Command) Subject

func EventSubject 

func EventSubject(evt Event) Subject

func QuerySubject 

func QuerySubject(query Command) Subject

func (Subject) String 

func (s Subject) String() string

func (Subject) Validate 

func (s Subject) Validate() error

type Subscription 

type Subscription interface {
	Stop() error
}

type SubscriptionError 

type SubscriptionError struct {
	Subject          Subject
	Queue            string
	NumDelivered     uint64
	RetriesExhausted bool
	Transport        bool
	Err              error
}

func (*SubscriptionError) Error 

func (e *SubscriptionError) Error() string

func (*SubscriptionError) Unwrap 

func (e *SubscriptionError) Unwrap() error

type SubscriptionFunc 

type SubscriptionFunc func() error

func (SubscriptionFunc) Stop 

func (f SubscriptionFunc) Stop() error

type TypeRegistry 

type TypeRegistry interface {
	RegisterMessageType(any) error
}

TypeRegistry is the public typed-registration target implemented by Bus and transport adapters that need to decode stored or incoming messages.

type UnknownCommandError 

type UnknownCommandError struct {
	Scope         string
	AggregateType string
	Name          string
}

func (*UnknownCommandError) Error 

func (e *UnknownCommandError) Error() string

func (*UnknownCommandError) Is 

func (e *UnknownCommandError) Is(target error) bool

type UnknownEventError 

type UnknownEventError struct {
	Scope         string
	AggregateType string
	Name          string
}

func (*UnknownEventError) Error 

func (e *UnknownEventError) Error() string

func (*UnknownEventError) Is 

func (e *UnknownEventError) Is(target error) bool

Source Files

View all Source files

Directories

Path Synopsis
natstest

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.