watch: test key watch

watch/funcs.go

@@ -35,6 +35,9 @@ func keyWatch(params map[string][]string) (WatchFunc, error) {
 		if err != nil {
 			return 0, nil, err
 		}
+		if pair == nil {
+			return meta.LastIndex, nil, err
+		}
 		return meta.LastIndex, pair, err
 	}
 	return fn, nil

watch/plan.go

@@ -34,6 +34,7 @@ func (p *WatchPlan) Run(address string) error {
 
 	// Loop until we are canceled
 	failures := 0
+OUTER:
 	for !p.shouldStop() {
 		// Invoke the handler
 		index, result, err := p.Func(p)
@@ -56,7 +57,7 @@ func (p *WatchPlan) Run(address string) error {
 			}
 			select {
 			case <-time.After(retry):
-				continue
+				continue OUTER
 			case <-p.stopCh:
 				return nil
 			}
@@ -78,7 +79,9 @@ func (p *WatchPlan) Run(address string) error {
 
 		// Handle the updated result
 		p.lastResult = result
-		p.Handler(index, result)
+		if p.Handler != nil {
+			p.Handler(index, result)
+		}
 	}
 	return nil
 }

watch/plan_test.go

@@ -17,11 +17,16 @@ func noopWatch(params map[string][]string) (WatchFunc, error) {
 	return fn, nil
 }
 
-func TestRun_Stop(t *testing.T) {
-	plan, err := Parse("type:noop")
+func mustParse(t *testing.T, q string) *WatchPlan {
+	plan, err := Parse(q)
 	if err != nil {
 		t.Fatalf("err: %v", err)
 	}
+	return plan
+}
+
+func TestRun_Stop(t *testing.T) {
+	plan := mustParse(t, "type:noop")
 	var expect uint64 = 1
 	plan.Handler = func(idx uint64, val interface{}) {
 		if idx != expect {
@@ -37,7 +42,7 @@ func TestRun_Stop(t *testing.T) {
 		plan.Stop()
 	})
 
-	err = plan.Run("127.0.0.1:8500")
+	err := plan.Run("127.0.0.1:8500")
 	if err != nil {
 		t.Fatalf("err: %v", err)
 	}