Ftr: add pass through proxy factory

common/constant/key.go

@@ -50,7 +50,8 @@ const (
 	PROTOCOL_KEY             = "protocol"
 	PATH_SEPARATOR           = "/"
 	//DUBBO_KEY                = "dubbo"
-	SSL_ENABLED_KEY = "ssl-enabled"
+	SSL_ENABLED_KEY  = "ssl-enabled"
+	ParameterTypeKey = "parameter-type-names" // ParameterType key used in dapr, to tranfer rsp param type
 )
 
 const (
@@ -199,6 +200,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,124 @@
+/*
+ * 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 = [][]byte{}
+	} else {
+		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 := []reflect.Value{srv.Rcvr()}
+	in = append(in, reflect.ValueOf(invocation.MethodName()))
+	in = append(in, reflect.ValueOf(invocation.Attachment(constant.ParameterTypeKey)))
+	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))
+	} else {
+		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 @@
+package proxy_factory
+
+/*
+ * 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.
+ */
+
+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())
+}

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",
 	}
 )