Merge ade5a1f into c5802c5

config/testdata/consumer_config.yml

@@ -87,7 +87,6 @@ protocol_conf:
       keep_alive_period: "120s"
       tcp_r_buf_size: 262144
       tcp_w_buf_size: 65536
-      pkg_wq_size: 512
       tcp_read_timeout: "1s"
       tcp_write_timeout: "5s"
       wait_timeout: "1s"

config/testdata/consumer_config_with_configcenter.yml

@@ -36,7 +36,6 @@ protocol_conf:
       keep_alive_period: "120s"
       tcp_r_buf_size: 262144
       tcp_w_buf_size: 65536
-      pkg_wq_size: 512
       tcp_read_timeout: "1s"
       tcp_write_timeout: "5s"
       wait_timeout: "1s"

config/testdata/consumer_config_withoutProtocol.yml

@@ -73,7 +73,6 @@ protocol_conf:
       keep_alive_period: "120s"
       tcp_r_buf_size: 262144
       tcp_w_buf_size: 65536
-      pkg_wq_size: 512
       tcp_read_timeout: "1s"
       tcp_write_timeout: "5s"
       wait_timeout: "1s"

config/testdata/provider_config.yml

@@ -92,7 +92,6 @@ protocol_conf:
       keep_alive_period: "120s"
       tcp_r_buf_size: 262144
       tcp_w_buf_size: 65536
-      pkg_wq_size: 512
       tcp_read_timeout: "1s"
       tcp_write_timeout: "5s"
       wait_timeout: "1s"

config/testdata/provider_config_withoutProtocol.yml

@@ -72,7 +72,6 @@ protocol_conf:
       keep_alive_period: "120s"
       tcp_r_buf_size: 262144
       tcp_w_buf_size: 65536
-      pkg_wq_size: 512
       tcp_read_timeout: "1s"
       tcp_write_timeout: "5s"
       wait_timeout: "1s"

config_center/apollo/impl_test.go

@@ -110,7 +110,6 @@ var mockConfigRes = `{
 		"application.environment": "dev",
 		"services.UserProvider.protocol": "dubbo",
 		"application.organization": "ikurento.com",
-		"protocol_conf.dubbo.getty_session_param.pkg_wq_size": "512",
 		"services.UserProvider.methods[0].loadbalance": "random"
 	},
 	"releaseKey": "20191104105242-0f13805d89f834a4"

go.mod

@@ -7,7 +7,7 @@ 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.2
-	github.com/apache/dubbo-getty v1.4.3
+	github.com/apache/dubbo-getty v1.4.5
 	github.com/apache/dubbo-go-hessian2 v1.9.2
 	github.com/creasty/defaults v1.5.1
 	github.com/dubbogo/go-zookeeper v1.0.3

go.sum

@@ -53,8 +53,8 @@ github.com/alibaba/sentinel-golang v1.0.2/go.mod h1:QsB99f/z35D2AiMrAWwgWE85kDTk
 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/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
-github.com/apache/dubbo-getty v1.4.3 h1:PCKpryDasKOxwT5MBC6MIMO+0NLOaHF6Xco9YXQw7HI=
-github.com/apache/dubbo-getty v1.4.3/go.mod h1:ansXgKxxyhCOiQL29nO5ce1MDcEKmCyZuNR9oMs3hek=
+github.com/apache/dubbo-getty v1.4.5 h1:MptKbjC0n2Mo/8eFPwirSInH2BfdNG4IZch43PdNvIM=
+github.com/apache/dubbo-getty v1.4.5/go.mod h1:mcDyiu7M/TVrYDyL8TxDemQkOdvEqqHSQ4jOuYejY1w=
 github.com/apache/dubbo-go-hessian2 v1.9.1/go.mod h1:xQUjE7F8PX49nm80kChFvepA/AvqAZ0oh/UaB6+6pBE=
 github.com/apache/dubbo-go-hessian2 v1.9.2 h1:XuI8KvENSfKiAhiCBS4RNihmQDoPNmGWKT3gTui0p9A=
 github.com/apache/dubbo-go-hessian2 v1.9.2/go.mod h1:xQUjE7F8PX49nm80kChFvepA/AvqAZ0oh/UaB6+6pBE=
@@ -148,7 +148,7 @@ github.com/docker/spdystream v0.0.0-20160310174837-449fdfce4d96/go.mod h1:Qh8CwZ
 github.com/dubbogo/go-zookeeper v1.0.3 h1:UkuY+rBsxdT7Bs63QAzp9z7XqQ53W1j8E5rwl83me8g=
 github.com/dubbogo/go-zookeeper v1.0.3/go.mod h1:fn6n2CAEer3novYgk9ULLwAjuV8/g4DdC2ENwRb6E+c=
 github.com/dubbogo/gost v1.9.0/go.mod h1:pPTjVyoJan3aPxBPNUX0ADkXjPibLo+/Ib0/fADXSG8=
-github.com/dubbogo/gost v1.10.1/go.mod h1:+mQGS51XQEUWZP2JeGZTxJwipjRKtJO7Tr+FOg+72rI=
+github.com/dubbogo/gost v1.11.12/go.mod h1:vIcP9rqz2KsXHPjsAwIUtfJIJjppQLQDcYaZTy/61jI=
 github.com/dubbogo/gost v1.11.14 h1:9lfcdILOmqTOVAW1fPHa5uf1NrD6jlIOBe4vf8576yQ=
 github.com/dubbogo/gost v1.11.14/go.mod h1:vIcP9rqz2KsXHPjsAwIUtfJIJjppQLQDcYaZTy/61jI=
 github.com/dubbogo/jsonparser v1.0.1/go.mod h1:tYAtpctvSP/tWw4MeelsowSPgXQRVHHWbqL6ynps8jU=
@@ -467,6 +467,7 @@ github.com/modern-go/reflect2 v0.0.0-20180320133207-05fbef0ca5da/go.mod h1:bx2lN
 github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
 github.com/modern-go/reflect2 v1.0.1 h1:9f412s+6RmYXLWZSEzVVgPGK7C2PphHj5RJrvfx9AWI=
 github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
+github.com/montanaflynn/stats v0.6.6/go.mod h1:etXPPgVO6n31NxCd9KQUMvCM+ve0ruNzt6R8Bnaayow=
 github.com/morikuni/aec v1.0.0/go.mod h1:BbKIizmSmc5MMPqRYbxO4ZU0S0+P200+tUnFx7PXmsc=
 github.com/mschoch/smat v0.2.0 h1:8imxQsjDm8yFEAVBe7azKmKSgzSkZXDuKkSq9374khM=
 github.com/mschoch/smat v0.2.0/go.mod h1:kc9mz7DoBKqDyiRL7VZN8KvXQMWeTaVnttLRXOlotKw=
@@ -595,7 +596,6 @@ github.com/samuel/go-zookeeper v0.0.0-20190923202752-2cc03de413da/go.mod h1:gi+0
 github.com/satori/go.uuid v1.2.0 h1:0uYX9dsZ2yD7q2RtLRtPSdGDWzjeM3TbMJP9utgA0ww=
 github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
 github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
-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/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
@@ -647,7 +647,6 @@ github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXf
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
-github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/subosito/gotenv v1.2.0 h1:Slr1R9HxAlEKefgq5jn9U+DnETlIUa6HfgEzj0g5d7s=

metadata/service/exporter/configurable/exporter_test.go

@@ -46,7 +46,6 @@ func TestConfigurableExporter(t *testing.T) {
 			KeepAlivePeriod:  "120s",
 			TcpRBufSize:      262144,
 			TcpWBufSize:      65536,
-			PkgWQSize:        512,
 			TcpReadTimeout:   "1s",
 			TcpWriteTimeout:  "5s",
 			WaitTimeout:      "1s",

protocol/dubbo/dubbo_invoker_test.go

@@ -112,7 +112,6 @@ func InitTest(t *testing.T) (protocol.Protocol, *common.URL) {
 			KeepAlivePeriod:  "120s",
 			TcpRBufSize:      262144,
 			TcpWBufSize:      65536,
-			PkgWQSize:        512,
 			TcpReadTimeout:   "4s",
 			TcpWriteTimeout:  "5s",
 			WaitTimeout:      "1s",
@@ -130,7 +129,6 @@ func InitTest(t *testing.T) (protocol.Protocol, *common.URL) {
 			KeepAlivePeriod:  "120s",
 			TcpRBufSize:      262144,
 			TcpWBufSize:      65536,
-			PkgWQSize:        512,
 			TcpReadTimeout:   "1s",
 			TcpWriteTimeout:  "5s",
 			WaitTimeout:      "1s",

protocol/dubbo/dubbo_protocol_test.go

@@ -52,7 +52,6 @@ func initDubboInvokerTest() {
 			KeepAlivePeriod:  "120s",
 			TcpRBufSize:      262144,
 			TcpWBufSize:      65536,
-			PkgWQSize:        512,
 			TcpReadTimeout:   "1s",
 			TcpWriteTimeout:  "5s",
 			WaitTimeout:      "1s",
@@ -71,7 +70,6 @@ func initDubboInvokerTest() {
 			KeepAlivePeriod:  "120s",
 			TcpRBufSize:      262144,
 			TcpWBufSize:      65536,
-			PkgWQSize:        512,
 			TcpReadTimeout:   "4s",
 			TcpWriteTimeout:  "5s",
 			WaitTimeout:      "1s",

registry/etcdv3/listener.go

@@ -24,6 +24,7 @@ import (
 
 import (
 	gxchan "github.com/dubbogo/gost/container/chan"
+
 	perrors "github.com/pkg/errors"
 )
 

registry/etcdv3/listener_test.go

@@ -24,8 +24,10 @@ import (
 )
 
 import (
-	getty "github.com/apache/dubbo-getty"
+	gxtime "github.com/dubbogo/gost/time"
+
 	"github.com/stretchr/testify/suite"
+
 	"go.etcd.io/etcd/server/v3/embed"
 )
 
@@ -57,7 +59,7 @@ func (suite *RegistryTestSuite) SetupSuite() {
 	select {
 	case <-e.Server.ReadyNotify():
 		t.Log("Server is ready!")
-	case <-getty.GetTimeWheel().After(60 * time.Second):
+	case <-gxtime.After(60 * time.Second):
 		e.Server.Stop() // trigger a shutdown
 		t.Logf("Server took too long to start!")
 	}

registry/etcdv3/registry.go

@@ -27,6 +27,7 @@ import (
 
 import (
 	gxetcd "github.com/dubbogo/gost/database/kv/etcd/v3"
+
 	perrors "github.com/pkg/errors"
 )
 

registry/etcdv3/service_discovery.go

@@ -28,7 +28,9 @@ import (
 	gxset "github.com/dubbogo/gost/container/set"
 	gxetcd "github.com/dubbogo/gost/database/kv/etcd/v3"
 	gxpage "github.com/dubbogo/gost/hash/page"
+
 	"github.com/hashicorp/vault/sdk/helper/jsonutil"
+
 	perrors "github.com/pkg/errors"
 )
 

registry/kubernetes/listener.go

@@ -23,6 +23,7 @@ import (
 
 import (
 	gxchan "github.com/dubbogo/gost/container/chan"
+
 	perrors "github.com/pkg/errors"
 )
 

registry/kubernetes/registry.go

@@ -25,8 +25,10 @@ import (
 )
 
 import (
-	getty "github.com/apache/dubbo-getty"
+	gxtime "github.com/dubbogo/gost/time"
+
 	perrors "github.com/pkg/errors"
+
 	v1 "k8s.io/api/core/v1"
 )
 
@@ -210,7 +212,7 @@ LOOP:
 			// try to connect to kubernetes,
 			failTimes = 0
 			for {
-				after := getty.GetTimeWheel().After(timeSecondDuration(failTimes * ConnDelay))
+				after := gxtime.After(timeSecondDuration(failTimes * ConnDelay))
 				select {
 				case <-r.Done():
 					logger.Warnf("(KubernetesProviderRegistry)reconnectKubernetes Registry goroutine exit now...")

registry/kubernetes/registry_test.go

@@ -28,6 +28,7 @@ import (
 
 import (
 	"github.com/stretchr/testify/assert"
+
 	v1 "k8s.io/api/core/v1"
 )
 

remoting/getty/config.go

@@ -39,7 +39,6 @@ type (
 		keepAlivePeriod  time.Duration
 		TcpRBufSize      int    `default:"262144" yaml:"tcp_r_buf_size" json:"tcp_r_buf_size,omitempty"`
 		TcpWBufSize      int    `default:"65536" yaml:"tcp_w_buf_size" json:"tcp_w_buf_size,omitempty"`
-		PkgWQSize        int    `default:"1024" yaml:"pkg_wq_size" json:"pkg_wq_size,omitempty"`
 		TcpReadTimeout   string `default:"1s" yaml:"tcp_read_timeout" json:"tcp_read_timeout,omitempty"`
 		tcpReadTimeout   time.Duration
 		TcpWriteTimeout  string `default:"5s" yaml:"tcp_write_timeout" json:"tcp_write_timeout,omitempty"`
@@ -122,7 +121,6 @@ func GetDefaultClientConfig() ClientConfig {
 			KeepAlivePeriod:  "180s",
 			TcpRBufSize:      262144,
 			TcpWBufSize:      65536,
-			PkgWQSize:        512,
 			TcpReadTimeout:   "1s",
 			TcpWriteTimeout:  "5s",
 			WaitTimeout:      "1s",
@@ -147,7 +145,6 @@ func GetDefaultServerConfig() ServerConfig {
 			KeepAlivePeriod:  "180s",
 			TcpRBufSize:      262144,
 			TcpWBufSize:      65536,
-			PkgWQSize:        512,
 			TcpReadTimeout:   "1s",
 			TcpWriteTimeout:  "5s",
 			WaitTimeout:      "1s",

remoting/getty/dubbo_codec_for_test.go

@@ -26,6 +26,7 @@ import (
 
 import (
 	hessian "github.com/apache/dubbo-go-hessian2"
+
 	perrors "github.com/pkg/errors"
 )
 

remoting/getty/getty_client.go

@@ -25,9 +25,14 @@ import (
 
 import (
 	"github.com/apache/dubbo-getty"
+
 	gxsync "github.com/dubbogo/gost/sync"
+	gxtime "github.com/dubbogo/gost/time"
+
 	perrors "github.com/pkg/errors"
+
 	"go.uber.org/atomic"
+
 	"gopkg.in/yaml.v2"
 )
 
@@ -201,7 +206,7 @@ func (c *Client) Request(request *remoting.Request, timeout time.Duration, respo
 	}
 
 	select {
-	case <-getty.GetTimeWheel().After(timeout):
+	case <-gxtime.After(timeout):
 		return perrors.WithStack(errClientReadTimeout)
 	case <-response.Done:
 		err = response.Err

remoting/getty/getty_client_test.go

@@ -28,7 +28,9 @@ import (
 
 import (
 	hessian "github.com/apache/dubbo-go-hessian2"
+
 	perrors "github.com/pkg/errors"
+
 	"github.com/stretchr/testify/assert"
 )
 
@@ -343,7 +345,6 @@ func InitTest(t *testing.T) (*Server, *common.URL) {
 			KeepAlivePeriod:  "120s",
 			TcpRBufSize:      262144,
 			TcpWBufSize:      65536,
-			PkgWQSize:        512,
 			TcpReadTimeout:   "4s",
 			TcpWriteTimeout:  "5s",
 			WaitTimeout:      "1s",
@@ -362,7 +363,6 @@ func InitTest(t *testing.T) (*Server, *common.URL) {
 			KeepAlivePeriod:  "120s",
 			TcpRBufSize:      262144,
 			TcpWBufSize:      65536,
-			PkgWQSize:        512,
 			TcpReadTimeout:   "1s",
 			TcpWriteTimeout:  "5s",
 			WaitTimeout:      "1s",

remoting/getty/getty_server.go

@@ -25,8 +25,11 @@ import (
 
 import (
 	"github.com/apache/dubbo-getty"
+
 	gxsync "github.com/dubbogo/gost/sync"
+
 	perrors "github.com/pkg/errors"
+
 	"gopkg.in/yaml.v2"
 )
 

remoting/getty/listener.go

@@ -26,7 +26,11 @@ import (
 
 import (
 	"github.com/apache/dubbo-getty"
+
 	hessian "github.com/apache/dubbo-go-hessian2"
+
+	gxtime "github.com/dubbogo/gost/time"
+
 	perrors "github.com/pkg/errors"
 )
 
@@ -378,7 +382,7 @@ func heartbeat(session getty.Session, timeout time.Duration, callBack func(err e
 	go func() {
 		var err1 error
 		select {
-		case <-getty.GetTimeWheel().After(timeout):
+		case <-gxtime.After(timeout):
 			err1 = errHeartbeatReadTimeout
 		case <-resp.Done:
 			err1 = resp.Err

remoting/getty/listener_test.go

@@ -25,6 +25,7 @@ import (
 import (
 	"github.com/opentracing/opentracing-go"
 	"github.com/opentracing/opentracing-go/mocktracer"
+
 	"github.com/stretchr/testify/assert"
 )
 

remoting/getty/pool.go

@@ -29,6 +29,7 @@ import (
 
 import (
 	"github.com/apache/dubbo-getty"
+
 	perrors "github.com/pkg/errors"
 )
 

remoting/getty/readwriter.go

@@ -24,7 +24,9 @@ import (
 
 import (
 	"github.com/apache/dubbo-getty"
+
 	hessian "github.com/apache/dubbo-go-hessian2"
+
 	perrors "github.com/pkg/errors"
 )
 
@@ -87,14 +89,6 @@ func (p *RpcClientPackageHandler) Write(ss getty.Session, pkg interface{}) ([]by
 	return nil, perrors.New("invalid rpc request")
 }
 
-////////////////////////////////////////////
-// RpcServerPackageHandler
-////////////////////////////////////////////
-
-//var (
-//	rpcServerPkgHandler = &RpcServerPackageHandler{}
-//)
-
 // RpcServerPackageHandler Read data from client and Write data to client
 type RpcServerPackageHandler struct {
 	server *Server