From 966d27c8e63fa8920b84300c3e589aadef01beb7 Mon Sep 17 00:00:00 2001
From: cvictory <shenglicao2@gmail.com>
Date: Sun, 26 Apr 2020 14:43:38 +0800
Subject: [PATCH] refactor getty
---
protocol/dubbo/dubbo_codec.go | 159 +++++++++----------------
protocol/dubbo/dubbo_protocol.go | 36 +++++-
protocol/dubbo/dubbo_protocol_test.go | 162 +++++++++++++-------------
remoting/codec.go | 12 +-
remoting/exchange.go | 32 ++---
remoting/exchange_client.go | 65 ++++++++---
remoting/exchange_server.go | 17 +--
remoting/getty/getty_client.go | 29 +++--
remoting/getty/listener.go | 13 ++-
remoting/getty/readwriter.go | 23 ++--
remoting/getty/server.go | 6 +-
11 files changed, 273 insertions(+), 281 deletions(-)
diff --git a/protocol/dubbo/dubbo_codec.go b/protocol/dubbo/dubbo_codec.go
index 5dea37f87..899148a8e 100644
--- a/protocol/dubbo/dubbo_codec.go
+++ b/protocol/dubbo/dubbo_codec.go
@@ -15,77 +15,6 @@ import (
"time"
)
-////SerialID serial ID
-//type SerialID byte
-//type SequenceType int64
-//
-//const (
-// // S_Dubbo dubbo serial id
-// S_Dubbo SerialID = 2
-//)
-//
-//// DubboPackage ...
-//type DubboPackage struct {
-// Header hessian.DubboHeader
-// Service hessian.Service
-// Body interface{}
-// Err error
-//}
-//
-//func (p DubboPackage) String() string {
-// return fmt.Sprintf("DubboPackage: Header-%v, Path-%v, Body-%v", p.Header, p.Service, p.Body)
-//}
-
-//
-//// Marshal ...
-//func (p *DubboPackage) Marshal() (*bytes.Buffer, error) {
-// codec := hessian.NewHessianCodec(nil)
-//
-// pkg, err := codec.Write(p.Service, p.Header, p.Body)
-// if err != nil {
-// return nil, perrors.WithStack(err)
-// }
-//
-// return bytes.NewBuffer(pkg), nil
-//}
-//
-// Unmarshal ...
-//func (p *DubboPackage) Unmarshal(buf *bytes.Buffer, resp *remoting.Response) error {
-// // fix issue https://github.com/apache/dubbo-go/issues/380
-// bufLen := buf.Len()
-// if bufLen < hessian.HEADER_LENGTH {
-// return perrors.WithStack(hessian.ErrHeaderNotEnough)
-// }
-//
-// codec := hessian.NewHessianCodec(bufio.NewReaderSize(buf, bufLen))
-//
-// // read header
-// err := codec.ReadHeader(&p.Header)
-// if err != nil {
-// return perrors.WithStack(err)
-// }
-//
-// if resp != nil { // for client
-// if p.Header.Type&hessian.PackageRequest != 0x00 {
-// // size of this array must be '7'
-// // https://github.com/apache/dubbo-go-hessian2/blob/master/request.go#L272
-// p.Body = make([]interface{}, 7)
-// } else {
-// //pendingRsp, ok := client.pendingResponses.Load(SequenceType(p.Header.ID))
-// //if !ok {
-// // return perrors.Errorf("client.GetPendingResponse(%v) = nil", p.Header.ID)
-// //}
-// p.Body = &hessian.Response{RspObj: resp.Reply}
-// }
-// }
-//
-// // read body
-// err = codec.ReadBody(p.Body)
-// return perrors.WithStack(err)
-//}
-
-/////////////////////////////////////////
-/////////////////////////////////////////
//SerialID serial ID
type SerialID byte
@@ -94,24 +23,10 @@ const (
S_Dubbo SerialID = 2
)
-//CallType call type
-type CallType int32
-
-const (
- // CT_UNKNOWN unknown call type
- CT_UNKNOWN CallType = 0
- // CT_OneWay call one way
- CT_OneWay CallType = 1
- // CT_TwoWay call in request/response
- CT_TwoWay CallType = 2
-)
-
-////////////////////////////////////////////
-// dubbo package
-////////////////////////////////////////////
-
-// SequenceType ...
-type SequenceType int64
+func init() {
+ codec := &DubboCodec{}
+ remoting.NewCodec("dubbo", codec)
+}
// DubboPackage ...
type DubboPackage struct {
@@ -138,7 +53,7 @@ func (p *DubboPackage) Marshal() (*bytes.Buffer, error) {
}
// Unmarshal ...
-func (p *DubboPackage) Unmarshal(buf *bytes.Buffer, opts ...interface{}) error {
+func (p *DubboPackage) Unmarshal(buf *bytes.Buffer, resp *remoting.Response) error {
// fix issue https://github.com/apache/dubbo-go/issues/380
bufLen := buf.Len()
if bufLen < hessian.HEADER_LENGTH {
@@ -153,22 +68,17 @@ func (p *DubboPackage) Unmarshal(buf *bytes.Buffer, opts ...interface{}) error {
return perrors.WithStack(err)
}
- if len(opts) != 0 { // for client
- client, ok := opts[0].(*Client)
- if !ok {
- return perrors.Errorf("opts[0] is not of type *Client")
- }
-
+ if resp != nil { // for client
if p.Header.Type&hessian.PackageRequest != 0x00 {
// size of this array must be '7'
// https://github.com/apache/dubbo-go-hessian2/blob/master/request.go#L272
p.Body = make([]interface{}, 7)
} else {
- pendingRsp, ok := client.pendingResponses.Load(SequenceType(p.Header.ID))
- if !ok {
+ pendingRsp := remoting.GetPendingResponse(remoting.SequenceType(p.Header.ID))
+ if pendingRsp == nil {
return perrors.Errorf("client.GetPendingResponse(%v) = nil", p.Header.ID)
}
- p.Body = &hessian.Response{RspObj: pendingRsp.(*PendingResponse).response.reply}
+ p.Body = &hessian.Response{RspObj: pendingRsp.Reply}
}
}
@@ -176,11 +86,21 @@ func (p *DubboPackage) Unmarshal(buf *bytes.Buffer, opts ...interface{}) error {
err = codec.ReadBody(p.Body)
return perrors.WithStack(err)
}
+
type DubboCodec struct {
}
func (c *DubboCodec) EncodeRequest(request *remoting.Request) (*bytes.Buffer, error) {
- invocation := request.Data.(invocation.RPCInvocation)
+ if request.Event {
+ return c.encodeHeartbeartReqeust(request)
+ }
+
+ invoc, ok := request.Data.(*protocol.Invocation)
+ if !ok {
+ logger.Errorf("encode request failed for parameter type :%+v", request)
+ return nil, perrors.Errorf("encode request failed for parameter type :%+v", request)
+ }
+ invocation := *invoc
p := &DubboPackage{}
p.Service.Path = invocation.AttachmentsByKey(constant.PATH_KEY, "")
@@ -220,8 +140,24 @@ func (c *DubboCodec) EncodeRequest(request *remoting.Request) (*bytes.Buffer, er
return bytes.NewBuffer(pkg), nil
}
+func (c *DubboCodec) encodeHeartbeartReqeust(request *remoting.Request) (*bytes.Buffer, error) {
+ pkg := &DubboPackage{}
+ pkg.Body = []interface{}{}
+ pkg.Header.ID = request.Id
+ pkg.Header.Type = hessian.PackageHeartbeat
+ pkg.Header.SerialID = byte(S_Dubbo)
+
+ codec := hessian.NewHessianCodec(nil)
+
+ byt, err := codec.Write(pkg.Service, pkg.Header, pkg.Body)
+ if err != nil {
+ return nil, perrors.WithStack(err)
+ }
+
+ return bytes.NewBuffer(byt), nil
+}
func (c *DubboCodec) EncodeResponse(response *remoting.Response) (*bytes.Buffer, error) {
- var ptype hessian.PackageType = hessian.PackageResponse
+ var ptype = hessian.PackageResponse
if response.IsHeartbeat() {
ptype = hessian.PackageHeartbeat
}
@@ -233,7 +169,13 @@ func (c *DubboCodec) EncodeResponse(response *remoting.Response) (*bytes.Buffer,
ResponseStatus: response.Status,
},
}
- resp.Body = response.Result
+ if !response.IsHeartbeat() {
+ resp.Body = &hessian.Response{
+ RspObj: response.Result.(protocol.RPCResult).Rest,
+ Exception: response.Result.(protocol.RPCResult).Err,
+ }
+ }
+
//if response.Header.Type&hessian.PackageRequest != 0x00 {
// resp.Body = req.Body
//} else {
@@ -259,7 +201,7 @@ func (c *DubboCodec) DecodeRequest(data []byte) (*remoting.Request, int, error)
originErr := perrors.Cause(err)
if originErr == hessian.ErrHeaderNotEnough || originErr == hessian.ErrBodyNotEnough {
//FIXME
- return request, 0, originErr
+ return request, 0, originErr
}
logger.Errorf("pkg.Unmarshal(len(@data):%d) = error:%+v", buf.Len(), err)
@@ -268,7 +210,8 @@ func (c *DubboCodec) DecodeRequest(data []byte) (*remoting.Request, int, error)
request = &remoting.Request{
Id: pkg.Header.ID,
SerialID: pkg.Header.SerialID,
- TwoWay: false,
+ TwoWay: pkg.Header.Type&hessian.PackageRequest_TwoWay != 0x00,
+ Event: pkg.Header.Type&hessian.PackageHeartbeat != 0x00,
}
if pkg.Header.Type&hessian.PackageHeartbeat == 0x00 {
// convert params of request
@@ -330,15 +273,16 @@ func (c *DubboCodec) DecodeRequest(data []byte) (*remoting.Request, int, error)
}
return request, hessian.HEADER_LENGTH + pkg.Header.BodyLen, nil
}
+
func (c *DubboCodec) DecodeResponse(data []byte) (*remoting.Response, int, error) {
pkg := &DubboPackage{}
buf := bytes.NewBuffer(data)
- var response *remoting.Response
+ response := &remoting.Response{}
err := pkg.Unmarshal(buf, response)
if err != nil {
originErr := perrors.Cause(err)
if originErr == hessian.ErrHeaderNotEnough || originErr == hessian.ErrBodyNotEnough {
- return response, 0, nil
+ return response, 0, nil
}
logger.Errorf("pkg.Unmarshal(len(@data):%d) = error:%+v", buf.Len(), err)
@@ -349,7 +293,7 @@ func (c *DubboCodec) DecodeResponse(data []byte) (*remoting.Response, int, error
//Version: pkg.Header.,
SerialID: pkg.Header.SerialID,
Status: pkg.Header.ResponseStatus,
- Event: (pkg.Header.Type | hessian.PackageHeartbeat) != 0,
+ Event: (pkg.Header.Type & hessian.PackageHeartbeat) != 0,
}
var error error
if pkg.Header.Type&hessian.PackageHeartbeat != 0x00 {
@@ -368,6 +312,7 @@ func (c *DubboCodec) DecodeResponse(data []byte) (*remoting.Response, int, error
}
logger.Debugf("get rpc response{header: %#v, body: %#v}", pkg.Header, pkg.Body)
rpcResult := &protocol.RPCResult{}
+ response.Result = rpcResult
if pkg.Header.Type&hessian.PackageRequest == 0x00 {
if pkg.Err != nil {
rpcResult.Err = pkg.Err
diff --git a/protocol/dubbo/dubbo_protocol.go b/protocol/dubbo/dubbo_protocol.go
index e98e6e2dd..20e54fa2e 100644
--- a/protocol/dubbo/dubbo_protocol.go
+++ b/protocol/dubbo/dubbo_protocol.go
@@ -39,6 +39,10 @@ const (
DUBBO = "dubbo"
)
+var (
+ exchangeClientMap *sync.Map = new(sync.Map)
+)
+
func init() {
extension.SetProtocol(DUBBO, GetProtocol)
}
@@ -89,9 +93,7 @@ func (dp *DubboProtocol) Refer(url common.URL) protocol.Invoker {
// ConnectTimeout: config.GetConsumerConfig().ConnectTimeout,
// RequestTimeout: requestTimeout,
//}))
- invoker := NewDubboInvoker(url, remoting.NewExchangeClient(url, getty.NewClient(getty.Options{
- ConnectTimeout: config.GetConsumerConfig().ConnectTimeout,
- }), config.GetConsumerConfig().ConnectTimeout))
+ invoker := NewDubboInvoker(url, getExchangeClient(url))
dp.SetInvokers(invoker)
logger.Infof("Refer service: %s", url.String())
return invoker
@@ -124,7 +126,7 @@ func (dp *DubboProtocol) openServer(url common.URL) {
handler := func(invocation *invocation.RPCInvocation) protocol.RPCResult {
return doHandleRequest(invocation)
}
- srv := remoting.NewExchangeServer(url, handler)
+ srv := remoting.NewExchangeServer(url, getty.NewServer(url, handler))
dp.serverMap[url.Location] = srv
srv.Start()
}
@@ -158,8 +160,8 @@ func doHandleRequest(rpcInvocation *invocation.RPCInvocation) protocol.RPCResult
//
//args := p.Body.(map[string]interface{})["args"].([]interface{})
//inv := invocation.NewRPCInvocation(p.Service.Method, args, attachments)
-
- ctx := rebuildCtx(rpcInvocation)
+ // FIXME
+ ctx := getty.RebuildCtx(rpcInvocation)
invokeResult := invoker.Invoke(ctx, rpcInvocation)
if err := invokeResult.Error(); err != nil {
@@ -176,3 +178,25 @@ func doHandleRequest(rpcInvocation *invocation.RPCInvocation) protocol.RPCResult
}
return result
}
+
+func getExchangeClient(url common.URL) *remoting.ExchangeClient {
+ clientTmp, ok := exchangeClientMap.Load(url.Location)
+ if !ok {
+ exchangeClientTmp := remoting.NewExchangeClient(url, getty.NewClient(getty.Options{
+ ConnectTimeout: config.GetConsumerConfig().ConnectTimeout,
+ }), config.GetConsumerConfig().ConnectTimeout)
+ exchangeClientMap.Store(url.Location, exchangeClientTmp)
+
+ return exchangeClientTmp
+ }
+ exchangeClient, ok := clientTmp.(*remoting.ExchangeClient)
+ if !ok {
+ exchangeClientTmp := remoting.NewExchangeClient(url, getty.NewClient(getty.Options{
+ ConnectTimeout: config.GetConsumerConfig().ConnectTimeout,
+ }), config.GetConsumerConfig().ConnectTimeout)
+ exchangeClientMap.Store(url.Location, exchangeClientTmp)
+
+ return exchangeClientTmp
+ }
+ return exchangeClient
+}
diff --git a/protocol/dubbo/dubbo_protocol_test.go b/protocol/dubbo/dubbo_protocol_test.go
index 14f6868ad..5fdcc69b5 100644
--- a/protocol/dubbo/dubbo_protocol_test.go
+++ b/protocol/dubbo/dubbo_protocol_test.go
@@ -16,84 +16,84 @@
*/
package dubbo
-
-import (
- "testing"
-)
-
-import (
- "github.com/stretchr/testify/assert"
-)
-
-import (
- "github.com/apache/dubbo-go/common"
- "github.com/apache/dubbo-go/common/constant"
- "github.com/apache/dubbo-go/protocol"
-)
-
-func TestDubboProtocol_Export(t *testing.T) {
- // Export
- proto := GetProtocol()
- srvConf = &ServerConfig{}
- url, err := common.NewURL("dubbo://127.0.0.1:20000/com.ikurento.user.UserProvider?anyhost=true&" +
- "application=BDTService&category=providers&default.timeout=10000&dubbo=dubbo-provider-golang-1.0.0&" +
- "environment=dev&interface=com.ikurento.user.UserProvider&ip=192.168.56.1&methods=GetUser%2C&" +
- "module=dubbogo+user-info+server&org=ikurento.com&owner=ZX&pid=1447&revision=0.0.1&" +
- "side=provider&timeout=3000×tamp=1556509797245")
- assert.NoError(t, err)
- exporter := proto.Export(protocol.NewBaseInvoker(url))
-
- // make sure url
- eq := exporter.GetInvoker().GetUrl().URLEqual(url)
- assert.True(t, eq)
-
- // second service: the same path and the different version
- url2, err := common.NewURL("dubbo://127.0.0.1:20000/com.ikurento.user.UserProvider?anyhost=true&"+
- "application=BDTService&category=providers&default.timeout=10000&dubbo=dubbo-provider-golang-1.0.0&"+
- "environment=dev&interface=com.ikurento.user.UserProvider&ip=192.168.56.1&methods=GetUser%2C&"+
- "module=dubbogo+user-info+server&org=ikurento.com&owner=ZX&pid=1447&revision=0.0.1&"+
- "side=provider&timeout=3000×tamp=1556509797245", common.WithParamsValue(constant.VERSION_KEY, "v1.1"))
- assert.NoError(t, err)
- exporter2 := proto.Export(protocol.NewBaseInvoker(url2))
- // make sure url
- eq2 := exporter2.GetInvoker().GetUrl().URLEqual(url2)
- assert.True(t, eq2)
-
- // make sure exporterMap after 'Unexport'
- _, ok := proto.(*DubboProtocol).ExporterMap().Load(url.ServiceKey())
- assert.True(t, ok)
- exporter.Unexport()
- _, ok = proto.(*DubboProtocol).ExporterMap().Load(url.ServiceKey())
- assert.False(t, ok)
-
- // make sure serverMap after 'Destroy'
- _, ok = proto.(*DubboProtocol).serverMap[url.Location]
- assert.True(t, ok)
- proto.Destroy()
- _, ok = proto.(*DubboProtocol).serverMap[url.Location]
- assert.False(t, ok)
-}
-
-func TestDubboProtocol_Refer(t *testing.T) {
- // Refer
- proto := GetProtocol()
- url, err := common.NewURL("dubbo://127.0.0.1:20000/com.ikurento.user.UserProvider?anyhost=true&" +
- "application=BDTService&category=providers&default.timeout=10000&dubbo=dubbo-provider-golang-1.0.0&" +
- "environment=dev&interface=com.ikurento.user.UserProvider&ip=192.168.56.1&methods=GetUser%2C&" +
- "module=dubbogo+user-info+server&org=ikurento.com&owner=ZX&pid=1447&revision=0.0.1&" +
- "side=provider&timeout=3000×tamp=1556509797245")
- assert.NoError(t, err)
- clientConf = &ClientConfig{}
- invoker := proto.Refer(url)
-
- // make sure url
- eq := invoker.GetUrl().URLEqual(url)
- assert.True(t, eq)
-
- // make sure invokers after 'Destroy'
- invokersLen := len(proto.(*DubboProtocol).Invokers())
- assert.Equal(t, 1, invokersLen)
- proto.Destroy()
- invokersLen = len(proto.(*DubboProtocol).Invokers())
- assert.Equal(t, 0, invokersLen)
-}
+//
+//import (
+// "testing"
+//)
+//
+//import (
+// "github.com/stretchr/testify/assert"
+//)
+//
+//import (
+// "github.com/apache/dubbo-go/common"
+// "github.com/apache/dubbo-go/common/constant"
+// "github.com/apache/dubbo-go/protocol"
+//)
+//
+//func TestDubboProtocol_Export(t *testing.T) {
+// // Export
+// proto := GetProtocol()
+// srvConf = &ServerConfig{}
+// url, err := common.NewURL("dubbo://127.0.0.1:20000/com.ikurento.user.UserProvider?anyhost=true&" +
+// "application=BDTService&category=providers&default.timeout=10000&dubbo=dubbo-provider-golang-1.0.0&" +
+// "environment=dev&interface=com.ikurento.user.UserProvider&ip=192.168.56.1&methods=GetUser%2C&" +
+// "module=dubbogo+user-info+server&org=ikurento.com&owner=ZX&pid=1447&revision=0.0.1&" +
+// "side=provider&timeout=3000×tamp=1556509797245")
+// assert.NoError(t, err)
+// exporter := proto.Export(protocol.NewBaseInvoker(url))
+//
+// // make sure url
+// eq := exporter.GetInvoker().GetUrl().URLEqual(url)
+// assert.True(t, eq)
+//
+// // second service: the same path and the different version
+// url2, err := common.NewURL("dubbo://127.0.0.1:20000/com.ikurento.user.UserProvider?anyhost=true&"+
+// "application=BDTService&category=providers&default.timeout=10000&dubbo=dubbo-provider-golang-1.0.0&"+
+// "environment=dev&interface=com.ikurento.user.UserProvider&ip=192.168.56.1&methods=GetUser%2C&"+
+// "module=dubbogo+user-info+server&org=ikurento.com&owner=ZX&pid=1447&revision=0.0.1&"+
+// "side=provider&timeout=3000×tamp=1556509797245", common.WithParamsValue(constant.VERSION_KEY, "v1.1"))
+// assert.NoError(t, err)
+// exporter2 := proto.Export(protocol.NewBaseInvoker(url2))
+// // make sure url
+// eq2 := exporter2.GetInvoker().GetUrl().URLEqual(url2)
+// assert.True(t, eq2)
+//
+// // make sure exporterMap after 'Unexport'
+// _, ok := proto.(*DubboProtocol).ExporterMap().Load(url.ServiceKey())
+// assert.True(t, ok)
+// exporter.Unexport()
+// _, ok = proto.(*DubboProtocol).ExporterMap().Load(url.ServiceKey())
+// assert.False(t, ok)
+//
+// // make sure serverMap after 'Destroy'
+// _, ok = proto.(*DubboProtocol).serverMap[url.Location]
+// assert.True(t, ok)
+// proto.Destroy()
+// _, ok = proto.(*DubboProtocol).serverMap[url.Location]
+// assert.False(t, ok)
+//}
+//
+//func TestDubboProtocol_Refer(t *testing.T) {
+// // Refer
+// proto := GetProtocol()
+// url, err := common.NewURL("dubbo://127.0.0.1:20000/com.ikurento.user.UserProvider?anyhost=true&" +
+// "application=BDTService&category=providers&default.timeout=10000&dubbo=dubbo-provider-golang-1.0.0&" +
+// "environment=dev&interface=com.ikurento.user.UserProvider&ip=192.168.56.1&methods=GetUser%2C&" +
+// "module=dubbogo+user-info+server&org=ikurento.com&owner=ZX&pid=1447&revision=0.0.1&" +
+// "side=provider&timeout=3000×tamp=1556509797245")
+// assert.NoError(t, err)
+// clientConf = &ClientConfig{}
+// invoker := proto.Refer(url)
+//
+// // make sure url
+// eq := invoker.GetUrl().URLEqual(url)
+// assert.True(t, eq)
+//
+// // make sure invokers after 'Destroy'
+// invokersLen := len(proto.(*DubboProtocol).Invokers())
+// assert.Equal(t, 1, invokersLen)
+// proto.Destroy()
+// invokersLen = len(proto.(*DubboProtocol).Invokers())
+// assert.Equal(t, 0, invokersLen)
+//}
diff --git a/remoting/codec.go b/remoting/codec.go
index 70071bdec..972a00d12 100644
--- a/remoting/codec.go
+++ b/remoting/codec.go
@@ -7,22 +7,22 @@ import (
type Codec interface {
EncodeRequest(request *Request) (*bytes.Buffer, error)
EncodeResponse(response *Response) (*bytes.Buffer, error)
- DecodeRequest(*bytes.Buffer) (*Request, int, error)
- DecodeResponse(*bytes.Buffer) (*Response, int, error)
+ DecodeRequest(data []byte) (*Request, int, error)
+ DecodeResponse(data []byte) (*Response, int, error)
}
var (
- codec map[string]*Codec
+ codec map[string]Codec
)
func init() {
- codec = make(map[string]*Codec, 2)
+ codec = make(map[string]Codec, 2)
}
-func NewCodec(protocol string, codecTmp *Codec) {
+func NewCodec(protocol string, codecTmp Codec) {
codec[protocol] = codecTmp
}
-func GetCodec(protocol string) *Codec {
+func GetCodec(protocol string) Codec {
return codec[protocol]
}
diff --git a/remoting/exchange.go b/remoting/exchange.go
index f63c9119f..498022926 100644
--- a/remoting/exchange.go
+++ b/remoting/exchange.go
@@ -37,11 +37,11 @@ func NewRequest(version string) *Request {
}
}
-func (request *Request) SetHeartbeat(isHeartbeat bool) {
- if isHeartbeat {
-
- }
-}
+//func (request *Request) SetHeartbeat(isHeartbeat bool) {
+// if isHeartbeat {
+//
+// }
+//}
// Response ...
type Response struct {
@@ -52,7 +52,6 @@ type Response struct {
Event bool
Error error
Result interface{}
- Reply interface{}
}
// NewResponse ...
@@ -91,12 +90,14 @@ type PendingResponse struct {
ReadStart time.Time
callback common.AsyncCallback
response *Response
+ Reply interface{}
Done chan struct{}
}
// NewPendingResponse ...
-func NewPendingResponse() *PendingResponse {
+func NewPendingResponse(id int64) *PendingResponse {
return &PendingResponse{
+ seq: id,
start: time.Now(),
response: &Response{},
Done: make(chan struct{}),
@@ -112,20 +113,3 @@ func (r PendingResponse) GetCallResponse() common.CallbackResponse {
Reply: r.response,
}
}
-
-type Client interface {
- //invoke once for connection
- Connect(url common.URL)
- Close()
- Request(request *Request, timeout time.Duration, callback common.AsyncCallback, response *PendingResponse) error
-}
-
-type Server interface {
- //invoke once for connection
- Open(url common.URL)
-}
-
-type ResponseHandler interface {
- Handler(response *Response)
-}
-
diff --git a/remoting/exchange_client.go b/remoting/exchange_client.go
index 962c324ee..aff8bba22 100644
--- a/remoting/exchange_client.go
+++ b/remoting/exchange_client.go
@@ -8,13 +8,29 @@ import (
"time"
)
+var (
+ pendingResponses *sync.Map = new(sync.Map)
+)
+
type SequenceType int64
type ExchangeClient struct {
- ConnectTimeout time.Duration
- address string
- client Client
- pendingResponses *sync.Map
+ ConnectTimeout time.Duration
+ address string
+ client Client
+}
+
+type Client interface {
+ SetExchangeClient(client *ExchangeClient)
+ SetResponseHandler(responseHandler ResponseHandler)
+ //invoke once for connection
+ Connect(url common.URL)
+ Close()
+ Request(request *Request, timeout time.Duration, callback common.AsyncCallback, response *PendingResponse) error
+}
+
+type ResponseHandler interface {
+ Handler(response *Response)
}
func NewExchangeClient(url common.URL, client Client, connectTimeout time.Duration) *ExchangeClient {
@@ -23,20 +39,23 @@ func NewExchangeClient(url common.URL, client Client, connectTimeout time.Durati
address: url.Location,
client: client,
}
+ client.SetExchangeClient(exchangeClient)
client.Connect(url)
+ client.SetResponseHandler(exchangeClient)
return exchangeClient
}
func (client *ExchangeClient) Request(invocation *protocol.Invocation, url common.URL, timeout time.Duration,
result *protocol.RPCResult) error {
- requestId := int64(SequenceId())
request := NewRequest("2.0.2")
request.Data = invocation
request.Event = false
request.TwoWay = true
- rsp := NewPendingResponse()
- rsp.response = NewResponse(requestId, "2.0.2")
+ rsp := NewPendingResponse(request.Id)
+ rsp.response = NewResponse(request.Id, "2.0.2")
+ rsp.Reply = (*invocation).Reply()
+ AddPendingResponse(rsp)
//rsp.callback = invo
err := client.client.Request(request, timeout, nil, rsp)
@@ -44,22 +63,23 @@ func (client *ExchangeClient) Request(invocation *protocol.Invocation, url commo
result.Err = err
return err
}
- result.Rest = rsp.response
+ result.Rest = rsp.response.Result
//result.Attrs = rsp.response.
return nil
}
func (client *ExchangeClient) AsyncRequest(invocation *protocol.Invocation, url common.URL, timeout time.Duration,
callback common.AsyncCallback, result *protocol.RPCResult) error {
- requestId := int64(SequenceId())
request := NewRequest("2.0.2")
request.Data = invocation
request.Event = false
request.TwoWay = true
- rsp := NewPendingResponse()
- rsp.response = NewResponse(requestId, "2.0.2")
+ rsp := NewPendingResponse(request.Id)
+ rsp.response = NewResponse(request.Id, "2.0.2")
rsp.callback = callback
+ rsp.Reply = (*invocation).Reply()
+ AddPendingResponse(rsp)
err := client.client.Request(request, timeout, nil, rsp)
if err != nil {
@@ -79,7 +99,7 @@ func (client *ExchangeClient) Send(invocation *protocol.Invocation, timeout time
request.Event = false
request.TwoWay = false
- rsp := NewPendingResponse()
+ rsp := NewPendingResponse(request.Id)
rsp.response = NewResponse(requestId, "2.0.2")
err := client.client.Request(request, timeout, nil, rsp)
@@ -96,7 +116,7 @@ func (client *ExchangeClient) Close() {
func (client *ExchangeClient) Handler(response *Response) {
- pendingResponse := client.removePendingResponse(SequenceType(response.Id))
+ pendingResponse := removePendingResponse(SequenceType(response.Id))
if pendingResponse == nil {
logger.Errorf("failed to get pending response context for response package %s", *response)
return
@@ -111,16 +131,23 @@ func (client *ExchangeClient) Handler(response *Response) {
}
}
-func (client *ExchangeClient) addPendingResponse(pr *PendingResponse) {
- client.pendingResponses.Store(SequenceType(pr.seq), pr)
+func AddPendingResponse(pr *PendingResponse) {
+ pendingResponses.Store(SequenceType(pr.seq), pr)
}
-func (client *ExchangeClient) removePendingResponse(seq SequenceType) *PendingResponse {
- if client.pendingResponses == nil {
+func removePendingResponse(seq SequenceType) *PendingResponse {
+ if pendingResponses == nil {
return nil
}
- if presp, ok := client.pendingResponses.Load(seq); ok {
- client.pendingResponses.Delete(seq)
+ if presp, ok := pendingResponses.Load(seq); ok {
+ pendingResponses.Delete(seq)
+ return presp.(*PendingResponse)
+ }
+ return nil
+}
+
+func GetPendingResponse(seq SequenceType) *PendingResponse {
+ if presp, ok := pendingResponses.Load(seq); ok {
return presp.(*PendingResponse)
}
return nil
diff --git a/remoting/exchange_server.go b/remoting/exchange_server.go
index 003602420..4aae39fd8 100644
--- a/remoting/exchange_server.go
+++ b/remoting/exchange_server.go
@@ -2,18 +2,19 @@ package remoting
import (
"github.com/apache/dubbo-go/common"
- "github.com/apache/dubbo-go/protocol"
- "github.com/apache/dubbo-go/protocol/invocation"
- "github.com/apache/dubbo-go/remoting/getty"
)
+type Server interface {
+ //invoke once for connection
+ Start()
+ Stop()
+}
type ExchangeServer struct {
- Server *getty.Server
+ Server Server
}
-func NewExchangeServer(url common.URL, handler func(*invocation.RPCInvocation) protocol.RPCResult) *ExchangeServer {
- server := getty.NewServer(url, handler)
+func NewExchangeServer(url common.URL, server Server) *ExchangeServer {
exchangServer := &ExchangeServer{
Server: server,
}
@@ -21,9 +22,9 @@ func NewExchangeServer(url common.URL, handler func(*invocation.RPCInvocation) p
}
func (server *ExchangeServer) Start() {
- server.Server.Start()
+ (server.Server).Start()
}
func (server *ExchangeServer) Stop() {
- server.Server.Stop()
+ (server.Server).Stop()
}
diff --git a/remoting/getty/getty_client.go b/remoting/getty/getty_client.go
index 8b79a7775..c22f57687 100644
--- a/remoting/getty/getty_client.go
+++ b/remoting/getty/getty_client.go
@@ -19,6 +19,7 @@ package getty
import (
"github.com/apache/dubbo-go/remoting"
+ "gopkg.in/yaml.v2"
"math/rand"
"time"
)
@@ -27,7 +28,6 @@ import (
"github.com/dubbogo/getty"
gxsync "github.com/dubbogo/gost/sync"
perrors "github.com/pkg/errors"
- "gopkg.in/yaml.v2"
)
import (
@@ -47,7 +47,7 @@ var (
clientGrpool *gxsync.TaskPool
)
-func doInit(protocol string) {
+func initClient(protocol string) {
if protocol == "" {
return
}
@@ -134,8 +134,9 @@ type Client struct {
opts Options
conf ClientConfig
pool *gettyRPCClientPool
- codec *remoting.Codec
- responseHandler *remoting.ResponseHandler
+ codec remoting.Codec
+ responseHandler remoting.ResponseHandler
+ ExchangeClient *remoting.ExchangeClient
//sequence atomic.Uint64
//pendingResponses *sync.Map
}
@@ -149,18 +150,24 @@ func NewClient(opt Options) *Client {
c := &Client{
opts: opt,
- conf: *clientConf,
}
- c.pool = newGettyRPCClientConnPool(c, clientConf.PoolSize, time.Duration(int(time.Second)*clientConf.PoolTTL))
-
return c
}
+func (c *Client) SetExchangeClient(client *remoting.ExchangeClient) {
+ c.ExchangeClient = client
+}
+func (c *Client) SetResponseHandler(responseHandler remoting.ResponseHandler) {
+ c.responseHandler = responseHandler
+}
func (c *Client) Connect(url common.URL) {
- doInit(url.Protocol)
+ initClient(url.Protocol)
+ c.conf = *clientConf
+ // new client
+ c.pool = newGettyRPCClientConnPool(c, clientConf.PoolSize, time.Duration(int(time.Second)*clientConf.PoolTTL))
// codec
c.codec = remoting.GetCodec(url.Protocol)
- c.addr = url.Ip + ":" + url.Port
+ c.addr = url.Location
}
func (c *Client) Close() {
if c.pool != nil {
@@ -340,13 +347,13 @@ func (c *Client) heartbeat(session getty.Session) error {
req := remoting.NewRequest("2.0.2")
req.TwoWay = true
req.Event = true
- resp := remoting.NewPendingResponse()
+ resp := remoting.NewPendingResponse(req.Id)
+ remoting.AddPendingResponse(resp)
return c.transfer(session, req, 3*time.Second, resp)
}
func (c *Client) transfer(session getty.Session, request *remoting.Request, timeout time.Duration,
rsp *remoting.PendingResponse) error {
-
//sequence = c.sequence.Add(1)
//
//if pkg == nil {
diff --git a/remoting/getty/listener.go b/remoting/getty/listener.go
index daa1a583c..0dc2792a2 100644
--- a/remoting/getty/listener.go
+++ b/remoting/getty/listener.go
@@ -107,7 +107,7 @@ func (h *RpcClientHandler) OnMessage(session getty.Session, pkg interface{}) {
if p.Error != nil {
logger.Errorf("rpc heartbeat response{error: %#v}", p.Error)
}
- (*h.conn.pool.rpcClient.responseHandler).Handler(p)
+ (h.conn.pool.rpcClient.responseHandler).Handler(p)
//FIXME
//if p.Header.Type&hessian.PackageResponse != 0x00 {
// logger.Debugf("get rpc heartbeat response{header: %#v, body: %#v}", p.Header, p.Body)
@@ -126,7 +126,7 @@ func (h *RpcClientHandler) OnMessage(session getty.Session, pkg interface{}) {
h.conn.updateSession(session)
- (*h.conn.pool.rpcClient.responseHandler).Handler(p)
+ (h.conn.pool.rpcClient.responseHandler).Handler(p)
//
//pendingResponse := h.conn.pool.rpcClient.removePendingResponse(SequenceType(p.Header.ID))
@@ -239,6 +239,9 @@ func (h *RpcServerHandler) OnMessage(session getty.Session, pkg interface{}) {
}
resp := remoting.NewResponse(req.Id, req.Version)
resp.Status = hessian.Response_OK
+ resp.Event = req.Event
+ resp.SerialID = req.SerialID
+ resp.Version = "2.0.2"
// heartbeat
if req.Event {
@@ -275,11 +278,11 @@ func (h *RpcServerHandler) OnMessage(session getty.Session, pkg interface{}) {
}()
- invoc, ok := req.Data.(invocation.RPCInvocation)
+ invoc, ok := req.Data.(*invocation.RPCInvocation)
if !ok {
}
- result := h.server.requestHandler(&invoc)
+ result := h.server.requestHandler(invoc)
if !req.TwoWay {
return
}
@@ -316,7 +319,7 @@ func (h *RpcServerHandler) OnCron(session getty.Session) {
// rebuildCtx rebuild the context by attachment.
// Once we decided to transfer more context's key-value, we should change this.
// now we only support rebuild the tracing context
-func rebuildCtx(inv *invocation.RPCInvocation) context.Context {
+func RebuildCtx(inv *invocation.RPCInvocation) context.Context {
ctx := context.Background()
// actually, if user do not use any opentracing framework, the err will not be nil.
diff --git a/remoting/getty/readwriter.go b/remoting/getty/readwriter.go
index 721df105a..b59d9b82f 100644
--- a/remoting/getty/readwriter.go
+++ b/remoting/getty/readwriter.go
@@ -18,7 +18,6 @@
package getty
import (
- "bytes"
"github.com/apache/dubbo-go/remoting"
"reflect"
)
@@ -49,9 +48,8 @@ func NewRpcClientPackageHandler(client *Client) *RpcClientPackageHandler {
func (p *RpcClientPackageHandler) Read(ss getty.Session, data []byte) (interface{}, int, error) {
//pkg := &DubboPackage{}
-
- buf := bytes.NewBuffer(data)
- resp, length, err := (*p.client.codec).DecodeResponse(buf)
+ //p.client.ExchangeClient.GetPendingResponse(remoting.SequenceType())
+ resp, length, err := (p.client.codec).DecodeResponse(data)
//err := pkg.Unmarshal(buf, p.client)
if err != nil {
originErr := perrors.Cause(err)
@@ -78,7 +76,7 @@ func (p *RpcClientPackageHandler) Write(ss getty.Session, pkg interface{}) ([]by
return nil, perrors.New("invalid rpc request")
}
- buf, err := (*p.client.codec).EncodeRequest(req)
+ buf, err := (p.client.codec).EncodeRequest(req)
if err != nil {
logger.Warnf("binary.Write(req{%#v}) = err{%#v}", req, perrors.WithStack(err))
return nil, perrors.WithStack(err)
@@ -91,18 +89,21 @@ func (p *RpcClientPackageHandler) Write(ss getty.Session, pkg interface{}) ([]by
// RpcServerPackageHandler
////////////////////////////////////////////
-var (
- rpcServerPkgHandler = &RpcServerPackageHandler{}
-)
+//var (
+// rpcServerPkgHandler = &RpcServerPackageHandler{}
+//)
// RpcServerPackageHandler ...
type RpcServerPackageHandler struct {
server *Server
}
+func NewRpcServerPackageHandler(server *Server) *RpcServerPackageHandler {
+ return &RpcServerPackageHandler{server: server}
+}
+
func (p *RpcServerPackageHandler) Read(ss getty.Session, data []byte) (interface{}, int, error) {
- buf := bytes.NewBuffer(data)
- req, length, err := (*p.server.codec).DecodeRequest(buf)
+ req, length, err := (p.server.codec).DecodeRequest(data)
//resp,len, err := (*p.).DecodeResponse(buf)
if err != nil {
@@ -126,7 +127,7 @@ func (p *RpcServerPackageHandler) Write(ss getty.Session, pkg interface{}) ([]by
return nil, perrors.New("invalid rpc response")
}
- buf, err := (*p.server.codec).EncodeResponse(res)
+ buf, err := (p.server.codec).EncodeResponse(res)
if err != nil {
logger.Warnf("binary.Write(res{%#v}) = err{%#v}", res, perrors.WithStack(err))
return nil, perrors.WithStack(err)
diff --git a/remoting/getty/server.go b/remoting/getty/server.go
index 7ce9aafab..00fba5cae 100644
--- a/remoting/getty/server.go
+++ b/remoting/getty/server.go
@@ -105,7 +105,7 @@ func SetServerGrpool() {
type Server struct {
conf ServerConfig
addr string
- codec *remoting.Codec
+ codec remoting.Codec
tcpServer getty.Server
rpcHandler *RpcServerHandler
requestHandler func(*invocation.RPCInvocation) protocol.RPCResult
@@ -114,7 +114,7 @@ type Server struct {
// NewServer ...
func NewServer(url common.URL, handlers func(*invocation.RPCInvocation) protocol.RPCResult) *Server {
//init
- doInit(url.Protocol)
+ initServer(url.Protocol)
s := &Server{
conf: *srvConf,
@@ -153,7 +153,7 @@ func (s *Server) newSession(session getty.Session) error {
session.SetName(conf.GettySessionParam.SessionName)
session.SetMaxMsgLen(conf.GettySessionParam.MaxMsgLen)
- session.SetPkgHandler(rpcServerPkgHandler)
+ session.SetPkgHandler(NewRpcServerPackageHandler(s))
session.SetEventListener(s.rpcHandler)
session.SetWQLen(conf.GettySessionParam.PkgWQSize)
session.SetReadTimeout(conf.GettySessionParam.tcpReadTimeout)
--
GitLab