Merge e345a5e into 679ce10

pkg/common/common.go

@@ -17,6 +17,8 @@
 
 package common
 
+import "time"
+
 //match mode
 const (
 	QueryParamQ      = "q"
@@ -43,3 +45,17 @@ const (
 	ContentTypeJSON = "application/json"
 	ContentTypeYaml = "text/yaml"
 )
+
+//const for server/resource/v1
+const (
+	PatternExact            = "exact"
+	MsgDomainMustNotBeEmpty = "domain must not be empty"
+	MsgIllegalLabels        = "label value can not be empty, " +
+		"label can not be duplicated, please check query parameters"
+	MsgIllegalDepth     = "X-Depth must be number"
+	MsgInvalidWait      = "wait param should be formed with number and time unit like 5s,100ms, and less than 5m"
+	MsgInvalidRev       = "revision param should be formed with number greater than 0"
+	ErrKvIDMustNotEmpty = "must supply kv id if you want to remove key"
+
+	MaxWait = 5 * time.Minute
+)

server/pubsub/struct.go

@@ -20,6 +20,8 @@ package pubsub
 import (
 	"encoding/json"
 	"errors"
+	"github.com/apache/servicecomb-kie/pkg/common"
+	"reflect"
 	"strings"
 )
 
@@ -46,6 +48,7 @@ type Topic struct {
 	LabelsFormat string            `json:"labels,omitempty"`
 	DomainID     string            `json:"domainID,omitempty"`
 	Project      string            `json:"project,omitempty"`
+	MatchType    string            `json:"match,omitempty"`
 }
 
 //ParseTopicString parse topic string to topic struct
@@ -78,6 +81,11 @@ func (t *Topic) Match(event *KVChangeEvent) bool {
 			match = true
 		}
 	}
+	if t.MatchType == common.PatternExact {
+		if !reflect.DeepEqual(t.Labels, event.Labels) {
+			return false
+		}
+	}
 	for k, v := range t.Labels {
 		if event.Labels[k] != v {
 			return false

server/resource/v1/common.go

@@ -37,22 +37,10 @@ import (
 )
 
 //const of server
-const (
-	PatternExact            = "exact"
-	MsgDomainMustNotBeEmpty = "domain must not be empty"
-	MsgIllegalLabels        = "label value can not be empty, " +
-		"label can not be duplicated, please check query parameters"
-	MsgIllegalDepth     = "X-Depth must be number"
-	MsgInvalidWait      = "wait param should be formed with number and time unit like 5s,100ms, and less than 5m"
-	MsgInvalidRev       = "revision param should be formed with number greater than 0"
-	ErrKvIDMustNotEmpty = "must supply kv id if you want to remove key"
-
-	MaxWait = 5 * time.Minute
-)
 
 //err
 var (
-	ErrInvalidRev = errors.New(MsgInvalidRev)
+	ErrInvalidRev = errors.New(common.MsgInvalidRev)
 )
 
 //ReadDomain get domain info from attribute
@@ -141,7 +129,7 @@ func getLabels(rctx *restful.Context) (map[string]string, error) {
 	for _, v := range labelSlice {
 		v := strings.Split(v, ":")
 		if len(v) != 2 {
-			return nil, errors.New(MsgIllegalLabels)
+			return nil, errors.New(common.MsgIllegalLabels)
 		}
 		labels[v[0]] = v[1]
 	}
@@ -163,15 +151,15 @@ func isRevised(ctx context.Context, revStr, domain string) (bool, error) {
 }
 func getMatchPattern(rctx *restful.Context) string {
 	m := rctx.ReadQueryParameter(common.QueryParamMatch)
-	if m != "" && m != PatternExact {
+	if m != "" && m != common.PatternExact {
 		return ""
 	}
 	return m
 }
 func eventHappened(rctx *restful.Context, waitStr string, topic *pubsub.Topic) (bool, error) {
 	d, err := time.ParseDuration(waitStr)
-	if err != nil || d > MaxWait {
-		return false, errors.New(MsgInvalidWait)
+	if err != nil || d > common.MaxWait {
+		return false, errors.New(common.MsgInvalidWait)
 	}
 	happened := true
 	o := &pubsub.Observer{
@@ -221,7 +209,7 @@ func queryAndResponse(rctx *restful.Context,
 		service.WithLimit(limit),
 		service.WithOffset(offset),
 	}
-	if m == PatternExact {
+	if m == common.PatternExact {
 		opts = append(opts, service.WithExactLabels())
 	}
 	kv, err := service.KVService.List(rctx.Ctx, domain.(string), project, opts...)

server/resource/v1/kv_resource.go

@@ -46,7 +46,7 @@ func (r *KVResource) Put(context *restful.Context) {
 	}
 	domain := ReadDomain(context)
 	if domain == nil {
-		WriteErrResponse(context, http.StatusInternalServerError, MsgDomainMustNotBeEmpty, common.ContentTypeText)
+		WriteErrResponse(context, http.StatusInternalServerError, common.MsgDomainMustNotBeEmpty, common.ContentTypeText)
 		return
 	}
 	kv.Key = key
@@ -88,12 +88,12 @@ func (r *KVResource) GetByKey(rctx *restful.Context) {
 	project := rctx.ReadPathParameter("project")
 	labels, err := getLabels(rctx)
 	if err != nil {
-		WriteErrResponse(rctx, http.StatusBadRequest, MsgIllegalLabels, common.ContentTypeText)
+		WriteErrResponse(rctx, http.StatusBadRequest, common.MsgIllegalLabels, common.ContentTypeText)
 		return
 	}
 	domain := ReadDomain(rctx)
 	if domain == nil {
-		WriteErrResponse(rctx, http.StatusInternalServerError, MsgDomainMustNotBeEmpty, common.ContentTypeText)
+		WriteErrResponse(rctx, http.StatusInternalServerError, common.MsgDomainMustNotBeEmpty, common.ContentTypeText)
 		return
 	}
 	limitStr := rctx.ReadQueryParameter("limit")
@@ -112,7 +112,7 @@ func (r *KVResource) List(rctx *restful.Context) {
 	project := rctx.ReadPathParameter("project")
 	domain := ReadDomain(rctx)
 	if domain == nil {
-		WriteErrResponse(rctx, http.StatusInternalServerError, MsgDomainMustNotBeEmpty, common.ContentTypeText)
+		WriteErrResponse(rctx, http.StatusInternalServerError, common.MsgDomainMustNotBeEmpty, common.ContentTypeText)
 		return
 	}
 	labels, err := getLabels(rctx)
@@ -139,9 +139,10 @@ func returnData(rctx *restful.Context, domain interface{}, project string, label
 			return
 		}
 		changed, err := eventHappened(rctx, wait, &pubsub.Topic{
-			Labels:   labels,
-			Project:  project,
-			DomainID: domain.(string),
+			Labels:    labels,
+			Project:   project,
+			MatchType: getMatchPattern(rctx),
+			DomainID:  domain.(string),
 		})
 		if err != nil {
 			WriteErrResponse(rctx, http.StatusBadRequest, err.Error(), common.ContentTypeText)
@@ -167,9 +168,10 @@ func returnData(rctx *restful.Context, domain interface{}, project string, label
 			return
 		} else if wait != "" {
 			changed, err := eventHappened(rctx, wait, &pubsub.Topic{
-				Labels:   labels,
-				Project:  project,
-				DomainID: domain.(string),
+				Labels:    labels,
+				Project:   project,
+				MatchType: getMatchPattern(rctx),
+				DomainID:  domain.(string),
 			})
 			if err != nil {
 				WriteErrResponse(rctx, http.StatusBadRequest, err.Error(), common.ContentTypeText)
@@ -198,7 +200,7 @@ func (r *KVResource) Search(context *restful.Context) {
 	project := context.ReadPathParameter("project")
 	domain := ReadDomain(context)
 	if domain == nil {
-		WriteErrResponse(context, http.StatusInternalServerError, MsgDomainMustNotBeEmpty, common.ContentTypeText)
+		WriteErrResponse(context, http.StatusInternalServerError, common.MsgDomainMustNotBeEmpty, common.ContentTypeText)
 		return
 	}
 	var kvs []*model.KVResponse
@@ -261,12 +263,12 @@ func (r *KVResource) Delete(context *restful.Context) {
 	project := context.ReadPathParameter("project")
 	domain := ReadDomain(context)
 	if domain == nil {
-		WriteErrResponse(context, http.StatusInternalServerError, MsgDomainMustNotBeEmpty, common.ContentTypeText)
+		WriteErrResponse(context, http.StatusInternalServerError, common.MsgDomainMustNotBeEmpty, common.ContentTypeText)
 		return
 	}
 	kvID := context.ReadQueryParameter(common.QueryParamKeyID)
 	if kvID == "" {
-		WriteErrResponse(context, http.StatusBadRequest, ErrKvIDMustNotEmpty, common.ContentTypeText)
+		WriteErrResponse(context, http.StatusBadRequest, common.ErrKvIDMustNotEmpty, common.ContentTypeText)
 		return
 	}
 	err := service.KVService.Delete(context.Ctx, kvID, domain.(string), project)

server/resource/v1/kv_resource_test.go

@@ -36,6 +36,7 @@ import (
 	"io/ioutil"
 	"net/http"
 	"net/http/httptest"
+	"sync"
 	"testing"
 	"time"
 
@@ -206,7 +207,7 @@ func TestKVResource_List(t *testing.T) {
 		t.Log(duration)
 		assert.Equal(t, http.StatusNotModified, resp.Result().StatusCode)
 	})
-	t.Run("list kv by service label, with wait param,will too 1s and return 304", func(t *testing.T) {
+	t.Run("list kv by service label, with wait param,will exceed 1s and return 304", func(t *testing.T) {
 		r, _ := http.NewRequest("GET", "/v1/test/kie/kv?label=service:utService&wait=1s", nil)
 		noopH := &handler2.NoopAuthHandler{}
 		chain, _ := handler.CreateChain(common.Provider, "testchain1", noopH.Name())
@@ -254,6 +255,47 @@ func TestKVResource_List(t *testing.T) {
 		assert.NoError(t, err)
 		assert.Equal(t, 1, len(result.Data))
 	})
+	t.Run("list kv by service label, with wait and match param,not exact match and return 304", func(t *testing.T) {
+		r, _ := http.NewRequest("GET", "/v1/test/kie/kv?label=match:test&wait=10s&match=exact", nil)
+		noopH := &handler2.NoopAuthHandler{}
+		chain, _ := handler.CreateChain(common.Provider, "testchain-match", noopH.Name())
+		r.Header.Set("Content-Type", "application/json")
+		kvr := &v1.KVResource{}
+		c, err := restfultest.New(kvr, chain)
+		assert.NoError(t, err)
+		resp := httptest.NewRecorder()
+		var wg sync.WaitGroup
+		wg.Add(1)
+		go func() {
+			kv := &model.KVDoc{
+				Value:  "val",
+				Labels: map[string]string{"dummy": "test", "match": "test"},
+			}
+			j, _ := json.Marshal(kv)
+			r2, _ := http.NewRequest("PUT", "/v1/test/kie/kv/testKey", bytes.NewBuffer(j))
+			noopH2 := &handler2.NoopAuthHandler{}
+			chain2, _ := handler.CreateChain(common.Provider, "testchain-match", noopH2.Name())
+			r2.Header.Set("Content-Type", "application/json")
+			kvr2 := &v1.KVResource{}
+			c2, _ := restfultest.New(kvr2, chain2)
+			resp2 := httptest.NewRecorder()
+			c2.ServeHTTP(resp2, r2)
+			body, _ := ioutil.ReadAll(resp2.Body)
+			data := &model.KVDoc{}
+			err = json.Unmarshal(body, data)
+			assert.NotEmpty(t, data.ID)
+			wg.Done()
+		}()
+		start := time.Now()
+		c.ServeHTTP(resp, r)
+		wg.Wait()
+		duration := time.Since(start)
+		body, _ := ioutil.ReadAll(resp.Body)
+		data := &model.KVDoc{}
+		err = json.Unmarshal(body, data)
+		assert.Equal(t, 304, resp.Code)
+		t.Log(duration)
+	})
 }
 func TestKVResource_GetByKey(t *testing.T) {
 	t.Run("get one key by label, exact match,should return 1 kv", func(t *testing.T) {

server/resource/v1/label_resouce.go

@@ -26,7 +26,7 @@ func (r *LabelResource) PutLabel(context *restful.Context) {
 	entity.Project = context.ReadPathParameter("project")
 	domain := ReadDomain(context)
 	if domain == nil {
-		WriteErrResponse(context, http.StatusInternalServerError, MsgDomainMustNotBeEmpty, common.ContentTypeText)
+		WriteErrResponse(context, http.StatusInternalServerError, common.MsgDomainMustNotBeEmpty, common.ContentTypeText)
 		return
 	}
 	entity.Domain = domain.(string)