reformat codes

rpc/client.go

@@ -255,7 +255,8 @@ func (c *Client) getClientRpcSession(session getty.Session) (rpcSession, error)
 			break
 		}
 	}
-	return rpcSession, err
+
+	return rpcSession, jerrors.Trace(err)
 }
 
 func (c *Client) ping(session getty.Session) error {

rpc/codec.go

@@ -53,15 +53,23 @@ const (
 )
 
 ////////////////////////////////////////////
+//  getty error code
+////////////////////////////////////////////
+
+type GettyErrorCode int32
+
+const (
+	GettyOK   GettyErrorCode = 0x00
+	GettyFail                = 0x01
+)
+
+////////////////////////////////////////////
 // GettyPackageHandler
 ////////////////////////////////////////////
 
 const (
 	gettyPackageMagic = 0x20160905
 	maxPackageLen     = 1024 * 1024
-
-	ReplyTypeData = 0x01
-	ReplyTypeAck  = 0x03
 )
 
 var (
@@ -90,12 +98,12 @@ type RPCPackage interface {
 }
 
 type GettyPackageHeader struct {
-	Magic    uint32
+	Magic    uint32 // magic number
 	LogID    uint32 // log id
 	Sequence uint64 // request/response sequence
 
-	Command gettyCommand // operation command code
-	Code    int32        // error code
+	Command gettyCommand   // operation command code
+	Code    GettyErrorCode // error code
 
 	ServiceID uint32 // service id
 	Len       uint32 // body length
@@ -121,16 +129,17 @@ func (p GettyPackage) Marshal() (*bytes.Buffer, error) {
 	buf = &bytes.Buffer{}
 	err = binary.Write(buf, binary.LittleEndian, p.H)
 	if err != nil {
-		return nil, err
+		return nil, jerrors.Trace(err)
 	}
 	if p.B != nil {
 		if err = p.B.Marshal(buf); err != nil {
 			return nil, jerrors.Trace(err)
 		}
 
+		// body length
 		length = buf.Len() - gettyPackageHeaderLen
 		size = (int)((uint)(unsafe.Sizeof(p.H.Len)))
-		buf0 = bytes.NewBuffer(buf.Bytes()[gettyPackageHeaderLen-size:])
+		buf0 = bytes.NewBuffer(buf.Bytes()[gettyPackageHeaderLen-size : size])
 		binary.Write(buf0, binary.LittleEndian, length)
 	}
 
@@ -143,8 +152,7 @@ func (p *GettyPackage) Unmarshal(buf *bytes.Buffer) (int, error) {
 	}
 
 	// header
-	err := binary.Read(buf, binary.LittleEndian, &(p.H))
-	if err != nil {
+	if err := binary.Read(buf, binary.LittleEndian, &(p.H)); err != nil {
 		return 0, jerrors.Trace(err)
 	}
 	if p.H.Magic != gettyPackageMagic {
@@ -159,7 +167,7 @@ func (p *GettyPackage) Unmarshal(buf *bytes.Buffer) (int, error) {
 	}
 
 	if p.H.Len != 0 {
-		if err = p.B.Unmarshal(bytes.NewBuffer(buf.Next(int(p.H.Len)))); err != nil {
+		if err := p.B.Unmarshal(bytes.NewBuffer(buf.Next(int(p.H.Len)))); err != nil {
 			return 0, jerrors.Trace(err)
 		}
 	}
@@ -171,6 +179,8 @@ func (p *GettyPackage) Unmarshal(buf *bytes.Buffer) (int, error) {
 // GettyRPCRequest
 ////////////////////////////////////////////
 
+type GettyRPCHeaderLenType uint16
+
 type GettyRPCRequestHeader struct {
 	Service  string        `json:"service,omitempty"`
 	Method   string        `json:"method,omitempty"`
@@ -188,7 +198,7 @@ type GettyRPCRequest struct {
 }
 
 // json rpc stream format
-// |-- 2B (GettyRPCRequestHeader length) --|-- GettyRPCRequestHeader --|
+// |-- 2B (GettyRPCRequestHeader length) --|-- GettyRPCRequestHeader --|-- rpc body --|
 
 func NewGettyRPCRequest(server *Server) *GettyRPCRequest {
 	return &GettyRPCRequest{
@@ -277,7 +287,7 @@ func (req *GettyRPCRequest) Unmarshal(buf *bytes.Buffer) error {
 ////////////////////////////////////////////
 
 type GettyRPCResponseHeader struct {
-	Error string `json:"error,omitempty"`
+	Error string `json:"error,omitempty"` // error string
 }
 
 type GettyRPCResponse struct {
@@ -296,16 +306,14 @@ func (resp *GettyRPCResponse) Marshal(buf *bytes.Buffer) error {
 		return jerrors.Trace(err)
 	}
 
-	err = binary.Write(buf, binary.LittleEndian, uint16(len(headerData)))
+	err = binary.Write(buf, binary.LittleEndian, (GettyRPCHeaderLenType)(len(headerData)))
 	if err != nil {
-		return err
+		return jerrors.Trace(err)
 	}
-	_, err = buf.Write(headerData)
-	if err != nil {
+	if _, err = buf.Write(headerData); err != nil {
 		return jerrors.Trace(err)
 	}
-	_, err = buf.Write(bodyData)
-	if err != nil {
+	if _, err = buf.Write(bodyData); err != nil {
 		return jerrors.Trace(err)
 	}
 
@@ -318,21 +326,17 @@ func (resp *GettyRPCResponse) Unmarshal(buf *bytes.Buffer) error {
 		return ErrNotEnoughStream
 	}
 
-	var headerLen uint16
+	var headerLen GettyRPCHeaderLenType
 	err := binary.Read(buf, binary.LittleEndian, &headerLen)
 	if err != nil {
-		return err
+		return jerrors.Trace(err)
 	}
 
 	header := buf.Next(int(headerLen))
 	if len(header) != int(headerLen) {
 		return ErrNotEnoughStream
 	}
-	body := buf.Next(int(buf.Len()))
-	if len(body) == 0 {
-		return ErrNotEnoughStream
-	}
-	resp.body = body
+	resp.body = buf.Next(int(buf.Len()))
 	err = json.Unmarshal(header, resp.header)
 	if err != nil {
 		return jerrors.Trace(err)

rpc/listener.go

@@ -57,7 +57,7 @@ func (h *RpcServerHandler) OnOpen(session getty.Session) error {
 	}
 	h.rwlock.RUnlock()
 	if err != nil {
-		return err
+		return jerrors.Trace(err)
 	}
 
 	log.Info("got session:%s", session.Stat())
@@ -143,6 +143,7 @@ func (h *RpcServerHandler) replyCmd(session getty.Session, reqPkg *GettyPackage,
 	rspPkg.H.Code = 0
 	rspPkg.H.Command = gettyCmdRPCResponse
 	if len(err) != 0 {
+		rspPkg.H.Code = GettyFail
 		rspPkg.B = &GettyRPCResponse{
 			header: GettyRPCResponseHeader{
 				Error: err,
@@ -227,7 +228,7 @@ func (h *RpcClientHandler) OnMessage(session getty.Session, pkg interface{}) {
 		log.Error("response body:{%#v} type is not *GettyRPCResponse", p.B)
 		return
 	}
-	if len(rsp.header.Error) > 0 {
+	if p.H.Code == GettyFail && len(rsp.header.Error) > 0 {
 		pendingResponse.err = jerrors.New(rsp.header.Error)
 	}
 	err := json.Unmarshal(rsp.body.([]byte), pendingResponse.reply)

rpc/readwriter.go

@@ -25,15 +25,10 @@ func NewRpcServerPackageHandler(server *Server) *RpcServerPackageHandler {
 }
 
 func (p *RpcServerPackageHandler) Read(ss getty.Session, data []byte) (interface{}, int, error) {
-	var (
-		err    error
-		length int
-		pkg    GettyPackage
-		buf    *bytes.Buffer
-	)
-
-	buf = bytes.NewBuffer(data)
-	length, err = pkg.Unmarshal(buf)
+	var pkg GettyPackage
+
+	buf := bytes.NewBuffer(data)
+	length, err := pkg.Unmarshal(buf)
 	if err != nil {
 		if jerrors.Cause(err) == ErrNotEnoughStream {
 			return nil, 0, nil
@@ -45,19 +40,13 @@ func (p *RpcServerPackageHandler) Read(ss getty.Session, data []byte) (interface
 }
 
 func (p *RpcServerPackageHandler) Write(ss getty.Session, pkg interface{}) error {
-	var (
-		ok   bool
-		err  error
-		resp *GettyPackage
-		buf  *bytes.Buffer
-	)
-
-	if resp, ok = pkg.(*GettyPackage); !ok {
+	resp, ok := pkg.(*GettyPackage)
+	if !ok {
 		log.Error("illegal pkg:%+v\n", pkg)
 		return jerrors.New("invalid rpc response")
 	}
 
-	buf, err = resp.Marshal()
+	buf, err := resp.Marshal()
 	if err != nil {
 		log.Warn("binary.Write(resp{%#v}) = err{%#v}", resp, err)
 		return jerrors.Trace(err)
@@ -78,15 +67,10 @@ func NewRpcClientPackageHandler() *RpcClientPackageHandler {
 }
 
 func (p *RpcClientPackageHandler) Read(ss getty.Session, data []byte) (interface{}, int, error) {
-	var (
-		err    error
-		length int
-		pkg    GettyPackage
-		buf    *bytes.Buffer
-	)
-
-	buf = bytes.NewBuffer(data)
-	length, err = pkg.Unmarshal(buf)
+	var pkg GettyPackage
+
+	buf := bytes.NewBuffer(data)
+	length, err := pkg.Unmarshal(buf)
 	if err != nil {
 		if err == ErrNotEnoughStream {
 			return nil, 0, nil
@@ -98,19 +82,13 @@ func (p *RpcClientPackageHandler) Read(ss getty.Session, data []byte) (interface
 }
 
 func (p *RpcClientPackageHandler) Write(ss getty.Session, pkg interface{}) error {
-	var (
-		ok  bool
-		err error
-		req *GettyPackage
-		buf *bytes.Buffer
-	)
-
-	if req, ok = pkg.(*GettyPackage); !ok {
+	req, ok := pkg.(*GettyPackage)
+	if !ok {
 		log.Error("illegal pkg:%+v\n", pkg)
 		return jerrors.New("invalid rpc request")
 	}
 
-	buf, err = req.Marshal()
+	buf, err := req.Marshal()
 	if err != nil {
 		log.Warn("binary.Write(req{%#v}) = err{%#v}", req, err)
 		return jerrors.Trace(err)

session.go

@@ -323,7 +323,7 @@ func (s *session) WritePkg(pkg interface{}, timeout time.Duration) error {
 			s.incWritePkgNum()
 			// gxlog.CError("after incWritePkgNum, ss:%s", s.Stat())
 		}
-		return err
+		return jerrors.Trace(err)
 	}
 	select {
 	case s.wQ <- pkg:
@@ -640,7 +640,7 @@ func (s *session) handleTCPPackage() error {
 		}
 	}
 
-	return err
+	return jerrors.Trace(err)
 }
 
 // get package from udp packet
@@ -709,7 +709,7 @@ func (s *session) handleUDPPackage() error {
 		s.rQ <- UDPContext{Pkg: pkg, PeerAddr: addr}
 	}
 
-	return err
+	return jerrors.Trace(err)
 }
 
 // get package from websocket stream
@@ -737,7 +737,7 @@ func (s *session) handleWSPackage() error {
 			log.Warn("%s, [session.handleWSPackage] = error{%s}",
 				s.sessionToken(), jerrors.ErrorStack(err))
 			// s.errFlag = true
-			return err
+			return jerrors.Trace(err)
 		}
 		s.UpdateActive()
 		if s.reader != nil {