diff --git a/dbus/dbus.go b/dbus/dbus.go index e966c156..ccb843a7 100644 --- a/dbus/dbus.go +++ b/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 } } diff --git a/dbus/subscription.go b/dbus/subscription.go index f9678585..01f5fe4c 100644 --- a/dbus/subscription.go +++ b/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) } diff --git a/dbus/subscription_set.go b/dbus/subscription_set.go index 173ca372..5172f23f 100644 --- a/dbus/subscription_set.go +++ b/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 diff --git a/dbus/subscription_set_test.go b/dbus/subscription_set_test.go index a37e4b9a..edd6a58d 100644 --- a/dbus/subscription_set_test.go +++ b/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") + } + } +}