Merge remote-tracking branch 'origin/feature-broker-25' into feature-broker-26

Author: JavaSaBr

src/main/java/com/ss/mqtt/broker/handler/packet/in/ConnectInPacketHandler.java

@@ -140,7 +140,7 @@ private Mono<Boolean> onConnected(
             packet.isRequestProblemInformation()
         );
 
-        var response = client.getPacketOutFactory().newConnectAck(
+        var connectAck = client.getPacketOutFactory().newConnectAck(
             client,
             ConnectAckReasonCode.SUCCESS,
             sessionRestored,
@@ -152,17 +152,19 @@ private Mono<Boolean> onConnected(
 
         subscriptionService.restoreSubscriptions(client, session);
 
-        return Mono.fromFuture(client.sendWithFeedback(response)
-            .thenApply(result -> {
+        return Mono.fromFuture(client.sendWithFeedback(connectAck)
+            .thenApply(result -> onSentConnAck(client, session, result)));
+    }
+
+    private boolean onSentConnAck(@NotNull UnsafeMqttClient client, @NotNull MqttSession session, boolean result) {
 
-                if (!result) {
-                    log.warn("Was issue with sending conn ack packet to client {}", client.getClientId());
-                    return false;
-                }
+        if (!result) {
+            log.warn("Was issue with sending conn ack packet to client {}", client.getClientId());
+            return false;
+        }
 
-                session.resendPendingPackets(client);
-                return true;
-            }));
+        session.resendPendingPackets(client);
+        return true;
     }
 
     private boolean checkPacketException(@NotNull UnsafeMqttClient client, @NotNull ConnectInPacket packet) {

src/main/java/com/ss/mqtt/broker/network/packet/out/Connect311OutPacket.java

@@ -72,13 +72,9 @@ protected void writePayload(@NotNull ByteBuffer buffer) {
 
         // http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718031
         writeString(buffer, clientId);
-
-        // for MQTT 5
+       
         if (StringUtils.isNotEmpty(willTopic)) {
             appendWillProperties(buffer);
-        }
-
-        if (StringUtils.isNotEmpty(willTopic)) {
             writeString(buffer, willTopic);
             writeBytes(buffer, willPayload);
         }