eventhorizon

type Aggregate ¶

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

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

	// Version returns the version of the aggregate.
	Version() int

	// IncrementVersion increments the aggregate version.
	IncrementVersion()

	// HandleCommand handles a command and stores events.
	HandleCommand(Command) error

	// ApplyEvent applies an event to the aggregate by setting its values.
	ApplyEvent(events Event)

	// StoreEvent stores an event until as uncommitted.
	StoreEvent(Event)

	// GetUncommittedEvents gets all uncommitted events for storing.
	GetUncommittedEvents() []Event

	// ClearUncommittedEvents clears all uncommitted events after storing.
	ClearUncommittedEvents()
}

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 imlement the full interface, or more commonly embed *AggregateBase to take care of the common methods.

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
}

The embedded aggregate is then initialized by the factory function in the callback repository.

func NewAggregateBase(id UUID) *AggregateBase

func (a *AggregateBase) AggregateID() UUID

func (a *AggregateBase) ClearUncommittedEvents()

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

func (a *AggregateBase) IncrementVersion()

func (a *AggregateBase) StoreEvent(event 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(command Command) error

func (h *AggregateCommandHandler) SetAggregate(aggregate Aggregate, command Command) error

type AggregateRecord ¶

type AggregateRecord interface {
	AggregateID() UUID
	Version() int
	EventRecords() []EventRecord
}

AggregateRecord is a stored record of an aggregate in form of its events.

type CallbackRepository ¶

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

CallbackRepository is an aggregate repository using factory functions.

func NewCallbackRepository(eventStore EventStore) (*CallbackRepository, error)

func (r *CallbackRepository) Load(aggregateType string, id UUID) (Aggregate, error)

func (r *CallbackRepository) RegisterAggregate(aggregate Aggregate, callback func(UUID) Aggregate) error

repository.RegisterAggregate(&Aggregate{}, func(id UUID) interface{} { return &Aggregate{id} })

func (r *CallbackRepository) Save(aggregate Aggregate) error

type Command ¶

type Command interface {
	AggregateID() UUID
	AggregateType() string
	CommandType() string
}

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"`.

type CommandBus ¶

type CommandBus interface {
	// HandleCommand handles a command on the event bus.
	HandleCommand(Command) error
	// SetHandler registers a handler with a command.
	SetHandler(CommandHandler, Command) error
}

CommandBus is an interface defining an event bus for distributing events.

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(Command) error
}

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

type Event ¶

type Event interface {
	AggregateID() UUID
	AggregateType() string
	EventType() string
}

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

An event name should 1) be in past tense and 2) contain the intent (CustomerMoved vs CustomerAddressCorrected).

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

type EventBus ¶

type EventBus interface {
	// PublishEvent publishes an event on the event bus.
	PublishEvent(Event)
	// AddHandler adds a handler for a specific local event.
	AddHandler(EventHandler, Event)
	// AddLocalHandler adds a handler for local events.
	AddLocalHandler(EventHandler)
	// AddGlobalHandler adds a handler for global (remote) events.
	AddGlobalHandler(EventHandler)
}

EventBus is an interface defining an event bus for distributing events.

type EventHandler ¶

type EventHandler interface {
	// HandleEvent handles an event.
	HandleEvent(Event)
}

EventHandler is an interface that all handlers of events should implement.

type EventRecord ¶

type EventRecord interface {
	Type() string
	Version() int
	Events() []Event
}

EventRecord is a single event record with timestamp

type EventStore ¶

type EventStore interface {
	// Save appends all events in the event stream to the store.
	Save([]Event) error

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

EventStore is an interface for an event sourcing event store.

type ReadRepository ¶

type ReadRepository interface {
	// Save saves a read model with id to the repository.
	Save(UUID, interface{}) error

	// Find returns one read model with using an id.
	Find(UUID) (interface{}, error)

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

	// Remove removes a read model with id from the repository.
	Remove(UUID) error
}

ReadRepository is a storage for read models.

type Repository ¶

type Repository interface {
	// Load loads an aggregate with a type and id.
	Load(string, UUID) (Aggregate, error)

	// Save saves an aggregets uncommitted events.
	Save(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