#69 Improve publish delivering to subscribers part 5 (#139)

JavaSaBr · web-flow · commit a9902009ac2c · 2025-12-14T10:46:40.000+01:00
diff --git a/application/src/main/java/javasabr/mqtt/broker/application/config/MqttBrokerSpringConfig.java b/application/src/main/java/javasabr/mqtt/broker/application/config/MqttBrokerSpringConfig.java
@@ -14,6 +14,7 @@
 import javasabr.mqtt.network.user.NetworkMqttUserFactory;
 import javasabr.mqtt.service.AuthorizationService;
 import javasabr.mqtt.service.AuthenticationService;
+import javasabr.mqtt.service.AuthorizationService;
 import javasabr.mqtt.service.ClientIdRegistry;
 import javasabr.mqtt.service.ConnectionService;
 import javasabr.mqtt.service.CredentialSource;
@@ -254,24 +255,20 @@ ConnectionService externalMqttConnectionService(Collection<? extends MqttInMessa
   }
 
   @Bean
-  MqttPublishOutMessageHandler qos0MqttPublishOutMessageHandler(
-      SubscriptionService subscriptionService,
-      MessageOutFactoryService messageOutFactoryService) {
-    return new Qos0MqttPublishOutMessageHandler(subscriptionService, messageOutFactoryService);
+  MqttPublishOutMessageHandler qos0MqttPublishOutMessageHandler(MessageOutFactoryService messageOutFactoryService) {
+    return new Qos0MqttPublishOutMessageHandler(messageOutFactoryService);
   }
 
   @Bean
   MqttPublishOutMessageHandler qos1MqttPublishOutMessageHandler(
-      SubscriptionService subscriptionService,
       MessageOutFactoryService messageOutFactoryService) {
-    return new Qos1MqttPublishOutMessageHandler(subscriptionService, messageOutFactoryService);
+    return new Qos1MqttPublishOutMessageHandler(messageOutFactoryService);
   }
 
   @Bean
   MqttPublishOutMessageHandler qos2MqttPublishOutMessageHandler(
-      SubscriptionService subscriptionService,
       MessageOutFactoryService messageOutFactoryService) {
-    return new Qos2MqttPublishOutMessageHandler(subscriptionService, messageOutFactoryService);
+    return new Qos2MqttPublishOutMessageHandler(messageOutFactoryService);
   }
 
   @Bean
diff --git a/core-service/src/main/java/javasabr/mqtt/service/PublishDeliveringService.java b/core-service/src/main/java/javasabr/mqtt/service/PublishDeliveringService.java
@@ -2,9 +2,8 @@
 
 import javasabr.mqtt.model.publishing.Publish;
 import javasabr.mqtt.model.subscriber.SingleSubscriber;
-import javasabr.mqtt.service.publish.handler.PublishHandlingResult;
 
 public interface PublishDeliveringService {
 
-  PublishHandlingResult startDelivering(Publish publish, SingleSubscriber subscriber);
+  void startDelivering(Publish publish, SingleSubscriber subscriber);
 }
diff --git a/core-service/src/main/java/javasabr/mqtt/service/impl/DefaultPublishDeliveringService.java b/core-service/src/main/java/javasabr/mqtt/service/impl/DefaultPublishDeliveringService.java
@@ -6,7 +6,6 @@
 import javasabr.mqtt.model.subscriber.SingleSubscriber;
 import javasabr.mqtt.service.PublishDeliveringService;
 import javasabr.mqtt.service.publish.handler.MqttPublishOutMessageHandler;
-import javasabr.mqtt.service.publish.handler.PublishHandlingResult;
 import lombok.AccessLevel;
 import lombok.CustomLog;
 import lombok.experimental.FieldDefaults;
@@ -45,13 +44,12 @@ public DefaultPublishDeliveringService(
   }
 
   @Override
-  public PublishHandlingResult startDelivering(Publish publish, SingleSubscriber subscriber) {
+  public void startDelivering(Publish publish, SingleSubscriber subscriber) {
     try {
       //noinspection DataFlowIssue
-      return publishOutMessageHandlers[subscriber.qos().level()].handle(publish, subscriber);
+      publishOutMessageHandlers[subscriber.qos().level()].handle(publish, subscriber);
     } catch (IndexOutOfBoundsException | NullPointerException ex) {
       log.warning(publish, "Received not supported publish message:[%s]"::formatted);
-      return PublishHandlingResult.UNSPECIFIED_ERROR;
     }
   }
 
diff --git a/core-service/src/main/java/javasabr/mqtt/service/impl/DefaultPublishReceivingService.java b/core-service/src/main/java/javasabr/mqtt/service/impl/DefaultPublishReceivingService.java
@@ -54,8 +54,7 @@ public void processPublish(NetworkMqttUser user, Publish publish) {
       log.warning(user.clientId(), publish, "[%s] Received not supported publish:%s"::formatted);
     }
   }
-
-
+  
   private static String buildServiceDescription(
       @Nullable MqttPublishInMessageHandler[] publishInMessageHandlers) {
     var builder = new StringBuilder();
diff --git a/core-service/src/main/java/javasabr/mqtt/service/message/handler/impl/AbstractMqttInMessageHandler.java b/core-service/src/main/java/javasabr/mqtt/service/message/handler/impl/AbstractMqttInMessageHandler.java
@@ -35,7 +35,7 @@ protected boolean requireSession() {
   }
 
   @Override
-  public void processValidMessage(MqttConnection connection, MqttInMessage mqttInMessage) {
+  public final void processValidMessage(MqttConnection connection, MqttInMessage mqttInMessage) {
     NetworkMqttUser user = connection.user();
     if (!expectedUser.isInstance(user)) {
       log.warning(user, "Received not expected user:[%s]"::formatted);
@@ -60,7 +60,7 @@ public void processValidMessage(MqttConnection connection, MqttInMessage mqttInM
   }
 
   @Override
-  public void processInvalidMessage(MqttConnection connection, MqttInMessage mqttInMessage) {
+  public final void processInvalidMessage(MqttConnection connection, MqttInMessage mqttInMessage) {
     NetworkMqttUser user = connection.user();
     if (!expectedUser.isInstance(user)) {
       log.warning(user, "Received not expected user:[%s]"::formatted);
diff --git a/core-service/src/main/java/javasabr/mqtt/service/message/handler/impl/PublishMqttInMessageHandler.java b/core-service/src/main/java/javasabr/mqtt/service/message/handler/impl/PublishMqttInMessageHandler.java
@@ -55,35 +55,35 @@ protected void processMessageWithValidFields(
       MqttConnection connection,
       ExternalNetworkMqttUser user,
       NetworkMqttSession session,
-      PublishMqttInMessage publishMessage) {
+      PublishMqttInMessage message) {
     
-    TopicName finalTopicName = resolveFinalTopicName(user, session, publishMessage);
+    TopicName finalTopicName = resolveFinalTopicName(user, session, message);
     if (finalTopicName == null) {
       return;
     } else if (!authorizationService.authorizePublish(user, finalTopicName)) {
       handleNotAuthorize(user);
       return;
     }
 
-    byte[] payload = publishMessage.payload();
-    TopicName responseTopicName = resolveResponseTopic(user, publishMessage);
+    byte[] payload = message.payload();
+    TopicName responseTopicName = resolveResponseTopic(user, message);
 
     //noinspection DataFlowIssue everything is already validated
     Publish publish = new Publish(
-        publishMessage.messageId(),
-        publishMessage.qos(),
+        message.messageId(),
+        message.qos(),
         finalTopicName,
         responseTopicName,
         payload,
-        publishMessage.duplicate(),
-        publishMessage.retain(),
-        publishMessage.contentType(),
-        publishMessage.subscriptionIds(),
-        publishMessage.correlationData(),
-        publishMessage.messageExpiryInterval(),
-        publishMessage.topicAlias(),
-        publishMessage.payloadFormat(),
-        publishMessage.userProperties());
+        message.duplicate(),
+        message.retain(),
+        message.contentType(),
+        message.subscriptionIds(),
+        message.correlationData(),
+        message.messageExpiryInterval(),
+        message.topicAlias(),
+        message.payloadFormat(),
+        message.userProperties());
 
     publishReceivingService.processPublish(user, publish);
   }
@@ -92,12 +92,12 @@ protected void processMessageWithValidFields(
   private TopicName resolveFinalTopicName(
       ExternalNetworkMqttUser user,
       NetworkMqttSession session,
-      PublishMqttInMessage publishMessage) {
+      PublishMqttInMessage message) {
 
     TopicNameMapping topicNameMapping = session.topicNameMapping();
-    String rawTopicName = publishMessage.rawTopicName();
+    String rawTopicName = message.rawTopicName();
     boolean providedRawTopicName = !StringUtils.isEmpty(rawTopicName);
-    int topicAlias = publishMessage.topicAlias();
+    int topicAlias = message.topicAlias();
 
     TopicName topicNameByAlias;
     TopicName finalTopicName;
diff --git a/core-service/src/main/java/javasabr/mqtt/service/publish/handler/MqttPublishOutMessageHandler.java b/core-service/src/main/java/javasabr/mqtt/service/publish/handler/MqttPublishOutMessageHandler.java
@@ -8,5 +8,5 @@ public interface MqttPublishOutMessageHandler {
 
   QoS qos();
 
-  PublishHandlingResult handle(Publish publish, SingleSubscriber subscriber);
+  void handle(Publish publish, SingleSubscriber subscriber);
 }
diff --git a/core-service/src/main/java/javasabr/mqtt/service/publish/handler/impl/AbstractMqttPublishInMessageHandler.java b/core-service/src/main/java/javasabr/mqtt/service/publish/handler/impl/AbstractMqttPublishInMessageHandler.java
@@ -31,7 +31,7 @@ public abstract class AbstractMqttPublishInMessageHandler<U extends NetworkMqttU
   MessageOutFactoryService messageOutFactoryService;
 
   @Override
-  public void handle(NetworkMqttUser user, Publish publish) {
+  public final void handle(NetworkMqttUser user, Publish publish) {
     if (!expectedUserType.isInstance(user)) {
       log.warning(user.clientId(), user.getClass(), "[%s] Not expected user of type:[%s]"::formatted);
       return;
diff --git a/core-service/src/main/java/javasabr/mqtt/service/publish/handler/impl/AbstractMqttPublishOutMessageHandler.java b/core-service/src/main/java/javasabr/mqtt/service/publish/handler/impl/AbstractMqttPublishOutMessageHandler.java
@@ -4,12 +4,9 @@
 import javasabr.mqtt.model.publishing.Publish;
 import javasabr.mqtt.model.session.MqttSession;
 import javasabr.mqtt.model.subscriber.SingleSubscriber;
-import javasabr.mqtt.network.message.out.MqttOutMessage;
 import javasabr.mqtt.network.user.NetworkMqttUser;
 import javasabr.mqtt.service.MessageOutFactoryService;
-import javasabr.mqtt.service.SubscriptionService;
 import javasabr.mqtt.service.publish.handler.MqttPublishOutMessageHandler;
-import javasabr.mqtt.service.publish.handler.PublishHandlingResult;
 import lombok.AccessLevel;
 import lombok.CustomLog;
 import lombok.RequiredArgsConstructor;
@@ -23,39 +20,36 @@ public abstract class AbstractMqttPublishOutMessageHandler<U extends NetworkMqtt
     implements MqttPublishOutMessageHandler {
 
   Class<U> expectedUserType;
-  SubscriptionService subscriptionService;
   MessageOutFactoryService messageOutFactoryService;
 
   @Override
-  public PublishHandlingResult handle(Publish publish, SingleSubscriber subscriber) {
+  public final void handle(Publish publish, SingleSubscriber subscriber) {
     MqttUser user = subscriber.resolveUser();
     if (!expectedUserType.isInstance(user)) {
       log.warning(user.clientId(), user.getClass(), "[%s] Not expected user of type:[%s]"::formatted);
-      return PublishHandlingResult.NOT_EXPECTED_CLIENT;
+      return;
     }
     U expectedUser = expectedUserType.cast(user);
     MqttSession session = expectedUser.session();
     if (session == null) {
       log.warning(user.clientId(), "[%s] Session is already closed"::formatted);
-      return PublishHandlingResult.SESSION_IS_ALREADY_CLOSED;
+      return;
     }
-    publish = reconstruct(expectedUser, session, publish);
-    if (publish == null) {
-      return PublishHandlingResult.SKIPPED;
+    Publish publishToSend = reconstruct(expectedUser, session, publish);
+    if (publishToSend != null) {
+      handleImpl(expectedUser, session, publishToSend);
     }
-    return handleImpl(expectedUser, session, publish);
   }
 
   @Nullable
   protected abstract Publish reconstruct(U user, MqttSession session, Publish original);
 
-  protected PublishHandlingResult handleImpl(U user, MqttSession session, Publish publish) {
+  protected void handleImpl(U user, MqttSession session, Publish publish) {
     send(user, publish);
-    return PublishHandlingResult.SUCCESS;
   }
 
   protected void send(U user, Publish publish) {
-    MqttOutMessage outMessage = messageOutFactoryService
+    user.sendInBackground(messageOutFactoryService
         .resolveFactory(user)
         .newPublish(
             publish.messageId(),
@@ -68,7 +62,6 @@ protected void send(U user, Publish publish) {
             publish.payloadFormat(),
             publish.responseTopicName(),
             publish.correlationData(),
-            publish.userProperties());
-    user.sendInBackground(outMessage);
+            publish.userProperties()));
   }
 }
diff --git a/core-service/src/main/java/javasabr/mqtt/service/publish/handler/impl/Qos0MqttPublishOutMessageHandler.java b/core-service/src/main/java/javasabr/mqtt/service/publish/handler/impl/Qos0MqttPublishOutMessageHandler.java
@@ -6,16 +6,13 @@
 import javasabr.mqtt.model.session.MqttSession;
 import javasabr.mqtt.network.impl.ExternalNetworkMqttUser;
 import javasabr.mqtt.service.MessageOutFactoryService;
-import javasabr.mqtt.service.SubscriptionService;
 import org.jspecify.annotations.Nullable;
 
 public class Qos0MqttPublishOutMessageHandler 
     extends AbstractMqttPublishOutMessageHandler<ExternalNetworkMqttUser> {
 
-  public Qos0MqttPublishOutMessageHandler(
-      SubscriptionService subscriptionService,
-      MessageOutFactoryService messageOutFactoryService) {
-    super(ExternalNetworkMqttUser.class, subscriptionService, messageOutFactoryService);
+  public Qos0MqttPublishOutMessageHandler(MessageOutFactoryService messageOutFactoryService) {
+    super(ExternalNetworkMqttUser.class, messageOutFactoryService);
   }
 
   @Override
diff --git a/core-service/src/main/java/javasabr/mqtt/service/publish/handler/impl/Qos1MqttPublishOutMessageHandler.java b/core-service/src/main/java/javasabr/mqtt/service/publish/handler/impl/Qos1MqttPublishOutMessageHandler.java
@@ -10,17 +10,14 @@
 import javasabr.mqtt.network.impl.ExternalNetworkMqttUser;
 import javasabr.mqtt.network.message.in.PublishAckMqttInMessage;
 import javasabr.mqtt.service.MessageOutFactoryService;
-import javasabr.mqtt.service.SubscriptionService;
 import lombok.CustomLog;
 import org.jspecify.annotations.Nullable;
 
 @CustomLog
 public class Qos1MqttPublishOutMessageHandler extends TrackableMqttPublishOutMessageHandler {
 
-  public Qos1MqttPublishOutMessageHandler(
-      SubscriptionService subscriptionService,
-      MessageOutFactoryService messageOutFactoryService) {
-    super(subscriptionService, messageOutFactoryService);
+  public Qos1MqttPublishOutMessageHandler(MessageOutFactoryService messageOutFactoryService) {
+    super(messageOutFactoryService);
   }
 
   @Override
diff --git a/core-service/src/main/java/javasabr/mqtt/service/publish/handler/impl/Qos2MqttPublishInMessageHandler.java b/core-service/src/main/java/javasabr/mqtt/service/publish/handler/impl/Qos2MqttPublishInMessageHandler.java
@@ -143,22 +143,23 @@ private void handleMessageIdIsInUse(ExternalNetworkMqttUser user, int messageId)
   }
 
   private boolean handleReceivedTrackableMessage(MqttUser user, MqttSession session, TrackableMqttMessage message) {
-    ExternalNetworkMqttUser networkUser = (ExternalNetworkMqttUser) user;
+    ExternalNetworkMqttUser networkMqttUser = expectedUserType.cast(user);
+    String clientId = networkMqttUser.clientId();
     int messageId = message.messageId();
 
     MessageTacker messageTacker = session.inMessageTracker();
     TrackedMessageMeta messageMeta = messageTacker.stored(messageId);
     if (messageMeta == null) {
-      log.warning(networkUser.clientId(), messageId, "[%s] No any stored information for messageId:[%d]"::formatted);
+      log.warning(clientId, messageId, "[%s] No any stored information for messageId:[%d]"::formatted);
       return true;
     }
 
     if (messageMeta.messageType() != MqttMessageType.PUBLISH) {
-      log.warning(networkUser.clientId(), messageMeta, messageId,
+      log.warning(clientId, messageMeta, messageId, 
           "[%s] Not expected tracked message meta:[%s] for messageId:[%d]"::formatted);
       return true;
     } else if (!(message instanceof PublishReleaseMqttInMessage release)) {
-      log.warning(networkUser.clientId(), message, "[%s] Not expected message:[%s]"::formatted);
+      log.warning(clientId, message.messageType(), "[%s] Not expected message:[%s]"::formatted);
       return true;
     }
 
@@ -168,10 +169,10 @@ private boolean handleReceivedTrackableMessage(MqttUser user, MqttSession sessio
         PublishCompletedReasonCode.SUCCESS);
 
     MqttOutMessage response = messageOutFactoryService
-        .resolveFactory(networkUser)
+        .resolveFactory(networkMqttUser)
         .newPublishCompleted(message.messageId(), PublishCompletedReasonCode.SUCCESS);
 
-    sendFeedback(networkUser, session, response, messageId);
+    sendFeedback(networkMqttUser, session, response, messageId);
     return true;
   }
 }
diff --git a/core-service/src/main/java/javasabr/mqtt/service/publish/handler/impl/Qos2MqttPublishOutMessageHandler.java b/core-service/src/main/java/javasabr/mqtt/service/publish/handler/impl/Qos2MqttPublishOutMessageHandler.java
@@ -13,17 +13,14 @@
 import javasabr.mqtt.network.message.in.PublishCompleteMqttInMessage;
 import javasabr.mqtt.network.message.in.PublishReceivedMqttInMessage;
 import javasabr.mqtt.service.MessageOutFactoryService;
-import javasabr.mqtt.service.SubscriptionService;
 import lombok.CustomLog;
 import org.jspecify.annotations.Nullable;
 
 @CustomLog
 public class Qos2MqttPublishOutMessageHandler extends TrackableMqttPublishOutMessageHandler {
 
-  public Qos2MqttPublishOutMessageHandler(
-      SubscriptionService subscriptionService,
-      MessageOutFactoryService messageOutFactoryService) {
-    super(subscriptionService, messageOutFactoryService);
+  public Qos2MqttPublishOutMessageHandler(MessageOutFactoryService messageOutFactoryService) {
+    super(messageOutFactoryService);
   }
 
   @Override
diff --git a/core-service/src/main/java/javasabr/mqtt/service/publish/handler/impl/TrackableMqttPublishOutMessageHandler.java b/core-service/src/main/java/javasabr/mqtt/service/publish/handler/impl/TrackableMqttPublishOutMessageHandler.java
diff --git a/core-service/src/main/java/javasabr/mqtt/service/session/impl/InMemoryNetworkMqttSession.java b/core-service/src/main/java/javasabr/mqtt/service/session/impl/InMemoryNetworkMqttSession.java
diff --git a/core-service/src/test/groovy/javasabr/mqtt/service/IntegrationServiceSpecification.groovy b/core-service/src/test/groovy/javasabr/mqtt/service/IntegrationServiceSpecification.groovy
diff --git a/core-service/src/test/groovy/javasabr/mqtt/service/publish/handler/impl/Qos0MqttPublishOutMessageHandlerTest.groovy b/core-service/src/test/groovy/javasabr/mqtt/service/publish/handler/impl/Qos0MqttPublishOutMessageHandlerTest.groovy
diff --git a/core-service/src/test/groovy/javasabr/mqtt/service/publish/handler/impl/Qos1MqttPublishOutMessageHandlerTest.groovy b/core-service/src/test/groovy/javasabr/mqtt/service/publish/handler/impl/Qos1MqttPublishOutMessageHandlerTest.groovy
diff --git a/core-service/src/test/groovy/javasabr/mqtt/service/publish/handler/impl/Qos2MqttPublishOutMessageHandlerTest.groovy b/core-service/src/test/groovy/javasabr/mqtt/service/publish/handler/impl/Qos2MqttPublishOutMessageHandlerTest.groovy

Original file line number	Diff line number	Diff line change
`@@ -2,9 +2,8 @@`
`2`	`2`
`3`	`3`	`import javasabr.mqtt.model.publishing.Publish;`
`4`	`4`	`import javasabr.mqtt.model.subscriber.SingleSubscriber;`
`5`		`-import javasabr.mqtt.service.publish.handler.PublishHandlingResult;`
`6`	`5`
`7`	`6`	`public interface PublishDeliveringService {`
`8`	`7`
`9`		`- PublishHandlingResult startDelivering(Publish publish, SingleSubscriber subscriber);`
	`8`	`+ void startDelivering(Publish publish, SingleSubscriber subscriber);`
`10`	`9`	`}`
Original file line number	Diff line number	Diff line change
`@@ -54,8 +54,7 @@ public void processPublish(NetworkMqttUser user, Publish publish) {`
`54`	`54`	`log.warning(user.clientId(), publish, "[%s] Received not supported publish:%s"::formatted);`
`55`	`55`	`}`
`56`	`56`	`}`
`57`		`-`
`58`		`-`
	`57`	`+`
`59`	`58`	`private static String buildServiceDescription(`
`60`	`59`	`@Nullable MqttPublishInMessageHandler[] publishInMessageHandlers) {`
`61`	`60`	`var builder = new StringBuilder();`
Original file line number	Diff line number	Diff line change
`@@ -8,5 +8,5 @@ public interface MqttPublishOutMessageHandler {`
`8`	`8`
`9`	`9`	`QoS qos();`
`10`	`10`
`11`		`- PublishHandlingResult handle(Publish publish, SingleSubscriber subscriber);`
	`11`	`+ void handle(Publish publish, SingleSubscriber subscriber);`
`12`	`12`	`}`