From 7207bc146a1cf144b15ce5ca4fdcd0a786d57729 Mon Sep 17 00:00:00 2001 From: Cathal Coffey Date: Sun, 31 Jan 2021 18:23:09 +0000 Subject: [PATCH] Make it easier to handle different onChangelogEvents --- go/logic/migrator.go | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/go/logic/migrator.go b/go/logic/migrator.go index 291a490..908a42f 100644 --- a/go/logic/migrator.go +++ b/go/logic/migrator.go @@ -207,12 +207,18 @@ func (this *Migrator) canStopStreaming() bool { return atomic.LoadInt64(&this.migrationContext.CutOverCompleteFlag) != 0 } -// onChangelogStateEvent is called when a binlog event operation on the changelog table is intercepted. -func (this *Migrator) onChangelogStateEvent(dmlEvent *binlog.BinlogDMLEvent) (err error) { +// onChangelogEvent is called when a binlog event operation on the changelog table is intercepted. +func (this *Migrator) onChangelogEvent(dmlEvent *binlog.BinlogDMLEvent) (err error) { // Hey, I created the changelog table, I know the type of columns it has! - if hint := dmlEvent.NewColumnValues.StringColumn(2); hint != "state" { + switch hint := dmlEvent.NewColumnValues.StringColumn(2); hint { + case "state": + return this.onChangelogStateEvent(dmlEvent) + default: return nil } +} + +func (this *Migrator) onChangelogStateEvent(dmlEvent *binlog.BinlogDMLEvent) (err error) { changelogStateString := dmlEvent.NewColumnValues.StringColumn(3) changelogState := ReadChangelogState(changelogStateString) this.migrationContext.Log.Infof("Intercepted changelog state %s", changelogState) @@ -995,7 +1001,7 @@ func (this *Migrator) initiateStreaming() error { this.migrationContext.DatabaseName, this.migrationContext.GetChangelogTableName(), func(dmlEvent *binlog.BinlogDMLEvent) error { - return this.onChangelogStateEvent(dmlEvent) + return this.onChangelogEvent(dmlEvent) }, )