crdb

Documentation

Index

• func AddCreateStatement(columns []handler.Column, opts ...execOption) func(eventstore.Event) Exec
• func AddDeleteStatement(conditions []handler.Condition, opts ...execOption) func(eventstore.Event) Exec
• func AddUpdateStatement(values []handler.Column, conditions []handler.Condition, opts ...execOption) func(eventstore.Event) Exec
• func AddUpsertStatement(values []handler.Column, opts ...execOption) func(eventstore.Event) Exec
• func NewArrayAppendCol(column string, value interface{}) handler.Column
• func NewArrayIntersectCol(column string, value interface{}) handler.Column
• func NewArrayRemoveCol(column string, value interface{}) handler.Column
• func NewCopyStatement(event eventstore.Event, cols []handler.Column, conds []handler.Condition, ...) *handler.Statement
• func NewCreateStatement(event eventstore.Event, values []handler.Column, opts ...execOption) *handler.Statement
• func NewDeleteStatement(event eventstore.Event, conditions []handler.Condition, opts ...execOption) *handler.Statement
• func NewMultiStatement(event eventstore.Event, opts ...func(eventstore.Event) Exec) *handler.Statement
• func NewNoOpStatement(event eventstore.Event) *handler.Statement
• func NewUpdateStatement(event eventstore.Event, values []handler.Column, ...) *handler.Statement
• func NewUpsertStatement(event eventstore.Event, values []handler.Column, opts ...execOption) *handler.Statement
• func WithTableSuffix(name string) func(*execConfig)
• type Exec
• type Locker
  • func NewLocker(client *sql.DB, lockTable, projectionName string) Locker
• type StatementHandler
  • func NewStatementHandler(ctx context.Context, config StatementHandlerConfig) StatementHandler
  • func (h *StatementHandler) SearchQuery() (*eventstore.SearchQueryBuilder, uint64, error)
  • func (h *StatementHandler) TriggerBulk(ctx context.Context)
  • func (h *StatementHandler) Update(ctx context.Context, stmts []*handler.Statement, reduce handler.Reduce) (unexecutedStmts []*handler.Statement, err error)
• type StatementHandlerConfig

Functions

func AddCreateStatement

func AddCreateStatement(columns []handler.Column, opts ...execOption) func(eventstore.Event) Exec

func AddDeleteStatement

func AddDeleteStatement(conditions []handler.Condition, opts ...execOption) func(eventstore.Event) Exec

func AddUpdateStatement

func AddUpdateStatement(values []handler.Column, conditions []handler.Condition, opts ...execOption) func(eventstore.Event) Exec

func AddUpsertStatement

func AddUpsertStatement(values []handler.Column, opts ...execOption) func(eventstore.Event) Exec

func NewArrayAppendCol

func NewArrayAppendCol(column string, value interface{}) handler.Column

func NewArrayIntersectCol

func NewArrayIntersectCol(column string, value interface{}) handler.Column

func NewArrayRemoveCol

func NewArrayRemoveCol(column string, value interface{}) handler.Column

func NewCopyStatement

func NewCopyStatement(event eventstore.Event, cols []handler.Column, conds []handler.Condition, opts ...execOption) *handler.Statement

NewCopyStatement creates a new upsert statement which updates a column from an existing row cols represent the columns which are objective to change. if the value of a col is empty the data will be copied from the selected row if the value of a col is not empty the data will be set by the static value conds represent the conditions for the selection subquery

func NewCreateStatement

func NewCreateStatement(event eventstore.Event, values []handler.Column, opts ...execOption) *handler.Statement

func NewDeleteStatement

func NewDeleteStatement(event eventstore.Event, conditions []handler.Condition, opts ...execOption) *handler.Statement

func NewMultiStatement

func NewMultiStatement(event eventstore.Event, opts ...func(eventstore.Event) Exec) *handler.Statement

func NewNoOpStatement

func NewNoOpStatement(event eventstore.Event) *handler.Statement

func NewUpdateStatement

func NewUpdateStatement(event eventstore.Event, values []handler.Column, conditions []handler.Condition, opts ...execOption) *handler.Statement

func NewUpsertStatement

func NewUpsertStatement(event eventstore.Event, values []handler.Column, opts ...execOption) *handler.Statement

func WithTableSuffix

func WithTableSuffix(name string) func(*execConfig)

Types

type Exec

type Exec func(ex handler.Executer, projectionName string) error

type Locker

type Locker interface {
	Lock(ctx context.Context, lockDuration time.Duration) <-chan error
	Unlock() error
}

func NewLocker

func NewLocker(client *sql.DB, lockTable, projectionName string) Locker

type StatementHandler

type StatementHandler struct {
	*handler.ProjectionHandler
	Locker
	// contains filtered or unexported fields
}

func NewStatementHandler

func NewStatementHandler(
	ctx context.Context,
	config StatementHandlerConfig,
) StatementHandler

func (*StatementHandler) SearchQuery

func (h *StatementHandler) SearchQuery() (*eventstore.SearchQueryBuilder, uint64, error)

func (*StatementHandler) TriggerBulk

func (h *StatementHandler) TriggerBulk(ctx context.Context)

func (*StatementHandler) Update

func (h *StatementHandler) Update(ctx context.Context, stmts []*handler.Statement, reduce handler.Reduce) (unexecutedStmts []*handler.Statement, err error)

Update implements handler.Update

type StatementHandlerConfig

type StatementHandlerConfig struct {
	handler.ProjectionHandlerConfig

	Client            *sql.DB
	SequenceTable     string
	LockTable         string
	FailedEventsTable string
	MaxFailureCount   uint
	BulkLimit         uint64

	Reducers []handler.AggregateReducer
}