Fixed bug

Author: dobyte

internal/transporter/internal/client/conn.go

@@ -159,18 +159,15 @@ func (c *Conn) read(conn net.Conn) {
 
 			c.lastHeartbeatTime.Store(xtime.Now().Unix())
 
-			isHeartbeat, _, seq := protocol.ParseBuffer(buf)
-
-			if isHeartbeat {
-				continue
-			}
-
-			call, ok := c.pending.extract(seq)
-			if !ok {
-				continue
+			if isHeartbeat, _, seq := protocol.ParseBuffer(buf.Bytes()); isHeartbeat {
+				buf.Release()
+			} else {
+				if call, ok := c.pending.extract(seq); ok {
+					call <- buf
+				} else {
+					buf.Release()
+				}
 			}
-
-			call <- buf
 		}
 	}
 }

internal/transporter/internal/protocol/reader.go

@@ -9,41 +9,30 @@ import (
 )
 
 // ReadMessage 读取消息
-func ReadMessage(reader io.Reader) (isHeartbeat bool, route uint8, seq uint64, data []byte, err error) {
+func ReadMessage(reader io.Reader) (bool, uint8, uint64, []byte, error) {
 	buf := buffer.MallocBytes(defaultSizeBytes)
 	defer buf.Release()
 
-	if _, err = io.ReadFull(reader, buf.Bytes()); err != nil {
-		return
+	if _, err := io.ReadFull(reader, buf.Bytes()); err != nil {
+		return false, 0, 0, nil, err
 	}
 
 	size := binary.BigEndian.Uint32(buf.Bytes())
 
 	if size == 0 {
-		err = errors.ErrInvalidMessage
-		return
+		return false, 0, 0, nil, errors.ErrInvalidMessage
 	}
 
-	data = make([]byte, defaultSizeBytes+size)
+	data := make([]byte, defaultSizeBytes+size)
 	copy(data[:defaultSizeBytes], buf.Bytes())
 
-	if _, err = io.ReadFull(reader, data[defaultSizeBytes:]); err != nil {
-		return
-	}
-
-	header := data[defaultSizeBytes : defaultSizeBytes+defaultHeaderBytes][0]
-
-	isHeartbeat = header&heartbeatBit == heartbeatBit
-
-	if isHeartbeat {
-		return
+	if _, err := io.ReadFull(reader, data[defaultSizeBytes:]); err != nil {
+		return false, 0, 0, nil, err
 	}
 
-	route = data[defaultSizeBytes+defaultHeaderBytes : defaultSizeBytes+defaultHeaderBytes+defaultRouteBytes][0]
+	isHeartbeat, route, seq := ParseBuffer(data)
 
-	seq = binary.BigEndian.Uint64(data[defaultSizeBytes+defaultHeaderBytes+defaultRouteBytes : defaultSizeBytes+defaultHeaderBytes+defaultRouteBytes+8])
-
-	return
+	return isHeartbeat, route, seq, data, nil
 }
 
 // ReadBuffer 以buffer的形式读取消息
@@ -75,18 +64,15 @@ func ReaderBuffer(reader io.Reader) (buffer.Buffer, error) {
 }
 
 // ParseBuffer 解析buffer
-func ParseBuffer(buf buffer.Buffer) (isHeartbeat bool, route uint8, seq uint64) {
-	data := buf.Bytes()
-	header := data[defaultSizeBytes+defaultHeaderBytes : defaultSizeBytes+defaultHeaderBytes+defaultRouteBytes][0]
-
-	isHeartbeat = header&heartbeatBit == heartbeatBit
-
-	if isHeartbeat {
-		return
+func ParseBuffer(data []byte) (bool, uint8, uint64) {
+	if header := data[defaultSizeBytes : defaultSizeBytes+defaultHeaderBytes][0]; header&heartbeatBit == heartbeatBit {
+		return true, 0, 0
+	} else {
+		var (
+			route = data[defaultSizeBytes+defaultHeaderBytes : defaultSizeBytes+defaultHeaderBytes+defaultRouteBytes][0]
+			seq   = binary.BigEndian.Uint64(data[defaultSizeBytes+defaultHeaderBytes+defaultRouteBytes : defaultSizeBytes+defaultHeaderBytes+defaultRouteBytes+8])
+		)
+
+		return false, route, seq
 	}
-
-	route = data[defaultSizeBytes+defaultHeaderBytes : defaultSizeBytes+defaultHeaderBytes+defaultRouteBytes][0]
-	seq = binary.BigEndian.Uint64(data[defaultSizeBytes+defaultHeaderBytes+defaultRouteBytes : defaultSizeBytes+defaultHeaderBytes+defaultRouteBytes+8])
-
-	return
 }