Merge branch 'develop' into heartbeat_for_server

README.md

@@ -15,6 +15,10 @@ Apache License, Version 2.0
 
 [v1.0.0 - May 29, 2019 compatible with dubbo v2.6.5](https://github.com/apache/dubbo-go/releases/tag/v1.0.0)
 
+[v1.1.0 - Sep 7, 2019 the first release after transferred to apache](https://github.com/apache/dubbo-go/releases/tag/v1.1.0)
+
+[v1.2.0 - Nov 15, 2019](https://github.com/apache/dubbo-go/releases/tag/v1.2.0)
+
 ## Project Architecture ##
 
 Both extension module and layered project architecture is according to Apache Dubbo (including protocol layer, registry layer, cluster layer, config layer and so on), the advantage of this arch is as following: you can implement these layered interfaces in your own way, override the default implementation of dubbo-go by calling 'extension.SetXXX' of extension, complete your special needs without modifying the source code. At the same time, you are welcome to contribute implementation of useful extension to the community.
@@ -27,43 +31,94 @@ If you wanna know more about dubbo-go, please visit this reference [Project Arch
 
 Finished List:
 
-- Role: Consumer, Provider
-- Transport: HTTP, TCP
-- Codec: JsonRPC v2, Hessian v2
-- Registry: ZooKeeper/[etcd v3](https://github.com/apache/dubbo-go/pull/148)/[nacos](https://github.com/apache/dubbo-go/pull/151)/[consul](https://github.com/apache/dubbo-go/pull/121)
-- Configure Center: Zookeeper
-- Cluster Strategy: Failover/[Failfast](https://github.com/apache/dubbo-go/pull/140)/[Failsafe/Failback](https://github.com/apache/dubbo-go/pull/136)/[Available](https://github.com/apache/dubbo-go/pull/155)/[Broadcast](https://github.com/apache/dubbo-go/pull/158)/[Forking](https://github.com/apache/dubbo-go/pull/161)
-- Load Balance: Random/[RoundRobin](https://github.com/apache/dubbo-go/pull/66)/[LeastActive](https://github.com/apache/dubbo-go/pull/65)
-- Filter: Echo Health Check/[Circuit break and service downgrade](https://github.com/apache/dubbo-go/pull/133)/[TokenFilter](https://github.com/apache/dubbo-go/pull/202)/[AccessLogFilter](https://github.com/apache/dubbo-go/pull/214)
-- Other feature: [generic invoke](https://github.com/apache/dubbo-go/pull/122)/start check/connecting certain provider/multi-protocols/multi-registries/multi-versions/service group
+- Role
+    * Consumer
+    * Provider
+
+- Transport
+    * HTTP
+    * TCP
+
+- Codec
+    * JsonRPC V2
+    * Hessian V2
+    
+- Registry
+    * ZooKeeper
+    * [etcd v3](https://github.com/apache/dubbo-go/pull/148)
+    * [nacos](https://github.com/apache/dubbo-go/pull/151)
+    * [consul](https://github.com/apache/dubbo-go/pull/121)
+    
+- Dynamic Configure Center & Service Management Configurator
+    * Zookeeper
+    * [apollo](https://github.com/apache/dubbo-go/pull/250)
+
+- Cluster Strategy
+    * Failover
+    * [Failfast](https://github.com/apache/dubbo-go/pull/140)
+    * [Failsafe/Failback](https://github.com/apache/dubbo-go/pull/136)
+    * [Available](https://github.com/apache/dubbo-go/pull/155)
+    * [Broadcast](https://github.com/apache/dubbo-go/pull/158)
+    * [Forking](https://github.com/apache/dubbo-go/pull/161)
+    
+- Load Balance
+    * Random
+    * [RoundRobin](https://github.com/apache/dubbo-go/pull/66)
+    * [LeastActive](https://github.com/apache/dubbo-go/pull/65)
+    
+- Filter
+    * Echo Health Check
+    * [Circuit break and service downgrade](https://github.com/apache/dubbo-go/pull/133)
+    * [TokenFilter](https://github.com/apache/dubbo-go/pull/202)
+    * [AccessLogFilter](https://github.com/apache/dubbo-go/pull/214)
+    * [TpsLimitFilter](https://github.com/apache/dubbo-go/pull/237)
+    * [ExecuteLimitFilter](https://github.com/apache/dubbo-go/pull/246)
+    
+- Invoke
+    * [generic invoke](https://github.com/apache/dubbo-go/pull/122)
+    
+- Others:
+    * start check
+    * connecting certain provider
+    * multi-protocols
+    * multi-registries
+    * multi-versions
+    * service group
 
 Working List:
 
 - Load Balance: ConsistentHash
-- Filter: CountFilter/ExecuteLimitFilter/TpsLimitFilter
 - Registry: k8s
-- Configure Center: apollo
-- Dynamic Configuration Center & Metadata Center (dubbo v2.7.x)
-- Metrics: Promethus(dubbo v2.7.x)
-
-Todo List:
-
-- Registry: kubernetes
-- Routing: istio
-- tracing (dubbo ecosystem)
+- Metadata Center (dubbo v2.7.x)
+- Metrics: Opentracing/Promethus(dubbo v2.7.x)
 
 You can know more about dubbo-go by its [roadmap](https://github.com/apache/dubbo-go/wiki/Roadmap).
 
+![feature](https://raw.githubusercontent.com/wiki/apache/dubbo-go/arch.png)
+
 ## Document
 
-TODO
+https://dubbogo.github.io/dubbo-go-website(**Improving**)
 
 ## Quick Start
 
-[dubbogo-samples](https://github.com/dubbogo/dubbogo-samples) shows how to use dubbo-go. Please read the [dubbogo-samples/README.md](https://github.com/dubbogo/dubbogo-samples/blob/master/README.md) carefully to learn how to dispose the configuration and compile the program.
+[dubbo-samples/golang](https://github.com/dubbogo/dubbo-samples) shows how to use dubbo-go. Please read the [dubbo-samples/golang/README.md](https://github.com/dubbogo/dubbo-samples/blob/master/golang/README.md) carefully to learn how to dispose the configuration and compile the program.
 
 ## Running unit tests
 
+### Prepare
+
+Mac/Linux
+```bash
+sh ./before_ut.sh
+```
+
+Windows
+```bash
+before_ut.bat
+```
+
+# Run
 ```bash
 go test ./...
 

README_CN.md

@@ -14,6 +14,10 @@ Apache License, Version 2.0
 
 [v1.0.0 - 2019年5月29日 兼容dubbo v2.6.5 版本](https://github.com/apache/dubbo-go/releases/tag/v1.0.0)
 
+[v1.1.0 - 2019年9月7日 捐献给Apache之后的第一次release](https://github.com/apache/dubbo-go/releases/tag/v1.1.0)
+
+[v1.2.0 - 2019年11月15日](https://github.com/apache/dubbo-go/releases/tag/v1.2.0)
+
 ## 工程架构 ##
 
 基于dubbo的extension模块和分层的代码设计(包括 protocol layer, registry layer, cluster layer, config 等等)。我们的目标是：你可以对这些分层接口进行新的实现，并通过调用 extension 模块的“ extension.SetXXX ”方法来覆盖 dubbo-go [同 go-for-apache-dubbo ]的默认实现，以完成自己的特殊需求而无需修改源代码。同时，欢迎你为社区贡献有用的拓展实现。
@@ -26,48 +30,98 @@ Apache License, Version 2.0
 
 实现列表:
 
-- 角色端: Consumer, Provider
-- 传输协议: HTTP, TCP
-- 序列化协议: JsonRPC v2, Hessian v2
-- 注册中心: ZooKeeper/[etcd v3](https://github.com/apache/dubbo-go/pull/148)/[nacos](https://github.com/apache/dubbo-go/pull/151)/[consul](https://github.com/apache/dubbo-go/pull/121)
-- 配置中心: Zookeeper
-- 集群策略: Failover/[Failfast](https://github.com/apache/dubbo-go/pull/140)/[Failsafe/Failback](https://github.com/apache/dubbo-go/pull/136)/[Available](https://github.com/apache/dubbo-go/pull/155)/[Broadcast](https://github.com/apache/dubbo-go/pull/158)/[Forking](https://github.com/apache/dubbo-go/pull/161)
-- 负载均衡策略: Random/[RoundRobin](https://github.com/apache/dubbo-go/pull/66)/[LeastActive](https://github.com/apache/dubbo-go/pull/65)
-- 过滤器: Echo Health Check/[服务熔断&降级](https://github.com/apache/dubbo-go/pull/133)/[TokenFilter](https://github.com/apache/dubbo-go/pull/202)/[AccessLogFilter](https://github.com/apache/dubbo-go/pull/214)
-- 其他功能支持: [泛化调用](https://github.com/apache/dubbo-go/pull/122)/启动时检查/服务直连/多服务协议/多注册中心/多服务版本/服务分组
+- 角色端
+    * Consumer
+    * Provider
+    
+- 传输协议
+    * HTTP
+    * TCP
+
+- 序列化协议
+    * JsonRPC V2
+    * Hessian V2
+    
+- 注册中心
+    * ZooKeeper
+    * [etcd v3](https://github.com/apache/dubbo-go/pull/148)
+    * [nacos](https://github.com/apache/dubbo-go/pull/151)
+    * [consul](https://github.com/apache/dubbo-go/pull/121)
+    
+- 动态配置中心与服务治理配置器
+    * Zookeeper
+    * [apollo](https://github.com/apache/dubbo-go/pull/250)
+    
+- 集群策略
+    * Failover
+    * [Failfast](https://github.com/apache/dubbo-go/pull/140)
+    * [Failsafe/Failback](https://github.com/apache/dubbo-go/pull/136)
+    * [Available](https://github.com/apache/dubbo-go/pull/155)
+    * [Broadcast](https://github.com/apache/dubbo-go/pull/158)
+    * [Forking](https://github.com/apache/dubbo-go/pull/161)
+   
+- 负载均衡策略
+    * Random
+    * [RoundRobin](https://github.com/apache/dubbo-go/pull/66)
+    * [LeastActive](https://github.com/apache/dubbo-go/pull/65)
+    
+- 过滤器
+    * Echo Health Check
+    * [服务熔断&降级](https://github.com/apache/dubbo-go/pull/133)
+    * [TokenFilter](https://github.com/apache/dubbo-go/pull/202)
+    * [AccessLogFilter](https://github.com/apache/dubbo-go/pull/214)
+    * [TpsLimitFilter](https://github.com/apache/dubbo-go/pull/237)
+    * [ExecuteLimitFilter](https://github.com/apache/dubbo-go/pull/246)
+    
+- 调用
+    * [泛化调用](https://github.com/apache/dubbo-go/pull/122)
+    
+- 其他功能支持:
+    * 启动时检查
+    * 服务直连
+    * 多服务协议
+    * 多注册中心
+    * 多服务版本
+    * 服务分组
 
 开发中列表:
 
-- 集群策略: Forking
 - 负载均衡策略: ConsistentHash
-- 过滤器: CountFilter/ExecuteLimitFilter/TpsLimitFilter
 - 注册中心: k8s
-- 配置中心: apollo
-- 动态配置中心 & 元数据中心 (dubbo v2.7.x)
-- Metrics: Promethus(dubbo v2.7.x)
-
-任务列表:
+- 元数据中心 (dubbo v2.7.x)
+- Metrics: Opentracing/Promethus(dubbo v2.7.x)
 
-- 注册中心: kubernetes
-- Routing: istio
-- tracing (dubbo ecosystem)
+你可以通过访问 [roadmap](https://github.com/apache/dubbo-go/wiki/Roadmap) 知道更多关于 dubbo-go 的信息。
 
-你可以通过访问 [roadmap](https://github.com/apache/dubbo-go/wiki/Roadmap) 知道更多关于 dubbo-go 的信息
+![feature](https://raw.githubusercontent.com/wiki/apache/dubbo-go/arch.png)
 
 ## 文档
 
-TODO
+https://dubbogo.github.io/dubbo-go-website(**完善中**)
 
 ## 快速开始 ##
 
-[dubbogo-samples](https://github.com/dubbogo/dubbogo-samples)这个项目的事例展示了如何使用 dubbo-go 。请仔细阅读 [dubbogo-samples/README.md](https://github.com/dubbogo/dubbogo-samples/blob/master/README.md) 学习如何处理配置并编译程序。
+[dubbo-samples/golang](https://github.com/dubbogo/dubbo-samples)这个项目的事例展示了如何使用 dubbo-go 。请仔细阅读 [dubbo-samples/golang/README.md](https://github.com/dubbogo/dubbo-samples/blob/master/golang/README.md) 学习如何处理配置并编译程序。
 
 ## 运行单测
 
+### 准备
+
+Mac/Linux
+```bash
+sh ./before_ut.sh
+```
+
+Windows
+```bash
+before_ut.bat
+```
+
+# 执行
 ```bash
 go test ./...
 
-# 覆盖率
+# coverage
 go test ./... -coverprofile=coverage.txt -covermode=atomic
 ```
 

before_ut.bat

+::
+::  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.
+
+set zkJar=zookeeper-3.4.9-fatjar.jar
+md remoting\zookeeper\zookeeper-4unittest\contrib\fatjar config_center\zookeeper\zookeeper-4unittest\contrib\fatjar registry\zookeeper\zookeeper-4unittest\contrib\fatjar
+curl -L https://github.com/dubbogo/resources/raw/master/zookeeper-4unitest/contrib/fatjar/%zkJar% -o remoting/zookeeper/zookeeper-4unittest/contrib/fatjar/%zkJar%
+xcopy /f "remoting/zookeeper/zookeeper-4unittest/contrib/fatjar/%zkJar%" "config_center/zookeeper/zookeeper-4unittest/contrib/fatjar/"
+xcopy /f "remoting/zookeeper/zookeeper-4unittest/contrib/fatjar/%zkJar%" "registry/zookeeper/zookeeper-4unittest/contrib/fatjar/"
\ No newline at end of file

before_ut.sh

+#
+#  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.
+
+
 mkdir -p remoting/zookeeper/zookeeper-4unittest/contrib/fatjar config_center/zookeeper/zookeeper-4unittest/contrib/fatjar registry/zookeeper/zookeeper-4unittest/contrib/fatjar
 wget -P "remoting/zookeeper/zookeeper-4unittest/contrib/fatjar" https://github.com/dubbogo/resources/raw/master/zookeeper-4unitest/contrib/fatjar/zookeeper-3.4.9-fatjar.jar
 cp remoting/zookeeper/zookeeper-4unittest/contrib/fatjar/zookeeper-3.4.9-fatjar.jar config_center/zookeeper/zookeeper-4unittest/contrib/fatjar/

cluster/cluster_impl/base_cluster_invoker.go

@@ -35,6 +35,7 @@ type baseClusterInvoker struct {
 	directory      cluster.Directory
 	availablecheck bool
 	destroyed      *atomic.Bool
+	stickyInvoker  protocol.Invoker
 }
 
 func newBaseClusterInvoker(directory cluster.Directory) baseClusterInvoker {
@@ -56,7 +57,9 @@ func (invoker *baseClusterInvoker) Destroy() {
 }
 
 func (invoker *baseClusterInvoker) IsAvailable() bool {
-	//TODO:sticky connection
+	if invoker.stickyInvoker != nil {
+		return invoker.stickyInvoker.IsAvailable()
+	}
 	return invoker.directory.IsAvailable()
 }
 
@@ -83,15 +86,42 @@ func (invoker *baseClusterInvoker) checkWhetherDestroyed() error {
 }
 
 func (invoker *baseClusterInvoker) doSelect(lb cluster.LoadBalance, invocation protocol.Invocation, invokers []protocol.Invoker, invoked []protocol.Invoker) protocol.Invoker {
-	//todo:sticky connect
+
+	var selectedInvoker protocol.Invoker
+	url := invokers[0].GetUrl()
+	sticky := url.GetParamBool(constant.STICKY_KEY, false)
+	//Get the service method sticky config if have
+	sticky = url.GetMethodParamBool(invocation.MethodName(), constant.STICKY_KEY, sticky)
+
+	if invoker.stickyInvoker != nil && !isInvoked(invoker.stickyInvoker, invokers) {
+		invoker.stickyInvoker = nil
+	}
+
+	if sticky && invoker.stickyInvoker != nil && (invoked == nil || !isInvoked(invoker.stickyInvoker, invoked)) {
+		if invoker.availablecheck && invoker.stickyInvoker.IsAvailable() {
+			return invoker.stickyInvoker
+		}
+	}
+
+	selectedInvoker = invoker.doSelectInvoker(lb, invocation, invokers, invoked)
+
+	if sticky {
+		invoker.stickyInvoker = selectedInvoker
+	}
+	return selectedInvoker
+
+}
+
+func (invoker *baseClusterInvoker) doSelectInvoker(lb cluster.LoadBalance, invocation protocol.Invocation, invokers []protocol.Invoker, invoked []protocol.Invoker) protocol.Invoker {
 	if len(invokers) == 1 {
 		return invokers[0]
 	}
+
 	selectedInvoker := lb.Select(invokers, invocation)
 
 	//judge to if the selectedInvoker is invoked
 
-	if !selectedInvoker.IsAvailable() || !invoker.availablecheck || isInvoked(selectedInvoker, invoked) {
+	if (!selectedInvoker.IsAvailable() && invoker.availablecheck) || isInvoked(selectedInvoker, invoked) {
 		// do reselect
 		var reslectInvokers []protocol.Invoker
 
@@ -106,13 +136,12 @@ func (invoker *baseClusterInvoker) doSelect(lb cluster.LoadBalance, invocation p
 		}
 
 		if len(reslectInvokers) > 0 {
-			return lb.Select(reslectInvokers, invocation)
+			selectedInvoker = lb.Select(reslectInvokers, invocation)
 		} else {
 			return nil
 		}
 	}
 	return selectedInvoker
-
 }
 
 func isInvoked(selectedInvoker protocol.Invoker, invoked []protocol.Invoker) bool {

cluster/cluster_impl/base_cluster_invoker_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 cluster_impl
+
+import (
+	"context"
+	"fmt"
+	"testing"
+)
+
+import (
+	"github.com/stretchr/testify/assert"
+)
+
+import (
+	"github.com/apache/dubbo-go/cluster/loadbalance"
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/protocol"
+	"github.com/apache/dubbo-go/protocol/invocation"
+)
+
+func Test_StickyNormal(t *testing.T) {
+	invokers := []protocol.Invoker{}
+	for i := 0; i < 10; i++ {
+		url, _ := common.NewURL(context.TODO(), fmt.Sprintf("dubbo://192.168.1.%v:20000/com.ikurento.user.UserProvider", i))
+		url.SetParam("sticky", "true")
+		invokers = append(invokers, NewMockInvoker(url, 1))
+	}
+	base := &baseClusterInvoker{}
+	base.availablecheck = true
+	invoked := []protocol.Invoker{}
+	result := base.doSelect(loadbalance.NewRandomLoadBalance(), invocation.NewRPCInvocation("getUser", nil, nil), invokers, invoked)
+	result1 := base.doSelect(loadbalance.NewRandomLoadBalance(), invocation.NewRPCInvocation("getUser", nil, nil), invokers, invoked)
+	assert.Equal(t, result, result1)
+}
+func Test_StickyNormalWhenError(t *testing.T) {
+	invokers := []protocol.Invoker{}
+	for i := 0; i < 10; i++ {
+		url, _ := common.NewURL(context.TODO(), fmt.Sprintf("dubbo://192.168.1.%v:20000/com.ikurento.user.UserProvider", i))
+		url.SetParam("sticky", "true")
+		invokers = append(invokers, NewMockInvoker(url, 1))
+	}
+	base := &baseClusterInvoker{}
+	base.availablecheck = true
+
+	invoked := []protocol.Invoker{}
+	result := base.doSelect(loadbalance.NewRandomLoadBalance(), invocation.NewRPCInvocation("getUser", nil, nil), invokers, invoked)
+	invoked = append(invoked, result)
+	result1 := base.doSelect(loadbalance.NewRandomLoadBalance(), invocation.NewRPCInvocation("getUser", nil, nil), invokers, invoked)
+	assert.NotEqual(t, result, result1)
+}

cluster/cluster_impl/failback_cluster_test.go

@@ -67,7 +67,7 @@ func Test_FailbackSuceess(t *testing.T) {
 	invoker := mock.NewMockInvoker(ctrl)
 	clusterInvoker := registerFailback(t, invoker).(*failbackClusterInvoker)
 
-	invoker.EXPECT().GetUrl().Return(failbackUrl).Times(1)
+	invoker.EXPECT().GetUrl().Return(failbackUrl).AnyTimes()
 
 	mockResult := &protocol.RPCResult{Rest: rest{tried: 0, success: true}}
 	invoker.EXPECT().Invoke(gomock.Any()).Return(mockResult)

cluster/cluster_impl/failfast_cluster_test.go

@@ -64,7 +64,7 @@ func Test_FailfastInvokeSuccess(t *testing.T) {
 	invoker := mock.NewMockInvoker(ctrl)
 	clusterInvoker := registerFailfast(t, invoker)
 
-	invoker.EXPECT().GetUrl().Return(failfastUrl)
+	invoker.EXPECT().GetUrl().Return(failfastUrl).AnyTimes()
 
 	mockResult := &protocol.RPCResult{Rest: rest{tried: 0, success: true}}
 
@@ -84,7 +84,7 @@ func Test_FailfastInvokeFail(t *testing.T) {
 	invoker := mock.NewMockInvoker(ctrl)
 	clusterInvoker := registerFailfast(t, invoker)
 
-	invoker.EXPECT().GetUrl().Return(failfastUrl)
+	invoker.EXPECT().GetUrl().Return(failfastUrl).AnyTimes()
 
 	mockResult := &protocol.RPCResult{Err: perrors.New("error")}
 

cluster/cluster_impl/failsafe_cluster_test.go

@@ -64,7 +64,7 @@ func Test_FailSafeInvokeSuccess(t *testing.T) {
 	invoker := mock.NewMockInvoker(ctrl)
 	clusterInvoker := register_failsafe(t, invoker)
 
-	invoker.EXPECT().GetUrl().Return(failsafeUrl)
+	invoker.EXPECT().GetUrl().Return(failsafeUrl).AnyTimes()
 
 	mockResult := &protocol.RPCResult{Rest: rest{tried: 0, success: true}}
 
@@ -83,7 +83,7 @@ func Test_FailSafeInvokeFail(t *testing.T) {
 	invoker := mock.NewMockInvoker(ctrl)
 	clusterInvoker := register_failsafe(t, invoker)
 
-	invoker.EXPECT().GetUrl().Return(failsafeUrl)
+	invoker.EXPECT().GetUrl().Return(failsafeUrl).AnyTimes()
 
 	mockResult := &protocol.RPCResult{Err: perrors.New("error")}
 

cluster/loadbalance/consistent_hash.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 loadbalance
+
+import (
+	"crypto/md5"
+	"encoding/json"
+	"fmt"
+	"hash/crc32"
+	"regexp"
+	"sort"
+	"strconv"
+	"strings"
+)
+
+import (
+	"github.com/apache/dubbo-go/cluster"
+	"github.com/apache/dubbo-go/common/constant"
+	"github.com/apache/dubbo-go/common/extension"
+	"github.com/apache/dubbo-go/protocol"
+)
+
+const (
+	ConsistentHash = "consistenthash"
+	HashNodes      = "hash.nodes"
+	HashArguments  = "hash.arguments"
+)
+
+var (
+	selectors = make(map[string]*ConsistentHashSelector)
+	re        = regexp.MustCompile(constant.COMMA_SPLIT_PATTERN)
+)
+
+func init() {
+	extension.SetLoadbalance(ConsistentHash, NewConsistentHashLoadBalance)
+}
+
+type ConsistentHashLoadBalance struct {
+}
+
+func NewConsistentHashLoadBalance() cluster.LoadBalance {
+	return &ConsistentHashLoadBalance{}
+}
+
+func (lb *ConsistentHashLoadBalance) Select(invokers []protocol.Invoker, invocation protocol.Invocation) protocol.Invoker {
+	methodName := invocation.MethodName()
+	key := invokers[0].GetUrl().ServiceKey() + "." + methodName
+
+	// hash the invokers
+	bs := make([]byte, 0)
+	for _, invoker := range invokers {
+		b, err := json.Marshal(invoker)
+		if err != nil {
+			return nil
+		}
+		bs = append(bs, b...)
+	}
+	hashCode := crc32.ChecksumIEEE(bs)
+	selector, ok := selectors[key]
+	if !ok || selector.hashCode != hashCode {
+		selectors[key] = newConsistentHashSelector(invokers, methodName, hashCode)
+		selector = selectors[key]
+	}
+	return selector.Select(invocation)
+}
+
+type Uint32Slice []uint32
+
+func (s Uint32Slice) Len() int {
+	return len(s)
+}
+
+func (s Uint32Slice) Less(i, j int) bool {
+	return s[i] < s[j]
+}
+
+func (s Uint32Slice) Swap(i, j int) {
+	s[i], s[j] = s[j], s[i]
+}
+
+type ConsistentHashSelector struct {
+	hashCode        uint32
+	replicaNum      int
+	virtualInvokers map[uint32]protocol.Invoker
+	keys            Uint32Slice
+	argumentIndex   []int
+}
+
+func newConsistentHashSelector(invokers []protocol.Invoker, methodName string,
+	hashCode uint32) *ConsistentHashSelector {
+
+	selector := &ConsistentHashSelector{}
+	selector.virtualInvokers = make(map[uint32]protocol.Invoker)
+	selector.hashCode = hashCode
+	url := invokers[0].GetUrl()
+	selector.replicaNum = int(url.GetMethodParamInt(methodName, HashNodes, 160))
+	indices := re.Split(url.GetMethodParam(methodName, HashArguments, "0"), -1)
+	for _, index := range indices {
+		i, err := strconv.Atoi(index)
+		if err != nil {
+			return nil
+		}
+		selector.argumentIndex = append(selector.argumentIndex, i)
+	}
+	for _, invoker := range invokers {
+		u := invoker.GetUrl()
+		address := u.Ip + ":" + u.Port
+		for i := 0; i < selector.replicaNum/4; i++ {
+			digest := md5.Sum([]byte(address + strconv.Itoa(i)))
+			for j := 0; j < 4; j++ {
+				key := selector.hash(digest, j)
+				selector.keys = append(selector.keys, key)
+				selector.virtualInvokers[key] = invoker
+			}
+		}
+	}
+	sort.Sort(selector.keys)
+	return selector
+}
+
+func (c *ConsistentHashSelector) Select(invocation protocol.Invocation) protocol.Invoker {
+	key := c.toKey(invocation.Arguments())
+	digest := md5.Sum([]byte(key))
+	return c.selectForKey(c.hash(digest, 0))
+}
+
+func (c *ConsistentHashSelector) toKey(args []interface{}) string {
+	var sb strings.Builder
+	for i := range c.argumentIndex {
+		if i >= 0 && i < len(args) {
+			fmt.Fprint(&sb, args[i].(string))
+		}
+	}
+	return sb.String()
+}
+
+func (c *ConsistentHashSelector) selectForKey(hash uint32) protocol.Invoker {
+	idx := sort.Search(len(c.keys), func(i int) bool {
+		return c.keys[i] >= hash
+	})
+	if idx == len(c.keys) {
+		idx = 0
+	}
+	return c.virtualInvokers[c.keys[idx]]
+}
+
+func (c *ConsistentHashSelector) hash(digest [16]byte, i int) uint32 {
+	return uint32((digest[3+i*4]&0xFF)<<24) | uint32((digest[2+i*4]&0xFF)<<16) |
+		uint32((digest[1+i*4]&0xFF)<<8) | uint32(digest[i*4]&0xFF)&0xFFFFFFF
+}

cluster/loadbalance/consistent_hash_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 loadbalance
+
+import (
+	"context"
+	"testing"
+)
+
+import (
+	"github.com/stretchr/testify/suite"
+)
+
+import (
+	"github.com/apache/dubbo-go/cluster"
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/protocol"
+	"github.com/apache/dubbo-go/protocol/invocation"
+)
+
+func TestConsistentHashSelectorSuite(t *testing.T) {
+	suite.Run(t, new(consistentHashSelectorSuite))
+}
+
+type consistentHashSelectorSuite struct {
+	suite.Suite
+	selector *ConsistentHashSelector
+}
+
+func (s *consistentHashSelectorSuite) SetupTest() {
+	var invokers []protocol.Invoker
+	url, _ := common.NewURL(context.TODO(),
+		"dubbo://192.168.1.0:20000/org.apache.demo.HelloService?methods.echo.hash.arguments=0,1")
+	invokers = append(invokers, protocol.NewBaseInvoker(url))
+	s.selector = newConsistentHashSelector(invokers, "echo", 999944)
+}
+
+func (s *consistentHashSelectorSuite) TestToKey() {
+	result := s.selector.toKey([]interface{}{"username", "age"})
+	s.Equal(result, "usernameage")
+}
+
+func (s *consistentHashSelectorSuite) TestSelectForKey() {
+	url1, _ := common.NewURL(context.TODO(), "dubbo://192.168.1.0:8080")
+	url2, _ := common.NewURL(context.TODO(), "dubbo://192.168.1.0:8081")
+	s.selector.virtualInvokers = make(map[uint32]protocol.Invoker)
+	s.selector.virtualInvokers[99874] = protocol.NewBaseInvoker(url1)
+	s.selector.virtualInvokers[9999945] = protocol.NewBaseInvoker(url2)
+	s.selector.keys = []uint32{99874, 9999945}
+	result := s.selector.selectForKey(9999944)
+	s.Equal(result.GetUrl().String(), "dubbo://192.168.1.0:8081?")
+}
+
+func TestConsistentHashLoadBalanceSuite(t *testing.T) {
+	suite.Run(t, new(consistentHashLoadBalanceSuite))
+}
+
+type consistentHashLoadBalanceSuite struct {
+	suite.Suite
+	url1     common.URL
+	url2     common.URL
+	url3     common.URL
+	invokers []protocol.Invoker
+	invoker1 protocol.Invoker
+	invoker2 protocol.Invoker
+	invoker3 protocol.Invoker
+	lb       cluster.LoadBalance
+}
+
+func (s *consistentHashLoadBalanceSuite) SetupTest() {
+	var err error
+	s.url1, err = common.NewURL(context.TODO(), "dubbo://192.168.1.0:8080/org.apache.demo.HelloService?methods.echo.hash.arguments=0,1")
+	s.NoError(err)
+	s.url2, err = common.NewURL(context.TODO(), "dubbo://192.168.1.0:8081/org.apache.demo.HelloService?methods.echo.hash.arguments=0,1")
+	s.NoError(err)
+	s.url3, err = common.NewURL(context.TODO(), "dubbo://192.168.1.0:8082/org.apache.demo.HelloService?methods.echo.hash.arguments=0,1")
+	s.NoError(err)
+
+	s.invoker1 = protocol.NewBaseInvoker(s.url1)
+	s.invoker2 = protocol.NewBaseInvoker(s.url2)
+	s.invoker3 = protocol.NewBaseInvoker(s.url3)
+
+	s.invokers = append(s.invokers, s.invoker1, s.invoker2, s.invoker3)
+	s.lb = NewConsistentHashLoadBalance()
+}
+
+func (s *consistentHashLoadBalanceSuite) TestSelect() {
+	args := []interface{}{"name", "password", "age"}
+	invoker := s.lb.Select(s.invokers, invocation.NewRPCInvocation("echo", args, nil))
+	s.Equal(invoker.GetUrl().Location, "192.168.1.0:8080")
+
+	args = []interface{}{"ok", "abc"}
+	invoker = s.lb.Select(s.invokers, invocation.NewRPCInvocation("echo", args, nil))
+	s.Equal(invoker.GetUrl().Location, "192.168.1.0:8082")
+}

cluster/router/condition_router.go

@@ -156,10 +156,13 @@ func parseRule(rule string) (map[string]MatchPair, error) {
 	if len(rule) == 0 {
 		return condition, nil
 	}
-	var pair MatchPair
+
+	var (
+		pair       MatchPair
+		startIndex int
+	)
 	values := gxset.NewSet()
 	reg := regexp.MustCompile(`([&!=,]*)\s*([^&!=,\s]+)`)
-	var startIndex = 0
 	if indexTuple := reg.FindIndex([]byte(rule)); len(indexTuple) > 0 {
 		startIndex = indexTuple[0]
 	}
@@ -227,7 +230,7 @@ func MatchCondition(pairs map[string]MatchPair, url *common.URL, param *common.U
 	if sample == nil {
 		return true, perrors.Errorf("url is not allowed be nil")
 	}
-	result := false
+	var result bool
 	for key, matchPair := range pairs {
 		var sampleValue string
 

common/constant/default.go

@@ -46,8 +46,8 @@ const (
 const (
 	DEFAULT_KEY               = "default"
 	PREFIX_DEFAULT_KEY        = "default."
-	DEFAULT_SERVICE_FILTERS   = "echo,token,accesslog"
-	DEFAULT_REFERENCE_FILTERS = ""
+	DEFAULT_SERVICE_FILTERS   = "echo,token,accesslog,tps,execute,pshutdown"
+	DEFAULT_REFERENCE_FILTERS = "cshutdown"
 	GENERIC_REFERENCE_FILTERS = "generic"
 	GENERIC                   = "$invoke"
 	ECHO                      = "$echo"
@@ -67,3 +67,7 @@ const (
 	APP_DYNAMIC_CONFIGURATORS_CATEGORY = "appdynamicconfigurators"
 	PROVIDER_CATEGORY                  = "providers"
 )
+
+const (
+	COMMA_SPLIT_PATTERN = "\\s*[,]+\\s*"
+)

common/constant/key.go

@@ -48,19 +48,32 @@ const (
 )
 
 const (
-	TIMESTAMP_KEY        = "timestamp"
-	REMOTE_TIMESTAMP_KEY = "remote.timestamp"
-	CLUSTER_KEY          = "cluster"
-	LOADBALANCE_KEY      = "loadbalance"
-	WEIGHT_KEY           = "weight"
-	WARMUP_KEY           = "warmup"
-	RETRIES_KEY          = "retries"
-	BEAN_NAME            = "bean.name"
-	FAIL_BACK_TASKS_KEY  = "failbacktasks"
-	FORKS_KEY            = "forks"
-	DEFAULT_FORKS        = 2
-	DEFAULT_TIMEOUT      = 1000
-	ACCESS_LOG_KEY       = "accesslog"
+	TIMESTAMP_KEY                          = "timestamp"
+	REMOTE_TIMESTAMP_KEY                   = "remote.timestamp"
+	CLUSTER_KEY                            = "cluster"
+	LOADBALANCE_KEY                        = "loadbalance"
+	WEIGHT_KEY                             = "weight"
+	WARMUP_KEY                             = "warmup"
+	RETRIES_KEY                            = "retries"
+	STICKY_KEY                             = "sticky"
+	BEAN_NAME                              = "bean.name"
+	FAIL_BACK_TASKS_KEY                    = "failbacktasks"
+	FORKS_KEY                              = "forks"
+	DEFAULT_FORKS                          = 2
+	DEFAULT_TIMEOUT                        = 1000
+	ACCESS_LOG_KEY                         = "accesslog"
+	TPS_LIMITER_KEY                        = "tps.limiter"
+	TPS_REJECTED_EXECUTION_HANDLER_KEY     = "tps.limit.rejected.handler"
+	TPS_LIMIT_RATE_KEY                     = "tps.limit.rate"
+	DEFAULT_TPS_LIMIT_RATE                 = "-1"
+	TPS_LIMIT_INTERVAL_KEY                 = "tps.limit.interval"
+	DEFAULT_TPS_LIMIT_INTERVAL             = "60000"
+	TPS_LIMIT_STRATEGY_KEY                 = "tps.limit.strategy"
+	EXECUTE_LIMIT_KEY                      = "execute.limit"
+	DEFAULT_EXECUTE_LIMIT                  = "-1"
+	EXECUTE_REJECTED_EXECUTION_HANDLER_KEY = "execute.limit.rejected.handler"
+	PROVIDER_SHUTDOWN_FILTER               = "pshutdown"
+	CONSUMER_SHUTDOWN_FILTER               = "cshutdown"
 )
 
 const (
@@ -90,6 +103,10 @@ const (
 
 const (
 	CONFIG_NAMESPACE_KEY  = "config.namespace"
+	CONFIG_GROUP_KEY      = "config.group"
+	CONFIG_APP_ID_KEY     = "config.appId"
+	CONFIG_CLUSTER_KEY    = "config.cluster"
+	CONFIG_CHECK_KEY      = "config.check"
 	CONFIG_TIMEOUT_KET    = "config.timeout"
 	CONFIG_VERSION_KEY    = "configVersion"
 	COMPATIBLE_CONFIG_KEY = "compatible_config"
@@ -102,6 +119,7 @@ const (
 	ProtocolConfigPrefix       = "dubbo.protocols."
 	ProviderConfigPrefix       = "dubbo.provider."
 	ConsumerConfigPrefix       = "dubbo.consumer."
+	ShutdownConfigPrefix       = "dubbo.shutdown."
 )
 
 const (

common/extension/filter.go

@@ -19,10 +19,12 @@ package extension
 
 import (
 	"github.com/apache/dubbo-go/filter"
+	"github.com/apache/dubbo-go/filter/common"
 )
 
 var (
-	filters = make(map[string]func() filter.Filter)
+	filters                  = make(map[string]func() filter.Filter)
+	rejectedExecutionHandler = make(map[string]func() common.RejectedExecutionHandler)
 )
 
 func SetFilter(name string, v func() filter.Filter) {
@@ -31,7 +33,20 @@ func SetFilter(name string, v func() filter.Filter) {
 
 func GetFilter(name string) filter.Filter {
 	if filters[name] == nil {
-		panic("filter for " + name + " is not existing, make sure you have import the package.")
+		panic("filter for " + name + " is not existing, make sure you have imported the package.")
 	}
 	return filters[name]()
 }
+
+func SetRejectedExecutionHandler(name string, creator func() common.RejectedExecutionHandler) {
+	rejectedExecutionHandler[name] = creator
+}
+
+func GetRejectedExecutionHandler(name string) common.RejectedExecutionHandler {
+	creator, ok := rejectedExecutionHandler[name]
+	if !ok {
+		panic("RejectedExecutionHandler for " + name + " is not existing, make sure you have import the package " +
+			"and you have register it by invoking extension.SetRejectedExecutionHandler.")
+	}
+	return creator()
+}

common/extension/graceful_shutdown.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 extension
+
+import (
+	"container/list"
+)
+
+var (
+	customShutdownCallbacks = list.New()
+)
+
+/**
+ * you should not make any assumption about the order.
+ * For example, if you have more than one callbacks, and you wish the order is:
+ * callback1()
+ * callback2()
+ * ...
+ * callbackN()
+ * Then you should put then together:
+ * func callback() {
+ *     callback1()
+ *     callback2()
+ *     ...
+ *     callbackN()
+ * }
+ * I think the order of custom callbacks should be decided by the users.
+ * Even though I can design a mechanism to support the ordered custom callbacks,
+ * the benefit of that mechanism is low.
+ * And it may introduce much complication for another users.
+ */
+func AddCustomShutdownCallback(callback func()) {
+	customShutdownCallbacks.PushBack(callback)
+}
+
+func GetAllCustomShutdownCallbacks() *list.List {
+	return customShutdownCallbacks
+}

common/extension/tps_limit.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 extension
+
+import (
+	"github.com/apache/dubbo-go/filter/impl/tps"
+)
+
+var (
+	tpsLimitStrategy = make(map[string]tps.TpsLimitStrategyCreator)
+	tpsLimiter       = make(map[string]func() tps.TpsLimiter)
+)
+
+func SetTpsLimiter(name string, creator func() tps.TpsLimiter) {
+	tpsLimiter[name] = creator
+}
+
+func GetTpsLimiter(name string) tps.TpsLimiter {
+	creator, ok := tpsLimiter[name]
+	if !ok {
+		panic("TpsLimiter for " + name + " is not existing, make sure you have import the package " +
+			"and you have register it by invoking extension.SetTpsLimiter.")
+	}
+	return creator()
+}
+
+func SetTpsLimitStrategy(name string, creator tps.TpsLimitStrategyCreator) {
+	tpsLimitStrategy[name] = creator
+}
+
+func GetTpsLimitStrategyCreator(name string) tps.TpsLimitStrategyCreator {
+	creator, ok := tpsLimitStrategy[name]
+	if !ok {
+		panic("TpsLimitStrategy for " + name + " is not existing, make sure you have import the package " +
+			"and you have register it by invoking extension.SetTpsLimitStrategy.")
+	}
+	return creator
+}

common/logger/logger.go

@@ -40,6 +40,11 @@ var (
 	logger Logger
 )
 
+type DubboLogger struct {
+	Logger
+	dynamicLevel zap.AtomicLevel
+}
+
 type Logger interface {
 	Info(args ...interface{})
 	Warn(args ...interface{})
@@ -109,7 +114,8 @@ func InitLogger(conf *zap.Config) {
 		zapLoggerConfig = *conf
 	}
 	zapLogger, _ := zapLoggerConfig.Build(zap.AddCallerSkip(1))
-	logger = zapLogger.Sugar()
+	//logger = zapLogger.Sugar()
+	logger = &DubboLogger{Logger: zapLogger.Sugar(), dynamicLevel: zapLoggerConfig.Level}
 
 	// set getty log
 	getty.SetLogger(logger)
@@ -123,3 +129,22 @@ func SetLogger(log Logger) {
 func GetLogger() Logger {
 	return logger
 }
+
+func SetLoggerLevel(level string) bool {
+	if l, ok := logger.(OpsLogger); ok {
+		l.SetLoggerLevel(level)
+		return true
+	}
+	return false
+}
+
+type OpsLogger interface {
+	Logger
+	SetLoggerLevel(level string)
+}
+
+func (dl *DubboLogger) SetLoggerLevel(level string) {
+	l := new(zapcore.Level)
+	l.Set(level)
+	dl.dynamicLevel.SetLevel(*l)
+}

common/logger/logger_test.go

@@ -65,3 +65,19 @@ func TestInitLog(t *testing.T) {
 	Warnf("%s", "warn")
 	Errorf("%s", "error")
 }
+
+func TestSetLevel(t *testing.T) {
+	err := InitLog("./log.yml")
+	assert.NoError(t, err)
+	Debug("debug")
+	Info("info")
+
+	assert.True(t, SetLoggerLevel("info"))
+	Debug("debug")
+	Info("info")
+
+	SetLogger(GetLogger().(*DubboLogger).Logger)
+	assert.False(t, SetLoggerLevel("debug"))
+	Debug("debug")
+	Info("info")
+}

common/proxy/proxy.go

@@ -181,3 +181,7 @@ func (p *Proxy) Implement(v common.RPCService) {
 func (p *Proxy) Get() common.RPCService {
 	return p.rpc
 }
+
+func (p *Proxy) GetCallback() interface{} {
+	return p.callBack
+}