Function NewBinlogReader

driver/mysql/binlog/binlog_reader.go:180–276

(
	execCtx *common.ExecContext, cfg *common.MySQLDriverConfig, logger g.LoggerType,
	replicateDoDb map[string]*common.SchemaContext, sqleContext *sqle.Context,
	memory *int64, db *gosql.DB, targetGtid string, lctn mysqlconfig.LowerCaseTableNamesValue,
	ctx context.Context)

}

func NewBinlogReader(
	execCtx *common.ExecContext, cfg *common.MySQLDriverConfig, logger g.LoggerType,
	replicateDoDb map[string]*common.SchemaContext, sqleContext *sqle.Context,
	memory *int64, db *gosql.DB, targetGtid string, lctn mysqlconfig.LowerCaseTableNamesValue,
	ctx context.Context) (binlogReader *BinlogReader, err error) {

	sqlFilter, err := parseSqlFilter(cfg.SqlFilter)
	if err != nil {
		return nil, err
	}

	binlogReader = &BinlogReader{
		ctx:                 ctx,
		execCtx:             execCtx,
		logger:              logger,
		currentCoord:        common.MySQLCoordinates{},
		currentCoordMutex:   &sync.Mutex{},
		mysqlContext:        cfg,
		ReMap:               make(map[string]*regexp.Regexp),
		shutdownCh:          make(chan struct{}),
		tables:              make(map[string]*common.SchemaContext),
		sqlFilter:           sqlFilter,
		maybeSqleContext:    sqleContext,
		memory:              memory,
		db:                  db,
		lowerCaseTableNames: lctn,
	}

	binlogReader.serverUUID, err = sql.GetServerUUID(db)
	if err != nil {
		return nil, err
	}

	gset, _ := gomysql.ParseMysqlGTIDSet("")
	binlogReader.CurrentGtidSet = gset.(*gomysql.MysqlGTIDSet)
	if targetGtid != "" {
		binlogReader.targetGtid, err = gomysql.ParseMysqlGTIDSet(targetGtid)
		if err != nil {
			return nil, errors.Wrap(err, "ParseMysqlGTIDSet")
		}
	}

	binlogReader.tables = replicateDoDb

	id, err := util.NewIdWorker(2, 3, util.SnsEpoch)
	if err != nil {
		return nil, err
	}
	sid, err := id.NextId()
	if err != nil {
		return nil, err
	}
	bid := []byte(strconv.FormatUint(uint64(sid), 10))
	binlogReader.serverId, err = strconv.ParseUint(string(bid), 10, 32)
	if err != nil {
		return nil, err
	}
	logger.Debug("got replication serverId", "id", binlogReader.serverId)