subscription: switched to event based subscription

dbus/dbus.go

@@ -106,8 +106,9 @@ type Conn struct {
 		cleanIgnore int64
 	}
 	propertiesSubscriber struct {
-		updateCh chan<- *PropertiesUpdate
-		errCh    chan<- error
+		setSubscriber []*setPropertiesSubscriber
+		updateCh      chan<- *PropertiesUpdate
+		errCh         chan<- error
 		sync.Mutex
 	}
 }

dbus/subscription.go

@@ -17,6 +17,7 @@ package dbus
 import (
 	"context"
 	"errors"
+	"fmt"
 	"log"
 	"time"
 
@@ -46,6 +47,18 @@ func (c *Conn) Unsubscribe() error {
 	return c.sigobj.Call("org.freedesktop.systemd1.Manager.Unsubscribe", 0).Store()
 }
 
+func (c *Conn) SubscribeUnit(unit string) error {
+	return c.sigconn.BusObject().Call("org.freedesktop.DBus.AddMatch", 0,
+		fmt.Sprintf("type='signal',interface='org.freedesktop.DBus.Properties',"+
+			"member='PropertiesChanged',path='/org/freedesktop/systemd1/unit/%s'", PathBusEscape(unit))).Store()
+}
+
+func (c *Conn) UnsubscribeUnit(unit string) error {
+	return c.sigconn.BusObject().Call("org.freedesktop.DBus.RemoveMatch", 0,
+		fmt.Sprintf("type='signal',interface='org.freedesktop.DBus.Properties',"+
+			"member='PropertiesChanged',path='/org/freedesktop/systemd1/unit/%s'", PathBusEscape(unit))).Store()
+}
+
 func (c *Conn) dispatch() {
 	ch := make(chan *dbus.Signal, signalBuffer)
 
@@ -63,7 +76,8 @@ func (c *Conn) dispatch() {
 			}
 
 			if c.subStateSubscriber.updateCh == nil &&
-				c.propertiesSubscriber.updateCh == nil {
+				c.propertiesSubscriber.updateCh == nil &&
+				len(c.propertiesSubscriber.setSubscriber) == 0 {
 				continue
 			}
 
@@ -77,7 +91,6 @@ func (c *Conn) dispatch() {
 			case "org.freedesktop.DBus.Properties.PropertiesChanged":
 				if signal.Body[0].(string) == "org.freedesktop.systemd1.Unit" {
 					unitPath = signal.Path
-
 					if len(signal.Body) >= 2 {
 						if changed, ok := signal.Body[1].(map[string]dbus.Variant); ok {
 							c.sendPropertiesUpdate(unitPath, changed)
@@ -339,18 +352,36 @@ func (c *Conn) sendPropertiesUpdate(unitPath dbus.ObjectPath, changedProps map[s
 	c.propertiesSubscriber.Lock()
 	defer c.propertiesSubscriber.Unlock()
 
+	// remove inactive set subscribers
+	var activeSetSubscribers []*setPropertiesSubscriber
+	update := &PropertiesUpdate{unitName(unitPath), changedProps}
+	for _, setSubscriber := range c.propertiesSubscriber.setSubscriber {
+		select {
+		case <-setSubscriber.cancel:
+			close(setSubscriber.updateCh)
+			close(setSubscriber.errCh)
+		default:
+			if setSubscriber.set.Contains(unitName(unitPath)) {
+				handleUpdate(update, setSubscriber.updateCh, setSubscriber.errCh)
+			}
+			activeSetSubscribers = append(activeSetSubscribers, setSubscriber)
+		}
+	}
+	c.propertiesSubscriber.setSubscriber = activeSetSubscribers
+
 	if c.propertiesSubscriber.updateCh == nil {
 		return
 	}
+	handleUpdate(update, c.propertiesSubscriber.updateCh, c.propertiesSubscriber.errCh)
+}
 
-	update := &PropertiesUpdate{unitName(unitPath), changedProps}
-
+func handleUpdate(update *PropertiesUpdate, updateCh chan<- *PropertiesUpdate, errCh chan<- error) {
 	select {
-	case c.propertiesSubscriber.updateCh <- update:
+	case updateCh <- update:
 	default:
 		msg := "update channel is full"
 		select {
-		case c.propertiesSubscriber.errCh <- errors.New(msg):
+		case errCh <- errors.New(msg):
 		default:
 			log.Printf("full error channel while reporting: %s\n", msg)
 		}

dbus/subscription_set.go

@@ -33,7 +33,6 @@ func (s *SubscriptionSet) filter(unit string) bool {
 // SubscribeContext starts listening for dbus events for all of the units in the set.
 // Returns channels identical to conn.SubscribeUnits.
 func (s *SubscriptionSet) SubscribeContext(ctx context.Context) (<-chan map[string]*UnitStatus, <-chan error) {
-	// TODO: Make fully evented by using systemd 209 with properties changed values
 	return s.conn.SubscribeUnitsCustomContext(ctx, time.Second, 0,
 		mismatchUnitStatus,
 		func(unit string) bool { return s.filter(unit) },
@@ -50,6 +49,25 @@ func (c *Conn) NewSubscriptionSet() *SubscriptionSet {
 	return &SubscriptionSet{newSet(), c}
 }
 
+// SetPropertiesSubscriber works the same as [Conn.SetPropertiesSubscriber] but
+// will send updates only for the units that are part of [SubscriptionSet]. It
+// is a caller's responsibility to call [Conn.SubscribeUnit]
+// and [Conn.UnsubscribeUnit] when new units are added or
+// removed from the SubscriptionSet.
+func (s *SubscriptionSet) SetPropertiesSubscriber(ctx context.Context, propertiesChangedCh chan<- *PropertiesUpdate, errorCh chan<- error) {
+	s.conn.propertiesSubscriber.Lock()
+	defer s.conn.propertiesSubscriber.Unlock()
+	s.conn.propertiesSubscriber.setSubscriber = append(s.conn.propertiesSubscriber.setSubscriber,
+		&setPropertiesSubscriber{updateCh: propertiesChangedCh, errCh: errorCh, set: s, cancel: ctx.Done()})
+}
+
+type setPropertiesSubscriber struct {
+	updateCh chan<- *PropertiesUpdate
+	errCh    chan<- error
+	set      *SubscriptionSet
+	cancel   <-chan struct{}
+}
+
 // mismatchUnitStatus returns true if the provided UnitStatus objects
 // are not equivalent. false is returned if the objects are equivalent.
 // Only the Name, Description and state-related fields are used in

dbus/subscription_set_test.go

@@ -15,6 +15,7 @@
 package dbus
 
 import (
+	"context"
 	"testing"
 	"time"
 )
@@ -76,3 +77,59 @@ func TestSubscriptionSetUnit(t *testing.T) {
 success:
 	return
 }
+
+// TestSubscriptionSetAddedUnit exercises the basics of properties change subscription
+func TestSubscriptionPropertiesSubscriber(t *testing.T) {
+	target := "subscribe-events-set.service"
+
+	conn := setupConn(t)
+
+	testCtx := context.Background()
+	subSet := conn.NewSubscriptionSet()
+
+	updateCh := make(chan *PropertiesUpdate, 256)
+	errCh := make(chan error, 256)
+	subSet.SetPropertiesSubscriber(testCtx, updateCh, errCh)
+
+	subSet.Add(target)
+	setupUnit(target, conn, t)
+	linkUnit(target, conn, t)
+
+	err := conn.SubscribeUnit(target)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	reschan := make(chan string)
+	_, err = conn.StartUnitContext(testCtx, target, "replace", reschan)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	job := <-reschan
+	if job != "done" {
+		t.Fatal("Couldn't start", target)
+	}
+
+	timeout := make(chan bool, 1)
+	go func() {
+		time.Sleep(3 * time.Second)
+		close(timeout)
+	}()
+
+	for {
+		select {
+		case update := <-updateCh:
+			if update.UnitName == target {
+				subState, ok := update.Changed["SubState"].Value().(string)
+				if ok && subState == "running" {
+					return // success
+				}
+			}
+		case err := <-errCh:
+			t.Fatal(err)
+		case <-time.After(10 * time.Second):
+			t.Fatal("Reached timeout")
+		}
+	}
+}