Merge remote-tracking branch 'upstream/1.5' into fix_zk_too_many_tcp_conn

.github/workflows/github-actions.yml

@@ -35,7 +35,7 @@ jobs:
       uses: actions/checkout@v2
 
     - name: Cache dependencies
-      uses: actions/cache@v2
+      uses: actions/cache@v2.1.4
       with:
         # Cache
         path: ~/go/pkg/mod

cluster/cluster_impl/zone_aware_cluster_invoker.go

@@ -43,7 +43,7 @@ func newZoneAwareClusterInvoker(directory cluster.Directory) protocol.Invoker {
 	invoke := &zoneAwareClusterInvoker{
 		baseClusterInvoker: newBaseClusterInvoker(directory),
 	}
-	// add self to interceptor
+	// add local to interceptor
 	invoke.interceptor = invoke
 	return invoke
 }

cluster/router/local/factory.go

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package local
+
+import (
+	"github.com/apache/dubbo-go/cluster/router"
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/common/constant"
+	"github.com/apache/dubbo-go/common/extension"
+)
+
+func init() {
+	extension.SetRouterFactory(constant.LocalPriorityRouterName, newLocalPriorityRouteFactory)
+}
+
+// LocalPriorityRouteFactory
+type LocalPriorityRouteFactory struct {
+}
+
+// newLocalPriorityRouteFactory construct a new LocalDiscRouteFactory
+func newLocalPriorityRouteFactory() router.PriorityRouterFactory {
+	return &LocalPriorityRouteFactory{}
+}
+
+// NewPriorityRouter construct a new NewLocalDiscRouter via url
+func (f *LocalPriorityRouteFactory) NewPriorityRouter(url *common.URL) (router.PriorityRouter, error) {
+	return NewLocalPriorityRouter(url)
+}

cluster/router/local/factory_test.go

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package local
+
+import (
+	"context"
+	"testing"
+)
+
+import (
+	"github.com/stretchr/testify/assert"
+)
+
+import (
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/protocol"
+)
+
+// nolint
+type MockInvoker struct {
+	url *common.URL
+}
+
+// nolint
+func NewMockInvoker(url *common.URL) *MockInvoker {
+	return &MockInvoker{
+		url: url,
+	}
+}
+
+// nolint
+func (bi *MockInvoker) GetUrl() *common.URL {
+	return bi.url
+}
+
+// nolint
+func (bi *MockInvoker) IsAvailable() bool {
+	return true
+}
+
+// nolint
+func (bi *MockInvoker) IsDestroyed() bool {
+	return true
+}
+
+// nolint
+func (bi *MockInvoker) Invoke(_ context.Context, _ protocol.Invocation) protocol.Result {
+	return nil
+}
+
+// nolint
+func (bi *MockInvoker) Destroy() {
+}
+
+// nolint
+func TestLocalDiscRouteFactory(t *testing.T) {
+	factory := newLocalPriorityRouteFactory()
+	assert.NotNil(t, factory)
+}

cluster/router/local/self_priority_route.go

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package local
+
+import (
+	"github.com/RoaringBitmap/roaring"
+)
+
+import (
+	"github.com/apache/dubbo-go/cluster/router"
+	"github.com/apache/dubbo-go/cluster/router/utils"
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/common/logger"
+	"github.com/apache/dubbo-go/protocol"
+)
+
+const (
+	localPriority = "local-priority"
+	name          = "local-priority-router"
+)
+
+// LocalPriorityRouter provides a ip-same-first routing logic
+// if there is not provider with same ip as consumer, it would not filter any invoker
+// if exists same ip invoker, it would retains this invoker only
+type LocalPriorityRouter struct {
+	url     *common.URL
+	localIP string
+}
+
+// NewLocalPriorityRouter construct an LocalPriorityRouter via url
+func NewLocalPriorityRouter(url *common.URL) (router.PriorityRouter, error) {
+	r := &LocalPriorityRouter{
+		url:     url,
+		localIP: url.Ip,
+	}
+	return r, nil
+}
+
+// Route gets a list of match-logic invoker
+func (r *LocalPriorityRouter) Route(invokers *roaring.Bitmap, cache router.Cache, url *common.URL, invocation protocol.Invocation) *roaring.Bitmap {
+	addrPool := cache.FindAddrPool(r)
+	// Add localPriority invoker to the list
+	selectedInvokers := utils.JoinIfNotEqual(addrPool[localPriority], invokers)
+	// If all invokers are considered not match, downgrade to all invoker
+	if selectedInvokers.IsEmpty() {
+		logger.Warnf(" Now all invokers are not match, so downgraded to all! Service: [%s]", url.ServiceKey())
+		return invokers
+	}
+	return selectedInvokers
+}
+
+// Pool separates same ip invoker from others.
+func (r *LocalPriorityRouter) Pool(invokers []protocol.Invoker) (router.AddrPool, router.AddrMetadata) {
+	rb := make(router.AddrPool, 8)
+	rb[localPriority] = roaring.NewBitmap()
+	localIpFound := false
+	for i, invoker := range invokers {
+		if invoker.GetUrl().Ip == r.localIP {
+			rb[localPriority].Add(uint32(i))
+			localIpFound = true
+		}
+	}
+	if localIpFound {
+		// found local desc
+		logger.Debug("found local ip ")
+		return rb, nil
+	}
+	for i, _ := range invokers {
+		rb[localPriority].Add(uint32(i))
+	}
+	return rb, nil
+}
+
+// ShouldPool will always return true to make sure local call logic constantly.
+func (r *LocalPriorityRouter) ShouldPool() bool {
+	return true
+}
+
+func (r *LocalPriorityRouter) Name() string {
+	return name
+}
+
+// Priority
+func (r *LocalPriorityRouter) Priority() int64 {
+	return 0
+}
+
+// URL Return URL in router
+func (r *LocalPriorityRouter) URL() *common.URL {
+	return r.url
+}

cluster/router/local/self_priority_route_test.go

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package local
+
+import (
+	"fmt"
+	"testing"
+)
+
+import (
+	"github.com/stretchr/testify/assert"
+)
+
+import (
+	"github.com/apache/dubbo-go/cluster/router"
+	"github.com/apache/dubbo-go/cluster/router/chain"
+	"github.com/apache/dubbo-go/cluster/router/utils"
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/protocol"
+	"github.com/apache/dubbo-go/protocol/invocation"
+)
+
+const (
+	localDiscRoute1010IP         = "192.168.10.10"
+	localDiscRoute1011IP         = "192.168.10.11"
+	localDiscRoute1012IP         = "192.168.10.12"
+	localDiscRouteMethodNameTest = "test"
+	localDiscRouteUrlFormat      = "dubbo://%s:20000/com.ikurento.user.UserProvider"
+)
+
+func TestLocalDiscRouterRoute(t *testing.T) {
+	defer protocol.CleanAllStatus()
+	consumerURL, _ := common.NewURL(fmt.Sprintf(localDiscRouteUrlFormat, localDiscRoute1010IP))
+	url1, _ := common.NewURL(fmt.Sprintf(localDiscRouteUrlFormat, localDiscRoute1010IP))
+	url2, _ := common.NewURL(fmt.Sprintf(localDiscRouteUrlFormat, localDiscRoute1011IP))
+	url3, _ := common.NewURL(fmt.Sprintf(localDiscRouteUrlFormat, localDiscRoute1012IP))
+	hcr, _ := NewLocalPriorityRouter(consumerURL)
+
+	var invokers []protocol.Invoker
+	invoker1 := NewMockInvoker(url1)
+	invoker2 := NewMockInvoker(url2)
+	invoker3 := NewMockInvoker(url3)
+	invokers = append(invokers, invoker1, invoker2, invoker3)
+	inv := invocation.NewRPCInvocation(localDiscRouteMethodNameTest, nil, nil)
+	res := hcr.Route(utils.ToBitmap(invokers), setUpAddrCache(hcr.(*LocalPriorityRouter), invokers), consumerURL, inv)
+	// now only same ip invoker is selected
+	assert.True(t, len(res.ToArray()) == 1)
+
+	// now all invoker with ip that not match client are selected
+	invokers = invokers[1:]
+	res = hcr.Route(utils.ToBitmap(invokers), setUpAddrCache(hcr.(*LocalPriorityRouter), invokers), consumerURL, inv)
+	assert.True(t, len(res.ToArray()) == 2)
+}
+
+func setUpAddrCache(r router.Poolable, addrs []protocol.Invoker) router.Cache {
+	pool, info := r.Pool(addrs)
+	cache := chain.BuildCache(addrs)
+	cache.SetAddrMeta(r.Name(), info)
+	cache.SetAddrPool(r.Name(), pool)
+	return cache
+}

common/constant/key.go

@@ -215,6 +215,8 @@ const (
 	ListenableRouterName = "listenable"
 	// HealthCheckRouterName Specify the name of HealthCheckRouter
 	HealthCheckRouterName = "health_check"
+	// LocalPriorityRouterName Specify the name of LocalPriorityRouter
+	LocalPriorityRouterName = "local_priority"
 	// ConnCheckRouterName Specify the name of ConnCheckRouter
 	ConnCheckRouterName = "conn_check"
 	// TagRouterName Specify the name of TagRouter

config/config_api.go

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package config
+
+import (
+	"context"
+	"time"
+)
+
+//////////////////////////////////// default registry config
+const (
+	// defaultZKAddr is the default registry address of zookeeper
+	defaultZKAddr = "127.0.0.1:2181"
+
+	// defaultConsulAddr is the default registry address of consul
+	defaultConsulAddr = "127.0.0.1:8500"
+
+	// defaultNacosAddr is the default registry address of nacos
+	defaultNacosAddr = "127.0.0.1:8848"
+
+	// defaultRegistryTimeout is the default registry timeout
+	defaultRegistryTimeout = "3s"
+)
+
+// NewDefaultRegistryConfig New default registry config
+// the input @protocol can only be:
+// "zookeeper" with default addr "127.0.0.1:2181"
+// "consul" with default addr "127.0.0.1:8500"
+// "nacos" with default addr "127.0.0.1:8848"
+func NewDefaultRegistryConfig(protocol string) *RegistryConfig {
+	switch protocol {
+	case "zookeeper":
+		return &RegistryConfig{
+			Protocol:   protocol,
+			Address:    defaultZKAddr,
+			TimeoutStr: defaultRegistryTimeout,
+		}
+	case "consul":
+		return &RegistryConfig{
+			Protocol:   protocol,
+			Address:    defaultConsulAddr,
+			TimeoutStr: defaultRegistryTimeout,
+		}
+	case "nacos":
+		return &RegistryConfig{
+			Protocol:   protocol,
+			Address:    defaultNacosAddr,
+			TimeoutStr: defaultRegistryTimeout,
+		}
+	default:
+		return &RegistryConfig{
+			Protocol: protocol,
+		}
+	}
+}
+
+///////////////////////////////////// registry config api
+type RegistryConfigOpt func(config *RegistryConfig) *RegistryConfig
+
+// NewRegistryConfig creates New RegistryConfig with @opts
+func NewRegistryConfig(opts ...RegistryConfigOpt) *RegistryConfig {
+	newRegistryConfig := NewDefaultRegistryConfig("")
+	for _, v := range opts {
+		newRegistryConfig = v(newRegistryConfig)
+	}
+	return newRegistryConfig
+}
+
+// WithRegistryProtocol returns RegistryConfigOpt with given @regProtocol name
+func WithRegistryProtocol(regProtocol string) RegistryConfigOpt {
+	return func(config *RegistryConfig) *RegistryConfig {
+		config.Protocol = regProtocol
+		return config
+	}
+}
+
+// WithRegistryAddress returns RegistryConfigOpt with given @addr registry address
+func WithRegistryAddress(addr string) RegistryConfigOpt {
+	return func(config *RegistryConfig) *RegistryConfig {
+		config.Address = addr
+		return config
+	}
+}
+
+// WithRegistryTimeOut returns RegistryConfigOpt with given @timeout registry config
+func WithRegistryTimeOut(timeout string) RegistryConfigOpt {
+	return func(config *RegistryConfig) *RegistryConfig {
+		config.TimeoutStr = timeout
+		return config
+	}
+}
+
+// WithRegistryGroup returns RegistryConfigOpt with given @group registry group
+func WithRegistryGroup(group string) RegistryConfigOpt {
+	return func(config *RegistryConfig) *RegistryConfig {
+		config.Group = group
+		return config
+	}
+}
+
+// WithRegistryTTL returns RegistryConfigOpt with given @ttl registry ttl
+func WithRegistryTTL(ttl string) RegistryConfigOpt {
+	return func(config *RegistryConfig) *RegistryConfig {
+		config.TTL = ttl
+		return config
+	}
+}
+
+// WithRegistryUserName returns RegistryConfigOpt with given @userName registry userName
+func WithRegistryUserName(userName string) RegistryConfigOpt {
+	return func(config *RegistryConfig) *RegistryConfig {
+		config.Username = userName
+		return config
+	}
+}
+
+// WithRegistryPassword returns RegistryConfigOpt with given @psw registry password
+func WithRegistryPassword(psw string) RegistryConfigOpt {
+	return func(config *RegistryConfig) *RegistryConfig {
+		config.Password = psw
+		return config
+	}
+}
+
+// WithRegistrySimplified returns RegistryConfigOpt with given @simplified registry simplified flag
+func WithRegistrySimplified(simplified bool) RegistryConfigOpt {
+	return func(config *RegistryConfig) *RegistryConfig {
+		config.Simplified = simplified
+		return config
+	}
+}
+
+// WithRegistryPreferred returns RegistryConfig with given @preferred registry preferred flag
+func WithRegistryPreferred(preferred bool) RegistryConfigOpt {
+	return func(config *RegistryConfig) *RegistryConfig {
+		config.Preferred = preferred
+		return config
+	}
+}
+
+// WithRegistryWeight returns RegistryConfigOpt with given @weight registry weight flag
+func WithRegistryWeight(weight int64) RegistryConfigOpt {
+	return func(config *RegistryConfig) *RegistryConfig {
+		config.Weight = weight
+		return config
+	}
+}
+
+// WithRegistryParams returns RegistryConfigOpt with given registry @params
+func WithRegistryParams(params map[string]string) RegistryConfigOpt {
+	return func(config *RegistryConfig) *RegistryConfig {
+		config.Params = params
+		return config
+	}
+}
+
+///////////////////////////////////// consumer config api
+// ConsumerConfigOpt is the options to init ConsumerConfig
+type ConsumerConfigOpt func(config *ConsumerConfig) *ConsumerConfig
+
+// NewDefaultConsumerConfig returns default ConsumerConfig
+// with connection timeout = 3s, request timeout = 3s
+func NewDefaultConsumerConfig() *ConsumerConfig {
+	check := true
+	newConsumerConfig := &ConsumerConfig{
+		BaseConfig:     BaseConfig{},
+		Registries:     make(map[string]*RegistryConfig, 8),
+		References:     make(map[string]*ReferenceConfig, 8),
+		ConnectTimeout: 3 * time.Second,
+		RequestTimeout: 3 * time.Second,
+		Check:          &check,
+	}
+	return newConsumerConfig
+}
+
+// NewConsumerConfig returns ConsumerConfig with @opts
+func NewConsumerConfig(opts ...ConsumerConfigOpt) *ConsumerConfig {
+	newConfig := NewDefaultConsumerConfig()
+	for _, v := range opts {
+		v(newConfig)
+	}
+	return newConfig
+}
+
+// WithConsumerAppConfig returns ConsumerConfigOpt with given @appConfig
+func WithConsumerAppConfig(appConfig *ApplicationConfig) ConsumerConfigOpt {
+	return func(config *ConsumerConfig) *ConsumerConfig {
+		config.ApplicationConfig = appConfig
+		return config
+	}
+}
+
+// WithConsumerRegistryConfig returns ConsumerConfigOpt with given @registryKey and @regConfig
+func WithConsumerRegistryConfig(registryKey string, regConfig *RegistryConfig) ConsumerConfigOpt {
+	return func(config *ConsumerConfig) *ConsumerConfig {
+		config.Registries[registryKey] = regConfig
+		return config
+	}
+}
+
+// WithConsumerReferenceConfig returns ConsumerConfigOpt with
+func WithConsumerReferenceConfig(referenceKey string, refConfig *ReferenceConfig) ConsumerConfigOpt {
+	return func(config *ConsumerConfig) *ConsumerConfig {
+		config.References[referenceKey] = refConfig
+		return config
+	}
+}
+
+// WithConsumerConnTimeout returns ConsumerConfigOpt with given consumer conn @timeout
+func WithConsumerConnTimeout(timeout time.Duration) ConsumerConfigOpt {
+	return func(config *ConsumerConfig) *ConsumerConfig {
+		config.ConnectTimeout = timeout
+		return config
+	}
+}
+
+// WithConsumerRequestTimeout returns ConsumerConfigOpt with given consumer request @timeout
+func WithConsumerRequestTimeout(timeout time.Duration) ConsumerConfigOpt {
+	return func(config *ConsumerConfig) *ConsumerConfig {
+		config.RequestTimeout = timeout
+		return config
+	}
+}
+
+// WithConsumerConfigCenterConfig returns ConsumerConfigOpt with given @configCenterConfig
+func WithConsumerConfigCenterConfig(configCenterConfig *ConfigCenterConfig) ConsumerConfigOpt {
+	return func(config *ConsumerConfig) *ConsumerConfig {
+		config.ConfigCenterConfig = configCenterConfig
+		return config
+	}
+}
+
+// WithConsumerConfigCheck returns ConsumerConfigOpt with given @check flag
+func WithConsumerConfigCheck(check bool) ConsumerConfigOpt {
+	return func(config *ConsumerConfig) *ConsumerConfig {
+		*config.Check = check
+		return config
+	}
+}
+
+//////////////////////////////////// reference config api
+// ReferenceConfigOpt is consumer's reference config
+type ReferenceConfigOpt func(config *ReferenceConfig) *ReferenceConfig
+
+// NewDefaultReferenceConfig returns empty ReferenceConfig
+func NewDefaultReferenceConfig() *ReferenceConfig {
+	newReferenceConfig := NewReferenceConfig("", context.Background())
+	newReferenceConfig.Methods = make([]*MethodConfig, 0, 8)
+	newReferenceConfig.Params = make(map[string]string, 8)
+	return newReferenceConfig
+}
+
+// NewReferenceConfigByAPI returns ReferenceConfig with given @opts
+func NewReferenceConfigByAPI(opts ...ReferenceConfigOpt) *ReferenceConfig {
+	newreferenceConfig := NewDefaultReferenceConfig()
+	for _, v := range opts {
+		v(newreferenceConfig)
+	}
+	return newreferenceConfig
+}
+
+// WithReferenceRegistry returns ReferenceConfigOpt with given registryKey: @registry
+func WithReferenceRegistry(registry string) ReferenceConfigOpt {
+	return func(config *ReferenceConfig) *ReferenceConfig {
+		config.Registry = registry
+		return config
+	}
+}
+
+// WithReferenceProtocol returns ReferenceConfigOpt with given protocolKey: @protocol
+func WithReferenceProtocol(protocol string) ReferenceConfigOpt {
+	return func(config *ReferenceConfig) *ReferenceConfig {
+		config.Protocol = protocol
+		return config
+	}
+}
+
+// WithReferenceInterface returns ReferenceConfigOpt with given @interfaceName
+func WithReferenceInterface(interfaceName string) ReferenceConfigOpt {
+	return func(config *ReferenceConfig) *ReferenceConfig {
+		config.InterfaceName = interfaceName
+		return config
+	}
+}
+
+// WithReferenceCluster returns ReferenceConfigOpt with given cluster name: @cluster
+func WithReferenceCluster(cluster string) ReferenceConfigOpt {
+	return func(config *ReferenceConfig) *ReferenceConfig {
+		config.Cluster = cluster
+		return config
+	}
+}
+
+// WithReferenceMethod returns ReferenceConfigOpt with given @method, @retries, and load balance: @lb
+func WithReferenceMethod(methodName, retries, lb string) ReferenceConfigOpt {
+	return func(config *ReferenceConfig) *ReferenceConfig {
+		config.Methods = append(config.Methods, &MethodConfig{
+			Name:        methodName,
+			Retries:     retries,
+			LoadBalance: lb,
+		})
+		return config
+	}
+}
+
+///////////////////////////////////// provider config api
+// ProviderConfigOpt is the
+type ProviderConfigOpt func(config *ProviderConfig) *ProviderConfig
+
+// NewDefaultProviderConfig returns ProviderConfig with default ApplicationConfig
+func NewDefaultProviderConfig() *ProviderConfig {
+	newConsumerConfig := &ProviderConfig{
+		BaseConfig: BaseConfig{
+			ApplicationConfig: &ApplicationConfig{
+				Name:         "dubbo",
+				Module:       "module",
+				Organization: "dubbo_org",
+				Owner:        "dubbo",
+			},
+		},
+		Services:   make(map[string]*ServiceConfig),
+		Registries: make(map[string]*RegistryConfig, 8),
+		Protocols:  make(map[string]*ProtocolConfig, 8),
+	}
+	return newConsumerConfig
+}
+
+// NewProviderConfig returns ProviderConfig with given @opts
+func NewProviderConfig(opts ...ProviderConfigOpt) *ProviderConfig {
+	newConfig := NewDefaultProviderConfig()
+	for _, v := range opts {
+		v(newConfig)
+	}
+	return newConfig
+}
+
+// WithPrividerRegistryConfig returns ProviderConfigOpt with given registry config: @regConfig
+func WithPrividerRegistryConfig(regConfig *RegistryConfig) ProviderConfigOpt {
+	return func(config *ProviderConfig) *ProviderConfig {
+		config.Registries[regConfig.Protocol] = regConfig
+		return config
+	}
+}
+
+// WithProviderAppConfig returns ProviderConfigOpt with given @appConfig
+func WithProviderAppConfig(appConfig *ApplicationConfig) ProviderConfigOpt {
+	return func(config *ProviderConfig) *ProviderConfig {
+		config.ApplicationConfig = appConfig
+		return config
+	}
+}
+
+// WithProviderServices returns ProviderConfig with given serviceNameKey @serviceName and @serviceConfig
+func WithProviderServices(serviceName string, serviceConfig *ServiceConfig) ProviderConfigOpt {
+	return func(config *ProviderConfig) *ProviderConfig {
+		config.Services[serviceName] = serviceConfig
+		return config
+	}
+}
+
+// WithProviderProtocol returns ProviderConfigOpt with given @protocolKey, protocolName @protocol and @port
+func WithProviderProtocol(protocolKey, protocol, port string) ProviderConfigOpt {
+	return func(config *ProviderConfig) *ProviderConfig {
+		config.Protocols[protocolKey] = &ProtocolConfig{
+			Name: protocol,
+			Port: port,
+		}
+		return config
+	}
+}
+
+// WithProviderRegistry returns ProviderConfigOpt with given @registryKey and registry @registryConfig
+func WithProviderRegistry(registryKey string, registryConfig *RegistryConfig) ProviderConfigOpt {
+	return func(config *ProviderConfig) *ProviderConfig {
+		config.Registries[registryKey] = registryConfig
+		return config
+	}
+}
+
+/////////////////////////////////////// service config api
+// ServiceConfigOpt is the option to init ServiceConfig
+type ServiceConfigOpt func(config *ServiceConfig) *ServiceConfig
+
+// NewDefaultServiceConfig returns default ServiceConfig
+func NewDefaultServiceConfig() *ServiceConfig {
+	newServiceConfig := NewServiceConfig("", context.Background())
+	newServiceConfig.Params = make(map[string]string)
+	newServiceConfig.Methods = make([]*MethodConfig, 0, 8)
+	return newServiceConfig
+}
+
+// NewServiceConfigByAPI is named as api, because there is NewServiceConfig func already declared
+// NewServiceConfigByAPI returns ServiceConfig with given @opts
+func NewServiceConfigByAPI(opts ...ServiceConfigOpt) *ServiceConfig {
+	defaultServiceConfig := NewDefaultServiceConfig()
+	for _, v := range opts {
+		v(defaultServiceConfig)
+	}
+	return defaultServiceConfig
+}
+
+// WithServiceRegistry returns ServiceConfigOpt with given registryKey @registry
+func WithServiceRegistry(registry string) ServiceConfigOpt {
+	return func(config *ServiceConfig) *ServiceConfig {
+		config.Registry = registry
+		return config
+	}
+}
+
+// WithServiceProtocol returns ServiceConfigOpt with given protocolKey @protocol
+func WithServiceProtocol(protocol string) ServiceConfigOpt {
+	return func(config *ServiceConfig) *ServiceConfig {
+		config.Protocol = protocol
+		return config
+	}
+}
+
+// WithServiceInterface returns ServiceConfigOpt with given @interfaceName
+func WithServiceInterface(interfaceName string) ServiceConfigOpt {
+	return func(config *ServiceConfig) *ServiceConfig {
+		config.InterfaceName = interfaceName
+		return config
+	}
+}
+
+// WithServiceLoadBalance returns ServiceConfigOpt with given load balance @lb
+func WithServiceLoadBalance(lb string) ServiceConfigOpt {
+	return func(config *ServiceConfig) *ServiceConfig {
+		config.Loadbalance = lb
+		return config
+	}
+}
+
+// WithServiceWarmUpTime returns ServiceConfigOpt with given @warmUp time
+func WithServiceWarmUpTime(warmUp string) ServiceConfigOpt {
+	return func(config *ServiceConfig) *ServiceConfig {
+		config.Warmup = warmUp
+		return config
+	}
+}
+
+// WithServiceCluster returns ServiceConfigOpt with given cluster name @cluster
+func WithServiceCluster(cluster string) ServiceConfigOpt {
+	return func(config *ServiceConfig) *ServiceConfig {
+		config.Cluster = cluster
+		return config
+	}
+}
+
+// WithServiceMethod returns ServiceConfigOpt with given @name, @retries and load balance @lb
+func WithServiceMethod(name, retries, lb string) ServiceConfigOpt {
+	return func(config *ServiceConfig) *ServiceConfig {
+		config.Methods = append(config.Methods, &MethodConfig{
+			Name:        name,
+			Retries:     retries,
+			LoadBalance: lb,
+		})
+		return config
+	}
+}
+
+///////////////////////////////////////// Application config api
+// ApplicationConfigOpt is option to init ApplicationConfig
+type ApplicationConfigOpt func(config *ApplicationConfig) *ApplicationConfig
+
+// NewDefaultApplicationConfig returns ApplicationConfig with default
+// name: dubbo.io
+// module: sample
+// organization: dubbo.io
+// owner: dubbogo
+// version: 0.0.1
+// environment dev
+func NewDefaultApplicationConfig() *ApplicationConfig {
+	newAppConfig := &ApplicationConfig{
+		Name:         "dubbo.io",
+		Module:       "sample",
+		Organization: "dubbo.io",
+		Owner:        "dubbogo",
+		Version:      "0.0.1",
+		Environment:  "dev",
+	}
+	return newAppConfig
+}
+
+// NewApplicationConfig is named as api, because there is NewServiceConfig func already declared
+// NewApplicationConfig returns ApplicationConfig wigh default application config
+func NewApplicationConfig(opts ...ApplicationConfigOpt) *ApplicationConfig {
+	defaultServiceConfig := NewDefaultApplicationConfig()
+	for _, v := range opts {
+		v(defaultServiceConfig)
+	}
+	return defaultServiceConfig
+}
+
+// WithAppName returns ApplicationConfigOpt with given @name
+func WithAppName(name string) ApplicationConfigOpt {
+	return func(config *ApplicationConfig) *ApplicationConfig {
+		config.Name = name
+		return config
+	}
+}
+
+// WithAppModule returns ApplicationConfigOpt with given @module
+func WithAppModule(module string) ApplicationConfigOpt {
+	return func(config *ApplicationConfig) *ApplicationConfig {
+		config.Module = module
+		return config
+	}
+}
+
+// WithAppOrganization returns ApplicationConfigOpt wight given organization @org
+func WithAppOrganization(org string) ApplicationConfigOpt {
+	return func(config *ApplicationConfig) *ApplicationConfig {
+		config.Organization = org
+		return config
+	}
+}
+
+// WithAppOwner returns ApplicationConfigOpt with given @owner
+func WithAppOwner(owner string) ApplicationConfigOpt {
+	return func(config *ApplicationConfig) *ApplicationConfig {
+		config.Owner = owner
+		return config
+	}
+}
+
+// WithAppVersion returns ApplicationConfigOpt with given version @version
+func WithAppVersion(version string) ApplicationConfigOpt {
+	return func(config *ApplicationConfig) *ApplicationConfig {
+		config.Version = version
+		return config
+	}
+}
+
+// WithAppEnvironment returns ApplicationConfigOpt wigh given environment @env
+func WithAppEnvironment(env string) ApplicationConfigOpt {
+	return func(config *ApplicationConfig) *ApplicationConfig {
+		config.Environment = env
+		return config
+	}
+}

config/config_api_test.go

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package config
+
+import (
+	"strconv"
+	"testing"
+	"time"
+)
+
+import (
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNewDefaultServiceConfig(t *testing.T) {
+	serviceConfig := NewServiceConfigByAPI(
+		WithServiceCluster("test-cluster"),
+		WithServiceInterface("test-interface"),
+		WithServiceLoadBalance("test-loadbalance"),
+		WithServiceMethod("test-method1", "test-retries1", "test-lb1"),
+		WithServiceMethod("test-method2", "test-retries2", "test-lb2"),
+		WithServiceMethod("test-method3", "test-retries3", "test-lb3"),
+		WithServiceProtocol("test-protocol"),
+		WithServiceRegistry("test-registry"),
+		WithServiceWarmUpTime("test-warmup"),
+	)
+	assert.Equal(t, serviceConfig.Cluster, "test-cluster")
+	assert.Equal(t, serviceConfig.InterfaceName, "test-interface")
+	assert.Equal(t, serviceConfig.Loadbalance, "test-loadbalance")
+	for i, v := range serviceConfig.Methods {
+		backFix := strconv.Itoa(i + 1)
+		assert.Equal(t, v.Name, "test-method"+backFix)
+		assert.Equal(t, v.Retries, "test-retries"+backFix)
+		assert.Equal(t, v.LoadBalance, "test-lb"+backFix)
+	}
+	assert.Equal(t, serviceConfig.Protocol, "test-protocol")
+	assert.Equal(t, serviceConfig.Registry, "test-registry")
+	assert.Equal(t, serviceConfig.Warmup, "test-warmup")
+}
+
+func TestNewReferenceConfigByAPI(t *testing.T) {
+	refConfig := NewReferenceConfigByAPI(
+		WithReferenceCluster("test-cluster"),
+		WithReferenceInterface("test-interface"),
+		WithReferenceMethod("test-method1", "test-retries1", "test-lb1"),
+		WithReferenceMethod("test-method2", "test-retries2", "test-lb2"),
+		WithReferenceMethod("test-method3", "test-retries3", "test-lb3"),
+		WithReferenceProtocol("test-protocol"),
+		WithReferenceRegistry("test-registry"),
+	)
+	assert.Equal(t, refConfig.Cluster, "test-cluster")
+	assert.Equal(t, refConfig.InterfaceName, "test-interface")
+	for i, v := range refConfig.Methods {
+		backFix := strconv.Itoa(i + 1)
+		assert.Equal(t, v.Name, "test-method"+backFix)
+		assert.Equal(t, v.Retries, "test-retries"+backFix)
+		assert.Equal(t, v.LoadBalance, "test-lb"+backFix)
+	}
+	assert.Equal(t, refConfig.Protocol, "test-protocol")
+	assert.Equal(t, refConfig.Registry, "test-registry")
+}
+
+func TestNewRegistryConfig(t *testing.T) {
+	regConfig := NewRegistryConfig(
+		WithRegistryTimeOut("test-timeout"),
+		WithRegistryProtocol("test-protocol"),
+		WithRegistryGroup("test-group"),
+		WithRegistryAddress("test-address"),
+		WithRegistrySimplified(true),
+		WithRegistryUserName("test-username"),
+		WithRegistryPassword("test-password"),
+	)
+	assert.Equal(t, regConfig.TimeoutStr, "test-timeout")
+	assert.Equal(t, regConfig.Protocol, "test-protocol")
+	assert.Equal(t, regConfig.Group, "test-group")
+	assert.Equal(t, regConfig.Address, "test-address")
+	assert.Equal(t, regConfig.Simplified, true)
+	assert.Equal(t, regConfig.Username, "test-username")
+	assert.Equal(t, regConfig.Password, "test-password")
+}
+
+func TestNewConsumerConfig(t *testing.T) {
+	referConfig := NewReferenceConfigByAPI(
+		WithReferenceCluster("test-cluster"),
+		WithReferenceInterface("test-interface"),
+		WithReferenceMethod("test-method1", "test-retries1", "test-lb1"),
+		WithReferenceMethod("test-method2", "test-retries2", "test-lb2"),
+		WithReferenceMethod("test-method3", "test-retries3", "test-lb3"),
+		WithReferenceProtocol("test-protocol"),
+		WithReferenceRegistry("test-registry"),
+	)
+	defaultZKRegistry := NewDefaultRegistryConfig("zookeeper")
+	assert.Equal(t, defaultZKRegistry.Address, defaultZKAddr)
+	assert.Equal(t, defaultZKRegistry.Protocol, "zookeeper")
+	assert.Equal(t, defaultZKRegistry.TimeoutStr, defaultRegistryTimeout)
+
+	testConsumerConfig := NewConsumerConfig(
+		WithConsumerConfigCheck(true),
+		WithConsumerConnTimeout(time.Minute),
+		WithConsumerRequestTimeout(time.Hour),
+		WithConsumerReferenceConfig("UserProvider", referConfig),
+		WithConsumerRegistryConfig("demoZK", defaultZKRegistry),
+	)
+
+	assert.Equal(t, *testConsumerConfig.Check, true)
+	assert.Equal(t, testConsumerConfig.ConnectTimeout, time.Minute)
+	assert.Equal(t, testConsumerConfig.RequestTimeout, time.Hour)
+	assert.Equal(t, testConsumerConfig.Registries["demoZK"], defaultZKRegistry)
+	assert.Equal(t, testConsumerConfig.References["UserProvider"], referConfig)
+}
+
+func TestNewProviderConfig(t *testing.T) {
+	serviceConfig := NewServiceConfigByAPI(
+		WithServiceCluster("test-cluster"),
+		WithServiceInterface("test-interface"),
+		WithServiceLoadBalance("test-loadbalance"),
+		WithServiceMethod("test-method1", "test-retries1", "test-lb1"),
+		WithServiceMethod("test-method2", "test-retries2", "test-lb2"),
+		WithServiceMethod("test-method3", "test-retries3", "test-lb3"),
+		WithServiceProtocol("test-protocol"),
+		WithServiceRegistry("test-registry"),
+		WithServiceWarmUpTime("test-warmup"),
+	)
+
+	defaultConsulRegistry := NewDefaultRegistryConfig("consul")
+	assert.Equal(t, defaultConsulRegistry.Address, defaultConsulAddr)
+	assert.Equal(t, defaultConsulRegistry.Protocol, "consul")
+	assert.Equal(t, defaultConsulRegistry.TimeoutStr, defaultRegistryTimeout)
+
+	defaultNacosRegistry := NewDefaultRegistryConfig("nacos")
+	assert.Equal(t, defaultNacosRegistry.Address, defaultNacosAddr)
+	assert.Equal(t, defaultNacosRegistry.Protocol, "nacos")
+	assert.Equal(t, defaultNacosRegistry.TimeoutStr, defaultRegistryTimeout)
+
+	testProviderConfig := NewProviderConfig(
+		WithProviderServices("UserProvider", serviceConfig),
+		WithProviderProtocol("dubbo", "dubbo", "20000"),
+		WithProviderRegistry("demoConsul", defaultConsulRegistry),
+		WithProviderRegistry("demoNacos", defaultNacosRegistry),
+	)
+
+	assert.NotNil(t, testProviderConfig.Services)
+	for k, v := range testProviderConfig.Services {
+		assert.Equal(t, k, "UserProvider")
+		assert.Equal(t, v, serviceConfig)
+	}
+	assert.NotNil(t, testProviderConfig.Registries)
+	i := 0
+	for k, v := range testProviderConfig.Registries {
+		if i == 0 {
+			assert.Equal(t, k, "demoConsul")
+			assert.Equal(t, v, defaultConsulRegistry)
+			i++
+		} else {
+			assert.Equal(t, k, "demoNacos")
+			assert.Equal(t, v, defaultNacosRegistry)
+		}
+	}
+
+	assert.NotNil(t, testProviderConfig.Protocols)
+	assert.Equal(t, testProviderConfig.Protocols["dubbo"].Name, "dubbo")
+	assert.Equal(t, testProviderConfig.Protocols["dubbo"].Port, "20000")
+}

config/graceful_shutdown.go

@@ -51,6 +51,7 @@ import (
  * The signals are different on different platforms.
  * We define them by using 'package build' feature https://golang.org/pkg/go/build/
  */
+const defaultShutDownTime = time.Second * 60
 
 // nolint
 func GracefulShutdownInit() {
@@ -64,20 +65,17 @@ func GracefulShutdownInit() {
 		case sig := <-signals:
 			logger.Infof("get signal %s, application will shutdown.", sig)
 			// gracefulShutdownOnce.Do(func() {
+			time.AfterFunc(totalTimeout(), func() {
+				logger.Warn("Shutdown gracefully timeout, application will shutdown immediately. ")
+				os.Exit(0)
+			})
 			BeforeShutdown()
-
 			// those signals' original behavior is exit with dump ths stack, so we try to keep the behavior
 			for _, dumpSignal := range DumpHeapShutdownSignals {
 				if sig == dumpSignal {
 					debug.WriteHeapDump(os.Stdout.Fd())
 				}
 			}
-
-			time.AfterFunc(totalTimeout(), func() {
-				logger.Warn("Shutdown gracefully timeout, application will shutdown immediately. ")
-				os.Exit(0)
-			})
-
 			os.Exit(0)
 		}
 	}()
@@ -196,7 +194,7 @@ func waitingProcessedTimeout(shutdownConfig *ShutdownConfig) {
 }
 
 func totalTimeout() time.Duration {
-	var providerShutdown time.Duration
+	var providerShutdown = defaultShutDownTime
 	if providerConfig != nil && providerConfig.ShutdownConfig != nil {
 		providerShutdown = providerConfig.ShutdownConfig.GetTimeout()
 	}

go.sum

@@ -174,7 +174,6 @@ 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.10.1 h1:39kF9Cd5JOiMpmwG6dX1/aLWNFqFv9gHp8HrhzMmjLY=
 github.com/dubbogo/gost v1.10.1/go.mod h1:+mQGS51XQEUWZP2JeGZTxJwipjRKtJO7Tr+FOg+72rI=
 github.com/dubbogo/gost v1.11.0 h1:9KtyWQz1gMlAfwzen5iyhMdoe08SPBBUVhco4rdgJ9I=
 github.com/dubbogo/gost v1.11.0/go.mod h1:w8Yw29eDWtRVo3tx9nPpHkNZnOi4SRx1fZf7eVlAAU4=
@@ -748,7 +747,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 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd0=
 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=

metadata/service/remote/service.go

@@ -107,7 +107,7 @@ func (mts *MetadataService) SubscribeURL(url *common.URL) (bool, error) {
 
 // UnsubscribeURL will be implemented by in memory service
 func (mts *MetadataService) UnsubscribeURL(url *common.URL) error {
-	// TODO remove call self.
+	// TODO remove call local.
 	return nil
 	//return mts.UnsubscribeURL(url)
 }

registry/base_registry.go

@@ -136,6 +136,13 @@ func (r *BaseRegistry) Register(conf *common.URL) error {
 		ok  bool
 		err error
 	)
+	// if developer define registry port and ip, use it first.
+	if ipToRegistry := os.Getenv("DUBBO_IP_TO_REGISTRY"); ipToRegistry != "" {
+		conf.Ip = ipToRegistry
+	}
+	if portToRegistry := os.Getenv("DUBBO_PORT_TO_REGISTRY"); portToRegistry != "" {
+		conf.Port = portToRegistry
+	}
 	role, _ := strconv.Atoi(r.URL.GetParam(constant.ROLE_KEY, ""))
 	// Check if the service has been registered
 	r.cltLock.Lock()

tools/cli/go.mod

@@ -4,7 +4,9 @@ go 1.13
 
 require (
 	github.com/apache/dubbo-go-hessian2 v1.8.0
-	github.com/dubbogo/gost v1.10.0
+	github.com/dubbogo/gost v1.11.1
 	github.com/pkg/errors v0.9.1
 	go.uber.org/atomic v1.7.0
 )
+
+replace github.com/coreos/bbolt => go.etcd.io/bbolt v1.3.4