Merge pull request #29 from JavaSaBr/feature-broker-25

[broker-25] Update publish re-try mechanism

Author: crazyrokr

build.gradle

@@ -32,7 +32,7 @@ allprojects {
 
     ext {
         annotationVersion = "17.0.0"
-        rlibVersion = "9.8.0"
+        rlibVersion = "9.9.0"
         lombokVersion = '1.18.4'
         springbootVersion = '2.2.0.RELEASE'
         springVersion = '5.1.6.RELEASE'
@@ -43,15 +43,17 @@ allprojects {
         projectReactorVersion = "3.3.0.RELEASE"
         byteBuddyVersion = "1.10.2"
         objenesisVersion = "3.1"
+        gsonVersion = "2.8.6"
     }
 
     dependencies {
 
         implementation "com.spaceshift:rlib.network:$rlibVersion"
         implementation "com.spaceshift:rlib.logger.slf4j:$rlibVersion"
         implementation "org.springframework.boot:spring-boot-starter:$springbootVersion"
-        implementation "org.springframework.boot:spring-boot-starter-logging:$springbootVersion"
+        implementation "org.springframework.boot:spring-boot-starter-log4j2:$springbootVersion"
         implementation "io.projectreactor:reactor-core:$projectReactorVersion"
+        implementation "com.google.code.gson:gson:$gsonVersion"
 
         compileOnly "org.jetbrains:annotations:$annotationVersion"
         compileOnly "org.projectlombok:lombok:$lombokVersion"
@@ -122,4 +124,5 @@ wrapper {
 
 configurations.each {
     it.exclude group: "org.slf4j", module: "slf4j-log4j12"
+    it.exclude group: "org.springframework.boot", module: "spring-boot-starter-logging"
 }

src/main/java/com/ss/mqtt/broker/MqttBrokerApplication.java

@@ -1,21 +1,18 @@
 package com.ss.mqtt.broker;
 
 import com.ss.mqtt.broker.config.MqttBrokerConfig;
-import com.ss.rlib.common.concurrent.util.ConcurrentUtils;
 import lombok.RequiredArgsConstructor;
 import org.jetbrains.annotations.NotNull;
 import org.springframework.boot.SpringApplication;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 
-@Import({
-    MqttBrokerConfig.class
-})
 @Configuration
 @RequiredArgsConstructor
+@Import(MqttBrokerConfig.class)
 public class MqttBrokerApplication {
 
     public static void main(@NotNull String[] args) {
-        ConcurrentUtils.wait(SpringApplication.run(MqttBrokerApplication.class, args));
+        SpringApplication.run(MqttBrokerApplication.class, args);
     }
 }

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

@@ -62,7 +62,7 @@ private interface ChannelFactory extends
         return new InMemoryClientIdRegistry(
             env.getProperty(
                 "client.id.available.chars",
-                "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ-"
+                "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ-_"
             ),
             env.getProperty("client.id.max.length", int.class, 36)
         );
@@ -94,16 +94,14 @@ private interface ChannelFactory extends
         @NotNull ClientIdRegistry clientIdRegistry,
         @NotNull SubscriptionService subscriptionService,
         @NotNull PublishingService publishingService,
-        @NotNull MqttSessionService mqttSessionService,
-        @NotNull PublishRetryService publishRetryService
+        @NotNull MqttSessionService mqttSessionService
     ) {
 
         var handlers = new PacketInHandler[PacketType.INVALID.ordinal()];
         handlers[PacketType.CONNECT.ordinal()] = new ConnectInPacketHandler(
             clientIdRegistry,
             authenticationService,
             mqttSessionService,
-            publishRetryService,
             subscriptionService
         );
         handlers[PacketType.SUBSCRIBE.ordinal()] = new SubscribeInPacketHandler(subscriptionService);
@@ -122,13 +120,11 @@ private interface ChannelFactory extends
     @NotNull MqttClientReleaseHandler defaultMqttClientReleaseHandler(
         @NotNull ClientIdRegistry clientIdRegistry,
         @NotNull MqttSessionService mqttSessionService,
-        @NotNull PublishRetryService publishRetryService,
         @NotNull SubscriptionService subscriptionService
     ) {
         return new DefaultMqttClientReleaseHandler(
             clientIdRegistry,
             mqttSessionService,
-            publishRetryService,
             subscriptionService
         );
     }
@@ -152,14 +148,6 @@ 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/factory/packet/out/Mqtt311PacketOutFactory.java

@@ -26,12 +26,11 @@ public class Mqtt311PacketOutFactory extends MqttPacketOutFactory {
         @NotNull byte[] authenticationData,
         @NotNull Array<StringPair> userProperties
     ) {
-        return new ConnectAck311OutPacket(client, reasonCode, sessionPresent);
+        return new ConnectAck311OutPacket(reasonCode, sessionPresent);
     }
 
     @Override
     public @NotNull PublishOutPacket newPublish(
-        @NotNull MqttClient client,
         int packetId,
         @NotNull QoS qos,
         boolean retained,
@@ -44,40 +43,44 @@ public class Mqtt311PacketOutFactory extends MqttPacketOutFactory {
         @NotNull byte[] correlationData,
         @NotNull Array<StringPair> userProperties
     ) {
-        return newPublish(client, packetId, qos, retained, duplicate, topicName, payload);
+        return new Publish311OutPacket(
+            packetId,
+            qos,
+            retained,
+            duplicate,
+            topicName,
+            payload
+        );
     }
 
     @Override
     public @NotNull MqttWritablePacket newPublishAck(
-        @NotNull MqttClient client,
         int packetId,
         @NotNull PublishAckReasonCode reasonCode,
         @NotNull String reason,
         @NotNull Array<StringPair> userProperties
     ) {
-        return new PublishAck311OutPacket(client, packetId);
+        return new PublishAck311OutPacket(packetId);
     }
 
     @Override
     public @NotNull MqttWritablePacket newSubscribeAck(
-        @NotNull MqttClient client,
         int packetId,
         @NotNull Array<SubscribeAckReasonCode> reasonCodes,
         @NotNull String reason,
         @NotNull Array<StringPair> userProperties
     ) {
-        return new SubscribeAck311OutPacket(client, packetId, reasonCodes);
+        return new SubscribeAck311OutPacket(reasonCodes, packetId);
     }
 
     @Override
     public @NotNull MqttWritablePacket newUnsubscribeAck(
-        @NotNull MqttClient client,
         int packetId,
         @NotNull Array<UnsubscribeAckReasonCode> reasonCodes,
         @NotNull Array<StringPair> userProperties,
         @NotNull String reason
     ) {
-        return new UnsubscribeAck311OutPacket(client, packetId);
+        return new UnsubscribeAck311OutPacket(packetId);
     }
 
     @Override
@@ -88,12 +91,11 @@ public class Mqtt311PacketOutFactory extends MqttPacketOutFactory {
         @NotNull String reason,
         @NotNull String serverReference
     ) {
-        return new Disconnect311OutPacket(client);
+        return new Disconnect311OutPacket();
     }
 
     @Override
     public @NotNull MqttWritablePacket newAuthenticate(
-        @NotNull MqttClient client,
         @NotNull AuthenticateReasonCode reasonCode,
         @NotNull String authenticateMethod,
         @NotNull byte[] authenticateData,
@@ -104,45 +106,42 @@ public class Mqtt311PacketOutFactory extends MqttPacketOutFactory {
     }
 
     @Override
-    public @NotNull MqttWritablePacket newPingRequest(@NotNull MqttClient client) {
-        return new PingRequest311OutPacket(client);
+    public @NotNull MqttWritablePacket newPingRequest() {
+        return new PingRequest311OutPacket();
     }
 
     @Override
-    public @NotNull MqttWritablePacket newPingResponse(@NotNull MqttClient client) {
-        return new PingResponse311OutPacket(client);
+    public @NotNull MqttWritablePacket newPingResponse() {
+        return new PingResponse311OutPacket();
     }
 
     @Override
     public @NotNull MqttWritablePacket newPublishRelease(
-        @NotNull MqttClient client,
         int packetId,
         @NotNull PublishReleaseReasonCode reasonCode,
         @NotNull Array<StringPair> userProperties,
         @NotNull String reason
     ) {
-        return new PublishRelease311OutPacket(client, packetId);
+        return new PublishRelease311OutPacket(packetId);
     }
 
     @Override
     public @NotNull MqttWritablePacket newPublishReceived(
-        @NotNull MqttClient client,
         int packetId,
         @NotNull PublishReceivedReasonCode reasonCode,
         @NotNull Array<StringPair> userProperties,
         @NotNull String reason
     ) {
-        return new PublishReceived311OutPacket(client, packetId);
+        return new PublishReceived311OutPacket(packetId);
     }
 
     @Override
     public @NotNull MqttWritablePacket newPublishCompleted(
-        @NotNull MqttClient client,
         int packetId,
         @NotNull PublishCompletedReasonCode reasonCode,
         @NotNull Array<StringPair> userProperties,
         @NotNull String reason
     ) {
-        return new PublishComplete311OutPacket(client, packetId);
+        return new PublishComplete311OutPacket(packetId);
     }
 }