migrate to updated rlib

Author: JavaSaBr

application/src/main/java/javasabr/mqtt/application/config/MqttNetworkConfig.java

@@ -21,13 +21,14 @@
 import javasabr.rlib.network.ServerNetworkConfig.SimpleServerNetworkConfig;
 import javasabr.rlib.network.impl.DefaultBufferAllocator;
 import javasabr.rlib.network.server.ServerNetwork;
+import lombok.CustomLog;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.log4j.Log4j2;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.core.env.Environment;
 
-@Log4j2
+@CustomLog
 @Configuration
 @RequiredArgsConstructor
 public class MqttNetworkConfig {
@@ -45,7 +46,7 @@ ServerNetworkConfig internalNetworkConfig() {
         .pendingBufferSize(env.getProperty("mqtt.internal.network.pending.buffer.size", int.class, 4096))
         .writeBufferSize(env.getProperty("mqtt.internal.network.write.buffer.size", int.class, 2048))
         .threadGroupName("InternalNetwork")
-        .threadGroupSize(env.getProperty("mqtt.internal.network.thread.count", int.class, 1))
+        .threadGroupMaxSize(env.getProperty("mqtt.internal.network.thread.count", int.class, 1))
         .build();
   }
 
@@ -57,7 +58,7 @@ ServerNetworkConfig externalNetworkConfig() {
         .pendingBufferSize(env.getProperty("mqtt.external.network.pending.buffer.size", int.class, 200))
         .writeBufferSize(env.getProperty("mqtt.external.network.write.buffer.size", int.class, 100))
         .threadGroupName("ExternalNetwork")
-        .threadGroupSize(env.getProperty("mqtt.external.network.thread.count", int.class, 1))
+        .threadGroupMaxSize(env.getProperty("mqtt.external.network.thread.count", int.class, 1))
         .build();
   }
 
@@ -136,7 +137,7 @@ InetSocketAddress internalNetworkAddress(
   @Bean
   Consumer<MqttConnection> externalConnectionConsumer() {
     return mqttConnection -> {
-      log.info("Accepted external connection: {}", mqttConnection);
+      log.info(mqttConnection.remoteAddress(), "[%s] Accepted external connection"::formatted);
       var client = (UnsafeMqttClient) mqttConnection.client();
       mqttConnection.onReceive((conn, packet) -> client.handle((MqttReadablePacket) packet));
     };
@@ -145,7 +146,7 @@ Consumer<MqttConnection> externalConnectionConsumer() {
   @Bean
   Consumer<MqttConnection> internalConnectionConsumer() {
     return mqttConnection -> {
-      log.info("Accepted internal connection: {}", mqttConnection);
+      log.info(mqttConnection.remoteAddress(), "[%s] Accepted internal connection"::formatted);
       var client = (UnsafeMqttClient) mqttConnection.client();
       mqttConnection.onReceive((conn, packet) -> client.handle((MqttReadablePacket) packet));
     };

application/src/test/groovy/javasabr/mqtt/application/integration/ConnectSubscribePublishTest.groovy

@@ -24,22 +24,17 @@ class ConnectSubscribePublishTest extends IntegrationSpecification {
     when:
         subscriber.connect().join()
         publisher.connect().join()
-
         def subscribeResult = subscribe(subscriber, subscriberId, MqttQos.AT_MOST_ONCE, received)
         def publishResult = publish(publisher, subscriberId, MqttQos.AT_MOST_ONCE)
-
         Thread.sleep(100)
     then:
         noExceptionThrown()
-
         subscribeResult != null
         subscribeResult.returnCodes.contains(Mqtt3SubAckReturnCode.SUCCESS_MAXIMUM_QOS_0)
         subscribeResult.type == Mqtt3MessageType.SUBACK
-
         publishResult != null
         publishResult.qos == MqttQos.AT_MOST_ONCE
         publishResult.type == Mqtt3MessageType.PUBLISH
-
         received.get() != null
         received.get().qos == MqttQos.AT_MOST_ONCE
         received.get().type == Mqtt3MessageType.PUBLISH
@@ -57,22 +52,17 @@ class ConnectSubscribePublishTest extends IntegrationSpecification {
     when:
         subscriber.connect().join()
         publisher.connect().join()
-
         def subscribeResult = subscribe(subscriber, subscriberId, MqttQos.AT_MOST_ONCE, received)
         def publishResult = publish(publisher, subscriberId, MqttQos.AT_MOST_ONCE)
-
         Thread.sleep(100)
     then:
         noExceptionThrown()
-
         subscribeResult != null
         subscribeResult.reasonCodes.contains(Mqtt5SubAckReasonCode.GRANTED_QOS_0)
         subscribeResult.type == Mqtt5MessageType.SUBACK
-
         publishResult != null
         publishResult.publish.qos == MqttQos.AT_MOST_ONCE
         publishResult.publish.type == Mqtt5MessageType.PUBLISH
-
         received.get() != null
         received.get().qos == MqttQos.AT_MOST_ONCE
         received.get().type == Mqtt5MessageType.PUBLISH
@@ -90,22 +80,17 @@ class ConnectSubscribePublishTest extends IntegrationSpecification {
     when:
         subscriber.connect().join()
         publisher.connect().join()
-
         def subscribeResult = subscribe(subscriber, subscriberId, MqttQos.AT_LEAST_ONCE, received)
         def publishResult = publish(publisher, subscriberId, MqttQos.AT_LEAST_ONCE)
-
         Thread.sleep(100)
     then:
         noExceptionThrown()
-
         subscribeResult != null
         subscribeResult.returnCodes.contains(Mqtt3SubAckReturnCode.SUCCESS_MAXIMUM_QOS_1)
         subscribeResult.type == Mqtt3MessageType.SUBACK
-
         publishResult != null
         publishResult.qos == MqttQos.AT_LEAST_ONCE
         publishResult.type == Mqtt3MessageType.PUBLISH
-
         received.get() != null
         received.get().qos == MqttQos.AT_LEAST_ONCE
         received.get().type == Mqtt3MessageType.PUBLISH
@@ -121,25 +106,19 @@ class ConnectSubscribePublishTest extends IntegrationSpecification {
         def subscriberId = subscriber.getConfig().clientIdentifier.get().toString()
         def publisher = buildExternalMqtt5Client()
     when:
-
         subscriber.connect().join()
         publisher.connect().join()
-
         def subscribeResult = subscribe(subscriber, subscriberId, MqttQos.AT_LEAST_ONCE, received)
         def publishResult = publish(publisher, subscriberId, MqttQos.AT_LEAST_ONCE)
-
         Thread.sleep(100)
     then:
         noExceptionThrown()
-
         subscribeResult != null
         subscribeResult.reasonCodes.contains(Mqtt5SubAckReasonCode.GRANTED_QOS_1)
         subscribeResult.type == Mqtt5MessageType.SUBACK
-
         publishResult != null
         publishResult.publish.qos == MqttQos.AT_LEAST_ONCE
         publishResult.publish.type == Mqtt5MessageType.PUBLISH
-
         received.get() != null
         received.get().qos == MqttQos.AT_LEAST_ONCE
         received.get().type == Mqtt5MessageType.PUBLISH
@@ -157,28 +136,20 @@ class ConnectSubscribePublishTest extends IntegrationSpecification {
     when:
         subscriber.connect().join()
         publisher.connect().join()
-
         def subscribeResult = subscribe(subscriber, subscriberId, MqttQos.EXACTLY_ONCE, received)
         def publishResult = publish(publisher, subscriberId, MqttQos.EXACTLY_ONCE)
-
         Thread.sleep(100)
     then:
         noExceptionThrown()
-
         subscribeResult != null
         subscribeResult.returnCodes.contains(Mqtt3SubAckReturnCode.SUCCESS_MAXIMUM_QOS_2)
         subscribeResult.type == Mqtt3MessageType.SUBACK
-
         publishResult != null
         publishResult.qos == MqttQos.EXACTLY_ONCE
         publishResult.type == Mqtt3MessageType.PUBLISH
-
         received.get() != null
         received.get().qos == MqttQos.EXACTLY_ONCE
         received.get().type == Mqtt3MessageType.PUBLISH
-    cleanup:
-        subscriber.disconnect().join()
-        publisher.disconnect().join()
   }
 
   def "publisher should publish message QoS 2 using mqtt 5"() {
@@ -188,25 +159,19 @@ class ConnectSubscribePublishTest extends IntegrationSpecification {
         def subscriberId = subscriber.getConfig().clientIdentifier.get().toString()
         def publisher = buildExternalMqtt5Client()
     when:
-
         subscriber.connect().join()
         publisher.connect().join()
-
         def subscribeResult = subscribe(subscriber, subscriberId, MqttQos.EXACTLY_ONCE, received)
         def publishResult = publish(publisher, subscriberId, MqttQos.EXACTLY_ONCE)
-
         Thread.sleep(100)
     then:
         noExceptionThrown()
-
         subscribeResult != null
         subscribeResult.reasonCodes.contains(Mqtt5SubAckReasonCode.GRANTED_QOS_2)
         subscribeResult.type == Mqtt5MessageType.SUBACK
-
         publishResult != null
         publishResult.publish.qos == MqttQos.EXACTLY_ONCE
         publishResult.publish.type == Mqtt5MessageType.PUBLISH
-
         received.get() != null
         received.get().qos == MqttQos.EXACTLY_ONCE
         received.get().type == Mqtt5MessageType.PUBLISH
@@ -229,8 +194,7 @@ class ConnectSubscribePublishTest extends IntegrationSpecification {
       Mqtt5AsyncClient subscriber,
       String subscriberId,
       MqttQos qos,
-      AtomicReference<Mqtt5Publish> received
-  ) {
+      AtomicReference<Mqtt5Publish> received) {
     return subscriber.subscribeWith()
         .topicFilter("test/$subscriberId")
         .qos(qos)
@@ -252,8 +216,7 @@ class ConnectSubscribePublishTest extends IntegrationSpecification {
       Mqtt3AsyncClient subscriber,
       String subscriberId,
       MqttQos qos,
-      AtomicReference<Mqtt3Publish> received
-  ) {
+      AtomicReference<Mqtt3Publish> received) {
     return subscriber.subscribeWith()
         .topicFilter("test/$subscriberId")
         .qos(qos)

application/src/test/groovy/javasabr/mqtt/application/integration/IntegrationSpecification.groovy

@@ -70,6 +70,9 @@ class IntegrationSpecification extends Specification {
         .serverHost(address.getHostName())
         .serverPort(address.getPort())
         .useMqttVersion3()
+        .addDisconnectedListener {
+          println "[${clientId}|mqtt311] disconnected:[$it.cause]"
+        }
         .build()
         .toAsync()
   }
@@ -90,6 +93,9 @@ class IntegrationSpecification extends Specification {
         .serverHost(address.getHostName())
         .serverPort(address.getPort())
         .useMqttVersion5()
+        .addDisconnectedListener {
+          println "[${clientId}|mqtt5] disconnected:[$it.cause]"
+        }
         .build()
         .toAsync()
   }