refactor(mqtt): update decode methods to include MQTT version parameter

- Modified decodeVariableHeader0 methods across multiple MQTT message classes to accept MqttVersion - Removed deprecated setVersion method from MqttMessage class - Updated MqttVariableMessage and OnlyFixedHeaderMessage to pass MqttVersion during decoding - Adjusted AbstractSession and MqttHeaderDecoder to remove obsolete version setting logic - Ensured consistent handling of MQTT protocol versions during message decoding - Added missing import for MqttVersion in relevant classes

Author: smthing

smart-mqtt-common/src/main/java/tech/smartboot/mqtt/common/AbstractSession.java

@@ -72,7 +72,6 @@ public final void notifyPubComp(int packetId) {
  public void write(MqttMessage mqttMessage, boolean autoFlush) {
  ValidateUtils.isTrue(!disconnect, "已断开连接,无法发送消息");
  try {
- mqttMessage.setVersion(mqttVersion);
  synchronized (mqttWriter) {
  if (disconnect) {
  ValidateUtils.throwException("session is disconnect");

smart-mqtt-common/src/main/java/tech/smartboot/mqtt/common/MqttHeaderDecoder.java

@@ -83,7 +83,6 @@ public Decoder decode(ByteBuffer buffer, AbstractSession session) {
 
  MqttFixedHeader mqttFixedHeader = MqttFixedHeader.getInstance(messageType, dupFlag, qosLevel, retain);
  session.mqttMessage = newMessage(mqttFixedHeader);
- session.mqttMessage.setVersion(session.getMqttVersion());
  session.mqttMessage.setRemainingLength(remainingLength);
  return mqttPayloadDecoder.decode(buffer, session);
  }

smart-mqtt-common/src/main/java/tech/smartboot/mqtt/common/MqttPayloadDecoder.java

@@ -46,7 +46,7 @@ public Decoder decode(ByteBuffer buffer, AbstractSession session) {
  return this;
  }
  int p = payloadBuffer.position();
- session.mqttMessage.decodeVariableHeader(payloadBuffer);
+ session.mqttMessage.decodeVariableHeader(payloadBuffer, session.getMqttVersion());
  session.mqttMessage.decodePlayLoad(payloadBuffer);
  ValidateUtils.isTrue((payloadBuffer.position() - p) == remainingLength, "Payload size is wrong");
  session.disposableBuffer = null;

smart-mqtt-common/src/main/java/tech/smartboot/mqtt/common/message/MqttConnAckMessage.java

@@ -34,7 +34,7 @@ public MqttConnAckMessage(MqttConnAckVariableHeader mqttConnAckVariableHeader) {
  }
 
  @Override
- public void decodeVariableHeader0(ByteBuffer buffer) {
+ public void decodeVariableHeader0(ByteBuffer buffer, MqttVersion version) {
  final boolean sessionPresent = (buffer.get() & 0x01) == 0x01;
  byte returnCode = buffer.get();
 

smart-mqtt-common/src/main/java/tech/smartboot/mqtt/common/message/MqttConnectMessage.java

@@ -33,6 +33,7 @@ public class MqttConnectMessage extends MqttVariableMessage<MqttConnectVariableH
  * 有效载荷
  */
  private MqttConnectPayload mqttConnectPayload;
+ private MqttVersion version;
 
  public MqttConnectMessage(MqttFixedHeader mqttFixedHeader) {
  super(mqttFixedHeader);
@@ -45,7 +46,7 @@ public MqttConnectMessage(MqttConnectVariableHeader mqttConnectVariableHeader, M
  }
 
  @Override
- public void decodeVariableHeader0(ByteBuffer buffer) {
+ public void decodeVariableHeader0(ByteBuffer buffer, MqttVersion mqttVersion) {
  //协议名
  //协议名是表示协议名 MQTT 的 UTF-8 编码的字符串。
  //MQTT 规范的后续版本不会改变这个字符串的偏移和长度。

smart-mqtt-common/src/main/java/tech/smartboot/mqtt/common/message/MqttDisconnectMessage.java

@@ -29,7 +29,7 @@ public MqttDisconnectMessage(MqttDisconnectVariableHeader mqttConnAckVariableHea
  }
 
  @Override
- protected void decodeVariableHeader0(ByteBuffer buffer) {
+ protected void decodeVariableHeader0(ByteBuffer buffer, final MqttVersion version) {
  if (version == MqttVersion.MQTT_5) {
  byte returnCode = buffer.get();
  DisConnectProperties properties = new DisConnectProperties();

smart-mqtt-common/src/main/java/tech/smartboot/mqtt/common/message/MqttMessage.java

@@ -40,7 +40,6 @@ protected void writeTo(MqttWriter mqttWriter) {
  * 固定报头
  */
  protected final MqttFixedHeader fixedHeader;
- protected MqttVersion version;
  /**
  * 剩余长度
  */
@@ -59,7 +58,7 @@ public final MqttFixedHeader getFixedHeader() {
  *
  * @param buffer
  */
- public abstract void decodeVariableHeader(ByteBuffer buffer);
+ public abstract void decodeVariableHeader(ByteBuffer buffer, MqttVersion mqttVersion);
 
  public void decodePlayLoad(ByteBuffer buffer) {
 
@@ -110,8 +109,4 @@ public int getRemainingLength() {
  public void setRemainingLength(int remainingLength) {
  this.remainingLength = remainingLength;
  }
-
- public final void setVersion(MqttVersion version) {
- this.version = version;
- }
 }

smart-mqtt-common/src/main/java/tech/smartboot/mqtt/common/message/MqttPubQosMessage.java

@@ -32,7 +32,7 @@ public MqttPubQosMessage(MqttFixedHeader pubRecHeader, MqttPubQosVariableHeader
  }
 
  @Override
- protected final void decodeVariableHeader0(ByteBuffer buffer) {
+ protected final void decodeVariableHeader0(ByteBuffer buffer, MqttVersion version) {
  int packetId = decodeMessageId(buffer);
  MqttPubQosVariableHeader header = null;
  if (version == MqttVersion.MQTT_5) {

smart-mqtt-common/src/main/java/tech/smartboot/mqtt/common/message/MqttPublishMessage.java

@@ -37,7 +37,7 @@ public MqttPublishMessage(MqttFixedHeader mqttFixedHeader, MqttPublishVariableHe
  }
 
  @Override
- public void decodeVariableHeader0(ByteBuffer buffer) {
+ public void decodeVariableHeader0(ByteBuffer buffer, final MqttVersion version) {
  final String topic = TopicByteTree.DEFAULT_INSTANCE.search(buffer);
  int packetId = -1;
  //只有当 QoS 等级是 1 或 2 时，报文标识符（Packet Identifier）字段才能出现在 PUBLISH 报文中。

smart-mqtt-common/src/main/java/tech/smartboot/mqtt/common/message/MqttSubAckMessage.java

@@ -35,7 +35,7 @@ public MqttSubAckMessage(MqttReasonVariableHeader variableHeader) {
  }
 
  @Override
- protected void decodeVariableHeader0(ByteBuffer buffer) {
+ protected void decodeVariableHeader0(ByteBuffer buffer, final MqttVersion version) {
  int packetId = decodeMessageId(buffer);
  MqttReasonVariableHeader header;
  if (version == MqttVersion.MQTT_5) {