Merge lp:~pedronis/ubuntu-push/cancel-cancel-cancel into lp:ubuntu-push/automatic

[Samuele Pedroni]

WatchSignal cancelling, and connectivity exposed cancelling, make connectivity start not leave watches behind, fixes deadlock XXX in test

WatchSignal cancelling, and connectivity exposed cancelling,

make connectivity start not leave watches behind, fixes deadlock XXX in test;

don't leave goroutines behind in touched bus packages

helper to dump goroutines to check whether tests leave them around

[John Lenton (chipaca)]

review: Approve

=== modified file 'bus/connectivity/connectivity.go'
--- bus/connectivity/connectivity.go	2015-02-24 14:10:05 +0000
+++ bus/connectivity/connectivity.go	2015-02-27 10:54:31 +0000
@@ -24,12 +24,14 @@
 
 import (
 	"errors"
+	"sync"
+	"time"
+
 	"launchpad.net/ubuntu-push/bus"
 	"launchpad.net/ubuntu-push/bus/networkmanager"
 	"launchpad.net/ubuntu-push/config"
 	"launchpad.net/ubuntu-push/logger"
 	"launchpad.net/ubuntu-push/util"
-	"time"
 )
 
 // The configuration for ConnectedState, intended to be populated from a config file.
@@ -45,7 +47,8 @@
 	ConnectivityCheckMD5 string `json:"connectivity_check_md5"`
 }
 
-type connectedState struct {
+// ConnectedState helps tracking connectivity.
+type ConnectedState struct {
 	networkStateCh <-chan networkmanager.State
 	networkConCh   <-chan string
 	config         ConnectivityConfig
@@ -57,11 +60,43 @@
 	currentState   networkmanager.State
 	lastSent       bool
 	timer          *time.Timer
+	doneLck        sync.Mutex
+	done           chan struct{}
+	canceled       bool
+	stateWatch     bus.Cancellable
+	conWatch       bus.Cancellable
+}
+
+// New makes a ConnectedState for connectivity tracking.
+//
+// The endpoint need not be dialed; Track()  will Dial() and
+// Close() it as it sees fit.
+func New(endp bus.Endpoint, config ConnectivityConfig, log logger.Logger) *ConnectedState {
+	wg := NewWebchecker(config.ConnectivityCheckURL, config.ConnectivityCheckMD5, 10*time.Second, log)
+	return &ConnectedState{
+		config: config,
+		log:    log,
+		endp:   endp,
+		webget: wg.Webcheck,
+		done:   make(chan struct{}),
+	}
+}
+
+// cancel watches if any
+func (cs *ConnectedState) reset() {
+	if cs.stateWatch != nil {
+		cs.stateWatch.Cancel()
+		cs.stateWatch = nil
+	}
+	if cs.conWatch != nil {
+		cs.conWatch.Cancel()
+		cs.conWatch = nil
+	}
 }
 
 // start connects to the bus, gets the initial NetworkManager state, and sets
 // up the watch.
-func (cs *connectedState) start() networkmanager.State {
+func (cs *ConnectedState) start() networkmanager.State {
 	var initial networkmanager.State
 	var stateCh <-chan networkmanager.State
 	var primary string
@@ -72,8 +107,9 @@
 		cs.connAttempts += ar.Redial()
 		nm := networkmanager.New(cs.endp, cs.log)
 
+		cs.reset()
 		// set up the watch
-		stateCh, err = nm.WatchState()
+		stateCh, cs.stateWatch, err = nm.WatchState()
 		if err != nil {
 			cs.log.Debugf("failed to set up the state watch: %s", err)
 			goto Continue
@@ -87,7 +123,7 @@
 		}
 		cs.log.Debugf("got initial state of %s", initial)
 
-		conCh, err = nm.WatchPrimaryConnection()
+		conCh, cs.conWatch, err = nm.WatchPrimaryConnection()
 		if err != nil {
 			cs.log.Debugf("failed to set up the connection watch: %s", err)
 			goto Continue
@@ -107,9 +143,11 @@
 	}
 }
 
-// connectedStateStep takes one step forwards in the “am I connected?”
+var errCanceled = errors.New("canceled")
+
+// step takes one step forwards in the “am I connected?”
 // answering state machine.
-func (cs *connectedState) connectedStateStep() (bool, error) {
+func (cs *ConnectedState) step() (bool, error) {
 	stabilizingTimeout := cs.config.StabilizingTimeout.Duration
 	recheckTimeout := cs.config.RecheckTimeout.Duration
 	log := cs.log
@@ -117,6 +155,8 @@
 Loop:
 	for {
 		select {
+		case <-cs.done:
+			return false, errCanceled
 		case <-cs.networkConCh:
 			cs.webgetCh = nil
 			cs.timer.Reset(stabilizingTimeout)
@@ -178,35 +218,48 @@
 	return cs.lastSent, nil
 }
 
-// ConnectedState sends the initial NetworkManager state and changes to it
+// Track sends the initial NetworkManager state and changes to it
 // over the "out" channel. Sends "false" as soon as it detects trouble, "true"
 // after checking actual connectivity.
 //
-// The endpoint need not be dialed; connectivity will Dial() and Close()
-// it as it sees fit.
-func ConnectedState(endp bus.Endpoint, config ConnectivityConfig, log logger.Logger, out chan<- bool) {
-	wg := NewWebchecker(config.ConnectivityCheckURL, config.ConnectivityCheckMD5, 10*time.Second, log)
-	cs := &connectedState{
-		config: config,
-		log:    log,
-		endp:   endp,
-		webget: wg.Webcheck,
-	}
+func (cs *ConnectedState) Track(out chan<- bool) {
 
 Start:
-	log.Debugf("sending initial 'disconnected'.")
-	out <- false
+	cs.log.Debugf("sending initial 'disconnected'.")
+	select {
+	case <-cs.done:
+		return
+	case out <- false:
+	}
 	cs.lastSent = false
 	cs.currentState = cs.start()
+	defer cs.reset()
 	cs.timer = time.NewTimer(cs.config.StabilizingTimeout.Duration)
 
 	for {
-		v, err := cs.connectedStateStep()
+		v, err := cs.step()
+		if err == errCanceled {
+			return
+		}
 		if err != nil {
 			// tear it all down and start over
-			log.Errorf("%s", err)
+			cs.log.Errorf("%s", err)
 			goto Start
 		}
-		out <- v
+		select {
+		case <-cs.done:
+			return
+		case out <- v:
+		}
+	}
+}
+
+// Cancel stops the ConnectedState machinary.
+func (cs *ConnectedState) Cancel() {
+	cs.doneLck.Lock()
+	defer cs.doneLck.Unlock()
+	if !cs.canceled {
+		cs.canceled = true
+		close(cs.done)
 	}
 }

=== modified file 'bus/connectivity/connectivity_test.go'
--- bus/connectivity/connectivity_test.go	2015-02-24 14:10:05 +0000
+++ bus/connectivity/connectivity_test.go	2015-02-27 10:54:31 +0000
@@ -64,13 +64,13 @@
 )
 
 /*
-   tests for connectedState's Start() method
+   tests for ConnectedState's Start() method
 */
 
 // when given a working config and bus, Start() will work
 func (s *ConnSuite) TestStartWorks(c *C) {
 	endp := testingbus.NewTestingEndpoint(condition.Work(true), condition.Work(true), uint32(networkmanager.Connecting), helloCon)
-	cs := connectedState{config: ConnectivityConfig{}, log: s.log, endp: endp}
+	cs := ConnectedState{config: ConnectivityConfig{}, log: s.log, endp: endp}
 
 	nopTicker := make(chan []interface{})
 	testingbus.SetWatchSource(endp, "StateChanged", nopTicker)
@@ -83,7 +83,7 @@
 // if the bus fails a couple of times, we're still OK
 func (s *ConnSuite) TestStartRetriesConnect(c *C) {
 	endp := testingbus.NewTestingEndpoint(condition.Fail2Work(2), condition.Work(true), uint32(networkmanager.Connecting), helloCon)
-	cs := connectedState{config: ConnectivityConfig{}, log: s.log, endp: endp}
+	cs := ConnectedState{config: ConnectivityConfig{}, log: s.log, endp: endp}
 
 	nopTicker := make(chan []interface{})
 	testingbus.SetWatchSource(endp, "StateChanged", nopTicker)
@@ -97,7 +97,7 @@
 // when the calls to NetworkManager fails for a bit, we're still OK
 func (s *ConnSuite) TestStartRetriesCall(c *C) {
 	endp := testingbus.NewTestingEndpoint(condition.Work(true), condition.Fail2Work(5), uint32(networkmanager.Connecting), helloCon)
-	cs := connectedState{config: ConnectivityConfig{}, log: s.log, endp: endp}
+	cs := ConnectedState{config: ConnectivityConfig{}, log: s.log, endp: endp}
 
 	nopTicker := make(chan []interface{})
 	testingbus.SetWatchSource(endp, "StateChanged", nopTicker)
@@ -111,14 +111,14 @@
 
 // when some of the calls to NetworkManager fails for a bit, we're still OK
 func (s *ConnSuite) TestStartRetriesCall2(c *C) {
-	cond := condition.Chain(3, condition.Work(true), 1, condition.Work(false),
+	cond := condition.Chain(1, condition.Work(true), 1, condition.Work(false),
 		1, condition.Work(true))
 
 	endp := testingbus.NewTestingEndpoint(condition.Work(true), cond,
 		uint32(networkmanager.Connecting), helloCon,
 		uint32(networkmanager.Connecting), helloCon,
 	)
-	cs := connectedState{config: ConnectivityConfig{}, log: s.log, endp: endp}
+	cs := ConnectedState{config: ConnectivityConfig{}, log: s.log, endp: endp}
 
 	nopTicker := make(chan []interface{})
 	testingbus.SetWatchSource(endp, "StateChanged", nopTicker)
@@ -141,7 +141,7 @@
 		uint32(networkmanager.Connecting),
 		helloCon,
 	)
-	cs := connectedState{config: ConnectivityConfig{}, log: s.log, endp: endp}
+	cs := ConnectedState{config: ConnectivityConfig{}, log: s.log, endp: endp}
 	watchTicker := make(chan []interface{}, 1)
 	nopTicker := make(chan []interface{})
 	testingbus.SetWatchSource(endp, "StateChanged", watchTicker)
@@ -151,7 +151,6 @@
 
 	c.Check(cs.start(), Equals, networkmanager.Connecting)
 	c.Check(cs.connAttempts, Equals, uint32(2))
-	// XXX this may be stolen by an old watch => dead lock
 	watchTicker <- []interface{}{uint32(networkmanager.ConnectedGlobal)}
 	c.Check(<-cs.networkStateCh, Equals, networkmanager.ConnectedGlobal)
 }
@@ -181,7 +180,7 @@
 	}
 }
 
-func (rep *racyEndpoint) WatchSignal(member string, f func(...interface{}), d func()) error {
+func (rep *racyEndpoint) WatchSignal(member string, f func(...interface{}), d func()) (bus.Cancellable, error) {
 	if member == "StateChanged" {
 		// we count never having gotten the state as happening "after" now.
 		rep.lock.RLock()
@@ -194,7 +193,7 @@
 			d()
 		}()
 	}
-	return nil
+	return nil, nil
 }
 
 func (*racyEndpoint) Close()                                                        {}
@@ -223,7 +222,7 @@
 func (s *ConnSuite) TestStartAvoidsRace(c *C) {
 	for delta := time.Second; delta > 1; delta /= 2 {
 		rep := &racyEndpoint{delta: delta}
-		cs := connectedState{config: ConnectivityConfig{}, log: s.log, endp: rep}
+		cs := ConnectedState{config: ConnectivityConfig{}, log: s.log, endp: rep}
 		f := Commentf("when delta=%s", delta)
 		c.Assert(cs.start(), Equals, networkmanager.Connecting, f)
 		c.Assert(takeNext(cs.networkStateCh), Equals, networkmanager.ConnectedGlobal, f)
@@ -231,7 +230,7 @@
 }
 
 /*
-   tests for connectedStateStep()
+   tests for step()
 */
 
 func (s *ConnSuite) TestSteps(c *C) {
@@ -242,7 +241,7 @@
 		RecheckTimeout: config.ConfigTimeDuration{recheck_timeout},
 	}
 	ch := make(chan networkmanager.State, 10)
-	cs := &connectedState{
+	cs := &ConnectedState{
 		config:         cfg,
 		networkStateCh: ch,
 		timer:          time.NewTimer(time.Second),
@@ -251,15 +250,15 @@
 		lastSent:       false,
 	}
 	ch <- networkmanager.ConnectedGlobal
-	f, e := cs.connectedStateStep()
+	f, e := cs.step()
 	c.Check(e, IsNil)
 	c.Check(f, Equals, true)
 	ch <- networkmanager.Disconnected
 	ch <- networkmanager.ConnectedGlobal
-	f, e = cs.connectedStateStep()
+	f, e = cs.step()
 	c.Check(e, IsNil)
 	c.Check(f, Equals, false)
-	f, e = cs.connectedStateStep()
+	f, e = cs.step()
 	c.Check(e, IsNil)
 	c.Check(f, Equals, true)
 
@@ -267,7 +266,7 @@
 	webget_p = condition.Fail2Work(1)
 	ch <- networkmanager.Disconnected
 	ch <- networkmanager.ConnectedGlobal
-	f, e = cs.connectedStateStep()
+	f, e = cs.step()
 	c.Check(e, IsNil)
 	c.Check(f, Equals, false) // first false is from the Disconnected
 
@@ -276,7 +275,7 @@
 	_t := time.NewTimer(recheck_timeout / 2)
 
 	go func() {
-		f, e := cs.connectedStateStep()
+		f, e := cs.step()
 		c.Check(e, IsNil)
 		_ch <- f
 	}()
@@ -294,15 +293,15 @@
 	ch <- networkmanager.Disconnected    // this should not
 	ch <- networkmanager.ConnectedGlobal // this should trigger a 'true'
 
-	f, e = cs.connectedStateStep()
+	f, e = cs.step()
 	c.Check(e, IsNil)
 	c.Check(f, Equals, false)
-	f, e = cs.connectedStateStep()
+	f, e = cs.step()
 	c.Check(e, IsNil)
 	c.Check(f, Equals, true)
 
 	close(ch) // this should make it error out
-	_, e = cs.connectedStateStep()
+	_, e = cs.step()
 	c.Check(e, NotNil)
 }
 
@@ -336,7 +335,9 @@
 	out := make(chan bool)
 	dt := time.Second / 10
 	timer := time.NewTimer(dt)
-	go ConnectedState(endp, cfg, s.log, out)
+	cs := New(endp, cfg, s.log)
+	defer cs.Cancel()
+	go cs.Track(out)
 	var v bool
 	expecteds := []struct {
 		p    bool
@@ -399,7 +400,9 @@
 	out := make(chan bool)
 	dt := time.Second / 10
 	timer := time.NewTimer(dt)
-	go ConnectedState(endp, cfg, s.log, out)
+	cs := New(endp, cfg, s.log)
+	defer cs.Cancel()
+	go cs.Track(out)
 	var v bool
 	expecteds := []struct {
 		p           bool

=== modified file 'bus/endpoint.go'
--- bus/endpoint.go	2015-01-22 09:52:07 +0000
+++ bus/endpoint.go	2015-02-27 10:54:31 +0000
@@ -35,10 +35,15 @@
 type BusMethod func(string, []interface{}, []interface{}) ([]interface{}, error)
 type DispatchMap map[string]BusMethod
 
+// Cancellable can be canceled.
+type Cancellable interface {
+	Cancel() error
+}
+
 // bus.Endpoint represents the DBus connection itself.
 type Endpoint interface {
 	GrabName(allowReplacement bool) <-chan error
-	WatchSignal(member string, f func(...interface{}), d func()) error
+	WatchSignal(member string, f func(...interface{}), d func()) (Cancellable, error)
 	WatchMethod(DispatchMap, string, ...interface{})
 	Signal(string, string, []interface{}) error
 	Call(member string, args []interface{}, rvs ...interface{}) error
@@ -123,16 +128,16 @@
 // sends the values over a channel, and d() would close the channel.
 //
 // XXX: untested
-func (endp *endpoint) WatchSignal(member string, f func(...interface{}), d func()) error {
+func (endp *endpoint) WatchSignal(member string, f func(...interface{}), d func()) (Cancellable, error) {
 	watch, err := endp.proxy.WatchSignal(endp.addr.Interface, member)
 	if err != nil {
 		endp.log.Debugf("failed to set up the watch: %s", err)
-		return err
+		return nil, err
 	}
 
 	go endp.unpackMessages(watch, f, d, member)
 
-	return nil
+	return watch, nil
 }
 
 // Call() invokes the provided member method (on the name, path and
@@ -324,6 +329,6 @@
 		}
 		f(endp.unpackOneMsg(msg, member)...)
 	}
-	endp.log.Errorf("got not-OK from %s watch", member)
+	endp.log.Debugf("got not-OK from %s watch", member)
 	d()
 }

=== modified file 'bus/networkmanager/networkmanager.go'
--- bus/networkmanager/networkmanager.go	2015-01-22 09:52:07 +0000
+++ bus/networkmanager/networkmanager.go	2015-02-27 10:54:31 +0000
@@ -42,13 +42,13 @@
 	GetState() State
 	// WatchState listens for changes to NetworkManager's state, and sends
 	// them out over the channel returned.
-	WatchState() (<-chan State, error)
+	WatchState() (<-chan State, bus.Cancellable, error)
 	// GetPrimaryConnection fetches and returns NetworkManager's current
 	// primary connection.
 	GetPrimaryConnection() string
 	// WatchPrimaryConnection listens for changes of NetworkManager's
 	// Primary Connection, and sends it out over the channel returned.
-	WatchPrimaryConnection() (<-chan string, error)
+	WatchPrimaryConnection() (<-chan string, bus.Cancellable, error)
 }
 
 type networkManager struct {
@@ -85,9 +85,9 @@
 	return State(v)
 }
 
-func (nm *networkManager) WatchState() (<-chan State, error) {
+func (nm *networkManager) WatchState() (<-chan State, bus.Cancellable, error) {
 	ch := make(chan State)
-	err := nm.bus.WatchSignal("StateChanged",
+	w, err := nm.bus.WatchSignal("StateChanged",
 		func(ns ...interface{}) {
 			stint, ok := ns[0].(uint32)
 			if !ok {
@@ -101,10 +101,10 @@
 		func() { close(ch) })
 	if err != nil {
 		nm.log.Debugf("Failed to set up the watch: %s", err)
-		return nil, err
+		return nil, nil, err
 	}
 
-	return ch, nil
+	return ch, w, nil
 }
 
 func (nm *networkManager) GetPrimaryConnection() string {
@@ -124,9 +124,9 @@
 	return string(v)
 }
 
-func (nm *networkManager) WatchPrimaryConnection() (<-chan string, error) {
+func (nm *networkManager) WatchPrimaryConnection() (<-chan string, bus.Cancellable, error) {
 	ch := make(chan string)
-	err := nm.bus.WatchSignal("PropertiesChanged",
+	w, err := nm.bus.WatchSignal("PropertiesChanged",
 		func(ppsi ...interface{}) {
 			pps, ok := ppsi[0].(map[string]dbus.Variant)
 			if !ok {
@@ -147,8 +147,8 @@
 		}, func() { close(ch) })
 	if err != nil {
 		nm.log.Debugf("failed to set up the watch: %s", err)
-		return nil, err
+		return nil, nil, err
 	}
 
-	return ch, nil
+	return ch, w, nil
 }

=== modified file 'bus/networkmanager/networkmanager_test.go'
--- bus/networkmanager/networkmanager_test.go	2014-04-04 12:01:42 +0000
+++ bus/networkmanager/networkmanager_test.go	2015-02-27 10:54:31 +0000
@@ -90,8 +90,9 @@
 func (s *NMSuite) TestWatchState(c *C) {
 	tc := testingbus.NewTestingEndpoint(nil, condition.Work(true), uint32(Unknown), uint32(Asleep), uint32(ConnectedGlobal))
 	nm := New(tc, s.log)
-	ch, err := nm.WatchState()
-	c.Check(err, IsNil)
+	ch, w, err := nm.WatchState()
+	c.Assert(err, IsNil)
+	defer w.Cancel()
 	l := []State{<-ch, <-ch, <-ch}
 	c.Check(l, DeepEquals, []State{Unknown, Asleep, ConnectedGlobal})
 }
@@ -99,7 +100,7 @@
 // WatchState returns on error if the dbus call fails
 func (s *NMSuite) TestWatchStateFails(c *C) {
 	nm := New(testingbus.NewTestingEndpoint(nil, condition.Work(false)), s.log)
-	_, err := nm.WatchState()
+	_, _, err := nm.WatchState()
 	c.Check(err, NotNil)
 }
 
@@ -107,8 +108,9 @@
 func (s *NMSuite) TestWatchStateClosesOnWatchBail(c *C) {
 	tc := testingbus.NewTestingEndpoint(nil, condition.Work(true))
 	nm := New(tc, s.log)
-	ch, err := nm.WatchState()
-	c.Check(err, IsNil)
+	ch, w, err := nm.WatchState()
+	c.Assert(err, IsNil)
+	defer w.Cancel()
 	_, ok := <-ch
 	c.Check(ok, Equals, false)
 }
@@ -117,8 +119,9 @@
 func (s *NMSuite) TestWatchStateSurvivesRubbishValues(c *C) {
 	tc := testingbus.NewTestingEndpoint(nil, condition.Work(true), "a")
 	nm := New(tc, s.log)
-	ch, err := nm.WatchState()
-	c.Check(err, IsNil)
+	ch, w, err := nm.WatchState()
+	c.Assert(err, IsNil)
+	defer w.Cancel()
 	_, ok := <-ch
 	c.Check(ok, Equals, false)
 }
@@ -164,8 +167,9 @@
 		mkPriConMap("/b/2"),
 		mkPriConMap("/c/3"))
 	nm := New(tc, s.log)
-	ch, err := nm.WatchPrimaryConnection()
-	c.Check(err, IsNil)
+	ch, w, err := nm.WatchPrimaryConnection()
+	c.Assert(err, IsNil)
+	defer w.Cancel()
 	l := []string{<-ch, <-ch, <-ch}
 	c.Check(l, DeepEquals, []string{"/a/1", "/b/2", "/c/3"})
 }
@@ -173,7 +177,7 @@
 // WatchPrimaryConnection returns on error if the dbus call fails
 func (s *NMSuite) TestWatchPrimaryConnectionFails(c *C) {
 	nm := New(testingbus.NewTestingEndpoint(nil, condition.Work(false)), s.log)
-	_, err := nm.WatchPrimaryConnection()
+	_, _, err := nm.WatchPrimaryConnection()
 	c.Check(err, NotNil)
 }
 
@@ -181,8 +185,9 @@
 func (s *NMSuite) TestWatchPrimaryConnectionClosesOnWatchBail(c *C) {
 	tc := testingbus.NewTestingEndpoint(nil, condition.Work(true))
 	nm := New(tc, s.log)
-	ch, err := nm.WatchPrimaryConnection()
-	c.Check(err, IsNil)
+	ch, w, err := nm.WatchPrimaryConnection()
+	c.Assert(err, IsNil)
+	defer w.Cancel()
 	_, ok := <-ch
 	c.Check(ok, Equals, false)
 }
@@ -191,8 +196,9 @@
 func (s *NMSuite) TestWatchPrimaryConnectionSurvivesRubbishValues(c *C) {
 	tc := testingbus.NewTestingEndpoint(nil, condition.Work(true), "a")
 	nm := New(tc, s.log)
-	ch, err := nm.WatchPrimaryConnection()
+	ch, w, err := nm.WatchPrimaryConnection()
 	c.Assert(err, IsNil)
+	defer w.Cancel()
 	_, ok := <-ch
 	c.Check(ok, Equals, false)
 }
@@ -204,8 +210,9 @@
 		map[string]dbus.Variant{"PrimaryConnection": dbus.Variant{dbus.ObjectPath("42")}},
 	)
 	nm := New(tc, s.log)
-	ch, err := nm.WatchPrimaryConnection()
+	ch, w, err := nm.WatchPrimaryConnection()
 	c.Assert(err, IsNil)
+	defer w.Cancel()
 	v, ok := <-ch
 	c.Check(ok, Equals, true)
 	c.Check(v, Equals, "42")
@@ -218,8 +225,9 @@
 		map[string]dbus.Variant{"PrimaryConnection": dbus.Variant{dbus.ObjectPath("42")}},
 	)
 	nm := New(tc, s.log)
-	ch, err := nm.WatchPrimaryConnection()
+	ch, w, err := nm.WatchPrimaryConnection()
 	c.Assert(err, IsNil)
+	defer w.Cancel()
 	v, ok := <-ch
 	c.Check(ok, Equals, true)
 	c.Check(v, Equals, "42")

=== modified file 'bus/notifications/raw.go'
--- bus/notifications/raw.go	2015-01-22 09:52:07 +0000
+++ bus/notifications/raw.go	2015-02-27 10:54:31 +0000
@@ -93,7 +93,7 @@
 // and sends them over the channel provided
 func (raw *RawNotifications) WatchActions() (<-chan *RawAction, error) {
 	ch := make(chan *RawAction)
-	err := raw.bus.WatchSignal("ActionInvoked",
+	_, err := raw.bus.WatchSignal("ActionInvoked",
 		func(ns ...interface{}) {
 			if len(ns) != 2 {
 				raw.log.Debugf("ActionInvoked delivered %d things instead of 2", len(ns))

=== modified file 'bus/testing/testing_endpoint.go'
--- bus/testing/testing_endpoint.go	2015-02-24 14:10:05 +0000
+++ bus/testing/testing_endpoint.go	2015-02-27 10:54:31 +0000
@@ -80,11 +80,32 @@
 	return tc.(*testingEndpoint).callArgs
 }
 
+type watchCancel struct {
+	done      chan struct{}
+	cancelled chan struct{}
+	lck       sync.Mutex
+	member    string
+}
+
+// this waits for actual cancelllation for test convenience
+func (wc *watchCancel) Cancel() error {
+	wc.lck.Lock()
+	defer wc.lck.Unlock()
+	if wc.cancelled != nil {
+		close(wc.cancelled)
+		wc.cancelled = nil
+		<-wc.done
+	}
+	return nil
+}
+
 // See Endpoint's WatchSignal. This WatchSignal will check its condition to
 // decide whether to return an error, or provide each of its return values
 // or values from the previously set watchSource for member.
-func (tc *testingEndpoint) WatchSignal(member string, f func(...interface{}), d func()) error {
+func (tc *testingEndpoint) WatchSignal(member string, f func(...interface{}), d func()) (bus.Cancellable, error) {
 	if tc.callCond.OK() {
+		cancelled := make(chan struct{})
+		done := make(chan struct{})
 		go func() {
 			tc.watchLck.RLock()
 			source := tc.watchSources[member]
@@ -96,21 +117,40 @@
 				tc.usedLck.Unlock()
 				source = make(chan []interface{})
 				go func() {
+				Feed:
 					for _, v := range tc.retvals[idx:] {
-						source <- v
-						time.Sleep(10 * time.Millisecond)
+						select {
+						case source <- v:
+						case <-cancelled:
+							break Feed
+						}
+						select {
+						case <-time.After(10 * time.Millisecond):
+						case <-cancelled:
+							break Feed
+						}
 					}
 					close(source)
 				}()
 			}
-			for v := range source {
-				f(v...)
+		Receive:
+			for {
+				select {
+				case v, ok := <-source:
+					if !ok {
+						break Receive
+					}
+					f(v...)
+				case <-cancelled:
+					break Receive
+				}
 			}
 			d()
+			close(done)
 		}()
-		return nil
+		return &watchCancel{cancelled: cancelled, done: done, member: member}, nil
 	} else {
-		return errors.New("no way")
+		return nil, errors.New("no way")
 	}
 }
 

=== modified file 'bus/testing/testing_endpoint_test.go'
--- bus/testing/testing_endpoint_test.go	2015-02-20 17:40:57 +0000
+++ bus/testing/testing_endpoint_test.go	2015-02-27 10:54:31 +0000
@@ -17,11 +17,13 @@
 package testing
 
 import (
+	"testing"
+	"time"
+
 	. "launchpad.net/gocheck"
+
 	"launchpad.net/ubuntu-push/bus"
 	"launchpad.net/ubuntu-push/testing/condition"
-	"testing"
-	"time"
 )
 
 // hook up gocheck
@@ -100,8 +102,9 @@
 	var m, n uint32 = 42, 17
 	endp := NewTestingEndpoint(nil, condition.Work(true), m, n)
 	ch := make(chan uint32)
-	e := endp.WatchSignal("what", func(us ...interface{}) { ch <- us[0].(uint32) }, func() { close(ch) })
-	c.Check(e, IsNil)
+	w, e := endp.WatchSignal("which", func(us ...interface{}) { ch <- us[0].(uint32) }, func() { close(ch) })
+	c.Assert(e, IsNil)
+	defer w.Cancel()
 	c.Check(<-ch, Equals, m)
 	c.Check(<-ch, Equals, n)
 }
@@ -110,8 +113,9 @@
 func (s *TestingEndpointSuite) TestWatchDestructor(c *C) {
 	endp := NewTestingEndpoint(nil, condition.Work(true))
 	ch := make(chan uint32)
-	e := endp.WatchSignal("what", func(us ...interface{}) {}, func() { close(ch) })
-	c.Check(e, IsNil)
+	w, e := endp.WatchSignal("what", func(us ...interface{}) {}, func() { close(ch) })
+	c.Assert(e, IsNil)
+	defer w.Cancel()
 	_, ok := <-ch
 	c.Check(ok, Equals, false)
 }
@@ -130,8 +134,9 @@
 // Test that WatchSignal() with a negative condition returns an error.
 func (s *TestingEndpointSuite) TestWatchFails(c *C) {
 	endp := NewTestingEndpoint(nil, condition.Work(false))
-	e := endp.WatchSignal("what", func(us ...interface{}) {}, func() {})
+	w, e := endp.WatchSignal("what", func(us ...interface{}) {}, func() {})
 	c.Check(e, NotNil)
+	c.Check(w, IsNil)
 }
 
 // Test WatchSignal can use a watchSource instead of a timeout and retvals (if
@@ -146,8 +151,9 @@
 	endp := NewTestingEndpoint(nil, condition.Work(true), 0, 0)
 	SetWatchSource(endp, "what", watchTicker)
 	ch := make(chan int)
-	e := endp.WatchSignal("what", func(us ...interface{}) {}, func() { close(ch) })
-	c.Check(e, IsNil)
+	w, e := endp.WatchSignal("what", func(us ...interface{}) {}, func() { close(ch) })
+	c.Assert(e, IsNil)
+	defer w.Cancel()
 
 	close(watchTicker)
 	// wait for the destructor to be called
@@ -161,6 +167,19 @@
 	c.Assert(len(watchTicker), Equals, 0)
 }
 
+// Test that WatchSignal() calls the destructor callback when canceled.
+func (s *TestingEndpointSuite) TestWatchCancel(c *C) {
+	endp := NewTestingEndpoint(nil, condition.Work(true))
+	ch := make(chan uint32)
+	w, e := endp.WatchSignal("what", func(us ...interface{}) {}, func() { close(ch) })
+	c.Assert(e, IsNil)
+	defer w.Cancel()
+	SetWatchSource(endp, "what", make(chan []interface{}))
+	w.Cancel()
+	_, ok := <-ch
+	c.Check(ok, Equals, false)
+}
+
 // Tests that GetProperty() works
 func (s *TestingEndpointSuite) TestGetProperty(c *C) {
 	var m uint32 = 42

=== modified file 'client/client.go'
--- client/client.go	2015-01-26 21:00:27 +0000
+++ client/client.go	2015-02-27 10:54:31 +0000
@@ -280,8 +280,10 @@
 
 // takeTheBus starts the connection(s) to D-Bus and sets up associated event channels
 func (client *PushClient) takeTheBus() error {
-	go connectivity.ConnectedState(client.connectivityEndp,
-		client.config.ConnectivityConfig, client.log, client.connCh)
+	fmt.Println("FOO")
+	cs := connectivity.New(client.connectivityEndp,
+		client.config.ConnectivityConfig, client.log)
+	go cs.Track(client.connCh)
 	util.NewAutoRedialer(client.systemImageEndp).Redial()
 	sysimg := systemimage.New(client.systemImageEndp, client.log)
 	info, err := sysimg.Info()

=== modified file 'client/client_test.go'
--- client/client_test.go	2015-02-24 14:10:05 +0000
+++ client/client_test.go	2015-02-27 10:54:31 +0000
@@ -206,12 +206,7 @@
 }
 
 func (cs *clientSuite) TearDownTest(c *C) {
-	//fmt.Println("GOROUTINE# ", runtime.NumGoroutine())
-	/*
-	   var x [16*1024]byte
-	   sz := runtime.Stack(x[:], true)
-	   fmt.Println(string(x[:sz]))
-	*/
+	//helpers.DumpGoroutines()
 }
 
 type sqlientSuite struct{ clientSuite }

=== modified file 'testing/helpers.go'
--- testing/helpers.go	2015-01-22 09:52:07 +0000
+++ testing/helpers.go	2015-02-27 10:54:31 +0000
@@ -170,3 +170,12 @@
 	}
 	return purl
 }
+
+// DumpGoroutines dumps current goroutines.
+func DumpGoroutines() {
+	var buf [64 * 1024]byte
+	sz := runtime.Stack(buf[:], true)
+	dump := string(buf[:sz])
+	fmt.Println(dump)
+	fmt.Println("#goroutines#", strings.Count("\n"+dump, "\ngoroutine "))
+}