watch: convert TestNodesWatch to use channels

watch/funcs_test.go

@@ -2,6 +2,8 @@ package watch_test
 
 import (
 	"encoding/json"
+	"errors"
+	"sync"
 	"testing"
 	"time"
 
@@ -10,6 +12,8 @@ import (
 	"github.com/hashicorp/consul/watch"
 )
 
+var errBadContent = errors.New("bad content")
+
 func TestKeyWatch(t *testing.T) {
 	a := agent.NewTestAgent(t.Name(), ``)
 	defer a.Shutdown()
@@ -227,49 +231,50 @@ func TestNodesWatch(t *testing.T) {
 	a := agent.NewTestAgent(t.Name(), ``)
 	defer a.Shutdown()
 
+	invoke := make(chan error)
 	plan := mustParse(t, `{"type":"nodes"}`)
-	invoke := 0
 	plan.Handler = func(idx uint64, raw interface{}) {
-		if invoke == 0 {
-			if raw == nil {
-				return
-			}
-			v, ok := raw.([]*consulapi.Node)
-			if !ok || len(v) == 0 {
-				t.Fatalf("Bad: %#v", raw)
-			}
-			invoke++
+		if raw == nil {
+			return // ignore
 		}
+		v, ok := raw.([]*consulapi.Node)
+		if !ok || len(v) == 0 {
+			return // ignore
+		}
+		invoke <- nil
 	}
 
+	var wg sync.WaitGroup
+	wg.Add(1)
 	go func() {
-		time.Sleep(20 * time.Millisecond)
-		plan.Stop()
-
+		defer wg.Done()
 		catalog := a.Client().Catalog()
+
+		time.Sleep(20 * time.Millisecond)
 		reg := &consulapi.CatalogRegistration{
 			Node:       "foobar",
 			Address:    "1.1.1.1",
 			Datacenter: "dc1",
 		}
-		catalog.Register(reg, nil)
-		time.Sleep(20 * time.Millisecond)
-		dereg := &consulapi.CatalogDeregistration{
-			Node:       "foobar",
-			Address:    "1.1.1.1",
-			Datacenter: "dc1",
+		if _, err := catalog.Register(reg, nil); err != nil {
+			t.Fatalf("err: %v", err)
 		}
-		catalog.Deregister(dereg, nil)
 	}()
 
-	err := plan.Run(a.HTTPAddr())
-	if err != nil {
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		if err := plan.Run(a.HTTPAddr()); err != nil {
+			t.Fatalf("err: %v", err)
+		}
+	}()
+
+	if err := <-invoke; err != nil {
 		t.Fatalf("err: %v", err)
 	}
 
-	if invoke == 0 {
-		t.Fatalf("bad: %v", invoke)
-	}
+	plan.Stop()
+	wg.Wait()
 }
 
 func TestServiceWatch(t *testing.T) {