Ftr: [refer dubbo 2.7.6] attachment type from map[string]stiring to m…

…ap[string]interface{} (apache#713)

* support attachment from map[sting]string to map[string]interface{} in invocation and result

cluster/cluster_impl/zone_aware_cluster_invoker_test.go

@@ -44,7 +44,7 @@ func TestZoneWareInvokerWithPreferredSuccess(t *testing.T) {
 	//defer ctrl.Finish()
 
 	mockResult := &protocol.RPCResult{
-		Attrs: map[string]string{constant.PREFERRED_KEY: "true"},
+		Attrs: map[string]interface{}{constant.PREFERRED_KEY: "true"},
 		Rest:  rest{tried: 0, success: true}}
 
 	var invokers []protocol.Invoker
@@ -99,15 +99,15 @@ func TestZoneWareInvokerWithWeightSuccess(t *testing.T) {
 			invoker.EXPECT().Invoke(gomock.Any()).DoAndReturn(
 				func(invocation protocol.Invocation) protocol.Result {
 					return &protocol.RPCResult{
-						Attrs: map[string]string{constant.WEIGHT_KEY: w1},
+						Attrs: map[string]interface{}{constant.WEIGHT_KEY: w1},
 						Rest:  rest{tried: 0, success: true}}
 				}).MaxTimes(100)
 		} else {
 			url.SetParam(constant.REGISTRY_KEY+"."+constant.WEIGHT_KEY, w2)
 			invoker.EXPECT().Invoke(gomock.Any()).DoAndReturn(
 				func(invocation protocol.Invocation) protocol.Result {
 					return &protocol.RPCResult{
-						Attrs: map[string]string{constant.WEIGHT_KEY: w2},
+						Attrs: map[string]interface{}{constant.WEIGHT_KEY: w2},
 						Rest:  rest{tried: 0, success: true}}
 				}).MaxTimes(100)
 		}
@@ -154,7 +154,7 @@ func TestZoneWareInvokerWithZoneSuccess(t *testing.T) {
 		invoker.EXPECT().Invoke(gomock.Any()).DoAndReturn(
 			func(invocation protocol.Invocation) protocol.Result {
 				return &protocol.RPCResult{
-					Attrs: map[string]string{constant.ZONE_KEY: zoneValue},
+					Attrs: map[string]interface{}{constant.ZONE_KEY: zoneValue},
 					Rest:  rest{tried: 0, success: true}}
 			})
 		invokers = append(invokers, invoker)

cluster/router/tag/tag_router.go

@@ -361,7 +361,9 @@ func isAnyHost(addr string) bool {
 func findTag(invocation protocol.Invocation, consumerUrl *common.URL) string {
 	tag, ok := invocation.Attachments()[constant.Tagkey]
 	if !ok {
-		tag = consumerUrl.GetParam(constant.Tagkey, "")
+		return consumerUrl.GetParam(constant.Tagkey, "")
+	} else if v, t := tag.(string); t {
+		return v
 	}
-	return tag
+	return ""
 }

common/proxy/proxy.go

@@ -145,12 +145,17 @@ func (p *Proxy) Implement(v common.RPCService) {
 				inv.SetAttachments(k, value)
 			}
 
-			// add user setAttachment
+			// add user setAttachment.  It is compatibility with previous versions.
 			atm := invCtx.Value(constant.AttachmentKey)
 			if m, ok := atm.(map[string]string); ok {
 				for k, value := range m {
 					inv.SetAttachments(k, value)
 				}
+			} else if m2, ok2 := atm.(map[string]interface{}); ok2 {
+				// it is support to transfer map[string]interface{}. It refers to dubbo-java 2.7.
+				for k, value := range m2 {
+					inv.SetAttachments(k, value)
+				}
 			}
 
 			result := p.invoke.Invoke(invCtx, inv)

common/proxy/proxy_test.go

@@ -24,6 +24,7 @@ import (
 )
 
 import (
+	"github.com/apache/dubbo-go/protocol/invocation"
 	perrors "github.com/pkg/errors"
 	"github.com/stretchr/testify/assert"
 )
@@ -32,6 +33,7 @@ import (
 	"github.com/apache/dubbo-go/common"
 	"github.com/apache/dubbo-go/common/constant"
 	"github.com/apache/dubbo-go/protocol"
+	"github.com/apache/dubbo-go/protocol/dubbo/hessian2"
 )
 
 type TestService struct {
@@ -40,6 +42,7 @@ type TestService struct {
 	MethodThree func(int, bool) (interface{}, error)
 	MethodFour  func(int, bool) (*interface{}, error) `dubbo:"methodFour"`
 	MethodFive  func() error
+	MethodSix   func(context.Context, string) (interface{}, error)
 	Echo        func(interface{}, *interface{}) error
 }
 
@@ -120,3 +123,34 @@ func TestProxyImplement(t *testing.T) {
 	assert.Nil(t, s3.MethodOne)
 
 }
+
+func TestProxyImplementForContext(t *testing.T) {
+	invoker := &TestProxyInvoker{
+		BaseInvoker: *protocol.NewBaseInvoker(common.URL{}),
+	}
+	p := NewProxy(invoker, nil, map[string]string{constant.ASYNC_KEY: "false"})
+	s := &TestService{}
+	p.Implement(s)
+	attahments1 := make(map[string]interface{}, 4)
+	attahments1["k1"] = "v1"
+	attahments1["k2"] = "v2"
+	context := context.WithValue(context.Background(), constant.AttachmentKey, attahments1)
+	r, err := p.Get().(*TestService).MethodSix(context, "xxx")
+	v1 := r.(map[string]interface{})
+	assert.NoError(t, err)
+	assert.Equal(t, v1["TestProxyInvoker"], "TestProxyInvokerValue")
+}
+
+type TestProxyInvoker struct {
+	protocol.BaseInvoker
+}
+
+func (bi *TestProxyInvoker) Invoke(context context.Context, inv protocol.Invocation) protocol.Result {
+	rpcInv := inv.(*invocation.RPCInvocation)
+	mapV := inv.Attachments()
+	mapV["TestProxyInvoker"] = "TestProxyInvokerValue"
+	hessian2.ReflectResponse(mapV, rpcInv.Reply())
+	return &protocol.RPCResult{
+		Rest: inv.Arguments(),
+	}
+}

filter/filter_impl/access_log_filter.go

@@ -105,13 +105,27 @@ func (ef *AccessLogFilter) logIntoChannel(accessLogData AccessLogData) {
 func (ef *AccessLogFilter) buildAccessLogData(_ protocol.Invoker, invocation protocol.Invocation) map[string]string {
 	dataMap := make(map[string]string, 16)
 	attachments := invocation.Attachments()
-	dataMap[constant.INTERFACE_KEY] = attachments[constant.INTERFACE_KEY]
-	dataMap[constant.METHOD_KEY] = invocation.MethodName()
-	dataMap[constant.VERSION_KEY] = attachments[constant.VERSION_KEY]
-	dataMap[constant.GROUP_KEY] = attachments[constant.GROUP_KEY]
-	dataMap[constant.TIMESTAMP_KEY] = time.Now().Format(MessageDateLayout)
-	dataMap[constant.LOCAL_ADDR], _ = attachments[constant.LOCAL_ADDR]
-	dataMap[constant.REMOTE_ADDR], _ = attachments[constant.REMOTE_ADDR]
+	if v, ok := attachments[constant.INTERFACE_KEY]; ok && v != nil {
+		dataMap[constant.INTERFACE_KEY] = v.(string)
+	}
+	if v, ok := attachments[constant.METHOD_KEY]; ok && v != nil {
+		dataMap[constant.METHOD_KEY] = v.(string)
+	}
+	if v, ok := attachments[constant.VERSION_KEY]; ok && v != nil {
+		dataMap[constant.VERSION_KEY] = v.(string)
+	}
+	if v, ok := attachments[constant.GROUP_KEY]; ok && v != nil {
+		dataMap[constant.GROUP_KEY] = v.(string)
+	}
+	if v, ok := attachments[constant.TIMESTAMP_KEY]; ok && v != nil {
+		dataMap[constant.TIMESTAMP_KEY] = v.(string)
+	}
+	if v, ok := attachments[constant.LOCAL_ADDR]; ok && v != nil {
+		dataMap[constant.LOCAL_ADDR] = v.(string)
+	}
+	if v, ok := attachments[constant.REMOTE_ADDR]; ok && v != nil {
+		dataMap[constant.REMOTE_ADDR] = v.(string)
+	}
 
 	if len(invocation.Arguments()) > 0 {
 		builder := strings.Builder{}

filter/filter_impl/access_log_filter_test.go

@@ -45,7 +45,7 @@ func TestAccessLogFilter_Invoke_Not_Config(t *testing.T) {
 			"service.filter=echo%2Ctoken%2Caccesslog&timestamp=1569153406&token=934804bf-b007-4174-94eb-96e3e1d60cc7&version=&warmup=100")
 	invoker := protocol.NewBaseInvoker(url)
 
-	attach := make(map[string]string, 10)
+	attach := make(map[string]interface{}, 10)
 	inv := invocation.NewRPCInvocation("MethodName", []interface{}{"OK", "Hello"}, attach)
 
 	accessLogFilter := GetAccessLogFilter()
@@ -64,7 +64,7 @@ func TestAccessLogFilterInvokeDefaultConfig(t *testing.T) {
 			"service.filter=echo%2Ctoken%2Caccesslog&timestamp=1569153406&token=934804bf-b007-4174-94eb-96e3e1d60cc7&version=&warmup=100")
 	invoker := protocol.NewBaseInvoker(url)
 
-	attach := make(map[string]string, 10)
+	attach := make(map[string]interface{}, 10)
 	attach[constant.VERSION_KEY] = "1.0"
 	attach[constant.GROUP_KEY] = "MyGroup"
 	inv := invocation.NewRPCInvocation("MethodName", []interface{}{"OK", "Hello"}, attach)

filter/filter_impl/active_filter_test.go

@@ -37,7 +37,7 @@ import (
 )
 
 func TestActiveFilterInvoke(t *testing.T) {
-	invoc := invocation.NewRPCInvocation("test", []interface{}{"OK"}, make(map[string]string, 0))
+	invoc := invocation.NewRPCInvocation("test", []interface{}{"OK"}, make(map[string]interface{}, 0))
 	url, _ := common.NewURL("dubbo://192.168.10.10:20000/com.ikurento.user.UserProvider")
 	filter := ActiveFilter{}
 	ctrl := gomock.NewController(t)
@@ -53,7 +53,7 @@ func TestActiveFilterInvoke(t *testing.T) {
 func TestActiveFilterOnResponse(t *testing.T) {
 	c := protocol.CurrentTimeMillis()
 	elapsed := 100
-	invoc := invocation.NewRPCInvocation("test", []interface{}{"OK"}, map[string]string{
+	invoc := invocation.NewRPCInvocation("test", []interface{}{"OK"}, map[string]interface{}{
 		dubboInvokeStartTime: strconv.FormatInt(c-int64(elapsed), 10),
 	})
 	url, _ := common.NewURL("dubbo://192.168.10.10:20000/com.ikurento.user.UserProvider")

filter/filter_impl/auth/default_authenticator_test.go

@@ -52,7 +52,7 @@ func TestDefaultAuthenticator_Authenticate(t *testing.T) {
 
 	var authenticator = &DefaultAuthenticator{}
 
-	invcation := invocation.NewRPCInvocation("test", parmas, map[string]string{
+	invcation := invocation.NewRPCInvocation("test", parmas, map[string]interface{}{
 		constant.REQUEST_SIGNATURE_KEY: signature,
 		constant.CONSUMER:              "test",
 		constant.REQUEST_TIMESTAMP_KEY: requestTime,
@@ -61,7 +61,7 @@ func TestDefaultAuthenticator_Authenticate(t *testing.T) {
 	err := authenticator.Authenticate(invcation, &testurl)
 	assert.Nil(t, err)
 	// modify the params
-	invcation = invocation.NewRPCInvocation("test", parmas[:1], map[string]string{
+	invcation = invocation.NewRPCInvocation("test", parmas[:1], map[string]interface{}{
 		constant.REQUEST_SIGNATURE_KEY: signature,
 		constant.CONSUMER:              "test",
 		constant.REQUEST_TIMESTAMP_KEY: requestTime,
@@ -119,7 +119,7 @@ func Test_getAccessKeyPairFailed(t *testing.T) {
 func Test_getSignatureWithinParams(t *testing.T) {
 	testurl, _ := common.NewURL("dubbo://127.0.0.1:20000/com.ikurento.user.UserProvider?interface=com.ikurento.user.UserProvider&group=gg&version=2.6.0")
 	testurl.SetParam(constant.PARAMTER_SIGNATURE_ENABLE_KEY, "true")
-	inv := invocation.NewRPCInvocation("test", []interface{}{"OK"}, map[string]string{
+	inv := invocation.NewRPCInvocation("test", []interface{}{"OK"}, map[string]interface{}{
 		"": "",
 	})
 	secret := "dubbo"

filter/filter_impl/auth/provider_auth_test.go

@@ -54,7 +54,7 @@ func TestProviderAuthFilter_Invoke(t *testing.T) {
 	requestTime := strconv.Itoa(int(time.Now().Unix() * 1000))
 	signature, _ := getSignature(&url, inv, secret, requestTime)
 
-	inv = invocation.NewRPCInvocation("test", []interface{}{"OK"}, map[string]string{
+	inv = invocation.NewRPCInvocation("test", []interface{}{"OK"}, map[string]interface{}{
 		constant.REQUEST_SIGNATURE_KEY: signature,
 		constant.CONSUMER:              "test",
 		constant.REQUEST_TIMESTAMP_KEY: requestTime,

filter/filter_impl/execute_limit_filter_test.go

@@ -36,7 +36,7 @@ import (
 
 func TestExecuteLimitFilterInvokeIgnored(t *testing.T) {
 	methodName := "hello"
-	invoc := invocation.NewRPCInvocation(methodName, []interface{}{"OK"}, make(map[string]string, 0))
+	invoc := invocation.NewRPCInvocation(methodName, []interface{}{"OK"}, make(map[string]interface{}, 0))
 
 	invokeUrl := common.NewURLWithOptions(
 		common.WithParams(url.Values{}),
@@ -51,7 +51,7 @@ func TestExecuteLimitFilterInvokeIgnored(t *testing.T) {
 
 func TestExecuteLimitFilterInvokeConfigureError(t *testing.T) {
 	methodName := "hello1"
-	invoc := invocation.NewRPCInvocation(methodName, []interface{}{"OK"}, make(map[string]string, 0))
+	invoc := invocation.NewRPCInvocation(methodName, []interface{}{"OK"}, make(map[string]interface{}, 0))
 
 	invokeUrl := common.NewURLWithOptions(
 		common.WithParams(url.Values{}),
@@ -68,7 +68,7 @@ func TestExecuteLimitFilterInvokeConfigureError(t *testing.T) {
 
 func TestExecuteLimitFilterInvoke(t *testing.T) {
 	methodName := "hello1"
-	invoc := invocation.NewRPCInvocation(methodName, []interface{}{"OK"}, make(map[string]string, 0))
+	invoc := invocation.NewRPCInvocation(methodName, []interface{}{"OK"}, make(map[string]interface{}, 0))
 
 	invokeUrl := common.NewURLWithOptions(
 		common.WithParams(url.Values{}),

filter/filter_impl/graceful_shutdown_filter_test.go

@@ -39,7 +39,7 @@ import (
 )
 
 func TestGenericFilterInvoke(t *testing.T) {
-	invoc := invocation.NewRPCInvocation("GetUser", []interface{}{"OK"}, make(map[string]string, 0))
+	invoc := invocation.NewRPCInvocation("GetUser", []interface{}{"OK"}, make(map[string]interface{}, 0))
 
 	invokeUrl := common.NewURLWithOptions(
 		common.WithParams(url.Values{}))

filter/filter_impl/metrics_filter_test.go

@@ -57,7 +57,7 @@ func TestMetricsFilterInvoke(t *testing.T) {
 			"service.filter=echo%2Ctoken%2Caccesslog&timestamp=1569153406&token=934804bf-b007-4174-94eb-96e3e1d60cc7&version=&warmup=100")
 	invoker := protocol.NewBaseInvoker(url)
 
-	attach := make(map[string]string, 10)
+	attach := make(map[string]interface{}, 10)
 	inv := invocation.NewRPCInvocation("MethodName", []interface{}{"OK", "Hello"}, attach)
 
 	ctx := context.Background()

filter/filter_impl/seata_filter_test.go

@@ -48,8 +48,9 @@ func (iv *testMockSeataInvoker) Invoke(ctx context.Context, _ protocol.Invocatio
 
 func TestSeataFilter_Invoke(t *testing.T) {
 	filter := getSeataFilter()
-	result := filter.Invoke(context.Background(), &testMockSeataInvoker{}, invocation.NewRPCInvocation("$echo", []interface{}{"OK"}, map[string]string{
-		SEATA_XID: "10.30.21.227:8091:2000047792",
-	}))
+	result := filter.Invoke(context.Background(), &testMockSeataInvoker{}, invocation.NewRPCInvocation("$echo",
+		[]interface{}{"OK"}, map[string]interface{}{
+			SEATA_XID: "10.30.21.227:8091:2000047792",
+		}))
 	assert.Equal(t, "10.30.21.227:8091:2000047792", result.Result())
 }

filter/filter_impl/token_filter.go

@@ -51,7 +51,7 @@ func (tf *TokenFilter) Invoke(ctx context.Context, invoker protocol.Invoker, inv
 	if len(invokerTkn) > 0 {
 		attachs := invocation.Attachments()
 		remoteTkn, exist := attachs[constant.TOKEN_KEY]
-		if exist && strings.EqualFold(invokerTkn, remoteTkn) {
+		if exist && remoteTkn != nil && strings.EqualFold(invokerTkn, remoteTkn.(string)) {
 			return invoker.Invoke(ctx, invocation)
 		}
 		return &protocol.RPCResult{Err: perrors.Errorf("Invalid token! Forbid invoke remote service %v method %s ",

filter/filter_impl/token_filter_test.go

@@ -40,7 +40,7 @@ func TestTokenFilterInvoke(t *testing.T) {
 	url := common.NewURLWithOptions(
 		common.WithParams(url.Values{}),
 		common.WithParamsValue(constant.TOKEN_KEY, "ori_key"))
-	attch := make(map[string]string, 0)
+	attch := make(map[string]interface{}, 0)
 	attch[constant.TOKEN_KEY] = "ori_key"
 	result := filter.Invoke(context.Background(),
 		protocol.NewBaseInvoker(*url),
@@ -54,7 +54,7 @@ func TestTokenFilterInvokeEmptyToken(t *testing.T) {
 	filter := GetTokenFilter()
 
 	testUrl := common.URL{}
-	attch := make(map[string]string, 0)
+	attch := make(map[string]interface{}, 0)
 	attch[constant.TOKEN_KEY] = "ori_key"
 	result := filter.Invoke(context.Background(), protocol.NewBaseInvoker(testUrl), invocation.NewRPCInvocation("MethodName", []interface{}{"OK"}, attch))
 	assert.Nil(t, result.Error())
@@ -67,7 +67,7 @@ func TestTokenFilterInvokeEmptyAttach(t *testing.T) {
 	testUrl := common.NewURLWithOptions(
 		common.WithParams(url.Values{}),
 		common.WithParamsValue(constant.TOKEN_KEY, "ori_key"))
-	attch := make(map[string]string, 0)
+	attch := make(map[string]interface{}, 0)
 	result := filter.Invoke(context.Background(), protocol.NewBaseInvoker(*testUrl), invocation.NewRPCInvocation("MethodName", []interface{}{"OK"}, attch))
 	assert.NotNil(t, result.Error())
 }
@@ -78,7 +78,7 @@ func TestTokenFilterInvokeNotEqual(t *testing.T) {
 	testUrl := common.NewURLWithOptions(
 		common.WithParams(url.Values{}),
 		common.WithParamsValue(constant.TOKEN_KEY, "ori_key"))
-	attch := make(map[string]string, 0)
+	attch := make(map[string]interface{}, 0)
 	attch[constant.TOKEN_KEY] = "err_key"
 	result := filter.Invoke(context.Background(),
 		protocol.NewBaseInvoker(*testUrl), invocation.NewRPCInvocation("MethodName", []interface{}{"OK"}, attch))

filter/filter_impl/tps/tps_limiter_method_service_test.go

@@ -36,7 +36,7 @@ import (
 
 func TestMethodServiceTpsLimiterImplIsAllowableOnlyServiceLevel(t *testing.T) {
 	methodName := "hello"
-	invoc := invocation.NewRPCInvocation(methodName, []interface{}{"OK"}, make(map[string]string, 0))
+	invoc := invocation.NewRPCInvocation(methodName, []interface{}{"OK"}, make(map[string]interface{}, 0))
 
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
@@ -63,7 +63,7 @@ func TestMethodServiceTpsLimiterImplIsAllowableOnlyServiceLevel(t *testing.T) {
 
 func TestMethodServiceTpsLimiterImplIsAllowableNoConfig(t *testing.T) {
 	methodName := "hello1"
-	invoc := invocation.NewRPCInvocation(methodName, []interface{}{"OK"}, make(map[string]string, 0))
+	invoc := invocation.NewRPCInvocation(methodName, []interface{}{"OK"}, make(map[string]interface{}, 0))
 	// ctrl := gomock.NewController(t)
 	// defer ctrl.Finish()
 
@@ -80,7 +80,7 @@ func TestMethodServiceTpsLimiterImplIsAllowableNoConfig(t *testing.T) {
 func TestMethodServiceTpsLimiterImplIsAllowableMethodLevelOverride(t *testing.T) {
 	methodName := "hello2"
 	methodConfigPrefix := "methods." + methodName + "."
-	invoc := invocation.NewRPCInvocation(methodName, []interface{}{"OK"}, make(map[string]string, 0))
+	invoc := invocation.NewRPCInvocation(methodName, []interface{}{"OK"}, make(map[string]interface{}, 0))
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
 
@@ -113,7 +113,7 @@ func TestMethodServiceTpsLimiterImplIsAllowableMethodLevelOverride(t *testing.T)
 func TestMethodServiceTpsLimiterImplIsAllowableBothMethodAndService(t *testing.T) {
 	methodName := "hello3"
 	methodConfigPrefix := "methods." + methodName + "."
-	invoc := invocation.NewRPCInvocation(methodName, []interface{}{"OK"}, make(map[string]string, 0))
+	invoc := invocation.NewRPCInvocation(methodName, []interface{}{"OK"}, make(map[string]interface{}, 0))
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
 

filter/filter_impl/tps_limit_filter_test.go

@@ -44,7 +44,7 @@ func TestTpsLimitFilterInvokeWithNoTpsLimiter(t *testing.T) {
 	invokeUrl := common.NewURLWithOptions(
 		common.WithParams(url.Values{}),
 		common.WithParamsValue(constant.TPS_LIMITER_KEY, ""))
-	attch := make(map[string]string, 0)
+	attch := make(map[string]interface{}, 0)
 
 	result := tpsFilter.Invoke(context.Background(),
 		protocol.NewBaseInvoker(*invokeUrl),
@@ -68,7 +68,7 @@ func TestGenericFilterInvokeWithDefaultTpsLimiter(t *testing.T) {
 	invokeUrl := common.NewURLWithOptions(
 		common.WithParams(url.Values{}),
 		common.WithParamsValue(constant.TPS_LIMITER_KEY, constant.DEFAULT_KEY))
-	attch := make(map[string]string, 0)
+	attch := make(map[string]interface{}, 0)
 
 	result := tpsFilter.Invoke(context.Background(),
 		protocol.NewBaseInvoker(*invokeUrl),
@@ -99,7 +99,7 @@ func TestGenericFilterInvokeWithDefaultTpsLimiterNotAllow(t *testing.T) {
 	invokeUrl := common.NewURLWithOptions(
 		common.WithParams(url.Values{}),
 		common.WithParamsValue(constant.TPS_LIMITER_KEY, constant.DEFAULT_KEY))
-	attch := make(map[string]string, 0)
+	attch := make(map[string]interface{}, 0)
 
 	result := tpsFilter.Invoke(context.Background(),
 		protocol.NewBaseInvoker(*invokeUrl), invocation.NewRPCInvocation("MethodName", []interface{}{"OK"}, attch))