eventhorizon

package module
v0.8.3 Latest Latest
Warning

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

 Go to latest Published: Nov 25, 2020 License: Apache-2.0 Imports: 9 Imported by: 0

README

Build Status Coverage Status GoDoc Go Report Card

Event Horizon

Event Horizon is a CQRS/ES toolkit for Go.

NOTE: Event Horizon is used in production systems but the API is not final!

CQRS stands for Command Query Responsibility Segregation and is a technique where object access (the Query part) and modification (the Command part) are separated from each other. This helps in designing complex data models where the actions can be totally independent from the data output.

ES stands for Event Sourcing and is a technique where all events that have happened in a system are recorded, and all future actions are based on the events instead of a single data model. The main benefit of adding Event Sourcing is traceability of changes which can be used for example in audit logging. Additionally, "incorrect" events that happened in the past (for example due to a bug) can be compensated for with an event which will make the current data "correct", as that is based on the events.

Read more about CQRS/ES from one of the major authors/contributors on the subject: http://codebetter.com/gregyoung/2010/02/16/cqrs-task-based-uis-event-sourcing-agh/

Other material on CQRS/ES:

Inspired by the following libraries/examples:

Suggestions are welcome!

Usage

See the example folder for a few examples to get you started.

Storage drivers

These are the drivers for storage of events and entities.

Local / in memory

There are simple in memory implementations of an event store and entity repo. These are meant for testing/experimentation.

MongoDB

Fairly mature, used in production. Note: This implementation has a document size limit of 16MB, which could lead to errors for large aggregates. If needed use the alternative version below

MongoDB-DPE (supporting larger aggregates)

MongoDB-Document Per Event is an alternative to the MongoDB driver mentioned above. In the MongoDB-DPE driver, events for an aggregate are each saved in a separate document. This allows aggregate event histories larger than the maximum document size of MongoDB, which is 16MB. https://github.com/gjongenelen/eh-mongodb

AWS DynamoDB

https://github.com/seedboxtech/eh-dynamo

Postgress

https://github.com/giautm/eh-pg

Redis

https://github.com/TerraSkye/eh-redis

Messaging drivers

These are the drivers for messaging, currently only publishers.

Local / in memory

Fully synchrounos. Useful for testing/experimentation.

GCP Cloud Pub/Sub

Experimental driver.

Kafka

https://github.com/Kistler-Group/eh-kafka

NATS Streaming

https://github.com/v0id3r/eh-nats

Development

To develop Event Horizon you need to have Docker and Docker Compose installed.

To start all needed services and run all tests, simply run make:

make

To manualy run the services and stop them:

make run_services
make stop_services

When the services are running testing can be done either locally or with Docker:

go test ./...
make test
make test_docker
make cover

The difference between make test and go test ./... is that make test also does race detection.

Get Involved

License

Event Horizon is licensed under Apache License 2.0

http://www.apache.org/licenses/LICENSE-2.0

Documentation

Overview

Package eventhorizon is a CQRS/ES toolkit.

Package eventhorizon is a CQRS/ES toolkit.

Index

Constants

View Source 
const DefaultMinVersionDeadline = 10 * time.Second

DefaultMinVersionDeadline is the deadline to use when creating a min version context that waits.

View Source 
const DefaultNamespace = "default"

DefaultNamespace is the namespace to use if not set in the context.

Variables

View Source 
var ErrAggregateNotFound = errors.New("aggregate not found")

ErrAggregateNotFound is when no aggregate can be found.

View Source 
var ErrAggregateNotRegistered = errors.New("aggregate not registered")

ErrAggregateNotRegistered is when no aggregate factory was registered.

View Source 
var ErrCommandNotRegistered = errors.New("command not registered")

ErrCommandNotRegistered is when no command factory was registered.

View Source 
var ErrCouldNotSaveEntity = errors.New("could not save entity")

ErrCouldNotSaveEntity is when a entity could not be saved.

View Source 
var ErrEntityHasNoVersion = errors.New("entity has no version")

ErrEntityHasNoVersion is when an entity has no version number.

View Source 
var ErrEntityNotFound = errors.New("could not find entity")

ErrEntityNotFound is when a entity could not be found.

View Source 
var ErrEventDataNotRegistered = errors.New("event data not registered")

ErrEventDataNotRegistered is when no event data factory was registered.

View Source 
var ErrHandlerAlreadyAdded = errors.New("handler already added")

ErrHandlerAlreadyAdded is returned when adding the same handler twice.

View Source 
var ErrIncorrectEntityVersion = errors.New("incorrect entity version")

ErrIncorrectEntityVersion is when an entity has an incorrect version.

View Source 
var ErrIncorrectEventVersion = errors.New("mismatching event version")

ErrIncorrectEventVersion is when an event is for an other version of the aggregate.

View Source 
var ErrInvalidEvent = errors.New("invalid event")

ErrInvalidEvent is when an event does not implement the Event interface.

View Source 
var ErrMissingEntityID = errors.New("missing entity ID")

ErrMissingEntityID is when a entity has no ID.

View Source 
var ErrMissingHandler = errors.New("missing handler")

ErrMissingHandler is returned when adding a handler with a nil handler.

View Source 
var ErrMissingMatcher = errors.New("missing matcher")

ErrMissingMatcher is returned when adding a handler without a matcher.

View Source 
var ErrNoEventsToAppend = errors.New("no events to append")

ErrNoEventsToAppend is when no events are available to append.

Functions

func CheckCommand 

func CheckCommand(cmd Command) error

CheckCommand checks a command for errors.

func MarshalContext 

func MarshalContext(ctx context.Context) map[string]interface{}

MarshalContext marshals a context into a map.

func NamespaceFromContext 

func NamespaceFromContext(ctx context.Context) string

NamespaceFromContext returns the namespace from the context, or the default namespace.

func NewContextWithNamespace 

func NewContextWithNamespace(ctx context.Context, namespace string) context.Context

NewContextWithNamespace sets the namespace to use in the context. The namespace is used to determine which database.

func RegisterAggregate 

func RegisterAggregate(factory func(uuid.UUID) Aggregate)

RegisterAggregate registers an aggregate factory for a type. The factory is used to create concrete aggregate types when loading from the database.

An example would be: 

RegisterAggregate(func(id UUID) Aggregate { return &MyAggregate{id} })

func RegisterCommand 

func RegisterCommand(factory func() Command)

RegisterCommand registers an command factory for a type. The factory is used to create concrete command types.

An example would be: 

RegisterCommand(func() Command { return &MyCommand{} })

func RegisterContextMarshaler 

func RegisterContextMarshaler(f ContextMarshalFunc)

RegisterContextMarshaler registers a marshaler function used by MarshalContext.

func RegisterContextUnmarshaler 

func RegisterContextUnmarshaler(f ContextUnmarshalFunc)

RegisterContextUnmarshaler registers a marshaler function used by UnmarshalContext.

func RegisterEventData 

func RegisterEventData(eventType EventType, factory func() EventData)

RegisterEventData registers an event data factory for a type. The factory is used to create concrete event data structs when loading from the database.

An example would be: 

RegisterEventData(MyEventType, func() Event { return &MyEventData{} })

func UnmarshalContext 

func UnmarshalContext(ctx context.Context, vals map[string]interface{}) context.Context

UnmarshalContext unmarshals a context from a map.

func UnregisterCommand 

func UnregisterCommand(commandType CommandType)

UnregisterCommand removes the registration of the command factory for a type. This is mainly useful in mainenance situations where the command type needs to be switched at runtime.

func UnregisterEventData 

func UnregisterEventData(eventType EventType)

UnregisterEventData removes the registration of the event data factory for a type. This is mainly useful in mainenance situations where the event data needs to be switched in a migrations.

Types

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 

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

CreateAggregate creates an aggregate of a type with an ID using the factory registered with RegisterAggregate.

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 uncommittend events for an aggregate.
	Save(context.Context, Aggregate) error
}

AggregateStore is responsible for loading and saving aggregates.

type AggregateType 

type AggregateType string

AggregateType is the type of an aggregate.

func (AggregateType) String added in v0.8.3 

func (at AggregateType) String() string

String returns the string representation of an aggregate type.

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 

func CreateCommand(commandType CommandType) (Command, error)

CreateCommand creates an command of a type with an ID using the factory registered with RegisterCommand.

type CommandFieldError 

type CommandFieldError struct {
	Field string
}

CommandFieldError is returned by Dispatch when a field is incorrect.

func (CommandFieldError) Error 

func (c CommandFieldError) Error() string

Error implements the Error method of the error interface.

type CommandHandler 

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

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

func UseCommandHandlerMiddleware 

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

UseCommandHandlerMiddleware wraps a CommandHandler in one or more middleware.

type CommandHandlerFunc 

type CommandHandlerFunc func(context.Context, Command) error

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

func (CommandHandlerFunc) HandleCommand 

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

HandleCommand implements the HandleCommand method of the CommandHandler.

type CommandHandlerMiddleware 

type CommandHandlerMiddleware func(CommandHandler) CommandHandler

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

type CommandType 

type CommandType string

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

func (CommandType) String added in v0.8.3 

func (ct CommandType) String() string

String returns the string representation of a command type.

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

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

NewEvent creates a new event with a type and data, setting its timestamp.

func NewEventForAggregate 

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

NewEventForAggregate creates a new event with a type and data, setting its timestamp. It also sets the aggregate data on it.

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 EventBusError

	// Wait wait for all handlers to be cancelled by their context.
	Wait()
}

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 garanteed to be handeled in order.

type EventBusError added in v0.8.2 

type EventBusError struct {
	Err   error
	Ctx   context.Context
	Event Event
}

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

func (EventBusError) Cause added in v0.8.3 

func (e EventBusError) Cause() error

Cause returns the cause of this error.

func (EventBusError) Error added in v0.8.2 

func (e EventBusError) Error() string

Error implements the Error method of the error interface.

type EventData 

type EventData interface{}

EventData is any additional data for an event.

func CreateEventData 

func CreateEventData(eventType EventType) (EventData, error)

CreateEventData creates an event data of a type using the factory registered with RegisterEventData.

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 

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

UseEventHandlerMiddleware wraps a EventHandler in one or more middleware.

type EventHandlerFunc 

type EventHandlerFunc func(context.Context, Event) error

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

func (EventHandlerFunc) HandleEvent 

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

HandleEvent implements the HandleEvent method of the EventHandler.

func (EventHandlerFunc) HandlerType added in v0.8.2 

func (f EventHandlerFunc) HandlerType() EventHandlerType

HandlerType implements the HandlerType method of the EventHandler by returning the name of the package and function: "github.com/looplab/eventhorizon.Function" becomes "eventhorizon-Function"

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.8.2 

type EventHandlerType string

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

func (EventHandlerType) String added in v0.8.3 

func (ht EventHandlerType) String() string

String returns the string representation of an event handler type.

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 EventSource added in v0.8.3 

type EventSource interface {
	// Events returns events. It should only return any new events since its
	// last invocation.
	Events() []Event
}

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

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 (EventStoreError) Cause added in v0.8.3 

func (e EventStoreError) Cause() error

Cause returns the cause of this error.

func (EventStoreError) Error 

func (e EventStoreError) Error() string

Error implements the Error method of the errors.Error interface.

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.

func (EventType) String added in v0.8.3 

func (et EventType) String() string

String returns the string representation of an event type.

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.3 

type MatchAggregates []AggregateType

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

func (MatchAggregates) Match added in v0.8.3 

func (types MatchAggregates) Match(e Event) bool

Match implements the Match method of the EventMatcher interface.

type MatchAll added in v0.8.3 

type MatchAll []EventMatcher

MatchAll matches all of the matchers.

func (MatchAll) Match added in v0.8.3 

func (matchers MatchAll) Match(e Event) bool

Match implements the Match method of the EventMatcher interface.

type MatchAny 

type MatchAny []EventMatcher

MatchAny matches any of the matchers.

func (MatchAny) Match added in v0.8.3 

func (matchers MatchAny) Match(e Event) bool

Match implements the Match method of the EventMatcher interface.

type MatchEvents added in v0.8.3 

type MatchEvents []EventType

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

func (MatchEvents) Match added in v0.8.3 

func (types MatchEvents) Match(e Event) bool

Match implements the Match method of the EventMatcher interface.

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 an entity for an ID.
	Find(context.Context, uuid.UUID) (Entity, error)

	// FindAll returns all entities in the repository.
	FindAll(context.Context) ([]Entity, 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
	// 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 (RepoError) Cause added in v0.8.3 

func (e RepoError) Cause() error

Cause returns the cause of this error.

func (RepoError) Error 

func (e RepoError) Error() string

Error implements the Error method of the errors.Error interface.

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.FindMinVersion().

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.

Source Files

View all Source files

Directories

Path Synopsis
aggregatestore
events
model
commandhandler
aggregate
bus
gcp
local
eventhandler
projector
saga
waiter
memory
mongodb
trace
examples
guestlist/domains/guestlist
guestlist/memory
Package memory contains an example of a CQRS/ES app using memory as DB.
 Package memory contains an example of a CQRS/ES app using memory as DB.
guestlist/mongodb
Package mongodb contains an example of a CQRS/ES app using the MongoDB adapter.
 Package mongodb contains an example of a CQRS/ES app using the MongoDB adapter.
todomvc/backend
todomvc/backend/domains/todo
todomvc/backend/handler
middleware
commandhandler/async
commandhandler/scheduler
commandhandler/validator
eventhandler/async
eventhandler/observer
eventhandler/scheduler
cache
memory
mongodb
version
types
mongodb

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.