// NewSourceWorker creates a new SourceWorker. The functionality of relay and subtask is disabled by default, need call EnableRelay
// and EnableSubtask later.
NewSourceWorker creates a new SourceWorker. The functionality of relay and subtask is disabled by default, need call EnableRelay
and EnableSubtask later.
// NewSourceWorker creates a new SourceWorker . The functionality of relay and subtask is disabled by default need call EnableRelay // and EnableSubtask later .
func NewSourceWorker(
	cfg *config.SourceConfig,
	etcdClient *clientv3.Client,
	name string,
	relayDir string,
) (w *SourceWorker, err error) {
	w = &SourceWorker{
		cfg:           cfg,
		subTaskHolder: newSubTaskHolder(),
		l:             log.With(zap.String("component", "worker controller")),
		etcdClient:    etcdClient,
		name:          name,
		relayDir:      relayDir,
	}
	w.ctx, w.cancel = context.WithCancel(context.Background())
	w.closed.Store(true)
	w.subTaskEnabled.Store(false)
	w.relayEnabled.Store(false)
	defer func(w2 *SourceWorker) {
		if err != nil { 
			w2.cancel()
			w2.subTaskHolder.closeAllSubTasks()
		}
	}(w)
	if w.cfg.Checker.CheckEnable {
		tsc := NewTaskStatusChecker(w.cfg.Checker, w)
		err = tsc.Init()
		if err != nil {
			return nil, err
		}
		w.taskStatusChecker = tsc
	}
	InitConditionHub(w)
	w.l.Info("initialized", zap.Stringer("cfg", cfg))
	return w, nil
}
func NewSourceWorker ( cfg * config . SourceConfig , etcdClient * clientv3 . Client , name string , relayDir string , ) ( w * SourceWorker , err error ) { w = & SourceWorker { cfg : cfg , subTaskHolder : newSubTaskHolder ( ) , l : log . With ( zap . String ( " " , " " ) ) , etcdClient : etcdClient , name : name , relayDir : relayDir , } w . ctx , w . cancel = context . WithCancel ( context . Background ( ) ) w . closed . Store ( true ) w . subTaskEnabled . Store ( false ) w . relayEnabled . Store ( false ) defer func ( w2 * SourceWorker ) { if err != nil { w2 . cancel ( ) w2 . subTaskHolder . closeAllSubTasks ( ) } } ( w ) if w . cfg . Checker . CheckEnable { tsc := NewTaskStatusChecker ( w . cfg . Checker , w ) err = tsc . Init ( ) if err != nil { return nil , err } w . taskStatusChecker = tsc } InitConditionHub ( w ) w . l . Info ( " " , zap . Stringer ( " " , cfg ) ) return w , nil }
NewSourceWorker creates a new SourceWorker.
NewSourceWorker creates a new SourceWorker .
// keep running until canceled in `Close`. // when err != nil, `w` will become nil in this func, so we pass `w` in defer. // release resources, NOTE: we need to refactor New/Init/Start/Close for components later. // initial task status checker