Method ConnectBinlogStreamer

driver/mysql/binlog/binlog_reader.go:282–414

(coordinates common.MySQLCoordinates)

}

func (b *BinlogReader) ConnectBinlogStreamer(coordinates common.MySQLCoordinates) (err error) {
	if coordinates.IsEmpty() {
		b.logger.Warn("Emptry coordinates at ConnectBinlogStreamer")
	}

	b.currentCoord = coordinates
	b.logger.Info("Connecting binlog streamer",
		"file", coordinates.LogFile, "pos", coordinates.LogPos, "gtid", coordinates.GtidSet)

	if b.mysqlContext.BinlogRelay {
		dbConfig := dmconfig.DBConfig{
			Host:     b.mysqlContext.SrcConnectionConfig.Host,
			Port:     b.mysqlContext.SrcConnectionConfig.Port,
			User:     b.mysqlContext.SrcConnectionConfig.User,
			Password: b.mysqlContext.SrcConnectionConfig.Password,
		}

		relayConfig := &dmrelay.Config{
			EnableGTID: true,
			RelayDir:   b.getBinlogDir(),
			ServerID:   uint32(b.serverId),
			Flavor:     "mysql",
			From:       dbConfig,
			BinLogName: "",
			BinlogGTID: coordinates.GtidSet,
			ReaderRetry: dmretry.ReaderRetryConfig{
				// value from dm/relay/relay_test.go
				BackoffRollback: 200 * time.Millisecond,
				BackoffMax:      1 * time.Second,
				BackoffMin:      1 * time.Millisecond,
				BackoffJitter:   true,
				BackoffFactor:   2,
			},
		}
		b.relay = dmrelay.NewRelay(relayConfig)
		err = b.relay.Init(b.ctx)
		if err != nil {
			return err
		}

		var ctx context.Context
		ctx, b.relayCancelF = context.WithCancel(b.ctx)

		{
			brConfig := &dmstreamer.BinlogReaderConfig{
				RelayDir: b.getBinlogDir(),
				Timezone: time.UTC,
			}
			b.binlogReader = dmstreamer.NewBinlogReader(dmlog.L(), brConfig)
		}

		go func() {
			b.logger.Info("starting BinlogRelay", "gtidSet", coordinates.GtidSet)
			pr := b.relay.Process(ctx)
			b.logger.Warn("relay.Process stopped", "isCancelled", pr.IsCanceled, "deail", string(pr.Detail))
			for _, prErr := range pr.Errors {
				b.logger.Error("relay.Process error", "err", prErr)
			}