Add routing-related features

cluster/router.go

@@ -25,11 +25,12 @@ import (
 // Extension - Router
 
 type RouterFactory interface {
-	Router(common.URL) Router
+	GetRouter(common.URL) (Router, error)
 }
 
 type Router interface {
-	Route([]protocol.Invoker, common.URL, protocol.Invocation) []protocol.Invoker
+	Route([]protocol.Invoker, common.URL, protocol.Invocation) ([]protocol.Invoker, error)
+	CompareTo(router Router) int
 }
 
 type RouterChain struct {

cluster/router/condition_router.go

+package router
+
+import (
+	"github.com/apache/dubbo-go/cluster"
+	"github.com/apache/dubbo-go/common/constant"
+	"regexp"
+	"strings"
+
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/protocol"
+	perrors "github.com/pkg/errors"
+)
+
+const (
+	RoutePattern = `([&!=,]*)\\s*([^&!=,\\s]+)`
+)
+
+type ConditionRouter struct {
+	Pattern       string
+	Url           common.URL
+	Priority      int64
+	Force         bool
+	WhenCondition map[string]MatchPair
+	ThenCondition map[string]MatchPair
+}
+
+func (c *ConditionRouter) Route(invokers []protocol.Invoker, url common.URL, invocation protocol.Invocation) ([]protocol.Invoker, error) {
+	if len(invokers) == 0 {
+		return invokers, nil
+	}
+	if !c.matchWhen(url, invocation) {
+		return invokers, nil
+	}
+	var result []protocol.Invoker
+	if len(c.ThenCondition) == 0 {
+		return result, nil
+	}
+	for _, invoker := range invokers {
+		if c.matchThen(invoker.GetUrl(), url) {
+			result = append(result, invoker)
+		}
+	}
+	if len(result) > 0 {
+		return result, nil
+	} else if c.Force {
+		//todo 日志
+		return result, nil
+	}
+	return result, nil
+}
+
+func (c ConditionRouter) CompareTo(r cluster.Router) int {
+	var result int
+	router, ok := r.(*ConditionRouter)
+	if r == nil || !ok {
+		return 1
+	}
+	if c.Priority == router.Priority {
+		result = strings.Compare(c.Url.String(), router.Url.String())
+	} else {
+		if c.Priority > router.Priority {
+			result = 1
+		} else {
+			result = -1
+		}
+	}
+	return result
+}
+
+func newConditionRouter(url common.URL) (*ConditionRouter, error) {
+	var whenRule string
+	var thenRule string
+	//rule := url.GetParam("rule", "")
+	w, err := parseRule(whenRule)
+	if err != nil {
+		return nil, perrors.Errorf("%s", "")
+	}
+
+	t, err := parseRule(thenRule)
+	if err != nil {
+		return nil, perrors.Errorf("%s", "")
+	}
+	when := If(whenRule == "" || "true" == whenRule, make(map[string]MatchPair), w).(map[string]MatchPair)
+	then := If(thenRule == "" || "false" == thenRule, make(map[string]MatchPair), t).(map[string]MatchPair)
+
+	return &ConditionRouter{
+		RoutePattern,
+		url,
+		url.GetParamInt("Priority", 0),
+		url.GetParamBool("Force", false),
+		when,
+		then,
+	}, nil
+}
+
+func parseRule(rule string) (map[string]MatchPair, error) {
+
+	condition := make(map[string]MatchPair)
+	if rule == "" {
+		return condition, nil
+	}
+	var pair MatchPair
+	values := make(map[string]interface{})
+
+	reg := regexp.MustCompile(`([&!=,]*)\s*([^&!=,\s]+)`)
+
+	startIndex := reg.FindIndex([]byte(rule))
+	matches := reg.FindAllSubmatch([]byte(rule), -1)
+	for _, groups := range matches {
+		separator := string(groups[1])
+		content := string(groups[2])
+
+		switch separator {
+		case "":
+			pair = MatchPair{}
+			condition[content] = pair
+		case "&":
+			if r, ok := condition[content]; ok {
+				pair = r
+			} else {
+				pair = MatchPair{}
+				condition[content] = pair
+			}
+		case "=":
+			if &pair == nil {
+				return nil, perrors.Errorf("Illegal route rule \"%s\", The error char '%s' at index %d before \"%d\".", rule, separator, startIndex[0], startIndex[0])
+			}
+			values = pair.Matches
+			values[content] = ""
+		case "!=":
+			if &pair == nil {
+				return nil, perrors.Errorf("Illegal route rule \"%s\", The error char '%s' at index %d before \"%d\".", rule, separator, startIndex[0], startIndex[0])
+			}
+			values = pair.Matches
+			values[content] = ""
+		case ",":
+			if len(values) == 0 {
+				return nil, perrors.Errorf("Illegal route rule \"%s\", The error char '%s' at index %d before \"%d\".", rule, separator, startIndex[0], startIndex[0])
+			}
+			values[content] = ""
+		default:
+			return nil, perrors.Errorf("Illegal route rule \"%s\", The error char '%s' at index %d before \"%d\".", rule, separator, startIndex[0], startIndex[0])
+
+		}
+	}
+
+	return condition, nil
+	//var pair MatchPair
+
+}
+
+func (c *ConditionRouter) matchWhen(url common.URL, invocation protocol.Invocation) bool {
+
+	return len(c.WhenCondition) == 0 || len(c.WhenCondition) == 0 || matchCondition(c.WhenCondition, &url, nil, invocation)
+}
+func (c *ConditionRouter) matchThen(url common.URL, param common.URL) bool {
+
+	return !(len(c.ThenCondition) == 0) && matchCondition(c.ThenCondition, &url, &param, nil)
+}
+
+func matchCondition(pairs map[string]MatchPair, url *common.URL, param *common.URL, invocation protocol.Invocation) bool {
+	sample := url.ToMap()
+	result := false
+	for key, matchPair := range pairs {
+		var sampleValue string
+
+		if invocation != nil && ((constant.METHOD_KEY == key) || (constant.METHOD_KEYS == key)) {
+			sampleValue = invocation.MethodName()
+		} else {
+			sampleValue = sample[key]
+			if &sampleValue == nil {
+				sampleValue = sample[constant.DEFAULT_KEY_PREFIX+key]
+			}
+		}
+		if &sampleValue != nil {
+			if !matchPair.isMatch(sampleValue, param) {
+				return false
+			} else {
+				result = true
+			}
+		} else {
+			if !(len(matchPair.Matches) == 0) {
+				return false
+			} else {
+				result = true
+			}
+		}
+
+	}
+	return result
+}
+
+func If(b bool, t, f interface{}) interface{} {
+	if b {
+		return t
+	}
+	return f
+}
+
+type MatchPair struct {
+	Matches    map[string]interface{}
+	Mismatches map[string]interface{}
+}
+
+func (pair MatchPair) isMatch(s string, param *common.URL) bool {
+
+	return false
+}

cluster/router/condition_router_test.go

+package router
+
+import (
+	"context"
+	perrors "errors"
+	"fmt"
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/common/logger"
+	"github.com/apache/dubbo-go/protocol"
+	"github.com/apache/dubbo-go/protocol/invocation"
+	"github.com/stretchr/testify/assert"
+	"net"
+	"testing"
+)
+
+type MockInvoker struct {
+	url       common.URL
+	available bool
+	destroyed bool
+
+	successCount int
+}
+
+func NewMockInvoker(url common.URL, successCount int) *MockInvoker {
+	return &MockInvoker{
+		url:          url,
+		available:    true,
+		destroyed:    false,
+		successCount: successCount,
+	}
+}
+
+func (bi *MockInvoker) GetUrl() common.URL {
+	return bi.url
+}
+
+func (bi *MockInvoker) IsAvailable() bool {
+	return bi.available
+}
+
+func (bi *MockInvoker) IsDestroyed() bool {
+	return bi.destroyed
+}
+
+type rest struct {
+	tried   int
+	success bool
+}
+
+var count int
+
+func (bi *MockInvoker) Invoke(invocation protocol.Invocation) protocol.Result {
+	count++
+	var success bool
+	var err error = nil
+	if count >= bi.successCount {
+		success = true
+	} else {
+		err = perrors.New("error")
+	}
+	result := &protocol.RPCResult{Err: err, Rest: rest{tried: count, success: success}}
+
+	return result
+}
+
+func (bi *MockInvoker) Destroy() {
+	logger.Infof("Destroy invoker: %v", bi.GetUrl().String())
+	bi.destroyed = true
+	bi.available = false
+}
+func Test_parseRule(t *testing.T) {
+	parseRule("host = 10.20.153.10 => host = 10.20.153.11")
+
+}
+func LocalIp() string {
+	addrs, err := net.InterfaceAddrs()
+	if err != nil {
+		fmt.Println(err)
+	}
+	var ip string = "localhost"
+	for _, address := range addrs {
+		if ipnet, ok := address.(*net.IPNet); ok && !ipnet.IP.IsLoopback() {
+			if ipnet.IP.To4() != nil {
+				ip = ipnet.IP.String()
+			}
+		}
+	}
+	return ip
+}
+func TestRoute_matchWhen(t *testing.T) {
+
+}
+func TestRoute_matchFilter(t *testing.T) {
+	url1, _ := common.NewURL(context.TODO(), "dubbo://10.20.3.3:20880/com.foo.BarService?default.serialization=fastjson")
+	url2, _ := common.NewURL(context.TODO(), fmt.Sprintf("dubbo://%s:20880/com.foo.BarService", LocalIp()))
+	url3, _ := common.NewURL(context.TODO(), fmt.Sprintf("dubbo://%s:20880/com.foo.BarService", LocalIp()))
+	invokers := []protocol.Invoker{NewMockInvoker(url1, 1), NewMockInvoker(url2, 2), NewMockInvoker(url3, 3)}
+	option := common.WithParamsValue("rule", "host = "+LocalIp()+" => "+" host = 10.20.3.3")
+	option1 := common.WithParamsValue("force", "true")
+	sUrl, _ := common.NewURL(context.TODO(), "condition://0.0.0.0/com.foo.BarService", option, option1)
+
+	router1, _ := NewConditionRouterFactory().GetRouter(sUrl)
+	cUrl, _ := common.NewURL(context.TODO(), "consumer://"+LocalIp()+"/com.foo.BarService")
+
+	routers, _ := router1.Route(invokers, cUrl, &invocation.RPCInvocation{})
+	assert.Equal(t, 1, len(routers))
+
+}

cluster/router/router_factory.go

+package router
+
+import (
+	"github.com/apache/dubbo-go/cluster"
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/common/extension"
+)
+
+func init() {
+	extension.SetRouterFactory("conditionRouterFactory", NewConditionRouterFactory)
+}
+
+type ConditionRouterFactory struct {
+}
+
+func NewConditionRouterFactory() cluster.RouterFactory {
+	return ConditionRouterFactory{}
+}
+func (c ConditionRouterFactory) GetRouter(url common.URL) (cluster.Router, error) {
+	return newConditionRouter(url)
+}

common/constant/default.go

@@ -33,6 +33,7 @@ const (
 
 const (
 	DEFAULT_KEY               = "default"
+	DEFAULT_KEY_PREFIX        = "default."
 	DEFAULT_SERVICE_FILTERS   = "echo"
 	DEFAULT_REFERENCE_FILTERS = ""
 	ECHO                      = "$echo"

common/constant/key.go

@@ -67,3 +67,7 @@ const (
 	OWNER_KEY        = "owner"
 	ENVIRONMENT_KEY  = "environment"
 )
+const (
+	METHOD_KEY  = "method"
+	METHOD_KEYS = "methods"
+)

common/extension/engine.go

+package extension
+
+import (
+	"github.com/apache/dubbo-go/cluster"
+	"github.com/apache/dubbo-go/common"
+)
+
+var (
+	engines = make(map[string]func(config *common.URL) (cluster.Router, error))
+)
+
+func SetEngine(name string, v func(config *common.URL) (cluster.Router, error)) {
+	engines[name] = v
+}
+
+func GetEngine(name string, config *common.URL) (cluster.Router, error) {
+	if engines[name] == nil {
+		panic("registry for " + name + " is not existing, make sure you have import the package.")
+	}
+	return engines[name](config)
+
+}

common/extension/router_factory.go

+package extension
+
+import (
+	"github.com/apache/dubbo-go/cluster"
+)
+
+var (
+	routers = make(map[string]func() cluster.RouterFactory)
+)
+
+func SetRouterFactory(name string, fun func() cluster.RouterFactory) {
+	routers[name] = fun
+}
+
+func GetRouterFactory(name string) cluster.RouterFactory {
+	if routers[name] == nil {
+		panic("router_factory for " + name + " is not existing, make sure you have import the package.")
+	}
+	return routers[name]()
+
+}

common/url.go

@@ -270,6 +270,17 @@ func (c URL) GetParam(s string, d string) string {
 	return r
 }
 
+// GetParamBool
+func (c URL) GetParamBool(s string, d bool) bool {
+
+	var r bool
+	var err error
+	if r, err = strconv.ParseBool(c.Params.Get(s)); err != nil {
+		return d
+	}
+	return r
+}
+
 func (c URL) GetParamInt(s string, d int64) int64 {
 	var r int
 	var err error
@@ -296,6 +307,31 @@ func (c URL) GetMethodParam(method string, key string, d string) string {
 	return r
 }
 
+// ToMap transfer URL to Map
+func (c URL) ToMap() map[string]string {
+
+	paramsMap := make(map[string]string)
+	if c.Protocol != "" {
+		paramsMap["protocol"] = c.Protocol
+	}
+	if c.Username != "" {
+		paramsMap["username"] = c.Username
+	}
+	if c.Password != "" {
+		paramsMap["password"] = c.Password
+	}
+	if c.Ip != "" {
+		paramsMap["host"] = c.Ip
+	}
+	if c.Protocol != "" {
+		paramsMap["protocol"] = c.Protocol
+	}
+	if c.Path != "" {
+		paramsMap["path"] = c.Path
+	}
+	return paramsMap
+}
+
 // configuration  > reference config >service config
 //  in this function we should merge the reference local url config into the service url from registry.
 //TODO configuration merge, in the future , the configuration center's config should merge too.

go.mod

@@ -4,9 +4,11 @@ require (
 	github.com/dubbogo/getty v0.0.0-20190523180329-bdf5e640ea53
 	github.com/dubbogo/hessian2 v0.0.0-20190526221400-d5610bbd0a41
 	github.com/pkg/errors v0.8.1
+	github.com/robertkrimen/otto v0.0.0-20180617131154-15f95af6e78d
 	github.com/samuel/go-zookeeper v0.0.0-20180130194729-c4fab1ac1bec
 	github.com/stretchr/testify v1.3.0
 	go.uber.org/atomic v1.4.0
 	go.uber.org/zap v1.10.0
+	gopkg.in/sourcemap.v1 v1.0.5 // indirect
 	gopkg.in/yaml.v2 v2.2.2
 )

go.sum

 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/dubbogo/getty v0.0.0-20190523180329-bdf5e640ea53 h1:bniSNoC4xnAbrx4estwc9F0qkWnh6ZDsAS0y9d7mPos=
 github.com/dubbogo/getty v0.0.0-20190523180329-bdf5e640ea53/go.mod h1:cRMSuoCmwc5lULFFnYZTxyCfZhObmRTNbS7XRnPNHSo=
 github.com/dubbogo/hessian2 v0.0.0-20190526221400-d5610bbd0a41 h1:lNtW7+aN8oBdCoEuny0rOqOkL5duI4Cu3+G8vqibX48=
@@ -15,19 +12,14 @@ github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoA
 github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
-github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/samuel/go-zookeeper v0.0.0-20180130194729-c4fab1ac1bec h1:6ncX5ko6B9LntYM0YBRXkiSaZMmLYeZ/NWcmeB43mMY=
+github.com/robertkrimen/otto v0.0.0-20180617131154-15f95af6e78d h1:1VUlQbCfkoSGv7qP7Y+ro3ap1P1pPZxgdGVqiTVy5C4=
+github.com/robertkrimen/otto v0.0.0-20180617131154-15f95af6e78d/go.mod h1:xvqspoSXJTIpemEonrMDFq6XzwHYYgToXWj5eRX1OtY=
 github.com/samuel/go-zookeeper v0.0.0-20180130194729-c4fab1ac1bec h1:6ncX5ko6B9LntYM0YBRXkiSaZMmLYeZ/NWcmeB43mMY=
 github.com/samuel/go-zookeeper v0.0.0-20180130194729-c4fab1ac1bec/go.mod h1:gi+0XIa01GRL2eRQVjQkKGqKF3SF9vZR/HnPullcV2E=
-github.com/samuel/go-zookeeper v0.0.0-20180130194729-c4fab1ac1bec/go.mod h1:gi+0XIa01GRL2eRQVjQkKGqKF3SF9vZR/HnPullcV2E=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
-github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
-github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q=
 github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
-github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 go.uber.org/atomic v1.4.0 h1:cxzIVoETapQEqDhQu3QfnvXAV4AlzcvUCxkVUFw3+EU=
 go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/multierr v1.1.0 h1:HoEmRHQPVSqub6w2z2d2EOVs2fjyFRGyofhKuyDq0QI=
@@ -41,8 +33,7 @@ golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5h
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw=
+gopkg.in/sourcemap.v1 v1.0.5 h1:inv58fC9f9J3TK2Y2R1NPntXEn3/wjWHkonhIUODNTI=
+gopkg.in/sourcemap.v1 v1.0.5/go.mod h1:2RlvNNSMglmRrcvhfuzp4hQHwOtjxlbjX7UPY/GXb78=
 gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
-gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=