Merge pull request #4 from lorenzoranucci/make-checkpointing-async

make checkpointing async

Makefile

@@ -35,4 +35,4 @@ test:
 	@find . -name go.mod \
      | sed 's/\/[^\/]*$$//' \
      | xargs -I {} bash -c \
-     'echo "testing {}" && go test {}/...'
+     'echo "testing {}" && go test --race {}/...'

example/tor/cmd/run.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"os"
 	"regexp"
+	"time"
 
 	"github.com/go-mysql-org/go-mysql/canal"
 	"github.com/go-redis/redis/v8"
@@ -40,7 +41,6 @@ var runCmd = &cobra.Command{
 
 		stateHandler := getRedisStateHandler()
 		handler, err := run.NewEventHandler(
-			stateHandler,
 			ed,
 			viper.GetString("dbAggregateIDColumnName"),
 			viper.GetString("dbAggregateTypeColumnName"),
@@ -51,7 +51,7 @@ var runCmd = &cobra.Command{
 			return err
 		}
 
-		runner := run.NewRunner(c, handler, stateHandler)
+		runner := run.NewRunner(c, handler, stateHandler, time.Second*5)
 
 		return runner.Run()
 	},
@@ -114,6 +114,8 @@ func getCanalConfig() *canal.Config {
 	cfg.Password = viper.GetString("dbPassword")
 	cfg.Dump.ExecutionPath = ""
 	cfg.IncludeTableRegex = []string{fmt.Sprintf("^%s$", viper.Get("dbOutboxTableRef"))}
+	cfg.MaxReconnectAttempts = 10
+	cfg.ReadTimeout = time.Second * 10
 
 	return cfg
 }

example/tor/go.mod

@@ -6,7 +6,7 @@ require (
 	github.com/go-mysql-org/go-mysql v1.6.0
 	github.com/go-redis/redis/v8 v8.11.5
 	github.com/lorenzoranucci/tor/adapters/kafka v0.1.0
-	github.com/lorenzoranucci/tor/adapters/redis v0.1.0
+	github.com/lorenzoranucci/tor/adapters/redis v0.2.0
 	github.com/lorenzoranucci/tor/router v0.2.0
 	github.com/sirupsen/logrus v1.9.0
 	github.com/spf13/cobra v1.6.1

go.work.sum

@@ -1,3 +1,4 @@
+github.com/lorenzoranucci/tor/adapters/redis v0.2.0/go.mod h1:U6idye39F8NfHtPeBv0MYD690+gnN75z9DDOe7XDNgY=
 golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
 golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=

router/pkg/run/event_handler.go

@@ -32,7 +32,6 @@ type EventDispatcher interface {
 }
 
 func NewEventHandler(
-	stateHandler StateHandler,
 	eventDispatcher EventDispatcher,
 	aggregateIDColumnName string,
 	aggregateTypeColumnName string,
@@ -60,7 +59,6 @@ func NewEventHandler(
 	}
 
 	return &EventHandler{
-		stateHandler: stateHandler,
 		eventMapper: &EventMapper{
 			aggregateIDColumnName:   actualAggregateIDColumnName,
 			aggregateTypeColumnName: actualAggregateTypeColumnName,
@@ -74,9 +72,9 @@ func NewEventHandler(
 type EventHandler struct {
 	canal.DummyEventHandler
 
-	stateHandler    StateHandler
 	eventMapper     *EventMapper
 	eventDispatcher EventDispatcher
+	positionChan    chan mysql.Position
 }
 
 func (h *EventHandler) OnRow(e *canal.RowsEvent) error {
@@ -105,20 +103,10 @@ func (h *EventHandler) OnRow(e *canal.RowsEvent) error {
 }
 
 func (h *EventHandler) OnPosSynced(p mysql.Position, g mysql.GTIDSet, f bool) error {
-	return h.setLastPosition(p)
+	h.positionChan <- p
+	return nil
 }
 
 func (h *EventHandler) String() string {
 	return "EventHandler"
 }
-
-func (h *EventHandler) setLastPosition(p mysql.Position) error {
-	err := h.stateHandler.SetLastPosition(p)
-	if err != nil {
-		return err
-	}
-	logrus.WithField("lastPosition", p).
-		Debug("last position set")
-
-	return nil
-}

router/pkg/run/event_handler_test.go

@@ -7,7 +7,6 @@ import (
 	"testing"
 
 	"github.com/go-mysql-org/go-mysql/canal"
-	"github.com/go-mysql-org/go-mysql/mysql"
 	"github.com/go-mysql-org/go-mysql/replication"
 	"github.com/go-mysql-org/go-mysql/schema"
 	"github.com/lorenzoranucci/tor/router/pkg/run"
@@ -16,8 +15,7 @@ import (
 
 func TestEventHandler_OnRow_HappyPaths(t *testing.T) {
 	type fields struct {
-		stateHandler            *StateHandlerStub
-		eventDispatcher         *EventDispatcherMock
+		eventDispatcher         *eventDispatcherMock
 		aggregateIdColumnName   string
 		aggregateTypeColumnName string
 		payloadColumnName       string
@@ -64,7 +62,7 @@ func TestEventHandler_OnRow_HappyPaths(t *testing.T) {
 				},
 			},
 			fields: fields{
-				eventDispatcher: &EventDispatcherMock{},
+				eventDispatcher: &eventDispatcherMock{},
 			},
 			wantDispatches: []dispatch{
 				{
@@ -104,7 +102,7 @@ func TestEventHandler_OnRow_HappyPaths(t *testing.T) {
 				},
 			},
 			fields: fields{
-				eventDispatcher: &EventDispatcherMock{},
+				eventDispatcher: &eventDispatcherMock{},
 			},
 			wantDispatches: []dispatch{
 				{
@@ -148,7 +146,7 @@ func TestEventHandler_OnRow_HappyPaths(t *testing.T) {
 				},
 			},
 			fields: fields{
-				eventDispatcher:         &EventDispatcherMock{},
+				eventDispatcher:         &eventDispatcherMock{},
 				aggregateIdColumnName:   "aggregateId",
 				aggregateTypeColumnName: "aggregateType",
 				payloadColumnName:       "payload_",
@@ -196,7 +194,7 @@ func TestEventHandler_OnRow_HappyPaths(t *testing.T) {
 				},
 			},
 			fields: fields{
-				eventDispatcher: &EventDispatcherMock{},
+				eventDispatcher: &eventDispatcherMock{},
 			},
 			wantDispatches: []dispatch{
 				{
@@ -245,7 +243,7 @@ func TestEventHandler_OnRow_HappyPaths(t *testing.T) {
 				},
 			},
 			fields: fields{
-				eventDispatcher:     &EventDispatcherMock{},
+				eventDispatcher:     &eventDispatcherMock{},
 				aggregateTypeRegexp: regexp.MustCompile("(?i)^order$"),
 			},
 			wantDispatches: []dispatch{
@@ -260,7 +258,6 @@ func TestEventHandler_OnRow_HappyPaths(t *testing.T) {
 		tt := tt
 		t.Run(tt.name, func(t *testing.T) {
 			h, err := run.NewEventHandler(
-				tt.fields.stateHandler,
 				tt.fields.eventDispatcher,
 				tt.fields.aggregateIdColumnName,
 				tt.fields.aggregateTypeColumnName,
@@ -282,7 +279,7 @@ func TestEventHandler_OnRow_HappyPaths(t *testing.T) {
 
 func TestEventHandler_OnRow_UnhappyPaths(t *testing.T) {
 	type fields struct {
-		eventDispatcher         *EventDispatcherMock
+		eventDispatcher         *eventDispatcherMock
 		aggregateIdColumnName   string
 		aggregateTypeColumnName string
 		payloadColumnName       string
@@ -330,7 +327,7 @@ func TestEventHandler_OnRow_UnhappyPaths(t *testing.T) {
 				},
 			},
 			fields: fields{
-				eventDispatcher: &EventDispatcherMock{},
+				eventDispatcher: &eventDispatcherMock{},
 			},
 		},
 		{
@@ -364,7 +361,7 @@ func TestEventHandler_OnRow_UnhappyPaths(t *testing.T) {
 				},
 			},
 			fields: fields{
-				eventDispatcher: &EventDispatcherMock{
+				eventDispatcher: &eventDispatcherMock{
 					err: errors.New(""),
 				},
 			},
@@ -406,7 +403,7 @@ func TestEventHandler_OnRow_UnhappyPaths(t *testing.T) {
 				},
 			},
 			fields: fields{
-				eventDispatcher: &EventDispatcherMock{},
+				eventDispatcher: &eventDispatcherMock{},
 			},
 			wantErr: true,
 		},
@@ -441,7 +438,7 @@ func TestEventHandler_OnRow_UnhappyPaths(t *testing.T) {
 				},
 			},
 			fields: fields{
-				eventDispatcher: &EventDispatcherMock{},
+				eventDispatcher: &eventDispatcherMock{},
 			},
 			wantErr: true,
 		},
@@ -476,7 +473,7 @@ func TestEventHandler_OnRow_UnhappyPaths(t *testing.T) {
 				},
 			},
 			fields: fields{
-				eventDispatcher: &EventDispatcherMock{},
+				eventDispatcher: &eventDispatcherMock{},
 			},
 			wantErr: true,
 		},
@@ -511,7 +508,7 @@ func TestEventHandler_OnRow_UnhappyPaths(t *testing.T) {
 				},
 			},
 			fields: fields{
-				eventDispatcher: &EventDispatcherMock{},
+				eventDispatcher: &eventDispatcherMock{},
 			},
 			wantErr: true,
 		},
@@ -546,7 +543,7 @@ func TestEventHandler_OnRow_UnhappyPaths(t *testing.T) {
 				},
 			},
 			fields: fields{
-				eventDispatcher: &EventDispatcherMock{},
+				eventDispatcher: &eventDispatcherMock{},
 			},
 			wantErr: true,
 		},
@@ -581,7 +578,7 @@ func TestEventHandler_OnRow_UnhappyPaths(t *testing.T) {
 				},
 			},
 			fields: fields{
-				eventDispatcher: &EventDispatcherMock{},
+				eventDispatcher: &eventDispatcherMock{},
 			},
 			wantErr: true,
 		},
@@ -616,7 +613,7 @@ func TestEventHandler_OnRow_UnhappyPaths(t *testing.T) {
 				},
 			},
 			fields: fields{
-				eventDispatcher: &EventDispatcherMock{},
+				eventDispatcher: &eventDispatcherMock{},
 			},
 			wantErr: true,
 		},
@@ -625,7 +622,6 @@ func TestEventHandler_OnRow_UnhappyPaths(t *testing.T) {
 		tt := tt
 		t.Run(tt.name, func(t *testing.T) {
 			h, err := run.NewEventHandler(
-				&StateHandlerStub{},
 				tt.fields.eventDispatcher,
 				tt.fields.aggregateIdColumnName,
 				tt.fields.aggregateTypeColumnName,
@@ -656,26 +652,16 @@ type dispatch struct {
 	event      []byte
 }
 
-type EventDispatcherMock struct {
+type eventDispatcherMock struct {
 	dispatches []dispatch
 	err        error
 }
 
-func (e *EventDispatcherMock) Dispatch(routingKey string, event []byte) error {
+func (e *eventDispatcherMock) Dispatch(routingKey string, event []byte) error {
 	e.dispatches = append(e.dispatches, dispatch{
 		routingKey: routingKey,
 		event:      event,
 	})
 
 	return e.err
 }
-
-type StateHandlerStub struct{}
-
-func (s *StateHandlerStub) GetLastPosition() (mysql.Position, error) {
-	return mysql.Position{}, nil
-}
-
-func (s *StateHandlerStub) SetLastPosition(_ mysql.Position) error {
-	return nil
-}

router/pkg/run/run.go

@@ -1,18 +1,39 @@
 package run
 
 import (
+	"context"
+	"time"
+
 	"github.com/go-mysql-org/go-mysql/canal"
+	"github.com/go-mysql-org/go-mysql/mysql"
+	"github.com/sirupsen/logrus"
 )
 
-func NewRunner(canal *canal.Canal, handler canal.EventHandler, stateHandler StateHandler) *Runner {
+func NewRunner(
+	canal Canal,
+	handler *EventHandler,
+	stateHandler StateHandler,
+	stateUpdateFrequency time.Duration,
+) *Runner {
+	p := make(chan mysql.Position)
+	handler.positionChan = p
+
 	canal.SetEventHandler(handler)
 
-	return &Runner{canal: canal, stateHandler: stateHandler}
+	return &Runner{canal: canal, stateHandler: stateHandler, positionChan: p, stateUpdateFrequency: stateUpdateFrequency}
 }
 
 type Runner struct {
-	canal        *canal.Canal
-	stateHandler StateHandler
+	canal                Canal
+	stateHandler         StateHandler
+	positionChan         chan mysql.Position
+	stateUpdateFrequency time.Duration
+}
+
+type Canal interface {
+	RunFrom(mysql.Position) error
+	SetEventHandler(handler canal.EventHandler)
+	Close()
 }
 
 func (r *Runner) Run() error {
@@ -21,5 +42,43 @@ func (r *Runner) Run() error {
 		return err
 	}
 
-	return r.canal.RunFrom(lastPosition)
+	errCh := make(chan error, 2)
+
+	go func() {
+		errCh <- r.canal.RunFrom(lastPosition)
+	}()
+
+	ctx, cf := context.WithCancel(context.Background())
+	ticker := time.NewTicker(r.stateUpdateFrequency)
+	go func() {
+		for {
+			select {
+			case lastPosition = <-r.positionChan:
+			case <-ticker.C:
+				err := r.setLastPosition(lastPosition)
+				if err != nil {
+					errCh <- err
+				}
+			case <-ctx.Done():
+				_ = r.setLastPosition(lastPosition)
+				return
+			}
+		}
+	}()
+
+	err = <-errCh
+	r.canal.Close()
+	cf()
+	return err
+}
+
+func (r *Runner) setLastPosition(p mysql.Position) error {
+	err := r.stateHandler.SetLastPosition(p)
+	if err != nil {
+		return err
+	}
+	logrus.WithField("lastPosition", p).
+		Debug("last position set")
+
+	return nil
 }