[broker-25] update logging, mqtt mock client

Author: JavaSaBr

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'

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

@@ -1,21 +1,20 @@
 package com.ss.mqtt.broker;
 
 import com.ss.mqtt.broker.config.MqttBrokerConfig;
-import com.ss.rlib.common.concurrent.util.ConcurrentUtils;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.log4j.Log4j2;
 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
-})
+@Log4j2
 @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)
         );

src/main/java/com/ss/mqtt/broker/model/impl/DefaultMqttSession.java

@@ -150,7 +150,7 @@ public boolean hasInPending(int packetId) {
     @Override
     public void resendPendingPackets(@NotNull MqttClient mqttClient) {
         pendingOutPublishes.forEachInReadLock(mqttClient, (client, pending) -> {
-            log.info("Re-try to send publish {}", pending.publish);
+            log.debug("Re-try to send publish {}", pending.publish);
             pending.handler.resend(client, pending.publish, pending.packetId);
         });
     }

src/main/java/com/ss/mqtt/broker/model/topic/AbstractTopic.java

@@ -1,5 +1,6 @@
 package com.ss.mqtt.broker.model.topic;
 
+import com.ss.mqtt.broker.util.DebugUtils;
 import com.ss.rlib.common.util.ArrayUtils;
 import com.ss.rlib.common.util.StringUtils;
 import lombok.EqualsAndHashCode;
@@ -10,6 +11,10 @@
 @EqualsAndHashCode(of = "rawTopic")
 public abstract class AbstractTopic {
 
+    static {
+        DebugUtils.registerIncludedFields("rawTopic");
+    }
+
     static final String DELIMITER = "/";
     static final String MULTI_LEVEL_WILDCARD = "#";
     static final String SINGLE_LEVEL_WILDCARD = "+";

src/main/java/com/ss/mqtt/broker/network/client/AbstractMqttClient.java

@@ -9,6 +9,7 @@
 import com.ss.mqtt.broker.factory.packet.out.MqttPacketOutFactory;
 import com.ss.mqtt.broker.network.packet.in.MqttReadablePacket;
 import com.ss.mqtt.broker.network.packet.out.MqttWritablePacket;
+import com.ss.mqtt.broker.util.DebugUtils;
 import com.ss.rlib.common.util.StringUtils;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
@@ -24,9 +25,12 @@
 
 @Getter
 @Log4j2
-@ToString(of = "clientId")
 public abstract class AbstractMqttClient implements UnsafeMqttClient {
 
+    static {
+        DebugUtils.registerIncludedFields("clientId");
+    }
+
     protected final @NotNull MqttConnection connection;
     protected final MqttClientReleaseHandler releaseHandler;
     protected final AtomicBoolean released;
@@ -58,7 +62,7 @@ public AbstractMqttClient(@NotNull MqttConnection connection, @NotNull MqttClien
 
     @Override
     public void handle(@NotNull MqttReadablePacket packet) {
-        log.info("Handle received packet: {} : {}", packet.getName(), packet);
+        log.debug("Client [{}] received packet: {} : {}", clientId, packet.getName(), packet);
 
         var packetHandler = connection.getPacketHandlers()[packet.getPacketType()];
 
@@ -90,11 +94,13 @@ public void configure(
 
     @Override
     public void send(@NotNull MqttWritablePacket packet) {
+        log.debug("Send to client [{}] packet: {} : {}", clientId, packet.getName(), packet);
         connection.send(packet);
     }
 
     @Override
     public @NotNull CompletableFuture<Boolean> sendWithFeedback(@NotNull MqttWritablePacket packet) {
+        log.debug("Send to client [{}] packet: {} : {}", clientId, packet.getName(), packet);
         return connection.sendWithFeedback(packet);
     }
 
@@ -122,4 +128,9 @@ public void reject(@NotNull ConnectAckReasonCode reasonCode) {
             return Mono.empty();
         }
     }
+
+    @Override
+    public @NotNull String toString() {
+        return DebugUtils.toJsonString(this);
+    }
 }

src/main/java/com/ss/mqtt/broker/network/client/DeviceMqttClient.java

@@ -2,10 +2,15 @@
 
 import com.ss.mqtt.broker.network.MqttConnection;
 import com.ss.mqtt.broker.handler.client.MqttClientReleaseHandler;
+import com.ss.mqtt.broker.util.DebugUtils;
 import org.jetbrains.annotations.NotNull;
 
 public class DeviceMqttClient extends AbstractMqttClient {
 
+    static {
+        DebugUtils.registerIncludedFields("clientId");
+    }
+
     public DeviceMqttClient(@NotNull MqttConnection connection, @NotNull MqttClientReleaseHandler releaseHandler) {
         super(connection, releaseHandler);
     }

src/main/java/com/ss/mqtt/broker/network/packet/in/ConnectInPacket.java

@@ -27,6 +27,10 @@ public class ConnectInPacket extends MqttReadablePacket {
 
     private static final byte PACKET_TYPE = (byte) PacketType.CONNECT.ordinal();
 
+    static {
+        DebugUtils.registerIncludedFields("clientId", "keepAlive", "cleanStart");
+    }
+
     private static final Set<PacketProperty> AVAILABLE_PROPERTIES = EnumSet.of(
         /*
           If the Session Expiry Interval is absent the value 0 is used. If it is set to 0,
@@ -398,9 +402,4 @@ protected void applyProperty(@NotNull PacketProperty property, long value) {
                 unexpectedProperty(property);
         }
     }
-
-    @Override
-    public @NotNull String toString() {
-        return DebugUtils.toJsonString(this);
-    }
 }

src/main/java/com/ss/mqtt/broker/network/packet/in/DisconnectInPacket.java

@@ -23,6 +23,10 @@ public class DisconnectInPacket extends MqttReadablePacket {
 
     public static final byte PACKET_TYPE = (byte) PacketType.DISCONNECT.ordinal();
 
+    static {
+        DebugUtils.registerIncludedFields("reasonCode");
+    }
+
     private static final Set<PacketProperty> AVAILABLE_PROPERTIES = EnumSet.of(
         /*
           If the Session Expiry Interval is absent, the Session Expiry Interval in the CONNECT packet is used.
@@ -126,9 +130,4 @@ protected void applyProperty(@NotNull PacketProperty property, @NotNull String v
                 unexpectedProperty(property);
         }
     }
-
-    @Override
-    public @NotNull String toString() {
-        return DebugUtils.toJsonString(this);
-    }
 }

src/main/java/com/ss/mqtt/broker/network/packet/in/MqttReadablePacket.java

@@ -8,6 +8,7 @@
 import com.ss.mqtt.broker.model.PacketProperty;
 import com.ss.mqtt.broker.model.data.type.StringPair;
 import com.ss.mqtt.broker.network.MqttConnection;
+import com.ss.mqtt.broker.util.DebugUtils;
 import com.ss.mqtt.broker.util.MqttDataUtils;
 import com.ss.rlib.common.util.ArrayUtils;
 import com.ss.rlib.common.util.array.Array;
@@ -232,4 +233,9 @@ protected long readUnsignedInt(@NotNull ByteBuffer buffer) {
     protected void unexpectedProperty(@NotNull PacketProperty property) {
         throw new IllegalArgumentException("Unsupported property: " + property);
     }
+
+    @Override
+    public @NotNull String toString() {
+        return DebugUtils.toJsonString(this);
+    }
 }