Merge pull request #1787 from fcrisciani/goroutine_leak

Fix leak of handleTableEvents

Author: mavenugo

agent.go

@@ -722,15 +722,13 @@ func (n *network) cancelDriverWatches() {
 	}
 }
 
-func (c *controller) handleTableEvents(ch chan events.Event, fn func(events.Event)) {
+func (c *controller) handleTableEvents(ch *events.Channel, fn func(events.Event)) {
 	for {
 		select {
-		case ev, ok := <-ch:
-			if !ok {
-				return
-			}
-
+		case ev := <-ch.C:
 			fn(ev)
+		case <-ch.Done():
+			return
 		}
 	}
 }

networkdb/networkdb_test.go

@@ -339,17 +339,17 @@ func TestNetworkDBWatch(t *testing.T) {
 	err = dbs[0].CreateEntry("test_table", "network1", "test_key", []byte("test_value"))
 	assert.NoError(t, err)
 
-	testWatch(t, ch, CreateEvent{}, "test_table", "network1", "test_key", "test_value")
+	testWatch(t, ch.C, CreateEvent{}, "test_table", "network1", "test_key", "test_value")
 
 	err = dbs[0].UpdateEntry("test_table", "network1", "test_key", []byte("test_updated_value"))
 	assert.NoError(t, err)
 
-	testWatch(t, ch, UpdateEvent{}, "test_table", "network1", "test_key", "test_updated_value")
+	testWatch(t, ch.C, UpdateEvent{}, "test_table", "network1", "test_key", "test_updated_value")
 
 	err = dbs[0].DeleteEntry("test_table", "network1", "test_key")
 	assert.NoError(t, err)
 
-	testWatch(t, ch, DeleteEvent{}, "test_table", "network1", "test_key", "")
+	testWatch(t, ch.C, DeleteEvent{}, "test_table", "network1", "test_key", "")
 
 	cancel()
 	closeNetworkDBInstances(dbs)

networkdb/watch.go

@@ -43,7 +43,7 @@ type DeleteEvent event
 // filter is an empty string it acts as a wildcard for that
 // field. Watch returns a channel of events, where the events will be
 // sent.
-func (nDB *NetworkDB) Watch(tname, nid, key string) (chan events.Event, func()) {
+func (nDB *NetworkDB) Watch(tname, nid, key string) (*events.Channel, func()) {
 	var matcher events.Matcher
 
 	if tname != "" || nid != "" || key != "" {
@@ -82,7 +82,7 @@ func (nDB *NetworkDB) Watch(tname, nid, key string) (chan events.Event, func())
 	}
 
 	nDB.broadcaster.Add(sink)
-	return ch.C, func() {
+	return ch, func() {
 		nDB.broadcaster.Remove(sink)
 		ch.Close()
 		sink.Close()