Merge pull request #294 from zouyx/feature/addRouter

Ftr: condition router

Author: fangyincheng

.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,17 @@
 ::  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
 curl -L https://github.com/dubbogo/resources/raw/master/zookeeper-4unitest/contrib/fatjar/%zkJar% -o remoting/zookeeper/zookeeper-4unittest/contrib/fatjar/%zkJar%
+
+md config_center\zookeeper\zookeeper-4unittest\contrib\fatjar
 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/"
+
+md registry\zookeeper\zookeeper-4unittest\contrib\fatjar
+xcopy /f "remoting/zookeeper/zookeeper-4unittest/contrib/fatjar/%zkJar%" "registry/zookeeper/zookeeper-4unittest/contrib/fatjar/"
+
+md cluster\router\chain\zookeeper-4unittest\contrib\fatjar
+xcopy /f "remoting/zookeeper/zookeeper-4unittest/contrib/fatjar/%zkJar%" "cluster/router/chain/zookeeper-4unittest/contrib/fatjar/"
+
+md cluster\router\condition\zookeeper-4unittest\contrib\fatjar
+xcopy /f "remoting/zookeeper/zookeeper-4unittest/contrib/fatjar/%zkJar%" "cluster/router/condition/zookeeper-4unittest/contrib/fatjar/"

before_ut.sh

@@ -15,7 +15,17 @@
 #  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
 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
+
+mkdir -p config_center/zookeeper/zookeeper-4unittest/contrib/fatjar
 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/
+
+mkdir -p 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/
+
+mkdir -p cluster/router/chain/zookeeper-4unittest/contrib/fatjar
+cp remoting/zookeeper/zookeeper-4unittest/contrib/fatjar/zookeeper-3.4.9-fatjar.jar cluster/router/chain/zookeeper-4unittest/contrib/fatjar
+
+mkdir -p cluster/router/condition/zookeeper-4unittest/contrib/fatjar
+cp remoting/zookeeper/zookeeper-4unittest/contrib/fatjar/zookeeper-3.4.9-fatjar.jar cluster/router/condition/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
 }
 
-// NewBaseDirectory ...
+// 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 ...
+// RouterChain Return router chain in directory
+func (dir *BaseDirectory) RouterChain() router.Chain {
+	return dir.routerChain
+}
+
+// SetRouterChain Set router chain in directory
+func (dir *BaseDirectory) SetRouterChain(routerChain router.Chain) {
+	dir.mutex.Lock()
+	defer dir.mutex.Unlock()
+	dir.routerChain = routerChain
+}
+
+// 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, 0, 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/base_directory_test.go

@@ -0,0 +1,71 @@
+/*
+ * 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 directory
+
+import (
+	"encoding/base64"
+	"fmt"
+	"testing"
+)
+
+import (
+	gxnet "github.com/dubbogo/gost/net"
+	"github.com/stretchr/testify/assert"
+)
+
+import (
+	_ "github.com/apache/dubbo-go/cluster/router/condition"
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/common/constant"
+)
+
+func TestNewBaseDirectory(t *testing.T) {
+	url, _ := common.NewURL(fmt.Sprintf("dubbo://192.168.1.1:20000/com.ikurento.user.UserProvider"))
+	directory := NewBaseDirectory(&url)
+
+	assert.NotNil(t, directory)
+
+	assert.Equal(t, url, directory.GetUrl())
+	assert.Equal(t, &url, directory.GetDirectoryUrl())
+
+}
+
+func TestBuildRouterChain(t *testing.T) {
+	url, _ := common.NewURL(fmt.Sprintf("dubbo://192.168.1.1:20000/com.ikurento.user.UserProvider"))
+	directory := NewBaseDirectory(&url)
+
+	assert.NotNil(t, directory)
+
+	localIP, _ := gxnet.GetLocalIP()
+	rule := base64.URLEncoding.EncodeToString([]byte("true => " + " host = " + localIP))
+	routeURL := getRouteUrl(rule)
+	routerURLs := make([]*common.URL, 0)
+	routerURLs = append(routerURLs, routeURL)
+	directory.SetRouters(routerURLs)
+	chain := directory.RouterChain()
+
+	assert.NotNil(t, chain)
+}
+
+func getRouteUrl(rule string) *common.URL {
+	url, _ := common.NewURL("condition://0.0.0.0/com.foo.BarService")
+	url.AddParam("rule", rule)
+	url.AddParam("force", "true")
+	url.AddParam(constant.ROUTER_KEY, "router")
+	return &url
+}

cluster/directory/static_directory.go

@@ -18,6 +18,11 @@
 package directory
 
 import (
+	perrors "github.com/pkg/errors"
+)
+
+import (
+	"github.com/apache/dubbo-go/cluster/router/chain"
 	"github.com/apache/dubbo-go/common"
 	"github.com/apache/dubbo-go/protocol"
 )
@@ -27,7 +32,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 +58,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 {
@@ -66,3 +81,17 @@ func (dir *staticDirectory) Destroy() {
 		dir.invokers = []protocol.Invoker{}
 	})
 }
+
+// BuildRouterChain build router chain by invokers
+func (dir *staticDirectory) BuildRouterChain(invokers []protocol.Invoker) error {
+	if len(invokers) == 0 {
+		return perrors.Errorf("invokers == null")
+	}
+	url := invokers[0].GetUrl()
+	routerChain, e := chain.NewRouterChain(&url)
+	if e != nil {
+		return e
+	}
+	dir.SetRouterChain(routerChain)
+	return nil
+}

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) {