cqrs

Documentation

Overview

Package cqrs is WIP.

Index

• type Aggregate
• type AggregateConfig
• type AggregateCursor
• type AggregateFactory
• type AggregateMatcher
• type AggregateQuery
• type AggregateRepository
• type AggregateType
  • func (t AggregateType) String() string
• type Command
• type CommandBus
• type CommandConfig
• type CommandHandler
• type CommandType
• type Container
  • func NewContainer(aggregates AggregateConfig, events EventConfig, commands CommandConfig) Container
• type Event
• type EventBus
• type EventConfig
• type EventCursor
• type EventData
• type EventMatcher
• type EventPublisher
• type EventQuery
• type EventStore
• type EventSubscriber
• type EventType
  • func (t EventType) String() string
• type SnapshotRepository

Types

type Aggregate

type Aggregate interface {
	AggregateID() uuid.UUID
	AggregateType() AggregateType
	OriginalVersion() int
	CurrentVersion() int
	Changes() []Event
	TrackChange(...Event)
	FlushChanges()
	ApplyEvent(Event) error
}

Aggregate is the aggregate of an event stream.

type AggregateConfig

type AggregateConfig interface {
	Register(AggregateType, AggregateFactory)
	New(AggregateType, uuid.UUID) (Aggregate, error)
	Factories() map[AggregateType]AggregateFactory
}

AggregateConfig ...

type AggregateCursor

type AggregateCursor interface {
	// Next moves the cursor to the next aggregate.
	Next(context.Context) bool
	// Aggregate fetches and returns the current aggregate at it's latest version.
	Aggregate(context.Context) (Aggregate, error)
	// Version fetches and returns the current aggregate at the provided version.
	Version(context.Context, int) (Aggregate, error)
	// Err returns the current error.
	Err() error
	// Close closes the cursor.
	Close(context.Context) error
}

AggregateCursor ...

type AggregateFactory

type AggregateFactory func(uuid.UUID) Aggregate

AggregateFactory ...

type AggregateMatcher

type AggregateMatcher func(Aggregate) bool

AggregateMatcher ...

type AggregateQuery

type AggregateQuery interface {
	// Types returns the aggregate types to query for.
	Types() []AggregateType
	// IDs returns the aggregate IDs to query for.
	IDs() []uuid.UUID
}

AggregateQuery ...

type AggregateRepository

type AggregateRepository interface {
	Save(ctx context.Context, aggregate Aggregate) error
	Fetch(ctx context.Context, typ AggregateType, id uuid.UUID, version int) (Aggregate, error)
	FetchWithBase(ctx context.Context, aggregate Aggregate, version int) (Aggregate, error)
	FetchLatest(ctx context.Context, typ AggregateType, id uuid.UUID) (Aggregate, error)
	FetchLatestWithBase(ctx context.Context, aggregate Aggregate) (Aggregate, error)
	Remove(ctx context.Context, aggregate Aggregate) error
	RemoveType(ctx context.Context, typ AggregateType) error
	Query(ctx context.Context, query AggregateQuery) (AggregateCursor, error)
}

AggregateRepository ...

type AggregateType

type AggregateType string

AggregateType is the type of an aggregate.

func (AggregateType) String

func (t AggregateType) String() string

type Command

type Command interface {
	Type() CommandType
	AggregateType() AggregateType
	AggregateID() uuid.UUID
}

Command is a command.

type CommandBus

type CommandBus interface {
	Dispatch(context.Context, Command) error
}

CommandBus dispatches commands.

type CommandConfig

type CommandConfig interface {
	Register(CommandType, CommandHandler)
	Handler(CommandType) (CommandHandler, error)
	Handlers() map[CommandType]CommandHandler
}

CommandConfig ...

type CommandHandler

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

CommandHandler handles commands.

type CommandType

type CommandType string

CommandType is a command type.

type Container

type Container interface {
	AggregateConfig() AggregateConfig
	SetAggregateConfig(AggregateConfig)
	EventConfig() EventConfig
	SetEventConfig(EventConfig)
	CommandConfig() CommandConfig
	SetCommandConfig(CommandConfig)
	EventBus() EventBus
	EventPublisher() EventPublisher
	EventSubscriber() EventSubscriber
	SetEventBus(EventBus)
	EventStore() EventStore
	SetEventStore(EventStore)
	CommandBus() CommandBus
	SetCommandBus(CommandBus)
	Snapshots() SnapshotRepository
	SetSnapshots(SnapshotRepository)
	Aggregates() AggregateRepository
	SetAggregates(AggregateRepository)
}

Container ...

func NewContainer

func NewContainer(aggregates AggregateConfig, events EventConfig, commands CommandConfig) Container

NewContainer ...

type Event

type Event interface {
	Type() EventType
	Data() EventData
	Time() time.Time

	AggregateType() AggregateType
	AggregateID() uuid.UUID
	Version() int
}

Event is an event.

type EventBus

type EventBus interface {
	EventPublisher
	EventSubscriber
}

EventBus is the event bus.

type EventConfig

type EventConfig interface {
	Register(EventType, EventData)
	// NewData creates an EventData instance for EventType.
	// The returned object must be a non-pointer struct.
	NewData(EventType) (EventData, error)
	Protos() map[EventType]EventData
}

EventConfig is the configuration for the events.

type EventCursor

type EventCursor interface {
	// Next moves the cursor to the next event.
	Next(context.Context) bool
	// Event returns the current event.
	Event() Event
	// Err returns the cursor error.
	Err() error
	// Close closes the cursor.
	Close(context.Context) error
}

EventCursor ...

type EventData

type EventData interface{}

EventData is the payload of an event.

type EventMatcher

type EventMatcher func(Event) bool

EventMatcher ...

type EventPublisher

type EventPublisher interface {
	Publish(ctx context.Context, events ...Event) error
}

EventPublisher publishes events.

type EventQuery

type EventQuery interface {
	// EventTypes returns the event types to query for.
	// Should query for all event types if none provided.
	EventTypes() []EventType
	// AggregateTypes returns the aggregate types to query for.
	// Should query for all aggregate types if none provided.
	AggregateTypes() []AggregateType
	// AggregateIDs returns the aggregate IDs to query for.
	// Should query for all IDs if none provided.
	AggregateIDs() []uuid.UUID
	// Versions returns the aggregate versions to query for.
	Versions() []int
	// VersionRanges returns version ranges to query for.
	VersionRanges() [][2]int
}

EventQuery ...

type EventStore

type EventStore interface {
	Save(ctx context.Context, originalVersion int, events ...Event) error
	Find(ctx context.Context, typ AggregateType, id uuid.UUID, version int) (Event, error)
	Fetch(ctx context.Context, typ AggregateType, id uuid.UUID, from int, to int) ([]Event, error)
	FetchAll(ctx context.Context, typ AggregateType, id uuid.UUID) ([]Event, error)
	FetchFrom(ctx context.Context, typ AggregateType, id uuid.UUID, from int) ([]Event, error)
	FetchTo(ctx context.Context, typ AggregateType, id uuid.UUID, to int) ([]Event, error)
	RemoveAggregate(ctx context.Context, typ AggregateType, id uuid.UUID) error
	RemoveAggregateType(ctx context.Context, typ AggregateType) error
	Query(ctx context.Context, query EventQuery) (EventCursor, error)
}

EventStore stores events in a database.

type EventSubscriber

type EventSubscriber interface {
	Subscribe(ctx context.Context, types ...EventType) (<-chan Event, error)
}

EventSubscriber subscribes to events.

type EventType

type EventType string

EventType is the type of an event.

func (EventType) String

func (t EventType) String() string

type SnapshotRepository

type SnapshotRepository interface {
	Save(ctx context.Context, snap Aggregate) error
	Find(ctx context.Context, typ AggregateType, id uuid.UUID, version int) (Aggregate, error)
	Latest(ctx context.Context, typ AggregateType, id uuid.UUID) (Aggregate, error)
	MaxVersion(ctx context.Context, typ AggregateType, id uuid.UUID, maxVersion int) (Aggregate, error)
	Remove(ctx context.Context, typ AggregateType, id uuid.UUID, version int) error
	RemoveAll(ctx context.Context, typ AggregateType, id uuid.UUID) error
}

SnapshotRepository ...