Add: add notify router for tag router

README.md

@@ -34,7 +34,7 @@ Both extension module and layered project architecture is according to Apache Du
 
 ![dubbo go extend](./doc/pic/arch/dubbo-go-ext.png)
 
-If you wanna know more about dubbo-go, please visit this reference [Project Architeture design](https://github.com/apache/dubbo-go/wiki/dubbo-go-V1.0-design)
+If you wanna know more about dubbo-go, please visit this reference [Project Architecture design](https://github.com/apache/dubbo-go/wiki/dubbo-go-V1.0-design)
 
 ## Feature list ##
 

cluster/cluster_impl/base_cluster_invoker.go

@@ -17,6 +17,10 @@
 
 package cluster_impl
 
+import (
+	"context"
+)
+
 import (
 	gxnet "github.com/dubbogo/gost/net"
 	perrors "github.com/pkg/errors"
@@ -36,6 +40,7 @@ type baseClusterInvoker struct {
 	availablecheck bool
 	destroyed      *atomic.Bool
 	stickyInvoker  protocol.Invoker
+	interceptor    cluster.ClusterInterceptor
 }
 
 func newBaseClusterInvoker(directory cluster.Directory) baseClusterInvoker {
@@ -146,6 +151,20 @@ func (invoker *baseClusterInvoker) doSelectInvoker(lb cluster.LoadBalance, invoc
 	return selectedInvoker
 }
 
+func (invoker *baseClusterInvoker) Invoke(ctx context.Context, invocation protocol.Invocation) protocol.Result {
+	if invoker.interceptor != nil {
+		invoker.interceptor.BeforeInvoker(ctx, invocation)
+
+		result := invoker.interceptor.DoInvoke(ctx, invocation)
+
+		invoker.interceptor.AfterInvoker(ctx, invocation)
+
+		return result
+	}
+
+	return nil
+}
+
 func isInvoked(selectedInvoker protocol.Invoker, invoked []protocol.Invoker) bool {
 	for _, i := range invoked {
 		if i == selectedInvoker {

cluster/cluster_impl/failover_cluster.go

@@ -19,16 +19,15 @@ package cluster_impl
 
 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"
 )
 
 type failoverCluster struct{}
 
-const name = "failover"
-
 func init() {
-	extension.SetCluster(name, NewFailoverCluster)
+	extension.SetCluster(constant.FAILOVER_CLUSTER_NAME, NewFailoverCluster)
 }
 
 // NewFailoverCluster returns a failover cluster instance

cluster/cluster_impl/registry_aware_cluster.go → cluster/cluster_impl/zone_aware_cluster.go

@@ -19,22 +19,26 @@ package cluster_impl
 
 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"
 )
 
-type registryAwareCluster struct{}
+type zoneAwareCluster struct{}
 
 func init() {
-	extension.SetCluster("registryAware", NewRegistryAwareCluster)
+	extension.SetCluster(constant.ZONEAWARE_CLUSTER_NAME, NewZoneAwareCluster)
 }
 
-// NewRegistryAwareCluster returns a registry aware cluster instance
-func NewRegistryAwareCluster() cluster.Cluster {
-	return &registryAwareCluster{}
+// NewZoneAwareCluster returns a zoneaware cluster instance.
+//
+// More than one registry for subscription.
+// Usually it is used for choose between registries.
+func NewZoneAwareCluster() cluster.Cluster {
+	return &zoneAwareCluster{}
 }
 
-// nolint
-func (cluster *registryAwareCluster) Join(directory cluster.Directory) protocol.Invoker {
-	return newRegistryAwareClusterInvoker(directory)
+// Join returns a zoneAwareClusterInvoker instance
+func (cluster *zoneAwareCluster) Join(directory cluster.Directory) protocol.Invoker {
+	return newZoneAwareClusterInvoker(directory)
 }

cluster/cluster_impl/zone_aware_cluster_invoker.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"
+)
+
+import (
+	"github.com/apache/dubbo-go/cluster"
+	"github.com/apache/dubbo-go/common/constant"
+	"github.com/apache/dubbo-go/protocol"
+)
+
+// When there're more than one registry for subscription.
+//
+// This extension provides a strategy to decide how to distribute traffics among them:
+// 1. registry marked as 'preferred=true' has the highest priority.
+// 2. check the zone the current request belongs, pick the registry that has the same zone first.
+// 3. Evenly balance traffic between all registries based on each registry's weight.
+// 4. Pick anyone that's available.
+type zoneAwareClusterInvoker struct {
+	baseClusterInvoker
+}
+
+func newZoneAwareClusterInvoker(directory cluster.Directory) protocol.Invoker {
+	invoke := &zoneAwareClusterInvoker{
+		baseClusterInvoker: newBaseClusterInvoker(directory),
+	}
+	// add self to interceptor
+	invoke.interceptor = invoke
+	return invoke
+}
+
+// nolint
+func (invoker *zoneAwareClusterInvoker) DoInvoke(ctx context.Context, invocation protocol.Invocation) protocol.Result {
+	invokers := invoker.directory.List(invocation)
+
+	err := invoker.checkInvokers(invokers, invocation)
+	if err != nil {
+		return &protocol.RPCResult{Err: err}
+	}
+
+	// First, pick the invoker (XXXClusterInvoker) that comes from the local registry, distinguish by a 'preferred' key.
+	for _, invoker := range invokers {
+		key := constant.REGISTRY_KEY + "." + constant.PREFERRED_KEY
+		if invoker.IsAvailable() && matchParam("true", key, "false", invoker) {
+			return invoker.Invoke(ctx, invocation)
+		}
+	}
+
+	// providers in the registry with the same zone
+	key := constant.REGISTRY_KEY + "." + constant.ZONE_KEY
+	zone := invocation.AttachmentsByKey(key, "")
+	if "" != zone {
+		for _, invoker := range invokers {
+			if invoker.IsAvailable() && matchParam(zone, key, "", invoker) {
+				return invoker.Invoke(ctx, invocation)
+			}
+		}
+
+		force := invocation.AttachmentsByKey(constant.REGISTRY_KEY+"."+constant.ZONE_FORCE_KEY, "")
+		if "true" == force {
+			return &protocol.RPCResult{
+				Err: fmt.Errorf("no registry instance in zone or "+
+					"no available providers in the registry, zone: %v, "+
+					" registries: %v", zone, invoker.GetUrl()),
+			}
+		}
+	}
+
+	// load balance among all registries, with registry weight count in.
+	loadBalance := getLoadBalance(invokers[0], invocation)
+	ivk := invoker.doSelect(loadBalance, invocation, invokers, nil)
+	if ivk != nil && ivk.IsAvailable() {
+		return ivk.Invoke(ctx, invocation)
+	}
+
+	// If none of the invokers has a preferred signal or is picked by the loadBalancer, pick the first one available.
+	for _, invoker := range invokers {
+		if invoker.IsAvailable() {
+			return invoker.Invoke(ctx, invocation)
+		}
+	}
+
+	return &protocol.RPCResult{
+		Err: fmt.Errorf("no provider available in %v", invokers),
+	}
+}
+
+func (invoker *zoneAwareClusterInvoker) BeforeInvoker(ctx context.Context, invocation protocol.Invocation) {
+	key := constant.REGISTRY_KEY + "." + constant.ZONE_FORCE_KEY
+	force := ctx.Value(key)
+
+	if force != nil {
+		switch value := force.(type) {
+		case bool:
+			if value {
+				invocation.SetAttachments(key, "true")
+			}
+		case string:
+			if "true" == value {
+				invocation.SetAttachments(key, "true")
+			}
+		default:
+			// ignore
+		}
+	}
+}
+
+func (invoker *zoneAwareClusterInvoker) AfterInvoker(ctx context.Context, invocation protocol.Invocation) {
+
+}
+
+func matchParam(target, key, def string, invoker protocol.Invoker) bool {
+	return target == invoker.GetUrl().GetParam(key, def)
+}

cluster/cluster_impl/zone_aware_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/golang/mock/gomock"
+	"github.com/stretchr/testify/assert"
+)
+
+import (
+	"github.com/apache/dubbo-go/cluster/directory"
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/common/constant"
+	"github.com/apache/dubbo-go/protocol"
+	"github.com/apache/dubbo-go/protocol/invocation"
+	"github.com/apache/dubbo-go/protocol/mock"
+)
+
+func TestZoneWareInvokerWithPreferredSuccess(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	// In Go versions 1.14+, if you pass a *testing.T
+	// into gomock.NewController(t) you no longer need to call ctrl.Finish().
+	//defer ctrl.Finish()
+
+	mockResult := &protocol.RPCResult{
+		Attrs: map[string]string{constant.PREFERRED_KEY: "true"},
+		Rest:  rest{tried: 0, success: true}}
+
+	var invokers []protocol.Invoker
+	for i := 0; i < 2; i++ {
+		url, _ := common.NewURL(fmt.Sprintf("dubbo://192.168.1.%v:20000/com.ikurento.user.UserProvider", i))
+		invoker := mock.NewMockInvoker(ctrl)
+		invoker.EXPECT().IsAvailable().Return(true).AnyTimes()
+		invoker.EXPECT().GetUrl().Return(url).AnyTimes()
+		if 0 == i {
+			url.SetParam(constant.REGISTRY_KEY+"."+constant.PREFERRED_KEY, "true")
+			invoker.EXPECT().Invoke(gomock.Any()).DoAndReturn(
+				func(invocation protocol.Invocation) protocol.Result {
+					return mockResult
+				})
+		} else {
+			invoker.EXPECT().Invoke(gomock.Any()).DoAndReturn(
+				func(invocation protocol.Invocation) protocol.Result {
+					return &protocol.RPCResult{}
+				})
+		}
+
+		invokers = append(invokers, invoker)
+	}
+
+	zoneAwareCluster := NewZoneAwareCluster()
+	staticDir := directory.NewStaticDirectory(invokers)
+	clusterInvoker := zoneAwareCluster.Join(staticDir)
+
+	result := clusterInvoker.Invoke(context.Background(), &invocation.RPCInvocation{})
+
+	assert.Equal(t, mockResult, result)
+}
+
+func TestZoneWareInvokerWithWeightSuccess(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	// In Go versions 1.14+, if you pass a *testing.T
+	// into gomock.NewController(t) you no longer need to call ctrl.Finish().
+	//defer ctrl.Finish()
+
+	w1 := "50"
+	w2 := "200"
+
+	var invokers []protocol.Invoker
+	for i := 0; i < 2; i++ {
+		url, _ := common.NewURL(fmt.Sprintf("dubbo://192.168.1.%v:20000/com.ikurento.user.UserProvider", i))
+		invoker := mock.NewMockInvoker(ctrl)
+		invoker.EXPECT().IsAvailable().Return(true).AnyTimes()
+		invoker.EXPECT().GetUrl().Return(url).AnyTimes()
+		url.SetParam(constant.REGISTRY_KEY+"."+constant.REGISTRY_LABEL_KEY, "true")
+		if 1 == i {
+			url.SetParam(constant.REGISTRY_KEY+"."+constant.WEIGHT_KEY, w1)
+			invoker.EXPECT().Invoke(gomock.Any()).DoAndReturn(
+				func(invocation protocol.Invocation) protocol.Result {
+					return &protocol.RPCResult{
+						Attrs: map[string]string{constant.WEIGHT_KEY: w1},
+						Rest:  rest{tried: 0, success: true}}
+				}).MaxTimes(100)
+		} else {
+			url.SetParam(constant.REGISTRY_KEY+"."+constant.WEIGHT_KEY, w2)
+			invoker.EXPECT().Invoke(gomock.Any()).DoAndReturn(
+				func(invocation protocol.Invocation) protocol.Result {
+					return &protocol.RPCResult{
+						Attrs: map[string]string{constant.WEIGHT_KEY: w2},
+						Rest:  rest{tried: 0, success: true}}
+				}).MaxTimes(100)
+		}
+		invokers = append(invokers, invoker)
+	}
+
+	zoneAwareCluster := NewZoneAwareCluster()
+	staticDir := directory.NewStaticDirectory(invokers)
+	clusterInvoker := zoneAwareCluster.Join(staticDir)
+
+	var w2Count, w1Count int
+	loop := 50
+	for i := 0; i < loop; i++ {
+		result := clusterInvoker.Invoke(context.Background(), &invocation.RPCInvocation{})
+		if w2 == result.Attachment(constant.WEIGHT_KEY, "0") {
+			w2Count++
+		}
+		if w1 == result.Attachment(constant.WEIGHT_KEY, "0") {
+			w1Count++
+		}
+		assert.NoError(t, result.Error())
+	}
+	t.Logf("loop count : %d, w1 height : %s | count : %d, w2 height : %s | count : %d", loop,
+		w1, w1Count, w2, w2Count)
+}
+
+func TestZoneWareInvokerWithZoneSuccess(t *testing.T) {
+	var zoneArray = []string{"hangzhou", "shanghai"}
+
+	ctrl := gomock.NewController(t)
+	// In Go versions 1.14+, if you pass a *testing.T
+	// into gomock.NewController(t) you no longer need to call ctrl.Finish().
+	//defer ctrl.Finish()
+
+	var invokers []protocol.Invoker
+	for i := 0; i < 2; i++ {
+		zoneValue := zoneArray[i]
+		url, _ := common.NewURL(fmt.Sprintf("dubbo://192.168.1.%v:20000/com.ikurento.user.UserProvider", i))
+		url.SetParam(constant.REGISTRY_KEY+"."+constant.ZONE_KEY, zoneValue)
+
+		invoker := mock.NewMockInvoker(ctrl)
+		invoker.EXPECT().IsAvailable().Return(true).AnyTimes()
+		invoker.EXPECT().GetUrl().Return(url).AnyTimes()
+		invoker.EXPECT().Invoke(gomock.Any()).DoAndReturn(
+			func(invocation protocol.Invocation) protocol.Result {
+				return &protocol.RPCResult{
+					Attrs: map[string]string{constant.ZONE_KEY: zoneValue},
+					Rest:  rest{tried: 0, success: true}}
+			})
+		invokers = append(invokers, invoker)
+	}
+
+	zoneAwareCluster := NewZoneAwareCluster()
+	staticDir := directory.NewStaticDirectory(invokers)
+	clusterInvoker := zoneAwareCluster.Join(staticDir)
+
+	inv := &invocation.RPCInvocation{}
+	// zone hangzhou
+	hz := zoneArray[0]
+	inv.SetAttachments(constant.REGISTRY_KEY+"."+constant.ZONE_KEY, hz)
+
+	result := clusterInvoker.Invoke(context.Background(), inv)
+
+	assert.Equal(t, hz, result.Attachment(constant.ZONE_KEY, ""))
+}
+
+func TestZoneWareInvokerWithZoneForceFail(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	// In Go versions 1.14+, if you pass a *testing.T
+	// into gomock.NewController(t) you no longer need to call ctrl.Finish().
+	//defer ctrl.Finish()
+
+	var invokers []protocol.Invoker
+	for i := 0; i < 2; i++ {
+		url, _ := common.NewURL(fmt.Sprintf("dubbo://192.168.1.%v:20000/com.ikurento.user.UserProvider", i))
+
+		invoker := mock.NewMockInvoker(ctrl)
+		invoker.EXPECT().IsAvailable().Return(true).AnyTimes()
+		invoker.EXPECT().GetUrl().Return(url).AnyTimes()
+		invokers = append(invokers, invoker)
+	}
+
+	zoneAwareCluster := NewZoneAwareCluster()
+	staticDir := directory.NewStaticDirectory(invokers)
+	clusterInvoker := zoneAwareCluster.Join(staticDir)
+
+	inv := &invocation.RPCInvocation{}
+	// zone hangzhou
+	inv.SetAttachments(constant.REGISTRY_KEY+"."+constant.ZONE_KEY, "hangzhou")
+	// zone force
+	inv.SetAttachments(constant.REGISTRY_KEY+"."+constant.ZONE_FORCE_KEY, "true")
+
+	result := clusterInvoker.Invoke(context.Background(), inv)
+
+	assert.NotNil(t, result.Error())
+}

cluster/cluster_impl/registry_aware_cluster_invoker.go → cluster/cluster_interceptor.go

@@ -15,42 +15,25 @@
  * limitations under the License.
  */
 
-package cluster_impl
+package cluster
 
 import (
 	"context"
 )
+
 import (
-	"github.com/apache/dubbo-go/cluster"
-	"github.com/apache/dubbo-go/common/constant"
 	"github.com/apache/dubbo-go/protocol"
 )
 
-type registryAwareClusterInvoker struct {
-	baseClusterInvoker
-}
+// ClusterInterceptor
+// Extension - ClusterInterceptor
+type ClusterInterceptor interface {
+	// Before DoInvoke method
+	BeforeInvoker(ctx context.Context, invocation protocol.Invocation)
 
-func newRegistryAwareClusterInvoker(directory cluster.Directory) protocol.Invoker {
-	return &registryAwareClusterInvoker{
-		baseClusterInvoker: newBaseClusterInvoker(directory),
-	}
-}
+	// After DoInvoke method
+	AfterInvoker(ctx context.Context, invocation protocol.Invocation)
 
-// nolint
-func (invoker *registryAwareClusterInvoker) Invoke(ctx context.Context, invocation protocol.Invocation) protocol.Result {
-	invokers := invoker.directory.List(invocation)
-	//First, pick the invoker (XXXClusterInvoker) that comes from the local registry, distinguish by a 'default' key.
-	for _, invoker := range invokers {
-		if invoker.IsAvailable() && invoker.GetUrl().GetParam(constant.REGISTRY_DEFAULT_KEY, "false") == "true" {
-			return invoker.Invoke(ctx, invocation)
-		}
-	}
-
-	//If none of the invokers has a local signal, pick the first one available.
-	for _, invoker := range invokers {
-		if invoker.IsAvailable() {
-			return invoker.Invoke(ctx, invocation)
-		}
-	}
-	return nil
+	// Corresponding cluster invoke
+	DoInvoke(ctx context.Context, invocation protocol.Invocation) protocol.Result
 }

cluster/directory/base_directory.go

@@ -85,7 +85,14 @@ func (dir *BaseDirectory) SetRouters(urls []*common.URL) {
 	for _, url := range urls {
 		routerKey := url.GetParam(constant.ROUTER_KEY, "")
 
-		if len(routerKey) > 0 {
+		if len(routerKey) == 0 {
+			continue
+		}
+		if url.Protocol == constant.CONDITION_ROUTE_PROTOCOL {
+			if !dir.isProperRouter(url) {
+				continue
+			}
+		}
 		factory := extension.GetRouterFactory(url.Protocol)
 		r, err := factory.NewPriorityRouter(url)
 		if err != nil {
@@ -94,7 +101,6 @@ func (dir *BaseDirectory) SetRouters(urls []*common.URL) {
 		}
 		routers = append(routers, r)
 	}
-	}
 
 	logger.Infof("Init file condition router success, size: %v", len(routers))
 	dir.mutex.Lock()
@@ -104,6 +110,21 @@ func (dir *BaseDirectory) SetRouters(urls []*common.URL) {
 	rc.AddRouters(routers)
 }
 
+func (dir *BaseDirectory) isProperRouter(url *common.URL) bool {
+	app := url.GetParam(constant.APPLICATION_KEY, "")
+	serviceKey := dir.GetUrl().ServiceKey()
+	if serviceKey == "" {
+		serviceKey = dir.GetUrl().SubURL.ServiceKey()
+	}
+	if len(app) > 0 && app == dir.GetUrl().GetParam(constant.APPLICATION_KEY, "") {
+		return true
+	}
+	if url.ServiceKey() == serviceKey {
+		return true
+	}
+	return false
+}
+
 // Destroy Destroy
 func (dir *BaseDirectory) Destroy(doDestroy func()) {
 	if dir.destroyed.CAS(false, true) {

cluster/directory/base_directory_test.go

@@ -37,7 +37,7 @@ import (
 var (
 	url, _ = common.NewURL(
 		fmt.Sprintf("dubbo://%s:%d/com.ikurento.user.UserProvider", constant.LOCAL_HOST_VALUE, constant.DEFAULT_PORT))
-	anyUrl, _ = common.NewURL(fmt.Sprintf("condition://%s/com.foo.BarService", constant.ANYHOST_VALUE))
+	anyURL, _ = common.NewURL(fmt.Sprintf("condition://%s/com.foo.BarService", constant.ANYHOST_VALUE))
 )
 
 func TestNewBaseDirectory(t *testing.T) {
@@ -48,13 +48,17 @@ func TestNewBaseDirectory(t *testing.T) {
 }
 
 func TestBuildRouterChain(t *testing.T) {
-	directory := NewBaseDirectory(&url)
+
+	regURL := url
+	regURL.AddParam(constant.INTERFACE_KEY, "mock-app")
+	directory := NewBaseDirectory(&regURL)
 
 	assert.NotNil(t, directory)
 
 	localIP, _ := gxnet.GetLocalIP()
 	rule := base64.URLEncoding.EncodeToString([]byte("true => " + " host = " + localIP))
-	routeURL := getRouteUrl(rule)
+	routeURL := getRouteURL(rule, anyURL)
+	routeURL.AddParam(constant.INTERFACE_KEY, "mock-app")
 	routerURLs := make([]*common.URL, 0)
 	routerURLs = append(routerURLs, routeURL)
 	directory.SetRouters(routerURLs)
@@ -63,9 +67,53 @@ func TestBuildRouterChain(t *testing.T) {
 	assert.NotNil(t, chain)
 }
 
-func getRouteUrl(rule string) *common.URL {
-	anyUrl.AddParam("rule", rule)
-	anyUrl.AddParam("force", "true")
-	anyUrl.AddParam(constant.ROUTER_KEY, "router")
-	return &url
+func getRouteURL(rule string, u common.URL) *common.URL {
+	ru := u
+	ru.AddParam("rule", rule)
+	ru.AddParam("force", "true")
+	ru.AddParam(constant.ROUTER_KEY, "router")
+	return &ru
+}
+
+func TestIsProperRouter(t *testing.T) {
+	regURL := url
+	regURL.AddParam(constant.APPLICATION_KEY, "mock-app")
+	d := NewBaseDirectory(&regURL)
+	localIP, _ := gxnet.GetLocalIP()
+	rule := base64.URLEncoding.EncodeToString([]byte("true => " + " host = " + localIP))
+	routeURL := getRouteURL(rule, anyURL)
+	routeURL.AddParam(constant.APPLICATION_KEY, "mock-app")
+	rst := d.isProperRouter(routeURL)
+	assert.True(t, rst)
+
+	regURL.AddParam(constant.APPLICATION_KEY, "")
+	regURL.AddParam(constant.INTERFACE_KEY, "com.foo.BarService")
+	d = NewBaseDirectory(&regURL)
+	routeURL = getRouteURL(rule, anyURL)
+	routeURL.AddParam(constant.INTERFACE_KEY, "com.foo.BarService")
+	rst = d.isProperRouter(routeURL)
+	assert.True(t, rst)
+
+	regURL.AddParam(constant.APPLICATION_KEY, "")
+	regURL.AddParam(constant.INTERFACE_KEY, "")
+	d = NewBaseDirectory(&regURL)
+	routeURL = getRouteURL(rule, anyURL)
+	rst = d.isProperRouter(routeURL)
+	assert.True(t, rst)
+
+	regURL.SetParam(constant.APPLICATION_KEY, "")
+	regURL.SetParam(constant.INTERFACE_KEY, "")
+	d = NewBaseDirectory(&regURL)
+	routeURL = getRouteURL(rule, anyURL)
+	routeURL.AddParam(constant.APPLICATION_KEY, "mock-service")
+	rst = d.isProperRouter(routeURL)
+	assert.False(t, rst)
+
+	regURL.SetParam(constant.APPLICATION_KEY, "")
+	regURL.SetParam(constant.INTERFACE_KEY, "")
+	d = NewBaseDirectory(&regURL)
+	routeURL = getRouteURL(rule, anyURL)
+	routeURL.AddParam(constant.INTERFACE_KEY, "mock-service")
+	rst = d.isProperRouter(routeURL)
+	assert.False(t, rst)
 }

cluster/loadbalance/util.go

@@ -28,8 +28,14 @@ import (
 
 // GetWeight gets weight for load balance strategy
 func GetWeight(invoker protocol.Invoker, invocation protocol.Invocation) int64 {
+	var weight int64
 	url := invoker.GetUrl()
-	weight := url.GetMethodParamInt64(invocation.MethodName(), constant.WEIGHT_KEY, constant.DEFAULT_WEIGHT)
+	// Multiple registry scenario, load balance among multiple registries.
+	isRegIvk := url.GetParamBool(constant.REGISTRY_KEY+"."+constant.REGISTRY_LABEL_KEY, false)
+	if isRegIvk {
+		weight = url.GetParamInt(constant.REGISTRY_KEY+"."+constant.WEIGHT_KEY, constant.DEFAULT_WEIGHT)
+	} else {
+		weight = url.GetMethodParamInt64(invocation.MethodName(), constant.WEIGHT_KEY, constant.DEFAULT_WEIGHT)
 
 		if weight > 0 {
 			//get service register time an do warm up time
@@ -46,5 +52,11 @@ func GetWeight(invoker protocol.Invoker, invocation protocol.Invocation) int64 {
 				}
 			}
 		}
+	}
+
+	if weight < 0 {
+		weight = 0
+	}
+
 	return weight
 }

cluster/router/chain/chain_test.go

@@ -66,7 +66,9 @@ func TestNewRouterChain(t *testing.T) {
 	err = z.Create(path)
 	assert.NoError(t, err)
 
-	testyml := `enabled: true
+	testyml := `scope: application
+key: mock-app
+enabled: true
 force: true
 runtime: false
 conditions:
@@ -93,7 +95,7 @@ conditions:
 	assert.NotNil(t, appRouter)
 	assert.NotNil(t, appRouter.RouterRule())
 	rule := appRouter.RouterRule()
-	assert.Equal(t, "", rule.Scope)
+	assert.Equal(t, "application", rule.Scope)
 	assert.True(t, rule.Force)
 	assert.True(t, rule.Enabled)
 	assert.True(t, rule.Valid)
@@ -101,7 +103,7 @@ conditions:
 	assert.Equal(t, testyml, rule.RawRule)
 	assert.Equal(t, false, rule.Runtime)
 	assert.Equal(t, false, rule.Dynamic)
-	assert.Equal(t, "", rule.Key)
+	assert.Equal(t, "mock-app", rule.Key)
 }
 
 func TestNewRouterChainURLNil(t *testing.T) {
@@ -116,7 +118,9 @@ func TestRouterChainAddRouters(t *testing.T) {
 	err = z.Create(path)
 	assert.NoError(t, err)
 
-	testyml := `enabled: true
+	testyml := `scope: application
+key: mock-app
+enabled: true
 force: true
 runtime: false
 conditions:
@@ -184,7 +188,9 @@ func TestRouterChainRouteAppRouter(t *testing.T) {
 	err = z.Create(path)
 	assert.NoError(t, err)
 
-	testyml := `enabled: true
+	testyml := `scope: application
+key: mock-app
+enabled: true
 force: true
 runtime: false
 conditions:

cluster/router/condition/app_router_test.go

@@ -52,7 +52,9 @@ var (
 
 func TestNewAppRouter(t *testing.T) {
 
-	testYML := `enabled: true
+	testYML := `scope: application
+key: mock-app
+enabled: true
 force: true
 runtime: false
 conditions:
@@ -83,7 +85,7 @@ conditions:
 	assert.NotNil(t, appRouter)
 	assert.NotNil(t, appRouter.RouterRule())
 	rule := appRouter.RouterRule()
-	assert.Equal(t, "", rule.Scope)
+	assert.Equal(t, "application", rule.Scope)
 	assert.True(t, rule.Force)
 	assert.True(t, rule.Enabled)
 	assert.True(t, rule.Valid)
@@ -91,13 +93,15 @@ conditions:
 	assert.Equal(t, testYML, rule.RawRule)
 	assert.Equal(t, false, rule.Runtime)
 	assert.Equal(t, false, rule.Dynamic)
-	assert.Equal(t, "", rule.Key)
+	assert.Equal(t, "mock-app", rule.Key)
 	assert.Equal(t, 0, rule.Priority)
 }
 
 func TestGenerateConditions(t *testing.T) {
 
-	testYML := `enabled: true
+	testYML := `scope: application
+key: mock-app
+enabled: true
 force: true
 runtime: false
 conditions:
@@ -135,7 +139,9 @@ conditions:
 
 func TestProcess(t *testing.T) {
 
-	testYML := `enabled: true
+	testYML := `scope: application
+key: mock-app
+enabled: true
 force: true
 runtime: false
 conditions:
@@ -165,7 +171,8 @@ conditions:
 
 	assert.Equal(t, 1, len(appRouter.conditionRouters))
 
-	testNewYML := `
+	testNewYML := `scope: application
+key: mock-app
 enabled: true
 force: true
 runtime: false

cluster/router/condition/file.go

@@ -20,6 +20,7 @@ package condition
 import (
 	"encoding/base64"
 	"net/url"
+	"regexp"
 	"strconv"
 	"strings"
 	"sync"
@@ -71,11 +72,53 @@ func (f *FileConditionRouter) URL() common.URL {
 			common.WithParamsValue(constant.RouterPriority, strconv.Itoa(routerRule.Priority)),
 			common.WithParamsValue(constant.RULE_KEY, base64.URLEncoding.EncodeToString([]byte(rule))),
 			common.WithParamsValue(constant.ROUTER_KEY, constant.CONDITION_ROUTE_PROTOCOL),
-			common.WithParamsValue(constant.CATEGORY_KEY, constant.ROUTERS_CATEGORY))
+			common.WithParamsValue(constant.CATEGORY_KEY, constant.ROUTERS_CATEGORY),
+		)
+		if routerRule.Scope == constant.RouterApplicationScope {
+			f.url.AddParam(constant.APPLICATION_KEY, routerRule.Key)
+			return
+		}
+		grp, srv, ver, e := parseServiceRouterKey(routerRule.Key)
+		if e != nil {
+			return
+		}
+		if len(grp) > 0 {
+			f.url.AddParam(constant.GROUP_KEY, grp)
+		}
+		if len(ver) > 0 {
+			f.url.AddParam(constant.VERSION_KEY, ver)
+		}
+		if len(srv) > 0 {
+			f.url.AddParam(constant.INTERFACE_KEY, srv)
+		}
 	})
 	return f.url
 }
 
+// The input value must follow [{group}/]{service}[:{version}] pattern
+// the returning strings are representing group, service, version respectively.
+// input: mock-group/mock-service:1.0.0 ==> "mock-group", "mock-service", "1.0.0"
+// input: mock-group/mock-service ==> "mock-group", "mock-service", ""
+// input: mock-service:1.0.0 ==> "", "mock-service", "1.0.0"
+// For more samples, please refer to unit test.
+func parseServiceRouterKey(key string) (string, string, string, error) {
+	if len(strings.TrimSpace(key)) == 0 {
+		return "", "", "", nil
+	}
+	reg := regexp.MustCompile(`(.*/{1})?([^:/]+)(:{1}[^:]*)?`)
+	strs := reg.FindAllStringSubmatch(key, -1)
+	if strs == nil || len(strs) > 1 {
+		return "", "", "", perrors.Errorf("Invalid key, service key must follow [{group}/]{service}[:{version}] pattern")
+	}
+	if len(strs[0]) != 4 {
+		return "", "", "", perrors.Errorf("Parse service router key failed")
+	}
+	grp := strings.TrimSpace(strings.TrimRight(strs[0][1], "/"))
+	srv := strings.TrimSpace(strs[0][2])
+	ver := strings.TrimSpace(strings.TrimLeft(strs[0][3], ":"))
+	return grp, srv, ver, nil
+}
+
 func parseCondition(conditions []string) string {
 	var when, then string
 	for _, condition := range conditions {

cluster/router/condition/file_test.go

@@ -26,7 +26,9 @@ import (
 )
 
 func TestLoadYmlConfig(t *testing.T) {
-	router, e := NewFileConditionRouter([]byte(`priority: 1
+	router, e := NewFileConditionRouter([]byte(`scope: application
+key: mock-app
+priority: 1
 force: true
 conditions :
   - "a => b"
@@ -47,12 +49,78 @@ func TestParseCondition(t *testing.T) {
 }
 
 func TestFileRouterURL(t *testing.T) {
-	router, e := NewFileConditionRouter([]byte(`priority: 1
+	router, e := NewFileConditionRouter([]byte(`scope: application
+key: mock-app
+priority: 1
 force: true
 conditions :
   - "a => b"
   - "c => d"`))
 	assert.Nil(t, e)
 	assert.NotNil(t, router)
-	assert.Equal(t, "condition://0.0.0.0:?category=routers&force=true&priority=1&router=condition&rule=YSAmIGMgPT4gYiAmIGQ%3D", router.URL().String())
+	assert.Equal(t, "condition://0.0.0.0:?application=mock-app&category=routers&force=true&priority=1&router=condition&rule=YSAmIGMgPT4gYiAmIGQ%3D", router.URL().String())
+
+	router, e = NewFileConditionRouter([]byte(`scope: service
+key: mock-service
+priority: 1
+force: true
+conditions :
+  - "a => b"
+  - "c => d"`))
+	assert.Nil(t, e)
+	assert.NotNil(t, router)
+	assert.Equal(t, "condition://0.0.0.0:?category=routers&force=true&interface=mock-service&priority=1&router=condition&rule=YSAmIGMgPT4gYiAmIGQ%3D", router.URL().String())
+
+	router, e = NewFileConditionRouter([]byte(`scope: service
+key: grp1/mock-service:v1
+priority: 1
+force: true
+conditions :
+  - "a => b"
+  - "c => d"`))
+	assert.Nil(t, e)
+	assert.NotNil(t, router)
+	assert.Equal(t, "condition://0.0.0.0:?category=routers&force=true&group=grp1&interface=mock-service&priority=1&router=condition&rule=YSAmIGMgPT4gYiAmIGQ%3D&version=v1", router.URL().String())
+}
+
+func TestParseServiceRouterKey(t *testing.T) {
+	testString := " mock-group / mock-service:1.0.0"
+	grp, srv, ver, err := parseServiceRouterKey(testString)
+	assert.Equal(t, "mock-group", grp)
+	assert.Equal(t, "mock-service", srv)
+	assert.Equal(t, "1.0.0", ver)
+
+	testString = "mock-group/mock-service"
+	grp, srv, ver, err = parseServiceRouterKey(testString)
+	assert.Equal(t, "mock-group", grp)
+	assert.Equal(t, "mock-service", srv)
+	assert.Equal(t, "", ver)
+
+	testString = "mock-service:1.0.0"
+	grp, srv, ver, err = parseServiceRouterKey(testString)
+	assert.Equal(t, "", grp)
+	assert.Equal(t, "mock-service", srv)
+	assert.Equal(t, "1.0.0", ver)
+
+	testString = "mock-service"
+	grp, srv, ver, err = parseServiceRouterKey(testString)
+	assert.Equal(t, "", grp)
+	assert.Equal(t, "mock-service", srv)
+	assert.Equal(t, "", ver)
+
+	testString = "/mock-service:"
+	grp, srv, ver, err = parseServiceRouterKey(testString)
+	assert.Equal(t, "", grp)
+	assert.Equal(t, "mock-service", srv)
+	assert.Equal(t, "", ver)
+
+	testString = "grp:mock-service:123"
+	grp, srv, ver, err = parseServiceRouterKey(testString)
+	assert.Error(t, err)
+
+	testString = ""
+	grp, srv, ver, err = parseServiceRouterKey(testString)
+	assert.Equal(t, "", grp)
+	assert.Equal(t, "", srv)
+	assert.Equal(t, "", ver)
 }

cluster/router/condition/router.go

@@ -117,7 +117,13 @@ func NewConditionRouter(url *common.URL) (*ConditionRouter, error) {
 	}
 
 	router.url = url
-	router.priority = url.GetParamInt(constant.RouterPriority, 0)
+	var defaultPriority int64 = 0
+	if url.GetParam(constant.APPLICATION_KEY, "") != "" {
+		defaultPriority = 150
+	} else if url.GetParam(constant.INTERFACE_KEY, "") != "" {
+		defaultPriority = 140
+	}
+	router.priority = url.GetParamInt(constant.RouterPriority, defaultPriority)
 	router.Force = url.GetParamBool(constant.RouterForce, false)
 	router.enabled = url.GetParamBool(constant.RouterEnabled, true)
 

cluster/router/condition/router_rule.go

@@ -28,12 +28,13 @@ import (
 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/yaml"
 )
 
 // RouterRule RouterRule config read from config file or config center
 type RouterRule struct {
-	router.BaseRouterRule `yaml:",inline""`
+	router.BaseRouterRule `yaml:",inline"`
 	Conditions            []string
 }
 
@@ -57,7 +58,7 @@ func getRule(rawRule string) (*RouterRule, error) {
 		return r, err
 	}
 	r.RawRule = rawRule
-	if len(r.Conditions) != 0 {
+	if len(r.Conditions) > 0 && len(r.Key) > 0 && (r.Scope == constant.RouterApplicationScope || r.Scope == constant.RouterServiceScope) {
 		r.Valid = true
 	}
 

cluster/router/condition/router_rule_test.go

@@ -32,6 +32,7 @@ import (
 func TestGetRule(t *testing.T) {
 	testyml := `
 scope: application
+key: test-provider
 runtime: true
 force: false
 conditions:
@@ -50,10 +51,31 @@ conditions:
 	assert.True(t, rule.Runtime)
 	assert.Equal(t, false, rule.Force)
 	assert.Equal(t, testyml, rule.RawRule)
-	assert.True(t, true, rule.Valid)
+	assert.True(t, rule.Valid)
 	assert.Equal(t, false, rule.Enabled)
 	assert.Equal(t, false, rule.Dynamic)
-	assert.Equal(t, "", rule.Key)
+	assert.Equal(t, "test-provider", rule.Key)
+
+	testyml = `
+key: test-provider
+runtime: true
+force: false
+conditions:
+  - >
+    method!=sayHello =>`
+	rule, e = getRule(testyml)
+	assert.Nil(t, e)
+	assert.False(t, rule.Valid)
+
+	testyml = `
+scope: noApplication
+key: test-provider
+conditions:
+  - >
+    method!=sayHello =>`
+	rule, e = getRule(testyml)
+	assert.Nil(t, e)
+	assert.False(t, rule.Valid)
 }
 
 func TestIsMatchGlobPattern(t *testing.T) {

cluster/router/condition/router_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 condition
+
+import (
+	"testing"
+)
+
+import (
+	"github.com/apache/dubbo-go/common"
+	"github.com/dubbogo/gost/container/set"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestParseRule(t *testing.T) {
+	testString := ``
+	matchPair, err := parseRule(testString)
+	assert.Nil(t, err)
+	assert.EqualValues(t, matchPair, make(map[string]MatchPair, 16))
+
+	testString = `method!=sayHello&application=sayGoodBye`
+	matchPair, err = parseRule(testString)
+	assert.Nil(t, err)
+	assert.EqualValues(t, matchPair["method"].Mismatches, gxset.NewSet("sayHello"))
+	assert.EqualValues(t, matchPair["application"].Matches, gxset.NewSet("sayGoodBye"))
+
+	testString = `noRule`
+	matchPair, err = parseRule(testString)
+	assert.Nil(t, err)
+	assert.EqualValues(t, matchPair["noRule"].Mismatches, gxset.NewSet())
+	assert.EqualValues(t, matchPair["noRule"].Matches, gxset.NewSet())
+
+	testString = `method!=sayHello,sayGoodBye`
+	matchPair, err = parseRule(testString)
+	assert.Nil(t, err)
+	assert.EqualValues(t, matchPair["method"].Mismatches, gxset.NewSet(`sayHello`, `sayGoodBye`))
+
+	testString = `method!=sayHello,sayGoodDay=sayGoodBye`
+	matchPair, err = parseRule(testString)
+	assert.Nil(t, err)
+	assert.EqualValues(t, matchPair["method"].Mismatches, gxset.NewSet(`sayHello`, `sayGoodDay`))
+	assert.EqualValues(t, matchPair["method"].Matches, gxset.NewSet(`sayGoodBye`))
+}
+
+func TestNewConditionRouter(t *testing.T) {
+	url, _ := common.NewURL(`condition://0.0.0.0:?application=mock-app&category=routers&force=true&priority=1&router=condition&rule=YSAmIGMgPT4gYiAmIGQ%3D`)
+	router, err := NewConditionRouter(&url)
+	assert.Nil(t, err)
+	assert.Equal(t, true, router.Enabled())
+	assert.Equal(t, true, router.Force)
+	assert.Equal(t, int64(1), router.Priority())
+	whenRule, _ := parseRule("a & c")
+	thenRule, _ := parseRule("b & d")
+	assert.EqualValues(t, router.WhenCondition, whenRule)
+	assert.EqualValues(t, router.ThenCondition, thenRule)
+
+	router, err = NewConditionRouter(nil)
+	assert.Error(t, err)
+
+	url, _ = common.NewURL(`condition://0.0.0.0:?application=mock-app&category=routers&force=true&priority=1&router=condition&rule=YSAmT4gYiAmIGQ%3D`)
+	router, err = NewConditionRouter(&url)
+	assert.Error(t, err)
+
+	url, _ = common.NewURL(`condition://0.0.0.0:?application=mock-app&category=routers&force=true&router=condition&rule=YSAmIGMgPT4gYiAmIGQ%3D`)
+	router, err = NewConditionRouter(&url)
+	assert.Nil(t, err)
+	assert.Equal(t, int64(150), router.Priority())
+
+	url, _ = common.NewURL(`condition://0.0.0.0:?category=routers&force=true&interface=mock-service&router=condition&rule=YSAmIGMgPT4gYiAmIGQ%3D`)
+	router, err = NewConditionRouter(&url)
+	assert.Nil(t, err)
+	assert.Equal(t, int64(140), router.Priority())
+}

cluster/router/healthcheck/default_health_check.go

@@ -49,7 +49,7 @@ type DefaultHealthChecker struct {
 // and the current active request
 func (c *DefaultHealthChecker) IsHealthy(invoker protocol.Invoker) bool {
 	urlStatus := protocol.GetURLStatus(invoker.GetUrl())
-	if c.isCircuitBreakerTripped(urlStatus) || urlStatus.GetActive() > c.GetOutStandingRequestConutLimit() {
+	if c.isCircuitBreakerTripped(urlStatus) || urlStatus.GetActive() > c.GetOutStandingRequestCountLimit() {
 		logger.Debugf("Invoker [%s] is currently in circuitbreaker tripped state", invoker.GetUrl().Key())
 		return false
 	}
@@ -92,18 +92,18 @@ func (c *DefaultHealthChecker) getCircuitBreakerSleepWindowTime(status *protocol
 	return int64(sleepWindow)
 }
 
-// GetOutStandingRequestConutLimit return the requestSuccessiveFailureThreshold bound to this DefaultHealthChecker
+// GetRequestSuccessiveFailureThreshold return the requestSuccessiveFailureThreshold bound to this DefaultHealthChecker
 func (c *DefaultHealthChecker) GetRequestSuccessiveFailureThreshold() int32 {
 	return c.requestSuccessiveFailureThreshold
 }
 
-// GetOutStandingRequestConutLimit return the circuitTrippedTimeoutFactor bound to this DefaultHealthChecker
+// GetCircuitTrippedTimeoutFactor return the circuitTrippedTimeoutFactor bound to this DefaultHealthChecker
 func (c *DefaultHealthChecker) GetCircuitTrippedTimeoutFactor() int32 {
 	return c.circuitTrippedTimeoutFactor
 }
 
-// GetOutStandingRequestConutLimit return the outStandingRequestConutLimit bound to this DefaultHealthChecker
-func (c *DefaultHealthChecker) GetOutStandingRequestConutLimit() int32 {
+// GetOutStandingRequestCountLimit return the outStandingRequestConutLimit bound to this DefaultHealthChecker
+func (c *DefaultHealthChecker) GetOutStandingRequestCountLimit() int32 {
 	return c.outStandingRequestConutLimit
 }