TheArchive/zitadel
1
0
Fork 0
mirror of https://github.com/zitadel/zitadel.git synced 2025-08-12 04:57:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(eventstore): rename EventPusher to Command, EventReader to Event, PushEvents to Push and FilterEvents to Filter (#2907)

Browse Source
This commit is contained in:
Silvan
2022-01-03 09:19:07 +01:00
committed by GitHub
parent 9a374f9c5c
commit 09be70949f
339 changed files with 1436 additions and 1346 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
internal/eventstore/handler/crdb/handler_stmt.go
View File

@@ -196,7 +196,7 @@ func (h *StatementHandler) fetchPreviousStmts(
return nil, nil
}
events, err := h.Eventstore.FilterEvents(ctx, query)
events, err := h.Eventstore.Filter(ctx, query)
if err != nil {
return nil, err
}

6
internal/eventstore/handler/crdb/handler_stmt_test.go
View File

@@ -21,7 +21,7 @@ var (
errReduce = errors.New("reduce err")
)
var _ eventstore.EventReader = &testEvent{}
var _ eventstore.Event = &testEvent{}
type testEvent struct {
eventstore.BaseEvent
@@ -1560,13 +1560,13 @@ func TestStatementHandler_updateCurrentSequence(t *testing.T) {
}
func testReduce() handler.Reduce {
return func(event eventstore.EventReader) (*handler.Statement, error) {
return func(event eventstore.Event) (*handler.Statement, error) {
return NewNoOpStatement(event), nil
}
}
func testReduceErr(err error) handler.Reduce {
return func(event eventstore.EventReader) (*handler.Statement, error) {
return func(event eventstore.Event) (*handler.Statement, error) {
return nil, err
}
}

2
internal/eventstore/handler/crdb/reduce.go
View File

@@ -6,7 +6,7 @@ import (
)
//reduce implements handler.Reduce function
func (h *StatementHandler) reduce(event eventstore.EventReader) (*handler.Statement, error) {
func (h *StatementHandler) reduce(event eventstore.Event) (*handler.Statement, error) {
reduce, ok := h.reduces[event.Type()]
if !ok {
return NewNoOpStatement(event), nil

30
internal/eventstore/handler/crdb/statement.go
View File

@@ -23,7 +23,7 @@ func WithTableSuffix(name string) func(*execConfig) {
}
}
func NewCreateStatement(event eventstore.EventReader, values []handler.Column, opts ...execOption) *handler.Statement {
func NewCreateStatement(event eventstore.Event, values []handler.Column, opts ...execOption) *handler.Statement {
cols, params, args := columnsToQuery(values)
columnNames := strings.Join(cols, ", ")
valuesPlaceholder := strings.Join(params, ", ")
@@ -48,7 +48,7 @@ func NewCreateStatement(event eventstore.EventReader, values []handler.Column, o
}
}
func NewUpsertStatement(event eventstore.EventReader, values []handler.Column, opts ...execOption) *handler.Statement {
func NewUpsertStatement(event eventstore.Event, values []handler.Column, opts ...execOption) *handler.Statement {
cols, params, args := columnsToQuery(values)
columnNames := strings.Join(cols, ", ")
valuesPlaceholder := strings.Join(params, ", ")
@@ -73,7 +73,7 @@ func NewUpsertStatement(event eventstore.EventReader, values []handler.Column, o
}
}
func NewUpdateStatement(event eventstore.EventReader, values []handler.Column, conditions []handler.Condition, opts ...execOption) *handler.Statement {
func NewUpdateStatement(event eventstore.Event, values []handler.Column, conditions []handler.Condition, opts ...execOption) *handler.Statement {
cols, params, args := columnsToQuery(values)
wheres, whereArgs := conditionsToWhere(conditions, len(params))
args = append(args, whereArgs...)
@@ -106,7 +106,7 @@ func NewUpdateStatement(event eventstore.EventReader, values []handler.Column, c
}
}
func NewDeleteStatement(event eventstore.EventReader, conditions []handler.Condition, opts ...execOption) *handler.Statement {
func NewDeleteStatement(event eventstore.Event, conditions []handler.Condition, opts ...execOption) *handler.Statement {
wheres, args := conditionsToWhere(conditions, 0)
wheresPlaceholders := strings.Join(wheres, " AND ")
@@ -131,7 +131,7 @@ func NewDeleteStatement(event eventstore.EventReader, conditions []handler.Condi
}
}
func NewNoOpStatement(event eventstore.EventReader) *handler.Statement {
func NewNoOpStatement(event eventstore.Event) *handler.Statement {
return &handler.Statement{
AggregateType: event.Aggregate().Type,
Sequence: event.Sequence(),
@@ -139,7 +139,7 @@ func NewNoOpStatement(event eventstore.EventReader) *handler.Statement {
}
}
func NewMultiStatement(event eventstore.EventReader, opts ...func(eventstore.EventReader) Exec) *handler.Statement {
func NewMultiStatement(event eventstore.Event, opts ...func(eventstore.Event) Exec) *handler.Statement {
if len(opts) == 0 {
return NewNoOpStatement(event)
}
@@ -157,26 +157,26 @@ func NewMultiStatement(event eventstore.EventReader, opts ...func(eventstore.Eve
type Exec func(ex handler.Executer, projectionName string) error
func AddCreateStatement(columns []handler.Column, opts ...execOption) func(eventstore.EventReader) Exec {
return func(event eventstore.EventReader) Exec {
func AddCreateStatement(columns []handler.Column, opts ...execOption) func(eventstore.Event) Exec {
return func(event eventstore.Event) Exec {
return NewCreateStatement(event, columns, opts...).Execute
}
}
func AddUpsertStatement(values []handler.Column, opts ...execOption) func(eventstore.EventReader) Exec {
return func(event eventstore.EventReader) Exec {
func AddUpsertStatement(values []handler.Column, opts ...execOption) func(eventstore.Event) Exec {
return func(event eventstore.Event) Exec {
return NewUpsertStatement(event, values, opts...).Execute
}
}
func AddUpdateStatement(values []handler.Column, conditions []handler.Condition, opts ...execOption) func(eventstore.EventReader) Exec {
return func(event eventstore.EventReader) Exec {
func AddUpdateStatement(values []handler.Column, conditions []handler.Condition, opts ...execOption) func(eventstore.Event) Exec {
return func(event eventstore.Event) Exec {
return NewUpdateStatement(event, values, conditions, opts...).Execute
}
}
func AddDeleteStatement(conditions []handler.Condition, opts ...execOption) func(eventstore.EventReader) Exec {
return func(event eventstore.EventReader) Exec {
func AddDeleteStatement(conditions []handler.Condition, opts ...execOption) func(eventstore.Event) Exec {
return func(event eventstore.Event) Exec {
return NewDeleteStatement(event, conditions, opts...).Execute
}
}
@@ -206,7 +206,7 @@ func NewArrayRemoveCol(column string, value interface{}) handler.Column {
// 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 NewCopyStatement(event eventstore.EventReader, cols []handler.Column, conds []handler.Condition, opts ...execOption) *handler.Statement {
func NewCopyStatement(event eventstore.Event, cols []handler.Column, conds []handler.Condition, opts ...execOption) *handler.Statement {
columnNames := make([]string, len(cols))
selectColumns := make([]string, len(cols))
argCounter := 0

6
internal/eventstore/handler/crdb/statement_test.go
View File

@@ -631,7 +631,7 @@ func TestNewMultiStatement(t *testing.T) {
type args struct {
table string
event *testEvent
execs []func(eventstore.EventReader) Exec
execs []func(eventstore.Event) Exec
}
type want struct {
@@ -671,7 +671,7 @@ func TestNewMultiStatement(t *testing.T) {
sequence: 1,
previousSequence: 0,
},
execs: []func(eventstore.EventReader) Exec{
execs: []func(eventstore.Event) Exec{
AddDeleteStatement(
[]handler.Condition{},
),
@@ -706,7 +706,7 @@ func TestNewMultiStatement(t *testing.T) {
previousSequence: 0,
aggregateType: "agg",
},
execs: []func(eventstore.EventReader) Exec{
execs: []func(eventstore.Event) Exec{
AddDeleteStatement(
[]handler.Condition{
{

4
internal/eventstore/handler/handler.go
View File

@@ -10,13 +10,13 @@ type HandlerConfig struct {
type Handler struct {
Eventstore *eventstore.Eventstore
Sub *eventstore.Subscription
EventQueue chan eventstore.EventReader
EventQueue chan eventstore.Event
}
func NewHandler(config HandlerConfig) Handler {
return Handler{
Eventstore: config.Eventstore,
EventQueue: make(chan eventstore.EventReader, 100),
EventQueue: make(chan eventstore.Event, 100),
}
}

6
internal/eventstore/handler/handler_projection.go
View File

@@ -23,7 +23,7 @@ type Update func(context.Context, []*Statement, Reduce) (unexecutedStmts []*Stat
//Reduce reduces the given event to a statement
//which is used to update the projection
type Reduce func(eventstore.EventReader) (*Statement, error)
type Reduce func(eventstore.Event) (*Statement, error)
//Lock is used for mutex handling if needed on the projection
type Lock func(context.Context, time.Duration) <-chan error
@@ -156,7 +156,7 @@ func (h *ProjectionHandler) Process(
func (h *ProjectionHandler) processEvent(
ctx context.Context,
event eventstore.EventReader,
event eventstore.Event,
reduce Reduce,
) error {
stmt, err := reduce(event)
@@ -262,7 +262,7 @@ func (h *ProjectionHandler) fetchBulkStmts(
return false, err
}
events, err := h.Eventstore.FilterEvents(ctx, eventQuery)
events, err := h.Eventstore.Filter(ctx, eventQuery)
if err != nil {
logging.LogWithFields("HANDL-X8vlo", "projection", h.ProjectionName).WithError(err).Info("Unable to bulk fetch events")
return false, err

6
internal/eventstore/handler/handler_projection_test.go
View File

@@ -46,7 +46,7 @@ func TestProjectionHandler_processEvent(t *testing.T) {
}
type args struct {
ctx context.Context
event eventstore.EventReader
event eventstore.Event
reduce Reduce
}
type want struct {
@@ -849,13 +849,13 @@ func testUpdate(t *testing.T, expectedStmtCount int, returnedErr error) Update {
}
func testReduce(stmts *Statement) Reduce {
return func(event eventstore.EventReader) (*Statement, error) {
return func(event eventstore.Event) (*Statement, error) {
return stmts, nil
}
}
func testReduceErr(err error) Reduce {
return func(event eventstore.EventReader) (*Statement, error) {
return func(event eventstore.Event) (*Statement, error) {
return nil, err
}
}