Fix:fix bug

protocol/dubbo/client.go

@@ -222,6 +222,7 @@ func (c *Client) call(ct CallType, addr string, svcUrl config.URL, method string
 	)
 	conn, session, err = c.selectSession(addr)
 	if err != nil || session == nil {
+		log.Warn(err)
 		return errSessionNotExist
 	}
 	defer c.pool.release(conn, err)
@@ -273,7 +274,6 @@ func (c *Client) transfer(session getty.Session, pkg *DubboPackage,
 	)
 
 	sequence = c.sequence.Add(1)
-	session.SetAttribute("seq", sequence) // store seq
 
 	if pkg == nil {
 		pkg = &DubboPackage{}

protocol/dubbo/codec.go

@@ -56,7 +56,7 @@ func (p *DubboPackage) Marshal() (*bytes.Buffer, error) {
 	return bytes.NewBuffer(pkg), nil
 }
 
-func (p *DubboPackage) Unmarshal(buf *bytes.Buffer) error {
+func (p *DubboPackage) Unmarshal(buf *bytes.Buffer, opts ...interface{}) error {
 	codec := hessian.NewHessianCodec(bufio.NewReader(buf))
 
 	// read header
@@ -65,6 +65,14 @@ func (p *DubboPackage) Unmarshal(buf *bytes.Buffer) error {
 		return jerrors.Trace(err)
 	}
 
+	if len(opts) != 0 { // for client
+		if client, ok := opts[0].(*Client); ok {
+			p.Body = client.pendingResponses[SequenceType(p.Header.ID)].reply
+		} else {
+			return fmt.Errorf("pendingResponses[%v] = nil", p.Header.ID)
+		}
+	}
+
 	if p.Header.Type&hessian.Heartbeat != 0x00 {
 		return nil
 	}

protocol/dubbo/readwriter.go

@@ -29,12 +29,10 @@ func NewRpcClientPackageHandler(client *Client) *RpcClientPackageHandler {
 func (p *RpcClientPackageHandler) Read(ss getty.Session, data []byte) (interface{}, int, error) {
 	p.client.pendingLock.RLock()
 	defer p.client.pendingLock.RUnlock()
-	pkg := &DubboPackage{
-		Body: p.client.pendingResponses[SequenceType(ss.GetAttribute("seq").(uint64))].reply,
-	}
+	pkg := &DubboPackage{}
 
 	buf := bytes.NewBuffer(data)
-	err := pkg.Unmarshal(buf)
+	err := pkg.Unmarshal(buf, p.client)
 	if err != nil {
 		pkg.Err = jerrors.Trace(err) // client will get this err
 		return pkg, len(data), nil