Use switch statements for readability, simplify .NewGoMySQLReader() (#1135)

* Use `switch` statements for readability

* Simplify initBinlogReader()

Author: timvaillancourt

go/binlog/gomysql_reader.go

@@ -28,31 +28,24 @@ type GoMySQLReader struct {
 	LastAppliedRowsEventHint mysql.BinlogCoordinates
 }
 
-func NewGoMySQLReader(migrationContext *base.MigrationContext) (binlogReader *GoMySQLReader, err error) {
-	binlogReader = &GoMySQLReader{
+func NewGoMySQLReader(migrationContext *base.MigrationContext) *GoMySQLReader {
+	connectionConfig := migrationContext.InspectorConnectionConfig
+	return &GoMySQLReader{
 		migrationContext:        migrationContext,
-		connectionConfig:        migrationContext.InspectorConnectionConfig,
+		connectionConfig:        connectionConfig,
 		currentCoordinates:      mysql.BinlogCoordinates{},
 		currentCoordinatesMutex: &sync.Mutex{},
-		binlogSyncer:            nil,
-		binlogStreamer:          nil,
+		binlogSyncer: replication.NewBinlogSyncer(replication.BinlogSyncerConfig{
+			ServerID:   uint32(migrationContext.ReplicaServerId),
+			Flavor:     gomysql.MySQLFlavor,
+			Host:       connectionConfig.Key.Hostname,
+			Port:       uint16(connectionConfig.Key.Port),
+			User:       connectionConfig.User,
+			Password:   connectionConfig.Password,
+			TLSConfig:  connectionConfig.TLSConfig(),
+			UseDecimal: true,
+		}),
 	}
-
-	serverId := uint32(migrationContext.ReplicaServerId)
-
-	binlogSyncerConfig := replication.BinlogSyncerConfig{
-		ServerID:   serverId,
-		Flavor:     "mysql",
-		Host:       binlogReader.connectionConfig.Key.Hostname,
-		Port:       uint16(binlogReader.connectionConfig.Key.Port),
-		User:       binlogReader.connectionConfig.User,
-		Password:   binlogReader.connectionConfig.Password,
-		TLSConfig:  binlogReader.connectionConfig.TLSConfig(),
-		UseDecimal: true,
-	}
-	binlogReader.binlogSyncer = replication.NewBinlogSyncer(binlogSyncerConfig)
-
-	return binlogReader, err
 }
 
 // ConnectBinlogStreamer
@@ -145,15 +138,17 @@ func (this *GoMySQLReader) StreamEvents(canStopStreaming func() bool, entriesCha
 			defer this.currentCoordinatesMutex.Unlock()
 			this.currentCoordinates.LogPos = int64(ev.Header.LogPos)
 		}()
-		if rotateEvent, ok := ev.Event.(*replication.RotateEvent); ok {
+
+		switch binlogEvent := ev.Event.(type) {
+		case *replication.RotateEvent:
 			func() {
 				this.currentCoordinatesMutex.Lock()
 				defer this.currentCoordinatesMutex.Unlock()
-				this.currentCoordinates.LogFile = string(rotateEvent.NextLogName)
+				this.currentCoordinates.LogFile = string(binlogEvent.NextLogName)
 			}()
-			this.migrationContext.Log.Infof("rotate to next log from %s:%d to %s", this.currentCoordinates.LogFile, int64(ev.Header.LogPos), rotateEvent.NextLogName)
-		} else if rowsEvent, ok := ev.Event.(*replication.RowsEvent); ok {
-			if err := this.handleRowsEvent(ev, rowsEvent, entriesChannel); err != nil {
+			this.migrationContext.Log.Infof("rotate to next log from %s:%d to %s", this.currentCoordinates.LogFile, int64(ev.Header.LogPos), binlogEvent.NextLogName)
+		case *replication.RowsEvent:
+			if err := this.handleRowsEvent(ev, binlogEvent, entriesChannel); err != nil {
 				return err
 			}
 		}

go/logic/migrator.go

@@ -537,19 +537,19 @@ func (this *Migrator) cutOver() (err error) {
 			}
 		}
 	}
-	if this.migrationContext.CutOverType == base.CutOverAtomic {
+
+	switch this.migrationContext.CutOverType {
+	case base.CutOverAtomic:
 		// Atomic solution: we use low timeout and multiple attempts. But for
 		// each failed attempt, we throttle until replication lag is back to normal
-		err := this.atomicCutOver()
-		this.handleCutOverResult(err)
-		return err
-	}
-	if this.migrationContext.CutOverType == base.CutOverTwoStep {
-		err := this.cutOverTwoStep()
-		this.handleCutOverResult(err)
-		return err
+		err = this.atomicCutOver()
+	case base.CutOverTwoStep:
+		err = this.cutOverTwoStep()
+	default:
+		return this.migrationContext.Log.Fatalf("Unknown cut-over type: %d; should never get here!", this.migrationContext.CutOverType)
 	}
-	return this.migrationContext.Log.Fatalf("Unknown cut-over type: %d; should never get here!", this.migrationContext.CutOverType)
+	this.handleCutOverResult(err)
+	return err
 }
 
 // Inject the "AllEventsUpToLockProcessed" state hint, wait for it to appear in the binary logs,

go/logic/streamer.go

@@ -123,10 +123,7 @@ func (this *EventsStreamer) InitDBConnections() (err error) {
 
 // initBinlogReader creates and connects the reader: we hook up to a MySQL server as a replica
 func (this *EventsStreamer) initBinlogReader(binlogCoordinates *mysql.BinlogCoordinates) error {
-	goMySQLReader, err := binlog.NewGoMySQLReader(this.migrationContext)
-	if err != nil {
-		return err
-	}
+	goMySQLReader := binlog.NewGoMySQLReader(this.migrationContext)
 	if err := goMySQLReader.ConnectBinlogStreamer(*binlogCoordinates); err != nil {
 		return err
 	}