eventhorizon

type Aggregate ¶

type Aggregate interface {
	// Entity provides the ID of the aggregate.
	Entity

	// AggregateType returns the type name of the aggregate.
	// AggregateType() string
	AggregateType() AggregateType

	// CommandHandler is used to handle commands.
	CommandHandler
}

Aggregate is an interface representing a versioned data entity created from events. It receives commands and generates events that are stored.

The aggregate is created/loaded and saved by the Repository inside the Dispatcher. A domain specific aggregate can either implement the full interface, or more commonly embed *AggregateBase to take care of the common methods.

func CreateAggregate(aggregateType AggregateType, id uuid.UUID) (Aggregate, error)

type AggregateError ¶ added in v0.9.0

type AggregateError struct {
	// Err is the error.
	Err error
}

AggregateError is an error caused in the aggregate when handling a command.

func (e *AggregateError) Cause() error

func (e *AggregateError) Error() string

func (e *AggregateError) Unwrap() error

type AggregateStore ¶

type AggregateStore interface {
	// Load loads the most recent version of an aggregate with a type and id.
	Load(context.Context, AggregateType, uuid.UUID) (Aggregate, error)

	// Save saves the uncommitted events for an aggregate.
	Save(context.Context, Aggregate) error
}

AggregateStore is responsible for loading and saving aggregates.

type AggregateStoreError ¶ added in v0.14.10

type AggregateStoreError struct {
	// Err is the error that happened when applying the event.
	Err error
	// Op is the operation for the error.
	Op AggregateStoreOperation
	// AggregateType of related operation.
	AggregateType AggregateType
	// AggregateID of related operation.
	AggregateID uuid.UUID
}

AggregateStoreError contains related info about errors in the store.

func (e *AggregateStoreError) Cause() error

func (e *AggregateStoreError) Error() string

func (e *AggregateStoreError) Unwrap() error

type AggregateStoreOperation ¶ added in v0.14.10

type AggregateStoreOperation string

AggregateStoreOperation is the operation done when an error happened.

type AggregateType ¶

type AggregateType string

AggregateType is the type of an aggregate.

func AggregateTypeFromContext(ctx context.Context) (AggregateType, bool)

func (at AggregateType) String() string

type Command ¶

type Command interface {
	// AggregateID returns the ID of the aggregate that the command should be
	// handled by.
	AggregateID() uuid.UUID

	// AggregateType returns the type of the aggregate that the command can be
	// handled by.
	AggregateType() AggregateType

	// CommandType returns the type of the command.
	CommandType() CommandType
}

Command is a domain command that is sent to a Dispatcher.

A command name should 1) be in present tense and 2) contain the intent (MoveCustomer vs CorrectCustomerAddress).

The command should contain all the data needed when handling it as fields. These fields can take an optional "eh" tag, which adds properties. For now only "optional" is a valid tag: `eh:"optional"`.

func CreateCommand(commandType CommandType) (Command, error)

type CommandCodec ¶ added in v0.15.0

type CommandCodec interface {
	// MarshalCommand marshals a command and the supported parts of context into bytes.
	MarshalCommand(context.Context, Command) ([]byte, error)
	// UnmarshalCommand unmarshals a command and supported parts of context from bytes.
	UnmarshalCommand(context.Context, []byte) (Command, context.Context, error)
}

CommandCodec is a codec for marshaling and unmarshaling commands to and from bytes.

type CommandFieldError ¶

type CommandFieldError struct {
	Field string
}

CommandFieldError is returned by Dispatch when a field is incorrect.

func (c *CommandFieldError) Error() string

type CommandHandler ¶

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

CommandHandler is an interface that all handlers of commands should implement.

func UseCommandHandlerMiddleware(h CommandHandler, middleware ...CommandHandlerMiddleware) CommandHandler

type CommandHandlerFunc ¶

type CommandHandlerFunc func(context.Context, Command) error

CommandHandlerFunc is a function that can be used as a command handler.

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

type CommandHandlerMiddleware ¶

type CommandHandlerMiddleware func(CommandHandler) CommandHandler

CommandHandlerMiddleware is a function that middlewares can implement to be able to chain.

type CommandIDer ¶ added in v0.9.0

type CommandIDer interface {
	// CommandID returns the ID of the command instance being handled.
	CommandID() uuid.UUID
}

CommandIDer provides a unique command ID to be used for request tracking etc.

type CommandType ¶

type CommandType string

CommandType is the type of a command, used as its unique identifier.

func CommandTypeFromContext(ctx context.Context) (CommandType, bool)

func (ct CommandType) String() string

type CompareConfig ¶ added in v0.11.0

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

CompareConfig is a config for the ComparEvents function.

type CompareOption ¶ added in v0.11.0

type CompareOption func(*CompareConfig)

CompareOption is an option setter used to configure comparing of events.

func IgnorePositionMetadata() CompareOption

func IgnoreTimestamp() CompareOption

func IgnoreVersion() CompareOption

type ContextMarshalFunc ¶

type ContextMarshalFunc func(context.Context, map[string]interface{})

ContextMarshalFunc is a function that marshalls any context values to a map, used for sending context on the wire.

type ContextUnmarshalFunc ¶

type ContextUnmarshalFunc func(context.Context, map[string]interface{}) context.Context

ContextUnmarshalFunc is a function that marshalls any context values to a map, used for sending context on the wire.

type Entity ¶

type Entity interface {
	// EntityID returns the ID of the entity.
	EntityID() uuid.UUID
}

Entity is an item which is identified by an ID.

From http://cqrs.nu/Faq: "Entities or reference types are characterized by having an identity that's not tied to their attribute values. All attributes in an entity can change and it's still "the same" entity. Conversely, two entities might be equivalent in all their attributes, but will still be distinct".

type Event ¶

type Event interface {
	// EventType returns the type of the event.
	EventType() EventType
	// The data attached to the event.
	Data() EventData
	// Timestamp of when the event was created.
	Timestamp() time.Time

	// AggregateType is the type of the aggregate that the event can be
	// applied to.
	AggregateType() AggregateType
	// AggregateID is the ID of the aggregate that the event belongs to.
	AggregateID() uuid.UUID
	// Version is the version of the aggregate after the event has been applied.
	Version() int

	// Metadata is app-specific metadata such as request ID, originating user etc.
	Metadata() map[string]interface{}

	// A string representation of the event.
	String() string
}

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

An event struct and type name should:

1. Be in past tense (CustomerMoved)
2. Contain the intent (CustomerMoved vs CustomerAddressCorrected).

The event should contain all the data needed when applying/handling it.

func NewEvent(eventType EventType, data EventData, timestamp time.Time, options ...EventOption) Event

func NewEventForAggregate(eventType EventType, data EventData, timestamp time.Time,
	aggregateType AggregateType, aggregateID uuid.UUID, version int, options ...EventOption) Event

type EventBus ¶

type EventBus interface {
	EventHandler

	// AddHandler adds a handler for an event. Returns an error if either the
	// matcher or handler is nil, the handler is already added or there was some
	// other problem adding the handler (for networked handlers for example).
	AddHandler(context.Context, EventMatcher, EventHandler) error

	// Errors returns an error channel where async handling errors are sent.
	Errors() <-chan error

	// Close closes the EventBus and waits for all handlers to finish.
	Close() error
}

EventBus is an EventHandler that distributes published events to all matching handlers that are registered, but only one of each type will handle the event. Events are not guaranteed to be handeled in order.

type EventBusError ¶ added in v0.5.0

type EventBusError struct {
	// Err is the error.
	Err error
	// Ctx is the context used when the error happened.
	Ctx context.Context
	// Event is the event handeled when the error happened.
	Event Event
}

EventBusError is an async error containing the error returned from a handler and the event that it happened on.

func (e *EventBusError) Cause() error

func (e *EventBusError) Error() string

func (e *EventBusError) Unwrap() error

type EventCodec ¶ added in v0.11.0

type EventCodec interface {
	// MarshalEvent marshals an event and the supported parts of context into bytes.
	MarshalEvent(context.Context, Event) ([]byte, error)
	// UnmarshalEvent unmarshals an event and supported parts of context from bytes.
	UnmarshalEvent(context.Context, []byte) (Event, context.Context, error)
}

EventCodec is a codec for marshaling and unmarshaling events to and from bytes.

type EventData ¶

type EventData interface{}

EventData is any additional data for an event.

func CreateEventData(eventType EventType) (EventData, error)

type EventHandler ¶

type EventHandler interface {
	// HandlerType is the type of the handler.
	HandlerType() EventHandlerType

	// HandleEvent handles an event.
	HandleEvent(context.Context, Event) error
}

EventHandler is a handler of events. If registered on a bus as a handler only one handler of the same type will receive each event. If registered on a bus as an observer all handlers of the same type will receive each event.

func UseEventHandlerMiddleware(h EventHandler, middleware ...EventHandlerMiddleware) EventHandler

type EventHandlerChain ¶ added in v0.16.0

type EventHandlerChain interface {
	InnerHandler() EventHandler
}

EventHandlerChain declares InnerHandler that returns the inner handler of a event handler middleware. This enables an endpoint or other middlewares to traverse the chain of handlers in order to find a specific middleware that can be interacted with.

For handlers who's intrinsic properties requires them to be the last responder of a chain, or can't produce an InnerHandler, a nil response can be implemented thereby hindering any further attempt to traverse the chain.

type EventHandlerError ¶ added in v0.14.10

type EventHandlerError struct {
	// Err is the error.
	Err error
	// Event is the event that failed to be handled.
	Event Event
}

EventHandlerError is an error returned when an event could not be handled by an event handler.

func (e *EventHandlerError) Cause() error

func (e *EventHandlerError) Error() string

func (e *EventHandlerError) Unwrap() error

type EventHandlerFunc ¶

type EventHandlerFunc func(context.Context, Event) error

EventHandlerFunc is a function that can be used as a event handler.

func (f EventHandlerFunc) HandleEvent(ctx context.Context, e Event) error

func (f EventHandlerFunc) HandlerType() EventHandlerType

type EventHandlerMiddleware ¶

type EventHandlerMiddleware func(EventHandler) EventHandler

EventHandlerMiddleware is a function that middlewares can implement to be able to chain.

type EventHandlerType ¶ added in v0.4.0

type EventHandlerType string

EventHandlerType is the type of an event handler, used as its unique identifier.

func (ht EventHandlerType) String() string

type EventMatcher ¶

type EventMatcher interface {
	// Match returns true if the matcher matches an event.
	Match(Event) bool
}

EventMatcher matches, for example on event types, aggregate types etc.

type EventOption ¶ added in v0.9.0

type EventOption func(Event)

EventOption is an option to use when creating events.

func ForAggregate(aggregateType AggregateType, aggregateID uuid.UUID, version int) EventOption

func FromCommand(cmd Command) EventOption

func WithGlobalPosition(position int) EventOption

func WithMetadata(metadata map[string]interface{}) EventOption

type EventSource ¶ added in v0.7.0

type EventSource interface {
	// UncommittedEvents returns events that are not committed to the event store,
	// or handeled in other ways (depending on the caller).
	UncommittedEvents() []Event
	// ClearUncommittedEvents clears uncommitted events, used after they have been
	// committed to the event store or handled in other ways (depending on the caller).
	ClearUncommittedEvents()
}

EventSource is a source of events, used for getting events for handling, storing, publishing etc. Mostly used in the aggregate stores.

type EventStore ¶

type EventStore interface {
	// Save appends all events in the event stream to the store.
	Save(ctx context.Context, events []Event, originalVersion int) error

	// Load loads all events for the aggregate id from the store.
	Load(context.Context, uuid.UUID) ([]Event, error)

	// LoadFrom loads all events from version for the aggregate id from the store.
	LoadFrom(ctx context.Context, id uuid.UUID, version int) ([]Event, error)

	// Close closes the EventStore.
	Close() error
}

EventStore is an interface for an event sourcing event store.

type EventStoreError ¶

type EventStoreError struct {
	// Err is the error.
	Err error
	// Op is the operation for the error.
	Op EventStoreOperation
	// AggregateType of related operation.
	AggregateType AggregateType
	// AggregateID of related operation.
	AggregateID uuid.UUID
	// AggregateVersion of related operation.
	AggregateVersion int
	// Events of the related operation.
	Events []Event
}

EventStoreError is an error in the event store.

func (e *EventStoreError) Cause() error

func (e *EventStoreError) Error() string

func (e *EventStoreError) Unwrap() error

type EventStoreMaintenance ¶ added in v0.13.0

type EventStoreMaintenance interface {
	// Replace an event, the version must match. Useful for maintenance actions.
	// Returns ErrAggregateNotFound if there is no aggregate.
	Replace(context.Context, Event) error

	// RenameEvent renames all instances of the event type.
	RenameEvent(ctx context.Context, from, to EventType) error
}

EventStoreMaintenance is an interface with maintenance tools for an EventStore. NOTE: Should not be used in apps, useful for migration tools etc.

type EventStoreOperation ¶ added in v0.14.10

type EventStoreOperation string

EventStoreOperation is the operation done when an error happened.

type EventType ¶

type EventType string

EventType is the type of an event, used as its unique identifier.

func (et EventType) String() string

type IsZeroer ¶ added in v0.14.4

type IsZeroer interface {
	IsZero() bool
}

IsZeroer is used to check if a type is zero-valued, and in that case is not allowed to be used in a command. See CheckCommand.

type Iter ¶

type Iter interface {
	Next(context.Context) bool
	Value() interface{}
	// Close must be called after the last Next() to retrieve error if any
	Close(context.Context) error
}

Iter is a stateful iterator object that when called Next() readies the next value that can be retrieved from Value(). Enables incremental object retrieval from repos that support it. You must call Close() on each Iter even when results were delivered without apparent error.

type MatchAggregates ¶ added in v0.8.0

type MatchAggregates []AggregateType

MatchAggregates matches any of the aggregate types, nil events never match.

func (types MatchAggregates) Match(e Event) bool

type MatchAll ¶ added in v0.8.0

type MatchAll []EventMatcher

MatchAll matches all of the matchers.

func (matchers MatchAll) Match(e Event) bool

type MatchAny ¶

type MatchAny []EventMatcher

MatchAny matches any of the matchers.

func (matchers MatchAny) Match(e Event) bool

type MatchEvents ¶ added in v0.8.0

type MatchEvents []EventType

MatchEvents matches any of the event types, nil events never match.

func (types MatchEvents) Match(e Event) bool

type Outbox ¶ added in v0.15.0

type Outbox interface {
	EventHandler

	// AddHandler adds a handler for an event. Returns an error if either the
	// matcher or handler is nil, the handler is already added or there was some
	// other problem adding the handler.
	AddHandler(context.Context, EventMatcher, EventHandler) error

	// Start starts processing the outbox until the Close() is cancelled by
	// handling all events using the added handlers. Should be called after all
	// handlers are setup but before new events are starting to be handled.
	Start()

	// Close closes the Outbox and waits for all handlers to finish.
	Close() error

	// Errors returns an error channel where async handling errors are sent.
	Errors() <-chan error
}

Outbox is an outbox for events. It ensures that all handled events get handled by all added handlers. Handling of events must be atomic to storing them for at least once handling of events, which is often provided by storage adapter using a transaction.

type OutboxError ¶ added in v0.15.0

type OutboxError struct {
	// Err is the error.
	Err error
	// Ctx is the context used when the error happened.
	Ctx context.Context
	// Event is the event handeled when the error happened.
	Event Event
}

OutboxError is an error in the outbox.

func (e *OutboxError) Cause() error

func (e *OutboxError) Error() string

func (e *OutboxError) Unwrap() error

type ReadRepo ¶

type ReadRepo interface {
	// InnerRepo returns the inner read repository, if there is one.
	// Useful for iterating a wrapped set of repositories to get a specific one.
	InnerRepo(context.Context) ReadRepo

	// Find returns an entity for an ID.
	Find(context.Context, uuid.UUID) (Entity, error)

	// FindAll returns all entities in the repository.
	FindAll(context.Context) ([]Entity, error)

	// Close closes the ReadRepo.
	Close() error
}

ReadRepo is a read repository for entities.

type ReadWriteRepo ¶

type ReadWriteRepo interface {
	ReadRepo
	WriteRepo
}

ReadWriteRepo is a combined read and write repo, mainly useful for testing.

type RepoError ¶

type RepoError struct {
	// Err is the error.
	Err error
	// Op is the operation for the error.
	Op RepoOperation
	// EntityID of related operation.
	EntityID uuid.UUID
}

RepoError is an error in the read repository.

func (e *RepoError) Cause() error

func (e *RepoError) Error() string

func (e *RepoError) Unwrap() error

type RepoOperation ¶ added in v0.14.10

type RepoOperation string

RepoOperation is the operation done when an error happened.

type Snapshot ¶ added in v0.16.0

type Snapshot struct {
	Version       int
	AggregateType AggregateType
	Timestamp     time.Time
	State         interface{}
}

Snapshot is a recording of the state of an aggregate at a point in time

type SnapshotData ¶ added in v0.16.0

type SnapshotData interface{}

func CreateSnapshotData(AggregateID uuid.UUID, aggregateType AggregateType) (SnapshotData, error)

type SnapshotStore ¶ added in v0.16.0

type SnapshotStore interface {
	LoadSnapshot(ctx context.Context, id uuid.UUID) (*Snapshot, error)
	SaveSnapshot(ctx context.Context, id uuid.UUID, snapshot Snapshot) error
}

SnapshotStore is an interface for snapshot store.

type SnapshotStrategy ¶ added in v0.16.0

type SnapshotStrategy interface {
	ShouldTakeSnapshot(lastSnapshotVersion int,
		lastSnapshotTimestamp time.Time,
		event Event) bool
}

SnapshotStrategy determines if a snapshot should be taken or not.

type Snapshotable ¶ added in v0.16.0

type Snapshotable interface {
	CreateSnapshot() *Snapshot
	ApplySnapshot(snapshot *Snapshot)
}

Snapshotable is an interface for creating and applying a Snapshot record.

type Versionable ¶

type Versionable interface {
	// AggregateVersion returns the version of the item.
	AggregateVersion() int
}

Versionable is an item that has a version number, used by version.ReadRepo and projector.EventHandler.

type WriteRepo ¶

type WriteRepo interface {
	// Save saves a entity in the storage.
	Save(context.Context, Entity) error

	// Remove removes a entity by ID from the storage.
	Remove(context.Context, uuid.UUID) error
}

WriteRepo is a write repository for entities.