cqrs

Documentation

Index

• func NewCommandJSONSerializer() *commandJSONSerializer
• func NewEventJSONSerializer() *eventJSONSerializer
• func NewInMemoryEventStore() *inMemoryStore
• func NewSimpleHandler(repo Repository) *simpleCommandHandler
• func NewSimpleRepository(store EventStore) *simpleRepository
• type AggregateRoot
• type AggregateRootCtor
• type BaseCommand
  • func (c *BaseCommand) GetAggregateID() string
  • func (c *BaseCommand) GetAggregateType() string
  • func (c *BaseCommand) GetCommandID() CommandID
  • func (c *BaseCommand) GetCorrelationID() string
  • func (c *BaseCommand) Validate(_ AggregateRoot) error
• type Command
• type CommandHandler
• type CommandID
• type CommandSerializer
• type CommandValidator
• type Event
  • func NewEvent(ID EventID, cmd Command, data interface{}) *Event
• type EventHook
• type EventID
• type EventSerializer
• type EventStore
• type Repository
• type Root
  • func (r *Root) ClearChanges()
  • func (r *Root) GetChanges() []*Event
  • func (r *Root) GetID() string

Functions

func NewCommandJSONSerializer

func NewCommandJSONSerializer() *commandJSONSerializer

NewCommandJSONSerializer is implementation of CommandSerializer that uses JSON as underlying format.

func NewEventJSONSerializer

func NewEventJSONSerializer() *eventJSONSerializer

NewEventJSONSerializer returns instance of EventSerializer that is using JSON as underlying format.

func NewInMemoryEventStore

func NewInMemoryEventStore() *inMemoryStore

NewInMemoryEventStore returns EventStore implementation that stores events only in memory.

func NewSimpleHandler

func NewSimpleHandler(repo Repository) *simpleCommandHandler

func NewSimpleRepository

func NewSimpleRepository(store EventStore) *simpleRepository

Types

type AggregateRoot

type AggregateRoot interface {
	// HandleCommand generates and applies relevant events for provided command to this aggregate root.
	HandleCommand(cmd Command) error

	// GetID returns unique ID for this aggregate root.
	GetID() string

	// GetChanges returns list of new events that have been applied to this aggregate root.
	GetChanges() []*Event

	// ClearChanges deletes all new events applied to this aggregate root.
	// This should be done after save to persistent database or in case of rollback.
	ClearChanges()

	// Apply changes state of this aggregate root to reflect desired outcome described by
	// provided event. Flag new is true in case new event is being added (typically during
	// HandleCommand), and it will be false if aggregate root is being reconstructed by past,
	// already stored events.
	Apply(new bool, ev *Event) error
}

AggregateRoot is an instance of a type that is entity being managed. For example, in user service, User type would implement and be AggregateRoot.

type AggregateRootCtor

type AggregateRootCtor func() AggregateRoot

AggregateRootCtor is a function that returns empty instance of an aggregate root.

type BaseCommand

type BaseCommand struct {
	CommandID     CommandID `json:"command_id" mapstructure:"command_id"`
	AggregateID   string    `json:"aggregate_id" mapstructure:"aggregate_id"`
	AggregateType string    `json:"aggregate_type" mapstructure:"aggregate_type"`
	CorrelationID string    `json:"correlation_id" mapstructure:"correlation_id"`
}

BaseCommand is utility, implementing common parts for each command. Intended usage is by embedding it to concrete command. E.g.

type MyCommand {
  cqrs.BaseCommand
  Param string
 }

func (*BaseCommand) GetAggregateID

func (c *BaseCommand) GetAggregateID() string

func (*BaseCommand) GetAggregateType

func (c *BaseCommand) GetAggregateType() string

func (*BaseCommand) GetCommandID

func (c *BaseCommand) GetCommandID() CommandID

func (*BaseCommand) GetCorrelationID

func (c *BaseCommand) GetCorrelationID() string

func (*BaseCommand) Validate

func (c *BaseCommand) Validate(_ AggregateRoot) error

type Command

type Command interface {
	// GetCommandID returns string uniquely identifying command in the system.
	GetCommandID() CommandID

	// Validate checks if this command is valid to be executed on provided aggregate root.
	// If not, error should be return, nil otherwise.
	Validate(root AggregateRoot) error

	// GetAggregateID returns unique ID of an object that this command should be applied on.
	// Only in case of create events this should return zero value and aggregate root
	// handler for such commands should expect that.
	GetAggregateID() string

	// GetAggregateType returns identification of a type to which this command should be
	// applied to. Together with GetAggregateID this should determine instance on which
	// command should be applied to.
	GetAggregateType() string

	// GetCorrelationID returns unique identifier associated with concrete command execution.
	// It is used to tie all events that were created from the same command, but also to identify
	// messages passed between systems that apply to the same command.
	GetCorrelationID() string
}

Command is something sent to a service to execute.

type CommandHandler

type CommandHandler interface {
	// HandleCommand processes provided command or dies (returns error) trying.
	HandleCommand(cmd Command) error
}

CommandHandler can process and execute a command.

type CommandID

type CommandID string

CommandID is identifier of a command across the system. Good idea is to use prefix with entity name.

type CommandSerializer

type CommandSerializer interface {
	Marshal(Command) ([]byte, error)
	Unmarshal([]byte) (Command, error)
}

CommandSerializer defines operations needed for command instance marshal and unmarshal operations.

type CommandValidator

type CommandValidator interface {
	Validate(Command) error
}

type Event

type Event struct {
	EventID       EventID     `json:"event_id" mapstructure:"event_id"`
	AggregateID   string      `json:"aggregate_id" mapstructure:"aggregate_id"`
	AggregateType string      `json:"aggregate_type" mapstructure:"aggregate_type"`
	CreatedAt     time.Time   `json:"created_at" mapstructure:"created_at"`
	CorrelationID string      `json:"correlation_id" mapstructure:"correlation_id"`
	Data          interface{} `json:"data" mapstructure:"data"`
}

Event is base event, containing all the needed information for something that had happened in the past.

func NewEvent

func NewEvent(ID EventID, cmd Command, data interface{}) *Event

NewEvent returns instance of an event with provided ID and data, and populates relevant fields from provided command.

type EventHook

type EventHook func(*Event)

EventHook is a function to be called during event lifecycle

type EventID

type EventID string

EventID is unique string representing an event. Should be unique per event type across the system. Good idea is to use prefix per entity.

type EventSerializer

type EventSerializer interface {
	Marshal(*Event) ([]byte, error)
	Unmarshal([]byte) (*Event, error)
	MarshalData(*Event) ([]byte, error)
	UnmarshalData(EventID, []byte) (interface{}, error)
}

EventSerializer defines operations needed for event instance marshal and unmarshal operations.

type EventStore

type EventStore interface {
	// Load returns all events for provided aggregate root id.
	Load(aggregateID string) ([]*Event, error)

	// Save persist all provided events.
	Save([]*Event) error
}

EventStore is description of persistence for events.

type Repository

type Repository interface {
	// Load creates and returns aggregate root with provided id.
	Load(typ, aggregateID string) (AggregateRoot, error)

	// Save stores new events from provided aggregate root.
	Save(root AggregateRoot) error
}

Repository manages aggregate root objects.

type Root

type Root struct {
	ID      string
	Changes []*Event
}

Root is partial implementation of AggregateRoot, made to make implementation of concrete aggregate roots easier by embedding.

func (*Root) ClearChanges

func (r *Root) ClearChanges()

func (*Root) GetChanges

func (r *Root) GetChanges() []*Event

func (*Root) GetID

func (r *Root) GetID() string