[broker-15] implement message expiration and re-trying for QoS 1+ messages

Author: JavaSaBr

src/main/java/com/ss/mqtt/broker/config/MqttBrokerConfig.java

@@ -1,6 +1,6 @@
 package com.ss.mqtt.broker.config;
 
-import com.ss.mqtt.broker.handler.client.DeviceMqttClientReleaseHandler;
+import com.ss.mqtt.broker.handler.client.DefaultMqttClientReleaseHandler;
 import com.ss.mqtt.broker.handler.client.MqttClientReleaseHandler;
 import com.ss.mqtt.broker.handler.packet.in.*;
 import com.ss.mqtt.broker.handler.publish.in.PublishInHandler;
@@ -94,14 +94,16 @@ private interface ChannelFactory extends
         @NotNull ClientIdRegistry clientIdRegistry,
         @NotNull SubscriptionService subscriptionService,
         @NotNull PublishingService publishingService,
-        @NotNull MqttSessionService mqttSessionService
+        @NotNull MqttSessionService mqttSessionService,
+        @NotNull PublishRetryService publishRetryService
     ) {
 
         var handlers = new PacketInHandler[PacketType.INVALID.ordinal()];
         handlers[PacketType.CONNECT.ordinal()] = new ConnectInPacketHandler(
             clientIdRegistry,
             authenticationService,
-            mqttSessionService
+            mqttSessionService,
+            publishRetryService
         );
         handlers[PacketType.SUBSCRIBE.ordinal()] = new SubscribeInPacketHandler(subscriptionService);
         handlers[PacketType.UNSUBSCRIBE.ordinal()] = new UnsubscribeInPacketHandler(subscriptionService);
@@ -113,11 +115,12 @@ private interface ChannelFactory extends
     }
 
     @Bean
-    @NotNull MqttClientReleaseHandler deviceMqttClientReleaseHandler(
+    @NotNull MqttClientReleaseHandler defaultMqttClientReleaseHandler(
         @NotNull ClientIdRegistry clientIdRegistry,
-        @NotNull MqttSessionService mqttSessionService
+        @NotNull MqttSessionService mqttSessionService,
+        @NotNull PublishRetryService publishRetryService
     ) {
-        return new DeviceMqttClientReleaseHandler(clientIdRegistry, mqttSessionService);
+        return new DefaultMqttClientReleaseHandler(clientIdRegistry, mqttSessionService, publishRetryService);
     }
 
     @Bean
@@ -139,6 +142,14 @@ private interface ChannelFactory extends
         );
     }
 
+    @Bean
+    @NotNull PublishRetryService publishRetryService() {
+        return new DefaultPublishRetryService(
+            env.getProperty("publish.pending.check.interval", int.class, 60 * 1000),
+            env.getProperty("publish.retry.interval", int.class, 60 * 1000)
+        );
+    }
+
     @Bean
     @NotNull InetSocketAddress deviceNetworkAddress(
         @NotNull ServerNetwork<@NotNull MqttConnection> deviceNetwork,

src/main/java/com/ss/mqtt/broker/handler/client/AbstractMqttClientReleaseHandler.java

@@ -4,6 +4,7 @@
 import com.ss.mqtt.broker.network.client.AbstractMqttClient;
 import com.ss.mqtt.broker.service.ClientIdRegistry;
 import com.ss.mqtt.broker.service.MqttSessionService;
+import com.ss.mqtt.broker.service.PublishRetryService;
 import com.ss.rlib.common.util.StringUtils;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.log4j.Log4j2;
@@ -17,6 +18,7 @@ public abstract class AbstractMqttClientReleaseHandler<T extends AbstractMqttCli
 
     private final @NotNull ClientIdRegistry clientIdRegistry;
     private final @NotNull MqttSessionService sessionService;
+    private final @NotNull PublishRetryService publishRetryService;
 
     @Override
     public @NotNull Mono<?> release(@NotNull UnsafeMqttClient client) {
@@ -25,6 +27,7 @@ public abstract class AbstractMqttClientReleaseHandler<T extends AbstractMqttCli
     }
 
     protected @NotNull Mono<?> releaseImpl(@NotNull T client) {
+        publishRetryService.unregister(client);
 
         var clientId = client.getClientId();
         client.setClientId(StringUtils.EMPTY);

src/main/java/com/ss/mqtt/broker/handler/client/DefaultMqttClientReleaseHandler.java

@@ -0,0 +1,18 @@
+package com.ss.mqtt.broker.handler.client;
+
+import com.ss.mqtt.broker.network.client.DeviceMqttClient;
+import com.ss.mqtt.broker.service.ClientIdRegistry;
+import com.ss.mqtt.broker.service.MqttSessionService;
+import com.ss.mqtt.broker.service.PublishRetryService;
+import org.jetbrains.annotations.NotNull;
+
+public class DefaultMqttClientReleaseHandler extends AbstractMqttClientReleaseHandler<DeviceMqttClient> {
+
+    public DefaultMqttClientReleaseHandler(
+        @NotNull ClientIdRegistry clientIdRegistry,
+        @NotNull MqttSessionService sessionService,
+        @NotNull PublishRetryService publishRetryService
+    ) {
+        super(clientIdRegistry, sessionService, publishRetryService);
+    }
+}

src/main/java/com/ss/mqtt/broker/handler/client/DeviceMqttClientReleaseHandler.java

@@ -13,6 +13,7 @@
 import com.ss.mqtt.broker.service.AuthenticationService;
 import com.ss.mqtt.broker.service.ClientIdRegistry;
 import com.ss.mqtt.broker.service.MqttSessionService;
+import com.ss.mqtt.broker.service.PublishRetryService;
 import com.ss.rlib.common.util.StringUtils;
 import lombok.RequiredArgsConstructor;
 import org.jetbrains.annotations.NotNull;
@@ -21,9 +22,10 @@
 @RequiredArgsConstructor
 public class ConnectInPacketHandler extends AbstractPacketHandler<UnsafeMqttClient, ConnectInPacket> {
 
-    private final ClientIdRegistry clientIdRegistry;
-    private final AuthenticationService authenticationService;
-    private final MqttSessionService mqttSessionService;
+    private final @NotNull ClientIdRegistry clientIdRegistry;
+    private final @NotNull AuthenticationService authenticationService;
+    private final @NotNull MqttSessionService mqttSessionService;
+    private final @NotNull PublishRetryService publishRetryService;
 
     @Override
     protected void handleImpl(@NotNull UnsafeMqttClient client, @NotNull ConnectInPacket packet) {
@@ -135,6 +137,8 @@ private Mono<Boolean> onConnected(
             packet.getReceiveMax()
         ));
 
+        publishRetryService.register(client);
+
         return Mono.just(Boolean.TRUE);
     }
 

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

@@ -19,7 +19,7 @@ public void handle(@NotNull PublishInPacket packet, @NotNull Subscriber subscrib
             MqttPropertyConstants.PACKET_ID_FOR_QOS_0,
             QoS.AT_MOST_ONCE_DELIVERY,
             packet.isRetained(),
-            packet.isDuplicate(),
+            false,
             packet.getTopicName(),
             MqttPropertyConstants.TOPIC_ALIAS_NOT_SET,
             packet.getPayload(),

src/main/java/com/ss/mqtt/broker/handler/publish/out/Qos0PublishOutHandler.java

@@ -1,39 +1,70 @@
 package com.ss.mqtt.broker.handler.publish.out;
 
 import com.ss.mqtt.broker.model.MqttPropertyConstants;
+import com.ss.mqtt.broker.model.MqttSession;
 import com.ss.mqtt.broker.model.QoS;
 import com.ss.mqtt.broker.model.Subscriber;
+import com.ss.mqtt.broker.network.client.MqttClient;
+import com.ss.mqtt.broker.network.packet.HasPacketId;
 import com.ss.mqtt.broker.network.packet.in.PublishInPacket;
 import lombok.RequiredArgsConstructor;
 import org.jetbrains.annotations.NotNull;
 
 @RequiredArgsConstructor
-public class Qos1PublishOutHandler extends AbstractPublishOutHandler {
+public class Qos1PublishOutHandler extends AbstractPublishOutHandler implements MqttSession.PendingPacketHandler {
 
     @Override
     public void handle(@NotNull PublishInPacket packet, @NotNull Subscriber subscriber) {
 
         var client = subscriber.getMqttClient();
         var session = client.getSession();
-        var packetOutFactory = client.getPacketOutFactory();
 
-        var publish = packetOutFactory.newPublish(
+        // it means this client was already closed
+        if (session == null) {
+            return;
+        }
+
+        var packetId = session.nextPacketId();
+        session.registerPendingPublish(packet, this, packetId);
+
+        var packetOutFactory = client.getPacketOutFactory();
+        client.send(packetOutFactory.newPublish(
             client,
-            session.nextPacketId(),
+            packetId,
             QoS.AT_LEAST_ONCE_DELIVERY,
             packet.isRetained(),
-            packet.isDuplicate(),
+            false,
             packet.getTopicName(),
             MqttPropertyConstants.TOPIC_ALIAS_NOT_SET,
             packet.getPayload(),
             packet.isPayloadFormatIndicator(),
             packet.getResponseTopic(),
             packet.getCorrelationData(),
             packet.getUserProperties()
-        );
+        ));
+    }
 
-        session.registerPendingPublish(publish);
+    @Override
+    public boolean handleResponse(@NotNull MqttClient client, @NotNull HasPacketId response) {
+        return true;
+    }
 
-        client.send(publish);
+    @Override
+    public void retryAsync(@NotNull MqttClient client, @NotNull PublishInPacket packet, int packetId) {
+        var packetOutFactory = client.getPacketOutFactory();
+        client.send(packetOutFactory.newPublish(
+            client,
+            packetId,
+            QoS.AT_LEAST_ONCE_DELIVERY,
+            packet.isRetained(),
+            true,
+            packet.getTopicName(),
+            MqttPropertyConstants.TOPIC_ALIAS_NOT_SET,
+            packet.getPayload(),
+            packet.isPayloadFormatIndicator(),
+            packet.getResponseTopic(),
+            packet.getCorrelationData(),
+            packet.getUserProperties()
+        ));
     }
 }

src/main/java/com/ss/mqtt/broker/handler/publish/out/Qos1PublishOutHandler.java

@@ -26,6 +26,7 @@ public interface MqttPropertyConstants {
     boolean PAYLOAD_FORMAT_INDICATOR_DEFAULT = false;
 
     long MESSAGE_EXPIRY_INTERVAL_DEFAULT = 0;
+    long MESSAGE_EXPIRY_INTERVAL_UNDEFINED = -1;
 
     int TOPIC_ALIAS_MAXIMUM_UNDEFINED = -1;
     int TOPIC_ALIAS_MAXIMUM_DISABLED = 0;

src/main/java/com/ss/mqtt/broker/model/MqttPropertyConstants.java

@@ -2,8 +2,7 @@
 
 import com.ss.mqtt.broker.network.client.MqttClient;
 import com.ss.mqtt.broker.network.packet.HasPacketId;
-import com.ss.mqtt.broker.network.packet.in.MqttReadablePacket;
-import com.ss.mqtt.broker.network.packet.out.PublishOutPacket;
+import com.ss.mqtt.broker.network.packet.in.PublishInPacket;
 import org.jetbrains.annotations.NotNull;
 
 public interface MqttSession {
@@ -15,14 +14,14 @@ interface UnsafeMqttSession extends MqttSession {
         void clear();
     }
 
-    interface PendingCallback<T extends HasPacketId> {
-
-        @NotNull PendingCallback<?> EMPTY = (client, feedback) -> true;
+    interface PendingPacketHandler {
 
         /**
-         * @return true of pending packet can be removed.
+         * @return true if pending packet can be removed.
          */
-        boolean handle(@NotNull MqttClient client, @NotNull T feedback);
+        boolean handleResponse(@NotNull MqttClient client, @NotNull HasPacketId response);
+
+        void retryAsync(@NotNull MqttClient client, @NotNull PublishInPacket packet, int packetId);
     }
 
     @NotNull String getClientId();
@@ -34,15 +33,9 @@ interface PendingCallback<T extends HasPacketId> {
      */
     long getExpirationTime();
 
-    void registerPendingPublish(@NotNull PublishOutPacket publish);
-
-    <T extends MqttReadablePacket & HasPacketId> void registerPendingPublish(
-        @NotNull PublishOutPacket publish,
-        @NotNull MqttSession.PendingCallback<T> callback
-    );
+    void removeExpiredPackets();
+    void resendPendingPacketsAsync(@NotNull MqttClient client, int retryInterval);
 
-    <T extends MqttReadablePacket & HasPacketId> void unregisterPendingPacket(
-        @NotNull MqttClient client,
-        @NotNull T feedback
-    );
+    void registerPendingPublish(@NotNull PublishInPacket publish, @NotNull PendingPacketHandler handler, int packetId);
+    void unregisterPendingPacket(@NotNull MqttClient client, @NotNull HasPacketId response);
 }