Merge pull request #64 from bcwaldon/two-conns-one-dbus

Use separate connection for dbus signals

Author: bcwaldon

dbus/dbus.go

@@ -64,8 +64,14 @@ func PathBusEscape(path string) string {
 
 // Conn is a connection to systemd's dbus endpoint.
 type Conn struct {
-	sysconn     *dbus.Conn
-	sysobj      *dbus.Object
+	// sysconn/sysobj are only used to call dbus methods
+	sysconn *dbus.Conn
+	sysobj  *dbus.Object
+
+	// sigconn/sigobj are only used to receive dbus signals
+	sigconn *dbus.Conn
+	sigobj  *dbus.Object
+
 	jobListener struct {
 		jobs map[dbus.ObjectPath]chan string
 		sync.Mutex
@@ -77,65 +83,74 @@ type Conn struct {
 		ignore      map[dbus.ObjectPath]int64
 		cleanIgnore int64
 	}
-	dispatch map[string]func(dbus.Signal)
 }
 
-// New() establishes a connection to the system bus and authenticates.
+// New establishes a connection to the system bus and authenticates.
 func New() (*Conn, error) {
-	c := new(Conn)
-
-	if err := c.initConnection(dbus.SystemBusPrivate); err != nil {
-		return nil, err
-	}
-
-	c.initJobs()
-	return c, nil
+	return newConnection(dbus.SystemBusPrivate)
 }
 
-// NewUserConnection() establishes a connection to the session bus and
+// NewUserConnection establishes a connection to the session bus and
 // authenticates. This can be used to connect to systemd user instances.
 func NewUserConnection() (*Conn, error) {
-	c := new(Conn)
+	return newConnection(dbus.SessionBusPrivate)
+}
+
+func newConnection(createBus func() (*dbus.Conn, error)) (*Conn, error) {
+	sysconn, err := dbusConnection(createBus)
+	if err != nil {
+		return nil, err
+	}
 
-	if err := c.initConnection(dbus.SessionBusPrivate); err != nil {
+	sigconn, err := dbusConnection(createBus)
+	if err != nil {
 		return nil, err
 	}
 
-	c.initJobs()
+	c := &Conn{
+		sysconn: sysconn,
+		sysobj:  systemdObject(sysconn),
+		sigconn: sigconn,
+		sigobj:  systemdObject(sigconn),
+	}
+
+	c.subscriber.ignore = make(map[dbus.ObjectPath]int64)
+	c.jobListener.jobs = make(map[dbus.ObjectPath]chan string)
+
+	// Setup the listeners on jobs so that we can get completions
+	c.sigconn.BusObject().Call("org.freedesktop.DBus.AddMatch", 0,
+		"type='signal', interface='org.freedesktop.systemd1.Manager', member='JobRemoved'")
+
+	c.dispatch()
 	return c, nil
 }
 
-func (c *Conn) initConnection(createBus func()(*dbus.Conn, error)) error {
-	var err error
-	c.sysconn, err = createBus()
+func dbusConnection(createBus func() (*dbus.Conn, error)) (*dbus.Conn, error) {
+	conn, err := createBus()
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	// Only use EXTERNAL method, and hardcode the uid (not username)
 	// to avoid a username lookup (which requires a dynamically linked
 	// libc)
 	methods := []dbus.Auth{dbus.AuthExternal(strconv.Itoa(os.Getuid()))}
 
-	err = c.sysconn.Auth(methods)
+	err = conn.Auth(methods)
 	if err != nil {
-		c.sysconn.Close()
-		return err
+		conn.Close()
+		return nil, err
 	}
 
-	err = c.sysconn.Hello()
+	err = conn.Hello()
 	if err != nil {
-		c.sysconn.Close()
-		return err
+		conn.Close()
+		return nil, err
 	}
 
-	c.sysobj = c.sysconn.Object("org.freedesktop.systemd1", dbus.ObjectPath("/org/freedesktop/systemd1"))
-
-	// Setup the listeners on jobs so that we can get completions
-	c.sysconn.BusObject().Call("org.freedesktop.DBus.AddMatch", 0,
-		"type='signal', interface='org.freedesktop.systemd1.Manager', member='JobRemoved'")
-	c.initSubscription()
-	c.initDispatch()
+	return conn, nil
+}
 
-	return nil
+func systemdObject(conn *dbus.Conn) *dbus.Object {
+	return conn.Object("org.freedesktop.systemd1", dbus.ObjectPath("/org/freedesktop/systemd1"))
 }

dbus/methods.go

@@ -22,10 +22,6 @@ import (
 	"github.com/godbus/dbus"
 )
 
-func (c *Conn) initJobs() {
-	c.jobListener.jobs = make(map[dbus.ObjectPath]chan string)
-}
-
 func (c *Conn) jobComplete(signal *dbus.Signal) {
 	var id uint32
 	var job dbus.ObjectPath

dbus/subscription.go

@@ -33,12 +33,12 @@ const (
 // systemd will automatically stop sending signals so there is no need to
 // explicitly call Unsubscribe().
 func (c *Conn) Subscribe() error {
-	c.sysconn.BusObject().Call("org.freedesktop.DBus.AddMatch", 0,
+	c.sigconn.BusObject().Call("org.freedesktop.DBus.AddMatch", 0,
 		"type='signal',interface='org.freedesktop.systemd1.Manager',member='UnitNew'")
-	c.sysconn.BusObject().Call("org.freedesktop.DBus.AddMatch", 0,
+	c.sigconn.BusObject().Call("org.freedesktop.DBus.AddMatch", 0,
 		"type='signal',interface='org.freedesktop.DBus.Properties',member='PropertiesChanged'")
 
-	err := c.sysobj.Call("org.freedesktop.systemd1.Manager.Subscribe", 0).Store()
+	err := c.sigobj.Call("org.freedesktop.systemd1.Manager.Subscribe", 0).Store()
 	if err != nil {
 		return err
 	}
@@ -48,22 +48,18 @@ func (c *Conn) Subscribe() error {
 
 // Unsubscribe this connection from systemd dbus events.
 func (c *Conn) Unsubscribe() error {
-	err := c.sysobj.Call("org.freedesktop.systemd1.Manager.Unsubscribe", 0).Store()
+	err := c.sigobj.Call("org.freedesktop.systemd1.Manager.Unsubscribe", 0).Store()
 	if err != nil {
 		return err
 	}
 
 	return nil
 }
 
-func (c *Conn) initSubscription() {
-	c.subscriber.ignore = make(map[dbus.ObjectPath]int64)
-}
-
-func (c *Conn) initDispatch() {
+func (c *Conn) dispatch() {
 	ch := make(chan *dbus.Signal, signalBuffer)
 
-	c.sysconn.Signal(ch)
+	c.sigconn.Signal(ch)
 
 	go func() {
 		for {
@@ -72,24 +68,32 @@ func (c *Conn) initDispatch() {
 				return
 			}
 
-			switch signal.Name {
-			case "org.freedesktop.systemd1.Manager.JobRemoved":
+			if signal.Name == "org.freedesktop.systemd1.Manager.JobRemoved" {
 				c.jobComplete(signal)
+			}
+
+			if c.subscriber.updateCh == nil {
+				continue
+			}
 
+			var unitPath dbus.ObjectPath
+			switch signal.Name {
+			case "org.freedesktop.systemd1.Manager.JobRemoved":
 				unitName := signal.Body[2].(string)
-				var unitPath dbus.ObjectPath
 				c.sysobj.Call("org.freedesktop.systemd1.Manager.GetUnit", 0, unitName).Store(&unitPath)
-				if unitPath != dbus.ObjectPath("") {
-					c.sendSubStateUpdate(unitPath)
-				}
 			case "org.freedesktop.systemd1.Manager.UnitNew":
-				c.sendSubStateUpdate(signal.Body[1].(dbus.ObjectPath))
+				unitPath = signal.Body[1].(dbus.ObjectPath)
 			case "org.freedesktop.DBus.Properties.PropertiesChanged":
 				if signal.Body[0].(string) == "org.freedesktop.systemd1.Unit" {
-					// we only care about SubState updates, which are a Unit property
-					c.sendSubStateUpdate(signal.Path)
+					unitPath = signal.Path
 				}
 			}
+
+			if unitPath == dbus.ObjectPath("") {
+				continue
+			}
+
+			c.sendSubStateUpdate(unitPath)
 		}
 	}()
 }
@@ -176,9 +180,6 @@ func (c *Conn) SetSubStateSubscriber(updateCh chan<- *SubStateUpdate, errCh chan
 func (c *Conn) sendSubStateUpdate(path dbus.ObjectPath) {
 	c.subscriber.Lock()
 	defer c.subscriber.Unlock()
-	if c.subscriber.updateCh == nil {
-		return
-	}
 
 	if c.shouldIgnore(path) {
 		return