Method Run

driver/mysql/extractor.go:152–367

Run executes the complete extract logic.

()

// Run executes the complete extract logic.
func (e *Extractor) Run() {
	var err error

	{
		jobStatus, _ := e.storeManager.GetJobStatus(e.subject)
		if jobStatus == common.TargetGtidFinished {
			_ = e.Shutdown()
			return
		}
	}

	{
		target, err := e.storeManager.GetTargetGtid(e.subject)
		if err != nil {
			e.onError(common.TaskStateDead, err)
			return
		}
		e.targetGtid = target

	}

	if e.mysqlContext.WaitOnJob != "" {
		jobStatus, err := e.storeManager.GetJobStatus(e.subject)
		if err != nil {
			e.onError(common.TaskStateDead, err)
			return
		}
		// ensure job status exist when first time to start a reverse task
		if jobStatus == "" {
			jobStatus = common.DtleJobStatusReverseInit
			e.storeManager.SaveJobInfo(common.JobListItemV2{JobId: e.subject, JobStatus: jobStatus})
		}
		firstWait := jobStatus == common.DtleJobStatusReverseInit
		if firstWait {
			// the first time to wait
			e.logger.Info("waiting for another job to finish", "job2", e.mysqlContext.WaitOnJob)
			err = e.storeManager.WaitOnJob(e.subject, e.mysqlContext.WaitOnJob, e.shutdownCh)
			if err != nil {
				e.onError(common.TaskStateDead, err)
				return
			}
		}
		e.logger.Info("after WaitOnJob", "job2", e.mysqlContext.WaitOnJob, "firstWait", firstWait)
	}

	// PutConfig before WatchNats
	err = e.storeManager.PutConfig(e.subject, e.mysqlContext)
	if err != nil {
		e.onError(common.TaskStateDead, errors.Wrap(err, "PutConfig"))
		return
	}

	e.natsAddr, err = e.storeManager.SrcWatchNats(e.subject, e.shutdownCh, func(err error) {
		e.onError(common.TaskStateDead, err)
	})
	if err != nil {
		e.onError(common.TaskStateDead, errors.Wrap(err, "SrcWatchNats"))
		return