aggregate

Documentation

Index

• Constants
• Variables
• func RecordChange(aggregateRoot Root, event interface{}) error
• type BaseRoot
  • func (b *BaseRoot) AggregateVersion() uint
• type Changed
  • func ReconstituteChange(aggregateID ID, uuid goengine.UUID, payload interface{}, ...) (*Changed, error)
  • func (a *Changed) AggregateID() ID
  • func (a *Changed) CreatedAt() time.Time
  • func (a *Changed) Metadata() metadata.Metadata
  • func (a *Changed) Payload() interface{}
  • func (a *Changed) UUID() goengine.UUID
  • func (a *Changed) Version() uint
  • func (a Changed) WithMetadata(key string, value interface{}) goengine.Message
• type EventApplier
• type ID
  • func GenerateID() ID
  • func IDFromString(str string) (ID, error)
• type Initiator
• type Repository
  • func NewRepository(eventStore goengine.EventStore, streamName goengine.StreamName, ...) (*Repository, error)
  • func (r *Repository) GetAggregateRoot(ctx context.Context, aggregateID ID) (Root, error)
  • func (r *Repository) SaveAggregateRoot(ctx context.Context, aggregateRoot Root) error
• type Root
• type Type
  • func NewType(name string, initiator Initiator) (*Type, error)
  • func (t *Type) CreateInstance() Root
  • func (t *Type) IsImplementedBy(root interface{}) bool
  • func (t *Type) String() string

Constants

const (
	// TypeKey is the metadata key to identify the aggregate type
	TypeKey = "_aggregate_type"
	// IDKey is the metadata key to identify the aggregate id
	IDKey = "_aggregate_id"
	// VersionKey is the metadata key to identify the aggregate version
	VersionKey = "_aggregate_version"
)

Variables

var (
	// ErrMissingAggregateID occurs when no or an invalid aggregate.ID was provided
	ErrMissingAggregateID = errors.New("goengine: no or empty aggregate ID was provided")
	// ErrMissingChangeUUID occurs when no or an invalid message.UUID was provided
	ErrMissingChangeUUID = errors.New("goengine: no or empty message UUID was provided")
	// ErrInvalidChangeVersion occurs since a version cannot be zero
	ErrInvalidChangeVersion = errors.New("goengine: a changed event must have a version number greater than zero")
	// ErrInvalidChangePayload occurs when no payload is provided
	ErrInvalidChangePayload = errors.New("goengine: a changed event must have a payload that is not nil")
)

var (
	// ErrUnsupportedAggregateType occurs when the given aggregateType is not handled by the AggregateRepository
	ErrUnsupportedAggregateType = errors.New("goengine: the given AggregateRoot is of a unsupported type")
	// ErrUnexpectedMessageType occurs when the event store returns a message that is not an *aggregate.Changed
	ErrUnexpectedMessageType = errors.New("goengine: event store returned an unsupported message type")
	// ErrEmptyEventStream occurs when the event stream returned by the event store is empty
	ErrEmptyEventStream = errors.New("goengine: unsupported empty event stream")
)

var ErrInitiatorMustReturnRoot = errors.New("goengine: the aggregate.Initiator must return a pointer to the aggregate.Root")

ErrInitiatorMustReturnRoot occurs when a aggregate.Initiator did not return a pointer

var ErrInvalidID = errors.New("goengine: an aggregate.ID must be a valid UUID")

ErrInvalidID occurs when a string is not a valid ID

Functions

func RecordChange

func RecordChange(aggregateRoot Root, event interface{}) error

RecordChange record the given event onto the aggregate.Root by wrapping it in an aggregate.Changed

Types

type BaseRoot

type BaseRoot struct {
	sync.Mutex
	// contains filtered or unexported fields
}

BaseRoot is the base struct to be embedded for any aggregate root

func (*BaseRoot) AggregateVersion

func (b *BaseRoot) AggregateVersion() uint

AggregateVersion returns the version of the aggregate

type Changed

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

Changed is a message indicating that a aggregate was changed

func ReconstituteChange

func ReconstituteChange(
	aggregateID ID,
	uuid goengine.UUID,
	payload interface{},
	metadata metadata.Metadata,
	createdAt time.Time,
	version uint,
) (*Changed, error)

ReconstituteChange recreates a previous aggregate Changed message based on the provided data

func (*Changed) AggregateID

func (a *Changed) AggregateID() ID

AggregateID returns the aggregate ID

func (*Changed) CreatedAt

func (a *Changed) CreatedAt() time.Time

CreatedAt returns the created time

func (*Changed) Metadata

func (a *Changed) Metadata() metadata.Metadata

Metadata return the change metadata

func (*Changed) Payload

func (a *Changed) Payload() interface{}

Payload returns the payload of the change This is the actual domain event

func (*Changed) UUID

func (a *Changed) UUID() goengine.UUID

UUID returns the unique message identifier

func (*Changed) Version

func (a *Changed) Version() uint

Version return the version of aggregate this change represents

func (Changed) WithMetadata

func (a Changed) WithMetadata(key string, value interface{}) goengine.Message

WithMetadata Returns new instance of the change with key and value added to metadata

type EventApplier

type EventApplier interface {
	// Apply updates the state of the aggregateRoot based on the recorded event
	// This method should only be called by the library
	Apply(event *Changed)
}

EventApplier is a interface so that a Changed event can be applied

type ID

type ID string

ID an UUID for a aggregate.Root instance

func GenerateID

func GenerateID() ID

GenerateID creates a new random UUID or panics

func IDFromString

func IDFromString(str string) (ID, error)

IDFromString creates a ID from a string

type Initiator

type Initiator func() Root

Initiator creates a new empty instance of a aggregate.Root this instance may then be used to reconstitute the state of the aggregate root

type Repository

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

Repository a repository to save and load aggregate.Root's of a specific type

func NewRepository

func NewRepository(
	eventStore goengine.EventStore,
	streamName goengine.StreamName,
	aggregateType *Type,
) (*Repository, error)

NewRepository instantiates a new AggregateRepository

func (*Repository) GetAggregateRoot

func (r *Repository) GetAggregateRoot(ctx context.Context, aggregateID ID) (Root, error)

GetAggregateRoot returns nil if no stream events can be found for aggregate id otherwise the reconstituted aggregate root

func (*Repository) SaveAggregateRoot

func (r *Repository) SaveAggregateRoot(ctx context.Context, aggregateRoot Root) error

SaveAggregateRoot stores the state changes of the aggregate.Root

type Root

type Root interface {
	EventApplier

	// ID returns the aggregateID
	AggregateID() ID
	// contains filtered or unexported methods
}

Root is a interface that a AggregateRoot must implement

type Type

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

Type an aggregate type represent the type of a aggregate

func NewType

func NewType(name string, initiator Initiator) (*Type, error)

NewType create a new aggregate type instance

func (*Type) CreateInstance

func (t *Type) CreateInstance() Root

CreateInstance returns an new empty/zero instance of the aggregate root that the type represents

func (*Type) IsImplementedBy

func (t *Type) IsImplementedBy(root interface{}) bool

IsImplementedBy returns whether the given aggregate root is a instance of this aggregateType

func (*Type) String

func (t *Type) String() string

String returns the name of the type