etcd watch + test (#168)

examples/complex-etcd/main.go

@@ -6,6 +6,7 @@ import (
 	"context"
 	"fmt"
 	"log"
+	"os/exec"
 	"strings"
 	"time"
 
@@ -227,5 +228,61 @@ func main() {
 	}
 
 	fmt.Printf("Third (combined) request test passed.\n")
+
+	kCheck.Delete("")
+
+	// Watch test
+
+	sKey = "child"
+	providerCfg = etcd.Config{
+		Endpoints:   []string{"localhost:2379"},
+		DialTimeout: time.Second * 5,
+		Prefix:      true,
+		Key:         "child",
+	}
+
+	provider = etcd.Provider(providerCfg)
+
+	if err := kCheck.Load(provider, nil); err != nil {
+		log.Fatalf("error loading config: %v", err)
+	}
+
+	changedC := make(chan string, 1)
+
+	provider.Watch(func(event interface{}, err error) {
+		if err != nil {
+			fmt.Printf("Unexpected error: %v", err)
+			return
+		}
+
+		kCheck.Load(provider, nil)
+		changedC <- kCheck.String(string(event.(*clientv3.Event).Kv.Key))
+	})
+
+	var newVal string = "Brian"
+	cmd := exec.Command("etcdctl", "put", "child1", newVal)
+	err = cmd.Run()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	if strings.Compare(newVal, <-changedC) != 0 {
+		fmt.Printf("Watch failed: new value comparison FAILED\n")
+		return
+	}
+
+	newVal = "Kate"
+	cmd = exec.Command("etcdctl", "put", "child2", newVal)
+	err = cmd.Run()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	if strings.Compare(newVal, <-changedC) != 0 {
+		fmt.Printf("Watch failed: new value comparison FAILED\n")
+		return
+	}
+	fmt.Printf("Watch test passed.\n")
+
 	fmt.Printf("ALL TESTS PASSED\n")
 }

providers/etcd/etcd.go

@@ -92,3 +92,23 @@ func (e *Etcd) Read() (map[string]interface{}, error) {
 
 	return mp, nil
 }
+
+func (e *Etcd) Watch(cb func(event interface{}, err error)) error {
+	var w clientv3.WatchChan
+
+	go func() {
+		if e.cfg.Prefix {
+			w = e.client.Watch(context.Background(), e.cfg.Key, clientv3.WithPrefix())
+		} else {
+			w = e.client.Watch(context.Background(), e.cfg.Key)
+		}
+
+		for wresp := range w {
+			for _, ev := range wresp.Events {
+				cb(ev, nil)
+			}
+		}
+	}()
+
+	return nil
+}