continue rework network part

JavaSaBr · JavaSaBr · commit a04e61710e47 · 2025-10-25T18:15:50.000+02:00
diff --git a/application/src/test/resources/log4j2.xml b/application/src/test/resources/log4j2.xml
@@ -9,6 +9,9 @@
     <Logger name="javasabr.mqtt.network.client.AbstractMqttClient" level="DEBUG" additivity="false">
       <AppenderRef ref="BrokerConsoleTest"/>
     </Logger>
+    <Logger name="javasabr.mqtt.service.impl.DefaultConnectionService" level="DEBUG" additivity="false">
+      <AppenderRef ref="BrokerConsoleTest"/>
+    </Logger>
     <Logger name="javasabr.mqtt.model.network.impl.DefaultMqttSession" level="DEBUG" additivity="false">
       <AppenderRef ref="BrokerConsoleTest"/>
     </Logger>
diff --git a/network/src/main/java/javasabr/mqtt/network/impl/DefaultMqttSession.java b/network/src/main/java/javasabr/mqtt/network/impl/DefaultMqttSession.java
@@ -64,7 +64,7 @@ private static void updatePendingPacket(
     }
 
     if (pendingPublish == null) {
-      log.warning(clientId , response, "Not found pending publish for client:[%s] by received packet:[%]"::formatted);
+      log.warning(clientId , response, "Not found pending publish for client:[%s] by received packet:[%s]"::formatted);
       return;
     }
 
diff --git a/network/src/main/java/javasabr/mqtt/network/impl/package-info.java b/network/src/main/java/javasabr/mqtt/network/impl/package-info.java
@@ -0,0 +1,4 @@
+@NullMarked
+package javasabr.mqtt.network.impl;
+
+import org.jspecify.annotations.NullMarked;
diff --git a/service/src/main/java/javasabr/mqtt/service/impl/DefaultConnectionService.java b/service/src/main/java/javasabr/mqtt/service/impl/DefaultConnectionService.java
@@ -56,6 +56,12 @@ protected void processReceivedMessage(
       return;
     }
 
+    log.debug(
+        connection.client().clientId(),
+        networkPacket.name(),
+        networkPacket,
+        "[%s] Received from client message:[%s] %s"::formatted);
+
     try {
       //noinspection DataFlowIssue
       inMessageHandlers[mrp.packetType()].processReceived(connection, mrp);
diff --git a/service/src/main/java/javasabr/mqtt/service/impl/DefaultPublishDeliveringService.java b/service/src/main/java/javasabr/mqtt/service/impl/DefaultPublishDeliveringService.java
@@ -48,7 +48,7 @@ public DefaultPublishDeliveringService(
   public PublishHandlingResult startDelivering(PublishInPacket publish, SingleSubscriber subscriber) {
     try {
       //noinspection DataFlowIssue
-      return publishOutMessageHandlers[publish.getQos().index()].handle(publish, subscriber);
+      return publishOutMessageHandlers[subscriber.getQos().index()].handle(publish, subscriber);
     } catch (IndexOutOfBoundsException | NullPointerException ex) {
       log.warning(publish, "Received not supported publish message:[%s]"::formatted);
       return PublishHandlingResult.UNSPECIFIED_ERROR;
diff --git a/service/src/main/java/javasabr/mqtt/service/message/handler/impl/DisconnectMqttInMessageHandler.java b/service/src/main/java/javasabr/mqtt/service/message/handler/impl/DisconnectMqttInMessageHandler.java
@@ -26,7 +26,7 @@ protected void processReceived(
       DisconnectInPacket networkPacket) {
     DisconnectReasonCode reasonCode = networkPacket.getReasonCode();
     if (reasonCode == DisconnectReasonCode.NORMAL_DISCONNECTION) {
-      log.info(client, "Disconnect client:[%s]"::formatted);
+      log.info(client.clientId(), "Disconnect client:[%s]"::formatted);
     } else {
       log.error("Disconnect client:[%s] by error reason:[%s]".formatted(client, reasonCode));
     }
diff --git a/service/src/main/java/javasabr/mqtt/service/message/handler/impl/PendingOutResponseMqttInMessageHandler.java b/service/src/main/java/javasabr/mqtt/service/message/handler/impl/PendingOutResponseMqttInMessageHandler.java
@@ -6,10 +6,10 @@
 import javasabr.mqtt.network.packet.HasPacketId;
 import javasabr.mqtt.network.packet.in.MqttReadablePacket;
 
-public abstract class PendingResponseMqttInMessageHandler<P extends MqttReadablePacket & HasPacketId>
+public abstract class PendingOutResponseMqttInMessageHandler<P extends MqttReadablePacket & HasPacketId>
     extends AbstractMqttInMessageHandler<ExternalMqttClient, P> {
 
-  protected PendingResponseMqttInMessageHandler(Class<P> expectedNetworkPacket) {
+  protected PendingOutResponseMqttInMessageHandler(Class<P> expectedNetworkPacket) {
     super(ExternalMqttClient.class, expectedNetworkPacket);
   }
 
diff --git a/service/src/main/java/javasabr/mqtt/service/message/handler/impl/PublishAckMqttInMessageHandler.java b/service/src/main/java/javasabr/mqtt/service/message/handler/impl/PublishAckMqttInMessageHandler.java
@@ -3,7 +3,7 @@
 import javasabr.mqtt.network.packet.MqttPacketType;
 import javasabr.mqtt.network.packet.in.PublishAckInPacket;
 
-public class PublishAckMqttInMessageHandler extends PendingResponseMqttInMessageHandler<PublishAckInPacket> {
+public class PublishAckMqttInMessageHandler extends PendingOutResponseMqttInMessageHandler<PublishAckInPacket> {
 
   public PublishAckMqttInMessageHandler() {
     super(PublishAckInPacket.class);
diff --git a/service/src/main/java/javasabr/mqtt/service/message/handler/impl/PublishCompleteMqttInMessageHandler.java b/service/src/main/java/javasabr/mqtt/service/message/handler/impl/PublishCompleteMqttInMessageHandler.java
@@ -3,7 +3,8 @@
 import javasabr.mqtt.network.packet.MqttPacketType;
 import javasabr.mqtt.network.packet.in.PublishCompleteInPacket;
 
-public class PublishCompleteMqttInMessageHandler extends PendingResponseMqttInMessageHandler<PublishCompleteInPacket>  {
+public class PublishCompleteMqttInMessageHandler extends
+    PendingOutResponseMqttInMessageHandler<PublishCompleteInPacket> {
 
   public PublishCompleteMqttInMessageHandler() {
     super(PublishCompleteInPacket.class);
diff --git a/service/src/main/java/javasabr/mqtt/service/message/handler/impl/PublishReceiveMqttInMessageHandler.java b/service/src/main/java/javasabr/mqtt/service/message/handler/impl/PublishReceiveMqttInMessageHandler.java
@@ -3,7 +3,8 @@
 import javasabr.mqtt.network.packet.MqttPacketType;
 import javasabr.mqtt.network.packet.in.PublishReceivedInPacket;
 
-public class PublishReceiveMqttInMessageHandler extends PendingResponseMqttInMessageHandler<PublishReceivedInPacket> {
+public class PublishReceiveMqttInMessageHandler extends
+    PendingOutResponseMqttInMessageHandler<PublishReceivedInPacket> {
 
   public PublishReceiveMqttInMessageHandler() {
     super(PublishReceivedInPacket.class);
diff --git a/service/src/main/java/javasabr/mqtt/service/message/handler/impl/PublishReleaseMqttInMessageHandler.java b/service/src/main/java/javasabr/mqtt/service/message/handler/impl/PublishReleaseMqttInMessageHandler.java
@@ -1,16 +1,31 @@
 package javasabr.mqtt.service.message.handler.impl;
 
+import javasabr.mqtt.network.MqttConnection;
+import javasabr.mqtt.network.MqttSession;
+import javasabr.mqtt.network.client.ExternalMqttClient;
 import javasabr.mqtt.network.packet.MqttPacketType;
 import javasabr.mqtt.network.packet.in.PublishReleaseInPacket;
 
-public class PublishReleaseMqttInMessageHandler extends PendingResponseMqttInMessageHandler<PublishReleaseInPacket> {
+public class PublishReleaseMqttInMessageHandler
+    extends AbstractMqttInMessageHandler<ExternalMqttClient, PublishReleaseInPacket> {
 
   public PublishReleaseMqttInMessageHandler() {
-    super(PublishReleaseInPacket.class);
+    super(ExternalMqttClient.class, PublishReleaseInPacket.class);
   }
 
   @Override
   public MqttPacketType messageType() {
     return MqttPacketType.PUBLISH_RELEASED;
   }
+
+  @Override
+  protected void processReceived(
+      MqttConnection connection,
+      ExternalMqttClient client,
+      PublishReleaseInPacket networkPacket) {
+    MqttSession session = client.session();
+    if (session != null) {
+      session.updateInPendingPacket(client, networkPacket);
+    }
+  }
 }
diff --git a/service/src/main/java/javasabr/mqtt/service/publish/handler/impl/AbstractMqttPublishInMessageHandler.java b/service/src/main/java/javasabr/mqtt/service/publish/handler/impl/AbstractMqttPublishInMessageHandler.java
@@ -39,20 +39,36 @@ protected void handleImpl(C client, PublishInPacket packet) {
       handleInvalidTopic(client, packet.getPacketId(), topicName);
       return;
     }
+
     Array<SingleSubscriber> subscribers = subscriptionService.findSubscribers(topicName);
     if (subscribers.isEmpty()) {
       handleEmptySubscriptions(client, packet.getPacketId(), topicName);
       return;
     }
+
     for (SingleSubscriber subscriber : subscribers) {
       PublishHandlingResult checkResult = checkSubscriber(client, packet, subscriber);
       if (checkResult.error()) {
         handleError(client, packet.getPacketId(), checkResult);
         return;
       }
     }
+
+    int count = 0;
+    PublishHandlingResult errorResult = null;
     for (SingleSubscriber subscriber : subscribers) {
-      startDelivering(client, packet, subscriber);
+      PublishHandlingResult result = startDelivering(client, packet, subscriber);
+      if (result.error()) {
+        errorResult = result;
+      } else if(result == PublishHandlingResult.SUCCESS) {
+        count++;
+      }
+    }
+
+    if (errorResult != null) {
+      handleError(client, packet.getPacketId(), errorResult);
+    } else {
+      handleSuccessfulResult(client, packet, count);
     }
   }
 
@@ -62,16 +78,16 @@ protected void handleEmptySubscriptions(C client, int messageId, TopicName topic
 
   protected void handleError(C client, int messageId, PublishHandlingResult handlingResult) {}
 
+  protected void handleSuccessfulResult(C client, PublishInPacket packet, int subscribers) {}
+
   protected PublishHandlingResult checkSubscriber(
       C client,
       PublishInPacket packet,
       SingleSubscriber subscriber) {
     return PublishHandlingResult.SUCCESS;
   }
 
-  protected void startDelivering(C client, PublishInPacket packet, SingleSubscriber subscriber) {
-    publishDeliveringService.startDelivering(packet, subscriber);
+  protected PublishHandlingResult startDelivering(C client, PublishInPacket packet, SingleSubscriber subscriber) {
+    return publishDeliveringService.startDelivering(packet, subscriber);
   }
-
-  protected void handleSuccessfulResult(C client, int messageId, int subscribers) {}
 }
diff --git a/service/src/main/java/javasabr/mqtt/service/publish/handler/impl/PersistedMqttPublishOutMessageHandler.java b/service/src/main/java/javasabr/mqtt/service/publish/handler/impl/PersistedMqttPublishOutMessageHandler.java
@@ -40,15 +40,14 @@ protected PublishHandlingResult handleImpl(PublishInPacket packet, ExternalMqttC
     if (session == null) {
       return PublishHandlingResult.SKIPPED;
     }
-    // generate new uniq packet id per client
-    var packetId = session.nextPacketId();
 
+    // generate new uniq packet id per client
+    int packetId = session.nextPacketId();
     // register waiting async response
     session.registerOutPublish(packet, pendingMessageHandler, packetId);
 
     // send publish
     startDelivering(client, packet, packetId, false);
-
     return PublishHandlingResult.SUCCESS;
   }
 
diff --git a/service/src/main/java/javasabr/mqtt/service/publish/handler/impl/Qos1MqttPublishInMessageHandler.java b/service/src/main/java/javasabr/mqtt/service/publish/handler/impl/Qos1MqttPublishInMessageHandler.java
@@ -4,6 +4,7 @@
 import javasabr.mqtt.model.reason.code.PublishAckReasonCode;
 import javasabr.mqtt.model.topic.TopicName;
 import javasabr.mqtt.network.client.ExternalMqttClient;
+import javasabr.mqtt.network.packet.in.PublishInPacket;
 import javasabr.mqtt.service.MessageOutFactoryService;
 import javasabr.mqtt.service.PublishDeliveringService;
 import javasabr.mqtt.service.SubscriptionService;
@@ -52,4 +53,12 @@ protected void handleError(ExternalMqttClient client, int messageId, PublishHand
         .resolveFactory(client)
         .newPublishAck(messageId, handlingResult.ackReasonCode()));
   }
+
+  @Override
+  protected void handleSuccessfulResult(ExternalMqttClient client, PublishInPacket packet, int subscribers) {
+    super.handleSuccessfulResult(client, packet, subscribers);
+    client.send(messageOutFactoryService
+        .resolveFactory(client)
+        .newPublishAck(packet.getPacketId(), PublishAckReasonCode.SUCCESS));
+  }
 }
diff --git a/service/src/main/java/javasabr/mqtt/service/publish/handler/impl/Qos2MqttPublishInMessageHandler.java b/service/src/main/java/javasabr/mqtt/service/publish/handler/impl/Qos2MqttPublishInMessageHandler.java
@@ -3,7 +3,6 @@
 import javasabr.mqtt.model.QoS;
 import javasabr.mqtt.model.reason.code.PublishCompletedReasonCode;
 import javasabr.mqtt.model.reason.code.PublishReceivedReasonCode;
-import javasabr.mqtt.model.subscriber.SingleSubscriber;
 import javasabr.mqtt.model.topic.TopicName;
 import javasabr.mqtt.network.MqttClient;
 import javasabr.mqtt.network.MqttSession;
@@ -82,15 +81,13 @@ protected void handleError(ExternalMqttClient client, int messageId, PublishHand
   }
 
   @Override
-  protected void startDelivering(ExternalMqttClient client, PublishInPacket packet, SingleSubscriber subscriber) {
-
+  protected void handleSuccessfulResult(ExternalMqttClient client, PublishInPacket packet, int subscribers) {
+    super.handleSuccessfulResult(client, packet, subscribers);
     MqttSession session = client.session();
     if (session == null) {
       return;
     }
-
     session.registerInPublish(packet, pendingMessageHandler, packet.getPacketId());
-    super.startDelivering(client, packet, subscriber);
     client.send(messageOutFactoryService
         .resolveFactory(client)
         .newPublishReceived(packet.getPacketId(), PublishReceivedReasonCode.SUCCESS));

Original file line number	Diff line number	Diff line change
`@@ -64,7 +64,7 @@ private static void updatePendingPacket(`
`64`	`64`	`}`
`65`	`65`
`66`	`66`	`if (pendingPublish == null) {`
`67`		`- log.warning(clientId , response, "Not found pending publish for client:[%s] by received packet:[%]"::formatted);`
	`67`	`+ log.warning(clientId , response, "Not found pending publish for client:[%s] by received packet:[%s]"::formatted);`
`68`	`68`	`return;`
`69`	`69`	`}`
`70`	`70`
Original file line number	Diff line number	Diff line change
`@@ -26,7 +26,7 @@ protected void processReceived(`
`26`	`26`	`DisconnectInPacket networkPacket) {`
`27`	`27`	`DisconnectReasonCode reasonCode = networkPacket.getReasonCode();`
`28`	`28`	`if (reasonCode == DisconnectReasonCode.NORMAL_DISCONNECTION) {`
`29`		`- log.info(client, "Disconnect client:[%s]"::formatted);`
	`29`	`+ log.info(client.clientId(), "Disconnect client:[%s]"::formatted);`
`30`	`30`	`} else {`
`31`	`31`	`log.error("Disconnect client:[%s] by error reason:[%s]".formatted(client, reasonCode));`
`32`	`32`	`}`