cqrs

README

cqrs

Documentation

Index

• type Aggregate
  • func NewAggregate(aggregateId string, retryFunc RetryFunc, numEventsInSnapshot int, ...) (*Aggregate, error)
  • func (a *Aggregate) HandleCommand(ctx context.Context, cmd Command) ([]event.Event, error)
• type AggregateModel
• type Command
• type FactoryModelFunc
• type Projection
  • func NewProjection(ctx context.Context, store eventstore.EventStore, subscriptionId string, ...) (*Projection, error)
  • func (p *Projection) Close() error
  • func (p *Projection) Forget(query []eventstore.SnapshotQuery) error
  • func (p *Projection) Handle(ctx context.Context, iter event.Iter) error
  • func (p *Projection) Models(queries []eventstore.SnapshotQuery) []eventstore.Model
  • func (p *Projection) Project(ctx context.Context, query []eventstore.SnapshotQuery) error
  • func (p *Projection) SubscribeTo(topics []string) error
• type RetryFunc
  • func NewDefaultRetryFunc(limit int) RetryFunc

Types

type Aggregate

type Aggregate struct {
	LogDebugfFunc eventstore.LogDebugfFunc
	// contains filtered or unexported fields
}

Aggregate holds data for Handle command

func NewAggregate

func NewAggregate(aggregateId string, retryFunc RetryFunc, numEventsInSnapshot int, store eventstore.EventStore, factoryModel FactoryModelFunc, LogDebugfFunc eventstore.LogDebugfFunc) (*Aggregate, error)

NewAggregate creates aggregate. it load and store events created from commands

func (*Aggregate) HandleCommand

func (a *Aggregate) HandleCommand(ctx context.Context, cmd Command) ([]event.Event, error)

HandleCommand transforms command to a event, store and publish event.

type AggregateModel

type AggregateModel interface {
	eventstore.Model

	HandleCommand(ctx context.Context, cmd Command, newVersion uint64) ([]event.Event, error)
	TakeSnapshot(version uint64) (snapshotEvent event.Event, ok bool)
	GroupId() string //defines group where model belows
}

AggregateModel user model for aggregate need to satisfy this interface.

type Command

type Command interface{}

Command user defined command that will handled in AggregateModel.HandleCommand

type FactoryModelFunc

type FactoryModelFunc func(ctx context.Context) (AggregateModel, error)

FactoryModelFunc creates model for aggregate

type Projection

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

Projection project events to user defined model from evenstore and update it by events from subscriber.

func NewProjection

func NewProjection(ctx context.Context, store eventstore.EventStore, subscriptionId string, subscriber eventbus.Subscriber, factoryModel eventstore.FactoryModelFunc, LogDebugfFunc eventstore.LogDebugfFunc) (*Projection, error)

NewProjection creates projection.

func (*Projection) Close

func (p *Projection) Close() error

Close cancel projection.

func (*Projection) Forget

func (p *Projection) Forget(query []eventstore.SnapshotQuery) error

Forget projection for certain query.

func (*Projection) Handle

func (p *Projection) Handle(ctx context.Context, iter event.Iter) error

Handle events to projection. This events comes from eventbus and it can trigger reload on eventstore.

func (*Projection) Models

func (p *Projection) Models(queries []eventstore.SnapshotQuery) []eventstore.Model

Models get models from projection

func (*Projection) Project

func (p *Projection) Project(ctx context.Context, query []eventstore.SnapshotQuery) error

Project load events from aggregates that below to path.

func (*Projection) SubscribeTo

func (p *Projection) SubscribeTo(topics []string) error

SubscribeTo set topics for observation for update events.

type RetryFunc

type RetryFunc func() (when time.Time, err error)

RetryFunc defines policy to repeat HandleCommand on concurrency exception.

func NewDefaultRetryFunc

func NewDefaultRetryFunc(limit int) RetryFunc

NewDefaultRetryFunc default retry function