Method readMessageAfterHeader

lib/protocol/protocol.go:515–567

(hdr *bep.Header, fourByteBuf []byte)

}

func (c *rawConnection) readMessageAfterHeader(hdr *bep.Header, fourByteBuf []byte) (proto.Message, error) {
	// First comes a 4 byte message length

	if _, err := io.ReadFull(c.cr, fourByteBuf[:4]); err != nil {
		return nil, fmt.Errorf("reading message length: %w", err)
	}
	msgLen := int32(binary.BigEndian.Uint32(fourByteBuf))
	if msgLen < 0 {
		return nil, fmt.Errorf("negative message length %d", msgLen)
	} else if msgLen > MaxMessageLen {
		return nil, fmt.Errorf("message length %d exceeds maximum %d", msgLen, MaxMessageLen)
	}

	// Then comes the message

	buf := BufferPool.Get(int(msgLen))
	defer BufferPool.Put(buf)

	if _, err := io.ReadFull(c.cr, buf); err != nil {
		return nil, fmt.Errorf("reading message: %w", err)
	}

	// ... which might be compressed

	switch hdr.Compression {
	case bep.MessageCompression_MESSAGE_COMPRESSION_NONE:
		// Nothing

	case bep.MessageCompression_MESSAGE_COMPRESSION_LZ4:
		decomp, err := lz4Decompress(buf)
		if err != nil {
			return nil, fmt.Errorf("decompressing message: %w", err)
		}
		buf = decomp

	default:
		return nil, fmt.Errorf("unknown message compression %d", hdr.Compression)
	}

	// ... and is then unmarshalled

	metricDeviceRecvDecompressedBytes.WithLabelValues(c.idString).Add(float64(4 + len(buf)))

	msg, err := newMessage(hdr.Type)
	if err != nil {
		return nil, err
	}
	if err := proto.Unmarshal(buf, msg); err != nil {
		return nil, fmt.Errorf("unmarshalling message: %w", err)
	}

	return msg, nil
}

func (c *rawConnection) readHeader(fourByteBuf []byte) (*bep.Header, error) {
	// First comes a 2 byte header length