watch: Remove DSL in place of JSON

watch/funcs.go

@@ -2,14 +2,13 @@ package watch
 
 import (
 	"fmt"
-	"strconv"
 
 	"github.com/armon/consul-api"
 )
 
 // watchFactory is a function that can create a new WatchFunc
 // from a parameter configuration
-type watchFactory func(params map[string][]string) (WatchFunc, error)
+type watchFactory func(params map[string]interface{}) (WatchFunc, error)
 
 // watchFuncFactory maps each type to a factory function
 var watchFuncFactory map[string]watchFactory
@@ -26,7 +25,7 @@ func init() {
 }
 
 // keyWatch is used to return a key watching function
-func keyWatch(params map[string][]string) (WatchFunc, error) {
+func keyWatch(params map[string]interface{}) (WatchFunc, error) {
 	var key string
 	if err := assignValue(params, "key", &key); err != nil {
 		return nil, err
@@ -51,7 +50,7 @@ func keyWatch(params map[string][]string) (WatchFunc, error) {
 }
 
 // keyPrefixWatch is used to return a key prefix watching function
-func keyPrefixWatch(params map[string][]string) (WatchFunc, error) {
+func keyPrefixWatch(params map[string]interface{}) (WatchFunc, error) {
 	var prefix string
 	if err := assignValue(params, "prefix", &prefix); err != nil {
 		return nil, err
@@ -73,7 +72,7 @@ func keyPrefixWatch(params map[string][]string) (WatchFunc, error) {
 }
 
 // servicesWatch is used to watch the list of available services
-func servicesWatch(params map[string][]string) (WatchFunc, error) {
+func servicesWatch(params map[string]interface{}) (WatchFunc, error) {
 	fn := func(p *WatchPlan) (uint64, interface{}, error) {
 		catalog := p.client.Catalog()
 		opts := consulapi.QueryOptions{WaitIndex: p.lastIndex}
@@ -87,7 +86,7 @@ func servicesWatch(params map[string][]string) (WatchFunc, error) {
 }
 
 // nodesWatch is used to watch the list of available nodes
-func nodesWatch(params map[string][]string) (WatchFunc, error) {
+func nodesWatch(params map[string]interface{}) (WatchFunc, error) {
 	fn := func(p *WatchPlan) (uint64, interface{}, error) {
 		catalog := p.client.Catalog()
 		opts := consulapi.QueryOptions{WaitIndex: p.lastIndex}
@@ -101,8 +100,8 @@ func nodesWatch(params map[string][]string) (WatchFunc, error) {
 }
 
 // serviceWatch is used to watch a specific service for changes
-func serviceWatch(params map[string][]string) (WatchFunc, error) {
+func serviceWatch(params map[string]interface{}) (WatchFunc, error) {
-	var service, tag, passingRaw string
+	var service, tag string
 	if err := assignValue(params, "service", &service); err != nil {
 		return nil, err
 	}
@@ -114,16 +113,9 @@ func serviceWatch(params map[string][]string) (WatchFunc, error) {
 		return nil, err
 	}
 
-	if err := assignValue(params, "passingonly", &passingRaw); err != nil {
-		return nil, err
-	}
 	passingOnly := false
-	if passingRaw != "" {
+	if err := assignValueBool(params, "passingonly", &passingOnly); err != nil {
-		b, err := strconv.ParseBool(passingRaw)
+		return nil, err
-		if err != nil {
-			return nil, fmt.Errorf("Failed to parse passingonly value: %v", err)
-		}
-		passingOnly = b
 	}
 
 	fn := func(p *WatchPlan) (uint64, interface{}, error) {
@@ -139,7 +131,7 @@ func serviceWatch(params map[string][]string) (WatchFunc, error) {
 }
 
 // checksWatch is used to watch a specific checks in a given state
-func checksWatch(params map[string][]string) (WatchFunc, error) {
+func checksWatch(params map[string]interface{}) (WatchFunc, error) {
 	var service, state string
 	if err := assignValue(params, "service", &service); err != nil {
 		return nil, err

watch/funcs_test.go

@@ -18,7 +18,7 @@ func TestKeyWatch(t *testing.T) {
 	if consulAddr == "" {
 		t.Skip()
 	}
-	plan := mustParse(t, "type:key key:foo/bar/baz")
+	plan := mustParse(t, `{"type":"key", "key":"foo/bar/baz"}`)
 	invoke := 0
 	plan.Handler = func(idx uint64, raw interface{}) {
 		if invoke == 0 {
@@ -72,7 +72,7 @@ func TestKeyPrefixWatch(t *testing.T) {
 	if consulAddr == "" {
 		t.Skip()
 	}
-	plan := mustParse(t, "type:keyprefix prefix:foo/")
+	plan := mustParse(t, `{"type":"keyprefix", "prefix":"foo/"}`)
 	invoke := 0
 	plan.Handler = func(idx uint64, raw interface{}) {
 		if invoke == 0 {
@@ -128,7 +128,7 @@ func TestServicesWatch(t *testing.T) {
 	if consulAddr == "" {
 		t.Skip()
 	}
-	plan := mustParse(t, "type:services")
+	plan := mustParse(t, `{"type":"services"}`)
 	invoke := 0
 	plan.Handler = func(idx uint64, raw interface{}) {
 		if invoke == 0 {
@@ -171,7 +171,7 @@ func TestNodesWatch(t *testing.T) {
 	if consulAddr == "" {
 		t.Skip()
 	}
-	plan := mustParse(t, "type:nodes")
+	plan := mustParse(t, `{"type":"nodes"}`)
 	invoke := 0
 	plan.Handler = func(idx uint64, raw interface{}) {
 		if invoke == 0 {
@@ -220,7 +220,7 @@ func TestServiceWatch(t *testing.T) {
 	if consulAddr == "" {
 		t.Skip()
 	}
-	plan := mustParse(t, "type:service service:foo tag:bar passingonly:true")
+	plan := mustParse(t, `{"type":"service", "service":"foo", "tag":"bar", "passingonly":true}`)
 	invoke := 0
 	plan.Handler = func(idx uint64, raw interface{}) {
 		if invoke == 0 {
@@ -269,7 +269,7 @@ func TestChecksWatch_State(t *testing.T) {
 	if consulAddr == "" {
 		t.Skip()
 	}
-	plan := mustParse(t, "type:checks state:warning")
+	plan := mustParse(t, `{"type":"checks", "state":"warning"}`)
 	invoke := 0
 	plan.Handler = func(idx uint64, raw interface{}) {
 		if invoke == 0 {
@@ -329,7 +329,7 @@ func TestChecksWatch_Service(t *testing.T) {
 	if consulAddr == "" {
 		t.Skip()
 	}
-	plan := mustParse(t, "type:checks service:foobar")
+	plan := mustParse(t, `{"type":"checks", "service":"foobar"}`)
 	invoke := 0
 	plan.Handler = func(idx uint64, raw interface{}) {
 		if invoke == 0 {

watch/plan.go

@@ -47,7 +47,7 @@ OUTER:
 
 		// Handle an error in the watch function
 		if err != nil {
-			log.Printf("consul.watch: Watch '%s' errored: %v", p.Query, err)
+			log.Printf("consul.watch: Watch (type: %s) errored: %v", p.Type, err)
 
 			// Perform an exponential backoff
 			failures++

watch/plan_test.go

@@ -9,7 +9,7 @@ func init() {
 	watchFuncFactory["noop"] = noopWatch
 }
 
-func noopWatch(params map[string][]string) (WatchFunc, error) {
+func noopWatch(params map[string]interface{}) (WatchFunc, error) {
 	fn := func(p *WatchPlan) (uint64, interface{}, error) {
 		idx := p.lastIndex + 1
 		return idx, idx, nil
@@ -18,7 +18,8 @@ func noopWatch(params map[string][]string) (WatchFunc, error) {
 }
 
 func mustParse(t *testing.T, q string) *WatchPlan {
-	plan, err := Parse(q)
+	params := makeParams(t, q)
+	plan, err := Parse(params)
 	if err != nil {
 		t.Fatalf("err: %v", err)
 	}
@@ -26,7 +27,7 @@ func mustParse(t *testing.T, q string) *WatchPlan {
 }
 
 func TestRun_Stop(t *testing.T) {
-	plan := mustParse(t, "type:noop")
+	plan := mustParse(t, `{"type":"noop"}`)
 	var expect uint64 = 1
 	plan.Handler = func(idx uint64, val interface{}) {
 		if idx != expect {

watch/watch.go

@@ -2,7 +2,6 @@ package watch
 
 import (
 	"fmt"
-	"strings"
 	"sync"
 
 	"github.com/armon/consul-api"
@@ -13,11 +12,10 @@ import (
 // This view is watched for changes and a handler is invoked to take any
 // appropriate actions.
 type WatchPlan struct {
-	Query      string
 	Datacenter string
 	Token      string
 	Type       string
-	Exempt     map[string][]string
+	Exempt     map[string]interface{}
 	Func       WatchFunc
 	Handler    HandlerFunc
 
@@ -38,20 +36,14 @@ type WatchFunc func(*WatchPlan) (uint64, interface{}, error)
 type HandlerFunc func(uint64, interface{})
 
 // Parse takes a watch query and compiles it into a WatchPlan or an error
-func Parse(query string) (*WatchPlan, error) {
+func Parse(params map[string]interface{}) (*WatchPlan, error) {
-	return ParseExempt(query, nil)
+	return ParseExempt(params, nil)
 }
 
 // ParseExempt takes a watch query and compiles it into a WatchPlan or an error
 // Any exempt parameters are stored in the Exempt map
-func ParseExempt(query string, exempt []string) (*WatchPlan, error) {
+func ParseExempt(params map[string]interface{}, exempt []string) (*WatchPlan, error) {
-	tokens, err := tokenize(query)
-	if err != nil {
-		return nil, fmt.Errorf("Failed to parse: %v", err)
-	}
-	params := collapse(tokens)
 	plan := &WatchPlan{
-		Query:  query,
 		stopCh: make(chan struct{}),
 	}
 
@@ -86,7 +78,7 @@ func ParseExempt(query string, exempt []string) (*WatchPlan, error) {
 
 	// Remove the exempt parameters
 	if len(exempt) > 0 {
-		plan.Exempt = make(map[string][]string)
+		plan.Exempt = make(map[string]interface{})
 		for _, ex := range exempt {
 			val, ok := params[ex]
 			if ok {
@@ -107,121 +99,28 @@ func ParseExempt(query string, exempt []string) (*WatchPlan, error) {
 	return plan, nil
 }
 
-// assignValue is used to extract a value ensuring it is only
+// assignValue is used to extract a value ensuring it is a string
-// defined once
+func assignValue(params map[string]interface{}, name string, out *string) error {
-func assignValue(params map[string][]string, name string, out *string) error {
+	if raw, ok := params[name]; ok {
-	if vals, ok := params[name]; ok {
+		val, ok := raw.(string)
-		if len(vals) != 1 {
+		if !ok {
-			return fmt.Errorf("Multiple definitions of %s", name)
+			return fmt.Errorf("Expecting %s to be a string")
 		}
-		*out = vals[0]
+		*out = val
 		delete(params, name)
 	}
 	return nil
 }
 
-// token is used to represent a "datacenter:foobar" pair, where
+// assignValueBool is used to extract a value ensuring it is a bool
-// datacenter is the param and foobar is the value
+func assignValueBool(params map[string]interface{}, name string, out *bool) error {
-type token struct {
+	if raw, ok := params[name]; ok {
-	param string
+		val, ok := raw.(bool)
-	val   string
+		if !ok {
-}
+			return fmt.Errorf("Expecting %s to be a boolean")
-
-func (t *token) GoString() string {
-	return fmt.Sprintf("%#v", *t)
-}
-
-// tokenize splits a query string into a slice of tokens
-func tokenize(query string) ([]*token, error) {
-	var tokens []*token
-	for i := 0; i < len(query); i++ {
-		char := query[i]
-
-		// Ignore whitespace
-		if char == ' ' || char == '\t' || char == '\n' {
-			continue
 		}
-
+		*out = val
-		// Read the next token
+		delete(params, name)
-		next, offset, err := readToken(query[i:])
-		if err != nil {
-			return nil, err
-		}
-
-		// Store the token
-		tokens = append(tokens, next)
-
-		// Increment the offset
-		i += offset
 	}
-	return tokens, nil
+	return nil
-}
-
-// readToken is used to read a single token
-func readToken(query string) (*token, int, error) {
-	// Get the token
-	param, offset, err := readParameter(query)
-	if err != nil {
-		return nil, 0, err
-	}
-
-	// Get the value
-	query = query[offset:]
-	val, offset2, err := readValue(query)
-	if err != nil {
-		return nil, 0, err
-	}
-
-	// Return the new token
-	token := &token{
-		param: param,
-		val:   val,
-	}
-	return token, offset + offset2, nil
-}
-
-// readParameter scans for the next parameter
-func readParameter(query string) (string, int, error) {
-	for i := 0; i < len(query); i++ {
-		char := query[i]
-		if char == ':' {
-			if i == 0 {
-				return "", 0, fmt.Errorf("Missing parameter name")
-			} else {
-				return query[:i], i + 1, nil
-			}
-		}
-	}
-	return "", 0, fmt.Errorf("Parameter delimiter not found")
-}
-
-// readValue is used to scan for the next value
-func readValue(query string) (string, int, error) {
-	// Handle quoted values
-	if query[0] == '\'' || query[0] == '"' {
-		quoteChar := query[0:1]
-		endChar := strings.Index(query[1:], quoteChar)
-		if endChar == -1 {
-			return "", 0, fmt.Errorf("Missing end of quotation")
-		}
-		return query[1 : endChar+1], endChar + 2, nil
-	}
-
-	// Look for white space
-	endChar := strings.IndexAny(query, " \t\n")
-	if endChar == -1 {
-		return query, len(query), nil
-	}
-	return query[:endChar], endChar, nil
-}
-
-// collapse is used to collapse a token stream into a map
-// of parameter name to list of values.
-func collapse(tokens []*token) map[string][]string {
-	out := make(map[string][]string)
-	for _, t := range tokens {
-		existing := out[t.param]
-		out[t.param] = append(existing, t.val)
-	}
-	return out
 }

watch/watch_test.go

@@ -1,96 +1,14 @@
 package watch
 
 import (
-	"fmt"
+	"bytes"
-	"reflect"
+	"encoding/json"
 	"testing"
 )
 
-func TestTokenize(t *testing.T) {
-	type tcase struct {
-		in  string
-		out []*token
-		err error
-	}
-	cases := []tcase{
-		tcase{
-			"",
-			nil,
-			nil,
-		},
-		tcase{
-			"foo:bar bar:baz	zip:zap",
-			[]*token{
-				&token{"foo", "bar"},
-				&token{"bar", "baz"},
-				&token{"zip", "zap"},
-			},
-			nil,
-		},
-		tcase{
-			"foo:\"long input here\" after:this",
-			[]*token{
-				&token{"foo", "long input here"},
-				&token{"after", "this"},
-			},
-			nil,
-		},
-		tcase{
-			"foo:'long input here' after:this",
-			[]*token{
-				&token{"foo", "long input here"},
-				&token{"after", "this"},
-			},
-			nil,
-		},
-		tcase{
-			"foo:'long input here after:this",
-			nil,
-			fmt.Errorf("Missing end of quotation"),
-		},
-		tcase{
-			"foo",
-			nil,
-			fmt.Errorf("Parameter delimiter not found"),
-		},
-		tcase{
-			":val",
-			nil,
-			fmt.Errorf("Missing parameter name"),
-		},
-	}
-
-	for _, tc := range cases {
-		tokens, err := tokenize(tc.in)
-		if err != nil && tc.err == nil {
-			t.Fatalf("%s: err: %v", tc.in, err)
-		} else if tc.err != nil && (err == nil || err.Error() != tc.err.Error()) {
-			t.Fatalf("%s: bad err: %v", tc.in, err)
-		}
-		if !reflect.DeepEqual(tokens, tc.out) {
-			t.Fatalf("%s: bad: %#v %#v", tc.in, tokens, tc.out)
-		}
-	}
-}
-
-func TestCollapse(t *testing.T) {
-	inp := "type:key key:foo key:bar"
-	tokens, err := tokenize(inp)
-	if err != nil {
-		t.Fatalf("err: %v", err)
-	}
-	out := collapse(tokens)
-	expect := map[string][]string{
-		"type": []string{"key"},
-		"key":  []string{"foo", "bar"},
-	}
-	if !reflect.DeepEqual(out, expect) {
-		t.Fatalf("bad: %#v", out)
-	}
-}
-
 func TestParseBasic(t *testing.T) {
-	p, err := Parse("type:key datacenter:dc2 token:12345 key:foo")
+	params := makeParams(t, `{"type":"key", "datacenter":"dc2", "token":"12345", "key":"foo"}`)
+	p, err := Parse(params)
 	if err != nil {
 		t.Fatalf("err: %v", err)
 	}
@@ -106,7 +24,8 @@ func TestParseBasic(t *testing.T) {
 }
 
 func TestParse_exempt(t *testing.T) {
-	p, err := ParseExempt("type:key key:foo handler:foobar", []string{"handler"})
+	params := makeParams(t, `{"type":"key", "key":"foo", "handler": "foobar"}`)
+	p, err := ParseExempt(params, []string{"handler"})
 	if err != nil {
 		t.Fatalf("err: %v", err)
 	}
@@ -114,7 +33,16 @@ func TestParse_exempt(t *testing.T) {
 		t.Fatalf("Bad: %#v", p)
 	}
 	ex := p.Exempt["handler"]
-	if len(ex) != 1 && ex[0] != "foobar" {
+	if ex != "foobar" {
 		t.Fatalf("bad: %v", ex)
 	}
 }
+
+func makeParams(t *testing.T, s string) map[string]interface{} {
+	var out map[string]interface{}
+	dec := json.NewDecoder(bytes.NewReader([]byte(s)))
+	if err := dec.Decode(&out); err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	return out
+}