Fix bugs; improve tests

client_test.go

@@ -557,7 +557,7 @@ func testServiceRegisterDeregister(t *testing.T, dbClient API) {
 	}
 
 	// Wait a while to make sure background refresh is working correctly
-	time.Sleep(5 * time.Millisecond)
+	time.Sleep(serviceTTL * 2)
 
 	resp, err = dbClient.GetService("athena")
 	if err != nil {
@@ -579,8 +579,60 @@ func testServiceRegisterDeregister(t *testing.T, dbClient API) {
 		t.Fatalf("Fatal deregistering service. Err: %+v\n", err)
 	}
 
-	// Wait a while to make sure background refresh is working correctly
-	time.Sleep(5 * time.Millisecond)
+	resp, err = dbClient.GetService("athena")
+	if err != nil {
+		t.Fatalf("Fatal getting service. Err: %+v\n", err)
+	}
+
+	log.Infof("Got service list: %+v\n", resp)
+
+	if len(resp) != 0 {
+		t.Fatalf("Service still in list after deregister")
+	}
+}
+
+func TestEtcdServiceRegisterMultiple(t *testing.T) {
+	testServiceMultipleRegister(t, etcdClient)
+}
+
+func TestConsulServiceRegisterMultiple(t *testing.T) {
+	testServiceMultipleRegister(t, consulClient)
+}
+
+func testServiceMultipleRegister(t *testing.T, dbClient API) {
+	// Service info
+	service1Info := ServiceInfo{
+		ServiceName: "athena",
+		HostAddr:    "10.10.10.10",
+		Port:        4567,
+	}
+
+	// register it multiple times
+	for i := 0; i < 5; i++ {
+		if err := dbClient.RegisterService(service1Info); err != nil {
+			t.Fatalf("Fatal registering service. Err: %+v\n", err)
+		}
+		log.Infof("Registered service: %+v", service1Info)
+
+		// sleep for a second
+		time.Sleep(time.Second)
+	}
+
+	resp, err := dbClient.GetService("athena")
+	if err != nil {
+		t.Fatalf("Fatal getting service. Err: %+v\n", err)
+	}
+
+	log.Infof("Got service list: %+v\n", resp)
+
+	if (len(resp) != 1) || (resp[0] != service1Info) {
+		t.Fatalf("Resp service list did not match input")
+	}
+
+	// deregister it
+	if err := dbClient.DeregisterService(service1Info); err != nil {
+		t.Fatalf("Fatal deregistering service. Err: %+v\n", err)
+	}
 
 	resp, err = dbClient.GetService("athena")
 	if err != nil {

consulClient.go

@@ -97,8 +97,8 @@ func (cp *ConsulClient) GetObj(key string, retVal interface{}) error {
 	}
 
 	// Parse JSON response
-	if err := json.Unmarshal([]byte(resp.Value), retVal); err != nil {
-		log.Errorf("Error parsing object %s, Err %v", resp.Value, err)
+	if err := json.Unmarshal(resp.Value, retVal); err != nil {
+		log.Errorf("Error parsing object %v, Err %v", resp.Value, err)
 		return err
 	}
 
@@ -121,7 +121,7 @@ func (cp *ConsulClient) ListDir(key string) ([]string, error) {
 
 	var keys []string
 	for _, kv := range kvs {
-		keys = append(keys, kv.Key)
+		keys = append(keys, string(kv.Value))
 	}
 
 	return keys, nil
@@ -152,5 +152,6 @@ func (cp *ConsulClient) DelObj(key string) error {
 
 // GetLocalAddr gets local address of the host
 func (cp *ConsulClient) GetLocalAddr() (string, error) {
+	log.Panic("Calling unsupported API")
 	return "", nil
 }

consulService.go

@@ -44,9 +44,23 @@ func (cp *ConsulClient) RegisterService(serviceInfo ServiceInfo) error {
 
 	log.Infof("Registering service key: %s, value: %+v", keyName, serviceInfo)
 
-	// if there is a previously registered service, stop refreshing it
+	// if there is a previously registered service, no need to register it again..
 	if cp.serviceDb[keyName] != nil {
-		close(cp.serviceDb[keyName].stopChan)
+		srvState := cp.serviceDb[keyName]
+		if (srvState.ServiceName == serviceInfo.ServiceName) && (srvState.HostAddr == serviceInfo.HostAddr) &&
+			(srvState.Port == serviceInfo.Port) {
+			return nil
+		}
+
+		// stop and release the old key
+		close(srvState.stopChan)
+
+		// Delete the service instance
+		_, err := cp.client.KV().Delete(keyName, nil)
+		if err != nil {
+			log.Errorf("Error deleting key %s. Err: %v", keyName, err)
+			return err
+		}
 	}
 
 	// JSON format the object

etcdClient.go

@@ -221,6 +221,8 @@ func (ep *EtcdClient) GetLocalAddr() (string, error) {
 		Name string `json:"name"`
 	}
 
+	log.Panic("Calling unsupported API")
+
 	// Get ep state from etcd
 	if _, err := httpGetJSON("http://localhost:2379/v2/stats/self", &epData); err != nil {
 		log.Errorf("Error getting self state. Err: %v", err)

modeldb/modeldb.go

@@ -108,6 +108,7 @@ func ReadObj(objType, objKey string, retVal interface{}) error {
 	err := cdb.GetObj(key, retVal)
 	if err != nil {
 		log.Errorf("Error reading object: %s. Err: %v", key, err)
+		return err
 	}
 
 	return nil

objdb.go

@@ -78,9 +78,6 @@ type Plugin interface {
 
 // API Plugin API
 type API interface {
-	// Return local address used by conf store
-	GetLocalAddr() (string, error)
-
 	// Get a Key from conf store
 	GetObj(key string, retValue interface{}) error