Skip to content

Imp: refactor the network transport layer #495

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 22 commits into from
Jun 4, 2020
Merged

Imp: refactor the network transport layer #495

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
22 commits
Select commit Hold shift + click to select a range
1eb4a83
getty refactor
cvictory Apr 24, 2020
fa61e9d
refactor getty
cvictory Apr 26, 2020
0ba15c8
refactor getty
cvictory May 8, 2020
9bc37f3
Merge branch 'master' of github.com:apache/dubbo-go into getty_reactor
cvictory May 8, 2020
c67d34a
fix review issue
cvictory May 8, 2020
acc7c79
fix review issue
cvictory May 8, 2020
d48fd1e
remove unused code, add license
cvictory May 9, 2020
58e7e86
fix review issue: add comment and some optimization
cvictory May 14, 2020
adb55ce
split import sentense, fix some review
cvictory May 18, 2020
1bbe0b4
modify log
cvictory May 18, 2020
dc2b68f
fix run issue
cvictory May 19, 2020
fe3c5b0
fix issue: create gettyClientConn every time
cvictory May 19, 2020
1e07952
fix : url.getPath contains slash, but it don't in dubbo-java
cvictory May 19, 2020
c6abc0d
unit test
cvictory May 20, 2020
15608af
fix review issue
cvictory May 27, 2020
cbac0c9
remove unused code
cvictory May 27, 2020
7f9c7a4
reactor import
cvictory May 28, 2020
7c30fdc
fix review issue
cvictory May 29, 2020
dd36c9c
fix review issue
cvictory May 31, 2020
713fcf9
fix review issue
cvictory Jun 2, 2020
a08d2c7
fix review issue
cvictory Jun 2, 2020
5ba2307
refactor ServiceKey func
cvictory Jun 2, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
696 changes: 348 additions & 348 deletions protocol/dubbo/client.go
View file
Open in desktop

Large diffs are not rendered by default.

564 changes: 282 additions & 282 deletions protocol/dubbo/client_test.go
View file
Open in desktop

Large diffs are not rendered by default.

276 changes: 138 additions & 138 deletions protocol/dubbo/codec.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,141 +16,141 @@
*/

package dubbo

import (
"bufio"
"bytes"
"fmt"
"time"
)

import (
"github.com/apache/dubbo-go-hessian2"
"github.com/apache/dubbo-go/common"
perrors "github.com/pkg/errors"
)

//SerialID serial ID
type SerialID byte

const (
// S_Dubbo dubbo serial id
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

// 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, opts ...interface{}) 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 len(opts) != 0 { // for client
client, ok := opts[0].(*Client)
if !ok {
return perrors.Errorf("opts[0] is not of type *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: pendingRsp.(*PendingResponse).response.reply}
}
}

// read body
err = codec.ReadBody(p.Body)
return perrors.WithStack(err)
}

////////////////////////////////////////////
// PendingResponse
////////////////////////////////////////////

// PendingResponse ...
type PendingResponse struct {
seq uint64
err error
start time.Time
readStart time.Time
callback common.AsyncCallback
response *Response
done chan struct{}
}

// NewPendingResponse ...
func NewPendingResponse() *PendingResponse {
return &PendingResponse{
start: time.Now(),
response: &Response{},
done: make(chan struct{}),
}
}

// GetCallResponse ...
func (r PendingResponse) GetCallResponse() common.CallbackResponse {
return AsyncCallbackResponse{
Cause: r.err,
Start: r.start,
ReadStart: r.readStart,
Reply: r.response,
}
}
//
//import (
// "bufio"
// "bytes"
// "fmt"
// "time"
//)
//
//import (
// "github.com/apache/dubbo-go-hessian2"
// "github.com/apache/dubbo-go/common"
// perrors "github.com/pkg/errors"
//)
//
////SerialID serial ID
//type SerialID byte
//
//const (
// // S_Dubbo dubbo serial id
// 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
//
//// 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, opts ...interface{}) 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 len(opts) != 0 { // for client
// client, ok := opts[0].(*Client)
// if !ok {
// return perrors.Errorf("opts[0] is not of type *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: pendingRsp.(*PendingResponse).response.reply}
// }
// }
//
// // read body
// err = codec.ReadBody(p.Body)
// return perrors.WithStack(err)
//}
//
//////////////////////////////////////////////
//// PendingResponse
//////////////////////////////////////////////
//
//// PendingResponse ...
//type PendingResponse struct {
// seq uint64
// err error
// start time.Time
// readStart time.Time
// callback common.AsyncCallback
// response *Response
// done chan struct{}
//}
//
//// NewPendingResponse ...
//func NewPendingResponse() *PendingResponse {
// return &PendingResponse{
// start: time.Now(),
// response: &Response{},
// done: make(chan struct{}),
// }
//}
//
//// GetCallResponse ...
//func (r PendingResponse) GetCallResponse() common.CallbackResponse {
// return AsyncCallbackResponse{
// Cause: r.err,
// Start: r.start,
// ReadStart: r.readStart,
// Reply: r.response,
// }
//}
Loading