Rft: refactoring code

cluster/cluster.go

@@ -5,4 +5,3 @@ import "github.com/dubbo/dubbo-go/protocol"
 type Cluster interface {
 	Join(Directory) protocol.Invoker
 }
-

common/constant/key.go

@@ -27,3 +27,7 @@ const (
 	WARMUP_KEY      = "warmup"
 	RETRIES_KEY     = "retries"
 )
+
+const (
+	DUBBOGO_CTX_KEY = "dubbogo-ctx"
+)

common/proxy/proxy.go

@@ -0,0 +1,108 @@
+package proxy
+
+import (
+	"fmt"
+	"reflect"
+)
+import (
+	log "github.com/AlexStocks/log4go"
+)
+
+import (
+	"github.com/dubbo/dubbo-go/protocol"
+)
+
+// Proxy struct
+type Proxy struct {
+	v           interface{}
+	invoke      protocol.Invoker
+	callBack    interface{}
+	attachments map[string]string
+}
+
+var typError = reflect.Zero(reflect.TypeOf((*error)(nil)).Elem()).Type()
+
+func NewProxy(invoke protocol.Invoker, callBack interface{}, attachments map[string]string) *Proxy {
+	return &Proxy{
+		invoke:      invoke,
+		callBack:    callBack,
+		attachments: attachments,
+	}
+}
+
+// proxy implement
+func (p *Proxy) Implement(v interface{}) error {
+
+	// check parameters, incoming interface must be a elem's pointer.
+	valueOf := reflect.ValueOf(v)
+	log.Debug("[Implement] reflect.TypeOf: %s", valueOf.String())
+	if valueOf.Kind() != reflect.Ptr {
+		return fmt.Errorf("%s must be a pointer", valueOf)
+	}
+
+	valueOfElem := valueOf.Elem()
+	typeOf := valueOfElem.Type()
+
+	// check incoming interface, incoming interface's elem must be a struct.
+	if typeOf.Kind() != reflect.Struct {
+		return fmt.Errorf("%s must be a struct ptr", valueOf.String())
+	}
+
+	makeDubboCallProxy := func(methodName string, outs []reflect.Type) func(in []reflect.Value) []reflect.Value {
+		return func(in []reflect.Value) []reflect.Value {
+			// Convert input parameters to interface.
+			var argsInterface = make([]interface{}, len(in))
+			for k, v := range in {
+				argsInterface[k] = v.Interface()
+			}
+
+			//todo:call
+			inv := protocol.NewRPCInvocationForConsumer(methodName, nil, argsInterface, in[2].Interface(), p.callBack, p.attachments, nil)
+			result := p.invoke.Invoke(inv)
+			var err error
+			err = result.Error()
+			return []reflect.Value{reflect.ValueOf(&err).Elem()}
+		}
+	}
+
+	numField := valueOfElem.NumField()
+	for i := 0; i < numField; i++ {
+		t := typeOf.Field(i)
+		f := valueOfElem.Field(i)
+		if f.Kind() == reflect.Func && f.IsValid() && f.CanSet() {
+
+			if t.Type.NumIn() != 3 && t.Type.NumIn() != 4 {
+				log.Error("method %s of mtype %v has wrong number of in parameters %d; needs exactly 3/4",
+					t.Name, t.Type.String(), t.Type.NumIn())
+				return fmt.Errorf("method %s of mtype %v has wrong number of in parameters %d; needs exactly 3/4",
+					t.Name, t.Type.String(), t.Type.NumIn())
+			}
+
+			// Method needs one out.
+			if t.Type.NumOut() != 1 {
+				log.Error("method %q has %d out parameters; needs exactly 1", t.Name, t.Type.NumOut())
+				return fmt.Errorf("method %q has %d out parameters; needs exactly 1", t.Name, t.Type.NumOut())
+			}
+			// The return type of the method must be error.
+			if returnType := t.Type.Out(0); returnType != typError {
+				log.Error("return type %s of method %q is not error", returnType, t.Name)
+				return fmt.Errorf("return type %s of method %q is not error", returnType, t.Name)
+			}
+
+			var funcOuts = make([]reflect.Type, t.Type.NumOut())
+			funcOuts[i] = t.Type.Out(0)
+
+			// do method proxy here:
+			f.Set(reflect.MakeFunc(f.Type(), makeDubboCallProxy(t.Name, funcOuts)))
+			log.Debug("set method [%s]", t.Name)
+		}
+	}
+
+	p.v = v
+
+	return nil
+}
+
+func (p *Proxy) Get() interface{} {
+	return p.v
+}

protocol/dubbo/dubbo_invoker.go

@@ -4,7 +4,6 @@ import (
 	"errors"
 	"strconv"
 	"sync"
-	"time"
 )
 
 import (
@@ -49,27 +48,15 @@ func (di *DubboInvoker) Invoke(invocation protocol.Invocation) protocol.Result {
 	}
 	if async {
 		if callBack, ok := inv.CallBack().(func(response CallResponse)); ok {
-			result.Err = di.client.AsyncCall(url.Location, *url, inv.MethodName(), inv.Arguments(), callBack, inv.Reply(),
-				WithCallRequestTimeout(inv.Params()["requestTimeout"].(time.Duration)),
-				WithCallResponseTimeout(inv.Params()["responseTimeout"].(time.Duration)),
-				WithCallSerialID(inv.Params()["serialID"].(SerialID)),
-				WithCallMeta_All(inv.Params()["callMeta"].(map[interface{}]interface{})))
+			result.Err = di.client.AsyncCall(url.Location, *url, inv.MethodName(), inv.Arguments(), callBack, inv.Reply())
 		} else {
-			result.Err = di.client.CallOneway(url.Location, *url, inv.MethodName(), inv.Arguments(),
-				WithCallRequestTimeout(inv.Params()["requestTimeout"].(time.Duration)),
-				WithCallResponseTimeout(inv.Params()["responseTimeout"].(time.Duration)),
-				WithCallSerialID(inv.Params()["serialID"].(SerialID)),
-				WithCallMeta_All(inv.Params()["callMeta"].(map[interface{}]interface{})))
+			result.Err = di.client.CallOneway(url.Location, *url, inv.MethodName(), inv.Arguments())
 		}
 	} else {
 		if inv.Reply() == nil {
 			result.Err = Err_No_Reply
 		} else {
-			result.Err = di.client.Call(url.Location, *url, inv.MethodName(), inv.Arguments(), inv.Reply(),
-				WithCallRequestTimeout(inv.Params()["requestTimeout"].(time.Duration)),
-				WithCallResponseTimeout(inv.Params()["responseTimeout"].(time.Duration)),
-				WithCallSerialID(inv.Params()["serialID"].(SerialID)),
-				WithCallMeta_All(inv.Params()["callMeta"].(map[interface{}]interface{})))
+			result.Err = di.client.Call(url.Location, *url, inv.MethodName(), inv.Arguments(), inv.Reply())
 			result.Rest = inv.Reply() // reply should be set to result.Rest when sync
 		}
 	}

protocol/invocation.go

@@ -26,12 +26,11 @@ type RPCInvocation struct {
 	callBack       interface{}
 	attachments    map[string]string
 	invoker        Invoker
-	params         map[string]interface{} // Store some parameters that are not easy to refine
 }
 
 // todo: arguments table is too many
 func NewRPCInvocationForConsumer(methodName string, parameterTypes []reflect.Type, arguments []interface{},
-	reply interface{}, callBack interface{}, attachments map[string]string, invoker Invoker, params map[string]interface{}) *RPCInvocation {
+	reply interface{}, callBack interface{}, attachments map[string]string, invoker Invoker) *RPCInvocation {
 	return &RPCInvocation{
 		methodName:     methodName,
 		parameterTypes: parameterTypes,
@@ -40,7 +39,6 @@ func NewRPCInvocationForConsumer(methodName string, parameterTypes []reflect.Typ
 		callBack:       callBack,
 		attachments:    attachments,
 		invoker:        invoker,
-		params:         params,
 	}
 }
 
@@ -71,6 +69,9 @@ func (r *RPCInvocation) Attachments() map[string]string {
 }
 
 func (r *RPCInvocation) AttachmentsByKey(key string, defaultValue string) string {
+	if r.attachments == nil {
+		return defaultValue
+	}
 	value, ok := r.attachments[key]
 	if ok {
 		return value
@@ -81,14 +82,12 @@ func (r *RPCInvocation) AttachmentsByKey(key string, defaultValue string) string
 func (r *RPCInvocation) Invoker() Invoker {
 	return r.invoker
 }
+
+// SetInvoker is called while getting url, maybe clusterInvoker?
 func (r *RPCInvocation) SetInvoker() Invoker {
 	return r.invoker
 }
 
-func (r *RPCInvocation) Params() map[string]interface{} {
-	return r.params
-}
-
 func (r *RPCInvocation) CallBack() interface{} {
 	return r.callBack
 }

protocol/jsonrpc/http.go

@@ -20,8 +20,8 @@ import (
 )
 
 import (
+	"github.com/dubbo/dubbo-go/common/constant"
 	"github.com/dubbo/dubbo-go/config"
-	"github.com/dubbo/dubbo-go/public"
 )
 
 //////////////////////////////////////////////
@@ -104,7 +104,7 @@ func (c *HTTPClient) Call(ctx context.Context, service config.URL, req *Request,
 		reqTimeout = 1e8
 	}
 	httpHeader.Set("Timeout", reqTimeout.String())
-	if md, ok := ctx.Value(public.DUBBOGO_CTX_KEY).(map[string]string); ok {
+	if md, ok := ctx.Value(constant.DUBBOGO_CTX_KEY).(map[string]string); ok {
 		for k := range md {
 			httpHeader.Set(k, md[k])
 		}

protocol/jsonrpc/jsonrpc_invoker.go

@@ -10,9 +10,9 @@ import (
 )
 
 import (
+	"github.com/dubbo/dubbo-go/common/constant"
 	"github.com/dubbo/dubbo-go/config"
 	"github.com/dubbo/dubbo-go/protocol"
-	"github.com/dubbo/dubbo-go/public"
 )
 
 type JsonrpcInvoker struct {
@@ -37,7 +37,7 @@ func (ji *JsonrpcInvoker) Invoke(invocation protocol.Invocation) protocol.Result
 	url := inv.Invoker().GetUrl().(*config.URL)
 
 	req := ji.client.NewRequest(*url, inv.MethodName(), inv.Arguments())
-	ctx := context.WithValue(context.Background(), public.DUBBOGO_CTX_KEY, map[string]string{
+	ctx := context.WithValue(context.Background(), constant.DUBBOGO_CTX_KEY, map[string]string{
 		"X-Proxy-Id": "dubbogo",
 		"X-Services": url.Service,
 		"X-Method":   inv.MethodName(),

protocol/jsonrpc/jsonrpc_protocol.go

@@ -7,6 +7,7 @@ import (
 import (
 	"github.com/dubbo/dubbo-go/common/extension"
 	"github.com/dubbo/dubbo-go/config"
+	"github.com/dubbo/dubbo-go/config/support"
 	"github.com/dubbo/dubbo-go/protocol"
 )
 
@@ -45,8 +46,8 @@ func (jp *JsonrpcProtocol) Export(invoker protocol.Invoker) protocol.Exporter {
 
 func (jp *JsonrpcProtocol) Refer(url config.IURL) protocol.Invoker {
 	invoker := NewJsonrpcInvoker(url, NewHTTPClient(&HTTPOptions{
-		HandshakeTimeout: config.GetConsumerConfig().ConnectTimeout,
-		HTTPTimeout:      config.GetConsumerConfig().RequestTimeout,
+		HandshakeTimeout: support.GetConsumerConfig().ConnectTimeout,
+		HTTPTimeout:      support.GetConsumerConfig().RequestTimeout,
 	}))
 	jp.SetInvokers(invoker)
 	log.Info("Refer service: ", url.(*config.URL).String())