Ftr: condition router

.travis.yml

@@ -10,10 +10,7 @@ install: true
 
 script:
   - go fmt ./... && [[ -z `git status -s` ]]
-  - 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/
-  - cp remoting/zookeeper/zookeeper-4unittest/contrib/fatjar/zookeeper-3.4.9-fatjar.jar registry/zookeeper/zookeeper-4unittest/contrib/fatjar/
+  - chmod u+x before_ut.sh && ./before_ut.sh
   - go mod vendor && go test ./... -coverprofile=coverage.txt -covermode=atomic
 
 after_success:

before_ut.bat

@@ -15,7 +15,8 @@
 ::  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
+md remoting\zookeeper\zookeeper-4unittest\contrib\fatjar config_center\zookeeper\zookeeper-4unittest\contrib\fatjar registry\zookeeper\zookeeper-4unittest\contrib\fatjar cluster\router\chain\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/"
+xcopy /f "remoting/zookeeper/zookeeper-4unittest/contrib/fatjar/%zkJar%" "registry/zookeeper/zookeeper-4unittest/contrib/fatjar/"
+xcopy /f "remoting/zookeeper/zookeeper-4unittest/contrib/fatjar/%zkJar%" "cluster/router/chain/zookeeper-4unittest/contrib/fatjar/"

before_ut.sh

@@ -15,7 +15,8 @@
 #  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
+mkdir -p remoting/zookeeper/zookeeper-4unittest/contrib/fatjar config_center/zookeeper/zookeeper-4unittest/contrib/fatjar registry/zookeeper/zookeeper-4unittest/contrib/fatjar cluster/router/chain/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/
-cp remoting/zookeeper/zookeeper-4unittest/contrib/fatjar/zookeeper-3.4.9-fatjar.jar registry/zookeeper/zookeeper-4unittest/contrib/fatjar/
+cp remoting/zookeeper/zookeeper-4unittest/contrib/fatjar/zookeeper-3.4.9-fatjar.jar registry/zookeeper/zookeeper-4unittest/contrib/fatjar/
+cp remoting/zookeeper/zookeeper-4unittest/contrib/fatjar/zookeeper-3.4.9-fatjar.jar cluster/router/chain/zookeeper-4unittest/contrib/fatjar

cluster/directory/base_directory.go

@@ -20,39 +20,94 @@ package directory
 import (
 	"sync"
 )
+
 import (
+	"github.com/dubbogo/gost/container/set"
 	"go.uber.org/atomic"
 )
+
 import (
+	"github.com/apache/dubbo-go/cluster/router"
+	"github.com/apache/dubbo-go/cluster/router/chain"
 	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/common/constant"
+	"github.com/apache/dubbo-go/common/extension"
+	"github.com/apache/dubbo-go/common/logger"
 )
 
-// BaseDirectory ...
+var routerURLSet = gxset.NewSet()
+
+// BaseDirectory Abstract implementation of Directory: Invoker list returned from this Directory's list method have been filtered by Routers
 type BaseDirectory struct {
 	url       *common.URL
 	destroyed *atomic.Bool
-	mutex     sync.Mutex
+	// this mutex for change the properties in BaseDirectory, like routerChain , destroyed etc
+	mutex       sync.Mutex
+	routerChain router.Chain
+}
+
+// RouterChain Return router chain in directory
+func (dir *BaseDirectory) RouterChain() router.Chain {
+	return dir.routerChain
 }
 
-// NewBaseDirectory ...
+// SetRouterChain Set router chain in directory
+func (dir *BaseDirectory) SetRouterChain(routerChain router.Chain) {
+	dir.mutex.Lock()
+	defer dir.mutex.Unlock()
+	dir.routerChain = routerChain
+}
+
+// NewBaseDirectory Create BaseDirectory with URL
 func NewBaseDirectory(url *common.URL) BaseDirectory {
 	return BaseDirectory{
-		url:       url,
-		destroyed: atomic.NewBool(false),
+		url:         url,
+		destroyed:   atomic.NewBool(false),
+		routerChain: &chain.RouterChain{},
 	}
 }
 
-// GetUrl ...
+// GetUrl Get URL
 func (dir *BaseDirectory) GetUrl() common.URL {
 	return *dir.url
 }
 
-// GetDirectoryUrl ...
+// GetDirectoryUrl Get URL instance
 func (dir *BaseDirectory) GetDirectoryUrl() *common.URL {
 	return dir.url
 }
 
-// Destroy ...
+// SetRouters Convert url to routers and add them into dir.routerChain
+func (dir *BaseDirectory) SetRouters(urls []*common.URL) {
+	if len(urls) == 0 {
+		return
+	}
+
+	routers := make([]router.Router, len(urls), len(urls))
+
+	for _, url := range urls {
+		routerKey := url.GetParam(constant.ROUTER_KEY, "")
+
+		if len(routerKey) > 0 {
+			factory := extension.GetRouterFactory(url.Protocol)
+			r, err := factory.NewRouter(url)
+			if err != nil {
+				logger.Errorf("Create router fail. router key: %s, error: %v", routerKey, url.Service(), err)
+				return
+			}
+			routers = append(routers, r)
+		}
+	}
+
+	logger.Infof("Init file condition router success, size: %v", len(routers))
+	dir.mutex.Lock()
+	rc := dir.routerChain
+	dir.mutex.Unlock()
+
+	rc.AddRouters(routers)
+}
+
+// Destroy Destroy
 func (dir *BaseDirectory) Destroy(doDestroy func()) {
 	if dir.destroyed.CAS(false, true) {
 		dir.mutex.Lock()
@@ -61,7 +116,18 @@ func (dir *BaseDirectory) Destroy(doDestroy func()) {
 	}
 }
 
-// IsAvailable ...
+// IsAvailable Once directory init finish, it will change to true
 func (dir *BaseDirectory) IsAvailable() bool {
 	return !dir.destroyed.Load()
 }
+
+// GetRouterURLSet Return router URL
+func GetRouterURLSet() *gxset.HashSet {
+	return routerURLSet
+}
+
+// AddRouterURLSet Add router URL
+// Router URL will init in config/config_loader.go
+func AddRouterURLSet(url *common.URL) {
+	routerURLSet.Add(url)
+}

cluster/directory/static_directory.go

@@ -27,7 +27,7 @@ type staticDirectory struct {
 	invokers []protocol.Invoker
 }
 
-// NewStaticDirectory ...
+// NewStaticDirectory Create a new staticDirectory with invokers
 func NewStaticDirectory(invokers []protocol.Invoker) *staticDirectory {
 	var url common.URL
 
@@ -53,11 +53,21 @@ func (dir *staticDirectory) IsAvailable() bool {
 	return true
 }
 
+// List List invokers
 func (dir *staticDirectory) List(invocation protocol.Invocation) []protocol.Invoker {
-	//TODO:Here should add router
-	return dir.invokers
+	l := len(dir.invokers)
+	invokers := make([]protocol.Invoker, l, l)
+	copy(invokers, dir.invokers)
+	routerChain := dir.RouterChain()
+
+	if routerChain == nil {
+		return invokers
+	}
+	dirUrl := dir.GetUrl()
+	return routerChain.Route(invokers, &dirUrl, invocation)
 }
 
+// Destroy Destroy
 func (dir *staticDirectory) Destroy() {
 	dir.BaseDirectory.Destroy(func() {
 		for _, ivk := range dir.invokers {

cluster/directory/static_directory_test.go

@@ -40,7 +40,9 @@ func Test_StaticDirList(t *testing.T) {
 	}
 
 	staticDir := NewStaticDirectory(invokers)
-	assert.Len(t, staticDir.List(&invocation.RPCInvocation{}), 10)
+	list := staticDir.List(&invocation.RPCInvocation{})
+
+	assert.Len(t, list, 10)
 }
 
 func Test_StaticDirDestroy(t *testing.T) {

cluster/router/chain/chain.go

@@ -0,0 +1,119 @@
+/*
+ * 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 chain
+
+import (
+	"sort"
+	"sync"
+)
+
+import (
+	perrors "github.com/pkg/errors"
+)
+
+import (
+	"github.com/apache/dubbo-go/cluster/router"
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/common/extension"
+	"github.com/apache/dubbo-go/common/logger"
+	"github.com/apache/dubbo-go/protocol"
+)
+
+// RouterChain Router chain
+type RouterChain struct {
+	// Full list of addresses from registry, classified by method name.
+	invokers []protocol.Invoker
+	// Containing all routers, reconstruct every time 'route://' urls change.
+	routers []router.Router
+	// Fixed router instances: ConfigConditionRouter, TagRouter, e.g., the rule for each instance may change but the
+	// instance will never delete or recreate.
+	builtinRouters []router.Router
+
+	mutex sync.RWMutex
+}
+
+// Route Loop routers in RouterChain and call Route method to determine the target invokers list.
+func (c RouterChain) Route(invoker []protocol.Invoker, url *common.URL, invocation protocol.Invocation) []protocol.Invoker {
+	finalInvokers := invoker
+	l := len(c.routers)
+	rs := make([]router.Router, l, l)
+	c.mutex.RLock()
+	copy(rs, c.routers)
+	c.mutex.RUnlock()
+
+	for _, r := range rs {
+		finalInvokers = r.Route(finalInvokers, url, invocation)
+	}
+	return finalInvokers
+}
+
+// AddRouters Add routers to router chain
+// New a array add builtinRouters which is not sorted in RouterChain and routers
+// Sort the array
+// Replace router array in RouterChain
+func (c RouterChain) AddRouters(routers []router.Router) {
+	newRouters := make([]router.Router, 0, len(c.builtinRouters)+len(routers))
+	newRouters = append(newRouters, c.builtinRouters...)
+	newRouters = append(newRouters, routers...)
+	sortRouter(newRouters)
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+	c.routers = newRouters
+}
+
+// NewRouterChain Use url to init router chain
+// Loop routerFactories and call NewRouter method
+func NewRouterChain(url *common.URL) (*RouterChain, error) {
+	routerFactories := extension.GetRouterFactories()
+	if len(routerFactories) == 0 {
+		return nil, perrors.Errorf("Illegal route rule!")
+	}
+	routers := make([]router.Router, 0, len(routerFactories))
+	for key, routerFactory := range routerFactories {
+		r, err := routerFactory().NewRouter(url)
+		if r == nil || err != nil {
+			logger.Errorf("router chain build router fail! routerFactories key:%s  error:%s", key, err.Error())
+			continue
+		}
+		routers = append(routers, r)
+	}
+
+	newRouters := make([]router.Router, len(routers))
+	copy(newRouters, routers)
+
+	sortRouter(newRouters)
+
+	chain := &RouterChain{
+		builtinRouters: routers,
+		routers:        newRouters,
+	}
+
+	return chain, nil
+}
+
+// sortRouter Sort router instance by priority with stable algorithm
+func sortRouter(routers []router.Router) {
+	sort.Stable(byPriority(routers))
+}
+
+// byPriority Sort by priority
+type byPriority []router.Router
+
+func (a byPriority) Len() int           { return len(a) }
+func (a byPriority) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
+func (a byPriority) Less(i, j int) bool { return a[i].Priority() < a[j].Priority() }