handler

Documentation

Index

• Variables
• type AggregateReducer
• type Column
  • func NewCol(name string, value interface{}) Column
  • func NewJSONCol(name string, value interface{}) Column
• type Condition
  • func NewCond(name string, value interface{}) Condition
• type EventReducer
• type Executer
• type Handler
  • func NewHandler(config HandlerConfig) Handler
  • func (h *Handler) Subscribe(aggregates ...eventstore.AggregateType)
  • func (h *Handler) SubscribeEvents(types map[eventstore.AggregateType][]eventstore.EventType)
• type HandlerConfig
• type Lock
• type ProjectionHandler
  • func NewProjectionHandler(config ProjectionHandlerConfig, reduce Reduce, update Update, ...) *ProjectionHandler
  • func (h *ProjectionHandler) Process(ctx context.Context, reduce Reduce, update Update, lock Lock, unlock Unlock)
  • func (h *ProjectionHandler) ResetShouldBulk()
  • func (h *ProjectionHandler) TriggerBulk(ctx context.Context, lock Lock, unlock Unlock) error
• type ProjectionHandlerConfig
• type Reduce
• type SearchQuery
• type Statement
  • func (s *Statement) IsNoop() bool
• type Statements
  • func (stmts Statements) Len() int
  • func (stmts Statements) Less(i, j int) bool
  • func (stmts Statements) Swap(i, j int)
• type Unlock
• type Update

Variables

var (
	ErrNoProjection    = errors.New("no projection")
	ErrNoValues        = errors.New("no values")
	ErrNoCondition     = errors.New("no condition")
	ErrSomeStmtsFailed = errors.New("some statements failed")
)

Types

type AggregateReducer

type AggregateReducer struct {
	Aggregate     eventstore.AggregateType
	EventRedusers []EventReducer
}

EventReducer represents the required data to work with aggregates

type Column

type Column struct {
	Name         string
	Value        interface{}
	ParameterOpt func(string) string
}

func NewCol

func NewCol(name string, value interface{}) Column

func NewJSONCol

func NewJSONCol(name string, value interface{}) Column

type Condition

type Condition Column

func NewCond

func NewCond(name string, value interface{}) Condition

type EventReducer

type EventReducer struct {
	Event  eventstore.EventType
	Reduce Reduce
}

EventReducer represents the required data to work with events

type Executer

type Executer interface {
	Exec(string, ...interface{}) (sql.Result, error)
}

type Handler

type Handler struct {
	Eventstore *eventstore.Eventstore
	Sub        *eventstore.Subscription
	EventQueue chan eventstore.Event
}

func NewHandler

func NewHandler(config HandlerConfig) Handler

func (*Handler) Subscribe

func (h *Handler) Subscribe(aggregates ...eventstore.AggregateType)

func (*Handler) SubscribeEvents

func (h *Handler) SubscribeEvents(types map[eventstore.AggregateType][]eventstore.EventType)

type HandlerConfig

type HandlerConfig struct {
	Eventstore *eventstore.Eventstore
}

type Lock

type Lock func(context.Context, time.Duration) <-chan error

Lock is used for mutex handling if needed on the projection

type ProjectionHandler

type ProjectionHandler struct {
	Handler

	ProjectionName string
	// contains filtered or unexported fields
}

func NewProjectionHandler

func NewProjectionHandler(
	config ProjectionHandlerConfig,
	reduce Reduce,
	update Update,
	query SearchQuery,
) *ProjectionHandler

func (*ProjectionHandler) Process

func (h *ProjectionHandler) Process(
	ctx context.Context,
	reduce Reduce,
	update Update,
	lock Lock,
	unlock Unlock,
)

Process waits for several conditions: if context is canceled the function gracefully shuts down if an event occures it reduces the event if the internal timer expires the handler will check for unprocessed events on eventstore

func (*ProjectionHandler) ResetShouldBulk

func (h *ProjectionHandler) ResetShouldBulk()

func (*ProjectionHandler) TriggerBulk

func (h *ProjectionHandler) TriggerBulk(
	ctx context.Context,
	lock Lock,
	unlock Unlock,
) error

type ProjectionHandlerConfig

type ProjectionHandlerConfig struct {
	HandlerConfig
	ProjectionName   string
	RequeueEvery     time.Duration
	RetryFailedAfter time.Duration
}

type Reduce

type Reduce func(eventstore.Event) (*Statement, error)

Reduce reduces the given event to a statement which is used to update the projection

type SearchQuery

type SearchQuery func() (query *eventstore.SearchQueryBuilder, queryLimit uint64, err error)

SearchQuery generates the search query to lookup for events

type Statement

type Statement struct {
	AggregateType    eventstore.AggregateType
	Sequence         uint64
	PreviousSequence uint64

	Execute func(ex Executer, projectionName string) error
}

func (*Statement) IsNoop

func (s *Statement) IsNoop() bool

type Statements

type Statements []Statement

func (Statements) Len

func (stmts Statements) Len() int

func (Statements) Less

func (stmts Statements) Less(i, j int) bool

func (Statements) Swap

func (stmts Statements) Swap(i, j int)

type Unlock

type Unlock func() error

Unlock releases the mutex of the projection

type Update

type Update func(context.Context, []*Statement, Reduce) (unexecutedStmts []*Statement, err error)

Update updates the projection with the given statements