fix: more invokers with different path (#2000)

apache · Aug 2, 2022 · 9334da0 · 9334da0
1 parent eae6b16
commit 9334da0
Show file tree

Hide file tree

Showing 12 changed files with 56 additions and 38 deletions.
diff --git a/cluster/cluster/available/cluster_invoker_test.go b/cluster/cluster/available/cluster_invoker_test.go
@@ -19,6 +19,7 @@ package available
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"strings"
 	"testing"
@@ -51,7 +52,8 @@ func registerAvailable(invoker *mock.MockInvoker) protocol.Invoker {
 
 	invokers := []protocol.Invoker{}
 	invokers = append(invokers, invoker)
-	invoker.EXPECT().GetUrl().Return(availableUrl)
+	invoker.EXPECT().GetUrl().Return(availableUrl).AnyTimes()
+	invoker.EXPECT().IsAvailable().Return(true).AnyTimes()
 
 	staticDir := static.NewDirectory(invokers)
 	clusterInvoker := availableCluster.Join(staticDir)
@@ -66,8 +68,8 @@ func TestAvailableClusterInvokerSuccess(t *testing.T) {
 	clusterInvoker := registerAvailable(invoker)
 
 	mockResult := &protocol.RPCResult{Rest: clusterpkg.Rest{Tried: 0, Success: true}}
-	invoker.EXPECT().IsAvailable().Return(true)
-	invoker.EXPECT().Invoke(gomock.Any()).Return(mockResult)
+	invoker.EXPECT().IsAvailable().Return(true).AnyTimes()
+	invoker.EXPECT().Invoke(gomock.Any()).Return(mockResult).AnyTimes()
 
 	result := clusterInvoker.Invoke(context.Background(), &invocation.RPCInvocation{})
 
@@ -81,7 +83,10 @@ func TestAvailableClusterInvokerNoAvail(t *testing.T) {
 	invoker := mock.NewMockInvoker(ctrl)
 	clusterInvoker := registerAvailable(invoker)
 
-	invoker.EXPECT().IsAvailable().Return(false)
+	invoker.EXPECT().IsAvailable().Return(false).AnyTimes()
+
+	res := &protocol.RPCResult{Err: errors.New("no provider available")}
+	invoker.EXPECT().Invoke(gomock.Any()).Return(res).AnyTimes()
 
 	result := clusterInvoker.Invoke(context.TODO(), &invocation.RPCInvocation{})
 

diff --git a/cluster/cluster/broadcast/cluster_invoker_test.go b/cluster/cluster/broadcast/cluster_invoker_test.go
@@ -49,11 +49,9 @@ func registerBroadcast(mockInvokers ...*mock.MockInvoker) protocol.Invoker {
 	extension.SetLoadbalance("random", random.NewRandomLoadBalance)
 
 	invokers := []protocol.Invoker{}
-	for i, ivk := range mockInvokers {
+	for _, ivk := range mockInvokers {
 		invokers = append(invokers, ivk)
-		if i == 0 {
-			ivk.EXPECT().GetUrl().Return(broadcastUrl)
-		}
+		ivk.EXPECT().GetUrl().Return(broadcastUrl).AnyTimes()
 	}
 	staticDir := static.NewDirectory(invokers)
 
@@ -72,7 +70,7 @@ func TestBroadcastInvokeSuccess(t *testing.T) {
 	for i := 0; i < 3; i++ {
 		invoker := mock.NewMockInvoker(ctrl)
 		invokers = append(invokers, invoker)
-		invoker.EXPECT().Invoke(gomock.Any()).Return(mockResult)
+		invoker.EXPECT().Invoke(gomock.Any()).Return(mockResult).AnyTimes()
 	}
 
 	clusterInvoker := registerBroadcast(invokers...)
@@ -92,17 +90,17 @@ func TestBroadcastInvokeFailed(t *testing.T) {
 	for i := 0; i < 10; i++ {
 		invoker := mock.NewMockInvoker(ctrl)
 		invokers = append(invokers, invoker)
-		invoker.EXPECT().Invoke(gomock.Any()).Return(mockResult)
+		invoker.EXPECT().Invoke(gomock.Any()).Return(mockResult).AnyTimes()
 	}
 	{
 		invoker := mock.NewMockInvoker(ctrl)
 		invokers = append(invokers, invoker)
-		invoker.EXPECT().Invoke(gomock.Any()).Return(mockFailedResult)
+		invoker.EXPECT().Invoke(gomock.Any()).Return(mockFailedResult).AnyTimes()
 	}
 	for i := 0; i < 10; i++ {
 		invoker := mock.NewMockInvoker(ctrl)
 		invokers = append(invokers, invoker)
-		invoker.EXPECT().Invoke(gomock.Any()).Return(mockResult)
+		invoker.EXPECT().Invoke(gomock.Any()).Return(mockResult).AnyTimes()
 	}
 
 	clusterInvoker := registerBroadcast(invokers...)

diff --git a/cluster/cluster/failback/cluster_test.go b/cluster/cluster/failback/cluster_test.go
@@ -56,7 +56,7 @@ func registerFailback(invoker *mock.MockInvoker) protocol.Invoker {
 	var invokers []protocol.Invoker
 	invokers = append(invokers, invoker)
 
-	invoker.EXPECT().GetUrl().Return(failbackUrl)
+	invoker.EXPECT().GetUrl().Return(failbackUrl).AnyTimes()
 
 	staticDir := static.NewDirectory(invokers)
 	clusterInvoker := failbackCluster.Join(staticDir)
@@ -73,10 +73,10 @@ func TestFailbackSuceess(t *testing.T) {
 
 	invoker.EXPECT().GetUrl().Return(failbackUrl).AnyTimes()
 
-	invoker.EXPECT().IsAvailable().Return(true)
+	invoker.EXPECT().IsAvailable().Return(true).AnyTimes()
 
 	mockResult := &protocol.RPCResult{Rest: clusterpkg.Rest{Tried: 0, Success: true}}
-	invoker.EXPECT().Invoke(gomock.Any()).Return(mockResult)
+	invoker.EXPECT().Invoke(gomock.Any()).Return(mockResult).AnyTimes()
 
 	result := clusterInvoker.Invoke(context.Background(), &invocation.RPCInvocation{})
 	assert.Equal(t, mockResult, result)
@@ -121,7 +121,7 @@ func TestFailbackRetryOneSuccess(t *testing.T) {
 	wg.Wait()
 	assert.Equal(t, int64(0), clusterInvoker.taskList.Len())
 
-	invoker.EXPECT().Destroy().Return()
+	invoker.EXPECT().Destroy().Return().AnyTimes()
 	clusterInvoker.Destroy()
 
 	assert.Equal(t, int64(0), clusterInvoker.taskList.Len())

diff --git a/cluster/cluster/failfast/cluster_test.go b/cluster/cluster/failfast/cluster_test.go
@@ -55,7 +55,7 @@ func registerFailfast(invoker *mock.MockInvoker) protocol.Invoker {
 	invokers = append(invokers, invoker)
 
 	invoker.EXPECT().IsAvailable().Return(true).AnyTimes()
-	invoker.EXPECT().GetUrl().Return(failfastUrl)
+	invoker.EXPECT().GetUrl().Return(failfastUrl).AnyTimes()
 
 	staticDir := static.NewDirectory(invokers)
 	clusterInvoker := failfastCluster.Join(staticDir)
@@ -74,7 +74,7 @@ func TestFailfastInvokeSuccess(t *testing.T) {
 
 	mockResult := &protocol.RPCResult{Rest: clusterpkg.Rest{Tried: 0, Success: true}}
 
-	invoker.EXPECT().Invoke(gomock.Any()).Return(mockResult)
+	invoker.EXPECT().Invoke(gomock.Any()).Return(mockResult).AnyTimes()
 	result := clusterInvoker.Invoke(context.Background(), &invocation.RPCInvocation{})
 
 	assert.NoError(t, result.Error())
@@ -95,7 +95,7 @@ func TestFailfastInvokeFail(t *testing.T) {
 
 	mockResult := &protocol.RPCResult{Err: perrors.New("error")}
 
-	invoker.EXPECT().Invoke(gomock.Any()).Return(mockResult)
+	invoker.EXPECT().Invoke(gomock.Any()).Return(mockResult).AnyTimes()
 	result := clusterInvoker.Invoke(context.Background(), &invocation.RPCInvocation{})
 
 	assert.NotNil(t, result.Error())

diff --git a/cluster/cluster/failsafe/cluster_test.go b/cluster/cluster/failsafe/cluster_test.go
@@ -55,7 +55,7 @@ func registerFailsafe(invoker *mock.MockInvoker) protocol.Invoker {
 	invokers = append(invokers, invoker)
 	invoker.EXPECT().IsAvailable().Return(true).AnyTimes()
 
-	invoker.EXPECT().GetUrl().Return(failsafeUrl)
+	invoker.EXPECT().GetUrl().Return(failsafeUrl).AnyTimes()
 
 	staticDir := static.NewDirectory(invokers)
 	clusterInvoker := failsafeCluster.Join(staticDir)
@@ -75,7 +75,7 @@ func TestFailSafeInvokeSuccess(t *testing.T) {
 
 	mockResult := &protocol.RPCResult{Rest: clusterpkg.Rest{Tried: 0, Success: true}}
 
-	invoker.EXPECT().Invoke(gomock.Any()).Return(mockResult)
+	invoker.EXPECT().Invoke(gomock.Any()).Return(mockResult).AnyTimes()
 	result := clusterInvoker.Invoke(context.Background(), &invocation.RPCInvocation{})
 
 	assert.NoError(t, result.Error())
@@ -95,7 +95,7 @@ func TestFailSafeInvokeFail(t *testing.T) {
 
 	mockResult := &protocol.RPCResult{Err: perrors.New("error")}
 
-	invoker.EXPECT().Invoke(gomock.Any()).Return(mockResult)
+	invoker.EXPECT().Invoke(gomock.Any()).Return(mockResult).AnyTimes()
 	result := clusterInvoker.Invoke(context.Background(), &invocation.RPCInvocation{})
 
 	assert.NoError(t, result.Error())

diff --git a/cluster/cluster/forking/cluster_test.go b/cluster/cluster/forking/cluster_test.go
@@ -51,11 +51,9 @@ func registerForking(mockInvokers ...*mock.MockInvoker) protocol.Invoker {
 	extension.SetLoadbalance(constant.LoadBalanceKeyRoundRobin, roundrobin.NewRRLoadBalance)
 
 	var invokers []protocol.Invoker
-	for i, ivk := range mockInvokers {
+	for _, ivk := range mockInvokers {
 		invokers = append(invokers, ivk)
-		if i == 0 {
-			ivk.EXPECT().GetUrl().Return(forkingUrl)
-		}
+		ivk.EXPECT().GetUrl().Return(forkingUrl).AnyTimes()
 	}
 	staticDir := static.NewDirectory(invokers)
 
@@ -145,14 +143,14 @@ func TestForkingInvokeHalfTimeout(t *testing.T) {
 				func(protocol.Invocation) protocol.Result {
 					wg.Done()
 					return mockResult
-				})
+				}).AnyTimes()
 		} else {
 			invoker.EXPECT().Invoke(gomock.Any()).DoAndReturn(
 				func(protocol.Invocation) protocol.Result {
 					time.Sleep(2 * time.Second)
 					wg.Done()
 					return mockResult
-				})
+				}).AnyTimes()
 		}
 	}
 

diff --git a/cluster/cluster/zoneaware/cluster_interceptor.go b/cluster/cluster/zoneaware/cluster_interceptor.go
@@ -27,8 +27,7 @@ import (
 	"dubbo.apache.org/dubbo-go/v3/protocol"
 )
 
-type interceptor struct {
-}
+type interceptor struct{}
 
 func (z *interceptor) Invoke(ctx context.Context, invoker protocol.Invoker, invocation protocol.Invocation) protocol.Result {
 	key := constant.RegistryKey + "." + constant.RegistryZoneForceKey

diff --git a/cluster/cluster/zoneaware/cluster_invoker_test.go b/cluster/cluster/zoneaware/cluster_invoker_test.go
@@ -63,7 +63,7 @@ func TestZoneWareInvokerWithPreferredSuccess(t *testing.T) {
 			invoker.EXPECT().Invoke(gomock.Any()).DoAndReturn(
 				func(invocation protocol.Invocation) protocol.Result {
 					return mockResult
-				})
+				}).AnyTimes()
 		} else {
 			invoker.EXPECT().Invoke(gomock.Any()).DoAndReturn(
 				func(invocation protocol.Invocation) protocol.Result {

diff --git a/cluster/router/chain/chain.go b/cluster/router/chain/chain.go
@@ -52,7 +52,19 @@ type RouterChain struct {
 
 // Route Loop routers in RouterChain and call Route method to determine the target invokers list.
 func (c *RouterChain) Route(url *common.URL, invocation protocol.Invocation) []protocol.Invoker {
-	finalInvokers := c.invokers
+	finalInvokers := make([]protocol.Invoker, 0, len(c.invokers))
+	// multiple invoker may include different methods, find correct invoker otherwise
+	// will return the invoker without methods
+	for _, invoker := range c.invokers {
+		if invoker.GetURL().ServiceKey() == url.ServiceKey() {
+			finalInvokers = append(finalInvokers, invoker)
+		}
+	}
+
+	if len(finalInvokers) == 0 {
+		finalInvokers = c.invokers
+	}
+
 	for _, r := range c.copyRouters() {
 		finalInvokers = r.Route(finalInvokers, url, invocation)
 	}

diff --git a/cluster/router/tag/router.go b/cluster/router/tag/router.go
@@ -94,7 +94,6 @@ func (p *PriorityRouter) Notify(invokers []protocol.Invoker) {
 		return
 	}
 	p.Process(&config_center.ConfigChangeEvent{Key: key, Value: value, ConfigType: remoting.EventTypeAdd})
-
 }
 
 func (p *PriorityRouter) Process(event *config_center.ConfigChangeEvent) {

diff --git a/cluster/router/tag/router_test.go b/cluster/router/tag/router_test.go
@@ -354,7 +354,8 @@ tags:
  - name: tag1
    addresses: [192.168.0.1:20881]
  - name: tag2
-   addresses: [192.168.0.2:20882]`}
+   addresses: [192.168.0.2:20882]`,
+		}
 		dc, _ := mockFactory.GetDynamicConfiguration(ccUrl)
 		common_cfg.GetEnvInstance().SetDynamicConfiguration(dc)
 		p.Notify(invokerList)
@@ -380,7 +381,8 @@ tags:
 		extension.SetDefaultConfigurator(configurator.NewMockConfigurator)
 		ccUrl, _ := common.NewURL("mock://127.0.0.1:1111")
 		mockFactory := &config_center.MockDynamicConfigurationFactory{
-			Content: `xxxxxx`}
+			Content: `xxxxxx`,
+		}
 		dc, _ := mockFactory.GetDynamicConfiguration(ccUrl)
 		common_cfg.GetEnvInstance().SetDynamicConfiguration(dc)
 		p.Notify(invokerList)

diff --git a/common/url.go b/common/url.go
@@ -284,6 +284,11 @@ func (c *URL) Group() string {
 	return c.GetParam(constant.GroupKey, "")
 }
 
+// Interface get interface
+func (c *URL) Interface() string {
+	return c.GetParam(constant.InterfaceKey, "")
+}
+
 // Version get group
 func (c *URL) Version() string {
 	return c.GetParam(constant.VersionKey, "")
@@ -356,7 +361,7 @@ func (c *URL) Key() string {
 	return buildString
 }
 
-//GetCacheInvokerMapKey get directory cacheInvokerMap key
+// GetCacheInvokerMapKey get directory cacheInvokerMap key
 func (c *URL) GetCacheInvokerMapKey() string {
 	urlNew, _ := NewURL(c.PrimitiveURL)
 
@@ -369,7 +374,7 @@ func (c *URL) GetCacheInvokerMapKey() string {
 
 // ServiceKey gets a unique key of a service.
 func (c *URL) ServiceKey() string {
-	return ServiceKey(c.GetParam(constant.InterfaceKey, strings.TrimPrefix(c.Path, "/")),
+	return ServiceKey(c.GetParam(constant.InterfaceKey, strings.TrimPrefix(c.Path, constant.PathSeparator)),
 		c.GetParam(constant.GroupKey, ""), c.GetParam(constant.VersionKey, ""))
 }
 
@@ -861,7 +866,7 @@ func GetCompareURLEqualFunc() CompareURLEqualFunc {
 	return compareURLEqualFunc
 }
 
-//GetParamDuration get duration if param is invalid or missing will return 3s
+// GetParamDuration get duration if param is invalid or missing will return 3s
 func (c *URL) GetParamDuration(s string, d string) time.Duration {
 	if t, err := time.ParseDuration(c.GetParam(s, d)); err == nil {
 		return t
-Original file line number
+Diff line change
@@ Expand Up @@
     		return
     	}
     	p.Process(&config_center.ConfigChangeEvent{Key: key, Value: value, ConfigType: remoting.EventTypeAdd})
     }
     func (p *PriorityRouter) Process(event *config_center.ConfigChangeEvent) {
@@ Expand Down @@