// EnableHandleSubtasks enables the functionality of start/watch/handle subtasks.
EnableHandleSubtasks enables the functionality of start/watch/handle subtasks.
// EnableHandleSubtasks enables the functionality of start / watch / handle subtasks .
func (w *SourceWorker) EnableHandleSubtasks() error {
	w.l.Info("enter EnableHandleSubtasks")
	w.Lock()
	defer w.Unlock()
	if w.subTaskEnabled.Load() {
		w.l.Warn("already enabled handling subtasks")
		return nil
	}
	w.subTaskCtx, w.subTaskCancel = context.WithCancel(w.ctx)
	subTaskStages, validatorStages, subTaskCfgM, revSubTask, err := w.fetchSubTasksAndAdjust()
	if err != nil {
		return err
	}
	w.l.Info("starting to handle mysql source", zap.String("sourceCfg", w.cfg.String()), zap.Any("subTasks", subTaskCfgM))
	for _, subTaskCfg := range subTaskCfgM {
		expectStage := subTaskStages[subTaskCfg.Name]
		if expectStage.IsDeleted {
			continue
		}
		validatorStage := pb.Stage_InvalidStage
		if s, ok := validatorStages[subTaskCfg.Name]; ok {
			validatorStage = s.Expect
		}
		w.l.Info("start to create subtask in EnableHandleSubtasks", zap.String("sourceID", subTaskCfg.SourceID), zap.String("task", subTaskCfg.Name))
		clone := subTaskCfg
		if err2 := w.StartSubTask(&clone, expectStage.Expect, validatorStage, false); err2 != nil {
			w.subTaskHolder.closeAllSubTasks()
			return err2
		}
	}
	w.subTaskWg.Add(1)
	go func() {
		defer w.subTaskWg.Done()
		w.observeSubtaskStage(w.subTaskCtx, w.etcdClient, revSubTask)
	}()
	w.subTaskWg.Add(1)
	go func() {
		defer w.subTaskWg.Done()
		w.observeValidatorStage(w.subTaskCtx, revSubTask)
	}()
	w.subTaskEnabled.Store(true)
	w.l.Info("handling subtask enabled")
	return nil
}
func ( w * SourceWorker ) EnableHandleSubtasks ( ) error { w . l . Info ( " " ) w . Lock ( ) defer w . Unlock ( ) if w . subTaskEnabled . Load ( ) { w . l . Warn ( " " ) return nil } w . subTaskCtx , w . subTaskCancel = context . WithCancel ( w . ctx ) subTaskStages , validatorStages , subTaskCfgM , revSubTask , err := w . fetchSubTasksAndAdjust ( ) if err != nil { return err } w . l . Info ( " " , zap . String ( " " , w . cfg . String ( ) ) , zap . Any ( " " , subTaskCfgM ) ) for _ , subTaskCfg := range subTaskCfgM { expectStage := subTaskStages [ subTaskCfg . Name ] if expectStage . IsDeleted { continue } validatorStage := pb . Stage_InvalidStage if s , ok := validatorStages [ subTaskCfg . Name ] ; ok { validatorStage = s . Expect } w . l . Info ( " " , zap . String ( " " , subTaskCfg . SourceID ) , zap . String ( " " , subTaskCfg . Name ) ) clone := subTaskCfg if err2 := w . StartSubTask ( & clone , expectStage . Expect , validatorStage , false ) ; err2 != nil { w . subTaskHolder . closeAllSubTasks ( ) return err2 } } w . subTaskWg . Add ( 1 ) go func ( ) { defer w . subTaskWg . Done ( ) w . observeSubtaskStage ( w . subTaskCtx , w . etcdClient , revSubTask ) } ( ) w . subTaskWg . Add ( 1 ) go func ( ) { defer w . subTaskWg . Done ( ) w . observeValidatorStage ( w . subTaskCtx , revSubTask ) } ( ) w . subTaskEnabled . Store ( true ) w . l . Info ( " " ) return nil }
EnableHandleSubtasks enables the functionality of start/watch/handle subtasks.
EnableHandleSubtasks enables the functionality of start / watch / handle subtasks .
// we get the newest subtask stages directly which will omit the subtask stage PUT/DELETE event // because triggering these events is useless now // "for range" of a map will use same value address, so we'd better not pass value address to other function // TODO: handle fatal error from observeSubtaskStage //nolint:errcheck // TODO: handle fatal error from observeValidatorStage //nolint:errcheck