opensource/grpc-go
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #401 from iamqizhao/master

Browse Source 
some cleanup
This commit is contained in:
Qi Zhao
2015-10-15 11:07:02 -07:00
parent ebeb34ad2f 61c1259226
commit a90aeac84e
1 changed files with 8 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
rpc_util.go
View File

@ -128,19 +128,14 @@ type parser struct {
s io.Reader s io.Reader
} }
// msgFixedHeader defines the header of a gRPC message. Find more detail
// at http://www.grpc.io/docs/guides/wire.html.
type msgFixedHeader struct {
T payloadFormat
Length uint32
}
// recvMsg is to read a complete gRPC message from the stream. It is blocking if // recvMsg is to read a complete gRPC message from the stream. It is blocking if
// the message has not been complete yet. It returns the message and its type, // the message has not been complete yet. It returns the message and its type,
// EOF is returned with nil msg and 0 pf if the entire stream is done. Other // EOF is returned with nil msg and 0 pf if the entire stream is done. Other
// non-nil error is returned if something is wrong on reading. // non-nil error is returned if something is wrong on reading.
func (p *parser) recvMsg() (pf payloadFormat, msg []byte, err error) { func (p *parser) recvMsg() (pf payloadFormat, msg []byte, err error) {
var buf [5]byte // see msgFixedHeader // The header of a gRPC message. Find more detail
// at http://www.grpc.io/docs/guides/wire.html.
var buf [5]byte
if _, err := io.ReadFull(p.s, buf[:]); err != nil { if _, err := io.ReadFull(p.s, buf[:]); err != nil {
return 0, nil, err return 0, nil, err
@ -169,7 +164,7 @@ func encode(c Codec, msg interface{}, pf payloadFormat) ([]byte, error) {
// Write message into the fixed header. // Write message into the fixed header.
buf.WriteByte(uint8(pf)) buf.WriteByte(uint8(pf))
var b []byte var b []byte
var length int var length uint32
if msg != nil { if msg != nil {
var err error var err error
// TODO(zhaoq): optimize to reduce memory alloc and copying. // TODO(zhaoq): optimize to reduce memory alloc and copying.
@ -177,13 +172,13 @@ func encode(c Codec, msg interface{}, pf payloadFormat) ([]byte, error) {
if err != nil { if err != nil {
return nil, err return nil, err
} }
length = len(b) length = uint32(len(b))
} }
if length > math.MaxUint32 { if length > math.MaxUint32 {
return nil, Errorf(codes.InvalidArgument, "grpc: message too large (%d bytes)", length) return nil, Errorf(codes.InvalidArgument, "grpc: message too large (%d bytes)", length)
} }
var szHdr [4]byte var szHdr [4]byte
binary.BigEndian.PutUint32(szHdr[:], uint32(length)) binary.BigEndian.PutUint32(szHdr[:], length)
buf.Write(szHdr[:]) buf.Write(szHdr[:])
buf.Write(b) buf.Write(b)
return buf.Bytes(), nil return buf.Bytes(), nil
@ -258,6 +253,8 @@ func Errorf(c codes.Code, format string, a ...interface{}) error {
// toRPCErr converts an error into a rpcError. // toRPCErr converts an error into a rpcError.
func toRPCErr(err error) error { func toRPCErr(err error) error {
switch e := err.(type) { switch e := err.(type) {
case rpcError:
return err
case transport.StreamError: case transport.StreamError:
return rpcError{ return rpcError{
code: e.Code, code: e.Code,