Merge pull request #6 from hudangwei/feature/rpc

LGTM

rpc/client.go

@@ -120,7 +120,7 @@ func NewClient(confFile string) (*Client, error) {
 	return c, nil
 }
 
-func (c *Client) Call(service, method string, args interface{}, reply interface{}) error {
+func (c *Client) Call(addr, protocol, service, method string, args interface{}, reply interface{}) error {
 	b := &GettyRPCRequest{}
 	b.header.Service = service
 	b.header.Method = method
@@ -133,7 +133,7 @@ func (c *Client) Call(service, method string, args interface{}, reply interface{
 	resp := NewPendingResponse()
 	resp.reply = reply
 
-	session := c.selectSession()
+	session := c.selectSession(protocol, addr)
 	if session == nil {
 		return errSessionNotExist
 	}
@@ -157,8 +157,12 @@ func (c *Client) Close() {
 	c.registry = nil
 }
 
-func (c *Client) selectSession() getty.Session {
-	return nil
+func (c *Client) selectSession(protocol, addr string) getty.Session {
+	rpcConn, err := c.pool.getConn(protocol, addr)
+	if err != nil {
+		return nil
+	}
+	return rpcConn.selectSession()
 }
 
 func (c *Client) heartbeat(session getty.Session) error {

rpc/codec.go

@@ -16,9 +16,7 @@ import (
 )
 
 import (
-	"github.com/AlexStocks/goext/log"
 	log "github.com/AlexStocks/log4go"
-	"time"
 )
 
 ////////////////////////////////////////////
@@ -225,14 +223,14 @@ func (p *GettyPackage) Marshal() (*bytes.Buffer, error) {
 		buf             *bytes.Buffer
 	)
 
-	packLen = gettyPackageHeaderLen
+	packLen = rpcPackagePlaceholderLen + gettyPackageHeaderLen
 	if p.B != nil {
 		buf = &bytes.Buffer{}
 		length, err = p.B.Marshal(p.H.CodecType, buf)
 		if err != nil {
 			return nil, jerrors.Trace(err)
 		}
-		packLen = gettyPackageHeaderLen + length
+		packLen = rpcPackagePlaceholderLen + gettyPackageHeaderLen + length
 	}
 	buf0 := &bytes.Buffer{}
 	err = binary.Write(buf0, binary.LittleEndian, uint16(packLen))
@@ -278,7 +276,7 @@ func (p *GettyPackage) Unmarshal(buf *bytes.Buffer) (int, error) {
 	}
 
 	if int(packLen) > rpcPackagePlaceholderLen+gettyPackageHeaderLen {
-		if err := p.B.Unmarshal(p.H.CodecType, bytes.NewBuffer(buf.Next(int(packLen)-gettyPackageHeaderLen))); err != nil {
+		if err := p.B.Unmarshal(p.H.CodecType, bytes.NewBuffer(buf.Next(int(packLen)-rpcPackagePlaceholderLen-gettyPackageHeaderLen))); err != nil {
 			return 0, jerrors.Trace(err)
 		}
 	}
@@ -473,7 +471,6 @@ func (resp *GettyRPCResponse) Unmarshal(sz gettyCodecType, buf *bytes.Buffer) er
 	if err != nil {
 		return jerrors.Trace(err)
 	}
-
 	body := make([]byte, bodyLen)
 	err = binary.Read(buf, binary.LittleEndian, body)
 	if err != nil {
@@ -495,8 +492,6 @@ func (resp *GettyRPCResponse) Unmarshal(sz gettyCodecType, buf *bytes.Buffer) er
 }
 
 func (resp *GettyRPCResponse) GetBody() []byte {
-	gxlog.CWarn("resp body %p", resp.body)
-	time.Sleep(5e9)
 	return resp.body.([]byte)
 }
 

rpc/example/client/client.go

@@ -23,7 +23,7 @@ func main() {
 	for i := 0; i < 100; i++ {
 		go func() {
 			var res string
-			err := client.Call("TestRpc", "Test", data.TestABC{"aaa", "bbb", "ccc"}, &res)
+			err := client.Call("127.0.0.1:20000", "json", "TestRpc", "Test", data.TestABC{"aaa", "bbb", "ccc"}, &res)
 			if err != nil {
 				log.Error(err)
 				return
@@ -35,7 +35,7 @@ func main() {
 	for i := 0; i < 100; i++ {
 		go func() {
 			var result int
-			err := client.Call("TestRpc", "Add", 1, &result)
+			err := client.Call("127.0.0.1:20000", "json", "TestRpc", "Add", 1, &result)
 			if err != nil {
 				log.Error(err)
 				return
@@ -45,7 +45,7 @@ func main() {
 	}
 
 	var errInt int
-	err = client.Call("TestRpc", "Err", 2, &errInt)
+	err = client.Call("127.0.0.1:20000", "json", "TestRpc", "Err", 2, &errInt)
 	if err != nil {
 		log.Error(jerrors.ErrorStack(err))
 	}

rpc/readwriter.go

@@ -123,7 +123,9 @@ func (p *RpcClientPackageHandler) Read(ss getty.Session, data []byte) (interface
 	resp := &GettyRPCResponsePackage{
 		H:      pkg.H,
 		header: pkg.B.GetHeader().(GettyRPCResponseHeader),
-		body:   pkg.B.GetBody(),
+	}
+	if pkg.H.Command != gettyCmdHbResponse {
+		resp.body = pkg.B.GetBody()
 	}
 	return resp, length, nil
 }