// DisableRelay disables the functionality of start/watch/handle relay.
// a source worker will disable relay by the reason of EnableRelay is no longer valid.
// The paths to DisableRelay are:
// - source config `enable-relay: true` no longer valid
//   - when DM-worker Server watches a SourceBound change, which is to notify that source config has changed, and the
//     worker has started relay by that bound
//   - when the source worker is unbound and has started relay by that bound
// - UpstreamRelayWorkerKeyAdapter no longer valid
//   - when DM-worker Server watches a UpstreamRelayWorkerKeyAdapter change
//   - when DM-worker Server fails watching and recovers from a snapshot
DisableRelay disables the functionality of start/watch/handle relay.
a source worker will disable relay by the reason of EnableRelay is no longer valid.
The paths to DisableRelay are:
- source config `enable-relay: true` no longer valid
- when DM-worker Server watches a SourceBound change, which is to notify that source config has changed, and the
worker has started relay by that bound
- when the source worker is unbound and has started relay by that bound
- UpstreamRelayWorkerKeyAdapter no longer valid
- when DM-worker Server watches a UpstreamRelayWorkerKeyAdapter change
- when DM-worker Server fails watching and recovers from a snapshot
// DisableRelay disables the functionality of start / watch / handle relay . // a source worker will disable relay by the reason of EnableRelay is no longer valid . // The paths to DisableRelay are : // - source config ` enable - relay : true ` no longer valid // - when DM - worker Server watches a SourceBound change which is to notify that source config has changed and the // worker has started relay by that bound // - when the source worker is unbound and has started relay by that bound // - UpstreamRelayWorkerKeyAdapter no longer valid // - when DM - worker Server watches a UpstreamRelayWorkerKeyAdapter change // - when DM - worker Server fails watching and recovers from a snapshot
func (w *SourceWorker) DisableRelay() {
	w.l.Info("enter DisableRelay")
	w.Lock()
	defer w.Unlock()
	w.startedRelayBySourceCfg = false
	if !w.relayEnabled.CAS(true, false) {
		w.l.Warn("already disabled relay")
		return
	}
	w.relayCancel()
	w.relayWg.Wait()
	if w.cfg.Checker.CheckEnable {
		w.l.Info("refresh task checker")
		w.taskStatusChecker.Close()
		w.taskStatusChecker.Start()
		w.l.Info("finish refreshing task checker")
	}
	w.subTaskHolder.resetAllSubTasks(nil)
	if w.relayHolder != nil {
		r := w.relayHolder
		w.relayHolder = nil
		r.Close()
	}
	if w.relayPurger != nil {
		r := w.relayPurger
		w.relayPurger = nil
		r.Close()
	}
	w.l.Info("relay disabled")
}
func ( w * SourceWorker ) DisableRelay ( ) { w . l . Info ( " " ) w . Lock ( ) defer w . Unlock ( ) w . startedRelayBySourceCfg = false if ! w . relayEnabled . CAS ( true , false ) { w . l . Warn ( " " ) return } w . relayCancel ( ) w . relayWg . Wait ( ) if w . cfg . Checker . CheckEnable { w . l . Info ( " " ) w . taskStatusChecker . Close ( ) w . taskStatusChecker . Start ( ) w . l . Info ( " " ) } w . subTaskHolder . resetAllSubTasks ( nil ) if w . relayHolder != nil { r := w . relayHolder w . relayHolder = nil r . Close ( ) } if w . relayPurger != nil { r := w . relayPurger w . relayPurger = nil r . Close ( ) } w . l . Info ( " " ) }
DisableRelay disables the functionality of start/watch/handle relay.
DisableRelay disables the functionality of start / watch / handle relay .
// refresh task checker know latest relayEnabled, to avoid accessing relayHolder