[broker-25] update mqtt mock client and test with it

Author: JavaSaBr

build.gradle

@@ -50,7 +50,7 @@ allprojects {
         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"
 
         compileOnly "org.jetbrains:annotations:$annotationVersion"
@@ -113,4 +113,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/factory/packet/out/Mqtt311PacketOutFactory.java

@@ -43,7 +43,14 @@ public class Mqtt311PacketOutFactory extends MqttPacketOutFactory {
         @NotNull byte[] correlationData,
         @NotNull Array<StringPair> userProperties
     ) {
-        return newPublish(packetId, qos, retained, duplicate, topicName, payload);
+        return new Publish311OutPacket(
+            packetId,
+            qos,
+            retained,
+            duplicate,
+            topicName,
+            payload
+        );
     }
 
     @Override

src/main/java/com/ss/mqtt/broker/factory/packet/out/Mqtt5PacketOutFactory.java

@@ -74,8 +74,8 @@ public class Mqtt5PacketOutFactory extends Mqtt311PacketOutFactory {
             retained,
             duplicate,
             topicName,
-            topicAlias,
             payload,
+            topicAlias,
             stringPayload,
             responseTopic,
             correlationData,

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

@@ -20,8 +20,10 @@ public abstract class AbstractMqttClientReleaseHandler<T extends AbstractMqttCli
 
     @Override
     public @NotNull Mono<?> release(@NotNull UnsafeMqttClient client) {
+        var clientId = client.getClientId();
         //noinspection unchecked
-        return releaseImpl((T) client);
+        return releaseImpl((T) client)
+            .doOnNext(aVoid -> log.info("Client {} was released.", clientId));
     }
 
     protected @NotNull Mono<?> releaseImpl(@NotNull T client) {

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

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

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

@@ -55,12 +55,14 @@ protected byte getPacketFlags() {
     protected void writeVariableHeader(@NotNull ByteBuffer buffer) {
         // http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc384800412
         writeString(buffer, topicName);
-        writeShort(buffer, packetId);
+        if (qos.ordinal() > QoS.AT_MOST_ONCE_DELIVERY.ordinal()) {
+            writeShort(buffer, packetId);
+        }
     }
 
     @Override
     protected void writePayload(@NotNull ByteBuffer buffer) {
         // https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc384800413
-        writeBytes(buffer, payload);
+        buffer.put(payload);
     }
 }

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

@@ -148,8 +148,8 @@ public Publish5OutPacket(
         boolean retained,
         boolean duplicate,
         @NotNull String topicName,
-        int topicAlias,
         @NotNull byte[] payload,
+        int topicAlias,
         boolean stringPayload,
         @NotNull String responseTopic,
         @NotNull byte[] correlationData,

src/main/resources/application.properties

@@ -1 +1,2 @@
 authentication.allow.anonymous=false
+

src/main/resources/log4j2.xml

@@ -0,0 +1,13 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<Configuration status="INFO">
+  <Appenders>
+    <Console name="Console" target="SYSTEM_OUT">
+      <PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
+    </Console>
+  </Appenders>
+  <Loggers>
+    <Root level="INFO">
+      <AppenderRef ref="Console"/>
+    </Root>
+  </Loggers>
+</Configuration>

src/test/groovy/com/ss/mqtt/broker/test/integration/PublishRetryTest.groovy

@@ -1,23 +1,82 @@
 package com.ss.mqtt.broker.test.integration
 
+import com.hivemq.client.mqtt.datatypes.MqttQos
+import com.ss.mqtt.broker.model.QoS
+import com.ss.mqtt.broker.model.SubscribeTopicFilter
+import com.ss.mqtt.broker.model.reason.code.ConnectAckReasonCode
+import com.ss.mqtt.broker.model.reason.code.SubscribeAckReasonCode
 import com.ss.mqtt.broker.network.packet.in.ConnectAckInPacket
+import com.ss.mqtt.broker.network.packet.in.PublishInPacket
+import com.ss.mqtt.broker.network.packet.in.SubscribeAckInPacket
 import com.ss.mqtt.broker.network.packet.out.Connect311OutPacket
+import com.ss.mqtt.broker.network.packet.out.Subscribe311OutPacket
 import com.ss.mqtt.broker.service.MqttSessionService
+import com.ss.rlib.common.util.array.Array
 import org.springframework.beans.factory.annotation.Autowired
 
 class PublishRetryTest extends IntegrationSpecification {
 
     @Autowired
     MqttSessionService mqttSessionService
 
-    def "mqtt5 client should be generate session with one pending packet"() {
+    def "mqtt 3.1.1 client should be generate session with one pending packet"() {
         given:
-            def client = buildMqtt311MockClient()
-            def clientId = generateClientId()
+            def publisher = buildClient()
+            def subscriber = buildMqtt311MockClient()
+            def subscriberId = generateClientId()
         when:
-            client.connect()
-            client.send(new Connect311OutPacket(clientId, keepAlive))
+            
+            publisher.connect().join()
+        
+            subscriber.connect()
+            subscriber.send(new Connect311OutPacket(subscriberId, keepAlive))
+        
+            def connectAck = subscriber.readNext() as ConnectAckInPacket
+        
         then:
-            client.readNext() instanceof ConnectAckInPacket
+            connectAck.reasonCode == ConnectAckReasonCode.SUCCESS
+        when:
+           
+            subscriber.send(new Subscribe311OutPacket(
+                Array.of(new SubscribeTopicFilter("/test/retry/$subscriberId", QoS.AT_LEAST_ONCE_DELIVERY)),
+                1
+            ))
+        
+            def subscribeAck = subscriber.readNext() as SubscribeAckInPacket
+        
+        then:
+            subscribeAck.reasonCodes.stream()
+                .allMatch({ it == SubscribeAckReasonCode.GRANTED_QOS_1 })
+        when:
+           
+            publisher.publishWith()
+                .topic("/test/retry/$subscriberId")
+                .qos(MqttQos.AT_MOST_ONCE)
+                .payload(publishPayload)
+                .send()
+                .join()
+        
+            def receivedPublish = subscriber.readNext() as PublishInPacket
+        
+        then:
+            receivedPublish.payload == publishPayload
+        when:
+           
+            subscriber.close()
+        
+            subscriber.connect()
+            subscriber.send(new Connect311OutPacket(subscriberId, keepAlive))
+        
+            connectAck = subscriber.readNext() as ConnectAckInPacket
+            def receivedDupPublish = subscriber.readNext() as PublishInPacket
+        
+        then:
+            connectAck.reasonCode == ConnectAckReasonCode.SUCCESS
+            receivedDupPublish.duplicate
+            receivedDupPublish.packetId == receivedPublish.packetId
+            receivedDupPublish.payload == publishPayload
+        cleanup:
+            subscriber.close()
+            publisher.disconnect().join()
     }
 }