Merge da6f6f6 into 4640167

.github/workflows/github-actions.yml

@@ -56,7 +56,7 @@ jobs:
 
     - name: gofmt
       run: |
-        go fmt ./... && git checkout -- go.mod && git status && [[ -z `git status -s` ]]
+        go fmt ./... && git status && [[ -z `git status -s` ]]
         # diff -u <(echo -n) <(gofmt -d -s .)
 
     - name: Install go ci lint
@@ -69,9 +69,9 @@ jobs:
       run: |
         make verify
 
-#    - name: Integrate Test
-#      run: |
-#        chmod +x integrate_test.sh && ./integrate_test.sh ${{github.event.pull_request.head.repo.full_name}} ${{github.event.pull_request.head.sha}}
+    - name: Integrate Test
+      run: |
+        chmod +x integrate_test.sh && ./integrate_test.sh ${{github.event.pull_request.head.repo.full_name}} ${{github.event.pull_request.head.sha}}
 
     - name: Post Coverage
       run: bash <(curl -s https://codecov.io/bash)

README.md

@@ -177,7 +177,7 @@ If you are willing to do some code contributions and document contributions to [
 
 ## Community
 
-If u want to communicate with our community, pls scan the following  [dubbobo Ding-Ding QR code](https://mmbiz.qpic.cn/mmbiz_jpg/yvBJb5IiafvnHVBdtia30dxA2hKotr9DEckWsZ7aOJcDWDaSVMGwLmYv8GRgIQtqb4C2svicp8nVkMmGy7yKC5tyA/640?wx_fmt=jpeg&tp=webp&wxfrom=5&wx_lazy=1&wx_co=1) or search our commnity DingDing group code 31363295.
+If u want to communicate with our community, pls scan the following dubbobo DingDing QR code or search our commnity DingDing group code 31363295.
 
 <div>
 <table>

common/constant/key.go

@@ -49,8 +49,10 @@ const (
 	PORT_KEY                 = "port"
 	PROTOCOL_KEY             = "protocol"
 	PATH_SEPARATOR           = "/"
-	//DUBBO_KEY                = "dubbo"
+	// DUBBO_KEY                = "dubbo"
 	SSL_ENABLED_KEY = "ssl-enabled"
+	// PARAMS_TYPE_Key key used in pass through invoker factory, to define param type
+	PARAMS_TYPE_Key = "parameter-type-names"
 )
 
 const (
@@ -199,6 +201,8 @@ const (
 	// default deregister critical server after
 	DEFAULT_DEREGISTER_TIME = "20s"
 	DEREGISTER_AFTER        = "consul-deregister-critical-service-after"
+	// PassThroughProxyFactoryKey is key of proxy factory with raw data input service
+	PassThroughProxyFactoryKey = "dubbo-raw"
 )
 
 const (

common/proxy/proxy_factory/pass_through.go

@@ -0,0 +1,123 @@
+/*
+ * 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 proxy_factory
+
+import (
+	"context"
+	"reflect"
+)
+
+import (
+	perrors "github.com/pkg/errors"
+)
+
+import (
+	"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/proxy"
+	"github.com/apache/dubbo-go/protocol"
+)
+
+func init() {
+	extension.SetProxyFactory(constant.PassThroughProxyFactoryKey, NewPassThroughProxyFactory)
+}
+
+// PassThroughProxyFactory is the factory of PassThroughProxyInvoker
+type PassThroughProxyFactory struct {
+}
+
+// NewPassThroughProxyFactory returns a proxy factory instance
+func NewPassThroughProxyFactory(_ ...proxy.Option) proxy.ProxyFactory {
+	return &PassThroughProxyFactory{}
+}
+
+// GetProxy gets a proxy
+func (factory *PassThroughProxyFactory) GetProxy(invoker protocol.Invoker, url *common.URL) *proxy.Proxy {
+	return factory.GetAsyncProxy(invoker, nil, url)
+}
+
+// GetAsyncProxy gets a async proxy
+func (factory *PassThroughProxyFactory) GetAsyncProxy(invoker protocol.Invoker, callBack interface{}, url *common.URL) *proxy.Proxy {
+	//create proxy
+	attachments := map[string]string{}
+	attachments[constant.ASYNC_KEY] = url.GetParam(constant.ASYNC_KEY, "false")
+	return proxy.NewProxy(invoker, callBack, attachments)
+}
+
+// GetInvoker gets a invoker
+func (factory *PassThroughProxyFactory) GetInvoker(url *common.URL) protocol.Invoker {
+	return &PassThroughProxyInvoker{
+		ProxyInvoker: &ProxyInvoker{
+			BaseInvoker: *protocol.NewBaseInvoker(url),
+		},
+	}
+}
+
+// PassThroughProxyInvoker is a invoker struct, it calls service with specific method 'Serivce' and params:
+// Service(method string, argsTypes []string, args [][]byte, attachment map[string]interface{})
+// PassThroughProxyInvoker pass through raw invocation data and method name to service, which will deal with them.
+type PassThroughProxyInvoker struct {
+	*ProxyInvoker
+}
+
+// Invoke is used to call service method by invocation
+func (pi *PassThroughProxyInvoker) Invoke(ctx context.Context, invocation protocol.Invocation) protocol.Result {
+	result := &protocol.RPCResult{}
+	result.SetAttachments(invocation.Attachments())
+	url := getProviderURL(pi.GetUrl())
+
+	arguments := invocation.Arguments()
+	srv := common.ServiceMap.GetServiceByServiceKey(url.Protocol, url.ServiceKey())
+
+	var args [][]byte
+	if len(arguments) > 0 {
+		args = make([][]byte, 0, len(arguments))
+		for _, arg := range arguments {
+			if v, ok := arg.([]byte); ok {
+				args = append(args, v)
+			} else {
+				result.Err = perrors.New("the param type is not []byte")
+				return result
+			}
+		}
+	}
+	method := srv.Method()["Service"]
+
+	in := make([]reflect.Value, 5)
+	in = append(in, srv.Rcvr())
+	in = append(in, reflect.ValueOf(invocation.MethodName()))
+	in = append(in, reflect.ValueOf(invocation.Attachment(constant.PARAMS_TYPE_Key)))
+	in = append(in, reflect.ValueOf(args))
+	in = append(in, reflect.ValueOf(invocation.Attachments()))
+
+	returnValues := method.Method().Func.Call(in)
+
+	var retErr interface{}
+	replyv := returnValues[0]
+	retErr = returnValues[1].Interface()
+
+	if retErr != nil {
+		result.SetError(retErr.(error))
+		return result
+	}
+	if replyv.IsValid() && (replyv.Kind() != reflect.Ptr || replyv.Kind() == reflect.Ptr && replyv.Elem().IsValid()) {
+		result.SetResult(replyv.Interface())
+	}
+	return result
+}

common/proxy/proxy_factory/pass_through_test.go

@@ -0,0 +1,61 @@
+/*
+ * 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 proxy_factory
+
+import (
+	"fmt"
+	"testing"
+)
+
+import (
+	"github.com/stretchr/testify/assert"
+)
+
+import (
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/protocol"
+)
+
+func TestPassThroughProxyFactoryGetProxy(t *testing.T) {
+	proxyFactory := NewPassThroughProxyFactory()
+	url := common.NewURLWithOptions()
+	proxy := proxyFactory.GetProxy(protocol.NewBaseInvoker(url), url)
+	assert.NotNil(t, proxy)
+}
+
+type TestPassThroughProxyFactoryAsync struct {
+}
+
+func (u *TestPassThroughProxyFactoryAsync) CallBack(res common.CallbackResponse) {
+	fmt.Println("CallBack res:", res)
+}
+
+func TestPassThroughProxyFactoryGetAsyncProxy(t *testing.T) {
+	proxyFactory := NewPassThroughProxyFactory()
+	url := common.NewURLWithOptions()
+	async := &TestPassThroughProxyFactoryAsync{}
+	proxy := proxyFactory.GetAsyncProxy(protocol.NewBaseInvoker(url), async.CallBack, url)
+	assert.NotNil(t, proxy)
+}
+
+func TestPassThroughProxyFactoryGetInvoker(t *testing.T) {
+	proxyFactory := NewPassThroughProxyFactory()
+	url := common.NewURLWithOptions()
+	invoker := proxyFactory.GetInvoker(url)
+	assert.True(t, invoker.IsAvailable())
+}

go.mod

@@ -1,6 +1,6 @@
 module github.com/apache/dubbo-go
 
-go 1.13
+go 1.15
 
 require (
 	github.com/NYTimes/gziphandler v1.1.1 // indirect

registry/directory/directory.go

@@ -187,9 +187,12 @@ func (dir *RegistryDirectory) refreshAllInvokers(events []*registry.ServiceEvent
 		// loop the updateEvents
 		for _, event := range addEvents {
 			logger.Debugf("registry update, result{%s}", event)
-			logger.Infof("selector add service url{%s}", event.Service)
-			// FIXME: routers are built in every address notification?
-			dir.configRouters()
+			if event != nil && event.Service != nil {
+				logger.Infof("selector add service url{%s}", event.Service.String())
+			}
+			if event != nil && event.Service != nil && constant.ROUTER_PROTOCOL == event.Service.Protocol {
+				dir.configRouters()
+			}
 			if oldInvoker, _ := dir.doCacheInvoker(event.Service); oldInvoker != nil {
 				oldInvokers = append(oldInvokers, oldInvoker)
 			}
@@ -241,8 +244,9 @@ func (dir *RegistryDirectory) cacheInvokerByEvent(event *registry.ServiceEvent)
 		switch event.Action {
 		case remoting.EventTypeAdd, remoting.EventTypeUpdate:
 			logger.Infof("selector add service url{%s}", event.Service)
-			// FIXME: routers are built in every address notification?
-			dir.configRouters()
+			if u != nil && constant.ROUTER_PROTOCOL == u.Protocol {
+				dir.configRouters()
+			}
 			return dir.cacheInvoker(u), nil
 		case remoting.EventTypeDel:
 			logger.Infof("selector delete service url{%s}", event.Service)

registry/protocol/protocol.go

@@ -48,7 +48,7 @@ var (
 	reserveParams = []string{
 		"application", "codec", "exchanger", "serialization", "cluster", "connections", "deprecated", "group",
 		"loadbalance", "mock", "path", "timeout", "token", "version", "warmup", "weight", "timestamp", "dubbo",
-		"release", "interface",
+		"release", "interface", "registry.role",
 	}
 )
 

test/integrate/dubbo/go-client/Dockerfile

@@ -33,6 +33,6 @@ RUN test ${PR_ORIGIN_REPO} && go mod edit  -replace=github.com/apache/dubbo-go=g
 
 ENV GO111MODULE on
 
-RUN go install github.com/apache/dubbo-go/test/integrate/dubbo/go-client
+RUN go mod tidy && go install github.com/apache/dubbo-go/test/integrate/dubbo/go-client
 
 CMD go-client

test/integrate/dubbo/go-client/go.mod

@@ -1,8 +1,3 @@
 module github.com/apache/dubbo-go/test/integrate/dubbo/go-client
 
-require (
-	github.com/apache/dubbo-go v1.5.6-rc1.0.20210220143153-9c8fc77f0381
-	github.com/apache/dubbo-go-hessian2 v1.8.2
-)
-
 go 1.13