🐛 修复 gitee #I6YOMD

mica-mqtt-codec/src/main/java/net/dreamlu/iot/mqtt/codec/MqttDecoder.java

@@ -86,10 +86,16 @@ public final class MqttDecoder {
 		if (mqttFixedHeader == null) {
 			return null;
 		}
+		// 包长度计算
+		int headLength = mqttFixedHeader.headLength();
 		int bytesRemainingInVariablePart = mqttFixedHeader.remainingLength();
+		int messageLength = headLength + bytesRemainingInVariablePart;
+		if (messageLength > maxBytesInMessage) {
+			throw new DecoderException("too large message: " + messageLength + " bytes but maxBytesInMessage is " + maxBytesInMessage);
+		}
 		// 3. 长度不够，直接返回 null
-		if (buffer.remaining() < bytesRemainingInVariablePart) {
-			ctx.setPacketNeededLength(buffer.position() + bytesRemainingInVariablePart);
+		if (readableLength < messageLength) {
+			ctx.setPacketNeededLength(messageLength);
 			return null;
 		}
 		// 4. 解析头信息
@@ -97,9 +103,6 @@ public final class MqttDecoder {
 		try {
 			Result<?> decodedVariableHeader = decodeVariableHeader(ctx, buffer, mqttFixedHeader, bytesRemainingInVariablePart);
 			variableHeader = decodedVariableHeader.value;
-			if (bytesRemainingInVariablePart > maxBytesInMessage) {
-				throw new DecoderException("too large message: " + bytesRemainingInVariablePart + " bytes but maxBytesInMessage is " + maxBytesInMessage);
-			}
 			bytesRemainingInVariablePart -= decodedVariableHeader.numberOfBytesConsumed;
 		} catch (Exception cause) {
 			return MqttMessageFactory.newInvalidMessage(mqttFixedHeader, variableHeader, cause);
@@ -149,7 +152,8 @@ public final class MqttDecoder {
 		if (loops == 4 && (digit & 128) != 0) {
 			throw new DecoderException("remaining length exceeds 4 digits (" + messageType + ')');
 		}
-		MqttFixedHeader decodedFixedHeader = new MqttFixedHeader(messageType, dupFlag, MqttQoS.valueOf(qosLevel), retain, remainingLength);
+		int headLength = 1 + loops;
+		MqttFixedHeader decodedFixedHeader = new MqttFixedHeader(messageType, dupFlag, MqttQoS.valueOf(qosLevel), retain, headLength, remainingLength);
 		return MqttCodecUtil.validateFixedHeader(ctx, MqttCodecUtil.resetUnusedFields(decodedFixedHeader));
 	}
 

mica-mqtt-codec/src/main/java/net/dreamlu/iot/mqtt/codec/MqttFixedHeader.java

@@ -22,7 +22,7 @@ import java.util.Objects;
  * See <a href="https://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#fixed-header">
  * MQTTV3.1/fixed-header</a>
  *
- * @author netty
+ * @author netty、L.cm
  */
 public final class MqttFixedHeader {
 
@@ -30,6 +30,7 @@ public final class MqttFixedHeader {
 	private final boolean isDup;
 	private MqttQoS qosLevel;
 	private final boolean isRetain;
+	private final int headLength;
 	private final int remainingLength;
 
 	public MqttFixedHeader(
@@ -38,10 +39,21 @@ public final class MqttFixedHeader {
 		MqttQoS qosLevel,
 		boolean isRetain,
 		int remainingLength) {
+		this(messageType, isDup, qosLevel, isRetain, 0, remainingLength);
+	}
+
+	public MqttFixedHeader(
+		MqttMessageType messageType,
+		boolean isDup,
+		MqttQoS qosLevel,
+		boolean isRetain,
+		int headLength,
+		int remainingLength) {
 		this.messageType = Objects.requireNonNull(messageType, "messageType is null.");
 		this.isDup = isDup;
 		this.qosLevel = Objects.requireNonNull(qosLevel, "qosLevel is null.");
 		this.isRetain = isRetain;
+		this.headLength = headLength;
 		this.remainingLength = remainingLength;
 	}
 
@@ -68,10 +80,18 @@ public final class MqttFixedHeader {
 		return isRetain;
 	}
 
+	public int headLength() {
+		return headLength;
+	}
+
 	public int remainingLength() {
 		return remainingLength;
 	}
 
+	public int getMessageLength() {
+		return headLength + remainingLength;
+	}
+
 	@Override
 	public String toString() {
 		return "MqttFixedHeader[" +