modify async callback

common/rpc_service.go

@@ -22,7 +22,6 @@ import (
 	"reflect"
 	"strings"
 	"sync"
-	"time"
 	"unicode"
 	"unicode/utf8"
 )
@@ -40,25 +39,17 @@ type RPCService interface {
 	Reference() string // rpc service id or reference id
 }
 
-// callback interface for async
+//AsyncCallbackService callback interface for async
 type AsyncCallbackService interface {
-	CallBack(response CallResponse) // callback
+	CallBack(response CallbackResponse) // callback
 }
 
-type Options struct {
-	// connect timeout
-	ConnectTimeout time.Duration
-	// request timeout
-	RequestTimeout time.Duration
+//CallbackResponse for different protocol
+type CallbackResponse interface {
 }
 
-type CallResponse struct {
-	Opts      Options
-	Cause     error
-	Start     time.Time // invoke(call) start time == write start time
-	ReadStart time.Time // read start time, write duration = ReadStart - Start
-	Reply     interface{}
-}
+//AsyncCallback async callback method
+type AsyncCallback func(response CallbackResponse)
 
 // for lowercase func
 // func MethodMapper() map[string][string] {

config/service.go

@@ -44,7 +44,7 @@ func GetProviderService(name string) common.RPCService {
 	return proServices[name]
 }
 
-func GetCallback(name string) func(common.CallResponse) {
+func GetCallback(name string) func(response common.CallbackResponse) {
 	service := GetConsumerService(name)
 	if sv, ok := service.(common.AsyncCallbackService); ok {
 		return sv.CallBack

protocol/dubbo/client.go

@@ -107,19 +107,22 @@ func setClientGrpool() {
 }
 
 type Options struct {
-	common.Options
 	// connect timeout
 	ConnectTimeout time.Duration
 	// request timeout
 	RequestTimeout time.Duration
 }
 
-type CallResponse struct {
-	common.CallResponse
+//AsyncCallbackResponse async response for dubbo
+type AsyncCallbackResponse struct {
+	common.CallbackResponse
+	Opts      Options
+	Cause     error
+	Start     time.Time // invoke(call) start time == write start time
+	ReadStart time.Time // read start time, write duration = ReadStart - Start
+	Reply     interface{}
 }
 
-type AsyncCallback func(response common.CallResponse)
-
 type Client struct {
 	opts     Options
 	conf     ClientConfig
@@ -196,12 +199,12 @@ func (c *Client) Call(request *Request, response *Response) error {
 	return perrors.WithStack(c.call(ct, request, response, nil))
 }
 
-func (c *Client) AsyncCall(request *Request, callback AsyncCallback, response *Response) error {
+func (c *Client) AsyncCall(request *Request, callback common.AsyncCallback, response *Response) error {
 
 	return perrors.WithStack(c.call(CT_TwoWay, request, response, callback))
 }
 
-func (c *Client) call(ct CallType, request *Request, response *Response, callback AsyncCallback) error {
+func (c *Client) call(ct CallType, request *Request, response *Response, callback common.AsyncCallback) error {
 
 	p := &DubboPackage{}
 	p.Service.Path = strings.TrimPrefix(request.svcUrl.Path, "/")

protocol/dubbo/client_test.go

@@ -144,8 +144,9 @@ func TestClient_AsyncCall(t *testing.T) {
 	user := &User{}
 	lock := sync.Mutex{}
 	lock.Lock()
-	err := c.AsyncCall(NewRequest("127.0.0.1:20000", url, "GetUser", []interface{}{"1", "username"}, nil), func(response common.CallResponse) {
-		assert.Equal(t, User{Id: "1", Name: "username"}, *response.Reply.(*Response).reply.(*User))
+	err := c.AsyncCall(NewRequest("127.0.0.1:20000", url, "GetUser", []interface{}{"1", "username"}, nil), func(response common.CallbackResponse) {
+		r := response.(AsyncCallbackResponse)
+		assert.Equal(t, User{Id: "1", Name: "username"}, *r.Reply.(*Response).reply.(*User))
 		lock.Unlock()
 	}, NewResponse(user, nil))
 	assert.NoError(t, err)

protocol/dubbo/codec.go

@@ -110,7 +110,7 @@ type PendingResponse struct {
 	err       error
 	start     time.Time
 	readStart time.Time
-	callback  AsyncCallback
+	callback  common.AsyncCallback
 	response  *Response
 	done      chan struct{}
 }
@@ -123,8 +123,8 @@ func NewPendingResponse() *PendingResponse {
 	}
 }
 
-func (r PendingResponse) GetCallResponse() common.CallResponse {
-	return common.CallResponse{
+func (r PendingResponse) GetCallResponse() common.CallbackResponse {
+	return AsyncCallbackResponse{
 		Cause:     r.err,
 		Start:     r.start,
 		ReadStart: r.readStart,

protocol/dubbo/dubbo_invoker.go

@@ -75,7 +75,7 @@ func (di *DubboInvoker) Invoke(invocation protocol.Invocation) protocol.Result {
 	}
 	response := NewResponse(inv.Reply(), nil)
 	if async {
-		if callBack, ok := inv.CallBack().(func(response common.CallResponse)); ok {
+		if callBack, ok := inv.CallBack().(func(response common.CallbackResponse)); ok {
 			result.Err = di.client.AsyncCall(NewRequest(url.Location, url, inv.MethodName(), inv.Arguments(), inv.Attachments()), callBack, response)
 		} else {
 			result.Err = di.client.CallOneway(NewRequest(url.Location, url, inv.MethodName(), inv.Arguments(), inv.Attachments()))

protocol/dubbo/dubbo_invoker_test.go

@@ -67,8 +67,9 @@ func TestDubboInvoker_Invoke(t *testing.T) {
 	// AsyncCall
 	lock := sync.Mutex{}
 	lock.Lock()
-	inv.SetCallBack(func(response common.CallResponse) {
-		assert.Equal(t, User{Id: "1", Name: "username"}, *response.Reply.(*Response).reply.(*User))
+	inv.SetCallBack(func(response common.CallbackResponse) {
+		r := response.(AsyncCallbackResponse)
+		assert.Equal(t, User{Id: "1", Name: "username"}, *r.Reply.(*Response).reply.(*User))
 		lock.Unlock()
 	})
 	res = invoker.Invoke(inv)