Routing: Add webhook to rules (#5722)

https://github.com/XTLS/Xray-core/pull/5722#issuecomment-3953836108

app/router/router.go

@@ -57,6 +57,7 @@ func (r *Router) Init(ctx context.Context, config *Config, d dns.Client, ohm out
 	for _, rule := range config.Rule {
 		cond, err := rule.BuildCondition()
 		if err != nil {
+			r.closeWebhooks()
 			return err
 		}
 		rr := &Rule{
@@ -64,10 +65,22 @@ func (r *Router) Init(ctx context.Context, config *Config, d dns.Client, ohm out
 			Tag:       rule.GetTag(),
 			RuleTag:   rule.GetRuleTag(),
 		}
+		if wh := rule.GetWebhook(); wh != nil {
+			notifier, err := NewWebhookNotifier(wh)
+			if err != nil {
+				r.closeWebhooks()
+				return err
+			}
+			rr.Webhook = notifier
+		}
 		btag := rule.GetBalancingTag()
 		if len(btag) > 0 {
 			brule, found := r.balancers[btag]
 			if !found {
+				if rr.Webhook != nil {
+					rr.Webhook.Close()
+				}
+				r.closeWebhooks()
 				return errors.New("balancer ", btag, " not found")
 			}
 			rr.Balancer = brule
@@ -80,6 +93,7 @@ func (r *Router) Init(ctx context.Context, config *Config, d dns.Client, ohm out
 
 // PickRoute implements routing.Router.
 func (r *Router) PickRoute(ctx routing.Context) (routing.Route, error) {
+	originalCtx := ctx
 	rule, ctx, err := r.pickRouteInternal(ctx)
 	if err != nil {
 		return nil, err
@@ -88,6 +102,9 @@ func (r *Router) PickRoute(ctx routing.Context) (routing.Route, error) {
 	if err != nil {
 		return nil, err
 	}
+	if rule.Webhook != nil {
+		rule.Webhook.Fire(originalCtx, tag)
+	}
 	return &Route{Context: ctx, outboundTag: tag, ruleTag: rule.RuleTag}, nil
 }
 
@@ -109,6 +126,11 @@ func (r *Router) ReloadRules(config *Config, shouldAppend bool) error {
 	defer r.mu.Unlock()
 
 	if !shouldAppend {
+		for _, rule := range r.rules {
+			if rule.Webhook != nil {
+				rule.Webhook.Close()
+			}
+		}
 		r.balancers = make(map[string]*Balancer, len(config.BalancingRule))
 		r.rules = make([]*Rule, 0, len(config.Rule))
 	}
@@ -125,12 +147,24 @@ func (r *Router) ReloadRules(config *Config, shouldAppend bool) error {
 		r.balancers[rule.Tag] = balancer
 	}
 
+	startIdx := len(r.rules)
+	closeNewWebhooks := func() {
+		for i := startIdx; i < len(r.rules); i++ {
+			if r.rules[i].Webhook != nil {
+				r.rules[i].Webhook.Close()
+			}
+		}
+		r.rules = r.rules[:startIdx]
+	}
+
 	for _, rule := range config.Rule {
 		if r.RuleExists(rule.GetRuleTag()) {
+			closeNewWebhooks()
 			return errors.New("duplicate ruleTag ", rule.GetRuleTag())
 		}
 		cond, err := rule.BuildCondition()
 		if err != nil {
+			closeNewWebhooks()
 			return err
 		}
 		rr := &Rule{
@@ -138,10 +172,22 @@ func (r *Router) ReloadRules(config *Config, shouldAppend bool) error {
 			Tag:       rule.GetTag(),
 			RuleTag:   rule.GetRuleTag(),
 		}
+		if wh := rule.GetWebhook(); wh != nil {
+			notifier, err := NewWebhookNotifier(wh)
+			if err != nil {
+				closeNewWebhooks()
+				return err
+			}
+			rr.Webhook = notifier
+		}
 		btag := rule.GetBalancingTag()
 		if len(btag) > 0 {
 			brule, found := r.balancers[btag]
 			if !found {
+				if rr.Webhook != nil {
+					rr.Webhook.Close()
+				}
+				closeNewWebhooks()
 				return errors.New("balancer ", btag, " not found")
 			}
 			rr.Balancer = brule
@@ -173,6 +219,8 @@ func (r *Router) RemoveRule(tag string) error {
 		for _, rule := range r.rules {
 			if rule.RuleTag != tag {
 				newRules = append(newRules, rule)
+			} else if rule.Webhook != nil {
+				rule.Webhook.Close()
 			}
 		}
 		r.rules = newRules
@@ -233,8 +281,20 @@ func (r *Router) Start() error {
 	return nil
 }
 
+// closeWebhooks closes all webhook notifiers in the current rule set.
+func (r *Router) closeWebhooks() {
+	for _, rule := range r.rules {
+		if rule.Webhook != nil {
+			rule.Webhook.Close()
+		}
+	}
+}
+
 // Close implements common.Closable.
 func (r *Router) Close() error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.closeWebhooks()
 	return nil
 }