eventhorizon

type Aggregate ¶

type Aggregate interface {
	// AggregateType returns the type name of the aggregate.
	// AggregateType() string
	AggregateType() AggregateType
	// AggregateID returns the id of the aggregate.
	AggregateID() UUID

	// Version returns the version of the aggregate.
	Version() int
	// Increment version increments the version of the aggregate. It should be
	// called after an event has been successfully applied.
	IncrementVersion()

	// CommandHandler is used to handle commands.
	CommandHandler

	// StoreEvent creates and stores a new event as uncommitted for the aggregate.
	StoreEvent(EventType, EventData) Event
	// UncommittedEvents gets all uncommitted events to commit to the store.
	UncommittedEvents() []Event
	// ClearUncommittedEvents clears all uncommitted events after committing to
	// the store.
	ClearUncommittedEvents()

	// ApplyEvent applies an event on the aggregate by setting its values.
	// If there are no errors the version shoudl be incremented by calling
	// IncrementVersion.
	ApplyEvent(context.Context, Event) error
}

Aggregate is an interface representing a versioned data entity created from events. It receives commands and generates evens 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) (Aggregate, error)

type AggregateBase ¶

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

AggregateBase is a CQRS aggregate base to embed in domain specific aggregates.

A typical aggregate example:

type UserAggregate struct {
    *eventhorizon.AggregateBase

    name string
}

Using a new function to create aggregates and setting up the aggregate base is recommended:

func NewUserAggregate(id eh.UUID) *InvitationAggregate {
    return &UserAggregate{
        AggregateBase: eh.NewAggregateBase(UserAggregateType, id),
    }
}

The aggregate must also be registered, in this case:

func init() {
    eh.RegisterAggregate(func(id eh.UUID) eh.Aggregate {
        return NewUserAggregate(id)
    })
}

The aggregate must return an error if the event can not be applied, or nil to signal success which will increment the version.

func (a *Aggregate) ApplyEvent(event Event) error {
    switch event.EventType() {
    case AddUserEvent:
        // Apply the event data to the aggregate.
    }
}

See the examples folder for a complete use case.

func NewAggregateBase(aggregateType AggregateType, id UUID) *AggregateBase

func (a *AggregateBase) AggregateID() UUID

func (a *AggregateBase) AggregateType() AggregateType

func (a *AggregateBase) ClearUncommittedEvents()

func (a *AggregateBase) IncrementVersion()

func (a *AggregateBase) StoreEvent(eventType EventType, data EventData) Event

func (a *AggregateBase) UncommittedEvents() []Event

func (a *AggregateBase) Version() int

type AggregateCommandHandler ¶

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

AggregateCommandHandler dispatches commands to registered aggregates.

The dispatch process is as follows: 1. The handler receives a command 2. An aggregate is created or rebuilt from previous events by the repository 3. The aggregate's command handler is called 4. The aggregate stores events in response to the command 5. The new events are stored in the event store by the repository 6. The events are published to the event bus when stored by the event store

func NewAggregateCommandHandler(repository Repository) (*AggregateCommandHandler, error)

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

func (h *AggregateCommandHandler) SetAggregate(aggregateType AggregateType, cmdType CommandType) error

type AggregateType ¶

type AggregateType string

AggregateType is the type of an aggregate.

type ApplyEventError ¶

type ApplyEventError struct {
	// Event is the event that caused the error.
	Event Event
	// Err is the error that happened when applying the event.
	Err error
}

ApplyEventError is when an event could not be applied. It contains the error and the event that caused it.

func (a ApplyEventError) Error() string

type Command ¶

type Command interface {
	// AggregateID returns the ID of the aggregate that the command should be
	// handled by.
	AggregateID() 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 CommandBus ¶

type CommandBus interface {
	CommandHandler

	// SetHandler registers a handler for a command.
	SetHandler(CommandHandler, CommandType) error
}

CommandBus is a command handler that delegates command handling to other handlers based on the command type.

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.

type CommandType ¶

type CommandType string

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

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 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 returns the type of the aggregate that the event can be
	// applied to.
	AggregateType() AggregateType
	// AggregateID returns the ID of the aggregate that the event should be
	// applied to.
	AggregateID() UUID
	// Version of the aggregate for this event (after it has been applied).
	Version() int

	// 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) Event

func NewEventForAggregate(eventType EventType, data EventData,
	aggregateType AggregateType, aggregateID UUID, version int) Event

type EventBus ¶

type EventBus interface {
	EventHandler

	// AddHandler adds a handler for an event.
	AddHandler(EventHandler, EventType)

	// SetPublisher sets the publisher to use for publishing the event after all
	// handlers have been run.
	SetPublisher(EventPublisher)

	// SetHandlingStrategy will set the strategy to use for handling events.
	SetHandlingStrategy(EventHandlingStrategy)
}

EventBus is an event handler that handles events with the correct subhandlers after which it publishes the event using the publisher.

type EventData ¶

type EventData interface{}

EventData is any additional data for an event.

func CreateEventData(eventType EventType) (EventData, error)

type EventHandler ¶

type EventHandler interface {
	// HandleEvent handles an event.
	HandleEvent(context.Context, Event) error

	// HandlerType returns the type of the handler.
	HandlerType() EventHandlerType
}

EventHandler is a handler of events. Only one handler of the same type will receive an event.

type EventHandlerType ¶

type EventHandlerType string

EventHandlerType is the type of an event handler. Used to serve only handle an event by one handler of each type.

type EventHandlingStrategy ¶

type EventHandlingStrategy int

EventHandlingStrategy is the strategy to use when handling events.

const (
	// SimpleEventHandlingStrategy will handle events in the same goroutine and
	// wait for them to complete before handling the next.
	SimpleEventHandlingStrategy EventHandlingStrategy = iota
	// AsyncEventHandlingStrategy will handle events concurrently in their own
	// goroutines and not wait for them to finish.
	AsyncEventHandlingStrategy
)

type EventObserver ¶

type EventObserver interface {
	// Notify is notifed about an event.
	Notify(context.Context, Event) error
}

EventObserver is an observer of events. All observers will receive an event.

type EventPublisher ¶

type EventPublisher interface {
	// PublishEvent publishes the event to all observers.
	PublishEvent(context.Context, Event) error

	// AddObserver adds an observer.
	AddObserver(EventObserver)

	// SetHandlingStrategy will set the strategy to use for handling events.
	SetHandlingStrategy(EventHandlingStrategy)
}

EventPublisher is a publisher of events to observers.

type EventSourcingRepository ¶

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

EventSourcingRepository is an aggregate repository using event sourcing. It uses an event store for loading and saving events used to build the aggregate.

func NewEventSourcingRepository(eventStore EventStore, eventBus EventBus) (*EventSourcingRepository, error)

func (r *EventSourcingRepository) Load(ctx context.Context, aggregateType AggregateType, id UUID) (Aggregate, error)

func (r *EventSourcingRepository) Save(ctx context.Context, aggregate Aggregate) error

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, AggregateType, UUID) ([]Event, error)
}

EventStore is an interface for an event sourcing event store.

type EventStoreError ¶

type EventStoreError struct {
	// Err is the error.
	Err error
	// BaseErr is an optional underlying error, for example from the DB driver.
	BaseErr error
	// Namespace is the namespace for the error.
	Namespace string
}

EventStoreError is an error in the event store, with the namespace.

func (e EventStoreError) Error() string

type EventStoreMaintainer ¶

type EventStoreMaintainer interface {
	EventStore

	// 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
}

EventStoreMaintainer is an interface for a maintainer of an EventStore. NOTE: Should not be used in apps, useful for migration tools etc.

type EventType ¶

type EventType string

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

type ReadRepo ¶

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

	// Find returns a read model with an id.
	Find(context.Context, UUID) (interface{}, error)

	// FindAll returns all read models in the repository.
	FindAll(context.Context) ([]interface{}, error)
}

ReadRepo is a storage for read models.

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
	// BaseErr is an optional underlying error, for example from the DB driver.
	BaseErr error
	// Namespace is the namespace for the error.
	Namespace string
}

RepoError is an error in the read repository, with the namespace.

func (e RepoError) Error() string

type Repository ¶

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

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

Repository is a repository responsible for loading and saving aggregates.

type UUID ¶

type UUID string

UUID is a unique identifier, based on the UUID spec. It must be exactly 16 bytes long.

func NewUUID() UUID

func ParseUUID(s string) (UUID, error)

ParseUUID("6ba7b814-9dad-11d1-80b4-00c04fd430c8")
ParseUUID("{6ba7b814-9dad-11d1-80b4-00c04fd430c8}")
ParseUUID("urn:uuid:6ba7b814-9dad-11d1-80b4-00c04fd430c8")

func (id UUID) MarshalJSON() ([]byte, error)

func (id UUID) String() string

func (id *UUID) UnmarshalJSON(data []byte) error

type Versionable ¶

type Versionable interface {
	// AggregateVersion returns the aggregate version that a read model represents.
	AggregateVersion() int
}

Versionable is a read model that has a version number saved, used by version.ReadRepo.FindMinVersion().

type WriteRepo ¶

type WriteRepo interface {
	// Save saves a model in the storage.
	Save(context.Context, UUID, interface{}) error

	// Remove removes a model from the storage.
	Remove(context.Context, UUID) error
}

WriteRepo is a write repo for a matching ReadRepo, used in projectors.