wgengine/router: use eventbus.Monitor in linuxRouter (#17232)

This commit does not change the order or meaning of any eventbus activity, it
only updates the way the plumbing is set up.

Updates #15160

Change-Id: I61b863f9c05459d530a4c34063a8bad9046c0e27
Signed-off-by: M. J. Fromberger <fromberger@tailscale.com>

wgengine/router/router_linux.go

@@ -49,8 +49,7 @@ type linuxRouter struct {
 	tunname           string
 	netMon            *netmon.Monitor
 	health            *health.Tracker
-	eventClient       *eventbus.Client
+	eventSubs         eventbus.Monitor
-	ruleDeletedSub    *eventbus.Subscriber[netmon.RuleDeleted]
 	rulesAddedPub     *eventbus.Publisher[AddIPRules]
 	unregNetMon       func()
 	addrs             map[netip.Prefix]bool
@@ -100,7 +99,6 @@ func newUserspaceRouterAdvanced(logf logger.Logf, tunname string, netMon *netmon
 		tunname:       tunname,
 		netfilterMode: netfilterOff,
 		netMon:        netMon,
-		eventClient:   bus.Client("router-linux"),
 		health:        health,
 
 		cmd: cmd,
@@ -108,9 +106,9 @@ func newUserspaceRouterAdvanced(logf logger.Logf, tunname string, netMon *netmon
 		ipRuleFixLimiter: rate.NewLimiter(rate.Every(5*time.Second), 10),
 		ipPolicyPrefBase: 5200,
 	}
-	r.ruleDeletedSub = eventbus.Subscribe[netmon.RuleDeleted](r.eventClient)
+	ec := bus.Client("router-linux")
-	r.rulesAddedPub = eventbus.Publish[AddIPRules](r.eventClient)
+	r.rulesAddedPub = eventbus.Publish[AddIPRules](ec)
-	go r.consumeEventbusTopics()
+	r.eventSubs = ec.Monitor(r.consumeEventbusTopics(ec))
 
 	if r.useIPCommand() {
 		r.ipRuleAvailable = (cmd.run("ip", "rule") == nil)
@@ -159,13 +157,16 @@ func newUserspaceRouterAdvanced(logf logger.Logf, tunname string, netMon *netmon
 // always handled in the order they are received, i.e. the next event is not
 // read until the previous event's handler has returned. It returns when the
 // [eventbus.Client] is closed.
-func (r *linuxRouter) consumeEventbusTopics() {
+func (r *linuxRouter) consumeEventbusTopics(ec *eventbus.Client) func(*eventbus.Client) {
-	for {
+	ruleDeletedSub := eventbus.Subscribe[netmon.RuleDeleted](ec)
-		select {
+	return func(ec *eventbus.Client) {
-		case <-r.eventClient.Done():
+		for {
-			return
+			select {
-		case rulesDeleted := <-r.ruleDeletedSub.Events():
+			case <-ec.Done():
-			r.onIPRuleDeleted(rulesDeleted.Table, rulesDeleted.Priority)
+				return
+			case rs := <-ruleDeletedSub.Events():
+				r.onIPRuleDeleted(rs.Table, rs.Priority)
+			}
 		}
 	}
 }
@@ -362,7 +363,7 @@ func (r *linuxRouter) Close() error {
 	if r.unregNetMon != nil {
 		r.unregNetMon()
 	}
-	r.eventClient.Close()
+	r.eventSubs.Close()
 	if err := r.downInterface(); err != nil {
 		return err
 	}