Test and improve NounDataSource

Summary:
- added first focused test that integrates DataSources, NounDataSource, and
  marshaled.DataSource
- added name field to update data
- simplified DataSources.Add method name
- implemented data source removal with observation
- made formatNames order stable under marshaled.NewDataSource
- addressed some govet errors, and started a makefile to ease running tests and
  linting

Reviewers: abg

Reviewed By: abg

Makefile

@@ -0,0 +1,7 @@
+.PHONY: lint
+
+lint:
+	go vet $$(glide novendor)
+
+test: lint
+	go test $$(glide novendor)

data_source.go

@@ -14,18 +14,18 @@ var DefaultDataSources *source.DataSources
 func init() {
 	DefaultDataSources = source.NewDataSources()
 	metaNouns := meta.NewNounDataSource(DefaultDataSources)
-	DefaultDataSources.AddDataSource(marshaled.NewDataSource(metaNouns, nil))
+	DefaultDataSources.Add(marshaled.NewDataSource(metaNouns, nil))
 	DefaultDataSources.SetObserver(metaNouns)
 }
 
 // AddDataSource adds a data source to the default data sources registry.
 func AddDataSource(ds source.DataSource) error {
-	return DefaultDataSources.AddDataSource(ds)
+	return DefaultDataSources.Add(ds)
 }
 
 // AddGenericDataSource adds a generic data source to the default data sources
 // registry.
 func AddGenericDataSource(gds source.GenericDataSource) error {
 	mds := marshaled.NewDataSource(gds, nil)
-	return DefaultDataSources.AddDataSource(mds)
+	return DefaultDataSources.Add(mds)
 }

glide.yaml

@@ -2,3 +2,6 @@ package: github.com/uber-go/gwr
 import:
 - package: github.com/uber-common/stacked
   version: ^1.0.2
+- package: github.com/stretchr/testify
+  subpackages:
+  - assert

internal/marshaled/source.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"io"
 	"log"
+	"sort"
 	"strings"
 
 	"github.com/uber-go/gwr/internal"
@@ -225,6 +226,7 @@ func NewDataSource(
 		formatNames = append(formatNames, name)
 		watchers[name] = newMarshaledWatcher(src, format)
 	}
+	sort.Strings(formatNames)
 	return &DataSource{
 		source:      src,
 		formats:     formats,

internal/meta/noun_data_source.go

@@ -16,11 +16,6 @@ var nounsTextTemplate = template.Must(template.New("meta_nouns_text").Parse(`
 {{- end -}}
 `))
 
-type dataSourceUpdate struct {
-	Type string      `json:"type"`
-	Info source.Info `json:"info"`
-}
-
 // NounDataSource provides a data source that describes other data sources.  It
 // is used to implement the "/meta/nouns" data source.
 type NounDataSource struct {
@@ -66,16 +61,28 @@ func (nds *NounDataSource) GetInit() interface{} {
 
 // SetWatcher implements GenericDataSource by retaining a reference to the
 // passed watcher.  Updates are later sent to the watcher when new data sources
-// are added.  Currently there is no data source removal, but when there is,
-// removal updates will be sent here (TODO change this once we implement source
-// removal).
+// are added and removed.
 func (nds *NounDataSource) SetWatcher(watcher source.GenericDataWatcher) {
 	nds.watcher = watcher
 }
 
 // SourceAdded is called whenever a source is added to the DataSources.
 func (nds *NounDataSource) SourceAdded(ds source.DataSource) {
 	if nds.watcher != nil {
-		nds.watcher.HandleItem(dataSourceUpdate{"add", source.GetInfo(ds)})
+		nds.watcher.HandleItem(struct {
+			Type string      `json:"type"`
+			Name string      `json:"name"`
+			Info source.Info `json:"info"`
+		}{"add", ds.Name(), source.GetInfo(ds)})
+	}
+}
+
+// SourceRemoved is called whenever a source is removed from the DataSources.
+func (nds *NounDataSource) SourceRemoved(ds source.DataSource) {
+	if nds.watcher != nil {
+		nds.watcher.HandleItem(struct {
+			Type string `json:"type"`
+			Name string `json:"name"`
+		}{"remove", ds.Name()})
 	}
 }

internal/meta/noun_data_source_test.go

@@ -0,0 +1,119 @@
+package meta_test
+
+import (
+	"bufio"
+	"os"
+	"strings"
+	"testing"
+	"text/template"
+
+	"github.com/uber-go/gwr/internal/marshaled"
+	"github.com/uber-go/gwr/internal/meta"
+	"github.com/uber-go/gwr/source"
+
+	"github.com/stretchr/testify/assert"
+)
+
+type dummyDataSource struct {
+	name  string
+	attrs map[string]interface{}
+	tmpl  *template.Template
+}
+
+func (dds *dummyDataSource) Name() string {
+	return dds.name
+}
+
+func (dds *dummyDataSource) Attrs() map[string]interface{} {
+	return dds.attrs
+}
+
+func (dds *dummyDataSource) TextTemplate() *template.Template {
+	return dds.tmpl
+}
+
+func (dds *dummyDataSource) Get() interface{} {
+	return nil
+}
+
+func (dds *dummyDataSource) GetInit() interface{} {
+	return nil
+}
+
+func (dds *dummyDataSource) SetWatcher(watcher source.GenericDataWatcher) {
+}
+
+func setup() *source.DataSources {
+	dss := source.NewDataSources()
+	nds := meta.NewNounDataSource(dss)
+	dss.Add(marshaled.NewDataSource(nds, nil))
+	dss.SetObserver(nds)
+	return dss
+}
+
+func TestNounDataSource_Watch(t *testing.T) {
+	dss := setup()
+	mds := dss.Get("/meta/nouns")
+
+	r, w, err := os.Pipe()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// buf := newInternalRWC()
+	sc := bufio.NewScanner(r)
+	if err := mds.Watch("json", w); err != nil {
+		t.Fatal(err)
+	}
+
+	// verify init data
+	assertJSONScanLine(t, sc,
+		`{"/meta/nouns":{"formats":["json","text"],"attrs":null}}`,
+		"should get /meta/nouns initially")
+
+	// add a data source, observe it
+	assert.NoError(t, dss.Add(marshaled.NewDataSource(&dummyDataSource{
+		name:  "/foo",
+		attrs: map[string]interface{}{"aKey": "aVal"},
+		tmpl:  nil,
+	}, nil)), "no add error expected")
+	assertJSONScanLine(t, sc,
+		`{"name":"/foo","type":"add","info":{"formats":["json"],"attrs":{"aKey":"aVal"}}}`,
+		"should get an add event for /foo")
+
+	// add another data source, observe it
+	assert.NoError(t, dss.Add(marshaled.NewDataSource(&dummyDataSource{
+		name:  "/bar",
+		attrs: map[string]interface{}{"bKey": 123},
+		tmpl:  template.Must(template.New("bar_tmpl").Parse("")),
+	}, nil)), "no add error expected")
+	assertJSONScanLine(t, sc,
+		`{"name":"/bar","type":"add","info":{"formats":["json","text"],"attrs":{"bKey":123}}}`,
+		"should get an add event for /bar")
+
+	// remove the /foo data source, observe it
+	assert.NotNil(t, dss.Remove("/foo"), "expected a removed data source")
+	assertJSONScanLine(t, sc,
+		`{"name":"/foo","type":"remove"}`,
+		"should get a remove event for /foo")
+
+	// remove the /bar data source, observe it
+	assert.NotNil(t, dss.Remove("/bar"), "expected a removed data source")
+	assertJSONScanLine(t, sc,
+		`{"name":"/bar","type":"remove"}`,
+		"should get a remove event for /bar")
+
+	// shutdown the watch stream
+	assert.NoError(t, r.Close())
+	assert.False(t, sc.Scan(), "no more scan")
+}
+
+func assertJSONScanLine(t *testing.T, sc *bufio.Scanner, expected string, msgAndArgs ...interface{}) {
+	if !sc.Scan() {
+		assert.Fail(t, "expected to scan a JSON line", msgAndArgs...)
+	} else {
+		expected = strings.Join([]string{expected, "\n"}, "")
+		assert.JSONEq(t, sc.Text(), expected, msgAndArgs...)
+	}
+	assert.NoError(t, sc.Err())
+}

internal/protocol/redis.go

@@ -237,8 +237,6 @@ func (rm *respModel) doWatch(rconn *resp.RedisConnection) error {
 			}
 		}
 	}
-
-	return nil
 }
 
 type multiJSONMessage struct {

internal/resp/connection.go

@@ -86,8 +86,6 @@ func (rconn *RedisConnection) Consume(handler RedisHandler) error {
 	default:
 		return fmt.Errorf("unknown RESP type %#v", string(c))
 	}
-
-	return nil
 }
 
 func (rconn *RedisConnection) consumeError(handler RedisHandler) error {

internal/resp/server.go

@@ -38,7 +38,6 @@ func (h RedisServer) Serve(ln net.Listener) error {
 		}
 		go NewRedisConnection(conn, nil).Handle(h.consumer)
 	}
-	return nil
 }
 
 // IsFirstByteRespTag returns true if the first byte in the passed slice is a

serve.go

@@ -8,6 +8,7 @@ import (
 	"github.com/uber-go/gwr/internal/protocol"
 	"github.com/uber-go/gwr/internal/resp"
 	"github.com/uber-go/gwr/source"
+
 	"github.com/uber-common/stacked"
 )
 

source/sources.go

@@ -3,9 +3,11 @@ package source
 import "fmt"
 
 // DataSourcesObserver is an interface to observe data sources changes.
+//
+// Observation happens after after the source has been added (resp. removed).
 type DataSourcesObserver interface {
 	SourceAdded(ds DataSource)
-	// TODO: add removal
+	SourceRemoved(ds DataSource)
 }
 
 // DataSources is a flat collection of DataSources
@@ -39,9 +41,8 @@ func (dss *DataSources) Get(name string) DataSource {
 	return nil
 }
 
-// AddDataSource adds a DataSource, if none is
-// already defined for the given name.
-func (dss *DataSources) AddDataSource(ds DataSource) error {
+// Add a DataSource, if none is already defined for the given name.
+func (dss *DataSources) Add(ds DataSource) error {
 	name := ds.Name()
 	if _, ok := dss.sources[name]; ok {
 		return fmt.Errorf("data source already defined")
@@ -52,3 +53,16 @@ func (dss *DataSources) AddDataSource(ds DataSource) error {
 	}
 	return nil
 }
+
+// Remove a DataSource by name, if any exsits.  Returns the source removed, nil
+// if none was defined.
+func (dss *DataSources) Remove(name string) DataSource {
+	ds, ok := dss.sources[name]
+	if ok {
+		delete(dss.sources, name)
+		if dss.obs != nil {
+			dss.obs.SourceRemoved(ds)
+		}
+	}
+	return ds
+}