diff --git a/.github/workflows/github-actions.yml b/.github/workflows/github-actions.yml index 48a3feba4b..40ef43c8c1 100644 --- a/.github/workflows/github-actions.yml +++ b/.github/workflows/github-actions.yml @@ -53,21 +53,26 @@ jobs: else go get -v -t -d ./... fi - - - name: Verify - run: | - make verify - - name: Integrate Test + - name: gofmt run: | - chmod +x integrate_test.sh && ./integrate_test.sh - + go fmt ./... && git checkout -- go.mod && git status && [[ -z `git status -s` ]] + # diff -u <(echo -n) <(gofmt -d -s .) + - name: Install go ci lint run: curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(go env GOPATH)/bin v1.27.0 - name: Run Linter run: golangci-lint run --timeout=10m -v + - name: Verify + run: | + make verify + + - name: Integrate Test + run: | + chmod +x integrate_test.sh && ./integrate_test.sh + - name: Post Coverage run: bash <(curl -s https://codecov.io/bash) diff --git a/.golangci.yml b/.golangci.yml new file mode 100644 index 0000000000..6f214f445a --- /dev/null +++ b/.golangci.yml @@ -0,0 +1,63 @@ +linters-settings: + govet: + check-shadowing: true + golint: + min-confidence: 0 + gocyclo: + min-complexity: 10 + maligned: + suggest-new: true + dupl: + threshold: 100 + goconst: + min-len: 2 + min-occurrences: 2 + depguard: + list-type: blacklist + packages: + # logging is allowed only by logutils.Log, logrus + # is allowed to use only in logutils package + - github.com/sirupsen/logrus + misspell: + locale: US + lll: + line-length: 140 + goimports: + local-prefixes: github.com/golangci/golangci-lint + gocritic: + enabled-tags: + - performance + - style + - experimental + disabled-checks: + - wrapperFunc + +linters: + disable-all: true + enable: + - govet + - staticcheck + - ineffassign + - misspell + +run: + skip-dirs: + - test/testdata_etc + - pkg/golinters/goanalysis/(checker|passes) + +issues: + exclude-rules: + - text: "weak cryptographic primitive" + linters: + - gosec + - linters: + - staticcheck + text: "SA1019:" + +# golangci.com configuration +# https://github.com/golangci/golangci/wiki/Configuration +service: + golangci-lint-version: 1.15.x # use the fixed version to not introduce new linters unexpectedly + prepare: + - echo "here I can run custom commands, but no preparation needed for this repo" + diff --git a/.travis.yml b/.travis.yml deleted file mode 100644 index 4f79ecf3bc..0000000000 --- a/.travis.yml +++ /dev/null @@ -1,28 +0,0 @@ -dist: trusty -sudo: required - -# define the dependence env -language: go -os: - - linux -go: - - "1.13" -services: - - docker -env: - - GO111MODULE=on -install: true - -# define ci-stage -script: - - go fmt ./... && [[ -z `git status -s` ]] - # license-check - - make verify - # integrate-test - - chmod +x integrate_test.sh && ./integrate_test.sh - -after_success: - - bash <(curl -s https://codecov.io/bash) - -notifications: - webhooks: https://oapi.dingtalk.com/robot/send?access_token=f5d6237f2c79db584e75604f7f88db1ce1673c8c0e98451217b28fde791e1d4f diff --git a/cluster/cluster_impl/failback_cluster_invoker.go b/cluster/cluster_impl/failback_cluster_invoker.go index 7654e42ecb..5e0d133a67 100644 --- a/cluster/cluster_impl/failback_cluster_invoker.go +++ b/cluster/cluster_impl/failback_cluster_invoker.go @@ -77,8 +77,7 @@ func (invoker *failbackClusterInvoker) tryTimerTaskProc(ctx context.Context, ret invoked = append(invoked, retryTask.lastInvoker) retryInvoker := invoker.doSelect(retryTask.loadbalance, retryTask.invocation, retryTask.invokers, invoked) - var result protocol.Result - result = retryInvoker.Invoke(ctx, retryTask.invocation) + result := retryInvoker.Invoke(ctx, retryTask.invocation) if result.Error() != nil { retryTask.lastInvoker = retryInvoker invoker.checkRetry(retryTask, result.Error()) diff --git a/cluster/router/chain/chain.go b/cluster/router/chain/chain.go index 8b98acb582..fccce838bb 100644 --- a/cluster/router/chain/chain.go +++ b/cluster/router/chain/chain.go @@ -195,7 +195,7 @@ func (c *RouterChain) copyInvokerIfNecessary(cache *InvokerCache) []protocol.Inv func (c *RouterChain) buildCache() { origin := c.loadCache() invokers := c.copyInvokerIfNecessary(origin) - if invokers == nil || len(invokers) == 0 { + if len(invokers) == 0 { return } diff --git a/cluster/router/chain/chain_test.go b/cluster/router/chain/chain_test.go index b21990b08c..1bb7124a4f 100644 --- a/cluster/router/chain/chain_test.go +++ b/cluster/router/chain/chain_test.go @@ -76,8 +76,10 @@ conditions: _, err = z.Conn.Set(path, []byte(testyml), 0) assert.NoError(t, err) - defer ts.Stop() - defer z.Close() + defer func() { + _ = ts.Stop() + z.Close() + }() zkUrl, _ := common.NewURL(fmt.Sprintf(zkFormat, localIP, ts.Servers[0].Port)) configuration, err := extension.GetConfigCenterFactory(zk).GetDynamicConfiguration(zkUrl) @@ -128,11 +130,15 @@ conditions: _, err = z.Conn.Set(path, []byte(testyml), 0) assert.NoError(t, err) - defer ts.Stop() - defer z.Close() + defer func() { + _ = ts.Stop() + assert.NoError(t, err) + z.Close() + }() zkUrl, _ := common.NewURL(fmt.Sprintf(zkFormat, localIP, ts.Servers[0].Port)) configuration, err := extension.GetConfigCenterFactory(zk).GetDynamicConfiguration(zkUrl) + assert.NoError(t, err) config.GetEnvInstance().SetDynamicConfiguration(configuration) chain, err := NewRouterChain(getConditionRouteUrl(applicationKey)) @@ -154,11 +160,15 @@ conditions: func TestRouterChainRoute(t *testing.T) { ts, z, _, err := zookeeper.NewMockZookeeperClient("test", 15*time.Second) - defer ts.Stop() - defer z.Close() + defer func() { + err = ts.Stop() + assert.NoError(t, err) + z.Close() + }() zkUrl, _ := common.NewURL(fmt.Sprintf(zkFormat, localIP, ts.Servers[0].Port)) configuration, err := extension.GetConfigCenterFactory(zk).GetDynamicConfiguration(zkUrl) + assert.NoError(t, err) config.GetEnvInstance().SetDynamicConfiguration(configuration) chain, err := NewRouterChain(getConditionRouteUrl(applicationKey)) @@ -198,11 +208,15 @@ conditions: _, err = z.Conn.Set(path, []byte(testyml), 0) assert.NoError(t, err) - defer ts.Stop() - defer z.Close() + defer func() { + _ = ts.Stop() + assert.NoError(t, err) + z.Close() + }() zkUrl, _ := common.NewURL(fmt.Sprintf(zkFormat, localIP, ts.Servers[0].Port)) configuration, err := extension.GetConfigCenterFactory(zk).GetDynamicConfiguration(zkUrl) + assert.NoError(t, err) config.GetEnvInstance().SetDynamicConfiguration(configuration) chain, err := NewRouterChain(getConditionRouteUrl(applicationKey)) @@ -224,12 +238,17 @@ conditions: func TestRouterChainRouteNoRoute(t *testing.T) { ts, z, _, err := zookeeper.NewMockZookeeperClient("test", 15*time.Second) - defer ts.Stop() - defer z.Close() + assert.Nil(t, err) + defer func() { + _ = ts.Stop() + assert.NoError(t, err) + z.Close() + }() zkUrl, _ := common.NewURL(fmt.Sprintf(zkFormat, localIP, ts.Servers[0].Port)) configuration, err := extension.GetConfigCenterFactory(zk).GetDynamicConfiguration(zkUrl) config.GetEnvInstance().SetDynamicConfiguration(configuration) + assert.Nil(t, err) chain, err := NewRouterChain(getConditionNoRouteUrl(applicationKey)) assert.Nil(t, err) diff --git a/cluster/router/condition/app_router_test.go b/cluster/router/condition/app_router_test.go index 879abc5cc8..c8bc43eb33 100644 --- a/cluster/router/condition/app_router_test.go +++ b/cluster/router/condition/app_router_test.go @@ -67,8 +67,11 @@ conditions: _, err = z.Conn.Set(routerPath, []byte(testYML), 0) assert.NoError(t, err) - defer ts.Stop() - defer z.Close() + defer func() { + err = ts.Stop() + assert.NoError(t, err) + z.Close() + }() zkUrl, _ := common.NewURL(fmt.Sprintf(zkFormat, routerLocalIP, ts.Servers[0].Port)) configuration, err := extension.GetConfigCenterFactory(routerZk).GetDynamicConfiguration(zkUrl) @@ -115,8 +118,11 @@ conditions: _, err = z.Conn.Set(routerPath, []byte(testYML), 0) assert.NoError(t, err) - defer ts.Stop() - defer z.Close() + defer func() { + err = ts.Stop() + assert.NoError(t, err) + z.Close() + }() zkUrl, _ := common.NewURL(fmt.Sprintf(zkFormat, routerLocalIP, ts.Servers[0].Port)) configuration, err := extension.GetConfigCenterFactory(routerZk).GetDynamicConfiguration(zkUrl) @@ -154,8 +160,11 @@ conditions: _, err = z.Conn.Set(routerPath, []byte(testYML), 0) assert.NoError(t, err) - defer ts.Stop() - defer z.Close() + defer func() { + err = ts.Stop() + assert.NoError(t, err) + z.Close() + }() zkUrl, _ := common.NewURL(fmt.Sprintf(zkFormat, routerLocalIP, ts.Servers[0].Port)) configuration, err := extension.GetConfigCenterFactory(routerZk).GetDynamicConfiguration(zkUrl) diff --git a/cluster/router/condition/file_test.go b/cluster/router/condition/file_test.go index 441e91ee5a..9035e95592 100644 --- a/cluster/router/condition/file_test.go +++ b/cluster/router/condition/file_test.go @@ -110,12 +110,14 @@ func TestParseServiceRouterKey(t *testing.T) { assert.Equal(t, "", grp) assert.Equal(t, "mock-service", srv) assert.Equal(t, "", ver) + assert.NoError(t, err) testString = "/mock-service:" grp, srv, ver, err = parseServiceRouterKey(testString) assert.Equal(t, "", grp) assert.Equal(t, "mock-service", srv) assert.Equal(t, "", ver) + assert.NoError(t, err) testString = "grp:mock-service:123" grp, srv, ver, err = parseServiceRouterKey(testString) @@ -129,4 +131,5 @@ func TestParseServiceRouterKey(t *testing.T) { assert.Equal(t, "", grp) assert.Equal(t, "", srv) assert.Equal(t, "", ver) + assert.NoError(t, err) } diff --git a/cluster/router/condition/listenable_router.go b/cluster/router/condition/listenable_router.go index 0b47310dbf..102d766a8a 100644 --- a/cluster/router/condition/listenable_router.go +++ b/cluster/router/condition/listenable_router.go @@ -47,8 +47,8 @@ type listenableRouter struct { conditionRouters []*ConditionRouter routerRule *RouterRule url *common.URL - force bool - priority int64 + //force bool + priority int64 } // RouterRule Get RouterRule instance from listenableRouter diff --git a/cluster/router/tag/file.go b/cluster/router/tag/file.go index 94daf1508e..a51ae69cf2 100644 --- a/cluster/router/tag/file.go +++ b/cluster/router/tag/file.go @@ -41,7 +41,7 @@ type FileTagRouter struct { router *tagRouter routerRule *RouterRule url *common.URL - force bool + //force bool } // NewFileTagRouter Create file tag router instance with content (from config file) diff --git a/cluster/router/tag/router_rule.go b/cluster/router/tag/router_rule.go index c0a2d763ae..512d8f1079 100644 --- a/cluster/router/tag/router_rule.go +++ b/cluster/router/tag/router_rule.go @@ -84,9 +84,9 @@ func (t *RouterRule) getTagNames() []string { return result } -func (t *RouterRule) hasTag(tag string) bool { - return len(t.TagNameToAddresses[tag]) > 0 -} +//func (t *RouterRule) hasTag(tag string) bool { +// return len(t.TagNameToAddresses[tag]) > 0 +//} func (t *RouterRule) getAddressToTagNames() map[string][]string { return t.AddressToTagNames @@ -96,10 +96,10 @@ func (t *RouterRule) getTagNameToAddresses() map[string][]string { return t.TagNameToAddresses } -func (t *RouterRule) getTags() []Tag { - return t.Tags -} +//func (t *RouterRule) getTags() []Tag { +// return t.Tags +//} -func (t *RouterRule) setTags(tags []Tag) { - t.Tags = tags -} +//func (t *RouterRule) setTags(tags []Tag) { +// t.Tags = tags +//} diff --git a/cluster/router/tag/tag_router.go b/cluster/router/tag/tag_router.go index c7f53047c1..984ecb4eef 100644 --- a/cluster/router/tag/tag_router.go +++ b/cluster/router/tag/tag_router.go @@ -53,7 +53,7 @@ type addrMetadata struct { // application name application string // is rule a runtime rule - ruleRuntime bool + //ruleRuntime bool // is rule a force rule ruleForce bool // is rule a valid rule @@ -227,7 +227,7 @@ func (c *tagRouter) Pool(invokers []protocol.Invoker) (router.AddrPool, router.A // fetchRuleIfNecessary fetches, parses rule and register listener for the further change func (c *tagRouter) fetchRuleIfNecessary(invokers []protocol.Invoker) { - if invokers == nil || len(invokers) == 0 { + if len(invokers) == 0 { return } diff --git a/cluster/router/tag/tag_router_test.go b/cluster/router/tag/tag_router_test.go index 20b511ddbd..26c4b8352b 100644 --- a/cluster/router/tag/tag_router_test.go +++ b/cluster/router/tag/tag_router_test.go @@ -237,7 +237,7 @@ func TestRouteBeijingInvoker(t *testing.T) { type DynamicTagRouter struct { suite.Suite - rule *RouterRule + //rule *RouterRule route *tagRouter zkClient *zookeeper.ZookeeperClient diff --git a/common/config/environment.go b/common/config/environment.go index 44cdd1fca1..c5651a3354 100644 --- a/common/config/environment.go +++ b/common/config/environment.go @@ -34,8 +34,8 @@ import ( // We just have config center configuration which can override configuration in consumer.yaml & provider.yaml. // But for add these features in future ,I finish the environment struct following Environment class in java. type Environment struct { - configCenterFirst bool - externalConfigs sync.Map + configCenterFirst bool + //externalConfigs sync.Map externalConfigMap sync.Map appExternalConfigMap sync.Map dynamicConfiguration config_center.DynamicConfiguration diff --git a/common/constant/key.go b/common/constant/key.go index d020b9d40b..50aea81371 100644 --- a/common/constant/key.go +++ b/common/constant/key.go @@ -17,6 +17,8 @@ package constant +type DubboCtxKey string + const ( ASYNC_KEY = "async" // it's value should be "true" or "false" of string type ) @@ -47,8 +49,8 @@ const ( PORT_KEY = "port" PROTOCOL_KEY = "protocol" PATH_SEPARATOR = "/" - DUBBO_KEY = "dubbo" - SSL_ENABLED_KEY = "ssl-enabled" + //DUBBO_KEY = "dubbo" + SSL_ENABLED_KEY = "ssl-enabled" ) const ( @@ -92,7 +94,7 @@ const ( ) const ( - DUBBOGO_CTX_KEY = "dubbogo-ctx" + DUBBOGO_CTX_KEY = DubboCtxKey("dubbogo-ctx") ) const ( @@ -198,11 +200,9 @@ const ( ) const ( - TRACING_REMOTE_SPAN_CTX = "tracing.remote.span.ctx" + TRACING_REMOTE_SPAN_CTX = DubboCtxKey("tracing.remote.span.ctx") ) -type AttachmentCtxKey string - // Use for router module const ( // ConditionRouterName Specify file condition router name @@ -240,7 +240,7 @@ const ( Tagkey = "dubbo.tag" // Attachment key in context in invoker - AttachmentKey = AttachmentCtxKey("attachment") + AttachmentKey = DubboCtxKey("attachment") ) const ( diff --git a/common/rpc_service.go b/common/rpc_service.go index 9f587f3a68..30d72c1420 100644 --- a/common/rpc_service.go +++ b/common/rpc_service.go @@ -106,10 +106,10 @@ func (m *MethodType) ReplyType() reflect.Type { return m.replyType } -// SuiteContext tranfers @ctx to reflect.Value type or get it from @m.ctxType. +// SuiteContext transfers @ctx to reflect.Value type or get it from @m.ctxType. func (m *MethodType) SuiteContext(ctx context.Context) reflect.Value { - if contextv := reflect.ValueOf(ctx); contextv.IsValid() { - return contextv + if ctxV := reflect.ValueOf(ctx); ctxV.IsValid() { + return ctxV } return reflect.Zero(m.ctxType) } diff --git a/common/rpc_service_test.go b/common/rpc_service_test.go index 048e623c64..e8bd3933da 100644 --- a/common/rpc_service_test.go +++ b/common/rpc_service_test.go @@ -27,6 +27,10 @@ import ( "github.com/stretchr/testify/assert" ) +import ( + "github.com/apache/dubbo-go/common/constant" +) + const ( referenceTestPath = "com.test.Path" referenceTestPathDistinct = "com.test.Path1" @@ -125,7 +129,7 @@ func TestServiceMapUnRegister(t *testing.T) { err = ServiceMap.UnRegister("", testProtocol, ServiceKey("TestService", "", "v0")) assert.EqualError(t, err, "no service for TestService:v0") - // succ + // success err = ServiceMap.UnRegister("TestService", testProtocol, ServiceKey("TestService", "", "v1")) assert.NoError(t, err) } @@ -133,16 +137,12 @@ func TestServiceMapUnRegister(t *testing.T) { func TestMethodTypeSuiteContext(t *testing.T) { mt := &MethodType{ctxType: reflect.TypeOf(context.TODO())} ctx := context.Background() - type ctxKey string - key := ctxKey("key") + key := constant.DubboCtxKey("key") ctx = context.WithValue(ctx, key, "value") assert.Equal(t, reflect.ValueOf(ctx), mt.SuiteContext(ctx)) - - assert.Equal(t, reflect.Zero(mt.ctxType), mt.SuiteContext(nil)) } func TestSuiteMethod(t *testing.T) { - s := &TestService{} method, ok := reflect.TypeOf(s).MethodByName("MethodOne") assert.True(t, ok) diff --git a/common/url.go b/common/url.go index 8faa0f08ce..ac49465346 100644 --- a/common/url.go +++ b/common/url.go @@ -90,9 +90,7 @@ type baseUrl struct { Location string // ip+port Ip string Port string - //url.Values is not safe map, add to avoid concurrent map read and map write error - paramsLock sync.RWMutex - params url.Values + PrimitiveURL string } @@ -116,6 +114,10 @@ type URL struct { noCopy noCopy baseUrl + //url.Values is not safe map, add to avoid concurrent map read and map write error + paramsLock sync.RWMutex + params url.Values + Path string // like /com.ikurento.dubbo.UserProvider Username string Password string @@ -415,6 +417,9 @@ func (c *URL) Service() string { func (c *URL) AddParam(key string, value string) { c.paramsLock.Lock() defer c.paramsLock.Unlock() + if c.params == nil { + c.params = url.Values{} + } c.params.Add(key, value) } @@ -433,6 +438,9 @@ func (c *URL) AddParamAvoidNil(key string, value string) { func (c *URL) SetParam(key string, value string) { c.paramsLock.Lock() defer c.paramsLock.Unlock() + if c.params == nil { + c.params = url.Values{} + } c.params.Set(key, value) } @@ -440,7 +448,9 @@ func (c *URL) SetParam(key string, value string) { func (c *URL) DelParam(key string) { c.paramsLock.Lock() defer c.paramsLock.Unlock() - c.params.Del(key) + if c.params != nil { + c.params.Del(key) + } } // ReplaceParams will replace the URL.params @@ -466,10 +476,15 @@ func (c *URL) RangeParams(f func(key, value string) bool) { func (c *URL) GetParam(s string, d string) string { c.paramsLock.RLock() defer c.paramsLock.RUnlock() - r := c.params.Get(s) + + var r string + if len(c.params) > 0 { + r = c.params.Get(s) + } if len(r) == 0 { r = d } + return r } diff --git a/common/url_test.go b/common/url_test.go index 9f413494c6..c645f1a046 100644 --- a/common/url_test.go +++ b/common/url_test.go @@ -68,12 +68,17 @@ func TestURL(t *testing.T) { "side=provider&timeout=3000×tamp=1556509797245") assert.NoError(t, err) + urlInst := URL{} + urlInst.noCopy.Lock() + urlInst.SetParam("hello", "world") + urlInst.noCopy.Unlock() + assert.Equal(t, "/com.ikurento.user.UserProvider", u.Path) assert.Equal(t, "127.0.0.1:20000", u.Location) assert.Equal(t, "dubbo", u.Protocol) assert.Equal(t, loopbackAddress, u.Ip) assert.Equal(t, "20000", u.Port) - assert.Equal(t, URL{}.Methods, u.Methods) + assert.Equal(t, urlInst.Methods, u.Methods) assert.Equal(t, "", u.Username) assert.Equal(t, "", u.Password) assert.Equal(t, "anyhost=true&application=BDTService&category=providers&default.timeout=10000&dubbo=dubbo-"+ @@ -156,7 +161,10 @@ func TestURLEqual(t *testing.T) { func TestURLGetParam(t *testing.T) { params := url.Values{} params.Set("key", "value") - u := URL{baseUrl: baseUrl{params: params}} + + u := URL{} + u.SetParams(params) + v := u.GetParam("key", "default") assert.Equal(t, "value", v) @@ -167,8 +175,11 @@ func TestURLGetParam(t *testing.T) { func TestURLGetParamInt(t *testing.T) { params := url.Values{} - params.Set("key", "") - u := URL{baseUrl: baseUrl{params: params}} + params.Set("key", "value") + + u := URL{} + u.SetParams(params) + v := u.GetParamInt("key", 1) assert.Equal(t, int64(1), v) @@ -180,7 +191,10 @@ func TestURLGetParamInt(t *testing.T) { func TestURLGetParamIntValue(t *testing.T) { params := url.Values{} params.Set("key", "0") - u := URL{baseUrl: baseUrl{params: params}} + + u := URL{} + u.SetParams(params) + v := u.GetParamInt("key", 1) assert.Equal(t, int64(0), v) @@ -192,7 +206,10 @@ func TestURLGetParamIntValue(t *testing.T) { func TestURLGetParamBool(t *testing.T) { params := url.Values{} params.Set("force", "true") - u := URL{baseUrl: baseUrl{params: params}} + + u := URL{} + u.SetParams(params) + v := u.GetParamBool("force", false) assert.Equal(t, true, v) @@ -205,7 +222,10 @@ func TestURLGetParamAndDecoded(t *testing.T) { rule := "host = 2.2.2.2,1.1.1.1,3.3.3.3 & host !=1.1.1.1 => host = 1.2.3.4" params := url.Values{} params.Set("rule", base64.URLEncoding.EncodeToString([]byte(rule))) - u := URL{baseUrl: baseUrl{params: params}} + + u := URL{} + u.SetParams(params) + v, _ := u.GetParamAndDecoded("rule") assert.Equal(t, rule, v) } @@ -242,7 +262,10 @@ func TestURLToMap(t *testing.T) { func TestURLGetMethodParamInt(t *testing.T) { params := url.Values{} params.Set("methods.GetValue.timeout", "3") - u := URL{baseUrl: baseUrl{params: params}} + + u := URL{} + u.SetParams(params) + v := u.GetMethodParamInt("GetValue", "timeout", 1) assert.Equal(t, int64(3), v) @@ -254,7 +277,10 @@ func TestURLGetMethodParamInt(t *testing.T) { func TestURLGetMethodParam(t *testing.T) { params := url.Values{} params.Set("methods.GetValue.timeout", "3s") - u := URL{baseUrl: baseUrl{params: params}} + + u := URL{} + u.SetParams(params) + v := u.GetMethodParam("GetValue", "timeout", "1s") assert.Equal(t, "3s", v) @@ -266,7 +292,10 @@ func TestURLGetMethodParam(t *testing.T) { func TestURLGetMethodParamBool(t *testing.T) { params := url.Values{} params.Set("methods.GetValue.async", "true") - u := URL{baseUrl: baseUrl{params: params}} + + u := URL{} + u.SetParams(params) + v := u.GetMethodParamBool("GetValue", "async", false) assert.Equal(t, true, v) diff --git a/config/base_config.go b/config/base_config.go index 0cc6eec26c..7cb25a5079 100644 --- a/config/base_config.go +++ b/config/base_config.go @@ -45,7 +45,7 @@ type BaseConfig struct { // application config ApplicationConfig *ApplicationConfig `yaml:"application" json:"application,omitempty" property:"application"` - prefix string + //prefix string fatherConfig interface{} EventDispatcherType string `default:"direct" yaml:"event_dispatcher_type" json:"event_dispatcher_type,omitempty"` MetricConfig *MetricConfig `yaml:"metrics" json:"metrics,omitempty"` diff --git a/config/base_config_test.go b/config/base_config_test.go index 9e5565fc11..566b49bea7 100644 --- a/config/base_config_test.go +++ b/config/base_config_test.go @@ -107,6 +107,7 @@ var baseMockRef = map[string]*ReferenceConfig{ func TestRefresh(t *testing.T) { c := &BaseConfig{} + c.fileStream = nil mockMap := getMockMap() mockMap["dubbo.shutdown.timeout"] = "12s" diff --git a/config/config_center_config.go b/config/config_center_config.go index 3bb856415d..940b7be951 100644 --- a/config/config_center_config.go +++ b/config/config_center_config.go @@ -18,10 +18,8 @@ package config import ( - "context" "net/url" "reflect" - "time" ) import ( @@ -46,7 +44,7 @@ import ( // // ConfigCenter has currently supported Zookeeper, Nacos, Etcd, Consul, Apollo type ConfigCenterConfig struct { - context context.Context + //context context.Context Protocol string `required:"true" yaml:"protocol" json:"protocol,omitempty"` Address string `yaml:"address" json:"address,omitempty"` Cluster string `yaml:"cluster" json:"cluster,omitempty"` @@ -60,7 +58,7 @@ type ConfigCenterConfig struct { AppId string `default:"dubbo" yaml:"app_id" json:"app_id,omitempty"` TimeoutStr string `yaml:"timeout" json:"timeout,omitempty"` RemoteRef string `required:"false" yaml:"remote_ref" json:"remote_ref,omitempty"` - timeout time.Duration + //timeout time.Duration } // UnmarshalYAML unmarshals the ConfigCenterConfig by @unmarshal function diff --git a/config/config_loader.go b/config/config_loader.go index ec591aad65..35910981d1 100644 --- a/config/config_loader.go +++ b/config/config_loader.go @@ -406,9 +406,9 @@ func GetBaseConfig() *BaseConfig { baseConfig = &BaseConfig{ MetricConfig: &MetricConfig{}, ConfigCenterConfig: &ConfigCenterConfig{}, - Remotes: make(map[string]*RemoteConfig, 0), + Remotes: make(map[string]*RemoteConfig), ApplicationConfig: &ApplicationConfig{}, - ServiceDiscoveries: make(map[string]*ServiceDiscoveryConfig, 0), + ServiceDiscoveries: make(map[string]*ServiceDiscoveryConfig), } } } diff --git a/config/config_loader_test.go b/config/config_loader_test.go index ef76bc0908..b9a025e8bc 100644 --- a/config/config_loader_test.go +++ b/config/config_loader_test.go @@ -27,7 +27,7 @@ import ( import ( "github.com/Workiva/go-datastructures/slice/skip" gxset "github.com/dubbogo/gost/container/set" - gxpage "github.com/dubbogo/gost/page" + gxpage "github.com/dubbogo/gost/hash/page" "github.com/stretchr/testify/assert" "go.uber.org/atomic" ) @@ -104,7 +104,8 @@ func TestLoad(t *testing.T) { conServices = map[string]common.RPCService{} proServices = map[string]common.RPCService{} - err := common.ServiceMap.UnRegister("com.MockService", "mock", common.ServiceKey("com.MockService", "huadong_idc", "1.0.0")) + err := common.ServiceMap.UnRegister("com.MockService", "mock", + common.ServiceKey("com.MockService", "huadong_idc", "1.0.0")) assert.Nil(t, err) consumerConfig = nil providerConfig = nil @@ -143,7 +144,8 @@ func TestLoadWithSingleReg(t *testing.T) { conServices = map[string]common.RPCService{} proServices = map[string]common.RPCService{} - err := common.ServiceMap.UnRegister("com.MockService", "mock", common.ServiceKey("com.MockService", "huadong_idc", "1.0.0")) + err := common.ServiceMap.UnRegister("com.MockService", "mock", + common.ServiceKey("com.MockService", "huadong_idc", "1.0.0")) assert.Nil(t, err) consumerConfig = nil providerConfig = nil @@ -183,7 +185,8 @@ func TestWithNoRegLoad(t *testing.T) { conServices = map[string]common.RPCService{} proServices = map[string]common.RPCService{} - err := common.ServiceMap.UnRegister("com.MockService", "mock", common.ServiceKey("com.MockService", "huadong_idc", "1.0.0")) + err := common.ServiceMap.UnRegister("com.MockService", "mock", + common.ServiceKey("com.MockService", "huadong_idc", "1.0.0")) assert.Nil(t, err) consumerConfig = nil providerConfig = nil diff --git a/config/service_config.go b/config/service_config.go index 8bc1b91ee5..fd49390aa2 100644 --- a/config/service_config.go +++ b/config/service_config.go @@ -176,7 +176,8 @@ func (c *ServiceConfig) Export() error { // registry the service reflect methods, err := common.ServiceMap.Register(c.InterfaceName, proto.Name, c.Group, c.Version, c.rpcService) if err != nil { - formatErr := perrors.Errorf("The service %v export the protocol %v error! Error message is %v.", c.InterfaceName, proto.Name, err.Error()) + formatErr := perrors.Errorf("The service %v export the protocol %v error! Error message is %v.", + c.InterfaceName, proto.Name, err.Error()) logger.Errorf(formatErr.Error()) return formatErr } diff --git a/config_center/apollo/impl.go b/config_center/apollo/impl.go index c69fc2f66b..5b74f5e150 100644 --- a/config_center/apollo/impl.go +++ b/config_center/apollo/impl.go @@ -146,9 +146,8 @@ func (c *apolloConfiguration) getAddressWithProtocolPrefix(url *common.URL) stri address := url.Location converted := address if len(address) != 0 { - reg := regexp.MustCompile("\\s+") - address = reg.ReplaceAllString(address, "") - parts := strings.Split(address, ",") + addr := regexp.MustCompile(`\s+`).ReplaceAllString(address, "") + parts := strings.Split(addr, ",") addrs := make([]string, 0) for _, part := range parts { addr := part diff --git a/config_center/apollo/impl_test.go b/config_center/apollo/impl_test.go index 38ef0082fd..3b2cb16840 100644 --- a/config_center/apollo/impl_test.go +++ b/config_center/apollo/impl_test.go @@ -143,7 +143,7 @@ func serviceConfigResponse(rw http.ResponseWriter, _ *http.Request) { // run mock config server func runMockConfigServer(handlerMap map[string]func(http.ResponseWriter, *http.Request), notifyHandler func(http.ResponseWriter, *http.Request)) *httptest.Server { - uriHandlerMap := make(map[string]func(http.ResponseWriter, *http.Request), 0) + uriHandlerMap := make(map[string]func(http.ResponseWriter, *http.Request)) for namespace, handler := range handlerMap { uri := fmt.Sprintf("/configs/%s/%s/%s", mockAppId, mockCluster, namespace) uriHandlerMap[uri] = handler diff --git a/config_center/apollo/listener.go b/config_center/apollo/listener.go index ace5ed0268..44d325582f 100644 --- a/config_center/apollo/listener.go +++ b/config_center/apollo/listener.go @@ -36,7 +36,7 @@ type apolloListener struct { // nolint func newApolloListener() *apolloListener { return &apolloListener{ - listeners: make(map[config_center.ConfigurationListener]struct{}, 0), + listeners: make(map[config_center.ConfigurationListener]struct{}), } } diff --git a/config_center/nacos/client.go b/config_center/nacos/client.go index c7d631c76f..1e96b36a4f 100644 --- a/config_center/nacos/client.go +++ b/config_center/nacos/client.go @@ -65,7 +65,7 @@ type option func(*options) type options struct { nacosName string - client *NacosClient + //client *NacosClient } // WithNacosName Set nacos name diff --git a/config_center/nacos/impl_test.go b/config_center/nacos/impl_test.go index 89917d14c2..b7bd94bf96 100644 --- a/config_center/nacos/impl_test.go +++ b/config_center/nacos/impl_test.go @@ -40,7 +40,7 @@ import ( // run mock config server func runMockConfigServer(configHandler func(http.ResponseWriter, *http.Request), configListenHandler func(http.ResponseWriter, *http.Request)) *httptest.Server { - uriHandlerMap := make(map[string]func(http.ResponseWriter, *http.Request), 0) + uriHandlerMap := make(map[string]func(http.ResponseWriter, *http.Request)) uriHandlerMap["/nacos/v1/cs/configs"] = configHandler uriHandlerMap["/nacos/v1/cs/configs/listener"] = configListenHandler diff --git a/config_center/zookeeper/impl.go b/config_center/zookeeper/impl.go index 17812e917f..f3b2235025 100644 --- a/config_center/zookeeper/impl.go +++ b/config_center/zookeeper/impl.go @@ -52,7 +52,7 @@ type zookeeperDynamicConfiguration struct { done chan struct{} client *zookeeper.ZookeeperClient - listenerLock sync.Mutex + //listenerLock sync.Mutex listener *zookeeper.ZkEventListener cacheListener *CacheListener parser parser.ConfigurationParser diff --git a/config_center/zookeeper/impl_test.go b/config_center/zookeeper/impl_test.go index bb0902766f..a518bb6ec0 100644 --- a/config_center/zookeeper/impl_test.go +++ b/config_center/zookeeper/impl_test.go @@ -100,9 +100,8 @@ func initZkData(group string, t *testing.T) (*zk.TestCluster, *zookeeperDynamicC func TestGetConfig(t *testing.T) { ts, reg := initZkData("dubbo", t) defer func() { - if err := ts.Stop(); err != nil { - t.Errorf("ts.Stop() = error: %v", err) - } + err := ts.Stop() + assert.NoError(t, err) }() configs, err := reg.GetProperties(dubboPropertyFileName, config_center.WithGroup("dubbo")) assert.NoError(t, err) @@ -122,7 +121,10 @@ func TestGetConfig(t *testing.T) { func TestAddListener(t *testing.T) { ts, reg := initZkData("", t) - defer ts.Stop() + defer func() { + err := ts.Stop() + assert.NoError(t, err) + }() listener := &mockDataListener{} reg.AddListener(dubboPropertyFileName, listener) listener.wg.Add(1) @@ -155,7 +157,10 @@ func TestAddListener(t *testing.T) { func TestRemoveListener(t *testing.T) { ts, reg := initZkData("", t) - defer ts.Stop() + defer func() { + err := ts.Stop() + assert.NoError(t, err) + }() listener := &mockDataListener{} reg.AddListener(dubboPropertyFileName, listener) listener.wg.Add(1) @@ -193,7 +198,10 @@ func TestZookeeperDynamicConfigurationPublishConfig(t *testing.T) { customGroup := "Custom Group" key := "myKey" ts, zk := initZkData(config_center.DEFAULT_GROUP, t) - defer ts.Stop() + defer func() { + err := ts.Stop() + assert.NoError(t, err) + }() err := zk.PublishConfig(key, customGroup, value) assert.Nil(t, err) result, err := zk.GetInternalProperty("myKey", config_center.WithGroup(customGroup)) diff --git a/filter/filter_impl/active_filter_test.go b/filter/filter_impl/active_filter_test.go index 9f024fec34..2397503dc8 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]interface{}, 0)) + invoc := invocation.NewRPCInvocation("test", []interface{}{"OK"}, make(map[string]interface{})) url, _ := common.NewURL("dubbo://192.168.10.10:20000/com.ikurento.user.UserProvider") filter := ActiveFilter{} ctrl := gomock.NewController(t) diff --git a/filter/filter_impl/auth/sign_util.go b/filter/filter_impl/auth/sign_util.go index cb185376cb..4c12c762ed 100644 --- a/filter/filter_impl/auth/sign_util.go +++ b/filter/filter_impl/auth/sign_util.go @@ -37,7 +37,7 @@ func Sign(metadata, key string) string { // SignWithParams returns a signature with giving params and metadata. func SignWithParams(params []interface{}, metadata, key string) (string, error) { - if params == nil || len(params) == 0 { + if len(params) == 0 { return Sign(metadata, key), nil } diff --git a/filter/filter_impl/execute_limit_filter_test.go b/filter/filter_impl/execute_limit_filter_test.go index 2aebcaa8fa..682f8fa8b4 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]interface{}, 0)) + invoc := invocation.NewRPCInvocation(methodName, []interface{}{"OK"}, make(map[string]interface{})) 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]interface{}, 0)) + invoc := invocation.NewRPCInvocation(methodName, []interface{}{"OK"}, make(map[string]interface{})) 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]interface{}, 0)) + invoc := invocation.NewRPCInvocation(methodName, []interface{}{"OK"}, make(map[string]interface{})) 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 c2c1ef874d..b16956e01c 100644 --- a/filter/filter_impl/graceful_shutdown_filter_test.go +++ b/filter/filter_impl/graceful_shutdown_filter_test.go @@ -39,10 +39,8 @@ import ( ) func TestGenericFilterInvoke(t *testing.T) { - invoc := invocation.NewRPCInvocation("GetUser", []interface{}{"OK"}, make(map[string]interface{}, 0)) - - invokeUrl := common.NewURLWithOptions( - common.WithParams(url.Values{})) + invoc := invocation.NewRPCInvocation("GetUser", []interface{}{"OK"}, make(map[string]interface{})) + invokeUrl := common.NewURLWithOptions(common.WithParams(url.Values{})) shutdownFilter := extension.GetFilter(constant.PROVIDER_SHUTDOWN_FILTER).(*gracefulShutdownFilter) diff --git a/filter/filter_impl/seata_filter.go b/filter/filter_impl/seata_filter.go index 7722d2954f..b7b7b0e94b 100644 --- a/filter/filter_impl/seata_filter.go +++ b/filter/filter_impl/seata_filter.go @@ -23,6 +23,7 @@ import ( ) import ( + "github.com/apache/dubbo-go/common/constant" "github.com/apache/dubbo-go/common/extension" "github.com/apache/dubbo-go/common/logger" "github.com/apache/dubbo-go/filter" @@ -30,12 +31,12 @@ import ( ) const ( - SEATA = "seata" - SEATA_XID = "SEATA_XID" + SEATA = constant.DubboCtxKey("seata") + SEATA_XID = constant.DubboCtxKey("SEATA_XID") ) func init() { - extension.SetFilter(SEATA, getSeataFilter) + extension.SetFilter(string(SEATA), getSeataFilter) } // SeataFilter when use seata-golang, use this filter to transfer xid @@ -45,7 +46,7 @@ type SeataFilter struct{} // Invoke Get Xid by attachment key `SEATA_XID` func (sf *SeataFilter) Invoke(ctx context.Context, invoker protocol.Invoker, invocation protocol.Invocation) protocol.Result { logger.Infof("invoking seata filter.") - xid := invocation.AttachmentsByKey(SEATA_XID, "") + xid := invocation.AttachmentsByKey(string(SEATA_XID), "") if strings.TrimSpace(xid) != "" { logger.Debugf("Method: %v,Xid: %v", invocation.MethodName(), xid) return invoker.Invoke(context.WithValue(ctx, SEATA_XID, xid), invocation) diff --git a/filter/filter_impl/seata_filter_test.go b/filter/filter_impl/seata_filter_test.go index 45817e95cb..1705eba787 100644 --- a/filter/filter_impl/seata_filter_test.go +++ b/filter/filter_impl/seata_filter_test.go @@ -50,7 +50,7 @@ func TestSeataFilter_Invoke(t *testing.T) { filter := getSeataFilter() result := filter.Invoke(context.Background(), &testMockSeataInvoker{}, invocation.NewRPCInvocation("$echo", []interface{}{"OK"}, map[string]interface{}{ - SEATA_XID: "10.30.21.227:8091:2000047792", + string(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/sentinel_filter.go b/filter/filter_impl/sentinel_filter.go index b5f8b45876..f662db3563 100644 --- a/filter/filter_impl/sentinel_filter.go +++ b/filter/filter_impl/sentinel_filter.go @@ -204,8 +204,8 @@ const ( DefaultProviderPrefix = "dubbo:provider:" DefaultConsumerPrefix = "dubbo:consumer:" - MethodEntryKey = "$$sentinelMethodEntry" - InterfaceEntryKey = "$$sentinelInterfaceEntry" + MethodEntryKey = constant.DubboCtxKey("$$sentinelMethodEntry") + InterfaceEntryKey = constant.DubboCtxKey("$$sentinelInterfaceEntry") ) func getResourceName(invoker protocol.Invoker, invocation protocol.Invocation, prefix string) (interfaceResourceName, methodResourceName string) { diff --git a/filter/filter_impl/token_filter_test.go b/filter/filter_impl/token_filter_test.go index 9ef8c98d68..024ae2ae31 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]interface{}, 0) + attch := make(map[string]interface{}) 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]interface{}, 0) + attch := make(map[string]interface{}) 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]interface{}, 0) + attch := make(map[string]interface{}) 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]interface{}, 0) + attch := make(map[string]interface{}) 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 7435d9b92a..4ff0a232e4 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]interface{}, 0)) + invoc := invocation.NewRPCInvocation(methodName, []interface{}{"OK"}, make(map[string]interface{})) 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]interface{}, 0)) + invoc := invocation.NewRPCInvocation(methodName, []interface{}{"OK"}, make(map[string]interface{})) // 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]interface{}, 0)) + invoc := invocation.NewRPCInvocation(methodName, []interface{}{"OK"}, make(map[string]interface{})) 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]interface{}, 0)) + invoc := invocation.NewRPCInvocation(methodName, []interface{}{"OK"}, make(map[string]interface{})) 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 88e7781050..55a3a55fab 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]interface{}, 0) + attch := make(map[string]interface{}) 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]interface{}, 0) + attch := make(map[string]interface{}) 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]interface{}, 0) + attch := make(map[string]interface{}) result := tpsFilter.Invoke(context.Background(), protocol.NewBaseInvoker( diff --git a/filter/filter_impl/tracing_filter_test.go b/filter/filter_impl/tracing_filter_test.go index e159b7400d..bf96a7d064 100644 --- a/filter/filter_impl/tracing_filter_test.go +++ b/filter/filter_impl/tracing_filter_test.go @@ -57,6 +57,6 @@ func TestTracingFilterInvoke(t *testing.T) { tf.Invoke(ctx, invoker, inv) // has remote ctx - ctx = context.WithValue(context.Background(), constant.TRACING_REMOTE_SPAN_CTX, span.Context()) + ctx = context.WithValue(context.Background(), constant.DubboCtxKey(constant.TRACING_REMOTE_SPAN_CTX), span.Context()) tf.Invoke(ctx, invoker, inv) } diff --git a/go.mod b/go.mod index e45db5a638..0e5736c0a8 100644 --- a/go.mod +++ b/go.mod @@ -8,12 +8,12 @@ require ( github.com/Workiva/go-datastructures v1.0.52 github.com/afex/hystrix-go v0.0.0-20180502004556-fa1af6a1f4f5 github.com/alibaba/sentinel-golang v1.0.1 - github.com/apache/dubbo-getty v1.3.10 + github.com/apache/dubbo-getty v1.4.1 github.com/apache/dubbo-go-hessian2 v1.8.0 github.com/coreos/etcd v3.3.25+incompatible github.com/creasty/defaults v1.5.1 github.com/dubbogo/go-zookeeper v1.0.2 - github.com/dubbogo/gost v1.9.5 + github.com/dubbogo/gost v1.10.1 github.com/elazarl/go-bindata-assetfs v1.0.0 // indirect github.com/emicklei/go-restful/v3 v3.4.0 github.com/frankban/quicktest v1.4.1 // indirect diff --git a/go.sum b/go.sum index 79f2cda7c9..d69824eb7f 100644 --- a/go.sum +++ b/go.sum @@ -64,6 +64,8 @@ github.com/Shopify/sarama v1.19.0/go.mod h1:FVkBWblsNy7DGZRfXLU0O9RCGt5g3g3yEuWX github.com/Shopify/toxiproxy v2.1.4+incompatible/go.mod h1:OXgGpZ6Cli1/URJOF1DMxUHB2q5Ap20/P/eIdh4G0pI= github.com/StackExchange/wmi v0.0.0-20180116203802-5d049714c4a6 h1:fLjPD/aNc3UIOA6tDi6QXUemppXK3P9BI7mr2hd6gx8= github.com/StackExchange/wmi v0.0.0-20180116203802-5d049714c4a6/go.mod h1:3eOhrUMpNV+6aFIbp5/iudMxNCF27Vw2OZgy4xEx0Fg= +github.com/StackExchange/wmi v0.0.0-20190523213315-cbe66965904d h1:G0m3OIz70MZUWq3EgK3CesDbo8upS2Vm9/P3FtgI+Jk= +github.com/StackExchange/wmi v0.0.0-20190523213315-cbe66965904d/go.mod h1:3eOhrUMpNV+6aFIbp5/iudMxNCF27Vw2OZgy4xEx0Fg= github.com/VividCortex/gohistogram v1.0.0/go.mod h1:Pf5mBqqDxYaXu3hDrrU+w6nw50o/4+TcAqDqk/vUH7g= github.com/Workiva/go-datastructures v1.0.52 h1:PLSK6pwn8mYdaoaCZEMsXBpBotr4HHn9abU0yMQt0NI= github.com/Workiva/go-datastructures v1.0.52/go.mod h1:Z+F2Rca0qCsVYDS8z7bAGm8f3UkzuWYS/oBZz5a7VVA= @@ -80,8 +82,8 @@ github.com/alibaba/sentinel-golang v1.0.1 h1:WlhN0XUxRyfkiDc8TO6CcRrnakwFP9zFtvJ github.com/alibaba/sentinel-golang v1.0.1/go.mod h1:QsB99f/z35D2AiMrAWwgWE85kDTkBUIkcmPrRt+61NI= github.com/aliyun/alibaba-cloud-sdk-go v1.61.18 h1:zOVTBdCKFd9JbCKz9/nt+FovbjPFmb7mUnp8nH9fQBA= 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-getty v1.4.1 h1:M9yaFhemThQSWtRwmJNrxNuv7FzydlFx5EY8oq1v+lw= +github.com/apache/dubbo-getty v1.4.1/go.mod h1:ansXgKxxyhCOiQL29nO5ce1MDcEKmCyZuNR9oMs3hek= github.com/apache/dubbo-go-hessian2 v1.8.0 h1:+GJQHxWd/WUw2p4hbfCal/zjKvGVb8yJZzOke8IEazc= github.com/apache/dubbo-go-hessian2 v1.8.0/go.mod h1:7rEw9guWABQa6Aqb8HeZcsYPHsOS7XT1qtJvkmI6c5w= github.com/apache/thrift v0.12.0/go.mod h1:cp2SuWMxlEZw2r+iP2GNCdIi4C1qmUzdZFSVb+bacwQ= @@ -173,8 +175,9 @@ github.com/docker/spdystream v0.0.0-20160310174837-449fdfce4d96/go.mod h1:Qh8CwZ github.com/dubbogo/go-zookeeper v1.0.2 h1:xmEnPL8SlCe3/+J5ZR9e8qE35LmFVYe8VVpDakjNM4A= github.com/dubbogo/go-zookeeper v1.0.2/go.mod h1:fn6n2CAEer3novYgk9ULLwAjuV8/g4DdC2ENwRb6E+c= github.com/dubbogo/gost v1.9.0/go.mod h1:pPTjVyoJan3aPxBPNUX0ADkXjPibLo+/Ib0/fADXSG8= -github.com/dubbogo/gost v1.9.5 h1:UeG4y0O55lR3dzgdmCm/7bMWvpKrlpR7fsfKjrcXq/g= -github.com/dubbogo/gost v1.9.5/go.mod h1:QNM5RaeRdNWehUu8S0hUP5Qa8QUfGf6KH1JhqOVFvEI= +github.com/dubbogo/gost v1.10.1 h1:39kF9Cd5JOiMpmwG6dX1/aLWNFqFv9gHp8HrhzMmjLY= +github.com/dubbogo/gost v1.10.1/go.mod h1:+mQGS51XQEUWZP2JeGZTxJwipjRKtJO7Tr+FOg+72rI= +github.com/dubbogo/jsonparser v1.0.1/go.mod h1:tYAtpctvSP/tWw4MeelsowSPgXQRVHHWbqL6ynps8jU= github.com/dustin/go-humanize v0.0.0-20171111073723-bb3d318650d4/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk= github.com/dustin/go-humanize v1.0.0 h1:VSnTsYCnlFHaM2/igO1h6X3HA71jcobQuxemgkq4zYo= github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk= @@ -233,6 +236,8 @@ github.com/go-logfmt/logfmt v0.5.0/go.mod h1:wCYkCAKZfumFQihp8CzCvQ3paCTfi41vtzG github.com/go-logr/logr v0.1.0/go.mod h1:ixOQHD9gLJUVQQ2ZOR7zLEifBX6tGkNJF4QyIY7sIas= github.com/go-ole/go-ole v1.2.1 h1:2lOsA72HgjxAuMlKpFiCbHTvu44PIVkZ5hqm3RSdI/E= github.com/go-ole/go-ole v1.2.1/go.mod h1:7FAglXiTm7HKlQRDeOQ6ZNUHidzCWXuZWq/1dTyBNF8= +github.com/go-ole/go-ole v1.2.4 h1:nNBDSCOigTSiarFpYE9J/KtEA1IOW4CNeqT9TQDqCxI= +github.com/go-ole/go-ole v1.2.4/go.mod h1:XCwSNxSkXRo4vlyPy93sltvi/qJq0jqQhjqQNIwKuxM= github.com/go-openapi/jsonpointer v0.0.0-20160704185906-46af16f9f7b1/go.mod h1:+35s3my2LFTysnkMfxsJBAMHj/DoqoB9knIWoYG/Vk0= github.com/go-openapi/jsonreference v0.0.0-20160704190145-13c6e3589ad9/go.mod h1:W3Z9FmVs9qj+KR4zFKmDPGiLdk1D9Rlm7cyMvf57TTg= github.com/go-openapi/spec v0.0.0-20160808142527-6aced65f8501/go.mod h1:J8+jY1nAiCcj+friV/PDoE1/3eeccG9LYBs0tYvLOWc= @@ -690,6 +695,7 @@ github.com/sean-/pager v0.0.0-20180208200047-666be9bf53b5/go.mod h1:BeybITEsBEg6 github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 h1:nn5Wsu0esKSJiIVhscUtVbo7ada43DJhG55ua/hjS5I= github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc= github.com/shirou/gopsutil v0.0.0-20181107111621-48177ef5f880/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA= +github.com/shirou/gopsutil v3.20.11-0.20201116082039-2fb5da2f2449+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA= github.com/shirou/gopsutil v3.20.11+incompatible h1:LJr4ZQK4mPpIV5gOa4jCOKOGb4ty4DZO54I4FGqIpto= github.com/shirou/gopsutil v3.20.11+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA= github.com/shirou/w32 v0.0.0-20160930032740-bb4de0191aa4/go.mod h1:qsXQc7+bwAM3Q1u/4XEfrquwF8Lw7D7y5cD8CuHnfIc= diff --git a/metadata/mapping/dynamic/service_name_mapping_test.go b/metadata/mapping/dynamic/service_name_mapping_test.go index 2896b0fd4a..af21704b69 100644 --- a/metadata/mapping/dynamic/service_name_mapping_test.go +++ b/metadata/mapping/dynamic/service_name_mapping_test.go @@ -39,6 +39,7 @@ func TestDynamicConfigurationServiceNameMapping(t *testing.T) { dc, err := (&config_center.MockDynamicConfigurationFactory{ Content: appName, }).GetDynamicConfiguration(nil) + assert.NoError(t, err) config.GetApplicationConfig().Name = appName mapping := &DynamicConfigurationServiceNameMapping{dc: dc} diff --git a/metadata/report/consul/report_test.go b/metadata/report/consul/report_test.go index 13d0c419bc..51e93179a2 100644 --- a/metadata/report/consul/report_test.go +++ b/metadata/report/consul/report_test.go @@ -143,7 +143,9 @@ func (suite *consulMetadataReportTestSuite) testGetServiceDefinition() { func test1(t *testing.T) { consulAgent := consul.NewConsulAgent(t, 8500) - defer consulAgent.Shutdown() + defer func() { + _ = consulAgent.Shutdown() + }() url := newProviderRegistryUrl("localhost", 8500) mf := extension.GetMetadataReportFactory("consul") diff --git a/metadata/report/delegate/delegate_report_test.go b/metadata/report/delegate/delegate_report_test.go index 9c30ed9ffd..f60acf6e11 100644 --- a/metadata/report/delegate/delegate_report_test.go +++ b/metadata/report/delegate/delegate_report_test.go @@ -46,13 +46,10 @@ func TestMetadataReport_MetadataReportRetry(t *testing.T) { }) assert.NoError(t, err) retry.startRetryTask() - itsTime := time.After(2500 * time.Millisecond) - select { - case <-itsTime: - retry.scheduler.Clear() - assert.Equal(t, counter.Load(), int64(3)) - logger.Info("over") - } + <-time.After(2500 * time.Millisecond) + retry.scheduler.Clear() + assert.Equal(t, counter.Load(), int64(3)) + logger.Info("over") } func TestMetadataReport_MetadataReportRetryWithLimit(t *testing.T) { @@ -64,13 +61,10 @@ func TestMetadataReport_MetadataReportRetryWithLimit(t *testing.T) { }) assert.NoError(t, err) retry.startRetryTask() - itsTime := time.After(2500 * time.Millisecond) - select { - case <-itsTime: - retry.scheduler.Clear() - assert.Equal(t, counter.Load(), int64(2)) - logger.Info("over") - } + <-time.After(2500 * time.Millisecond) + retry.scheduler.Clear() + assert.Equal(t, counter.Load(), int64(2)) + logger.Info("over") } func mockNewMetadataReport(t *testing.T) *MetadataReport { diff --git a/metadata/report/etcd/report_test.go b/metadata/report/etcd/report_test.go index 28c0486982..59d0975ca6 100644 --- a/metadata/report/etcd/report_test.go +++ b/metadata/report/etcd/report_test.go @@ -82,8 +82,9 @@ func TestEtcdMetadataReport_CRUD(t *testing.T) { assert.Nil(t, err) serviceMi := newServiceMetadataIdentifier() - serviceUrl, _ := common.NewURL("registry://localhost:8848", common.WithParamsValue(constant.ROLE_KEY, strconv.Itoa(common.PROVIDER))) - metadataReport.SaveServiceMetadata(serviceMi, serviceUrl) + serviceUrl, err := common.NewURL("registry://localhost:8848", common.WithParamsValue(constant.ROLE_KEY, strconv.Itoa(common.PROVIDER))) + assert.Nil(t, err) + err = metadataReport.SaveServiceMetadata(serviceMi, serviceUrl) assert.Nil(t, err) subMi := newSubscribeMetadataIdentifier() diff --git a/metadata/service/inmemory/service_proxy.go b/metadata/service/inmemory/service_proxy.go index 7e14293a67..8b93aab9ac 100644 --- a/metadata/service/inmemory/service_proxy.go +++ b/metadata/service/inmemory/service_proxy.go @@ -39,8 +39,8 @@ import ( // this is the stub, or proxy // for now, only GetExportedURLs need to be implemented type MetadataServiceProxy struct { - invkr protocol.Invoker - golangServer bool + invkr protocol.Invoker + //golangServer bool } func (m *MetadataServiceProxy) GetExportedURLs(serviceInterface string, group string, version string, protocol string) ([]interface{}, error) { @@ -67,11 +67,7 @@ func (m *MetadataServiceProxy) GetExportedURLs(serviceInterface string, group st urlStrs := res.Result().(*[]interface{}) ret := make([]interface{}, 0, len(*urlStrs)) - - for _, s := range *urlStrs { - ret = append(ret, s) - } - return ret, nil + return append(ret, *urlStrs...), nil } func (m *MetadataServiceProxy) MethodMapper() map[string]string { diff --git a/metadata/service/inmemory/service_test.go b/metadata/service/inmemory/service_test.go index 40dfee5d15..e50cd6208a 100644 --- a/metadata/service/inmemory/service_test.go +++ b/metadata/service/inmemory/service_test.go @@ -95,13 +95,16 @@ func TestMetadataService(t *testing.T) { assert.Equal(t, 0, len(list4)) userProvider := &definition.UserProvider{} - common.ServiceMap.Register(serviceName, protocol, group, version, userProvider) - mts.PublishServiceDefinition(u) + _, err = common.ServiceMap.Register(serviceName, protocol, group, version, userProvider) + assert.NoError(t, err) + err = mts.PublishServiceDefinition(u) + assert.NoError(t, err) expected := "{\"CanonicalName\":\"com.ikurento.user.UserProvider\",\"CodeSource\":\"\"," + "\"Methods\":[{\"Name\":\"GetUser\",\"ParameterTypes\":[\"slice\"],\"ReturnType\":\"ptr\"," + "\"Parameters\":null}],\"Types\":null}" - def1, _ := mts.GetServiceDefinition(serviceName, group, version) + def1, err := mts.GetServiceDefinition(serviceName, group, version) assert.Equal(t, expected, def1) + assert.NoError(t, err) serviceKey := definition.ServiceDescriperBuild(serviceName, group, version) def2, err := mts.GetServiceDefinitionByServiceKey(serviceKey) assert.Equal(t, expected, def2) diff --git a/metadata/service/remote/service.go b/metadata/service/remote/service.go index ae1c344d27..d21cc88a3d 100644 --- a/metadata/service/remote/service.go +++ b/metadata/service/remote/service.go @@ -188,7 +188,7 @@ func (mts *MetadataService) RefreshMetadata(exportedRevision string, subscribedR logger.Errorf("Error occur when execute remote.MetadataService.RefreshMetadata, error message is %v+", err) return false, err } - if urls != nil && len(urls) > 0 { + if len(urls) > 0 { id := &identifier.SubscriberMetadataIdentifier{ MetadataIdentifier: identifier.MetadataIdentifier{ Application: config.GetApplicationConfig().Name, diff --git a/metadata/service/remote/service_proxy.go b/metadata/service/remote/service_proxy.go index 3199aa6dff..d0caa182b2 100644 --- a/metadata/service/remote/service_proxy.go +++ b/metadata/service/remote/service_proxy.go @@ -146,7 +146,7 @@ func newMetadataServiceProxy(ins registry.ServiceInstance) service.MetadataServi } func parse(key string) []string { - arr := make([]string, 3, 3) + arr := make([]string, 3) tmp := strings.SplitN(key, "/", 2) if len(tmp) > 1 { arr[0] = tmp[0] diff --git a/metadata/service/remote/service_proxy_test.go b/metadata/service/remote/service_proxy_test.go index 8bccbb8bbb..1899d02205 100644 --- a/metadata/service/remote/service_proxy_test.go +++ b/metadata/service/remote/service_proxy_test.go @@ -54,18 +54,28 @@ func TestMetadataServiceProxy_GetServiceDefinition(t *testing.T) { // in fact, we don't use them func TestMetadataServiceProxy(t *testing.T) { pxy := createProxy() - pxy.ServiceName() - pxy.PublishServiceDefinition(&common.URL{}) - pxy.Version() - pxy.GetSubscribedURLs() - pxy.UnsubscribeURL(&common.URL{}) - pxy.GetServiceDefinitionByServiceKey("any") - pxy.ExportURL(&common.URL{}) - pxy.SubscribeURL(&common.URL{}) - pxy.MethodMapper() - pxy.UnexportURL(&common.URL{}) - pxy.Reference() - pxy.RefreshMetadata(constant.ANY_VALUE, constant.ANY_VALUE) + _, err := pxy.ServiceName() + assert.NoError(t, err) + err = pxy.PublishServiceDefinition(&common.URL{}) + assert.NoError(t, err) + _, err = pxy.Version() + assert.NoError(t, err) + _, err = pxy.GetSubscribedURLs() + assert.NoError(t, err) + err = pxy.UnsubscribeURL(&common.URL{}) + assert.NoError(t, err) + _, err = pxy.GetServiceDefinitionByServiceKey("any") + assert.NoError(t, err) + _, err = pxy.ExportURL(&common.URL{}) + assert.NoError(t, err) + _, err = pxy.SubscribeURL(&common.URL{}) + assert.NoError(t, err) + _ = pxy.MethodMapper() + err = pxy.UnexportURL(&common.URL{}) + assert.NoError(t, err) + _ = pxy.Reference() + _, err = pxy.RefreshMetadata(constant.ANY_VALUE, constant.ANY_VALUE) + assert.NoError(t, err) } func createProxy() service.MetadataService { diff --git a/metadata/service/remote/service_test.go b/metadata/service/remote/service_test.go index 71586cc1dc..d6028152cc 100644 --- a/metadata/service/remote/service_test.go +++ b/metadata/service/remote/service_test.go @@ -95,7 +95,7 @@ func (metadataReport) GetServiceDefinition(*identifier.MetadataIdentifier) (stri func TestMetadataService(t *testing.T) { extension.SetMetadataReportFactory("mock", getMetadataReportFactory) - u, err := common.NewURL(fmt.Sprintf("mock://127.0.0.1:20000/?sync.report=true")) + u, err := common.NewURL("mock://127.0.0.1:20000/?sync.report=true") assert.NoError(t, err) instance.GetMetadataReportInstance(u) mts, err := newMetadataService() diff --git a/protocol/dubbo/dubbo_codec.go b/protocol/dubbo/dubbo_codec.go index ccecbee2e2..21376c3145 100644 --- a/protocol/dubbo/dubbo_codec.go +++ b/protocol/dubbo/dubbo_codec.go @@ -105,7 +105,7 @@ func (c *DubboCodec) EncodeRequest(request *remoting.Request) (*bytes.Buffer, er return pkg.Marshal() } -// encode heartbeart request +// encode heartbeat request func (c *DubboCodec) encodeHeartbeartReqeust(request *remoting.Request) (*bytes.Buffer, error) { header := impl.DubboHeader{ Type: impl.PackageHeartbeat, @@ -177,10 +177,7 @@ func (c *DubboCodec) Decode(data []byte) (remoting.DecodeResult, int, error) { } func (c *DubboCodec) isRequest(data []byte) bool { - if data[2]&byte(0x80) == 0x00 { - return false - } - return true + return data[2]&byte(0x80) != 0x00 } // decode request diff --git a/protocol/dubbo/dubbo_invoker_test.go b/protocol/dubbo/dubbo_invoker_test.go index c7a9a26975..fecb3b01ab 100644 --- a/protocol/dubbo/dubbo_invoker_test.go +++ b/protocol/dubbo/dubbo_invoker_test.go @@ -169,7 +169,7 @@ type ( } UserProvider struct { - user map[string]User + //user map[string]User } ) diff --git a/protocol/dubbo/dubbo_protocol.go b/protocol/dubbo/dubbo_protocol.go index 4f03b8aba0..1f7cd50187 100644 --- a/protocol/dubbo/dubbo_protocol.go +++ b/protocol/dubbo/dubbo_protocol.go @@ -222,13 +222,13 @@ func getExchangeClient(url *common.URL) *remoting.ExchangeClient { // Once we decided to transfer more context's key-value, we should change this. // now we only support rebuild the tracing context func rebuildCtx(inv *invocation.RPCInvocation) context.Context { - ctx := context.WithValue(context.Background(), "attachment", inv.Attachments()) + ctx := context.WithValue(context.Background(), constant.DubboCtxKey("attachment"), 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(filterContext(inv.Attachments()))) if err == nil { - ctx = context.WithValue(ctx, constant.TRACING_REMOTE_SPAN_CTX, spanCtx) + ctx = context.WithValue(ctx, constant.DubboCtxKey(constant.TRACING_REMOTE_SPAN_CTX), spanCtx) } return ctx } diff --git a/protocol/dubbo/hessian2/const.go b/protocol/dubbo/hessian2/const.go index 0c95390fbf..96e6eea8ea 100644 --- a/protocol/dubbo/hessian2/const.go +++ b/protocol/dubbo/hessian2/const.go @@ -75,10 +75,10 @@ const ( 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) + BC_LIST_VARIABLE = byte(0x55) + BC_LIST_FIXED = byte('V') + BC_LIST_VARIABLE_UNTYPED = byte(0x57) + BC_LIST_FIXED_UNTYPED = byte(0x58) BC_LIST_DIRECT = byte(0x70) BC_LIST_DIRECT_UNTYPED = byte(0x78) diff --git a/protocol/dubbo/hessian2/hessian_dubbo.go b/protocol/dubbo/hessian2/hessian_dubbo.go index 1afa4ec96e..5ffebde54b 100644 --- a/protocol/dubbo/hessian2/hessian_dubbo.go +++ b/protocol/dubbo/hessian2/hessian_dubbo.go @@ -99,7 +99,7 @@ func (h *HessianCodec) Write(service Service, header DubboHeader, body interface return packResponse(header, body) default: - return nil, perrors.Errorf("Unrecognised message type: %v", header.Type) + return nil, perrors.Errorf("Unrecognized message type: %v", header.Type) } // unreachable return nil, nil @@ -190,9 +190,9 @@ func (h *HessianCodec) ReadBody(rspObj interface{}) error { 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) + exception, exceptionErr := decoder.Decode() + if exceptionErr != nil { + return perrors.WithStack(exceptionErr) } rsp, ok := rspObj.(*DubboResponse) if !ok { diff --git a/protocol/dubbo/hessian2/hessian_request.go b/protocol/dubbo/hessian2/hessian_request.go index af603014c8..94aa34dee8 100644 --- a/protocol/dubbo/hessian2/hessian_request.go +++ b/protocol/dubbo/hessian2/hessian_request.go @@ -43,7 +43,7 @@ func getArgType(v interface{}) string { return "V" } - switch v.(type) { + switch v := v.(type) { // Serialized tags for base types case nil: return "V" @@ -220,23 +220,21 @@ func packRequest(service Service, header DubboHeader, req interface{}) ([]byte, // body ////////////////////////////////////////// if hb { - if err := encoder.Encode(nil); err != nil { - logger.Warnf("Encode(nil) = error: %v", err) - } + _ = encoder.Encode(nil) goto END } // dubbo version + path + version + method - if err := encoder.Encode(DEFAULT_DUBBO_PROTOCOL_VERSION); err != nil { + if err = encoder.Encode(DEFAULT_DUBBO_PROTOCOL_VERSION); err != nil { logger.Warnf("Encode(DEFAULT_DUBBO_PROTOCOL_VERSION) = error: %v", err) } - if err := encoder.Encode(service.Path); err != nil { + if err = encoder.Encode(service.Path); err != nil { logger.Warnf("Encode(service.Path) = error: %v", err) } - if err := encoder.Encode(service.Version); err != nil { + if err = encoder.Encode(service.Version); err != nil { logger.Warnf("Encode(service.Version) = error: %v", err) } - if err := encoder.Encode(service.Method); err != nil { + if err = encoder.Encode(service.Method); err != nil { logger.Warnf("Encode(service.Method) = error: %v", err) } @@ -244,12 +242,9 @@ func packRequest(service Service, header DubboHeader, req interface{}) ([]byte, if types, err = getArgsTypeList(args); err != nil { return nil, perrors.Wrapf(err, " PackRequest(args:%+v)", args) } - if err := encoder.Encode(types); err != nil { - logger.Warnf("Encode(types:%v) = error: %v", types, err) - } - + _ = encoder.Encode(types) for _, v := range args { - encoder.Encode(v) + _ = encoder.Encode(v) } request.Attachments[PATH_KEY] = service.Path @@ -264,7 +259,7 @@ func packRequest(service Service, header DubboHeader, req interface{}) ([]byte, request.Attachments[TIMEOUT_KEY] = strconv.Itoa(int(service.Timeout / time.Millisecond)) } - encoder.Encode(request.Attachments) + _ = encoder.Encode(request.Attachments) END: byteArray = encoder.Buffer() diff --git a/protocol/dubbo/hessian2/hessian_response.go b/protocol/dubbo/hessian2/hessian_response.go index 982960ed87..b95e1c2711 100644 --- a/protocol/dubbo/hessian2/hessian_response.go +++ b/protocol/dubbo/hessian2/hessian_response.go @@ -18,6 +18,7 @@ package hessian2 import ( "encoding/binary" + "github.com/apache/dubbo-go/common/logger" "math" "reflect" "strconv" @@ -93,7 +94,9 @@ func packResponse(header DubboHeader, ret interface{}) ([]byte, error) { if header.ResponseStatus == Response_OK { if hb { - encoder.Encode(nil) + if err := encoder.Encode(nil); err != nil { + logger.Warnf("Encode(nil) = %v", err) + } } else { atta := isSupportResponseAttachment(response.Attachments[DUBBO_VERSION_KEY]) @@ -109,30 +112,30 @@ func packResponse(header DubboHeader, ret interface{}) ([]byte, error) { } if response.Exception != nil { // throw error - encoder.Encode(resWithException) + _ = encoder.Encode(resWithException) if t, ok := response.Exception.(java_exception.Throwabler); ok { - encoder.Encode(t) + _ = encoder.Encode(t) } else { - encoder.Encode(java_exception.NewThrowable(response.Exception.Error())) + _ = encoder.Encode(java_exception.NewThrowable(response.Exception.Error())) } } else { if response.RspObj == nil { - encoder.Encode(resNullValue) + _ = encoder.Encode(resNullValue) } else { - encoder.Encode(resValue) - encoder.Encode(response.RspObj) // result + _ = encoder.Encode(resValue) + _ = encoder.Encode(response.RspObj) // result } } if atta { - encoder.Encode(response.Attachments) // attachments + _ = encoder.Encode(response.Attachments) // attachments } } } else { if response.Exception != nil { // throw error - encoder.Encode(response.Exception.Error()) + _ = encoder.Encode(response.Exception.Error()) } else { - encoder.Encode(response.RspObj) + _ = encoder.Encode(response.RspObj) } } @@ -145,7 +148,6 @@ func packResponse(header DubboHeader, ret interface{}) ([]byte, error) { // byteArray{body length} binary.BigEndian.PutUint32(byteArray[12:], uint32(pkgLen-HEADER_LENGTH)) return byteArray, nil - } // hessian decode response body diff --git a/protocol/dubbo/impl/codec.go b/protocol/dubbo/impl/codec.go index 17e7b57b45..6c9816f0ff 100644 --- a/protocol/dubbo/impl/codec.go +++ b/protocol/dubbo/impl/codec.go @@ -146,7 +146,7 @@ func (c *ProtocolCodec) Encode(p DubboPackage) ([]byte, error) { return packResponse(p, c.serializer) default: - return nil, perrors.Errorf("Unrecognised message type: %v", header.Type) + return nil, perrors.Errorf("Unrecognized message type: %v", header.Type) } } diff --git a/protocol/dubbo/impl/const.go b/protocol/dubbo/impl/const.go index 492a87a194..c9b92bad3b 100644 --- a/protocol/dubbo/impl/const.go +++ b/protocol/dubbo/impl/const.go @@ -75,10 +75,10 @@ const ( 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) + BC_LIST_VARIABLE = byte(0x55) + BC_LIST_FIXED = byte('V') + BC_LIST_VARIABLE_UNTYPED = byte(0x57) + BC_LIST_FIXED_UNTYPED = byte(0x58) BC_LIST_DIRECT = byte(0x70) BC_LIST_DIRECT_UNTYPED = byte(0x78) diff --git a/protocol/dubbo/impl/hessian.go b/protocol/dubbo/impl/hessian.go index 5fa1f2ece3..e355276c37 100644 --- a/protocol/dubbo/impl/hessian.go +++ b/protocol/dubbo/impl/hessian.go @@ -63,7 +63,7 @@ func marshalResponse(encoder *hessian.Encoder, p DubboPackage) ([]byte, error) { response := EnsureResponsePayload(p.Body) if header.ResponseStatus == Response_OK { if p.IsHeartBeat() { - encoder.Encode(nil) + _ = encoder.Encode(nil) } else { var version string if attachmentVersion, ok := response.Attachments[DUBBO_VERSION_KEY]; ok { @@ -83,30 +83,30 @@ func marshalResponse(encoder *hessian.Encoder, p DubboPackage) ([]byte, error) { } if response.Exception != nil { // throw error - encoder.Encode(resWithException) + _ = encoder.Encode(resWithException) if t, ok := response.Exception.(java_exception.Throwabler); ok { - encoder.Encode(t) + _ = encoder.Encode(t) } else { - encoder.Encode(java_exception.NewThrowable(response.Exception.Error())) + _ = encoder.Encode(java_exception.NewThrowable(response.Exception.Error())) } } else { if response.RspObj == nil { - encoder.Encode(resNullValue) + _ = encoder.Encode(resNullValue) } else { - encoder.Encode(resValue) - encoder.Encode(response.RspObj) // result + _ = encoder.Encode(resValue) + _ = encoder.Encode(response.RspObj) // result } } if atta { - encoder.Encode(response.Attachments) // attachments + _ = encoder.Encode(response.Attachments) // attachments } } } else { if response.Exception != nil { // throw error - encoder.Encode(response.Exception.Error()) + _ = encoder.Encode(response.Exception.Error()) } else { - encoder.Encode(response.RspObj) + _ = encoder.Encode(response.RspObj) } } bs := encoder.Buffer() @@ -118,10 +118,10 @@ func marshalResponse(encoder *hessian.Encoder, p DubboPackage) ([]byte, error) { func marshalRequest(encoder *hessian.Encoder, p DubboPackage) ([]byte, error) { service := p.Service request := EnsureRequestPayload(p.Body) - encoder.Encode(DEFAULT_DUBBO_PROTOCOL_VERSION) - encoder.Encode(service.Path) - encoder.Encode(service.Version) - encoder.Encode(service.Method) + _ = encoder.Encode(DEFAULT_DUBBO_PROTOCOL_VERSION) + _ = encoder.Encode(service.Path) + _ = encoder.Encode(service.Version) + _ = encoder.Encode(service.Method) args, ok := request.Params.([]interface{}) @@ -133,9 +133,9 @@ func marshalRequest(encoder *hessian.Encoder, p DubboPackage) ([]byte, error) { if err != nil { return nil, perrors.Wrapf(err, " PackRequest(args:%+v)", args) } - encoder.Encode(types) + _ = encoder.Encode(types) for _, v := range args { - encoder.Encode(v) + _ = encoder.Encode(v) } request.Attachments[PATH_KEY] = service.Path @@ -150,9 +150,8 @@ func marshalRequest(encoder *hessian.Encoder, p DubboPackage) ([]byte, error) { request.Attachments[TIMEOUT_KEY] = strconv.Itoa(int(service.Timeout / time.Millisecond)) } - encoder.Encode(request.Attachments) + _ = encoder.Encode(request.Attachments) return encoder.Buffer(), nil - } var versionInt = make(map[string]int) @@ -418,7 +417,7 @@ func getArgType(v interface{}) string { return "V" } - switch v.(type) { + switch v := v.(type) { // Serialized tags for base types case nil: return "V" diff --git a/protocol/grpc/protoc-gen-dubbo/main.go b/protocol/grpc/protoc-gen-dubbo/main.go index fbcfa6f9d4..fe3e38dddd 100644 --- a/protocol/grpc/protoc-gen-dubbo/main.go +++ b/protocol/grpc/protoc-gen-dubbo/main.go @@ -43,7 +43,7 @@ func main() { g.Error(err, "reading input") } - if err := proto.Unmarshal(data, g.Request); err != nil { + if err = proto.Unmarshal(data, g.Request); err != nil { g.Error(err, "parsing input proto") } diff --git a/protocol/grpc/protoc-gen-dubbo/plugin/dubbo/dubbo.go b/protocol/grpc/protoc-gen-dubbo/plugin/dubbo/dubbo.go index 78847282d5..ba84f36b74 100644 --- a/protocol/grpc/protoc-gen-dubbo/plugin/dubbo/dubbo.go +++ b/protocol/grpc/protoc-gen-dubbo/plugin/dubbo/dubbo.go @@ -254,8 +254,8 @@ func (g *dubboGrpc) generateClientSignature(servName string, method *pb.MethodDe return fmt.Sprintf("%s func(ctx %s.Context%s, %s) error", methName, contextPkg, reqArg, respName) } -func (g *dubboGrpc) generateClientMethod(servName, fullServName, serviceDescVar string, method *pb.MethodDescriptorProto, descExpr string) { -} +//func (g *dubboGrpc) generateClientMethod(servName, fullServName, serviceDescVar string, method *pb.MethodDescriptorProto, descExpr string) { +//} func (g *dubboGrpc) generateServerMethod(servName, fullServName string, method *pb.MethodDescriptorProto) string { methName := generator.CamelCase(method.GetName()) diff --git a/protocol/jsonrpc/http.go b/protocol/jsonrpc/http.go index 7ab1a8942d..11051df8ac 100644 --- a/protocol/jsonrpc/http.go +++ b/protocol/jsonrpc/http.go @@ -49,14 +49,14 @@ import ( // Request is HTTP protocol request type Request struct { - ID int64 - group string - protocol string - version string - service string - method string - args interface{} - contentType string + ID int64 + group string + protocol string + version string + service string + method string + args interface{} + //contentType string } // //////////////////////////////////////////// @@ -189,7 +189,7 @@ func (c *HTTPClient) Do(addr, path string, httpHeader http.Header, body []byte) return conn.SetDeadline(t) } - if err := setNetConnTimeout(tcpConn, c.options.HTTPTimeout); err != nil { + if err = setNetConnTimeout(tcpConn, c.options.HTTPTimeout); err != nil { return nil, err } diff --git a/protocol/jsonrpc/http_test.go b/protocol/jsonrpc/http_test.go index c4801c8db8..5ef40649b1 100644 --- a/protocol/jsonrpc/http_test.go +++ b/protocol/jsonrpc/http_test.go @@ -44,7 +44,7 @@ type ( } UserProvider struct { - user map[string]User + //user map[string]User } ) diff --git a/protocol/jsonrpc/json.go b/protocol/jsonrpc/json.go index 57d36d2ead..81ca512271 100644 --- a/protocol/jsonrpc/json.go +++ b/protocol/jsonrpc/json.go @@ -48,11 +48,11 @@ type CodecData struct { const ( // Errors defined in the JSON-RPC spec. See // http://www.jsonrpc.org/specification#error_object. - CodeParseError = -32700 - CodeInvalidRequest = -32600 - CodeMethodNotFound = -32601 - CodeInvalidParams = -32602 - CodeInternalError = -32603 + CodeParseError = -32700 + CodeInvalidRequest = -32600 + CodeMethodNotFound = -32601 + CodeInvalidParams = -32602 + CodeInternalError = -32603 ) // Error response Error diff --git a/protocol/protocolwrapper/mock_protocol_filter.go b/protocol/protocolwrapper/mock_protocol_filter.go index 18a4e15621..d1baba8134 100644 --- a/protocol/protocolwrapper/mock_protocol_filter.go +++ b/protocol/protocolwrapper/mock_protocol_filter.go @@ -45,5 +45,4 @@ func (pfw *mockProtocolFilter) Refer(url *common.URL) protocol.Invoker { // Destroy will do nothing func (pfw *mockProtocolFilter) Destroy() { - return } diff --git a/registry/base_registry.go b/registry/base_registry.go index ee0d340387..e97cc1a4ed 100644 --- a/registry/base_registry.go +++ b/registry/base_registry.go @@ -18,7 +18,6 @@ package registry import ( - "context" "fmt" "net/url" "os" @@ -93,7 +92,7 @@ type FacadeBasedRegistry interface { // BaseRegistry is a common logic abstract for registry. It implement Registry interface. type BaseRegistry struct { - context context.Context + //context context.Context facadeBasedRegistry FacadeBasedRegistry *common.URL birth int64 // time of file birth, seconds since Epoch; 0 if unknown diff --git a/registry/consul/service_discovery.go b/registry/consul/service_discovery.go index a64fabb1b9..fba142e04e 100644 --- a/registry/consul/service_discovery.go +++ b/registry/consul/service_discovery.go @@ -27,7 +27,7 @@ import ( import ( "github.com/dubbogo/gost/container/set" - "github.com/dubbogo/gost/page" + "github.com/dubbogo/gost/hash/page" consul "github.com/hashicorp/consul/api" "github.com/hashicorp/consul/api/watch" perrors "github.com/pkg/errors" @@ -252,7 +252,7 @@ func (csd *consulServiceDiscovery) GetServices() *gxset.HashSet { return res } - for service, _ := range services { + for service := range services { res.Add(service) } return res @@ -339,7 +339,7 @@ func (csd *consulServiceDiscovery) GetInstancesByPage(serviceName string, offset for i := offset; i < len(all) && i < offset+pageSize; i++ { res = append(res, all[i]) } - return gxpage.New(offset, pageSize, res, len(all)) + return gxpage.NewPage(offset, pageSize, res, len(all)) } func (csd *consulServiceDiscovery) GetHealthyInstancesByPage(serviceName string, offset int, pageSize int, healthy bool) gxpage.Pager { @@ -358,7 +358,7 @@ func (csd *consulServiceDiscovery) GetHealthyInstancesByPage(serviceName string, } i++ } - return gxpage.New(offset, pageSize, res, len(all)) + return gxpage.NewPage(offset, pageSize, res, len(all)) } func (csd *consulServiceDiscovery) GetRequestInstances(serviceNames []string, offset int, requestedSize int) map[string]gxpage.Pager { diff --git a/registry/consul/service_discovery_test.go b/registry/consul/service_discovery_test.go index 2169857ee8..3f97d841d5 100644 --- a/registry/consul/service_discovery_test.go +++ b/registry/consul/service_discovery_test.go @@ -84,7 +84,10 @@ func TestConsulServiceDiscovery_Destroy(t *testing.T) { func TestConsulServiceDiscovery_CRUD(t *testing.T) { // start consul agent consulAgent := consul.NewConsulAgent(t, registryPort) - defer consulAgent.Shutdown() + defer func() { + err := consulAgent.Shutdown() + assert.NoError(t, err) + }() prepareData() var eventDispatcher = MockEventDispatcher{Notify: make(chan struct{}, 1)} @@ -138,7 +141,8 @@ func TestConsulServiceDiscovery_CRUD(t *testing.T) { assert.Equal(t, 1, len(page.GetData())) instanceResult = page.GetData()[0].(*registry.DefaultServiceInstance) - v, _ := instanceResult.Metadata["aaa"] + v, ok := instanceResult.Metadata["aaa"] + assert.True(t, ok) assert.Equal(t, "bbb", v) // test dispatcher event diff --git a/registry/consul/utils_test.go b/registry/consul/utils_test.go index d78c534e93..b7e2929cec 100644 --- a/registry/consul/utils_test.go +++ b/registry/consul/utils_test.go @@ -19,6 +19,8 @@ package consul import ( "fmt" + "github.com/apache/dubbo-go/common/logger" + "github.com/stretchr/testify/assert" "net" "net/url" "strconv" @@ -113,15 +115,24 @@ func (server *testServer) serve() { if err != nil { continue } - conn.Write([]byte("Hello World")) - conn.Close() + _, err = conn.Write([]byte("Hello World")) + if err != nil { + logger.Warnf("conn.Write() = error: %v", err) + } + err = conn.Close() + if err != nil { + logger.Warnf("conn.Close() = error: %v", err) + } } } } func (server *testServer) close() { close(server.done) - server.listener.Close() + if err := server.listener.Close(); err != nil { + fmt.Printf("server.listener.Close() = error:%v\n", err) + } + server.wg.Wait() } @@ -148,7 +159,10 @@ func (suite *consulRegistryTestSuite) close() { // register -> subscribe -> unregister func test1(t *testing.T) { consulAgent := consul.NewConsulAgent(t, registryPort) - defer consulAgent.Shutdown() + defer func() { + err := consulAgent.Shutdown() + assert.NoError(t, err) + }() server := newServer(providerHost, providerPort) defer server.close() @@ -169,7 +183,10 @@ func test1(t *testing.T) { // subscribe -> register -> unregister func test2(t *testing.T) { consulAgent := consul.NewConsulAgent(t, registryPort) - defer consulAgent.Shutdown() + defer func() { + err := consulAgent.Shutdown() + assert.NoError(t, err) + }() server := newServer(providerHost, providerPort) defer server.close() diff --git a/registry/directory/directory.go b/registry/directory/directory.go index f55bd4abfd..1b607351af 100644 --- a/registry/directory/directory.go +++ b/registry/directory/directory.go @@ -26,7 +26,6 @@ import ( import ( perrors "github.com/pkg/errors" - "go.uber.org/atomic" ) import ( @@ -64,9 +63,9 @@ type RegistryDirectory struct { configurators []config_center.Configurator consumerConfigurationListener *consumerConfigurationListener referenceConfigurationListener *referenceConfigurationListener - serviceKey string - forbidden atomic.Bool - registerLock sync.Mutex // this lock if for register + //serviceKey string + //forbidden atomic.Bool + registerLock sync.Mutex // this lock if for register } // NewRegistryDirectory will create a new RegistryDirectory @@ -154,7 +153,6 @@ func (dir *RegistryDirectory) refreshAllInvokers(events []*registry.ServiceEvent if event.Action != remoting.EventTypeUpdate { panic("Your implements of register center is wrong, " + "please check the Action of ServiceEvent should be EventTypeUpdate") - return } // Originally it will Merge URL many times, now we just execute once. // MergeUrl is executed once and put the result into Event. After this, the key will get from Event.Key(). @@ -297,11 +295,7 @@ func (dir *RegistryDirectory) toGroupInvokers() []protocol.Invoker { for _, invoker := range newInvokersList { group := invoker.GetUrl().GetParam(constant.GROUP_KEY, "") - if _, ok := groupInvokersMap[group]; ok { - groupInvokersMap[group] = append(groupInvokersMap[group], invoker) - } else { - groupInvokersMap[group] = []protocol.Invoker{invoker} - } + groupInvokersMap[group] = append(groupInvokersMap[group], invoker) } groupInvokersList := make([]protocol.Invoker, 0, len(groupInvokersMap)) if len(groupInvokersMap) == 1 { diff --git a/registry/etcdv3/service_discovery.go b/registry/etcdv3/service_discovery.go index e8d4aea9a4..ca6016e49b 100644 --- a/registry/etcdv3/service_discovery.go +++ b/registry/etcdv3/service_discovery.go @@ -26,7 +26,7 @@ import ( import ( gxset "github.com/dubbogo/gost/container/set" - gxpage "github.com/dubbogo/gost/page" + gxpage "github.com/dubbogo/gost/hash/page" "github.com/hashicorp/vault/sdk/helper/jsonutil" perrors "github.com/pkg/errors" ) @@ -108,8 +108,11 @@ func (e *etcdV3ServiceDiscovery) Update(instance registry.ServiceInstance) error if nil != e.client { ins, err := jsonutil.EncodeJSON(instance) - if nil == err { - e.client.RegisterTemp(path, string(ins)) + if err == nil { + if err = e.client.RegisterTemp(path, string(ins)); err != nil { + logger.Warnf("etcdV3ServiceDiscovery.client.RegisterTemp(path:%v, instance:%v) = error:%v", + path, string(ins), err) + } e.services.Add(instance.GetServiceName()) } } @@ -162,7 +165,7 @@ func (e *etcdV3ServiceDiscovery) GetInstances(serviceName string) []registry.Ser logger.Infof("could not getChildrenKVList the err is:%v", err) } - return make([]registry.ServiceInstance, 0, 0) + return make([]registry.ServiceInstance, 0) } // GetInstancesByPage will return a page containing instances of ServiceInstance with the serviceName @@ -177,7 +180,7 @@ func (e *etcdV3ServiceDiscovery) GetInstancesByPage(serviceName string, offset i res = append(res, all[i]) } - return gxpage.New(offset, pageSize, res, len(all)) + return gxpage.NewPage(offset, pageSize, res, len(all)) } // GetHealthyInstancesByPage will return a page containing instances of ServiceInstance. @@ -199,7 +202,7 @@ func (e *etcdV3ServiceDiscovery) GetHealthyInstancesByPage(serviceName string, o } i++ } - return gxpage.New(offset, pageSize, res, len(all)) + return gxpage.NewPage(offset, pageSize, res, len(all)) } // Batch get all instances by the specified service names @@ -319,5 +322,5 @@ func newEtcdV3ServiceDiscovery(name string) (registry.ServiceDiscovery, error) { descriptor := fmt.Sprintf("etcd-service-discovery[%s]", remoteConfig.Address) - return &etcdV3ServiceDiscovery{descriptor, client, nil, gxset.NewSet(), make(map[string]*etcdv3.EventListener, 0)}, nil + return &etcdV3ServiceDiscovery{descriptor, client, nil, gxset.NewSet(), make(map[string]*etcdv3.EventListener)}, nil } diff --git a/registry/event/event_publishing_service_deiscovery_test.go b/registry/event/event_publishing_service_deiscovery_test.go index 54752c03c0..504f7b5fae 100644 --- a/registry/event/event_publishing_service_deiscovery_test.go +++ b/registry/event/event_publishing_service_deiscovery_test.go @@ -24,7 +24,7 @@ import ( import ( gxset "github.com/dubbogo/gost/container/set" - gxpage "github.com/dubbogo/gost/page" + gxpage "github.com/dubbogo/gost/hash/page" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/suite" ) diff --git a/registry/event/event_publishing_service_discovery.go b/registry/event/event_publishing_service_discovery.go index 3ee2f4a449..773eee6e83 100644 --- a/registry/event/event_publishing_service_discovery.go +++ b/registry/event/event_publishing_service_discovery.go @@ -19,7 +19,7 @@ package event import ( gxset "github.com/dubbogo/gost/container/set" - gxpage "github.com/dubbogo/gost/page" + gxpage "github.com/dubbogo/gost/hash/page" ) import ( diff --git a/registry/file/service_discovery.go b/registry/file/service_discovery.go index d19d1f310c..c3b77f6849 100644 --- a/registry/file/service_discovery.go +++ b/registry/file/service_discovery.go @@ -28,7 +28,7 @@ import ( import ( gxset "github.com/dubbogo/gost/container/set" - gxpage "github.com/dubbogo/gost/page" + gxpage "github.com/dubbogo/gost/hash/page" perrors "github.com/pkg/errors" ) @@ -212,7 +212,7 @@ func (fssd *fileSystemServiceDiscovery) GetInstances(serviceName string) []regis if err != nil { logger.Errorf("[FileServiceDiscovery] Could not query the instances for service{%s}, error = err{%v} ", serviceName, err) - return make([]registry.ServiceInstance, 0, 0) + return make([]registry.ServiceInstance, 0) } res := make([]registry.ServiceInstance, 0, set.Size()) @@ -223,7 +223,7 @@ func (fssd *fileSystemServiceDiscovery) GetInstances(serviceName string) []regis logger.Errorf("[FileServiceDiscovery] Could not get the properties for id{%s}, service{%s}, "+ "error = err{%v} ", id, serviceName, err) - return make([]registry.ServiceInstance, 0, 0) + return make([]registry.ServiceInstance, 0) } dsi := ®istry.DefaultServiceInstance{} @@ -232,7 +232,7 @@ func (fssd *fileSystemServiceDiscovery) GetInstances(serviceName string) []regis logger.Errorf("[FileServiceDiscovery] Could not unmarshal the properties for id{%s}, service{%s}, "+ "error = err{%v} ", id, serviceName, err) - return make([]registry.ServiceInstance, 0, 0) + return make([]registry.ServiceInstance, 0) } res = append(res, dsi) diff --git a/registry/file/service_discovery_test.go b/registry/file/service_discovery_test.go index 2f52eb52e1..0062eae32c 100644 --- a/registry/file/service_discovery_test.go +++ b/registry/file/service_discovery_test.go @@ -44,7 +44,7 @@ func TestNewFileSystemServiceDiscoveryAndDestroy(t *testing.T) { serviceDiscovery, err := newFileSystemServiceDiscovery(testName) assert.NoError(t, err) assert.NotNil(t, serviceDiscovery) - defer func () { + defer func() { err = serviceDiscovery.Destroy() assert.Nil(t, err) }() @@ -82,7 +82,7 @@ func TestCURDFileSystemServiceDiscovery(t *testing.T) { err = serviceDiscovery.Register(r1) assert.NoError(t, err) - defer func () { + defer func() { err = serviceDiscovery.Destroy() assert.NoError(t, err) }() diff --git a/registry/kubernetes/registry.go b/registry/kubernetes/registry.go index c1e559e48d..c4a0352cfa 100644 --- a/registry/kubernetes/registry.go +++ b/registry/kubernetes/registry.go @@ -19,7 +19,6 @@ package kubernetes import ( "fmt" - "os" "path" "sync" "time" @@ -40,11 +39,6 @@ import ( "github.com/apache/dubbo-go/remoting/kubernetes" ) -var ( - processID = "" - localIP = "" -) - const ( Name = "kubernetes" ConnDelay = 3 @@ -52,8 +46,8 @@ const ( ) func init() { - processID = fmt.Sprintf("%d", os.Getpid()) - localIP = common.GetLocalIp() + //processID = fmt.Sprintf("%d", os.Getpid()) + //localIP = common.GetLocalIp() extension.SetRegistry(Name, newKubernetesRegistry) } diff --git a/registry/mock_registry.go b/registry/mock_registry.go index 7c269c3a40..6287bb0d16 100644 --- a/registry/mock_registry.go +++ b/registry/mock_registry.go @@ -137,10 +137,7 @@ type listener struct { } func (l *listener) Next() (*ServiceEvent, error) { - select { - case e := <-l.listenChan: - return e, nil - } + return <-l.listenChan, nil } func (*listener) Close() { diff --git a/registry/nacos/listener.go b/registry/nacos/listener.go index 7f27326d6d..6783bf4684 100644 --- a/registry/nacos/listener.go +++ b/registry/nacos/listener.go @@ -188,7 +188,9 @@ func (nl *nacosListener) startListen() error { } serviceName := getSubscribeName(nl.listenUrl) nl.subscribeParam = &vo.SubscribeParam{ServiceName: serviceName, SubscribeCallback: nl.Callback} - go nl.namingClient.Subscribe(nl.subscribeParam) + go func() { + _ = nl.namingClient.Subscribe(nl.subscribeParam) + }() return nil } diff --git a/registry/nacos/registry_test.go b/registry/nacos/registry_test.go index 078b8ce59c..a852c00b14 100644 --- a/registry/nacos/registry_test.go +++ b/registry/nacos/registry_test.go @@ -167,7 +167,9 @@ func TestNacosRegistry_Subscribe_del(t *testing.T) { nacosReg := reg.(*nacosRegistry) //deregister instance to mock instance offline - nacosReg.namingClient.DeregisterInstance(vo.DeregisterInstanceParam{Ip: "127.0.0.2", Port: 20000, ServiceName: "providers:com.ikurento.user.UserProvider:2.0.0:guangzhou-idc"}) + _, err = nacosReg.namingClient.DeregisterInstance(vo.DeregisterInstanceParam{Ip: "127.0.0.2", Port: 20000, + ServiceName: "providers:com.ikurento.user.UserProvider:2.0.0:guangzhou-idc"}) + assert.NoError(t, err) serviceEvent3, _ := listener.Next() assert.NoError(t, err) diff --git a/registry/nacos/service_discovery.go b/registry/nacos/service_discovery.go index b38e150e51..4533e7bc6b 100644 --- a/registry/nacos/service_discovery.go +++ b/registry/nacos/service_discovery.go @@ -24,7 +24,7 @@ import ( import ( "github.com/dubbogo/gost/container/set" - "github.com/dubbogo/gost/page" + "github.com/dubbogo/gost/hash/page" "github.com/nacos-group/nacos-sdk-go/clients/naming_client" "github.com/nacos-group/nacos-sdk-go/model" "github.com/nacos-group/nacos-sdk-go/vo" @@ -143,7 +143,7 @@ func (n *nacosServiceDiscovery) GetInstances(serviceName string) []registry.Serv if err != nil { logger.Errorf("Could not query the instances for service: %+v, group: %+v . It happened err %+v", serviceName, n.group, err) - return make([]registry.ServiceInstance, 0, 0) + return make([]registry.ServiceInstance, 0) } res := make([]registry.ServiceInstance, 0, len(instances)) for _, ins := range instances { @@ -175,7 +175,7 @@ func (n *nacosServiceDiscovery) GetInstancesByPage(serviceName string, offset in for i := offset; i < len(all) && i < offset+pageSize; i++ { res = append(res, all[i]) } - return gxpage.New(offset, pageSize, res, len(all)) + return gxpage.NewPage(offset, pageSize, res, len(all)) } // GetHealthyInstancesByPage will return the instance @@ -198,7 +198,7 @@ func (n *nacosServiceDiscovery) GetHealthyInstancesByPage(serviceName string, of } i++ } - return gxpage.New(offset, pageSize, res, len(all)) + return gxpage.NewPage(offset, pageSize, res, len(all)) } // GetRequestInstances will return the instances diff --git a/registry/nacos/service_discovery_test.go b/registry/nacos/service_discovery_test.go index 3b09136d2e..2af1d5b6f5 100644 --- a/registry/nacos/service_discovery_test.go +++ b/registry/nacos/service_discovery_test.go @@ -151,7 +151,8 @@ func TestNacosServiceDiscovery_CRUD(t *testing.T) { assert.Equal(t, 1, len(page.GetData())) instance = page.GetData()[0].(*registry.DefaultServiceInstance) - v, _ := instance.Metadata["a"] + v, ok := instance.Metadata["a"] + assert.True(t, ok) assert.Equal(t, "b", v) // test dispatcher event diff --git a/registry/protocol/protocol.go b/registry/protocol/protocol.go index 3be88c3beb..4fcdf93c0f 100644 --- a/registry/protocol/protocol.go +++ b/registry/protocol/protocol.go @@ -212,7 +212,11 @@ func (proto *registryProtocol) Export(invoker protocol.Invoker) protocol.Exporte logger.Infof("The exporter has not been cached, and will return a new exporter!") } - go reg.Subscribe(overriderUrl, overrideSubscribeListener) + go func() { + if err = reg.Subscribe(overriderUrl, overrideSubscribeListener); err != nil { + logger.Warnf("reg.subscribe(overriderUrl:%v) = error:%v", overriderUrl, err) + } + }() return cachedExporter.(protocol.Exporter) } diff --git a/registry/service_discovery.go b/registry/service_discovery.go index cb7a3c0182..5ab7683515 100644 --- a/registry/service_discovery.go +++ b/registry/service_discovery.go @@ -23,7 +23,7 @@ import ( import ( gxset "github.com/dubbogo/gost/container/set" - gxpage "github.com/dubbogo/gost/page" + gxpage "github.com/dubbogo/gost/hash/page" ) const DefaultPageSize = 100 diff --git a/registry/service_instance.go b/registry/service_instance.go index dbb458284d..43a1640eea 100644 --- a/registry/service_instance.go +++ b/registry/service_instance.go @@ -91,7 +91,7 @@ func (d *DefaultServiceInstance) IsHealthy() bool { // GetMetadata will return the metadata, it will never return nil func (d *DefaultServiceInstance) GetMetadata() map[string]string { if d.Metadata == nil { - d.Metadata = make(map[string]string, 0) + d.Metadata = make(map[string]string) } return d.Metadata } diff --git a/registry/servicediscovery/service_discovery_registry.go b/registry/servicediscovery/service_discovery_registry.go index ad6ec981de..2ca4b8015b 100644 --- a/registry/servicediscovery/service_discovery_registry.go +++ b/registry/servicediscovery/service_discovery_registry.go @@ -462,7 +462,7 @@ func (s *serviceDiscoveryRegistry) initRevisionExportedURLsByInst(serviceInstanc } revisionExportedURLs := revisionExportedURLsMap[revision] firstGet := false - if revisionExportedURLs == nil || len(revisionExportedURLs) == 0 { + if len(revisionExportedURLs) == 0 { if len(revisionExportedURLsMap) > 0 { // The case is that current ServiceInstance with the different revision logger.Warnf("The ServiceInstance[id: %s, host : %s , port : %s] has different revision : %s"+ diff --git a/registry/servicediscovery/service_discovery_registry_test.go b/registry/servicediscovery/service_discovery_registry_test.go index ad6b73d3b4..391d92c8ad 100644 --- a/registry/servicediscovery/service_discovery_registry_test.go +++ b/registry/servicediscovery/service_discovery_registry_test.go @@ -23,7 +23,7 @@ import ( import ( "github.com/dubbogo/gost/container/set" - "github.com/dubbogo/gost/page" + "github.com/dubbogo/gost/hash/page" "github.com/stretchr/testify/assert" ) import ( @@ -79,18 +79,17 @@ func TestServiceDiscoveryRegistry_Register(t *testing.T) { registry, err := newServiceDiscoveryRegistry(registryURL) assert.Nil(t, err) assert.NotNil(t, registry) - registry.Register(url) + err = registry.Register(url) + assert.NoError(t, err) } type mockEventDispatcher struct { } func (m *mockEventDispatcher) AddEventListener(observer.EventListener) { - } func (m *mockEventDispatcher) AddEventListeners([]observer.EventListener) { - } func (m *mockEventDispatcher) RemoveEventListener(observer.EventListener) { diff --git a/registry/servicediscovery/synthesizer/rest/rest_subscribed_urls_synthesizer.go b/registry/servicediscovery/synthesizer/rest/rest_subscribed_urls_synthesizer.go index d1ab6113b3..c6b3aeaad0 100644 --- a/registry/servicediscovery/synthesizer/rest/rest_subscribed_urls_synthesizer.go +++ b/registry/servicediscovery/synthesizer/rest/rest_subscribed_urls_synthesizer.go @@ -38,14 +38,11 @@ type RestSubscribedURLsSynthesizer struct { } func (r RestSubscribedURLsSynthesizer) Support(subscribedURL *common.URL) bool { - if "rest" == subscribedURL.Protocol { - return true - } - return false + return "rest" == subscribedURL.Protocol } func (r RestSubscribedURLsSynthesizer) Synthesize(subscribedURL *common.URL, serviceInstances []registry.ServiceInstance) []*common.URL { - urls := make([]*common.URL, len(serviceInstances), len(serviceInstances)) + urls := make([]*common.URL, len(serviceInstances)) for i, s := range serviceInstances { splitHost := strings.Split(s.GetHost(), ":") u := common.NewURLWithOptions(common.WithProtocol(subscribedURL.Protocol), common.WithIp(splitHost[0]), diff --git a/registry/zookeeper/registry.go b/registry/zookeeper/registry.go index fe492c2b12..3232ee624d 100644 --- a/registry/zookeeper/registry.go +++ b/registry/zookeeper/registry.go @@ -129,7 +129,7 @@ func (r *zkRegistry) InitListeners() { defer oldDataListener.mutex.Unlock() r.dataListener.closed = true recovered := r.dataListener.subscribed - if recovered != nil && len(recovered) > 0 { + if len(recovered) > 0 { // recover all subscribed url for _, oldListener := range recovered { var ( @@ -251,8 +251,7 @@ func (r *zkRegistry) getListener(conf *common.URL) (*RegistryConfigurationListen dataListener.mutex.Lock() defer dataListener.mutex.Unlock() if r.dataListener.subscribed[conf.ServiceKey()] != nil { - - zkListener, _ := r.dataListener.subscribed[conf.ServiceKey()].(*RegistryConfigurationListener) + zkListener, _ = r.dataListener.subscribed[conf.ServiceKey()].(*RegistryConfigurationListener) if zkListener != nil { r.listenerLock.Lock() defer r.listenerLock.Unlock() @@ -284,7 +283,11 @@ func (r *zkRegistry) getListener(conf *common.URL) (*RegistryConfigurationListen //Interested register to dataconfig. r.dataListener.SubscribeURL(conf, zkListener) - go r.listener.ListenServiceEvent(conf, fmt.Sprintf("/dubbo/%s/"+constant.DEFAULT_CATEGORY, url.QueryEscape(conf.Service())), r.dataListener) + go r.listener.ListenServiceEvent( + conf, + fmt.Sprintf("/dubbo/%s/"+constant.DEFAULT_CATEGORY, url.QueryEscape(conf.Service())), + r.dataListener, + ) return zkListener, nil } @@ -295,9 +298,9 @@ func (r *zkRegistry) getCloseListener(conf *common.URL) (*RegistryConfigurationL r.dataListener.mutex.Lock() configurationListener := r.dataListener.subscribed[conf.ServiceKey()] if configurationListener != nil { - zkListener, _ := configurationListener.(*RegistryConfigurationListener) - if zkListener != nil { - if zkListener.isClosed { + rcListener, _ := configurationListener.(*RegistryConfigurationListener) + if rcListener != nil { + if rcListener.isClosed { r.dataListener.mutex.Unlock() return nil, perrors.New("configListener already been closed") } diff --git a/registry/zookeeper/registry_test.go b/registry/zookeeper/registry_test.go index d7a011ec4c..5959a903bc 100644 --- a/registry/zookeeper/registry_test.go +++ b/registry/zookeeper/registry_test.go @@ -34,11 +34,13 @@ import ( ) func Test_Register(t *testing.T) { - regurl, _ := common.NewURL("registry://127.0.0.1:1111", common.WithParamsValue(constant.ROLE_KEY, strconv.Itoa(common.PROVIDER))) + regURL, _ := common.NewURL("registry://127.0.0.1:1111", common.WithParamsValue(constant.ROLE_KEY, strconv.Itoa(common.PROVIDER))) url, _ := common.NewURL("dubbo://127.0.0.1:20000/com.ikurento.user.UserProvider", common.WithParamsValue(constant.CLUSTER_KEY, "mock"), common.WithParamsValue("serviceid", "soa.mock"), common.WithMethods([]string{"GetUser", "AddUser"})) - ts, reg, _ := newMockZkRegistry(regurl) - defer ts.Stop() + ts, reg, _ := newMockZkRegistry(regURL) + defer func() { + _ = ts.Stop() + }() err := reg.Register(url) children, _ := reg.client.GetChildren("/dubbo/com.ikurento.user.UserProvider/providers") assert.Regexp(t, ".*dubbo%3A%2F%2F127.0.0.1%3A20000%2Fcom.ikurento.user.UserProvider%3Fanyhost%3Dtrue%26cluster%3Dmock%26.*.serviceid%3Dsoa.mock", children) @@ -47,17 +49,21 @@ func Test_Register(t *testing.T) { func Test_UnRegister(t *testing.T) { // register - regurl, _ := common.NewURL("registry://127.0.0.1:1111", common.WithParamsValue(constant.ROLE_KEY, strconv.Itoa(common.PROVIDER))) + regURL, _ := common.NewURL("registry://127.0.0.1:1111", common.WithParamsValue(constant.ROLE_KEY, strconv.Itoa(common.PROVIDER))) url, _ := common.NewURL("dubbo://127.0.0.1:20000/com.ikurento.user.UserProvider", common.WithParamsValue(constant.CLUSTER_KEY, "mock"), common.WithParamsValue("serviceid", "soa.mock"), common.WithMethods([]string{"GetUser", "AddUser"})) - ts, reg, _ := newMockZkRegistry(regurl) - defer ts.Stop() + ts, reg, _ := newMockZkRegistry(regURL) + defer func() { + _ = ts.Stop() + }() err := reg.Register(url) + assert.NoError(t, err) children, _ := reg.client.GetChildren("/dubbo/com.ikurento.user.UserProvider/providers") assert.Regexp(t, ".*dubbo%3A%2F%2F127.0.0.1%3A20000%2Fcom.ikurento.user.UserProvider%3Fanyhost%3Dtrue%26cluster%3Dmock%26.*.serviceid%3Dsoa.mock", children) assert.NoError(t, err) err = reg.UnRegister(url) + assert.NoError(t, err) children, err = reg.client.GetChildren("/dubbo/com.ikurento.user.UserProvider/providers") assert.Equal(t, 0, len(children)) assert.Error(t, err) @@ -67,13 +73,12 @@ func Test_UnRegister(t *testing.T) { children, _ = reg.client.GetChildren("/dubbo/com.ikurento.user.UserProvider/providers") assert.Regexp(t, ".*dubbo%3A%2F%2F127.0.0.1%3A20000%2Fcom.ikurento.user.UserProvider%3Fanyhost%3Dtrue%26cluster%3Dmock%26.*.serviceid%3Dsoa.mock", children) assert.NoError(t, err) - } func Test_Subscribe(t *testing.T) { - regurl, _ := common.NewURL("registry://127.0.0.1:1111", common.WithParamsValue(constant.ROLE_KEY, strconv.Itoa(common.PROVIDER))) + regURL, _ := common.NewURL("registry://127.0.0.1:1111", common.WithParamsValue(constant.ROLE_KEY, strconv.Itoa(common.PROVIDER))) url, _ := common.NewURL("dubbo://127.0.0.1:20000/com.ikurento.user.UserProvider", common.WithParamsValue(constant.CLUSTER_KEY, "mock"), common.WithMethods([]string{"GetUser", "AddUser"})) - ts, reg, _ := newMockZkRegistry(regurl) + ts, reg, _ := newMockZkRegistry(regURL) //provider register err := reg.Register(url) @@ -84,8 +89,8 @@ func Test_Subscribe(t *testing.T) { } //consumer register - regurl.SetParam(constant.ROLE_KEY, strconv.Itoa(common.CONSUMER)) - _, reg2, _ := newMockZkRegistry(regurl, zookeeper.WithTestCluster(ts)) + regURL.SetParam(constant.ROLE_KEY, strconv.Itoa(common.CONSUMER)) + _, reg2, _ := newMockZkRegistry(regURL, zookeeper.WithTestCluster(ts)) err = reg2.Register(url) assert.Nil(t, err) @@ -97,13 +102,15 @@ func Test_Subscribe(t *testing.T) { return } assert.Regexp(t, ".*ServiceEvent{Action{add}.*", serviceEvent.String()) - defer ts.Stop() + defer func() { + _ = ts.Stop() + }() } func Test_UnSubscribe(t *testing.T) { - regurl, _ := common.NewURL("registry://127.0.0.1:1111", common.WithParamsValue(constant.ROLE_KEY, strconv.Itoa(common.PROVIDER))) + regURL, _ := common.NewURL("registry://127.0.0.1:1111", common.WithParamsValue(constant.ROLE_KEY, strconv.Itoa(common.PROVIDER))) url, _ := common.NewURL("dubbo://127.0.0.1:20000/com.ikurento.user.UserProvider", common.WithParamsValue(constant.CLUSTER_KEY, "mock"), common.WithMethods([]string{"GetUser", "AddUser"})) - ts, reg, _ := newMockZkRegistry(regurl) + ts, reg, _ := newMockZkRegistry(regURL) //provider register err := reg.Register(url) @@ -114,8 +121,8 @@ func Test_UnSubscribe(t *testing.T) { } //consumer register - regurl.SetParam(constant.ROLE_KEY, strconv.Itoa(common.CONSUMER)) - _, reg2, _ := newMockZkRegistry(regurl, zookeeper.WithTestCluster(ts)) + regURL.SetParam(constant.ROLE_KEY, strconv.Itoa(common.CONSUMER)) + _, reg2, _ := newMockZkRegistry(regURL, zookeeper.WithTestCluster(ts)) err = reg2.Register(url) assert.Nil(t, err) @@ -131,16 +138,19 @@ func Test_UnSubscribe(t *testing.T) { err = reg2.UnSubscribe(url, nil) assert.Nil(t, err) assert.Nil(t, reg2.listener) - - defer ts.Stop() + defer func() { + _ = ts.Stop() + }() } -func Test_ConsumerDestory(t *testing.T) { - regurl, _ := common.NewURL("registry://127.0.0.1:1111", common.WithParamsValue(constant.ROLE_KEY, strconv.Itoa(common.CONSUMER))) +func Test_ConsumerDestroy(t *testing.T) { + regURL, _ := common.NewURL("registry://127.0.0.1:1111", common.WithParamsValue(constant.ROLE_KEY, strconv.Itoa(common.CONSUMER))) url, _ := common.NewURL("dubbo://127.0.0.1:20000/com.ikurento.user.UserProvider", common.WithParamsValue(constant.CLUSTER_KEY, "mock"), common.WithMethods([]string{"GetUser", "AddUser"})) - ts, reg, err := newMockZkRegistry(regurl) - defer ts.Stop() + ts, reg, err := newMockZkRegistry(regURL) + defer func() { + _ = ts.Stop() + }() assert.NoError(t, err) err = reg.Register(url) @@ -152,15 +162,16 @@ func Test_ConsumerDestory(t *testing.T) { time.Sleep(1e9) reg.Destroy() assert.Equal(t, false, reg.IsAvailable()) - } -func Test_ProviderDestory(t *testing.T) { - regurl, _ := common.NewURL("registry://127.0.0.1:1111", common.WithParamsValue(constant.ROLE_KEY, strconv.Itoa(common.PROVIDER))) +func Test_ProviderDestroy(t *testing.T) { + regURL, _ := common.NewURL("registry://127.0.0.1:1111", common.WithParamsValue(constant.ROLE_KEY, strconv.Itoa(common.PROVIDER))) url, _ := common.NewURL("dubbo://127.0.0.1:20000/com.ikurento.user.UserProvider", common.WithParamsValue(constant.CLUSTER_KEY, "mock"), common.WithMethods([]string{"GetUser", "AddUser"})) - ts, reg, err := newMockZkRegistry(regurl) - defer ts.Stop() + ts, reg, err := newMockZkRegistry(regURL) + defer func() { + _ = ts.Stop() + }() assert.NoError(t, err) err = reg.Register(url) diff --git a/registry/zookeeper/service_discovery.go b/registry/zookeeper/service_discovery.go index 6d9582f33a..095a78dc3c 100644 --- a/registry/zookeeper/service_discovery.go +++ b/registry/zookeeper/service_discovery.go @@ -27,7 +27,7 @@ import ( import ( "github.com/dubbogo/gost/container/set" - "github.com/dubbogo/gost/page" + "github.com/dubbogo/gost/hash/page" perrors "github.com/pkg/errors" ) @@ -60,9 +60,9 @@ func init() { } type zookeeperServiceDiscovery struct { - client *zookeeper.ZookeeperClient - csd *curator_discovery.ServiceDiscovery - listener *zookeeper.ZkEventListener + client *zookeeper.ZookeeperClient + csd *curator_discovery.ServiceDiscovery + //listener *zookeeper.ZkEventListener url *common.URL wg sync.WaitGroup cltLock sync.Mutex @@ -214,7 +214,7 @@ func (zksd *zookeeperServiceDiscovery) GetInstances(serviceName string) []regist if err != nil { logger.Errorf("[zkServiceDiscovery] Could not query the instances for service{%s}, error = err{%v} ", serviceName, err) - return make([]registry.ServiceInstance, 0, 0) + return make([]registry.ServiceInstance, 0) } iss := make([]registry.ServiceInstance, 0, len(criss)) for _, cris := range criss { @@ -231,7 +231,7 @@ func (zksd *zookeeperServiceDiscovery) GetInstancesByPage(serviceName string, of for i := offset; i < len(all) && i < offset+pageSize; i++ { res = append(res, all[i]) } - return gxpage.New(offset, pageSize, res, len(all)) + return gxpage.NewPage(offset, pageSize, res, len(all)) } // GetHealthyInstancesByPage will return the instance @@ -254,7 +254,7 @@ func (zksd *zookeeperServiceDiscovery) GetHealthyInstancesByPage(serviceName str } i++ } - return gxpage.New(offset, pageSize, res, len(all)) + return gxpage.NewPage(offset, pageSize, res, len(all)) } // GetRequestInstances will return the instances diff --git a/registry/zookeeper/service_discovery_test.go b/registry/zookeeper/service_discovery_test.go index 60b3d13759..c9e14afd50 100644 --- a/registry/zookeeper/service_discovery_test.go +++ b/registry/zookeeper/service_discovery_test.go @@ -75,12 +75,13 @@ func TestNewZookeeperServiceDiscovery(t *testing.T) { func TestCURDZookeeperServiceDiscovery(t *testing.T) { ts := prepareData(t) - defer ts.Stop() + defer func() { + _ = ts.Stop() + }() sd, err := newZookeeperServiceDiscovery(testName) assert.Nil(t, err) defer func() { - err := sd.Destroy() - assert.Nil(t, err) + _ = sd.Destroy() }() md := make(map[string]string) md["t1"] = "test1" @@ -143,12 +144,13 @@ func TestCURDZookeeperServiceDiscovery(t *testing.T) { func TestAddListenerZookeeperServiceDiscovery(t *testing.T) { ts := prepareData(t) - defer ts.Stop() + defer func() { + _ = ts.Stop() + }() sd, err := newZookeeperServiceDiscovery(testName) assert.Nil(t, err) defer func() { - err := sd.Destroy() - assert.Nil(t, err) + _ = sd.Destroy() }() err = sd.Register(®istry.DefaultServiceInstance{ @@ -160,8 +162,6 @@ func TestAddListenerZookeeperServiceDiscovery(t *testing.T) { Healthy: true, Metadata: nil, }) - assert.Nil(t, err) - assert.Nil(t, err) wg := &sync.WaitGroup{} wg.Add(1) @@ -176,7 +176,7 @@ func TestAddListenerZookeeperServiceDiscovery(t *testing.T) { extension.SetAndInitGlobalDispatcher("direct") extension.GetGlobalDispatcher().AddEventListener(sicl) err = sd.AddListener(sicl) - assert.Nil(t, err) + assert.NoError(t, err) err = sd.Update(®istry.DefaultServiceInstance{ Id: "testId", @@ -187,6 +187,7 @@ func TestAddListenerZookeeperServiceDiscovery(t *testing.T) { Healthy: true, Metadata: nil, }) + assert.NoError(t, err) tn.wg.Wait() } diff --git a/remoting/etcdv3/client.go b/remoting/etcdv3/client.go index ebd454242d..34ee31bd94 100644 --- a/remoting/etcdv3/client.go +++ b/remoting/etcdv3/client.go @@ -416,7 +416,9 @@ func (c *Client) keepAliveKV(k string, v string) error { keepAlive, err := c.rawClient.KeepAlive(c.ctx, lease.ID) if err != nil || keepAlive == nil { - c.rawClient.Revoke(c.ctx, lease.ID) + if _, revokeErr := c.rawClient.Revoke(c.ctx, lease.ID); revokeErr != nil { + logger.Warnf("rawClient.Revoke() = error:%v", revokeErr) + } if err != nil { return perrors.WithMessage(err, "keep alive lease") } else { diff --git a/remoting/etcdv3/client_test.go b/remoting/etcdv3/client_test.go index da31136ed3..787c24d45f 100644 --- a/remoting/etcdv3/client_test.go +++ b/remoting/etcdv3/client_test.go @@ -108,7 +108,6 @@ func (suite *ClientTestSuite) SetupSuite() { } suite.etcd = e - return } // stop etcd server @@ -133,9 +132,9 @@ func (suite *ClientTestSuite) setUpClient() *Client { // set up a client for suite func (suite *ClientTestSuite) SetupTest() { c := suite.setUpClient() - c.CleanKV() + err := c.CleanKV() + suite.Nil(err) suite.client = c - return } func (suite *ClientTestSuite) TestClientClose() { diff --git a/remoting/getty/dubbo_codec_for_test.go b/remoting/getty/dubbo_codec_for_test.go index b33fb7422e..9afc18a9aa 100644 --- a/remoting/getty/dubbo_codec_for_test.go +++ b/remoting/getty/dubbo_codec_for_test.go @@ -99,7 +99,7 @@ func (c *DubboTestCodec) EncodeRequest(request *remoting.Request) (*bytes.Buffer return pkg.Marshal() } -// encode heartbeart request +// encode heartbeat request func (c *DubboTestCodec) encodeHeartbeartReqeust(request *remoting.Request) (*bytes.Buffer, error) { header := impl.DubboHeader{ Type: impl.PackageHeartbeat, @@ -171,10 +171,7 @@ func (c *DubboTestCodec) Decode(data []byte) (remoting.DecodeResult, int, error) } func (c *DubboTestCodec) isRequest(data []byte) bool { - if data[2]&byte(0x80) == 0x00 { - return false - } - return true + return data[2]&byte(0x80) != 0x00 } // decode request diff --git a/remoting/getty/getty_client.go b/remoting/getty/getty_client.go index da31c0bf4c..57221cc6d7 100644 --- a/remoting/getty/getty_client.go +++ b/remoting/getty/getty_client.go @@ -40,10 +40,10 @@ import ( var ( errSessionNotExist = perrors.New("session not exist") errClientClosed = perrors.New("client closed") - errClientReadTimeout = perrors.New("client read timeout") + errClientReadTimeout = perrors.New("maybe the client read timeout or fail to decode tcp stream in Writer.Write") clientConf *ClientConfig - clientGrpool *gxsync.TaskPool + clientGrpool gxsync.GenericTaskPool ) // it is init client for single protocol. @@ -99,10 +99,7 @@ func SetClientConf(c ClientConfig) { } func setClientGrpool() { - if clientConf.GrPoolSize > 1 { - clientGrpool = gxsync.NewTaskPool(gxsync.WithTaskPoolTaskPoolSize(clientConf.GrPoolSize), gxsync.WithTaskPoolTaskQueueLength(clientConf.QueueLen), - gxsync.WithTaskPoolTaskQueueNumber(clientConf.QueueNumber)) - } + clientGrpool = gxsync.NewTaskPoolSimple(clientConf.GrPoolSize) } // Options : param config diff --git a/remoting/getty/getty_client_test.go b/remoting/getty/getty_client_test.go index 982c509b32..c32e0c23f4 100644 --- a/remoting/getty/getty_client_test.go +++ b/remoting/getty/getty_client_test.go @@ -396,6 +396,7 @@ func InitTest(t *testing.T) (*Server, *common.URL) { "environment=dev&interface=com.ikurento.user.UserProvider&ip=127.0.0.1&methods=GetUser%2C&" + "module=dubbogo+user-info+server&org=ikurento.com&owner=ZX&pid=1447&revision=0.0.1&" + "side=provider&timeout=3000×tamp=1556509797245&bean.name=UserProvider") + assert.NoError(t, err) // init server userProvider := &UserProvider{} _, err = common.ServiceMap.Register("", url.Protocol, "", "0.0.1", userProvider) @@ -432,7 +433,7 @@ type ( } UserProvider struct { - user map[string]User + //user map[string]User } ) diff --git a/remoting/getty/getty_server.go b/remoting/getty/getty_server.go index d7582e2d14..4930a6ad0e 100644 --- a/remoting/getty/getty_server.go +++ b/remoting/getty/getty_server.go @@ -41,8 +41,7 @@ import ( ) var ( - srvConf *ServerConfig - srvGrpool *gxsync.TaskPool + srvConf *ServerConfig ) func initServer(protocol string) { @@ -76,7 +75,6 @@ func initServer(protocol string) { if err := srvConf.CheckValidity(); err != nil { panic(err) } - SetServerGrpool() } // SetServerConfig set dubbo server config. @@ -87,7 +85,6 @@ func SetServerConfig(s ServerConfig) { logger.Warnf("[ServerConfig CheckValidity] error: %v", err) return } - SetServerGrpool() } // GetServerConfig get getty server config. @@ -95,17 +92,6 @@ func GetServerConfig() ServerConfig { return *srvConf } -// SetServerGrpool set getty server GrPool -func SetServerGrpool() { - if srvConf.GrPoolSize > 1 { - srvGrpool = gxsync.NewTaskPool( - gxsync.WithTaskPoolTaskPoolSize(srvConf.GrPoolSize), - gxsync.WithTaskPoolTaskQueueLength(srvConf.QueueLen), - gxsync.WithTaskPoolTaskQueueNumber(srvConf.QueueNumber), - ) - } -} - // Server define getty server type Server struct { conf ServerConfig @@ -151,13 +137,11 @@ func (s *Server) newSession(session getty.Session) error { session.SetMaxMsgLen(conf.GettySessionParam.MaxMsgLen) session.SetPkgHandler(NewRpcServerPackageHandler(s)) session.SetEventListener(s.rpcHandler) - session.SetWQLen(conf.GettySessionParam.PkgWQSize) session.SetReadTimeout(conf.GettySessionParam.tcpReadTimeout) session.SetWriteTimeout(conf.GettySessionParam.tcpWriteTimeout) session.SetCronPeriod((int)(conf.heartbeatPeriod.Nanoseconds() / 1e6)) session.SetWaitTime(conf.GettySessionParam.waitTimeout) logger.Debugf("server accepts new session:%s\n", session.Stat()) - session.SetTaskPool(srvGrpool) return nil } if _, ok = session.Conn().(*net.TCPConn); !ok { @@ -192,13 +176,11 @@ func (s *Server) newSession(session getty.Session) error { session.SetMaxMsgLen(conf.GettySessionParam.MaxMsgLen) session.SetPkgHandler(NewRpcServerPackageHandler(s)) session.SetEventListener(s.rpcHandler) - session.SetWQLen(conf.GettySessionParam.PkgWQSize) session.SetReadTimeout(conf.GettySessionParam.tcpReadTimeout) session.SetWriteTimeout(conf.GettySessionParam.tcpWriteTimeout) session.SetCronPeriod((int)(conf.heartbeatPeriod.Nanoseconds() / 1e6)) session.SetWaitTime(conf.GettySessionParam.waitTimeout) logger.Debugf("server accepts new session: %s", session.Stat()) - session.SetTaskPool(srvGrpool) return nil } @@ -210,17 +192,15 @@ func (s *Server) Start() { ) addr = s.addr + serverOpts := []getty.ServerOption{getty.WithLocalAddress(addr)} if s.conf.SSLEnabled { - tcpServer = getty.NewTCPServer( - getty.WithLocalAddress(addr), - getty.WithServerSslEnabled(s.conf.SSLEnabled), - getty.WithServerTlsConfigBuilder(config.GetServerTlsConfigBuilder()), - ) - } else { - tcpServer = getty.NewTCPServer( - getty.WithLocalAddress(addr), - ) + serverOpts = append(serverOpts, getty.WithServerSslEnabled(s.conf.SSLEnabled), + getty.WithServerTlsConfigBuilder(config.GetServerTlsConfigBuilder())) } + + serverOpts = append(serverOpts, getty.WithServerTaskPool(gxsync.NewTaskPoolSimple(s.conf.GrPoolSize))) + + tcpServer = getty.NewTCPServer(serverOpts...) tcpServer.RunEventLoop(s.newSession) logger.Debugf("s bind addr{%s} ok!", s.addr) s.tcpServer = tcpServer diff --git a/remoting/getty/listener.go b/remoting/getty/listener.go index b2f7790f2f..fd4c4898f2 100644 --- a/remoting/getty/listener.go +++ b/remoting/getty/listener.go @@ -101,7 +101,7 @@ func (h *RpcClientHandler) OnMessage(session getty.Session, pkg interface{}) { logger.Errorf("illegal package") return } - // get heartbeart request from server + // get heartbeat request from server if result.IsRequest { req := result.Result.(*remoting.Request) if req.Event { @@ -114,12 +114,12 @@ func (h *RpcClientHandler) OnMessage(session getty.Session, pkg interface{}) { reply(session, resp) return } - logger.Errorf("illegal request but not heartbeart. {%#v}", req) + logger.Errorf("illegal request but not heartbeat. {%#v}", req) return } h.timeoutTimes = 0 p := result.Result.(*remoting.Response) - // get heartbeart + // get heartbeat if p.Event { logger.Debugf("get rpc heartbeat response{%#v}", p) if p.Error != nil { @@ -138,7 +138,7 @@ func (h *RpcClientHandler) OnMessage(session getty.Session, pkg interface{}) { // OnCron check the session health periodic. if the session's sessionTimeout has reached, just close the session func (h *RpcClientHandler) OnCron(session getty.Session) { - rpcSession, err := h.conn.getClientRpcSession(session) + rs, err := h.conn.getClientRpcSession(session) if err != nil { logger.Errorf("client.getClientSession(session{%s}) = error{%v}", session.Stat(), perrors.WithStack(err)) @@ -146,7 +146,7 @@ func (h *RpcClientHandler) OnCron(session getty.Session) { } if h.conn.pool.rpcClient.conf.sessionTimeout.Nanoseconds() < time.Since(session.GetActive()).Nanoseconds() { logger.Warnf("session{%s} timeout{%s}, reqNum{%d}", - session.Stat(), time.Since(session.GetActive()).String(), rpcSession.reqNum) + session.Stat(), time.Since(session.GetActive()).String(), rs.reqNum) h.conn.removeSession(session) // -> h.conn.close() -> h.conn.pool.remove(h.conn) return } @@ -237,8 +237,8 @@ func (h *RpcServerHandler) OnMessage(session getty.Session, pkg interface{}) { } h.rwlock.Unlock() - decodeResult, ok := pkg.(remoting.DecodeResult) - if !ok { + decodeResult, drOK := pkg.(remoting.DecodeResult) + if !drOK { logger.Errorf("illegal package{%#v}", pkg) return } @@ -252,7 +252,7 @@ func (h *RpcServerHandler) OnMessage(session getty.Session, pkg interface{}) { res.Handle() return } - logger.Errorf("illegal package but not heartbeart. {%#v}", pkg) + logger.Errorf("illegal package but not heartbeat. {%#v}", pkg) return } req := decodeResult.Result.(*remoting.Request) diff --git a/remoting/getty/listener_test.go b/remoting/getty/listener_test.go index 956ecf9849..2700ed8cd8 100644 --- a/remoting/getty/listener_test.go +++ b/remoting/getty/listener_test.go @@ -23,14 +23,14 @@ import ( ) import ( + "github.com/opentracing/opentracing-go" + "github.com/opentracing/opentracing-go/mocktracer" "github.com/stretchr/testify/assert" ) import ( "github.com/apache/dubbo-go/common/constant" "github.com/apache/dubbo-go/protocol/invocation" - "github.com/opentracing/opentracing-go" - "github.com/opentracing/opentracing-go/mocktracer" ) // test rebuild the ctx @@ -63,7 +63,7 @@ func TestRebuildCtx(t *testing.T) { // Once we decided to transfer more context's key-value, we should change this. // now we only support rebuild the tracing context func rebuildCtx(inv *invocation.RPCInvocation) context.Context { - ctx := context.WithValue(context.Background(), "attachment", inv.Attachments()) + ctx := context.WithValue(context.Background(), constant.DubboCtxKey("attachment"), inv.Attachments()) // actually, if user do not use any opentracing framework, the err will not be nil. spanCtx, err := opentracing.GlobalTracer().Extract(opentracing.TextMap, diff --git a/remoting/getty/pool.go b/remoting/getty/pool.go index c70aeea516..2b1cdfe2f4 100644 --- a/remoting/getty/pool.go +++ b/remoting/getty/pool.go @@ -60,21 +60,20 @@ func newGettyRPCClientConn(pool *gettyRPCClientPool, addr string) (*gettyRPCClie sslEnabled bool ) sslEnabled = pool.sslEnabled + clientOpts := []getty.ClientOption{ + getty.WithServerAddress(addr), + getty.WithConnectionNumber((int)(pool.rpcClient.conf.ConnectionNum)), + getty.WithReconnectInterval(pool.rpcClient.conf.ReconnectInterval), + } if sslEnabled { - gettyClient = getty.NewTCPClient( - getty.WithServerAddress(addr), - getty.WithConnectionNumber((int)(pool.rpcClient.conf.ConnectionNum)), - getty.WithReconnectInterval(pool.rpcClient.conf.ReconnectInterval), - getty.WithClientSslEnabled(pool.sslEnabled), - getty.WithClientTlsConfigBuilder(config.GetClientTlsConfigBuilder()), - ) - } else { - gettyClient = getty.NewTCPClient( - getty.WithServerAddress(addr), - getty.WithConnectionNumber((int)(pool.rpcClient.conf.ConnectionNum)), - getty.WithReconnectInterval(pool.rpcClient.conf.ReconnectInterval), - ) + clientOpts = append(clientOpts, getty.WithClientSslEnabled(pool.sslEnabled), getty.WithClientTlsConfigBuilder(config.GetClientTlsConfigBuilder())) } + + if clientGrpool != nil { + clientOpts = append(clientOpts, getty.WithClientTaskPool(clientGrpool)) + } + + gettyClient = getty.NewTCPClient(clientOpts...) c := &gettyRPCClient{ addr: addr, pool: pool, @@ -91,7 +90,7 @@ func newGettyRPCClientConn(pool *gettyRPCClientPool, addr string) (*gettyRPCClie break } - if time.Now().Sub(start) > connectTimeout { + if time.Since(start) > connectTimeout { c.gettyClient.Close() return nil, perrors.New(fmt.Sprintf("failed to create client connection to %s in %s", addr, connectTimeout)) } @@ -136,13 +135,11 @@ func (c *gettyRPCClient) newSession(session getty.Session) error { session.SetMaxMsgLen(conf.GettySessionParam.MaxMsgLen) session.SetPkgHandler(NewRpcClientPackageHandler(c.pool.rpcClient)) session.SetEventListener(NewRpcClientHandler(c)) - session.SetWQLen(conf.GettySessionParam.PkgWQSize) session.SetReadTimeout(conf.GettySessionParam.tcpReadTimeout) session.SetWriteTimeout(conf.GettySessionParam.tcpWriteTimeout) session.SetCronPeriod((int)(conf.heartbeatPeriod.Nanoseconds() / 1e6)) session.SetWaitTime(conf.GettySessionParam.waitTimeout) logger.Debugf("client new session:%s\n", session.Stat()) - session.SetTaskPool(clientGrpool) return nil } if tcpConn, ok = session.Conn().(*net.TCPConn); !ok { @@ -171,15 +168,11 @@ func (c *gettyRPCClient) newSession(session getty.Session) error { session.SetMaxMsgLen(conf.GettySessionParam.MaxMsgLen) session.SetPkgHandler(NewRpcClientPackageHandler(c.pool.rpcClient)) session.SetEventListener(NewRpcClientHandler(c)) - session.SetWQLen(conf.GettySessionParam.PkgWQSize) session.SetReadTimeout(conf.GettySessionParam.tcpReadTimeout) session.SetWriteTimeout(conf.GettySessionParam.tcpWriteTimeout) session.SetCronPeriod((int)(conf.heartbeatPeriod.Nanoseconds() / 1e6)) session.SetWaitTime(conf.GettySessionParam.waitTimeout) logger.Debugf("client new session:%s\n", session.Stat()) - - session.SetTaskPool(clientGrpool) - return nil } @@ -270,33 +263,29 @@ func (c *gettyRPCClient) updateSession(session getty.Session) { func (c *gettyRPCClient) getClientRpcSession(session getty.Session) (rpcSession, error) { var ( - err error - rpcSession rpcSession + err error + rs rpcSession ) c.lock.RLock() defer c.lock.RUnlock() if c.sessions == nil { - return rpcSession, errClientClosed + return rs, errClientClosed } err = errSessionNotExist for _, s := range c.sessions { if s.session == session { - rpcSession = *s + rs = *s err = nil break } } - return rpcSession, perrors.WithStack(err) + return rs, perrors.WithStack(err) } func (c *gettyRPCClient) isAvailable() bool { - if c.selectSession() == nil { - return false - } - - return true + return c.selectSession() != nil } func (c *gettyRPCClient) close() error { @@ -314,9 +303,7 @@ func (c *gettyRPCClient) close() error { c.gettyClient = nil sessions = make([]*rpcSession, 0, len(c.sessions)) - for _, s := range c.sessions { - sessions = append(sessions, s) - } + sessions = append(sessions, c.sessions...) c.sessions = c.sessions[:0] }() @@ -369,16 +356,16 @@ func (p *gettyRPCClientPool) close() { } func (p *gettyRPCClientPool) getGettyRpcClient(addr string) (*gettyRPCClient, error) { - conn, err := p.get() - if err == nil && conn == nil { + conn, connErr := p.get() + if connErr == nil && conn == nil { // create new conn - rpcClientConn, err := newGettyRPCClientConn(p, addr) - if err == nil { + rpcClientConn, rpcErr := newGettyRPCClientConn(p, addr) + if rpcErr == nil { p.put(rpcClientConn) } - return rpcClientConn, perrors.WithStack(err) + return rpcClientConn, perrors.WithStack(rpcErr) } - return conn, perrors.WithStack(err) + return conn, perrors.WithStack(connErr) } func (p *gettyRPCClientPool) get() (*gettyRPCClient, error) { diff --git a/remoting/kubernetes/client.go b/remoting/kubernetes/client.go index 5399e5f21b..ce6bcccea5 100644 --- a/remoting/kubernetes/client.go +++ b/remoting/kubernetes/client.go @@ -186,12 +186,12 @@ func ValidateClient(container clientFacade) error { // NewMockClient exports for registry package test func NewMockClient(podList *v1.PodList) (*Client, error) { - ctx, cancel := context.WithCancel(context.Background()) controller, err := newDubboRegistryController(ctx, common.CONSUMER, func() (kubernetes.Interface, error) { return fake.NewSimpleClientset(podList), nil }) if err != nil { + cancel() return nil, perrors.WithMessage(err, "new dubbo-registry controller") } diff --git a/remoting/kubernetes/client_test.go b/remoting/kubernetes/client_test.go index fdaee96b44..9cc421225e 100644 --- a/remoting/kubernetes/client_test.go +++ b/remoting/kubernetes/client_test.go @@ -317,7 +317,8 @@ func TestClientGetChildrenKVList(t *testing.T) { wc, done, err := client.WatchWithPrefix(prefix) if err != nil { - t.Fatal(err) + t.Error(err) + return } wg.Done() diff --git a/remoting/kubernetes/facade_test.go b/remoting/kubernetes/facade_test.go index 00e2e1171c..a6c6c02b51 100644 --- a/remoting/kubernetes/facade_test.go +++ b/remoting/kubernetes/facade_test.go @@ -19,7 +19,6 @@ package kubernetes import ( "strconv" - "sync" "testing" ) @@ -30,9 +29,9 @@ import ( type mockFacade struct { *common.URL - client *Client - cltLock sync.Mutex - done chan struct{} + client *Client + //cltLock sync.Mutex + //done chan struct{} } func (r *mockFacade) Client() *Client { diff --git a/remoting/kubernetes/watch.go b/remoting/kubernetes/watch.go index 07eeb09b4d..7bb5ef1754 100644 --- a/remoting/kubernetes/watch.go +++ b/remoting/kubernetes/watch.go @@ -116,21 +116,17 @@ type watcherSetImpl struct { // closeWatchers // when the watcher-set was closed func (s *watcherSetImpl) closeWatchers() { - - select { - case <-s.ctx.Done(): - - // parent ctx be canceled, close the watch-set's watchers - s.lock.Lock() - watchers := s.watchers - s.lock.Unlock() - - for _, w := range watchers { - // stop data stream - // close(w.ch) - // stop watcher - w.stop() - } + <-s.ctx.Done() + // parent ctx be canceled, close the watch-set's watchers + s.lock.Lock() + watchers := s.watchers + s.lock.Unlock() + + for _, w := range watchers { + // stop data stream + // close(w.ch) + // stop watcher + w.stop() } } diff --git a/remoting/kubernetes/watch_test.go b/remoting/kubernetes/watch_test.go index 8889103be2..efefcc5dc9 100644 --- a/remoting/kubernetes/watch_test.go +++ b/remoting/kubernetes/watch_test.go @@ -42,7 +42,8 @@ func TestWatchSet(t *testing.T) { defer wg.Done() w, err := s.Watch("key-1", false) if err != nil { - t.Fatal(err) + t.Error(err) + return } for { select { @@ -64,7 +65,8 @@ func TestWatchSet(t *testing.T) { defer wg.Done() w, err := s.Watch("key", true) if err != nil { - t.Fatal(err) + t.Error(err) + return } for { @@ -86,7 +88,8 @@ func TestWatchSet(t *testing.T) { Key: "key-" + strconv.Itoa(i), Value: strconv.Itoa(i), }); err != nil { - t.Fatal(err) + t.Error(err) + return } }(i) } diff --git a/remoting/zookeeper/client.go b/remoting/zookeeper/client.go index fbd90762eb..852a5564f9 100644 --- a/remoting/zookeeper/client.go +++ b/remoting/zookeeper/client.go @@ -128,18 +128,18 @@ func ValidateZookeeperClient(container ZkClientFacade, opts ...Option) error { if container.ZkClient() == nil { // in dubbo, every registry only connect one node, so this is []string{r.Address} - timeout, err := time.ParseDuration(url.GetParam(constant.REGISTRY_TIMEOUT_KEY, constant.DEFAULT_REG_TIMEOUT)) - if err != nil { - logger.Errorf("timeout config %v is invalid ,err is %v", - url.GetParam(constant.REGISTRY_TIMEOUT_KEY, constant.DEFAULT_REG_TIMEOUT), err.Error()) - return perrors.WithMessagef(err, "newZookeeperClient(address:%+v)", url.Location) + timeout, paramErr := time.ParseDuration(url.GetParam(constant.REGISTRY_TIMEOUT_KEY, constant.DEFAULT_REG_TIMEOUT)) + if paramErr != nil { + logger.Errorf("timeout config %v is invalid, err is %v", + url.GetParam(constant.REGISTRY_TIMEOUT_KEY, constant.DEFAULT_REG_TIMEOUT), paramErr.Error()) + return perrors.WithMessagef(paramErr, "newZookeeperClient(address:%+v)", url.Location) } zkAddresses := strings.Split(url.Location, ",") - newClient, err := NewZookeeperClient(options.zkName, zkAddresses, timeout) - if err != nil { + newClient, cltErr := NewZookeeperClient(options.zkName, zkAddresses, timeout) + if cltErr != nil { logger.Warnf("newZookeeperClient(name{%s}, zk address{%v}, timeout{%d}) = error{%v}", - options.zkName, url.Location, timeout.String(), err) - return perrors.WithMessagef(err, "newZookeeperClient(address:%+v)", url.Location) + options.zkName, url.Location, timeout.String(), cltErr) + return perrors.WithMessagef(cltErr, "newZookeeperClient(address:%+v)", url.Location) } container.SetZkClient(newClient) connected = true diff --git a/remoting/zookeeper/client_test.go b/remoting/zookeeper/client_test.go index af09edba48..3d0949973d 100644 --- a/remoting/zookeeper/client_test.go +++ b/remoting/zookeeper/client_test.go @@ -96,7 +96,7 @@ func TestCreate(t *testing.T) { ts, z, event, err := NewMockZookeeperClient("test", 15*time.Second) assert.NoError(t, err) defer func() { - err := ts.Stop() + _ = ts.Stop() assert.Nil(t, err) }() err = z.Create("test1/test2/test3/test4") @@ -110,7 +110,7 @@ func TestCreateDelete(t *testing.T) { ts, z, event, err := NewMockZookeeperClient("test", 15*time.Second) assert.NoError(t, err) defer func() { - err := ts.Stop() + _ = ts.Stop() assert.Nil(t, err) }() @@ -127,7 +127,7 @@ func TestRegisterTemp(t *testing.T) { ts, z, event, err := NewMockZookeeperClient("test", 15*time.Second) assert.NoError(t, err) defer func() { - err := ts.Stop() + _ = ts.Stop() assert.Nil(t, err) }() err = z.Create("/test1/test2/test3") @@ -144,7 +144,7 @@ func TestRegisterTempSeq(t *testing.T) { ts, z, event, err := NewMockZookeeperClient("test", 15*time.Second) assert.NoError(t, err) defer func() { - err := ts.Stop() + _ = ts.Stop() assert.Nil(t, err) }() err = z.Create("/test1/test2/test3") diff --git a/remoting/zookeeper/listener.go b/remoting/zookeeper/listener.go index e5ddcadeac..3eda6d9698 100644 --- a/remoting/zookeeper/listener.go +++ b/remoting/zookeeper/listener.go @@ -135,9 +135,10 @@ func (l *ZkEventListener) handleZkNodeEvent(zkPath string, children []string, li newChildren, err := l.client.GetChildren(zkPath) if err != nil { if err == errNilChildren { - content, _, err := l.client.Conn.Get(zkPath) - if err != nil { - logger.Errorf("Get new node path {%v} 's content error,message is {%v}", zkPath, perrors.WithStack(err)) + content, _, connErr := l.client.Conn.Get(zkPath) + if connErr != nil { + logger.Errorf("Get new node path {%v} 's content error,message is {%v}", + zkPath, perrors.WithStack(connErr)) } else { listener.DataChange(remoting.Event{Path: zkPath, Action: remoting.EventTypeUpdate, Content: string(content)}) } @@ -158,9 +159,10 @@ func (l *ZkEventListener) handleZkNodeEvent(zkPath string, children []string, li newNode = path.Join(zkPath, n) logger.Infof("add zkNode{%s}", newNode) - content, _, err := l.client.Conn.Get(newNode) - if err != nil { - logger.Errorf("Get new node path {%v} 's content error,message is {%v}", newNode, perrors.WithStack(err)) + content, _, connErr := l.client.Conn.Get(newNode) + if connErr != nil { + logger.Errorf("Get new node path {%v} 's content error,message is {%v}", + newNode, perrors.WithStack(connErr)) } if !listener.DataChange(remoting.Event{Path: zkPath, Action: remoting.EventTypeAdd, Content: string(content)}) { @@ -362,9 +364,9 @@ func (l *ZkEventListener) ListenServiceEvent(conf *common.URL, zkPath string, li }(zkPath, listener) } -func (l *ZkEventListener) valid() bool { - return l.client.ZkConnValid() -} +//func (l *ZkEventListener) valid() bool { +// return l.client.ZkConnValid() +//} // Close will let client listen exit func (l *ZkEventListener) Close() { diff --git a/remoting/zookeeper/listener_test.go b/remoting/zookeeper/listener_test.go index 07976dc31d..896bdc6070 100644 --- a/remoting/zookeeper/listener_test.go +++ b/remoting/zookeeper/listener_test.go @@ -95,7 +95,7 @@ func TestListener(t *testing.T) { ` var wait sync.WaitGroup ts, client, event := initZkData(t) - defer func () { + defer func() { if err := ts.Stop(); err != nil { t.Errorf("ts.Stop() = error: %v", err) }