eventsourcing

package module

Go to main page

Versions in this module

v0

v0.1.7

Jul 30, 2026

Changes in this version

+ func NewBusinessRuleViolation(err error) error

v0.1.6

Jun 11, 2026

Changes in this version

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

type CommandBus

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

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

type EventBus

+ Use func(middlewares ...EventHandlerMiddleware)

+ type EventHandlerMiddleware func(next EventHandler) EventHandler

+ type EventStoreMiddleware func(next EventStore) EventStore

type QueryBus

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

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

v0.1.5

May 12, 2026

v0.1.4

May 7, 2026

v0.1.3

May 1, 2026

v0.1.2

Apr 29, 2026

Changes in this version

+ type QueryGateway func(ctx context.Context, qry T) (R, error)

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

v0.1.1

Apr 13, 2026

v0.1.0

Apr 8, 2026

Changes in this version

+ const InstrumentationVersion

+ var ErrCommandBusClosed = errors.New("command bus is closed")

+ var ErrDuplicateHandler = errors.New("duplicate handler registered ")

+ var ErrEventNotRegistered = errors.New("event not registered")

+ var ErrHandlerNotFound = errors.New("handler not registered")

+ var ErrHandlerNotRegistered = errors.New("no handler registered for type")

+ var ErrHandlerPanicked = errors.New("handler panicked when handling command")

+ var ErrInvalidEventBatch = errors.New("invalid event batch")

+ var ErrInvalidRevision = errors.New("invalid revision")

+ var ErrStreamExists = errors.New("stream already exists")

+ var ErrStreamNotFound = errors.New("stream not found")

+ var EventNamesFor func(event Event) []string = func(event Event) []string

+ var NewEventByName func(name string) (Event, error) = newEventByNameDefault

+ var RegisterEvent func(event Event) = func(event Event)

+ var RegisterEventByName func(name string, fn func() Event) = func(name string, fn func() Event)

+ var RegisterEventByType func(fn func() Event) = func(fn func() Event)

+ func AggregateIDFromContext(ctx context.Context) string

+ func CausationFromContext(ctx context.Context) string

+ func EventIDFromContext(ctx context.Context) uuid.UUID

+ func GlobalVersionFromContext(ctx context.Context) uint64

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

+ func OccurredAtFromContext(ctx context.Context) time.Time

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

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

+ func StreamIDFromContext(ctx context.Context) string

+ func TypeName[T any](t T) string

+ func VersionFromContext(ctx context.Context) uint64

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

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

+ type Any struct

+ func (Any) ToRawInt64() int64

+ type AppendResult struct

+ NextExpectedVersion uint64

+ StreamID string

+ Successful bool

+ type Command interface

+ AggregateID func() string

+ type CommandBus struct

+ func NewCommandBus(bufferSize int, shardCount int) *CommandBus

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

+ func (b *CommandBus) Stop()

+ type CommandHandler func(ctx context.Context, command C) (AppendResult, error)

+ func NewCommandHandler[T any, C Command](store EventStore, initialState InitialState[T], evolve Evolver[T], ...) CommandHandler[C]

+ type CommandHandlerOption func(configuration *handlerOptions)

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

+ func WithRetryStrategy(strategy backoff.BackOff) CommandHandlerOption

+ func WithStreamNamer(namer StreamNamer) CommandHandlerOption

+ func WithStreamState(rev StreamState) CommandHandlerOption

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

+ type Dispatcher interface

+ Dispatch func(ctx context.Context, cmd Command) (AppendResult, error)

+ type Envelope struct

+ Event Event

+ EventID uuid.UUID

+ GlobalVersion uint64

+ Metadata map[string]any

+ OccurredAt time.Time

+ StreamID string

+ Version uint64

+ type ErrBusinessRuleViolation struct

+ Err error

+ func (e ErrBusinessRuleViolation) Cause() error

+ func (e ErrBusinessRuleViolation) Error() string

+ func (e ErrBusinessRuleViolation) Unwrap() error

+ type ErrSkippedEvent struct

+ Event Event

+ func (e ErrSkippedEvent) Error() string

+ type Event interface

+ AggregateID func() string

+ EventType func() string

+ type EventBus interface

+ Close func() error

+ Errors func() <-chan error

+ Subscribe func(ctx context.Context, name string, handler EventHandler, ...) error

+ type EventGroupProcessor struct

+ func NewEventGroupProcessor(handlers ...EventHandler) *EventGroupProcessor

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

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

+ type EventHandler interface

+ Handle func(ctx context.Context, event Event) error

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

+ func OnEvent[T Event](fn func(ctx context.Context, ev T) error) EventHandler

+ type EventStore interface

+ Close func() error

+ LoadFromAll func(ctx context.Context, version StreamState) (*Iterator[*Envelope], error)

+ LoadStream func(ctx context.Context, id string) (*Iterator[*Envelope], error)

+ LoadStreamFrom func(ctx context.Context, id string, version StreamState) (*Iterator[*Envelope], error)

+ Save func(ctx context.Context, events []Envelope, revision StreamState) (AppendResult, error)

+ type Evolver func(currentState T, envelope *Envelope) T

+ type GenericQueryGateway struct

+ func NewQueryGateway[T Query, R any | Iterator[any]](bus *QueryBus) GenericQueryGateway[T, R]

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

+ type HandlerOption func(*handlerSettings)

+ type InitialState func() T

+ type IterFunc func(ctx context.Context) (T, error)

+ type Iterator struct

+ func NewIteratorFunc[T any](nextFunc func(ctx context.Context) (T, error)) *Iterator[T]

+ func NewSliceIterator[T any](slice []T) *Iterator[T]

+ func (it *Iterator[T]) All(ctx context.Context) ([]T, error)

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

+ func (it *Iterator[T]) Next(ctx context.Context) bool

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

+ type NoStream struct

+ func (NoStream) ToRawInt64() int64

+ type Query interface

+ ID func() []byte

+ type QueryBus struct

+ func NewQueryBus() *QueryBus

+ func (q QueryBus) Validate() error

+ type QueryHandler interface

+ HandleQuery func(ctx context.Context, qry T) (R, error)

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

+ type Revision uint64

+ func (r Revision) ToRawInt64() int64

+ type StreamExists struct

+ func (StreamExists) ToRawInt64() int64

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

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

+ type StreamRevisionConflictError struct

+ ActualRevision StreamState

+ ExpectedRevision StreamState

+ Stream string

+ func (s StreamRevisionConflictError) Error() string

+ type StreamState interface

+ ToRawInt64 func() int64

+ type SubscriberOption func(cfg any)

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL