From ddbead73a134dded9c6501fe790f193e9bf6b75e Mon Sep 17 00:00:00 2001 From: cvictory Date: Mon, 7 Sep 2020 10:54:46 +0800 Subject: [PATCH] Ftr: [refer dubbo 2.7.6] attachment type from map[string]stiring to map[string]interface{} (#713) * support attachment from map[sting]string to map[string]interface{} in invocation and result --- .../zone_aware_cluster_invoker_test.go | 8 +- cluster/router/tag/tag_router.go | 6 +- common/proxy/proxy.go | 7 +- common/proxy/proxy_test.go | 34 ++ filter/filter_impl/access_log_filter.go | 28 +- filter/filter_impl/access_log_filter_test.go | 4 +- filter/filter_impl/active_filter_test.go | 4 +- .../auth/default_authenticator_test.go | 6 +- filter/filter_impl/auth/provider_auth_test.go | 2 +- .../filter_impl/execute_limit_filter_test.go | 6 +- .../graceful_shutdown_filter_test.go | 2 +- filter/filter_impl/metrics_filter_test.go | 2 +- filter/filter_impl/seata_filter_test.go | 7 +- filter/filter_impl/token_filter.go | 2 +- filter/filter_impl/token_filter_test.go | 8 +- .../tps/tps_limiter_method_service_test.go | 8 +- filter/filter_impl/tps_limit_filter_test.go | 6 +- filter/filter_impl/tracing_filter_test.go | 2 +- go.mod | 2 +- go.sum | 10 +- metadata/service/inmemory/service_proxy.go | 2 +- metrics/prometheus/reporter_test.go | 2 +- protocol/dubbo/client.go | 20 +- protocol/dubbo/client_test.go | 32 +- protocol/dubbo/codec.go | 21 +- protocol/dubbo/codec_test.go | 17 +- protocol/dubbo/dubbo_invoker.go | 3 +- protocol/dubbo/dubbo_invoker_test.go | 2 +- protocol/dubbo/hessian2/const.go | 243 +++++++++++ protocol/dubbo/hessian2/hessian_dubbo.go | 251 ++++++++++++ protocol/dubbo/hessian2/hessian_dubbo_test.go | 231 +++++++++++ protocol/dubbo/hessian2/hessian_request.go | 350 ++++++++++++++++ .../dubbo/hessian2/hessian_request_test.go | 158 ++++++++ protocol/dubbo/hessian2/hessian_response.go | 377 ++++++++++++++++++ .../dubbo/hessian2/hessian_response_test.go | 225 +++++++++++ protocol/dubbo/listener.go | 48 ++- protocol/dubbo/listener_test.go | 5 +- protocol/dubbo/opentracing.go | 60 +++ protocol/dubbo/readwriter.go | 34 +- protocol/invocation.go | 7 +- protocol/invocation/rpcinvocation.go | 28 +- protocol/jsonrpc/server.go | 2 +- protocol/rest/server/rest_server.go | 2 +- protocol/result.go | 19 +- test/integrate/dubbo/go-client/go.mod | 4 + test/integrate/dubbo/go-server/go.mod | 4 + 46 files changed, 2151 insertions(+), 150 deletions(-) create mode 100644 protocol/dubbo/hessian2/const.go create mode 100644 protocol/dubbo/hessian2/hessian_dubbo.go create mode 100644 protocol/dubbo/hessian2/hessian_dubbo_test.go create mode 100644 protocol/dubbo/hessian2/hessian_request.go create mode 100644 protocol/dubbo/hessian2/hessian_request_test.go create mode 100644 protocol/dubbo/hessian2/hessian_response.go create mode 100644 protocol/dubbo/hessian2/hessian_response_test.go create mode 100644 protocol/dubbo/opentracing.go diff --git a/cluster/cluster_impl/zone_aware_cluster_invoker_test.go b/cluster/cluster_impl/zone_aware_cluster_invoker_test.go index cd201a42c7..7f77f33166 100644 --- a/cluster/cluster_impl/zone_aware_cluster_invoker_test.go +++ b/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,7 +99,7 @@ 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 { @@ -107,7 +107,7 @@ 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: 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) diff --git a/cluster/router/tag/tag_router.go b/cluster/router/tag/tag_router.go index 12e6eda3e7..9f92e83b21 100644 --- a/cluster/router/tag/tag_router.go +++ b/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 "" } diff --git a/common/proxy/proxy.go b/common/proxy/proxy.go index ce0f4d1d3f..d51ce1cc1b 100644 --- a/common/proxy/proxy.go +++ b/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) diff --git a/common/proxy/proxy_test.go b/common/proxy/proxy_test.go index 14b2befbc4..c6066157fd 100644 --- a/common/proxy/proxy_test.go +++ b/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(), + } +} diff --git a/filter/filter_impl/access_log_filter.go b/filter/filter_impl/access_log_filter.go index 621012c24c..6eaf9cb00b 100644 --- a/filter/filter_impl/access_log_filter.go +++ b/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{} diff --git a/filter/filter_impl/access_log_filter_test.go b/filter/filter_impl/access_log_filter_test.go index 55c328cc30..a3a6151aa1 100644 --- a/filter/filter_impl/access_log_filter_test.go +++ b/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×tamp=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×tamp=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) diff --git a/filter/filter_impl/active_filter_test.go b/filter/filter_impl/active_filter_test.go index 6b72830e6a..9837a49c72 100644 --- a/filter/filter_impl/active_filter_test.go +++ b/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") diff --git a/filter/filter_impl/auth/default_authenticator_test.go b/filter/filter_impl/auth/default_authenticator_test.go index 5b107b5960..8b0fb6b0e3 100644 --- a/filter/filter_impl/auth/default_authenticator_test.go +++ b/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" diff --git a/filter/filter_impl/auth/provider_auth_test.go b/filter/filter_impl/auth/provider_auth_test.go index 626782ae83..f6ebfcd7ab 100644 --- a/filter/filter_impl/auth/provider_auth_test.go +++ b/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, diff --git a/filter/filter_impl/execute_limit_filter_test.go b/filter/filter_impl/execute_limit_filter_test.go index d36d6edef1..953f5e1cc8 100644 --- a/filter/filter_impl/execute_limit_filter_test.go +++ b/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{}), diff --git a/filter/filter_impl/graceful_shutdown_filter_test.go b/filter/filter_impl/graceful_shutdown_filter_test.go index 87ac2eac61..220ef6f208 100644 --- a/filter/filter_impl/graceful_shutdown_filter_test.go +++ b/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{})) diff --git a/filter/filter_impl/metrics_filter_test.go b/filter/filter_impl/metrics_filter_test.go index 881106f4bc..ac10d52cf3 100644 --- a/filter/filter_impl/metrics_filter_test.go +++ b/filter/filter_impl/metrics_filter_test.go @@ -57,7 +57,7 @@ func TestMetricsFilterInvoke(t *testing.T) { "service.filter=echo%2Ctoken%2Caccesslog×tamp=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() diff --git a/filter/filter_impl/seata_filter_test.go b/filter/filter_impl/seata_filter_test.go index 6c39897f7a..45817e95cb 100644 --- a/filter/filter_impl/seata_filter_test.go +++ b/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()) } diff --git a/filter/filter_impl/token_filter.go b/filter/filter_impl/token_filter.go index fe4e38747e..b5e05605c2 100644 --- a/filter/filter_impl/token_filter.go +++ b/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 ", diff --git a/filter/filter_impl/token_filter_test.go b/filter/filter_impl/token_filter_test.go index c2f69bd039..cd1bba3d4a 100644 --- a/filter/filter_impl/token_filter_test.go +++ b/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)) diff --git a/filter/filter_impl/tps/tps_limiter_method_service_test.go b/filter/filter_impl/tps/tps_limiter_method_service_test.go index edae99ec2d..61f28e442f 100644 --- a/filter/filter_impl/tps/tps_limiter_method_service_test.go +++ b/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() diff --git a/filter/filter_impl/tps_limit_filter_test.go b/filter/filter_impl/tps_limit_filter_test.go index 274e4e6de6..da0fc482ce 100644 --- a/filter/filter_impl/tps_limit_filter_test.go +++ b/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)) diff --git a/filter/filter_impl/tracing_filter_test.go b/filter/filter_impl/tracing_filter_test.go index 57f4095d49..e159b7400d 100644 --- a/filter/filter_impl/tracing_filter_test.go +++ b/filter/filter_impl/tracing_filter_test.go @@ -42,7 +42,7 @@ func TestTracingFilterInvoke(t *testing.T) { "service.filter=echo%2Ctoken%2Caccesslog×tamp=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() tf := newTracingFilter() diff --git a/go.mod b/go.mod index a26aec144e..9e18cc276a 100644 --- a/go.mod +++ b/go.mod @@ -7,7 +7,7 @@ require ( github.com/Workiva/go-datastructures v1.0.50 github.com/afex/hystrix-go v0.0.0-20180502004556-fa1af6a1f4f5 github.com/apache/dubbo-getty v1.3.10 - github.com/apache/dubbo-go-hessian2 v1.6.2 + github.com/apache/dubbo-go-hessian2 v1.6.0-rc1.0.20200906044240-6c1fb5c3bd44 github.com/coreos/bbolt v1.3.3 // indirect github.com/coreos/etcd v3.3.13+incompatible github.com/coreos/go-systemd v0.0.0-20190719114852-fd7a80b32e1f // indirect diff --git a/go.sum b/go.sum index c5ab85db14..f744fc313b 100644 --- a/go.sum +++ b/go.sum @@ -74,8 +74,8 @@ github.com/aliyun/alibaba-cloud-sdk-go v1.61.18 h1:zOVTBdCKFd9JbCKz9/nt+FovbjPFm github.com/aliyun/alibaba-cloud-sdk-go v1.61.18/go.mod h1:v8ESoHo4SyHmuB4b1tJqDHxfTGEciD+yhvOU/5s1Rfk= github.com/apache/dubbo-getty v1.3.10 h1:ys5mwjPdxG/KwkPjS6EI0RzQtU6p6FCPoKpaFEzpAL0= github.com/apache/dubbo-getty v1.3.10/go.mod h1:x6rraK01BL5C7jUM2fPl5KMkAxLVIx54ZB8/XEOik9Y= -github.com/apache/dubbo-go-hessian2 v1.6.2 h1:i7F5GjVaUatLQz1x9vUmmSIFj49L8J6rVICdF6xw4qw= -github.com/apache/dubbo-go-hessian2 v1.6.2/go.mod h1:7rEw9guWABQa6Aqb8HeZcsYPHsOS7XT1qtJvkmI6c5w= +github.com/apache/dubbo-go-hessian2 v1.6.0-rc1.0.20200906044240-6c1fb5c3bd44 h1:9biQu3Z0PjDN1m8h6poo76dFkvaIpfryUVpJ5VsYVrM= +github.com/apache/dubbo-go-hessian2 v1.6.0-rc1.0.20200906044240-6c1fb5c3bd44/go.mod h1:7rEw9guWABQa6Aqb8HeZcsYPHsOS7XT1qtJvkmI6c5w= github.com/armon/circbuf v0.0.0-20150827004946-bbbad097214e h1:QEF07wC0T1rKkctt1RINW/+RMTVmiwxETico2l3gxJA= github.com/armon/circbuf v0.0.0-20150827004946-bbbad097214e/go.mod h1:3U/XgcO3hCbHZ8TKRvWD2dDTCfh9M9ya+I9JpbB7O8o= github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8= @@ -150,7 +150,6 @@ github.com/docker/go-connections v0.4.0/go.mod h1:Gbd7IOopHjR8Iph03tsViu4nIes5Xh github.com/docker/spdystream v0.0.0-20160310174837-449fdfce4d96/go.mod h1:Qh8CwZgvJUkLughtfhJv5dyTYa91l1fOUCrgjqmcifM= github.com/dubbogo/go-zookeeper v1.0.1 h1:irLzvOsDOTNsN8Sv9tvYYxVu6DCQfLtziZQtUHmZgz8= github.com/dubbogo/go-zookeeper v1.0.1/go.mod h1:fn6n2CAEer3novYgk9ULLwAjuV8/g4DdC2ENwRb6E+c= -github.com/dubbogo/gost v1.9.0 h1:UT+dWwvLyJiDotxJERO75jB3Yxgsdy10KztR5ycxRAk= github.com/dubbogo/gost v1.9.0/go.mod h1:pPTjVyoJan3aPxBPNUX0ADkXjPibLo+/Ib0/fADXSG8= github.com/dubbogo/gost v1.9.1 h1:0/PPFo13zPbjt4Ia0zYWMFi3C6rAe9X7O1J2Iv+BHNM= github.com/dubbogo/gost v1.9.1/go.mod h1:pPTjVyoJan3aPxBPNUX0ADkXjPibLo+/Ib0/fADXSG8= @@ -805,13 +804,11 @@ golang.org/x/tools v0.0.0-20190907020128-2ca718005c18/go.mod h1:b+2E5dAYhXwXZwtn golang.org/x/tools v0.0.0-20190911174233-4f2ddba30aff/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.0.0-20191012152004-8de300cfc20a/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= -golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5 h1:hKsoRgsbwY1NafxrwTs+k64bikrLBkAgPir1TNCj3Zs= golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.0.0-20191112195655-aa38f8e97acc h1:NCy3Ohtk6Iny5V/reW2Ktypo4zIpWBdRJ1uFMjBxdg8= golang.org/x/tools v0.0.0-20191112195655-aa38f8e97acc/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= google.golang.org/api v0.4.0/go.mod h1:8k5glujaEP+g9n7WNsDg8QP6cUVNI86fCNMcbazEtwE= -google.golang.org/api v0.7.0 h1:9sdfJOzWlkqPltHAuzT2Cp+yrBeY1KRVYgms8soxMwM= google.golang.org/api v0.7.0/go.mod h1:WtwebWUNSVBH/HAw79HIFXZNqEvBhG+Ra+ax0hx3E3M= google.golang.org/api v0.8.0/go.mod h1:o4eAsZoiT+ibD93RtjEohWalFOjRDx6CVaqeizhEnKg= google.golang.org/api v0.9.0/go.mod h1:o4eAsZoiT+ibD93RtjEohWalFOjRDx6CVaqeizhEnKg= @@ -820,7 +817,6 @@ google.golang.org/api v0.13.0/go.mod h1:iLdEw5Ide6rF15KTC1Kkl0iskquN2gFfn9o9XIsb google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM= google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4= google.golang.org/appengine v1.5.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4= -google.golang.org/appengine v1.6.0 h1:Tfd7cKwKbFRsI8RMAD3oqqw7JPFRrvFlOsfbgVkjOOw= google.golang.org/appengine v1.6.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4= google.golang.org/appengine v1.6.1 h1:QzqyMA1tlu6CgqCDUtU9V+ZKhLFT2dkJuANu5QaxI3I= google.golang.org/appengine v1.6.1/go.mod h1:i06prIuMbXzDqacNJfV5OdTW448YApPu5ww/cMBSeb0= @@ -831,7 +827,6 @@ google.golang.org/genproto v0.0.0-20190418145605-e7d98fc518a7/go.mod h1:VzzqZJRn google.golang.org/genproto v0.0.0-20190425155659-357c62f0e4bb/go.mod h1:VzzqZJRnGkLBvHegQrXjBqPurQTc5/KpmUdxsrq26oE= google.golang.org/genproto v0.0.0-20190502173448-54afdca5d873/go.mod h1:VzzqZJRnGkLBvHegQrXjBqPurQTc5/KpmUdxsrq26oE= google.golang.org/genproto v0.0.0-20190530194941-fb225487d101/go.mod h1:z3L6/3dTEVtUr6QSP8miRzeRqwQOioJ9I66odjN4I7s= -google.golang.org/genproto v0.0.0-20190801165951-fa694d86fc64 h1:iKtrH9Y8mcbADOP0YFaEMth7OfuHY9xHOwNj4znpM1A= google.golang.org/genproto v0.0.0-20190801165951-fa694d86fc64/go.mod h1:DMBHOl98Agz4BDEuKkezgsaosCRResVns1a3J2ZsMNc= google.golang.org/genproto v0.0.0-20190819201941-24fa4b261c55/go.mod h1:DMBHOl98Agz4BDEuKkezgsaosCRResVns1a3J2ZsMNc= google.golang.org/genproto v0.0.0-20190911173649-1774047e7e51/go.mod h1:IbNlFCBrqXvoKpeg0TB2l7cyZUmoaFKYIwrEpbDKLA8= @@ -859,7 +854,6 @@ gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMy gopkg.in/gemnasium/logrus-airbrake-hook.v2 v2.1.2/go.mod h1:Xk6kEKp8OKb+X14hQBKWaSkCsqBpgog8nAV2xsGOxlo= gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc= gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw= -gopkg.in/ini.v1 v1.42.0 h1:7N3gPTt50s8GuLortA00n8AqRTk75qOP98+mTPpgzRk= gopkg.in/ini.v1 v1.42.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k= gopkg.in/ini.v1 v1.51.0 h1:AQvPpx3LzTDM0AjnIRlVFwFFGC+npRopjZxLJj6gdno= gopkg.in/ini.v1 v1.51.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k= diff --git a/metadata/service/inmemory/service_proxy.go b/metadata/service/inmemory/service_proxy.go index 7e01439f04..e2b29686f4 100644 --- a/metadata/service/inmemory/service_proxy.go +++ b/metadata/service/inmemory/service_proxy.go @@ -55,7 +55,7 @@ func (m *MetadataServiceProxy) GetExportedURLs(serviceInterface string, group st inv := invocation.NewRPCInvocationWithOptions(invocation.WithMethodName(methodName), invocation.WithArguments([]interface{}{siV.Interface(), gV.Interface(), vV.Interface(), pV.Interface()}), invocation.WithReply(reflect.ValueOf(&[]interface{}{}).Interface()), - invocation.WithAttachments(map[string]string{constant.ASYNC_KEY: "false"}), + invocation.WithAttachments(map[string]interface{}{constant.ASYNC_KEY: "false"}), invocation.WithParameterValues([]reflect.Value{siV, gV, vV, pV})) res := m.invkr.Invoke(context.Background(), inv) diff --git a/metrics/prometheus/reporter_test.go b/metrics/prometheus/reporter_test.go index 0cb7d09a2c..eaba0e324f 100644 --- a/metrics/prometheus/reporter_test.go +++ b/metrics/prometheus/reporter_test.go @@ -43,7 +43,7 @@ func TestPrometheusReporter_Report(t *testing.T) { "service.filter=echo%2Ctoken%2Caccesslog×tamp=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) assert.False(t, isConsumer(url)) diff --git a/protocol/dubbo/client.go b/protocol/dubbo/client.go index 530beba351..b6e4618bb5 100644 --- a/protocol/dubbo/client.go +++ b/protocol/dubbo/client.go @@ -26,7 +26,6 @@ import ( import ( "github.com/apache/dubbo-getty" - hessian "github.com/apache/dubbo-go-hessian2" gxsync "github.com/dubbogo/gost/sync" perrors "github.com/pkg/errors" "go.uber.org/atomic" @@ -38,6 +37,7 @@ import ( "github.com/apache/dubbo-go/common/constant" "github.com/apache/dubbo-go/common/logger" "github.com/apache/dubbo-go/config" + "github.com/apache/dubbo-go/protocol/dubbo/hessian2" ) var ( @@ -173,11 +173,11 @@ type Request struct { svcUrl common.URL method string args interface{} - atta map[string]string + atta map[string]interface{} } // NewRequest create a new Request. -func NewRequest(addr string, svcUrl common.URL, method string, args interface{}, atta map[string]string) *Request { +func NewRequest(addr string, svcUrl common.URL, method string, args interface{}, atta map[string]interface{}) *Request { return &Request{ addr: addr, svcUrl: svcUrl, @@ -190,11 +190,11 @@ func NewRequest(addr string, svcUrl common.URL, method string, args interface{}, // Response is dubbo protocol response. type Response struct { reply interface{} - atta map[string]string + atta map[string]interface{} } // NewResponse creates a new Response. -func NewResponse(reply interface{}, atta map[string]string) *Response { +func NewResponse(reply interface{}, atta map[string]interface{}) *Response { return &Response{ reply: reply, atta: atta, @@ -240,16 +240,16 @@ func (c *Client) call(ct CallType, request *Request, response *Response, callbac } p.Header.SerialID = byte(S_Dubbo) - p.Body = hessian.NewRequest(request.args, request.atta) + p.Body = hessian2.NewRequest(request.args, request.atta) var rsp *PendingResponse if ct != CT_OneWay { - p.Header.Type = hessian.PackageRequest_TwoWay + p.Header.Type = hessian2.PackageRequest_TwoWay rsp = NewPendingResponse() rsp.response = response rsp.callback = callback } else { - p.Header.Type = hessian.PackageRequest + p.Header.Type = hessian2.PackageRequest } var ( @@ -323,9 +323,9 @@ func (c *Client) transfer(session getty.Session, pkg *DubboPackage, if pkg == nil { pkg = &DubboPackage{} - pkg.Body = hessian.NewRequest([]interface{}{}, nil) + pkg.Body = hessian2.NewRequest([]interface{}{}, nil) pkg.Body = []interface{}{} - pkg.Header.Type = hessian.PackageHeartbeat + pkg.Header.Type = hessian2.PackageHeartbeat pkg.Header.SerialID = byte(S_Dubbo) } pkg.Header.ID = int64(sequence) diff --git a/protocol/dubbo/client_test.go b/protocol/dubbo/client_test.go index 8b0ba169b8..a3b194ad40 100644 --- a/protocol/dubbo/client_test.go +++ b/protocol/dubbo/client_test.go @@ -20,6 +20,7 @@ package dubbo import ( "bytes" "context" + "fmt" "sync" "testing" "time" @@ -129,6 +130,15 @@ func TestClientCall(t *testing.T) { assert.NoError(t, err) assert.Equal(t, User{Id: "1", Name: ""}, *user) + user = &User{} + r1 := "v1" + r2 := &AttaTestObject{Ti: 45, Desc: "v2"} + err = c.Call(NewRequest(mockAddress, url, "GetUserForAttachment", []interface{}{1}, map[string]interface{}{"sim": r1, "comp": r2}), NewResponse(user, nil)) + assert.NoError(t, err) + // the param is transfered from client to server by attachment, and the server will return the attachment value. + // the test should check the value came back from server. + assert.Equal(t, User{Id: r1, Name: fmt.Sprintf("%+v", *r2)}, *user) + // destroy proto.Destroy() } @@ -166,10 +176,11 @@ func TestClientAsyncCall(t *testing.T) { func InitTest(t *testing.T) (protocol.Protocol, common.URL) { hessian.RegisterPOJO(&User{}) + hessian.RegisterPOJO(&AttaTestObject{}) methods, err := common.ServiceMap.Register("com.ikurento.user.UserProvider", "dubbo", &UserProvider{}) assert.NoError(t, err) - assert.Equal(t, "GetBigPkg,GetUser,GetUser0,GetUser1,GetUser2,GetUser3,GetUser4,GetUser5,GetUser6", methods) + assert.Equal(t, "GetBigPkg,GetUser,GetUser0,GetUser1,GetUser2,GetUser3,GetUser4,GetUser5,GetUser6,GetUserForAttachment", methods) // config SetClientConf(ClientConfig{ @@ -296,6 +307,16 @@ func (u *UserProvider) GetUser6(id int64) (*User, error) { return &User{Id: "1"}, nil } +func (u *UserProvider) GetUserForAttachment(context context.Context, id int64) (*User, error) { + if id == 0 { + return nil, nil + } + var attachments = context.Value("attachment").(map[string]interface{}) + Id := attachments["sim"].(string) + name := fmt.Sprintf("%+v", *(attachments["comp"].(*AttaTestObject))) + return &User{Id: Id, Name: name}, nil +} + func (u *UserProvider) Reference() string { return "UserProvider" } @@ -303,3 +324,12 @@ func (u *UserProvider) Reference() string { func (u User) JavaClassName() string { return "com.ikurento.user.User" } + +type AttaTestObject struct { + Ti int64 + Desc string +} + +func (u *AttaTestObject) JavaClassName() string { + return "UserProvider" +} diff --git a/protocol/dubbo/codec.go b/protocol/dubbo/codec.go index 9781c70115..c33c92dfd9 100644 --- a/protocol/dubbo/codec.go +++ b/protocol/dubbo/codec.go @@ -25,11 +25,14 @@ import ( ) import ( - "github.com/apache/dubbo-go-hessian2" "github.com/apache/dubbo-go/common" perrors "github.com/pkg/errors" ) +import ( + "github.com/apache/dubbo-go/protocol/dubbo/hessian2" +) + //SerialID serial ID type SerialID byte @@ -59,8 +62,8 @@ type SequenceType int64 // nolint type DubboPackage struct { - Header hessian.DubboHeader - Service hessian.Service + Header hessian2.DubboHeader + Service hessian2.Service Body interface{} Err error } @@ -72,7 +75,7 @@ func (p DubboPackage) String() string { // Marshal encode hessian package. func (p *DubboPackage) Marshal() (*bytes.Buffer, error) { - codec := hessian.NewHessianCodec(nil) + codec := hessian2.NewHessianCodec(nil) pkg, err := codec.Write(p.Service, p.Header, p.Body) if err != nil { @@ -86,11 +89,11 @@ func (p *DubboPackage) Marshal() (*bytes.Buffer, error) { func (p *DubboPackage) Unmarshal(buf *bytes.Buffer, opts ...interface{}) error { // fix issue https://github.com/apache/dubbo-go/issues/380 bufLen := buf.Len() - if bufLen < hessian.HEADER_LENGTH { - return perrors.WithStack(hessian.ErrHeaderNotEnough) + if bufLen < hessian2.HEADER_LENGTH { + return perrors.WithStack(hessian2.ErrHeaderNotEnough) } - codec := hessian.NewHessianCodec(bufio.NewReaderSize(buf, bufLen)) + codec := hessian2.NewHessianCodec(bufio.NewReaderSize(buf, bufLen)) // read header err := codec.ReadHeader(&p.Header) @@ -104,7 +107,7 @@ func (p *DubboPackage) Unmarshal(buf *bytes.Buffer, opts ...interface{}) error { return perrors.Errorf("opts[0] is not of type *Client") } - if p.Header.Type&hessian.PackageRequest != 0x00 { + if p.Header.Type&hessian2.PackageRequest != 0x00 { // size of this array must be '7' // https://github.com/apache/dubbo-go-hessian2/blob/master/request.go#L272 p.Body = make([]interface{}, 7) @@ -113,7 +116,7 @@ func (p *DubboPackage) Unmarshal(buf *bytes.Buffer, opts ...interface{}) error { if !ok { return perrors.Errorf("client.GetPendingResponse(%v) = nil", p.Header.ID) } - p.Body = &hessian.Response{RspObj: pendingRsp.(*PendingResponse).response.reply} + p.Body = &hessian2.DubboResponse{RspObj: pendingRsp.(*PendingResponse).response.reply} } } diff --git a/protocol/dubbo/codec_test.go b/protocol/dubbo/codec_test.go index c2ca443637..6dfb87ea00 100644 --- a/protocol/dubbo/codec_test.go +++ b/protocol/dubbo/codec_test.go @@ -24,15 +24,18 @@ import ( ) import ( - hessian "github.com/apache/dubbo-go-hessian2" perrors "github.com/pkg/errors" "github.com/stretchr/testify/assert" ) +import ( + "github.com/apache/dubbo-go/protocol/dubbo/hessian2" +) + func TestDubboPackageMarshalAndUnmarshal(t *testing.T) { pkg := &DubboPackage{} pkg.Body = []interface{}{"a"} - pkg.Header.Type = hessian.PackageHeartbeat + pkg.Header.Type = hessian2.PackageHeartbeat pkg.Header.SerialID = byte(S_Dubbo) pkg.Header.ID = 10086 @@ -44,13 +47,13 @@ func TestDubboPackageMarshalAndUnmarshal(t *testing.T) { pkgres.Body = []interface{}{} err = pkgres.Unmarshal(data) assert.NoError(t, err) - assert.Equal(t, hessian.PackageHeartbeat|hessian.PackageRequest|hessian.PackageRequest_TwoWay, pkgres.Header.Type) + assert.Equal(t, hessian2.PackageHeartbeat|hessian2.PackageRequest|hessian2.PackageRequest_TwoWay, pkgres.Header.Type) assert.Equal(t, byte(S_Dubbo), pkgres.Header.SerialID) assert.Equal(t, int64(10086), pkgres.Header.ID) assert.Equal(t, 0, len(pkgres.Body.([]interface{}))) // request - pkg.Header.Type = hessian.PackageRequest + pkg.Header.Type = hessian2.PackageRequest pkg.Service.Interface = "Service" pkg.Service.Path = "path" pkg.Service.Version = "2.6" @@ -63,7 +66,7 @@ func TestDubboPackageMarshalAndUnmarshal(t *testing.T) { pkgres.Body = make([]interface{}, 7) err = pkgres.Unmarshal(data) assert.NoError(t, err) - assert.Equal(t, hessian.PackageRequest, pkgres.Header.Type) + assert.Equal(t, hessian2.PackageRequest, pkgres.Header.Type) assert.Equal(t, byte(S_Dubbo), pkgres.Header.SerialID) assert.Equal(t, int64(10086), pkgres.Header.ID) assert.Equal(t, "2.0.2", pkgres.Body.([]interface{})[0]) @@ -72,12 +75,12 @@ func TestDubboPackageMarshalAndUnmarshal(t *testing.T) { assert.Equal(t, "Method", pkgres.Body.([]interface{})[3]) assert.Equal(t, "Ljava/lang/String;", pkgres.Body.([]interface{})[4]) assert.Equal(t, []interface{}{"a"}, pkgres.Body.([]interface{})[5]) - assert.Equal(t, map[string]string{"dubbo": "2.0.2", "interface": "Service", "path": "path", "timeout": "1000", "version": "2.6"}, pkgres.Body.([]interface{})[6]) + assert.Equal(t, map[string]interface{}{"dubbo": "2.0.2", "interface": "Service", "path": "path", "timeout": "1000", "version": "2.6"}, pkgres.Body.([]interface{})[6]) } func TestIssue380(t *testing.T) { pkg := &DubboPackage{} buf := bytes.NewBuffer([]byte("hello")) err := pkg.Unmarshal(buf) - assert.True(t, perrors.Cause(err) == hessian.ErrHeaderNotEnough) + assert.True(t, perrors.Cause(err) == hessian2.ErrHeaderNotEnough) } diff --git a/protocol/dubbo/dubbo_invoker.go b/protocol/dubbo/dubbo_invoker.go index 59202d5f49..983a05dcec 100644 --- a/protocol/dubbo/dubbo_invoker.go +++ b/protocol/dubbo/dubbo_invoker.go @@ -150,8 +150,7 @@ func (di *DubboInvoker) appendCtx(ctx context.Context, inv *invocation_impl.RPCI // inject opentracing ctx currentSpan := opentracing.SpanFromContext(ctx) if currentSpan != nil { - carrier := opentracing.TextMapCarrier(inv.Attachments()) - err := opentracing.GlobalTracer().Inject(currentSpan.Context(), opentracing.TextMap, carrier) + err := injectTraceCtx(currentSpan, inv) if err != nil { logger.Errorf("Could not inject the span context into attachments: %v", err) } diff --git a/protocol/dubbo/dubbo_invoker_test.go b/protocol/dubbo/dubbo_invoker_test.go index c0640d5558..bf352c082c 100644 --- a/protocol/dubbo/dubbo_invoker_test.go +++ b/protocol/dubbo/dubbo_invoker_test.go @@ -52,7 +52,7 @@ func TestDubboInvokerInvoke(t *testing.T) { user := &User{} inv := invocation.NewRPCInvocationWithOptions(invocation.WithMethodName(mockMethodNameGetUser), invocation.WithArguments([]interface{}{"1", "username"}), - invocation.WithReply(user), invocation.WithAttachments(map[string]string{"test_key": "test_value"})) + invocation.WithReply(user), invocation.WithAttachments(map[string]interface{}{"test_key": "test_value"})) // Call res := invoker.Invoke(context.Background(), inv) diff --git a/protocol/dubbo/hessian2/const.go b/protocol/dubbo/hessian2/const.go new file mode 100644 index 0000000000..74a00b601d --- /dev/null +++ b/protocol/dubbo/hessian2/const.go @@ -0,0 +1,243 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +// This constants are also defined in dubbo constants. But we will still used these until hessian is stable. + +package hessian2 + +import ( + "reflect" + "regexp" +) + +import ( + perrors "github.com/pkg/errors" +) + +const ( + mask = byte(127) + flag = byte(128) +) + +const ( + // Zero : byte zero + Zero = byte(0x00) +) + +// constansts +const ( + TAG_READ = int32(-1) + ASCII_GAP = 32 + CHUNK_SIZE = 4096 + BC_BINARY = byte('B') // final chunk + BC_BINARY_CHUNK = byte('A') // non-final chunk + + BC_BINARY_DIRECT = byte(0x20) // 1-byte length binary + BINARY_DIRECT_MAX = byte(0x0f) + BC_BINARY_SHORT = byte(0x34) // 2-byte length binary + BINARY_SHORT_MAX = 0x3ff // 0-1023 binary + + BC_DATE = byte(0x4a) // 64-bit millisecond UTC date + BC_DATE_MINUTE = byte(0x4b) // 32-bit minute UTC date + + BC_DOUBLE = byte('D') // IEEE 64-bit double + + BC_DOUBLE_ZERO = byte(0x5b) + BC_DOUBLE_ONE = byte(0x5c) + BC_DOUBLE_BYTE = byte(0x5d) + BC_DOUBLE_SHORT = byte(0x5e) + BC_DOUBLE_MILL = byte(0x5f) + + BC_FALSE = byte('F') // boolean false + + BC_INT = byte('I') // 32-bit int + + INT_DIRECT_MIN = -0x10 + INT_DIRECT_MAX = byte(0x2f) + BC_INT_ZERO = byte(0x90) + + INT_BYTE_MIN = -0x800 + INT_BYTE_MAX = 0x7ff + BC_INT_BYTE_ZERO = byte(0xc8) + + BC_END = byte('Z') + + INT_SHORT_MIN = -0x40000 + INT_SHORT_MAX = 0x3ffff + BC_INT_SHORT_ZERO = byte(0xd4) + + BC_LIST_VARIABLE = byte(0x55) + BC_LIST_FIXED = byte('V') + BC_LIST_VARIABLE_UNTYPED = byte(0x57) + BC_LIST_FIXED_UNTYPED = byte(0x58) + _listFixedTypedLenTagMin = byte(0x70) + _listFixedTypedLenTagMax = byte(0x77) + _listFixedUntypedLenTagMin = byte(0x78) + _listFixedUntypedLenTagMax = byte(0x7f) + + BC_LIST_DIRECT = byte(0x70) + BC_LIST_DIRECT_UNTYPED = byte(0x78) + LIST_DIRECT_MAX = byte(0x7) + + BC_LONG = byte('L') // 64-bit signed integer + LONG_DIRECT_MIN = -0x08 + LONG_DIRECT_MAX = byte(0x0f) + BC_LONG_ZERO = byte(0xe0) + + LONG_BYTE_MIN = -0x800 + LONG_BYTE_MAX = 0x7ff + BC_LONG_BYTE_ZERO = byte(0xf8) + + LONG_SHORT_MIN = -0x40000 + LONG_SHORT_MAX = 0x3ffff + BC_LONG_SHORT_ZERO = byte(0x3c) + + BC_LONG_INT = byte(0x59) + + BC_MAP = byte('M') + BC_MAP_UNTYPED = byte('H') + + BC_NULL = byte('N') // x4e + + BC_OBJECT = byte('O') + BC_OBJECT_DEF = byte('C') + + BC_OBJECT_DIRECT = byte(0x60) + OBJECT_DIRECT_MAX = byte(0x0f) + + BC_REF = byte(0x51) + + BC_STRING = byte('S') // final string + BC_STRING_CHUNK = byte('R') // non-final string + + BC_STRING_DIRECT = byte(0x00) + STRING_DIRECT_MAX = byte(0x1f) + BC_STRING_SHORT = byte(0x30) + STRING_SHORT_MAX = 0x3ff + + BC_TRUE = byte('T') + + P_PACKET_CHUNK = byte(0x4f) + P_PACKET = byte('P') + + P_PACKET_DIRECT = byte(0x80) + PACKET_DIRECT_MAX = byte(0x7f) + + P_PACKET_SHORT = byte(0x70) + PACKET_SHORT_MAX = 0xfff + ARRAY_STRING = "[string" + ARRAY_INT = "[int" + ARRAY_DOUBLE = "[double" + ARRAY_FLOAT = "[float" + ARRAY_BOOL = "[boolean" + ARRAY_LONG = "[long" + + PATH_KEY = "path" + GROUP_KEY = "group" + INTERFACE_KEY = "interface" + VERSION_KEY = "version" + TIMEOUT_KEY = "timeout" + + STRING_NIL = "" + STRING_TRUE = "true" + STRING_FALSE = "false" + STRING_ZERO = "0.0" + STRING_ONE = "1.0" +) + +// DubboResponse related consts +const ( + Response_OK byte = 20 + Response_CLIENT_TIMEOUT byte = 30 + Response_SERVER_TIMEOUT byte = 31 + Response_BAD_REQUEST byte = 40 + Response_BAD_RESPONSE byte = 50 + Response_SERVICE_NOT_FOUND byte = 60 + Response_SERVICE_ERROR byte = 70 + Response_SERVER_ERROR byte = 80 + Response_CLIENT_ERROR byte = 90 + + // According to "java dubbo" There are two cases of response: + // 1. with attachments + // 2. no attachments + RESPONSE_WITH_EXCEPTION int32 = 0 + RESPONSE_VALUE int32 = 1 + RESPONSE_NULL_VALUE int32 = 2 + RESPONSE_WITH_EXCEPTION_WITH_ATTACHMENTS int32 = 3 + RESPONSE_VALUE_WITH_ATTACHMENTS int32 = 4 + RESPONSE_NULL_VALUE_WITH_ATTACHMENTS int32 = 5 +) + +/** + * the dubbo protocol header length is 16 Bytes. + * the first 2 Bytes is magic code '0xdabb' + * the next 1 Byte is message flags, in which its 16-20 bit is serial id, 21 for event, 22 for two way, 23 for request/response flag + * the next 1 Bytes is response state. + * the next 8 Bytes is package DI. + * the next 4 Bytes is package length. + **/ +const ( + // header length. + HEADER_LENGTH = 16 + + // magic header + MAGIC = uint16(0xdabb) + MAGIC_HIGH = byte(0xda) + MAGIC_LOW = byte(0xbb) + + // message flag. + FLAG_REQUEST = byte(0x80) + FLAG_TWOWAY = byte(0x40) + FLAG_EVENT = byte(0x20) // for heartbeat + SERIAL_MASK = 0x1f + + DUBBO_VERSION = "2.5.4" + DUBBO_VERSION_KEY = "dubbo" + DEFAULT_DUBBO_PROTOCOL_VERSION = "2.0.2" // Dubbo RPC protocol version, for compatibility, it must not be between 2.0.10 ~ 2.6.2 + LOWEST_VERSION_FOR_RESPONSE_ATTACHMENT = 2000200 + DEFAULT_LEN = 8388608 // 8 * 1024 * 1024 default body max length +) + +// regular +const ( + JAVA_IDENT_REGEX = "(?:[_$a-zA-Z][_$a-zA-Z0-9]*)" + CLASS_DESC = "(?:L" + JAVA_IDENT_REGEX + "(?:\\/" + JAVA_IDENT_REGEX + ")*;)" + ARRAY_DESC = "(?:\\[+(?:(?:[VZBCDFIJS])|" + CLASS_DESC + "))" + DESC_REGEX = "(?:(?:[VZBCDFIJS])|" + CLASS_DESC + "|" + ARRAY_DESC + ")" +) + +// Dubbo request response related consts +var ( + DubboRequestHeaderBytesTwoWay = [HEADER_LENGTH]byte{MAGIC_HIGH, MAGIC_LOW, FLAG_REQUEST | FLAG_TWOWAY} + DubboRequestHeaderBytes = [HEADER_LENGTH]byte{MAGIC_HIGH, MAGIC_LOW, FLAG_REQUEST} + DubboResponseHeaderBytes = [HEADER_LENGTH]byte{MAGIC_HIGH, MAGIC_LOW, Zero, Response_OK} + DubboRequestHeartbeatHeader = [HEADER_LENGTH]byte{MAGIC_HIGH, MAGIC_LOW, FLAG_REQUEST | FLAG_TWOWAY | FLAG_EVENT} + DubboResponseHeartbeatHeader = [HEADER_LENGTH]byte{MAGIC_HIGH, MAGIC_LOW, FLAG_EVENT} +) + +// Error part +var ( + ErrHeaderNotEnough = perrors.New("header buffer too short") + ErrBodyNotEnough = perrors.New("body buffer too short") + ErrJavaException = perrors.New("got java exception") + ErrIllegalPackage = perrors.New("illegal package!") +) + +// nolint +var DescRegex, _ = regexp.Compile(DESC_REGEX) + +var NilValue = reflect.Zero(reflect.TypeOf((*interface{})(nil)).Elem()) diff --git a/protocol/dubbo/hessian2/hessian_dubbo.go b/protocol/dubbo/hessian2/hessian_dubbo.go new file mode 100644 index 0000000000..1afa4ec96e --- /dev/null +++ b/protocol/dubbo/hessian2/hessian_dubbo.go @@ -0,0 +1,251 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package hessian2 + +import ( + "bufio" + "encoding/binary" + "time" +) + +import ( + "github.com/apache/dubbo-go-hessian2" + perrors "github.com/pkg/errors" +) + +// enum part +const ( + PackageError = PackageType(0x01) + PackageRequest = PackageType(0x02) + PackageResponse = PackageType(0x04) + PackageHeartbeat = PackageType(0x08) + PackageRequest_TwoWay = PackageType(0x10) + PackageResponse_Exception = PackageType(0x20) + PackageType_BitSize = 0x2f +) + +// PackageType nolint +type PackageType int + +// DubboHeader dubbo header +type DubboHeader struct { + SerialID byte + Type PackageType + ID int64 + BodyLen int + ResponseStatus byte +} + +// Service defines service instance +type Service struct { + Path string + Interface string + Group string + Version string + Method string + Timeout time.Duration // request timeout +} + +// HessianCodec defines hessian codec +type HessianCodec struct { + pkgType PackageType + reader *bufio.Reader + bodyLen int +} + +// NewHessianCodec generate a new hessian codec instance +func NewHessianCodec(reader *bufio.Reader) *HessianCodec { + return &HessianCodec{ + reader: reader, + } +} + +// NewHessianCodec generate a new hessian codec instance +func NewHessianCodecCustom(pkgType PackageType, reader *bufio.Reader, bodyLen int) *HessianCodec { + return &HessianCodec{ + pkgType: pkgType, + reader: reader, + bodyLen: bodyLen, + } +} + +func (h *HessianCodec) Write(service Service, header DubboHeader, body interface{}) ([]byte, error) { + switch header.Type { + case PackageHeartbeat: + if header.ResponseStatus == Zero { + return packRequest(service, header, body) + } + return packResponse(header, body) + + case PackageRequest, PackageRequest_TwoWay: + return packRequest(service, header, body) + + case PackageResponse: + return packResponse(header, body) + + default: + return nil, perrors.Errorf("Unrecognised message type: %v", header.Type) + } + + // unreachable return nil, nil +} + +// ReadHeader uses hessian codec to read dubbo header +func (h *HessianCodec) ReadHeader(header *DubboHeader) error { + + var err error + + if h.reader.Size() < HEADER_LENGTH { + return ErrHeaderNotEnough + } + buf, err := h.reader.Peek(HEADER_LENGTH) + if err != nil { // this is impossible + return perrors.WithStack(err) + } + _, err = h.reader.Discard(HEADER_LENGTH) + if err != nil { // this is impossible + return perrors.WithStack(err) + } + + //// read header + + if buf[0] != MAGIC_HIGH && buf[1] != MAGIC_LOW { + return ErrIllegalPackage + } + + // Header{serialization id(5 bit), event, two way, req/response} + if header.SerialID = buf[2] & SERIAL_MASK; header.SerialID == Zero { + return perrors.Errorf("serialization ID:%v", header.SerialID) + } + + flag := buf[2] & FLAG_EVENT + if flag != Zero { + header.Type |= PackageHeartbeat + } + flag = buf[2] & FLAG_REQUEST + if flag != Zero { + header.Type |= PackageRequest + flag = buf[2] & FLAG_TWOWAY + if flag != Zero { + header.Type |= PackageRequest_TwoWay + } + } else { + header.Type |= PackageResponse + header.ResponseStatus = buf[3] + if header.ResponseStatus != Response_OK { + header.Type |= PackageResponse_Exception + } + } + + // Header{req id} + header.ID = int64(binary.BigEndian.Uint64(buf[4:])) + + // Header{body len} + header.BodyLen = int(binary.BigEndian.Uint32(buf[12:])) + if header.BodyLen < 0 { + return ErrIllegalPackage + } + + h.pkgType = header.Type + h.bodyLen = header.BodyLen + + if h.reader.Buffered() < h.bodyLen { + return ErrBodyNotEnough + } + + return perrors.WithStack(err) + +} + +// ReadBody uses hessian codec to read response body +func (h *HessianCodec) ReadBody(rspObj interface{}) error { + + if h.reader.Buffered() < h.bodyLen { + return ErrBodyNotEnough + } + buf, err := h.reader.Peek(h.bodyLen) + if err != nil { + return perrors.WithStack(err) + } + _, err = h.reader.Discard(h.bodyLen) + if err != nil { // this is impossible + return perrors.WithStack(err) + } + + switch h.pkgType & PackageType_BitSize { + case PackageResponse | PackageHeartbeat | PackageResponse_Exception, PackageResponse | PackageResponse_Exception: + decoder := hessian.NewDecoder(buf[:]) + exception, err := decoder.Decode() + if err != nil { + return perrors.WithStack(err) + } + rsp, ok := rspObj.(*DubboResponse) + if !ok { + return perrors.Errorf("java exception:%s", exception.(string)) + } + rsp.Exception = perrors.Errorf("java exception:%s", exception.(string)) + return nil + case PackageRequest | PackageHeartbeat, PackageResponse | PackageHeartbeat: + case PackageRequest: + if rspObj != nil { + if err = unpackRequestBody(hessian.NewDecoder(buf[:]), rspObj); err != nil { + return perrors.WithStack(err) + } + } + case PackageResponse: + if rspObj != nil { + if err = unpackResponseBody(hessian.NewDecoder(buf[:]), rspObj); err != nil { + return perrors.WithStack(err) + } + } + } + + return nil +} + +// ignore body, but only read attachments +func (h *HessianCodec) ReadAttachments() (map[string]interface{}, error) { + if h.reader.Buffered() < h.bodyLen { + return nil, ErrBodyNotEnough + } + buf, err := h.reader.Peek(h.bodyLen) + if err != nil { + return nil, perrors.WithStack(err) + } + _, err = h.reader.Discard(h.bodyLen) + if err != nil { // this is impossible + return nil, perrors.WithStack(err) + } + + switch h.pkgType & PackageType_BitSize { + case PackageRequest: + rspObj := make([]interface{}, 7) + if err = unpackRequestBody(hessian.NewDecoderWithSkip(buf[:]), rspObj); err != nil { + return nil, perrors.WithStack(err) + } + return rspObj[6].(map[string]interface{}), nil + case PackageResponse: + rspObj := &DubboResponse{} + if err = unpackResponseBody(hessian.NewDecoderWithSkip(buf[:]), rspObj); err != nil { + return nil, perrors.WithStack(err) + } + return rspObj.Attachments, nil + } + + return nil, nil +} diff --git a/protocol/dubbo/hessian2/hessian_dubbo_test.go b/protocol/dubbo/hessian2/hessian_dubbo_test.go new file mode 100644 index 0000000000..c3f19f0453 --- /dev/null +++ b/protocol/dubbo/hessian2/hessian_dubbo_test.go @@ -0,0 +1,231 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package hessian2 + +import ( + "bufio" + "bytes" + "reflect" + "testing" + "time" +) + +import ( + hessian "github.com/apache/dubbo-go-hessian2" + "github.com/stretchr/testify/assert" +) + +type Case struct { + A string + B int +} + +type CaseA struct { + A string + B int + C Case +} + +type CaseB struct { + A string + B CaseA +} + +func (c *CaseB) JavaClassName() string { + return "com.test.caseb" +} + +func (c CaseA) JavaClassName() string { + return "com.test.casea" +} + +//JavaClassName java fully qualified path +func (c Case) JavaClassName() string { + return "com.test.case" +} + +func doTestHessianEncodeHeader(t *testing.T, packageType PackageType, responseStatus byte, body interface{}) ([]byte, error) { + hessian.RegisterPOJO(&Case{}) + codecW := NewHessianCodec(nil) + resp, err := codecW.Write(Service{ + Path: "test", + Interface: "ITest", + Version: "v1.0", + Method: "test", + Timeout: time.Second * 10, + }, DubboHeader{ + SerialID: 2, + Type: packageType, + ID: 1, + ResponseStatus: responseStatus, + }, body) + assert.Nil(t, err) + return resp, err +} + +func doTestResponse(t *testing.T, packageType PackageType, responseStatus byte, body interface{}, decodedResponse *DubboResponse, assertFunc func()) { + resp, err := doTestHessianEncodeHeader(t, packageType, responseStatus, body) + + codecR := NewHessianCodec(bufio.NewReader(bytes.NewReader(resp))) + + h := &DubboHeader{} + err = codecR.ReadHeader(h) + assert.Nil(t, err) + + assert.Equal(t, byte(2), h.SerialID) + assert.Equal(t, packageType, h.Type&(PackageRequest|PackageResponse|PackageHeartbeat)) + assert.Equal(t, int64(1), h.ID) + assert.Equal(t, responseStatus, h.ResponseStatus) + + err = codecR.ReadBody(decodedResponse) + assert.Nil(t, err) + t.Log(decodedResponse) + + if assertFunc != nil { + assertFunc() + return + } + + if h.ResponseStatus != Zero && h.ResponseStatus != Response_OK { + assert.Equal(t, "java exception:"+body.(string), decodedResponse.Exception.Error()) + return + } + + in, _ := hessian.EnsureInterface(hessian.UnpackPtrValue(hessian.EnsurePackValue(body)), nil) + out, _ := hessian.EnsureInterface(hessian.UnpackPtrValue(hessian.EnsurePackValue(decodedResponse.RspObj)), nil) + assert.Equal(t, in, out) +} + +func TestResponse(t *testing.T) { + caseObj := Case{A: "a", B: 1} + decodedResponse := &DubboResponse{} + + arr := []*Case{&caseObj} + var arrRes []interface{} + decodedResponse.RspObj = &arrRes + doTestResponse(t, PackageResponse, Response_OK, arr, decodedResponse, func() { + assert.Equal(t, 1, len(arrRes)) + assert.Equal(t, &caseObj, arrRes[0]) + }) + + decodedResponse.RspObj = &Case{} + doTestResponse(t, PackageResponse, Response_OK, &Case{A: "a", B: 1}, decodedResponse, nil) + + s := "ok!!!!!" + strObj := "" + decodedResponse.RspObj = &strObj + doTestResponse(t, PackageResponse, Response_OK, s, decodedResponse, nil) + + var intObj int64 + decodedResponse.RspObj = &intObj + doTestResponse(t, PackageResponse, Response_OK, int64(3), decodedResponse, nil) + + boolObj := false + decodedResponse.RspObj = &boolObj + doTestResponse(t, PackageResponse, Response_OK, true, decodedResponse, nil) + + strObj = "" + decodedResponse.RspObj = &strObj + doTestResponse(t, PackageResponse, hessian.Response_SERVER_ERROR, "error!!!!!", decodedResponse, nil) + + mapObj := map[string][]*Case{"key": {&caseObj}} + mapRes := map[interface{}]interface{}{} + decodedResponse.RspObj = &mapRes + doTestResponse(t, PackageResponse, Response_OK, mapObj, decodedResponse, func() { + c, ok := mapRes["key"] + if !ok { + assert.FailNow(t, "no key in decoded response map") + } + + mapValueArr, ok := c.([]*Case) + if !ok { + assert.FailNow(t, "invalid decoded response map value", "expect []*Case, but get %v", reflect.TypeOf(c)) + } + assert.Equal(t, 1, len(mapValueArr)) + assert.Equal(t, &caseObj, mapValueArr[0]) + }) +} + +func doTestRequest(t *testing.T, packageType PackageType, responseStatus byte, body interface{}) { + resp, err := doTestHessianEncodeHeader(t, packageType, responseStatus, body) + + codecR := NewHessianCodec(bufio.NewReader(bytes.NewReader(resp))) + + h := &DubboHeader{} + err = codecR.ReadHeader(h) + assert.Nil(t, err) + assert.Equal(t, byte(2), h.SerialID) + assert.Equal(t, packageType, h.Type&(PackageRequest|PackageResponse|PackageHeartbeat)) + assert.Equal(t, int64(1), h.ID) + assert.Equal(t, responseStatus, h.ResponseStatus) + + c := make([]interface{}, 7) + err = codecR.ReadBody(c) + assert.Nil(t, err) + t.Log(c) + assert.True(t, len(body.([]interface{})) == len(c[5].([]interface{}))) +} + +func TestRequest(t *testing.T) { + doTestRequest(t, PackageRequest, Zero, []interface{}{"a"}) + doTestRequest(t, PackageRequest, Zero, []interface{}{"a", 3}) + doTestRequest(t, PackageRequest, Zero, []interface{}{"a", true}) + doTestRequest(t, PackageRequest, Zero, []interface{}{"a", 3, true}) + doTestRequest(t, PackageRequest, Zero, []interface{}{3.2, true}) + doTestRequest(t, PackageRequest, Zero, []interface{}{"a", 3, true, &Case{A: "a", B: 3}}) + doTestRequest(t, PackageRequest, Zero, []interface{}{"a", 3, true, []*Case{{A: "a", B: 3}}}) + doTestRequest(t, PackageRequest, Zero, []interface{}{map[string][]*Case{"key": {{A: "a", B: 3}}}}) +} + +func TestHessianCodec_ReadAttachments(t *testing.T) { + hessian.RegisterPOJO(&AttachTestObject{}) + body := &DubboResponse{ + RspObj: &CaseB{A: "A", B: CaseA{A: "a", B: 1, C: Case{A: "c", B: 2}}}, + Exception: nil, + Attachments: map[string]interface{}{DUBBO_VERSION_KEY: "2.6.4", "att": AttachTestObject{Id: 23, Name: "haha"}}, + } + resp, err := doTestHessianEncodeHeader(t, PackageResponse, Response_OK, body) + assert.NoError(t, err) + hessian.UnRegisterPOJOs(&CaseB{}, &CaseA{}) + codecR1 := NewHessianCodec(bufio.NewReader(bytes.NewReader(resp))) + codecR2 := NewHessianCodec(bufio.NewReader(bytes.NewReader(resp))) + h := &DubboHeader{} + assert.NoError(t, codecR1.ReadHeader(h)) + t.Log(h) + assert.NoError(t, codecR2.ReadHeader(h)) + t.Log(h) + + err = codecR1.ReadBody(body) + assert.Equal(t, "can not find go type name com.test.caseb in registry", err.Error()) + attrs, err := codecR2.ReadAttachments() + assert.NoError(t, err) + assert.Equal(t, "2.6.4", attrs[DUBBO_VERSION_KEY]) + assert.Equal(t, AttachTestObject{Id: 23, Name: "haha"}, *(attrs["att"].(*AttachTestObject))) + assert.NotEqual(t, AttachTestObject{Id: 24, Name: "nohaha"}, *(attrs["att"].(*AttachTestObject))) + + t.Log(attrs) +} + +type AttachTestObject struct { + Id int32 + Name string `dubbo:name` +} + +func (AttachTestObject) JavaClassName() string { + return "com.test.Test" +} diff --git a/protocol/dubbo/hessian2/hessian_request.go b/protocol/dubbo/hessian2/hessian_request.go new file mode 100644 index 0000000000..4ebb4aa1be --- /dev/null +++ b/protocol/dubbo/hessian2/hessian_request.go @@ -0,0 +1,350 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package hessian2 + +import ( + "encoding/binary" + "reflect" + "strconv" + "strings" + "time" +) + +import ( + hessian "github.com/apache/dubbo-go-hessian2" + perrors "github.com/pkg/errors" +) + +///////////////////////////////////////// +// dubbo +///////////////////////////////////////// + +func getArgType(v interface{}) string { + if v == nil { + return "V" + } + + switch v.(type) { + // Serialized tags for base types + case nil: + return "V" + case bool: + return "Z" + case []bool: + return "[Z" + case byte: + return "B" + case []byte: + return "[B" + case int8: + return "B" + case []int8: + return "[B" + case int16: + return "S" + case []int16: + return "[S" + case uint16: // Equivalent to Char of Java + return "C" + case []uint16: + return "[C" + // case rune: + // return "C" + case int: + return "J" + case []int: + return "[J" + case int32: + return "I" + case []int32: + return "[I" + case int64: + return "J" + case []int64: + return "[J" + case time.Time: + return "java.util.Date" + case []time.Time: + return "[Ljava.util.Date" + case float32: + return "F" + case []float32: + return "[F" + case float64: + return "D" + case []float64: + return "[D" + case string: + return "java.lang.String" + case []string: + return "[Ljava.lang.String;" + case []hessian.Object: + return "[Ljava.lang.Object;" + case map[interface{}]interface{}: + // return "java.util.HashMap" + return "java.util.Map" + case hessian.POJOEnum: + return v.(hessian.POJOEnum).JavaClassName() + // Serialized tags for complex types + default: + t := reflect.TypeOf(v) + if reflect.Ptr == t.Kind() { + t = reflect.TypeOf(reflect.ValueOf(v).Elem()) + } + switch t.Kind() { + case reflect.Struct: + return "java.lang.Object" + case reflect.Slice, reflect.Array: + if t.Elem().Kind() == reflect.Struct { + return "[Ljava.lang.Object;" + } + // return "java.util.ArrayList" + return "java.util.List" + case reflect.Map: // Enter here, map may be map[string]int + return "java.util.Map" + default: + return "" + } + } + + // unreachable + // return "java.lang.RuntimeException" +} + +func getArgsTypeList(args []interface{}) (string, error) { + var ( + typ string + types string + ) + + for i := range args { + typ = getArgType(args[i]) + if typ == "" { + return types, perrors.Errorf("cat not get arg %#v type", args[i]) + } + if !strings.Contains(typ, ".") { + types += typ + } else if strings.Index(typ, "[") == 0 { + types += strings.Replace(typ, ".", "/", -1) + } else { + // java.util.List -> Ljava/util/List; + types += "L" + strings.Replace(typ, ".", "/", -1) + ";" + } + } + + return types, nil +} + +type DubboRequest struct { + Params interface{} + Attachments map[string]interface{} +} + +// NewRequest create a new DubboRequest +func NewRequest(params interface{}, atta map[string]interface{}) *DubboRequest { + if atta == nil { + atta = make(map[string]interface{}) + } + return &DubboRequest{ + Params: params, + Attachments: atta, + } +} + +func EnsureRequest(body interface{}) *DubboRequest { + if req, ok := body.(*DubboRequest); ok { + return req + } + return NewRequest(body, nil) +} + +func packRequest(service Service, header DubboHeader, req interface{}) ([]byte, error) { + var ( + err error + types string + byteArray []byte + pkgLen int + ) + + request := EnsureRequest(req) + + args, ok := request.Params.([]interface{}) + if !ok { + return nil, perrors.Errorf("@params is not of type: []interface{}") + } + + hb := header.Type == PackageHeartbeat + + ////////////////////////////////////////// + // byteArray + ////////////////////////////////////////// + // magic + switch header.Type { + case PackageHeartbeat: + byteArray = append(byteArray, DubboRequestHeartbeatHeader[:]...) + case PackageRequest_TwoWay: + byteArray = append(byteArray, DubboRequestHeaderBytesTwoWay[:]...) + default: + byteArray = append(byteArray, DubboRequestHeaderBytes[:]...) + } + + // serialization id, two way flag, event, request/response flag + // SerialID is id of serialization approach in java dubbo + byteArray[2] |= header.SerialID & SERIAL_MASK + // request id + binary.BigEndian.PutUint64(byteArray[4:], uint64(header.ID)) + + encoder := hessian.NewEncoder() + encoder.Append(byteArray[:HEADER_LENGTH]) + + ////////////////////////////////////////// + // body + ////////////////////////////////////////// + if hb { + encoder.Encode(nil) + goto END + } + + // dubbo version + path + version + method + encoder.Encode(DEFAULT_DUBBO_PROTOCOL_VERSION) + encoder.Encode(service.Path) + encoder.Encode(service.Version) + encoder.Encode(service.Method) + + // args = args type list + args value list + if types, err = getArgsTypeList(args); err != nil { + return nil, perrors.Wrapf(err, " PackRequest(args:%+v)", args) + } + encoder.Encode(types) + for _, v := range args { + encoder.Encode(v) + } + + request.Attachments[PATH_KEY] = service.Path + request.Attachments[VERSION_KEY] = service.Version + if len(service.Group) > 0 { + request.Attachments[GROUP_KEY] = service.Group + } + if len(service.Interface) > 0 { + request.Attachments[INTERFACE_KEY] = service.Interface + } + if service.Timeout != 0 { + request.Attachments[TIMEOUT_KEY] = strconv.Itoa(int(service.Timeout / time.Millisecond)) + } + + encoder.Encode(request.Attachments) + +END: + byteArray = encoder.Buffer() + pkgLen = len(byteArray) + if pkgLen > int(DEFAULT_LEN) { // 8M + return nil, perrors.Errorf("Data length %d too large, max payload %d", pkgLen, DEFAULT_LEN) + } + // byteArray{body length} + binary.BigEndian.PutUint32(byteArray[12:], uint32(pkgLen-HEADER_LENGTH)) + return byteArray, nil +} + +// hessian decode request body +func unpackRequestBody(decoder *hessian.Decoder, reqObj interface{}) error { + + if decoder == nil { + return perrors.Errorf("@decoder is nil") + } + + req, ok := reqObj.([]interface{}) + if !ok { + return perrors.Errorf("@reqObj is not of type: []interface{}") + } + if len(req) < 7 { + return perrors.New("length of @reqObj should be 7") + } + + var ( + err error + dubboVersion, target, serviceVersion, method, argsTypes interface{} + args []interface{} + ) + + dubboVersion, err = decoder.Decode() + if err != nil { + return perrors.WithStack(err) + } + req[0] = dubboVersion + + target, err = decoder.Decode() + if err != nil { + return perrors.WithStack(err) + } + req[1] = target + + serviceVersion, err = decoder.Decode() + if err != nil { + return perrors.WithStack(err) + } + req[2] = serviceVersion + + method, err = decoder.Decode() + if err != nil { + return perrors.WithStack(err) + } + req[3] = method + + argsTypes, err = decoder.Decode() + if err != nil { + return perrors.WithStack(err) + } + req[4] = argsTypes + + ats := DescRegex.FindAllString(argsTypes.(string), -1) + var arg interface{} + for i := 0; i < len(ats); i++ { + arg, err = decoder.Decode() + if err != nil { + return perrors.WithStack(err) + } + args = append(args, arg) + } + req[5] = args + + attachments, err := decoder.Decode() + if err != nil { + return perrors.WithStack(err) + } + if v, ok := attachments.(map[interface{}]interface{}); ok { + v[DUBBO_VERSION_KEY] = dubboVersion + req[6] = ToMapStringInterface(v) + return nil + } + + return perrors.Errorf("get wrong attachments: %+v", attachments) +} + +func ToMapStringInterface(origin map[interface{}]interface{}) map[string]interface{} { + dest := make(map[string]interface{}, len(origin)) + for k, v := range origin { + if kv, ok := k.(string); ok { + if v == nil { + dest[kv] = "" + continue + } + dest[kv] = v + } + } + return dest +} diff --git a/protocol/dubbo/hessian2/hessian_request_test.go b/protocol/dubbo/hessian2/hessian_request_test.go new file mode 100644 index 0000000000..98d5f2399c --- /dev/null +++ b/protocol/dubbo/hessian2/hessian_request_test.go @@ -0,0 +1,158 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package hessian2 + +import ( + "reflect" + "strconv" + "testing" + "time" +) + +import ( + hessian "github.com/apache/dubbo-go-hessian2" + "github.com/stretchr/testify/assert" +) + +type TestEnumGender hessian.JavaEnum + +const ( + MAN hessian.JavaEnum = iota + WOMAN +) + +var genderName = map[hessian.JavaEnum]string{ + MAN: "MAN", + WOMAN: "WOMAN", +} + +var genderValue = map[string]hessian.JavaEnum{ + "MAN": MAN, + "WOMAN": WOMAN, +} + +func (g TestEnumGender) JavaClassName() string { + return "com.ikurento.test.TestEnumGender" +} + +func (g TestEnumGender) String() string { + s, ok := genderName[hessian.JavaEnum(g)] + if ok { + return s + } + + return strconv.Itoa(int(g)) +} + +func (g TestEnumGender) EnumValue(s string) hessian.JavaEnum { + v, ok := genderValue[s] + if ok { + return v + } + + return hessian.InvalidJavaEnum +} + +func TestPackRequest(t *testing.T) { + bytes, err := packRequest(Service{ + Path: "test", + Interface: "ITest", + Version: "v1.0", + Method: "test", + Timeout: time.Second * 10, + }, DubboHeader{ + SerialID: 0, + Type: PackageRequest, + ID: 123, + }, []interface{}{1, 2}) + + assert.Nil(t, err) + + if bytes != nil { + t.Logf("pack request: %s", string(bytes)) + } +} + +func TestGetArgsTypeList(t *testing.T) { + type Test struct{} + str, err := getArgsTypeList([]interface{}{nil, 1, []int{2}, true, []bool{false}, "a", []string{"b"}, Test{}, &Test{}, []Test{}, map[string]Test{}, TestEnumGender(MAN)}) + assert.NoError(t, err) + assert.Equal(t, "VJ[JZ[ZLjava/lang/String;[Ljava/lang/String;Ljava/lang/Object;Ljava/lang/Object;[Ljava/lang/Object;Ljava/util/Map;Lcom/ikurento/test/TestEnumGender;", str) +} + +func TestDescRegex(t *testing.T) { + results := DescRegex.FindAllString("Ljava/lang/String;", -1) + assert.Equal(t, 1, len(results)) + assert.Equal(t, "Ljava/lang/String;", results[0]) + + results = DescRegex.FindAllString("Ljava/lang/String;I", -1) + assert.Equal(t, 2, len(results)) + assert.Equal(t, "Ljava/lang/String;", results[0]) + assert.Equal(t, "I", results[1]) + + results = DescRegex.FindAllString("ILjava/lang/String;", -1) + assert.Equal(t, 2, len(results)) + assert.Equal(t, "I", results[0]) + assert.Equal(t, "Ljava/lang/String;", results[1]) + + results = DescRegex.FindAllString("ILjava/lang/String;IZ", -1) + assert.Equal(t, 4, len(results)) + assert.Equal(t, "I", results[0]) + assert.Equal(t, "Ljava/lang/String;", results[1]) + assert.Equal(t, "I", results[2]) + assert.Equal(t, "Z", results[3]) + + results = DescRegex.FindAllString("[Ljava/lang/String;[I", -1) + assert.Equal(t, 2, len(results)) + assert.Equal(t, "[Ljava/lang/String;", results[0]) + assert.Equal(t, "[I", results[1]) +} + +func TestIssue192(t *testing.T) { + type args struct { + origin map[interface{}]interface{} + } + tests := []struct { + name string + args args + want map[string]interface{} + }{ + { + name: "not null", + args: args{ + origin: map[interface{}]interface{}{ + "1": nil, + "2": "3", + "": "", + }, + }, + want: map[string]interface{}{ + "1": "", + "2": "3", + "": "", + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := ToMapStringInterface(tt.args.origin); !reflect.DeepEqual(got, tt.want) { + t.Errorf("ToMapStringString() = %v, want %v", got, tt.want) + } + }) + } +} diff --git a/protocol/dubbo/hessian2/hessian_response.go b/protocol/dubbo/hessian2/hessian_response.go new file mode 100644 index 0000000000..982960ed87 --- /dev/null +++ b/protocol/dubbo/hessian2/hessian_response.go @@ -0,0 +1,377 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package hessian2 + +import ( + "encoding/binary" + "math" + "reflect" + "strconv" + "strings" +) + +import ( + hessian "github.com/apache/dubbo-go-hessian2" + "github.com/apache/dubbo-go-hessian2/java_exception" + perrors "github.com/pkg/errors" +) + +// DubboResponse dubbo response +type DubboResponse struct { + RspObj interface{} + Exception error + Attachments map[string]interface{} +} + +// NewResponse create a new DubboResponse +func NewResponse(rspObj interface{}, exception error, attachments map[string]interface{}) *DubboResponse { + if attachments == nil { + attachments = make(map[string]interface{}, 8) + } + return &DubboResponse{ + RspObj: rspObj, + Exception: exception, + Attachments: attachments, + } +} + +// EnsureResponse check body type, make sure it's a DubboResponse or package it as a DubboResponse +func EnsureResponse(body interface{}) *DubboResponse { + if res, ok := body.(*DubboResponse); ok { + return res + } + if exp, ok := body.(error); ok { + return NewResponse(nil, exp, nil) + } + return NewResponse(body, nil, nil) +} + +// https://github.com/apache/dubbo/blob/dubbo-2.7.1/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/codec/ExchangeCodec.java#L256 +// hessian encode response +func packResponse(header DubboHeader, ret interface{}) ([]byte, error) { + var ( + byteArray []byte + ) + + response := EnsureResponse(ret) + + hb := header.Type == PackageHeartbeat + + // magic + if hb { + byteArray = append(byteArray, DubboResponseHeartbeatHeader[:]...) + } else { + byteArray = append(byteArray, DubboResponseHeaderBytes[:]...) + } + // set serialID, identify serialization types, eg: fastjson->6, hessian2->2 + byteArray[2] |= header.SerialID & SERIAL_MASK + // response status + if header.ResponseStatus != 0 { + byteArray[3] = header.ResponseStatus + } + + // request id + binary.BigEndian.PutUint64(byteArray[4:], uint64(header.ID)) + + // body + encoder := hessian.NewEncoder() + encoder.Append(byteArray[:HEADER_LENGTH]) + + if header.ResponseStatus == Response_OK { + if hb { + encoder.Encode(nil) + } else { + atta := isSupportResponseAttachment(response.Attachments[DUBBO_VERSION_KEY]) + + var resWithException, resValue, resNullValue int32 + if atta { + resWithException = RESPONSE_WITH_EXCEPTION_WITH_ATTACHMENTS + resValue = RESPONSE_VALUE_WITH_ATTACHMENTS + resNullValue = RESPONSE_NULL_VALUE_WITH_ATTACHMENTS + } else { + resWithException = RESPONSE_WITH_EXCEPTION + resValue = RESPONSE_VALUE + resNullValue = RESPONSE_NULL_VALUE + } + + if response.Exception != nil { // throw error + encoder.Encode(resWithException) + if t, ok := response.Exception.(java_exception.Throwabler); ok { + encoder.Encode(t) + } else { + encoder.Encode(java_exception.NewThrowable(response.Exception.Error())) + } + } else { + if response.RspObj == nil { + encoder.Encode(resNullValue) + } else { + encoder.Encode(resValue) + encoder.Encode(response.RspObj) // result + } + } + + if atta { + encoder.Encode(response.Attachments) // attachments + } + } + } else { + if response.Exception != nil { // throw error + encoder.Encode(response.Exception.Error()) + } else { + encoder.Encode(response.RspObj) + } + } + + byteArray = encoder.Buffer() + byteArray = hessian.EncNull(byteArray) // if not, "java client" will throw exception "unexpected end of file" + pkgLen := len(byteArray) + if pkgLen > int(DEFAULT_LEN) { // 8M + return nil, perrors.Errorf("Data length %d too large, max payload %d", pkgLen, DEFAULT_LEN) + } + // byteArray{body length} + binary.BigEndian.PutUint32(byteArray[12:], uint32(pkgLen-HEADER_LENGTH)) + return byteArray, nil + +} + +// hessian decode response body +func unpackResponseBody(decoder *hessian.Decoder, resp interface{}) error { + // body + if decoder == nil { + return perrors.Errorf("@decoder is nil") + } + rspType, err := decoder.Decode() + if err != nil { + return perrors.WithStack(err) + } + + response := EnsureResponse(resp) + + switch rspType { + case RESPONSE_WITH_EXCEPTION, RESPONSE_WITH_EXCEPTION_WITH_ATTACHMENTS: + expt, err := decoder.Decode() + if err != nil { + return perrors.WithStack(err) + } + if rspType == RESPONSE_WITH_EXCEPTION_WITH_ATTACHMENTS { + attachments, err := decoder.Decode() + if err != nil { + return perrors.WithStack(err) + } + if v, ok := attachments.(map[interface{}]interface{}); ok { + atta := ToMapStringInterface(v) + response.Attachments = atta + } else { + return perrors.Errorf("get wrong attachments: %+v", attachments) + } + } + + if e, ok := expt.(error); ok { + response.Exception = e + } else { + response.Exception = perrors.Errorf("got exception: %+v", expt) + } + return nil + + case RESPONSE_VALUE, RESPONSE_VALUE_WITH_ATTACHMENTS: + rsp, err := decoder.Decode() + if err != nil { + return perrors.WithStack(err) + } + if rspType == RESPONSE_VALUE_WITH_ATTACHMENTS { + attachments, err := decoder.Decode() + if err != nil { + return perrors.WithStack(err) + } + if v, ok := attachments.(map[interface{}]interface{}); ok { + response.Attachments = ToMapStringInterface(v) + } else { + return perrors.Errorf("get wrong attachments: %+v", attachments) + } + } + + // If the return value is nil, + // we should consider it normal + if rsp == nil { + return nil + } + + return perrors.WithStack(ReflectResponse(rsp, response.RspObj)) + + case RESPONSE_NULL_VALUE, RESPONSE_NULL_VALUE_WITH_ATTACHMENTS: + if rspType == RESPONSE_NULL_VALUE_WITH_ATTACHMENTS { + attachments, err := decoder.Decode() + if err != nil { + return perrors.WithStack(err) + } + if v, ok := attachments.(map[interface{}]interface{}); ok { + atta := ToMapStringInterface(v) + response.Attachments = atta + } else { + return perrors.Errorf("get wrong attachments: %+v", attachments) + } + } + return nil + } + + return nil +} + +// CopySlice copy from inSlice to outSlice +func CopySlice(inSlice, outSlice reflect.Value) error { + if inSlice.IsNil() { + return perrors.New("@in is nil") + } + if inSlice.Kind() != reflect.Slice { + return perrors.Errorf("@in is not slice, but %v", inSlice.Kind()) + } + + for outSlice.Kind() == reflect.Ptr { + outSlice = outSlice.Elem() + } + + size := inSlice.Len() + outSlice.Set(reflect.MakeSlice(outSlice.Type(), size, size)) + + for i := 0; i < size; i++ { + inSliceValue := inSlice.Index(i) + if !inSliceValue.Type().AssignableTo(outSlice.Index(i).Type()) { + return perrors.Errorf("in element type [%s] can not assign to out element type [%s]", + inSliceValue.Type().String(), outSlice.Type().String()) + } + outSlice.Index(i).Set(inSliceValue) + } + + return nil +} + +// CopyMap copy from in map to out map +func CopyMap(inMapValue, outMapValue reflect.Value) error { + if inMapValue.IsNil() { + return perrors.New("@in is nil") + } + if !inMapValue.CanInterface() { + return perrors.New("@in's Interface can not be used.") + } + if inMapValue.Kind() != reflect.Map { + return perrors.Errorf("@in is not map, but %v", inMapValue.Kind()) + } + + outMapType := hessian.UnpackPtrType(outMapValue.Type()) + hessian.SetValue(outMapValue, reflect.MakeMap(outMapType)) + + outKeyType := outMapType.Key() + + outMapValue = hessian.UnpackPtrValue(outMapValue) + outValueType := outMapValue.Type().Elem() + + for _, inKey := range inMapValue.MapKeys() { + inValue := inMapValue.MapIndex(inKey) + + if !inKey.Type().AssignableTo(outKeyType) { + return perrors.Errorf("in Key:{type:%s, value:%#v} can not assign to out Key:{type:%s} ", + inKey.Type().String(), inKey, outKeyType.String()) + } + if !inValue.Type().AssignableTo(outValueType) { + return perrors.Errorf("in Value:{type:%s, value:%#v} can not assign to out value:{type:%s}", + inValue.Type().String(), inValue, outValueType.String()) + } + outMapValue.SetMapIndex(inKey, inValue) + } + + return nil +} + +// ReflectResponse reflect return value +// TODO response object should not be copied again to another object, it should be the exact type of the object +func ReflectResponse(in interface{}, out interface{}) error { + if in == nil { + return perrors.Errorf("@in is nil") + } + + if out == nil { + return perrors.Errorf("@out is nil") + } + if reflect.TypeOf(out).Kind() != reflect.Ptr { + return perrors.Errorf("@out should be a pointer") + } + + inValue := hessian.EnsurePackValue(in) + outValue := hessian.EnsurePackValue(out) + + outType := outValue.Type().String() + if outType == "interface {}" || outType == "*interface {}" { + hessian.SetValue(outValue, inValue) + return nil + } + + switch inValue.Type().Kind() { + case reflect.Slice, reflect.Array: + return CopySlice(inValue, outValue) + case reflect.Map: + return CopyMap(inValue, outValue) + default: + hessian.SetValue(outValue, inValue) + } + + return nil +} + +var versionInt = make(map[string]int) + +// https://github.com/apache/dubbo/blob/dubbo-2.7.1/dubbo-common/src/main/java/org/apache/dubbo/common/Version.java#L96 +// isSupportResponseAttachment is for compatibility among some dubbo version +func isSupportResponseAttachment(ver interface{}) bool { + version, ok := ver.(string) + if !ok || len(version) == 0 { + return false + } + + v, ok := versionInt[version] + if !ok { + v = version2Int(version) + if v == -1 { + return false + } + } + + if v >= 2001000 && v <= 2060200 { // 2.0.10 ~ 2.6.2 + return false + } + return v >= LOWEST_VERSION_FOR_RESPONSE_ATTACHMENT +} + +func version2Int(ver interface{}) int { + version, ok := ver.(string) + if !ok || len(version) == 0 { + return 0 + } + var v = 0 + varr := strings.Split(version, ".") + length := len(varr) + for key, value := range varr { + v0, err := strconv.Atoi(value) + if err != nil { + return -1 + } + v += v0 * int(math.Pow10((length-key-1)*2)) + } + if length == 3 { + return v * 100 + } + return v +} diff --git a/protocol/dubbo/hessian2/hessian_response_test.go b/protocol/dubbo/hessian2/hessian_response_test.go new file mode 100644 index 0000000000..f5c84baa90 --- /dev/null +++ b/protocol/dubbo/hessian2/hessian_response_test.go @@ -0,0 +1,225 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package hessian2 + +import ( + "reflect" + "testing" +) + +import ( + hessian "github.com/apache/dubbo-go-hessian2" + "github.com/stretchr/testify/assert" +) + +func doTestReflectResponse(t *testing.T, in interface{}, out interface{}) { + err := ReflectResponse(in, out) + if err != nil { + t.Error(err) + t.FailNow() + } + + result := hessian.UnpackPtrValue(reflect.ValueOf(out)).Interface() + + equal := reflect.DeepEqual(in, result) + if !equal { + t.Errorf("expect [%v]: %v, but got [%v]: %v", reflect.TypeOf(in), in, reflect.TypeOf(result), result) + } +} + +func TestReflectResponse(t *testing.T) { + var b bool + doTestReflectResponse(t, true, &b) + doTestReflectResponse(t, false, &b) + + var i int + doTestReflectResponse(t, 123, &i) + doTestReflectResponse(t, 234, &i) + + var i16 int16 + doTestReflectResponse(t, int16(456), &i16) + + var i64 int64 + doTestReflectResponse(t, int64(789), &i64) + + var s string + doTestReflectResponse(t, "hello world", &s) + + type rr struct { + Name string + Num int + } + + var r1 rr + doTestReflectResponse(t, rr{"dubbogo", 32}, &r1) + + // ------ map test ------- + m1 := make(map[interface{}]interface{}) + var m1r map[interface{}]interface{} + m1["hello"] = "world" + m1[1] = "go" + m1["dubbo"] = 666 + doTestReflectResponse(t, m1, &m1r) + + m2 := make(map[string]string) + var m2r map[string]string + m2["hello"] = "world" + m2["dubbo"] = "666" + doTestReflectResponse(t, m2, &m2r) + + m3 := make(map[string]rr) + var m3r map[string]rr + m3["dubbo"] = rr{"hello", 123} + m3["go"] = rr{"world", 456} + doTestReflectResponse(t, m3, &m3r) + + // ------ slice test ------- + s1 := []string{"abc", "def", "hello", "world"} + var s1r []string + doTestReflectResponse(t, s1, &s1r) + + s2 := []rr{rr{"dubbo", 666}, rr{"go", 999}} + var s2r []rr + doTestReflectResponse(t, s2, &s2r) + + s3 := []interface{}{rr{"dubbo", 666}, 123, "hello"} + var s3r []interface{} + doTestReflectResponse(t, s3, &s3r) + + // ------ interface test ------- + in1 := []interface{}{rr{"dubbo", 666}, 123, "hello"} + var inr1 *interface{} + doTestReflectResponse(t, in1, reflect.New(reflect.TypeOf(inr1).Elem()).Interface()) + + in2 := make(map[string]rr) + var inr2 map[string]rr + m3["dubbo"] = rr{"hello", 123} + m3["go"] = rr{"world", 456} + doTestReflectResponse(t, in2, &inr2) +} + +// separately test copy normal map to map[interface{}]interface{} +func TestCopyMap(t *testing.T) { + type rr struct { + Name string + Num int + } + + m3 := make(map[string]rr) + var m3r map[interface{}]interface{} + r1 := rr{"hello", 123} + r2 := rr{"world", 456} + m3["dubbo"] = r1 + m3["go"] = r2 + + err := ReflectResponse(m3, &m3r) + if err != nil { + t.Error(err) + t.FailNow() + } + + assert.Equal(t, 2, len(m3r)) + + rr1, ok := m3r["dubbo"] + assert.True(t, ok) + assert.True(t, reflect.DeepEqual(r1, rr1)) + + rr2, ok := m3r["go"] + assert.True(t, ok) + assert.True(t, reflect.DeepEqual(r2, rr2)) +} + +// separately test copy normal slice to []interface{} +func TestCopySlice(t *testing.T) { + type rr struct { + Name string + Num int + } + + r1 := rr{"hello", 123} + r2 := rr{"world", 456} + + s1 := []rr{r1, r2} + var s1r []interface{} + + err := ReflectResponse(s1, &s1r) + if err != nil { + t.Error(err) + t.FailNow() + } + + assert.Equal(t, 2, len(s1r)) + assert.True(t, reflect.DeepEqual(r1, s1r[0])) + assert.True(t, reflect.DeepEqual(r2, s1r[1])) +} + +func TestIsSupportResponseAttachment(t *testing.T) { + is := isSupportResponseAttachment("2.X") + assert.False(t, is) + + is = isSupportResponseAttachment("2.0.10") + assert.False(t, is) + + is = isSupportResponseAttachment("2.5.3") + assert.False(t, is) + + is = isSupportResponseAttachment("2.6.2") + assert.False(t, is) + + is = isSupportResponseAttachment("1.5.5") + assert.False(t, is) + + is = isSupportResponseAttachment("0.0.0") + assert.False(t, is) + + is = isSupportResponseAttachment("2.0.2") + assert.True(t, is) + + is = isSupportResponseAttachment("2.7.2") + assert.True(t, is) +} + +func TestVersion2Int(t *testing.T) { + v := version2Int("2.1.3") + assert.Equal(t, 2010300, v) + + v = version2Int("22.11.33") + assert.Equal(t, 22113300, v) + + v = version2Int("222.111.333") + assert.Equal(t, 223143300, v) + + v = version2Int("220.110.333") + assert.Equal(t, 221133300, v) + + v = version2Int("229.119.333") + assert.Equal(t, 230223300, v) + + v = version2Int("2222.1111.3333") + assert.Equal(t, 2233443300, v) + + v = version2Int("2.11") + assert.Equal(t, 211, v) + + v = version2Int("2.1.3.4") + assert.Equal(t, 2010304, v) + + v = version2Int("2.1.3.4.5") + assert.Equal(t, 201030405, v) + +} diff --git a/protocol/dubbo/listener.go b/protocol/dubbo/listener.go index a17b282fdf..180fd176f9 100644 --- a/protocol/dubbo/listener.go +++ b/protocol/dubbo/listener.go @@ -28,8 +28,6 @@ import ( import ( "github.com/apache/dubbo-getty" - "github.com/apache/dubbo-go-hessian2" - "github.com/opentracing/opentracing-go" perrors "github.com/pkg/errors" ) @@ -38,6 +36,7 @@ import ( "github.com/apache/dubbo-go/common/constant" "github.com/apache/dubbo-go/common/logger" "github.com/apache/dubbo-go/protocol" + "github.com/apache/dubbo-go/protocol/dubbo/hessian2" "github.com/apache/dubbo-go/protocol/invocation" ) @@ -105,8 +104,8 @@ func (h *RpcClientHandler) OnMessage(session getty.Session, pkg interface{}) { return } - if p.Header.Type&hessian.PackageHeartbeat != 0x00 { - if p.Header.Type&hessian.PackageResponse != 0x00 { + if p.Header.Type&hessian2.PackageHeartbeat != 0x00 { + if p.Header.Type&hessian2.PackageResponse != 0x00 { logger.Debugf("get rpc heartbeat response{header: %#v, body: %#v}", p.Header, p.Body) if p.Err != nil { logger.Errorf("rpc heartbeat response{error: %#v}", p.Err) @@ -114,8 +113,8 @@ func (h *RpcClientHandler) OnMessage(session getty.Session, pkg interface{}) { h.conn.pool.rpcClient.removePendingResponse(SequenceType(p.Header.ID)) } else { logger.Debugf("get rpc heartbeat request{header: %#v, service: %#v, body: %#v}", p.Header, p.Service, p.Body) - p.Header.ResponseStatus = hessian.Response_OK - reply(session, p, hessian.PackageHeartbeat) + p.Header.ResponseStatus = hessian2.Response_OK + reply(session, p, hessian2.PackageHeartbeat) } return } @@ -229,24 +228,24 @@ func (h *RpcServerHandler) OnMessage(session getty.Session, pkg interface{}) { logger.Errorf("illegal package{%#v}", pkg) return } - p.Header.ResponseStatus = hessian.Response_OK + p.Header.ResponseStatus = hessian2.Response_OK // heartbeat - if p.Header.Type&hessian.PackageHeartbeat != 0x00 { + if p.Header.Type&hessian2.PackageHeartbeat != 0x00 { logger.Debugf("get rpc heartbeat request{header: %#v, service: %#v, body: %#v}", p.Header, p.Service, p.Body) - reply(session, p, hessian.PackageHeartbeat) + reply(session, p, hessian2.PackageHeartbeat) return } twoway := true // not twoway - if p.Header.Type&hessian.PackageRequest_TwoWay == 0x00 { + if p.Header.Type&hessian2.PackageRequest_TwoWay == 0x00 { twoway = false } defer func() { if e := recover(); e != nil { - p.Header.ResponseStatus = hessian.Response_SERVER_ERROR + p.Header.ResponseStatus = hessian2.Response_SERVER_ERROR if err, ok := e.(error); ok { logger.Errorf("OnMessage panic: %+v", perrors.WithStack(err)) p.Body = perrors.WithStack(err) @@ -261,7 +260,7 @@ func (h *RpcServerHandler) OnMessage(session getty.Session, pkg interface{}) { if !twoway { return } - reply(session, p, hessian.PackageResponse) + reply(session, p, hessian2.PackageResponse) } }() @@ -274,14 +273,14 @@ func (h *RpcServerHandler) OnMessage(session getty.Session, pkg interface{}) { if exporter == nil { err := fmt.Errorf("don't have this exporter, key: %s", u.ServiceKey()) logger.Errorf(err.Error()) - p.Header.ResponseStatus = hessian.Response_OK + p.Header.ResponseStatus = hessian2.Response_OK p.Body = err - reply(session, p, hessian.PackageResponse) + reply(session, p, hessian2.PackageResponse) return } invoker := exporter.(protocol.Exporter).GetInvoker() if invoker != nil { - attachments := p.Body.(map[string]interface{})["attachments"].(map[string]string) + attachments := p.Body.(map[string]interface{})["attachments"].(map[string]interface{}) attachments[constant.LOCAL_ADDR] = session.LocalAddr() attachments[constant.REMOTE_ADDR] = session.RemoteAddr() @@ -292,19 +291,19 @@ func (h *RpcServerHandler) OnMessage(session getty.Session, pkg interface{}) { result := invoker.Invoke(ctx, inv) if err := result.Error(); err != nil { - p.Header.ResponseStatus = hessian.Response_OK - p.Body = hessian.NewResponse(nil, err, result.Attachments()) + p.Header.ResponseStatus = hessian2.Response_OK + p.Body = hessian2.NewResponse(nil, err, result.Attachments()) } else { res := result.Result() - p.Header.ResponseStatus = hessian.Response_OK - p.Body = hessian.NewResponse(res, nil, result.Attachments()) + p.Header.ResponseStatus = hessian2.Response_OK + p.Body = hessian2.NewResponse(res, nil, result.Attachments()) } } if !twoway { return } - reply(session, p, hessian.PackageResponse) + reply(session, p, hessian2.PackageResponse) } // OnCron notified when RPC server session got any message in cron job @@ -340,17 +339,16 @@ func rebuildCtx(inv *invocation.RPCInvocation) context.Context { ctx := context.Background() // actually, if user do not use any opentracing framework, the err will not be nil. - spanCtx, err := opentracing.GlobalTracer().Extract(opentracing.TextMap, - opentracing.TextMapCarrier(inv.Attachments())) + spanCtx, err := extractTraceCtx(inv) if err == nil { ctx = context.WithValue(ctx, constant.TRACING_REMOTE_SPAN_CTX, spanCtx) } return ctx } -func reply(session getty.Session, req *DubboPackage, tp hessian.PackageType) { +func reply(session getty.Session, req *DubboPackage, tp hessian2.PackageType) { resp := &DubboPackage{ - Header: hessian.DubboHeader{ + Header: hessian2.DubboHeader{ SerialID: req.Header.SerialID, Type: tp, ID: req.Header.ID, @@ -358,7 +356,7 @@ func reply(session getty.Session, req *DubboPackage, tp hessian.PackageType) { }, } - if req.Header.Type&hessian.PackageRequest != 0x00 { + if req.Header.Type&hessian2.PackageRequest != 0x00 { resp.Body = req.Body } else { resp.Body = nil diff --git a/protocol/dubbo/listener_test.go b/protocol/dubbo/listener_test.go index 5f80981460..5ab73fd465 100644 --- a/protocol/dubbo/listener_test.go +++ b/protocol/dubbo/listener_test.go @@ -35,7 +35,7 @@ import ( // test rebuild the ctx func TestRebuildCtx(t *testing.T) { opentracing.SetGlobalTracer(mocktracer.New()) - 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) @@ -47,8 +47,7 @@ func TestRebuildCtx(t *testing.T) { span, ctx := opentracing.StartSpanFromContext(ctx, "Test-Client") - opentracing.GlobalTracer().Inject(span.Context(), opentracing.TextMap, - opentracing.TextMapCarrier(inv.Attachments())) + injectTraceCtx(span, inv) // rebuild the context success inv = invocation.NewRPCInvocation("MethodName", []interface{}{"OK", "Hello"}, attach) ctx = rebuildCtx(inv) diff --git a/protocol/dubbo/opentracing.go b/protocol/dubbo/opentracing.go new file mode 100644 index 0000000000..2dcd6a4d0d --- /dev/null +++ b/protocol/dubbo/opentracing.go @@ -0,0 +1,60 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package dubbo + +import ( + "github.com/opentracing/opentracing-go" +) +import ( + invocation_impl "github.com/apache/dubbo-go/protocol/invocation" +) + +func injectTraceCtx(currentSpan opentracing.Span, inv *invocation_impl.RPCInvocation) error { + // inject opentracing ctx + traceAttachments := filterContext(inv.Attachments()) + carrier := opentracing.TextMapCarrier(traceAttachments) + err := opentracing.GlobalTracer().Inject(currentSpan.Context(), opentracing.TextMap, carrier) + if err == nil { + fillTraceAttachments(inv.Attachments(), traceAttachments) + } + return err +} + +func extractTraceCtx(inv *invocation_impl.RPCInvocation) (opentracing.SpanContext, error) { + traceAttachments := filterContext(inv.Attachments()) + // actually, if user do not use any opentracing framework, the err will not be nil. + spanCtx, err := opentracing.GlobalTracer().Extract(opentracing.TextMap, + opentracing.TextMapCarrier(traceAttachments)) + return spanCtx, err +} + +func filterContext(attachments map[string]interface{}) map[string]string { + var traceAttchment = make(map[string]string) + for k, v := range attachments { + if r, ok := v.(string); ok { + traceAttchment[k] = r + } + } + return traceAttchment +} + +func fillTraceAttachments(attachments map[string]interface{}, traceAttachment map[string]string) { + for k, v := range traceAttachment { + attachments[k] = v + } +} diff --git a/protocol/dubbo/readwriter.go b/protocol/dubbo/readwriter.go index adc6311b0e..a7b37aae76 100644 --- a/protocol/dubbo/readwriter.go +++ b/protocol/dubbo/readwriter.go @@ -24,7 +24,6 @@ import ( import ( "github.com/apache/dubbo-getty" - "github.com/apache/dubbo-go-hessian2" perrors "github.com/pkg/errors" ) @@ -32,6 +31,7 @@ import ( "github.com/apache/dubbo-go/common" "github.com/apache/dubbo-go/common/constant" "github.com/apache/dubbo-go/common/logger" + "github.com/apache/dubbo-go/protocol/dubbo/hessian2" ) //////////////////////////////////////////// @@ -56,7 +56,7 @@ func (p *RpcClientPackageHandler) Read(ss getty.Session, data []byte) (interface err := pkg.Unmarshal(buf, p.client) if err != nil { originErr := perrors.Cause(err) - if originErr == hessian.ErrHeaderNotEnough || originErr == hessian.ErrBodyNotEnough { + if originErr == hessian2.ErrHeaderNotEnough || originErr == hessian2.ErrBodyNotEnough { return nil, 0, nil } @@ -65,12 +65,12 @@ func (p *RpcClientPackageHandler) Read(ss getty.Session, data []byte) (interface return nil, 0, perrors.WithStack(err) } - if pkg.Header.Type&hessian.PackageRequest == 0x00 { - pkg.Err = pkg.Body.(*hessian.Response).Exception - pkg.Body = NewResponse(pkg.Body.(*hessian.Response).RspObj, pkg.Body.(*hessian.Response).Attachments) + if pkg.Header.Type&hessian2.PackageRequest == 0x00 { + pkg.Err = pkg.Body.(*hessian2.DubboResponse).Exception + pkg.Body = NewResponse(pkg.Body.(*hessian2.DubboResponse).RspObj, pkg.Body.(*hessian2.DubboResponse).Attachments) } - return pkg, hessian.HEADER_LENGTH + pkg.Header.BodyLen, nil + return pkg, hessian2.HEADER_LENGTH + pkg.Header.BodyLen, nil } // Write encode @pkg. @@ -111,7 +111,7 @@ func (p *RpcServerPackageHandler) Read(ss getty.Session, data []byte) (interface err := pkg.Unmarshal(buf) if err != nil { originErr := perrors.Cause(err) - if originErr == hessian.ErrHeaderNotEnough || originErr == hessian.ErrBodyNotEnough { + if originErr == hessian2.ErrHeaderNotEnough || originErr == hessian2.ErrBodyNotEnough { return nil, 0, nil } @@ -120,13 +120,13 @@ func (p *RpcServerPackageHandler) Read(ss getty.Session, data []byte) (interface return nil, 0, perrors.WithStack(err) } - if pkg.Header.Type&hessian.PackageHeartbeat == 0x00 { + if pkg.Header.Type&hessian2.PackageHeartbeat == 0x00 { // convert params of request req := pkg.Body.([]interface{}) // length of body should be 7 if len(req) > 0 { var dubboVersion, argsTypes string var args []interface{} - var attachments map[string]string + var attachments map[string]interface{} if req[0] != nil { dubboVersion = req[0].(string) } @@ -146,18 +146,18 @@ func (p *RpcServerPackageHandler) Read(ss getty.Session, data []byte) (interface args = req[5].([]interface{}) } if req[6] != nil { - attachments = req[6].(map[string]string) + attachments = req[6].(map[string]interface{}) } - if pkg.Service.Path == "" && len(attachments[constant.PATH_KEY]) > 0 { - pkg.Service.Path = attachments[constant.PATH_KEY] + if pkg.Service.Path == "" && attachments[constant.PATH_KEY] != nil && len(attachments[constant.PATH_KEY].(string)) > 0 { + pkg.Service.Path = attachments[constant.PATH_KEY].(string) } - if _, ok := attachments[constant.INTERFACE_KEY]; ok { - pkg.Service.Interface = attachments[constant.INTERFACE_KEY] + if inter, ok := attachments[constant.INTERFACE_KEY]; ok && inter != nil { + pkg.Service.Interface = inter.(string) } else { pkg.Service.Interface = pkg.Service.Path } - if len(attachments[constant.GROUP_KEY]) > 0 { - pkg.Service.Group = attachments[constant.GROUP_KEY] + if attachments[constant.GROUP_KEY] != nil && len(attachments[constant.GROUP_KEY].(string)) > 0 { + pkg.Service.Group = attachments[constant.GROUP_KEY].(string) } pkg.Body = map[string]interface{}{ "dubboVersion": dubboVersion, @@ -169,7 +169,7 @@ func (p *RpcServerPackageHandler) Read(ss getty.Session, data []byte) (interface } } - return pkg, hessian.HEADER_LENGTH + pkg.Header.BodyLen, nil + return pkg, hessian2.HEADER_LENGTH + pkg.Header.BodyLen, nil } // Write encode @pkg. diff --git a/protocol/invocation.go b/protocol/invocation.go index 296ec0540c..452f619e2d 100644 --- a/protocol/invocation.go +++ b/protocol/invocation.go @@ -34,15 +34,16 @@ type Invocation interface { // Reply gets response of request Reply() interface{} // Attachments gets all attachments - Attachments() map[string]string - // AttachmentsByKey gets attachment by key , if nil then return default value + Attachments() map[string]interface{} + // AttachmentsByKey gets attachment by key , if nil then return default value. （It will be deprecated in the future） AttachmentsByKey(string, string) string + Attachment(string) interface{} // Attributes refers to dubbo 2.7.6. It is different from attachment. It is used in internal process. Attributes() map[string]interface{} // AttributeByKey gets attribute by key , if nil then return default value AttributeByKey(string, interface{}) interface{} // SetAttachments sets attribute by @key and @value. - SetAttachments(key string, value string) + SetAttachments(key string, value interface{}) // Invoker gets the invoker in current context. Invoker() Invoker } diff --git a/protocol/invocation/rpcinvocation.go b/protocol/invocation/rpcinvocation.go index c72e105d35..35d12965e8 100644 --- a/protocol/invocation/rpcinvocation.go +++ b/protocol/invocation/rpcinvocation.go @@ -39,7 +39,7 @@ type RPCInvocation struct { arguments []interface{} reply interface{} callBack interface{} - attachments map[string]string + attachments map[string]interface{} // Refer to dubbo 2.7.6. It is different from attachment. It is used in internal process. attributes map[string]interface{} invoker protocol.Invoker @@ -47,7 +47,7 @@ type RPCInvocation struct { } // NewRPCInvocation creates a RPC invocation. -func NewRPCInvocation(methodName string, arguments []interface{}, attachments map[string]string) *RPCInvocation { +func NewRPCInvocation(methodName string, arguments []interface{}, attachments map[string]interface{}) *RPCInvocation { return &RPCInvocation{ methodName: methodName, arguments: arguments, @@ -99,7 +99,7 @@ func (r *RPCInvocation) SetReply(reply interface{}) { } // Attachments gets all attachments of RPC. -func (r *RPCInvocation) Attachments() map[string]string { +func (r *RPCInvocation) Attachments() map[string]interface{} { return r.attachments } @@ -112,11 +112,25 @@ func (r *RPCInvocation) AttachmentsByKey(key string, defaultValue string) string } value, ok := r.attachments[key] if ok { - return value + return value.(string) } return defaultValue } +// Attachment returns the corresponding value from dubbo's attachment with the given key. +func (r *RPCInvocation) Attachment(key string) interface{} { + r.lock.RLock() + defer r.lock.RUnlock() + if r.attachments == nil { + return nil + } + value, ok := r.attachments[key] + if ok { + return value + } + return nil +} + // Attributes gets all attributes of RPC. func (r *RPCInvocation) Attributes() map[string]interface{} { return r.attributes @@ -134,11 +148,11 @@ func (r *RPCInvocation) AttributeByKey(key string, defaultValue interface{}) int } // SetAttachments sets attribute by @key and @value. -func (r *RPCInvocation) SetAttachments(key string, value string) { +func (r *RPCInvocation) SetAttachments(key string, value interface{}) { r.lock.Lock() defer r.lock.Unlock() if r.attachments == nil { - r.attachments = make(map[string]string) + r.attachments = make(map[string]interface{}) } r.attachments[key] = value } @@ -221,7 +235,7 @@ func WithCallBack(callBack interface{}) option { } // WithAttachments creates option with @attachments. -func WithAttachments(attachments map[string]string) option { +func WithAttachments(attachments map[string]interface{}) option { return func(invo *RPCInvocation) { invo.attachments = attachments } diff --git a/protocol/jsonrpc/server.go b/protocol/jsonrpc/server.go index 29eba0223e..9755a481fd 100644 --- a/protocol/jsonrpc/server.go +++ b/protocol/jsonrpc/server.go @@ -343,7 +343,7 @@ func serveRequest(ctx context.Context, header map[string]string, body []byte, co exporter, _ := jsonrpcProtocol.ExporterMap().Load(path) invoker := exporter.(*JsonrpcExporter).GetInvoker() if invoker != nil { - result := invoker.Invoke(ctx, invocation.NewRPCInvocation(methodName, args, map[string]string{ + result := invoker.Invoke(ctx, invocation.NewRPCInvocation(methodName, args, map[string]interface{}{ constant.PATH_KEY: path, constant.VERSION_KEY: codec.req.Version})) if err := result.Error(); err != nil { diff --git a/protocol/rest/server/rest_server.go b/protocol/rest/server/rest_server.go index fbd6fb7ad9..d9542bb876 100644 --- a/protocol/rest/server/rest_server.go +++ b/protocol/rest/server/rest_server.go @@ -111,7 +111,7 @@ func GetRouteFunc(invoker protocol.Invoker, methodConfig *rest_config.RestMethod logger.Errorf("[Go Restful] WriteErrorString error:%v", err) } } - result := invoker.Invoke(context.Background(), invocation.NewRPCInvocation(methodConfig.MethodName, args, make(map[string]string))) + result := invoker.Invoke(context.Background(), invocation.NewRPCInvocation(methodConfig.MethodName, args, make(map[string]interface{}))) if result.Error() != nil { err = resp.WriteError(http.StatusInternalServerError, result.Error()) if err != nil { diff --git a/protocol/result.go b/protocol/result.go index 2a33be612f..a36b16d1cc 100644 --- a/protocol/result.go +++ b/protocol/result.go @@ -28,13 +28,14 @@ type Result interface { // Result gets invoker result. Result() interface{} // SetAttachments replaces the existing attachments with the specified param. - SetAttachments(map[string]string) + SetAttachments(map[string]interface{}) // Attachments gets all attachments - Attachments() map[string]string + Attachments() map[string]interface{} + // AddAttachment adds the specified map to existing attachments in this instance. - AddAttachment(string, string) + AddAttachment(string, interface{}) // Attachment gets attachment by key with default value. - Attachment(string, string) string + Attachment(string, interface{}) interface{} } ///////////////////////////// @@ -43,7 +44,7 @@ type Result interface { // RPCResult is default RPC result. type RPCResult struct { - Attrs map[string]string + Attrs map[string]interface{} Err error Rest interface{} } @@ -69,22 +70,22 @@ func (r *RPCResult) Result() interface{} { } // SetAttachments replaces the existing attachments with the specified param. -func (r *RPCResult) SetAttachments(attr map[string]string) { +func (r *RPCResult) SetAttachments(attr map[string]interface{}) { r.Attrs = attr } // Attachments gets all attachments -func (r *RPCResult) Attachments() map[string]string { +func (r *RPCResult) Attachments() map[string]interface{} { return r.Attrs } // AddAttachment adds the specified map to existing attachments in this instance. -func (r *RPCResult) AddAttachment(key, value string) { +func (r *RPCResult) AddAttachment(key string, value interface{}) { r.Attrs[key] = value } // Attachment gets attachment by key with default value. -func (r *RPCResult) Attachment(key, defaultValue string) string { +func (r *RPCResult) Attachment(key string, defaultValue interface{}) interface{} { v, ok := r.Attrs[key] if !ok { v = defaultValue diff --git a/test/integrate/dubbo/go-client/go.mod b/test/integrate/dubbo/go-client/go.mod index 4708eb1f0f..b0be45ae9c 100644 --- a/test/integrate/dubbo/go-client/go.mod +++ b/test/integrate/dubbo/go-client/go.mod @@ -1,3 +1,7 @@ module github.com/apache/dubbo-go/test/integrate/dubbo/go-client +require ( + github.com/apache/dubbo-go-hessian2 v1.6.0-rc1.0.20200906044240-6c1fb5c3bd44 +) + go 1.13 diff --git a/test/integrate/dubbo/go-server/go.mod b/test/integrate/dubbo/go-server/go.mod index 9e1162327d..6c530f6a59 100644 --- a/test/integrate/dubbo/go-server/go.mod +++ b/test/integrate/dubbo/go-server/go.mod @@ -1,3 +1,7 @@ module github.com/apache/dubbo-go/test/integrate/dubbo/go-server +require ( + github.com/apache/dubbo-go-hessian2 v1.6.0-rc1.0.20200906044240-6c1fb5c3bd44 +) + go 1.13