// updateSourceStatus updates w.sourceStatus.
updateSourceStatus updates w.sourceStatus.
// updateSourceStatus updates w . sourceStatus .
func (w *SourceWorker) updateSourceStatus(ctx context.Context) error {
	w.sourceDBMu.Lock()
	if w.sourceDB == nil {
		var err error
		w.sourceDB, err = conn.DefaultDBProvider.Apply(&w.cfg.DecryptPassword().From)
		if err != nil {
			w.sourceDBMu.Unlock()
			return err
		}
	}
	w.sourceDBMu.Unlock()
	var status binlog.SourceStatus
	ctx, cancel := context.WithTimeout(ctx, utils.DefaultDBTimeout)
	defer cancel()
	pos, gtidSet, err := utils.GetPosAndGs(ctx, w.sourceDB.DB, w.cfg.Flavor)
	if err != nil {
		return err
	}
	status.Location = binlog.InitLocation(pos, gtidSet)
	ctx2, cancel2 := context.WithTimeout(ctx, utils.DefaultDBTimeout)
	defer cancel2()
	binlogs, err := binlog.GetBinaryLogs(ctx2, w.sourceDB.DB)
	if err != nil {
		return err
	}
	status.Binlogs = binlogs
	status.UpdateTime = time.Now()
	w.sourceStatus.Store(&status)
	return nil
}
func ( w * SourceWorker ) updateSourceStatus ( ctx context . Context ) error { w . sourceDBMu . Lock ( ) if w . sourceDB == nil { var err error w . sourceDB , err = conn . DefaultDBProvider . Apply ( & w . cfg . DecryptPassword ( ) . From ) if err != nil { w . sourceDBMu . Unlock ( ) return err } } w . sourceDBMu . Unlock ( ) var status binlog . SourceStatus ctx , cancel := context . WithTimeout ( ctx , utils . DefaultDBTimeout ) defer cancel ( ) pos , gtidSet , err := utils . GetPosAndGs ( ctx , w . sourceDB . DB , w . cfg . Flavor ) if err != nil { return err } status . Location = binlog . InitLocation ( pos , gtidSet ) ctx2 , cancel2 := context . WithTimeout ( ctx , utils . DefaultDBTimeout ) defer cancel2 ( ) binlogs , err := binlog . GetBinaryLogs ( ctx2 , w . sourceDB . DB ) if err != nil { return err } status . Binlogs = binlogs status . UpdateTime = time . Now ( ) w . sourceStatus . Store ( & status ) return nil }
updateSourceStatus updates w.sourceStatus.
updateSourceStatus updates w . sourceStatus .