Merge pull request #291 from Patrick0308/generic-service

Ftr: Generic Implement

common/constant/default.go

@@ -46,7 +46,7 @@ const (
 const (
 	DEFAULT_KEY               = "default"
 	PREFIX_DEFAULT_KEY        = "default."
-	DEFAULT_SERVICE_FILTERS   = "echo,token,accesslog,tps,execute,pshutdown"
+	DEFAULT_SERVICE_FILTERS   = "echo,token,accesslog,tps,generic_service,execute,pshutdown"
 	DEFAULT_REFERENCE_FILTERS = "cshutdown"
 	GENERIC_REFERENCE_FILTERS = "generic"
 	GENERIC                   = "$invoke"

common/extension/filter.go

@@ -19,12 +19,11 @@ package extension
 
 import (
 	"github.com/apache/dubbo-go/filter"
-	"github.com/apache/dubbo-go/filter/common"
 )
 
 var (
 	filters                  = make(map[string]func() filter.Filter)
-	rejectedExecutionHandler = make(map[string]func() common.RejectedExecutionHandler)
+	rejectedExecutionHandler = make(map[string]func() filter.RejectedExecutionHandler)
 )
 
 func SetFilter(name string, v func() filter.Filter) {
@@ -38,11 +37,11 @@ func GetFilter(name string) filter.Filter {
 	return filters[name]()
 }
 
-func SetRejectedExecutionHandler(name string, creator func() common.RejectedExecutionHandler) {
+func SetRejectedExecutionHandler(name string, creator func() filter.RejectedExecutionHandler) {
 	rejectedExecutionHandler[name] = creator
 }
 
-func GetRejectedExecutionHandler(name string) common.RejectedExecutionHandler {
+func GetRejectedExecutionHandler(name string) filter.RejectedExecutionHandler {
 	creator, ok := rejectedExecutionHandler[name]
 	if !ok {
 		panic("RejectedExecutionHandler for " + name + " is not existing, make sure you have import the package " +

common/extension/tps_limit.go

@@ -18,19 +18,19 @@
 package extension
 
 import (
-	"github.com/apache/dubbo-go/filter/impl/tps"
+	"github.com/apache/dubbo-go/filter"
 )
 
 var (
-	tpsLimitStrategy = make(map[string]tps.TpsLimitStrategyCreator)
-	tpsLimiter       = make(map[string]func() tps.TpsLimiter)
+	tpsLimitStrategy = make(map[string]filter.TpsLimitStrategyCreator)
+	tpsLimiter       = make(map[string]func() filter.TpsLimiter)
 )
 
-func SetTpsLimiter(name string, creator func() tps.TpsLimiter) {
+func SetTpsLimiter(name string, creator func() filter.TpsLimiter) {
 	tpsLimiter[name] = creator
 }
 
-func GetTpsLimiter(name string) tps.TpsLimiter {
+func GetTpsLimiter(name string) filter.TpsLimiter {
 	creator, ok := tpsLimiter[name]
 	if !ok {
 		panic("TpsLimiter for " + name + " is not existing, make sure you have import the package " +
@@ -39,11 +39,11 @@ func GetTpsLimiter(name string) tps.TpsLimiter {
 	return creator()
 }
 
-func SetTpsLimitStrategy(name string, creator tps.TpsLimitStrategyCreator) {
+func SetTpsLimitStrategy(name string, creator filter.TpsLimitStrategyCreator) {
 	tpsLimitStrategy[name] = creator
 }
 
-func GetTpsLimitStrategyCreator(name string) tps.TpsLimitStrategyCreator {
+func GetTpsLimitStrategyCreator(name string) filter.TpsLimitStrategyCreator {
 	creator, ok := tpsLimitStrategy[name]
 	if !ok {
 		panic("TpsLimitStrategy for " + name + " is not existing, make sure you have import the package " +

filter/common/impl/rejected_execution_handler_only_log.go → filter/common/rejected_execution_handler_only_log.go

@@ -15,9 +15,10 @@
  * limitations under the License.
  */
 
-package impl
+package common
 
 import (
+	"github.com/apache/dubbo-go/filter"
 	"sync"
 )
 
@@ -26,7 +27,6 @@ import (
 	"github.com/apache/dubbo-go/common/constant"
 	"github.com/apache/dubbo-go/common/extension"
 	"github.com/apache/dubbo-go/common/logger"
-	filterCommon "github.com/apache/dubbo-go/filter/common"
 	"github.com/apache/dubbo-go/protocol"
 )
 
@@ -61,7 +61,7 @@ func (handler *OnlyLogRejectedExecutionHandler) RejectedExecution(url common.URL
 	return &protocol.RPCResult{}
 }
 
-func GetOnlyLogRejectedExecutionHandler() filterCommon.RejectedExecutionHandler {
+func GetOnlyLogRejectedExecutionHandler() filter.RejectedExecutionHandler {
 	onlyLogHandlerOnce.Do(func() {
 		onlyLogHandlerInstance = &OnlyLogRejectedExecutionHandler{}
 	})

filter/common/impl/rejected_execution_handler_only_log_test.go → filter/common/rejected_execution_handler_only_log_test.go

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package impl
+package common
 
 import (
 	"net/url"

filter/impl/access_log_filter.go → filter/filter/access_log_filter.go

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package impl
+package filter
 
 import (
 	"os"

filter/impl/access_log_filter_test.go → filter/filter/access_log_filter_test.go

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package impl
+package filter
 
 import (
 	"context"

filter/impl/active_filter.go → filter/filter/active_filter.go

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package impl
+package filter
 
 import (
 	"github.com/apache/dubbo-go/common/extension"

filter/impl/echo_filter.go → filter/filter/echo_filter.go

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package impl
+package filter
 
 import (
 	"github.com/apache/dubbo-go/common/constant"

filter/impl/echo_filter_test.go → filter/filter/echo_filter_test.go

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package impl
+package filter
 
 import (
 	"testing"

filter/impl/execute_limit_filter.go → filter/filter/execute_limit_filter.go

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package impl
+package filter
 
 import (
 	"strconv"
@@ -32,7 +32,7 @@ import (
 	"github.com/apache/dubbo-go/common/extension"
 	"github.com/apache/dubbo-go/common/logger"
 	"github.com/apache/dubbo-go/filter"
-	_ "github.com/apache/dubbo-go/filter/common/impl"
+	_ "github.com/apache/dubbo-go/filter/common"
 	"github.com/apache/dubbo-go/protocol"
 )
 

filter/impl/execute_limit_filter_test.go → filter/filter/execute_limit_filter_test.go

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package impl
+package filter
 
 import (
 	"net/url"

filter/impl/generic_filter.go → filter/filter/generic_filter.go

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package impl
+package filter
 
 import (
 	"reflect"

filter/impl/generic_filter_test.go → filter/filter/generic_filter_test.go

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package impl
+package filter
 
 import (
 	"reflect"

filter/filter/generic_service_filter.go

@@ -0,0 +1,126 @@
+/*
+ * 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 filter
+
+import (
+	"reflect"
+	"strings"
+)
+
+import (
+	hessian "github.com/apache/dubbo-go-hessian2"
+	"github.com/mitchellh/mapstructure"
+	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/logger"
+	"github.com/apache/dubbo-go/filter"
+	"github.com/apache/dubbo-go/protocol"
+	invocation2 "github.com/apache/dubbo-go/protocol/invocation"
+)
+
+const (
+	GENERIC_SERVICE               = "generic_service"
+	GENERIC_SERIALIZATION_DEFAULT = "true"
+)
+
+func init() {
+	extension.SetFilter(GENERIC_SERVICE, GetGenericServiceFilter)
+}
+
+type GenericServiceFilter struct{}
+
+func (ef *GenericServiceFilter) Invoke(invoker protocol.Invoker, invocation protocol.Invocation) protocol.Result {
+	logger.Infof("invoking generic service filter.")
+	logger.Debugf("generic service filter methodName:%v,args:%v", invocation.MethodName(), len(invocation.Arguments()))
+
+	if invocation.MethodName() != constant.GENERIC || len(invocation.Arguments()) != 3 {
+		return invoker.Invoke(invocation)
+	}
+
+	var (
+		ok         bool
+		err        error
+		methodName string
+		newParams  []interface{}
+		genericKey string
+		argsType   []reflect.Type
+		oldParams  []hessian.Object
+	)
+
+	url := invoker.GetUrl()
+	methodName = invocation.Arguments()[0].(string)
+	// get service
+	svc := common.ServiceMap.GetService(url.Protocol, strings.TrimPrefix(url.Path, "/"))
+	// get method
+	method := svc.Method()[methodName]
+	if method == nil {
+		logger.Errorf("[Generic Service Filter] Don't have this method: %s", methodName)
+		return &protocol.RPCResult{}
+	}
+	argsType = method.ArgsType()
+	genericKey = invocation.AttachmentsByKey(constant.GENERIC_KEY, GENERIC_SERIALIZATION_DEFAULT)
+	if genericKey == GENERIC_SERIALIZATION_DEFAULT {
+		oldParams, ok = invocation.Arguments()[2].([]hessian.Object)
+	} else {
+		logger.Errorf("[Generic Service Filter] Don't support this generic: %s", genericKey)
+		return &protocol.RPCResult{}
+	}
+	if !ok {
+		logger.Errorf("[Generic Service Filter] wrong serialization")
+		return &protocol.RPCResult{}
+	}
+	if len(oldParams) != len(argsType) {
+		logger.Errorf("[Generic Service Filter] method:%s invocation arguments number was wrong", methodName)
+		return &protocol.RPCResult{}
+	}
+	// oldParams convert to newParams
+	newParams = make([]interface{}, len(oldParams))
+	for i := range argsType {
+		newParam := reflect.New(argsType[i]).Interface()
+		err = mapstructure.Decode(oldParams[i], newParam)
+		newParam = reflect.ValueOf(newParam).Elem().Interface()
+		if err != nil {
+			logger.Errorf("[Generic Service Filter] decode arguments map to struct wrong: error{%v}", perrors.WithStack(err))
+			return &protocol.RPCResult{}
+		}
+		newParams[i] = newParam
+	}
+	newInvocation := invocation2.NewRPCInvocation(methodName, newParams, invocation.Attachments())
+	newInvocation.SetReply(invocation.Reply())
+	return invoker.Invoke(newInvocation)
+}
+
+func (ef *GenericServiceFilter) OnResponse(result protocol.Result, invoker protocol.Invoker, invocation protocol.Invocation) protocol.Result {
+	if invocation.MethodName() == constant.GENERIC && len(invocation.Arguments()) == 3 && result.Result() != nil {
+		v := reflect.ValueOf(result.Result())
+		if v.Kind() == reflect.Ptr {
+			v = v.Elem()
+		}
+		result.SetResult(struct2MapAll(v.Interface()))
+	}
+	return result
+}
+
+func GetGenericServiceFilter() filter.Filter {
+	return &GenericServiceFilter{}
+}