diff --git a/.github/workflows/github-actions.yml b/.github/workflows/github-actions.yml index 48a3feba4b29e9de039cbf73b4ab76f7e1d6ac30..40ef43c8c125fdcad481f1436e64006149ca3611 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 0000000000000000000000000000000000000000..6f214f445ac5544b26d4fa704b7ea4524535f8bd --- /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 4f79ecf3bc1439e00c88c4d0de74cc7a4bf26909..0000000000000000000000000000000000000000 --- 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/NOTICE b/NOTICE index 1120c200c997fe6befbe3f78d95e9bdb8a05a487..e0f4af68f6ce22fb817ba1173663395ee8f2bfd9 100644 --- a/NOTICE +++ b/NOTICE @@ -1,5 +1,5 @@ Apache Dubbo-go -Copyright 2018-2020 The Apache Software Foundation +Copyright 2018-2021 The Apache Software Foundation This product includes software developed at The Apache Software Foundation (http://www.apache.org/). diff --git a/README.md b/README.md index f950a4b3e9c4e115eda2f0e99379286c4c66d24f..8af021bb079f47939e5f3d69037906a0c0587d56 100644 --- a/README.md +++ b/README.md @@ -16,14 +16,14 @@ Apache License, Version 2.0 ## Release note ## +[v1.5.5 - Jan 5, 2021](https://github.com/apache/dubbo-go/releases/tag/v1.5.5) + [v1.4.5 - Nov 18, 2020](https://github.com/apache/dubbo-go/releases/tag/v1.4.5) [v1.5.4 - Nov 1, 2020](https://github.com/apache/dubbo-go/releases/tag/v1.5.4) [v1.5.3 - Sep 23, 2020](https://github.com/apache/dubbo-go/releases/tag/v1.5.3) -[v1.5.2 - discard]() - [v1.5.1 - Aug 23, 2020](https://github.com/apache/dubbo-go/releases/tag/v1.5.1) [v1.5.0 - July 24, 2020](https://github.com/apache/dubbo-go/releases/tag/v1.5.0) @@ -260,7 +260,7 @@ If you are using [apache/dubbo-go](github.com/apache/dubbo-go) and think that it </a> </td> </tr> - <tr></tr> + <tr></tr> </tbody> </table> </div> diff --git a/README_CN.md b/README_CN.md index ff72428d977f8b24c7c60b228108987c823c9efa..079cfa57449734f291ab5252ed86b01ac59c542c 100644 --- a/README_CN.md +++ b/README_CN.md @@ -15,14 +15,14 @@ Apache License, Version 2.0 ## 鍙戝竷鏃ュ織 ## +[v1.5.5 - 2021骞�1鏈�5鏃(https://github.com/apache/dubbo-go/releases/tag/v1.5.5) + [v1.4.5 - 2020骞�11鏈�18鏃(https://github.com/apache/dubbo-go/releases/tag/v1.4.5) [v1.5.4 - 2020骞�11鏈�1鏃(https://github.com/apache/dubbo-go/releases/tag/v1.5.4) [v1.5.3 - 2020骞�9鏈�23鏃(https://github.com/apache/dubbo-go/releases/tag/v1.5.3) -[v1.5.2 - 鑸嶅純]() - [v1.5.1 - 2020骞�8鏈�23鏃(https://github.com/apache/dubbo-go/releases/tag/v1.5.1) [v1.5.0 - 2020骞�7鏈�24鏃(https://github.com/apache/dubbo-go/releases/tag/v1.5.0) @@ -67,11 +67,11 @@ Apache License, Version 2.0 * Jsonrpc2.0 * [gRPC](https://github.com/apache/dubbo-go/pull/311) * [RESTful](https://github.com/apache/dubbo-go/pull/352) - + - 璺敱鍣� * [Condition router](https://github.com/apache/dubbo-go/pull/294) * [Health check router](https://github.com/apache/dubbo-go/pull/389) - + - 娉ㄥ唽涓績 * ZooKeeper * [etcd v3](https://github.com/apache/dubbo-go/pull/148) @@ -112,7 +112,7 @@ Apache License, Version 2.0 - 璋冪敤 * [娉涘寲璋冪敤](https://github.com/apache/dubbo-go/pull/122) - + - 鐩戞帶 * Opentracing API * [Prometheus](https://github.com/apache/dubbo-go/pull/342) @@ -258,7 +258,7 @@ make test </a> </td> </tr> - <tr></tr> + <tr></tr> </tbody> </table> -</div> \ No newline at end of file +</div> diff --git a/cluster/cluster_impl/available_cluster.go b/cluster/cluster_impl/available_cluster.go index ebd5767e4c320f10c8911cf9ac3f2c81deaafb0e..1f41890f5824ff0e2ca542b923b7cf85a6e3a2a1 100644 --- a/cluster/cluster_impl/available_cluster.go +++ b/cluster/cluster_impl/available_cluster.go @@ -39,6 +39,6 @@ func NewAvailableCluster() cluster.Cluster { } // Join returns a baseClusterInvoker instance -func (cluser *availableCluster) Join(directory cluster.Directory) protocol.Invoker { +func (cluster *availableCluster) Join(directory cluster.Directory) protocol.Invoker { return NewAvailableClusterInvoker(directory) } diff --git a/cluster/router/chain/chain_test.go b/cluster/router/chain/chain_test.go index 06e905a351d9b7f508342e73ddbdc092601c6299..d56a1f4416152c4c4ea5bd7dc25e42439720369a 100644 --- a/cluster/router/chain/chain_test.go +++ b/cluster/router/chain/chain_test.go @@ -131,7 +131,7 @@ conditions: _, err = z.Conn.Set(path, []byte(testyml), 0) assert.NoError(t, err) defer func() { - err := ts.Stop() + _ = ts.Stop() assert.NoError(t, err) z.Close() }() @@ -215,7 +215,7 @@ conditions: _, err = z.Conn.Set(path, []byte(testyml), 0) assert.NoError(t, err) defer func() { - err := ts.Stop() + _ = ts.Stop() assert.NoError(t, err) z.Close() }() @@ -246,7 +246,7 @@ func TestRouterChainRouteNoRoute(t *testing.T) { ts, z, _, err := zookeeper.NewMockZookeeperClient("test", 15*time.Second) assert.Nil(t, err) defer func() { - err := ts.Stop() + _ = ts.Stop() assert.NoError(t, err) z.Close() }() diff --git a/common/constant/default.go b/common/constant/default.go index 4165942a615e220f6384a898b07c04bafd39c3b0..bbe022cb2efebf7dc716f324253743a5a5a0aff5 100644 --- a/common/constant/default.go +++ b/common/constant/default.go @@ -81,7 +81,7 @@ const ( const ( SIMPLE_METADATA_SERVICE_NAME = "MetadataService" - DEFAULT_REVIESION = "N/A" + DEFAULT_REVISION = "N/A" ) const ( diff --git a/common/constant/version.go b/common/constant/version.go index 730224376054a36b0c7cfeda7d5ea5e7ce058618..0403e3aff7d77ed614c727addf9e9a162e60b69c 100644 --- a/common/constant/version.go +++ b/common/constant/version.go @@ -19,9 +19,9 @@ package constant const ( // Version apache/dubbo-go version - Version = "1.3.0" + Version = "1.5.5" // Name module name Name = "dubbogo" // Date release date - DATE = "2020/01/12" + DATE = "2021/01/05" ) diff --git a/common/proxy/proxy_test.go b/common/proxy/proxy_test.go index d91db4841252abe4ee97c855062e01a93f62b757..9ce36835c896ef7172bb4b9e49a35bcc4218c9d7 100644 --- a/common/proxy/proxy_test.go +++ b/common/proxy/proxy_test.go @@ -132,11 +132,19 @@ func TestProxyImplementForContext(t *testing.T) { p := NewProxy(invoker, nil, map[string]string{constant.ASYNC_KEY: "false"}) s := &TestService{} p.Implement(s) +<<<<<<< HEAD attahments1 := make(map[string]interface{}, 4) attahments1["k1"] = "v1" attahments1["k2"] = "v2" ctx := context.WithValue(context.Background(), constant.AttachmentKey, attahments1) r, err := p.Get().(*TestService).MethodSix(ctx, "xxx") +======= + attachments1 := make(map[string]interface{}, 4) + attachments1["k1"] = "v1" + attachments1["k2"] = "v2" + context := context.WithValue(context.Background(), constant.AttachmentKey, attachments1) + r, err := p.Get().(*TestService).MethodSix(context, "xxx") +>>>>>>> develop v1 := r.(map[string]interface{}) assert.NoError(t, err) assert.Equal(t, v1["TestProxyInvoker"], "TestProxyInvokerValue") diff --git a/common/rpc_service.go b/common/rpc_service.go index 9f587f3a689f8a8d6b257799233c0c7b6dc3a4ac..30d72c1420f254182f5d2f16c6fc9444881f0b71 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/config/config_loader_test.go b/config/config_loader_test.go index ef76bc090877281d2e12b48378476b4042b55943..b9a025e8bc7e146558249ca6d8aa00455ff55523 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 8bc1b91ee5cd35b423039ef7e2442df7d2a103f1..fd49390aa2b15938a8a4aa321d47ceafef4d9be9 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 cb0f03d54a2ea180543ec528e3a46be45067179c..5b74f5e15013c07de000cc18cab98238b96ac060 100644 --- a/config_center/apollo/impl.go +++ b/config_center/apollo/impl.go @@ -146,8 +146,8 @@ func (c *apolloConfiguration) getAddressWithProtocolPrefix(url *common.URL) stri address := url.Location converted := address if len(address) != 0 { - address := regexp.MustCompile(`\s+`).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/configurator/override_test.go b/config_center/configurator/override_test.go index bb9c36776337c5dde1c5b2dcc9dd59ee82067a02..4d2552d1845e57ada38b635bf3cb047891e73abf 100644 --- a/config_center/configurator/override_test.go +++ b/config_center/configurator/override_test.go @@ -37,7 +37,7 @@ const ( failover = "failover" ) -func TestConfigureVerison2p6(t *testing.T) { +func TestConfigureVersion2p6(t *testing.T) { url, err := common.NewURL("override://0.0.0.0:0/com.xxx.mock.userProvider?group=1&version=1&cluster=failfast&application=BDTService") assert.NoError(t, err) configurator := extension.GetConfigurator(defaults, url) @@ -49,7 +49,7 @@ func TestConfigureVerison2p6(t *testing.T) { assert.Equal(t, failfast, providerUrl.GetParam(constant.CLUSTER_KEY, "")) } -func TestConfigureVerisonOverrideAddr(t *testing.T) { +func TestConfigureVersionOverrideAddr(t *testing.T) { url, err := common.NewURL("override://0.0.0.0:0/com.xxx.mock.userProvider?group=1&version=1&cluster=failfast&application=BDTService&providerAddresses=127.0.0.2:20001|127.0.0.3:20001") assert.NoError(t, err) configurator := extension.GetConfigurator(defaults, url) @@ -61,7 +61,7 @@ func TestConfigureVerisonOverrideAddr(t *testing.T) { assert.Equal(t, failover, providerUrl.GetParam(constant.CLUSTER_KEY, "")) } -func TestConfigureVerison2p6WithIp(t *testing.T) { +func TestConfigureVersion2p6WithIp(t *testing.T) { url, err := common.NewURL("override://127.0.0.1:20001/com.xxx.mock.userProvider?group=1&version=1&cluster=failfast&application=BDTService") assert.NoError(t, err) configurator := extension.GetConfigurator(defaults, url) @@ -74,7 +74,7 @@ func TestConfigureVerison2p6WithIp(t *testing.T) { } -func TestConfigureVerison2p7(t *testing.T) { +func TestConfigureVersion2p7(t *testing.T) { url, err := common.NewURL("jsonrpc://0.0.0.0:20001/com.xxx.mock.userProvider?group=1&version=1&cluster=failfast&application=BDTService&configVersion=1.0&side=provider") assert.NoError(t, err) configurator := extension.GetConfigurator(defaults, url) diff --git a/config_center/file/impl.go b/config_center/file/impl.go index 9afe7c6524ab02d368f91283e39e3e4c1deacbac..6489a073deff643ad7eecc7f3c26ef0b8899ac53 100644 --- a/config_center/file/impl.go +++ b/config_center/file/impl.go @@ -24,7 +24,6 @@ import ( "os" "os/exec" "os/user" - "path" "path/filepath" "runtime" "strings" @@ -41,11 +40,19 @@ import ( "github.com/apache/dubbo-go/config_center/parser" ) +var ( + osType = runtime.GOOS +) + +const ( + windowsOS = "windows" +) + const ( - PARAM_NAME_PREFIX = "dubbo.config-center." - CONFIG_CENTER_DIR_PARAM_NAME = PARAM_NAME_PREFIX + "dir" - CONFIG_CENTER_ENCODING_PARAM_NAME = PARAM_NAME_PREFIX + "encoding" - DEFAULT_CONFIG_CENTER_ENCODING = "UTF-8" + ParamNamePrefix = "dubbo.config-center." + ConfigCenterDirParamName = ParamNamePrefix + "dir" + ConfigCenterEncodingParamName = ParamNamePrefix + "encoding" + defaultConfigCenterEncoding = "UTF-8" ) // FileSystemDynamicConfiguration @@ -59,24 +66,14 @@ type FileSystemDynamicConfiguration struct { } func newFileSystemDynamicConfiguration(url *common.URL) (*FileSystemDynamicConfiguration, error) { - encode := url.GetParam(CONFIG_CENTER_ENCODING_PARAM_NAME, DEFAULT_CONFIG_CENTER_ENCODING) + encode := url.GetParam(ConfigCenterEncodingParamName, defaultConfigCenterEncoding) - root := url.GetParam(CONFIG_CENTER_DIR_PARAM_NAME, "") + root := url.GetParam(ConfigCenterDirParamName, "") var c *FileSystemDynamicConfiguration - if _, err := os.Stat(root); err != nil { - // not exist, use default, /XXX/xx/.dubbo/config-center - if rp, err := Home(); err != nil { - return nil, perrors.WithStack(err) - } else { - root = path.Join(rp, ".dubbo", "config-center") - } - } - if _, err := os.Stat(root); err != nil { - // it must be dir, if not exist, will create - if err = createDir(root); err != nil { - return nil, perrors.WithStack(err) - } + root, err := mkdirIfNecessary(root) + if err != nil { + return nil, err } c = &FileSystemDynamicConfiguration{ @@ -195,18 +192,18 @@ func (fsdc *FileSystemDynamicConfiguration) Close() error { // GetPath get path func (fsdc *FileSystemDynamicConfiguration) GetPath(key string, group string) string { if len(key) == 0 { - return path.Join(fsdc.rootPath, group) + return filepath.Join(fsdc.rootPath, group) } if len(group) == 0 { group = config_center.DEFAULT_GROUP } - return path.Join(fsdc.rootPath, group, key) + return filepath.Join(fsdc.rootPath, group, adapterKey(key)) } func (fsdc *FileSystemDynamicConfiguration) deleteDelay(path string) (bool, error) { - if path == "" { + if len(path) == 0 { return false, nil } @@ -226,9 +223,7 @@ func (fsdc *FileSystemDynamicConfiguration) write2File(fp string, value string) } func forceMkdirParent(fp string) error { - pd := getParentDirectory(fp) - - return createDir(pd) + return createDir(getParentDirectory(fp)) } func createDir(path string) error { @@ -250,6 +245,7 @@ func substr(s string, pos, length int) string { if l > len(runes) { l = len(runes) } + return string(runes[pos:l]) } @@ -264,7 +260,7 @@ func Home() (string, error) { } // cross compile support - if "windows" == runtime.GOOS { + if windowsOS == osType { return homeWindows() } @@ -287,7 +283,7 @@ func homeUnix() (string, error) { } result := strings.TrimSpace(stdout.String()) - if result == "" { + if len(result) == 0 { return "", errors.New("blank output when reading home directory") } @@ -298,12 +294,66 @@ func homeWindows() (string, error) { drive := os.Getenv("HOMEDRIVE") homePath := os.Getenv("HOMEPATH") home := drive + homePath - if drive == "" || homePath == "" { + if len(drive) == 0 || len(homePath) == 0 { home = os.Getenv("USERPROFILE") } - if home == "" { + if len(home) == 0 { return "", errors.New("HOMEDRIVE, HOMEPATH, and USERPROFILE are blank") } return home, nil } + +func mkdirIfNecessary(urlRoot string) (string, error) { + if !legalPath(urlRoot) { + // not exist, use default, mac is: /XXX/xx/.dubbo/config-center + rp, err := Home() + if err != nil { + return "", perrors.WithStack(err) + } + + urlRoot = adapterUrl(rp) + } + + if _, err := os.Stat(urlRoot); err != nil { + // it must be dir, if not exist, will create + if err = createDir(urlRoot); err != nil { + return "", perrors.WithStack(err) + } + } + + return urlRoot, nil +} + +func legalPath(path string) bool { + if len(path) == 0 { + return false + } + if _, err := os.Stat(path); err != nil { + return false + } + + return true +} + +func adapterUrl(rp string) string { + if osType == windowsOS { + return filepath.Join(rp, "_dubbo", "config-center") + } + + return filepath.Join(rp, ".dubbo", "config-center") +} + +// used for GetPath. param key default is instance's id. +// e.g: (ip:port) 127.0.0.1:20081, in windows env, will change to 127_0_0_1_20081 +func adapterKey(key string) string { + if len(key) == 0 { + return "" + } + + if osType == windowsOS { + return strings.ReplaceAll(strings.ReplaceAll(key, ".", "_"), ":", "_") + } + + return key +} 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 4ff0a232e47e76a46db64812cf26f3d09b063b03..5baa70a5bc4eebe162243a485521543802a51f2c 100644 --- a/filter/filter_impl/tps/tps_limiter_method_service_test.go +++ b/filter/filter_impl/tps/tps_limiter_method_service_test.go @@ -22,7 +22,6 @@ import ( "testing" ) import ( - "github.com/apache/dubbo-go/filter" "github.com/golang/mock/gomock" "github.com/stretchr/testify/assert" ) @@ -31,6 +30,7 @@ import ( "github.com/apache/dubbo-go/common" "github.com/apache/dubbo-go/common/constant" "github.com/apache/dubbo-go/common/extension" + "github.com/apache/dubbo-go/filter" "github.com/apache/dubbo-go/protocol/invocation" ) diff --git a/go.mod b/go.mod index 68465d19463939ffefe15a88eb0b9a82274803a6..18d85a650f27430f163b78f3bba92236c1a3b524 100644 --- a/go.mod +++ b/go.mod @@ -1,19 +1,17 @@ module github.com/apache/dubbo-go -go 1.13 - require ( github.com/NYTimes/gziphandler v1.1.1 // indirect github.com/RoaringBitmap/roaring v0.5.5 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-go-hessian2 v1.8.0 + github.com/apache/dubbo-getty v1.4.1 + github.com/apache/dubbo-go-hessian2 v1.8.2 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 @@ -32,7 +30,7 @@ require ( github.com/magiconair/properties v1.8.4 github.com/mitchellh/mapstructure v1.4.0 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd - github.com/nacos-group/nacos-sdk-go v1.0.1 + github.com/nacos-group/nacos-sdk-go v1.0.3 github.com/opentracing/opentracing-go v1.2.0 github.com/pierrec/lz4 v2.2.6+incompatible // indirect github.com/pkg/errors v0.9.1 diff --git a/go.sum b/go.sum index 7b1c1716aa6701a495dbed2f54d411927c7bd890..2c07089b681b1944ae8fd33fac7dbf18857d7155 100644 --- a/go.sum +++ b/go.sum @@ -62,8 +62,9 @@ github.com/RoaringBitmap/roaring v0.5.5 h1:naNqvO1mNnghk2UvcsqnzHDBn9DRbCIRy94Gm github.com/RoaringBitmap/roaring v0.5.5/go.mod h1:puNo5VdzwbaIQxSiDIwfXl4Hnc+fbovcX4IW/dSTtUk= github.com/Shopify/sarama v1.19.0/go.mod h1:FVkBWblsNy7DGZRfXLU0O9RCGt5g3g3yEuWXgklEdEo= 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,10 +81,10 @@ 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-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/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.2 h1:CQq2Mmlrk6Fqmudwl9Dqps8drTrBFnmXRlzgOjj0FqA= +github.com/apache/dubbo-go-hessian2 v1.8.2/go.mod h1:xQUjE7F8PX49nm80kChFvepA/AvqAZ0oh/UaB6+6pBE= github.com/apache/thrift v0.12.0/go.mod h1:cp2SuWMxlEZw2r+iP2GNCdIi4C1qmUzdZFSVb+bacwQ= github.com/apache/thrift v0.13.0/go.mod h1:cp2SuWMxlEZw2r+iP2GNCdIi4C1qmUzdZFSVb+bacwQ= github.com/armon/circbuf v0.0.0-20150827004946-bbbad097214e h1:QEF07wC0T1rKkctt1RINW/+RMTVmiwxETico2l3gxJA= @@ -173,8 +174,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= @@ -231,8 +233,9 @@ github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9 github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk= github.com/go-logfmt/logfmt v0.5.0/go.mod h1:wCYkCAKZfumFQihp8CzCvQ3paCTfi41vtzG1KdI/P7A= 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= @@ -563,8 +566,8 @@ github.com/munnerz/goautoneg v0.0.0-20120707110453-a547fc61f48d/go.mod h1:+n7T8m github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= github.com/mxk/go-flowrate v0.0.0-20140419014527-cca7078d478f/go.mod h1:ZdcZmHo+o7JKHSa8/e818NopupXU1YMK5fe1lsApnBw= -github.com/nacos-group/nacos-sdk-go v1.0.1 h1:VNmXGlSS28xOmkO5Nxk5WRp6f1HMosAmG9pDtcnUFcw= -github.com/nacos-group/nacos-sdk-go v1.0.1/go.mod h1:hlAPn3UdzlxIlSILAyOXKxjFSvDJ9oLzTJ9hLAK1KzA= +github.com/nacos-group/nacos-sdk-go v1.0.3 h1:A2tCWcjuP6bSEjEfNwNnrY+9M0h13XRMDyLY+DPqHMI= +github.com/nacos-group/nacos-sdk-go v1.0.3/go.mod h1:hlAPn3UdzlxIlSILAyOXKxjFSvDJ9oLzTJ9hLAK1KzA= github.com/nats-io/jwt v0.3.0/go.mod h1:fRYCDE99xlTsqUzISS1Bi75UBJ6ljOJQOAAu5VglpSg= github.com/nats-io/jwt v0.3.2/go.mod h1:/euKqTS1ZD+zzjYrY7pseZrTtWQSjujC7xjPc8wL6eU= github.com/nats-io/nats-server/v2 v2.1.2/go.mod h1:Afk+wRZqkMQs/p45uXdrVLuab3gwv3Z8C4HTBu8GD/k= @@ -689,6 +692,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/service/remote/service_proxy.go b/metadata/service/remote/service_proxy.go index d0caa182b2efdc709af11e4bc40f807e89276a56..e0cd6e0783fe4572e0a69cf0694d70f74ba46b42 100644 --- a/metadata/service/remote/service_proxy.go +++ b/metadata/service/remote/service_proxy.go @@ -135,7 +135,7 @@ func (m metadataServiceProxy) Version() (string, error) { func newMetadataServiceProxy(ins registry.ServiceInstance) service.MetadataService { revision := ins.GetMetadata()[constant.EXPORTED_SERVICES_REVISION_PROPERTY_NAME] if len(revision) == 0 { - revision = constant.DEFAULT_REVIESION + revision = constant.DEFAULT_REVISION } return &metadataServiceProxy{ diff --git a/protocol/dubbo/hessian2/const.go b/protocol/dubbo/hessian2/const.go index 0c95390fbfff152a926ca2d1ff09682a4cb893ae..96e6eea8ea4cd3088e0d32157c013e93a93cad4e 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 1afa4ec96eccbb8077852dfcc020e0eb05be3257..5ffebde54b951997057a1f161dbcf6aae1d4a7fd 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 efcbd95d293fa84e8dfd6eddd9fdc14f05c3e6a2..94aa34dee895bc8d441f710edecabe0326da1400 100644 --- a/protocol/dubbo/hessian2/hessian_request.go +++ b/protocol/dubbo/hessian2/hessian_request.go @@ -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) } diff --git a/protocol/dubbo/impl/codec.go b/protocol/dubbo/impl/codec.go index 17e7b57b45f2334e06b757ee07e3a9ef034c920a..6c9816f0ff9a4faee7750d16468fa65b9aa803f2 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 492a87a194eeb9ef673f1c83a499872cfb9724cb..c9b92bad3b402f6f8656ca47c1a2954c7881d68b 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/grpc/client.go b/protocol/grpc/client.go index 33f0a088dcd6800ce7376faa3d5e05ab5d62596d..24ab125625465068e90f2b0d96033b6fc3ffd2f6 100644 --- a/protocol/grpc/client.go +++ b/protocol/grpc/client.go @@ -93,15 +93,15 @@ type Client struct { func NewClient(url *common.URL) *Client { // if global trace instance was set , it means trace function enabled. If not , will return Nooptracer tracer := opentracing.GlobalTracer() - dailOpts := make([]grpc.DialOption, 0, 4) + dialOpts := make([]grpc.DialOption, 0, 4) maxMessageSize, _ := strconv.Atoi(url.GetParam(constant.MESSAGE_SIZE_KEY, "4")) - dailOpts = append(dailOpts, grpc.WithInsecure(), grpc.WithBlock(), grpc.WithUnaryInterceptor( + dialOpts = append(dialOpts, grpc.WithInsecure(), grpc.WithBlock(), grpc.WithUnaryInterceptor( otgrpc.OpenTracingClientInterceptor(tracer, otgrpc.LogPayloads())), grpc.WithDefaultCallOptions( grpc.CallContentSubtype(clientConf.ContentSubType), grpc.MaxCallRecvMsgSize(1024*1024*maxMessageSize), grpc.MaxCallSendMsgSize(1024*1024*maxMessageSize))) - conn, err := grpc.Dial(url.Location, dailOpts...) + conn, err := grpc.Dial(url.Location, dialOpts...) if err != nil { panic(err) } diff --git a/protocol/jsonrpc/http.go b/protocol/jsonrpc/http.go index 037744cf09c4aa229fd7d5f0a7444fd86e70f06f..11051df8ac05010e9e07fe14d44768dda9342201 100644 --- a/protocol/jsonrpc/http.go +++ b/protocol/jsonrpc/http.go @@ -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/json.go b/protocol/jsonrpc/json.go index 57d36d2eadd761aa359d55c350772745f1cde819..81ca512271baa19a103b42c059ba63fd14513377 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/registry/consul/registry.go b/registry/consul/registry.go index 869e1c1d75769447f8e3d2c0fb7817b1dc2db163..4d3718e58cd0f98b931ed525fe61f33713b6e71f 100644 --- a/registry/consul/registry.go +++ b/registry/consul/registry.go @@ -23,6 +23,7 @@ import ( ) import ( + getty "github.com/apache/dubbo-getty" consul "github.com/hashicorp/consul/api" perrors "github.com/pkg/errors" ) @@ -56,6 +57,10 @@ type consulRegistry struct { // Done field represents whether // consul registry is closed. done chan struct{} + + // registeredURLs field represents all URLs that have been registered + // will be unregistered when destroyed + registeredURLs []*common.URL } func newConsulRegistry(url *common.URL) (registry.Registry, error) { @@ -91,6 +96,7 @@ func (r *consulRegistry) Register(url *common.URL) error { // register actually register the @url func (r *consulRegistry) register(url *common.URL) error { + r.registeredURLs = append(r.registeredURLs, url.Clone()) service, err := buildService(url) if err != nil { return err @@ -188,25 +194,26 @@ func (r *consulRegistry) IsAvailable() bool { // Destroy consul registry center func (r *consulRegistry) Destroy() { - if r.URL != nil { - done := make(chan struct{}, 1) - go func() { - defer func() { - if e := recover(); e != nil { - logger.Errorf("consulRegistry destroy with panic: %v", e) - } - done <- struct{}{} - }() - if err := r.UnRegister(r.URL); err != nil { - logger.Errorf("consul registry unregister with err: %s", err.Error()) + done := make(chan struct{}, 1) + go func() { + defer func() { + if e := recover(); e != nil { + logger.Errorf("consulRegistry destroy with panic: %v", e) } + done <- struct{}{} }() - select { - case <-done: - logger.Infof("consulRegistry unregister done") - case <-time.After(registryDestroyDefaultTimeout): - logger.Errorf("consul unregister timeout") + for _, url := range r.registeredURLs { + if err := r.UnRegister(url); err != nil { + logger.Errorf("consul registry unregister with err: %s", err.Error()) + } } + }() + select { + case <-done: + logger.Infof("consulRegistry unregister done") + case <-getty.GetTimeWheel().After(registryDestroyDefaultTimeout): + logger.Errorf("consul unregister timeout") } + close(r.done) } diff --git a/registry/consul/service_discovery.go b/registry/consul/service_discovery.go index 6674909b15afc225515ff080fb341da65c17a05f..fba142e04e4979b9c4fe123cc121111838ae2d4e 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" @@ -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/etcdv3/service_discovery.go b/registry/etcdv3/service_discovery.go index b9d8b5f895cf8257fa75da87c5548c7a8bdad3b9..ca6016e49b4e174e5843b3c08beea7b559bbfb9b 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" ) @@ -180,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. @@ -202,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 diff --git a/registry/etcdv3/service_discovery_test.go b/registry/etcdv3/service_discovery_test.go index d8e3f1a2864150cc1f1e8996bc7c53e115dbef45..5609cf7be6831d93b78945fa7bf8959c321f0d1e 100644 --- a/registry/etcdv3/service_discovery_test.go +++ b/registry/etcdv3/service_discovery_test.go @@ -75,6 +75,6 @@ func Test_newEtcdV3ServiceDiscovery(t *testing.T) { func TestEtcdV3ServiceDiscovery_GetDefaultPageSize(t *testing.T) { setUp() - serviceDiscovry := &etcdV3ServiceDiscovery{} - assert.Equal(t, registry.DefaultPageSize, serviceDiscovry.GetDefaultPageSize()) + serviceDiscovery := &etcdV3ServiceDiscovery{} + assert.Equal(t, registry.DefaultPageSize, serviceDiscovery.GetDefaultPageSize()) } diff --git a/registry/event/event_publishing_service_deiscovery_test.go b/registry/event/event_publishing_service_deiscovery_test.go index 54752c03c0de598226270b27c8d7d0f3621d07d1..504f7b5faed374795bd534c9e97cf5b42fc94d4c 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 3ee2f4a44946065cdf7489abc391df41f251d810..773eee6e83c6b3e24c293053b55b028d0e7b2e9a 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 21482d329e0e252a4af612e872c4153a5afdd25c..768a1c2a3d8a208d637e374632da530a43465a4e 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" ) @@ -69,7 +69,7 @@ func newFileSystemServiceDiscovery(name string) (registry.ServiceDiscovery, erro fdcf := extension.GetConfigCenterFactory(constant.FILE_KEY) p := path.Join(rp, ".dubbo", constant.REGISTRY_KEY) url, _ := common.NewURL("") - url.AddParamAvoidNil(file.CONFIG_CENTER_DIR_PARAM_NAME, p) + url.AddParamAvoidNil(file.ConfigCenterDirParamName, p) c, err := fdcf.GetDynamicConfiguration(url) if err != nil { return nil, perrors.WithStack(err) diff --git a/registry/file/service_discovery_test.go b/registry/file/service_discovery_test.go index 2f52eb52e1be1f7f06f29d30cd7761e71ecd1bee..0062eae32cc04fd58ae4398ac7a28aabc54892c1 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 55be4c392abcc14557c60af15b92bc84855b4b31..c4a0352cfaa3273c92296be6279c0aae71f0448b 100644 --- a/registry/kubernetes/registry.go +++ b/registry/kubernetes/registry.go @@ -39,11 +39,6 @@ import ( "github.com/apache/dubbo-go/remoting/kubernetes" ) -//var ( -// processID = "" -// localIP = "" -//) - const ( Name = "kubernetes" ConnDelay = 3 diff --git a/registry/nacos/service_discovery.go b/registry/nacos/service_discovery.go index 9a2e1f11729d33e6b22bcd2514d97ac5adc29888..4533e7bc6b1427f1713e8b61bf2f125c0d0bde99 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" @@ -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 2af1d5b6f564b5389574cc9107ef912cd506db8a..aa044ad04a4f5d982914535a2f7fec55beed304c 100644 --- a/registry/nacos/service_discovery_test.go +++ b/registry/nacos/service_discovery_test.go @@ -166,8 +166,8 @@ func TestNacosServiceDiscovery_CRUD(t *testing.T) { func TestNacosServiceDiscovery_GetDefaultPageSize(t *testing.T) { prepareData() - serviceDiscovry, _ := extension.GetServiceDiscovery(constant.NACOS_KEY, testName) - assert.Equal(t, registry.DefaultPageSize, serviceDiscovry.GetDefaultPageSize()) + serviceDiscovery, _ := extension.GetServiceDiscovery(constant.NACOS_KEY, testName) + assert.Equal(t, registry.DefaultPageSize, serviceDiscovery.GetDefaultPageSize()) } func prepareData() { diff --git a/registry/service_discovery.go b/registry/service_discovery.go index cb7a3c0182ff88995ab9dd6c920523225c3cb36c..5ab768351508ed289cbe9b17a75ea9b8e95fff72 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/servicediscovery/service_discovery_registry_test.go b/registry/servicediscovery/service_discovery_registry_test.go index a65e6e404a22c2304b4f2ad44fd0a0bc5ede2d19..391d92c8ad19c871756ca7426531bb01a7d54876 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 ( diff --git a/registry/zookeeper/registry.go b/registry/zookeeper/registry.go index 8b61e8092ce5b0d8081095555ed49734db9a48b4..3232ee624de909320b129a15adebef5aec5108ac 100644 --- a/registry/zookeeper/registry.go +++ b/registry/zookeeper/registry.go @@ -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/service_discovery.go b/registry/zookeeper/service_discovery.go index a515e22d041859211a7a8dcb28bb94b32ed6ae42..095a78dc3c1650197ba1ded75218e8fd0020b6ea 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" ) @@ -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 b7d4677deb4efc56f1d246dc1eabdb0033c94702..c9e14afd50af36e5e729f0efb768b37c56f0adf3 100644 --- a/registry/zookeeper/service_discovery_test.go +++ b/registry/zookeeper/service_discovery_test.go @@ -81,8 +81,7 @@ func TestCURDZookeeperServiceDiscovery(t *testing.T) { 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" @@ -151,8 +150,7 @@ func TestAddListenerZookeeperServiceDiscovery(t *testing.T) { sd, err := newZookeeperServiceDiscovery(testName) assert.Nil(t, err) defer func() { - err := sd.Destroy() - assert.Nil(t, err) + _ = sd.Destroy() }() err = sd.Register(®istry.DefaultServiceInstance{ @@ -164,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) diff --git a/remoting/exchange.go b/remoting/exchange.go index 5fbd8ae9b449103d5f7e3b19db19bc6a9c73593d..ad136a7c5ea4946e8467c69cfbf43b275df8ceef 100644 --- a/remoting/exchange.go +++ b/remoting/exchange.go @@ -129,7 +129,7 @@ type AsyncCallbackResponse struct { Reply interface{} } -// the client sends requst to server, there is one pendingResponse at client side to wait the response from server +// the client sends request to server, there is one pendingResponse at client side to wait the response from server type PendingResponse struct { seq int64 Err error diff --git a/remoting/getty/getty_client.go b/remoting/getty/getty_client.go index da31c0bf4c3a0fec8284f613a6409b100ffbb12f..57221cc6d7f2b9668e6a0518b6678b37e4924bf1 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_server.go b/remoting/getty/getty_server.go index d7582e2d14b92b11fe41008eebf0c10c401df272..4930a6ad0ee0dbcde50a16e8df6b6245fb87a617 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 a07726e5a609a40734946028f69de4d1653521a5..fd4c4898f2a74fae0cca1cf6c2e86e3e2f75017e 100644 --- a/remoting/getty/listener.go +++ b/remoting/getty/listener.go @@ -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 } diff --git a/remoting/getty/pool.go b/remoting/getty/pool.go index 63c9c1a1fb4659ec2d4b07f14008acc99b9b9c3a..2b1cdfe2f49bfee0d5291f2eaf4183db71b8aa31 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, @@ -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,25 +263,25 @@ 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 { @@ -363,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/zookeeper/client.go b/remoting/zookeeper/client.go index fbd90762eb34f361a38486ef2d8f5f10699a96f7..852a5564f955e9734be58515cf0e6840e69f2f78 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 af09edba48815044ff9002a68b438463113c4652..3d0949973d0aa01acf5178d20afb854debb5c763 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 7b90e523364b6a1a77b8e30ef73ddca6caffda6a..3eda6d9698e5ec244bb6c5763ebc8e459e03310e 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)}) { diff --git a/remoting/zookeeper/listener_test.go b/remoting/zookeeper/listener_test.go index 07976dc31db5909a261ed7f76223a12220249207..896bdc60702d17520c62cd0621b00da795c90e6d 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) }