watch: convert TestKeyWatch to use channels

watch/funcs_test.go

@@ -18,54 +18,53 @@ func TestKeyWatch(t *testing.T) {
 	a := agent.NewTestAgent(t.Name(), ``)
 	defer a.Shutdown()
 
+	invoke := make(chan error)
 	plan := mustParse(t, `{"type":"key", "key":"foo/bar/baz"}`)
-	invoke := 0
 	plan.Handler = func(idx uint64, raw interface{}) {
-		if invoke == 0 {
+		if raw == nil {
-			if raw == nil {
+			return // ignore
-				return
-			}
-			v, ok := raw.(*consulapi.KVPair)
-			if !ok || v == nil || string(v.Value) != "test" {
-				t.Fatalf("Bad: %#v", raw)
-			}
-			invoke++
 		}
+		v, ok := raw.(*consulapi.KVPair)
+		if !ok || v == nil {
+			return // ignore
+		}
+		if string(v.Value) != "test" {
+			invoke <- errBadContent
+			return
+		}
+		invoke <- nil
 	}
 
+	var wg sync.WaitGroup
+	wg.Add(1)
 	go func() {
-		defer plan.Stop()
+		defer wg.Done()
-		time.Sleep(20 * time.Millisecond)
-
 		kv := a.Client().KV()
+
+		time.Sleep(20 * time.Millisecond)
 		pair := &consulapi.KVPair{
 			Key:   "foo/bar/baz",
 			Value: []byte("test"),
 		}
-		_, err := kv.Put(pair, nil)
+		if _, err := kv.Put(pair, nil); err != nil {
-		if err != nil {
-			t.Fatalf("err: %v", err)
-		}
-
-		// Wait for the query to run
-		time.Sleep(20 * time.Millisecond)
-		plan.Stop()
-
-		// Delete the key
-		_, err = kv.Delete("foo/bar/baz", nil)
-		if err != nil {
 			t.Fatalf("err: %v", err)
 		}
 	}()
 
-	err := plan.Run(a.HTTPAddr())
+	wg.Add(1)
-	if err != nil {
+	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 {
+	plan.Stop()
-		t.Fatalf("bad: %v", invoke)
+	wg.Wait()
-	}
 }
 
 func TestKeyWatch_With_PrefixDelete(t *testing.T) {