feat(mqtt): add mqtt version configuration in client options

- Add setMqttVersion method to MqttSessionImpl - Remove redundant setMqttVersion call in MqttClient - Change mqttVersion field access to protected in AbstractSession - Remove setMqttVersion method from AbstractSession - Update MqttSpecComplianceTest to use mqtt version in client options - Clean up lambda expressions in MqttSpecComplianceTest subscribe calls - Add blank lines for better code readability in tests

Author: smthing

smart-mqtt-broker/src/main/java/tech/smartboot/mqtt/broker/MqttSessionImpl.java

@@ -334,4 +334,8 @@ public long getLatestReceiveMessageTime() {
  public void setLatestReceiveMessageTime(long latestReceiveMessageTime) {
  this.latestReceiveMessageTime = latestReceiveMessageTime;
  }
+
+ public final void setMqttVersion(MqttVersion mqttVersion) {
+ this.mqttVersion = mqttVersion;
+ }
 }

smart-mqtt-client/src/main/java/tech/smartboot/mqtt/client/MqttClient.java

@@ -131,7 +131,7 @@ public MqttClient(String uri, Consumer<Options> opt) {
  options.setPort(MqttUtil.toInt(array[2]));
  opt.accept(options);
  this.clientId = options.getClientId();
- setMqttVersion(options.getMqttVersion());
+ mqttVersion = options.getMqttVersion();
  }
 
 

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

@@ -38,7 +38,7 @@ public abstract class AbstractSession {
  protected boolean disconnect = false;
  protected MqttWriter mqttWriter;
 
- private MqttVersion mqttVersion;
+ protected MqttVersion mqttVersion;
 
  protected InflightQueue inflightQueue;
  private final Hashtable<Integer, MqttPublishMessage> ackMessageCacheMap = new Hashtable<>();
@@ -121,10 +121,6 @@ public final MqttVersion getMqttVersion() {
  return mqttVersion;
  }
 
- public final void setMqttVersion(MqttVersion mqttVersion) {
- this.mqttVersion = mqttVersion;
- }
-
  public final void setInflightQueue(InflightQueue inflightQueue) {
  this.inflightQueue = inflightQueue;
  }

smart-mqtt-test/src/test/java/org/smartboot/mqtt/test/MqttSpecComplianceTest.java

@@ -388,8 +388,7 @@ public void testSharedSubscription_MQTT5() throws InterruptedException, Executio
  int numSubscribers = 3;
  int numMessages = numSubscribers * 2; // Send enough messages to likely hit all subscribers
 
- MqttClient publisher = new MqttClient(host, port);
- publisher.setMqttVersion(MqttVersion.MQTT_5);
+ MqttClient publisher = new MqttClient(host, port, options -> options.setMqttVersion(MqttVersion.MQTT_5));
  CompletableFuture<Void> pubConnectFuture = new CompletableFuture<>();
  publisher.connect(connAck -> {
  if (connAck.getVariableHeader().connectReturnCode() == MqttConnectReturnCode.CONNECTION_ACCEPTED) {
@@ -407,9 +406,8 @@ public void testSharedSubscription_MQTT5() throws InterruptedException, Executio
  for (int i = 0; i < numSubscribers; i++) {
  int j = i;
  MqttClient subscriber = new MqttClient(host, port, options -> {
- options.setClientId("shared-sub-" + j);
+ options.setClientId("shared-sub-" + j).setMqttVersion(MqttVersion.MQTT_5);
  });
- subscriber.setMqttVersion(MqttVersion.MQTT_5);
  subscribers.add(subscriber);
  CompletableFuture<String> messageFuture = new CompletableFuture<>();
  messageFutures.add(messageFuture); // We expect each subscriber to get at least one message
@@ -631,7 +629,8 @@ public void testTopicWildcard_InvalidSubscription() throws InterruptedException,
 
  // Test wildcard in the middle of a topic level (invalid)
  CompletableFuture<Boolean> invalidWildcardFuture = new CompletableFuture<>();
- subscriber.subscribe("sport/+/tennis", MqttQoS.AT_MOST_ONCE, (client, msg) -> {}, (client, qos) -> {
+ subscriber.subscribe("sport/+/tennis", MqttQoS.AT_MOST_ONCE, (client, msg) -> {
+ }, (client, qos) -> {
  // This subscription should be rejected or handled properly
  invalidWildcardFuture.complete(true);
  });
@@ -645,7 +644,8 @@ public void testTopicWildcard_InvalidSubscription() throws InterruptedException,
 
  // Test hash wildcard not at the end (invalid)
  CompletableFuture<Boolean> invalidHashFuture = new CompletableFuture<>();
- subscriber.subscribe("sport/#/tennis", MqttQoS.AT_MOST_ONCE, (client, msg) -> {}, (client, qos) -> {
+ subscriber.subscribe("sport/#/tennis", MqttQoS.AT_MOST_ONCE, (client, msg) -> {
+ }, (client, qos) -> {
  invalidHashFuture.complete(true);
  });
 
@@ -667,7 +667,7 @@ public void testTopicWildcard_MixedSubscriptions() throws InterruptedException,
  // Test subscribing to both specific topic and wildcard topic
  String specificTopic = "sport/tennis/results";
  String wildcardTopic = "sport/tennis/+";
- 
+
  String matchingTopic1 = "sport/tennis/results";
  String matchingTopic2 = "sport/tennis/scores";
 
@@ -687,7 +687,7 @@ public void testTopicWildcard_MixedSubscriptions() throws InterruptedException,
  specificMessageFuture.complete(payload);
  }
  });
- 
+
  // Subscribe to wildcard topic
  subscriber.subscribe(wildcardTopic, MqttQoS.AT_MOST_ONCE, (client, msg) -> {
  String topic = msg.getVariableHeader().getTopicName();
@@ -697,7 +697,7 @@ public void testTopicWildcard_MixedSubscriptions() throws InterruptedException,
  }
  });
  });
- 
+
  Thread.sleep(500); // Allow subscriptions to complete
 
  // Publish to matching topics
@@ -736,11 +736,11 @@ public void testTopicWildcard_EmptyLevel() throws InterruptedException, Executio
  }
  });
  });
- 
+
  Thread.sleep(500); // Allow subscription to complete
 
  publisher.publish(matchingTopic, MqttQoS.AT_MOST_ONCE, "empty_level_payload".getBytes());
- 
+
  // Check if message with empty level is received
  try {
  Assert.assertEquals("empty_level_payload", messageFuture.get(5, TimeUnit.SECONDS));