continue working on reorganazing network

JavaSaBr · JavaSaBr · commit d5ad2fa625cc · 2025-10-24T19:46:28.000+02: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
@@ -22,6 +22,7 @@
 import javasabr.mqtt.service.handler.client.ExternalMqttClientReleaseHandler;
 import javasabr.mqtt.service.impl.DefaultConnectionService;
 import javasabr.mqtt.service.impl.DefaultMqttConnectionFactory;
+import javasabr.mqtt.service.impl.DefaultPublishDeliveringService;
 import javasabr.mqtt.service.impl.DefaultPublishReceivingService;
 import javasabr.mqtt.service.impl.DefaultPublishingService;
 import javasabr.mqtt.service.impl.ExternalMqttClientFactory;
@@ -41,9 +42,13 @@
 import javasabr.mqtt.service.message.handler.impl.SubscribeMqttInMessageHandler;
 import javasabr.mqtt.service.message.handler.impl.UnsubscribeMqttInMessageHandler;
 import javasabr.mqtt.service.publish.handler.MqttPublishInMessageHandler;
-import javasabr.mqtt.service.publish.handler.impl.Qos0PublishInMessageHandler;
-import javasabr.mqtt.service.publish.handler.impl.Qos1PublishInMessageHandler;
-import javasabr.mqtt.service.publish.handler.impl.Qos2PublishInMessageHandler;
+import javasabr.mqtt.service.publish.handler.MqttPublishOutMessageHandler;
+import javasabr.mqtt.service.publish.handler.impl.Qos0MqttPublishInMessageHandler;
+import javasabr.mqtt.service.publish.handler.impl.Qos0MqttPublishOutMessageHandler;
+import javasabr.mqtt.service.publish.handler.impl.Qos1MqttPublishInMessageHandler;
+import javasabr.mqtt.service.publish.handler.impl.Qos1MqttPublishOutMessageHandler;
+import javasabr.mqtt.service.publish.handler.impl.Qos2MqttPublishInMessageHandler;
+import javasabr.mqtt.service.publish.handler.impl.Qos2MqttPublishOutMessageHandler;
 import javasabr.rlib.network.NetworkFactory;
 import javasabr.rlib.network.ServerNetworkConfig;
 import javasabr.rlib.network.server.ServerNetwork;
@@ -156,29 +161,45 @@ ConnectionService mqttConnectionService(Collection<? extends MqttInMessageHandle
   }
 
   @Bean
-  PublishDeliveringService publishDeliveringService() {
-    return null;
+  MqttPublishOutMessageHandler qos0MqttPublishOutMessageHandler(SubscriptionService subscriptionService) {
+    return new Qos0MqttPublishOutMessageHandler(subscriptionService);
   }
 
   @Bean
-  MqttPublishInMessageHandler qos0PublishInMessageHandler(
+  MqttPublishOutMessageHandler qos1MqttPublishOutMessageHandler(SubscriptionService subscriptionService) {
+    return new Qos1MqttPublishOutMessageHandler(subscriptionService);
+  }
+
+  @Bean
+  MqttPublishOutMessageHandler qos2MqttPublishOutMessageHandler(SubscriptionService subscriptionService) {
+    return new Qos2MqttPublishOutMessageHandler(subscriptionService);
+  }
+
+  @Bean
+  PublishDeliveringService publishDeliveringService(
+      Collection<? extends MqttPublishOutMessageHandler> knownPublishOutHandlers) {
+    return new DefaultPublishDeliveringService(knownPublishOutHandlers);
+  }
+
+  @Bean
+  MqttPublishInMessageHandler qos0MqttPublishInMessageHandler(
       SubscriptionService subscriptionService,
       PublishDeliveringService publishDeliveringService) {
-    return new Qos0PublishInMessageHandler(subscriptionService, publishDeliveringService);
+    return new Qos0MqttPublishInMessageHandler(subscriptionService, publishDeliveringService);
   }
 
   @Bean
-  MqttPublishInMessageHandler qos1PublishInMessageHandler(
+  MqttPublishInMessageHandler qos1MqttPublishInMessageHandler(
       SubscriptionService subscriptionService,
       PublishDeliveringService publishDeliveringService) {
-    return new Qos1PublishInMessageHandler(subscriptionService, publishDeliveringService);
+    return new Qos1MqttPublishInMessageHandler(subscriptionService, publishDeliveringService);
   }
 
   @Bean
-  MqttPublishInMessageHandler qos2PublishInMessageHandler(
+  MqttPublishInMessageHandler qos2MqttPublishInMessageHandler(
       SubscriptionService subscriptionService,
       PublishDeliveringService publishDeliveringService) {
-    return new Qos2PublishInMessageHandler(subscriptionService, publishDeliveringService);
+    return new Qos2MqttPublishInMessageHandler(subscriptionService, publishDeliveringService);
   }
 
   @Bean
diff --git a/service/src/main/java/javasabr/mqtt/service/PublishDeliveringService.java b/service/src/main/java/javasabr/mqtt/service/PublishDeliveringService.java
@@ -1,10 +1,10 @@
 package javasabr.mqtt.service;
 
-import javasabr.mqtt.model.ActionResult;
 import javasabr.mqtt.model.subscriber.SingleSubscriber;
 import javasabr.mqtt.network.packet.in.PublishInPacket;
+import javasabr.mqtt.service.publish.handler.PublishHandlingResult;
 
 public interface PublishDeliveringService {
 
-  ActionResult startDelivering(PublishInPacket publish, SingleSubscriber subscriber);
+  PublishHandlingResult startDelivering(PublishInPacket publish, SingleSubscriber subscriber);
 }
diff --git a/service/src/main/java/javasabr/mqtt/service/SubscriptionService.java b/service/src/main/java/javasabr/mqtt/service/SubscriptionService.java
@@ -6,6 +6,7 @@
 import javasabr.mqtt.model.reason.code.UnsubscribeAckReasonCode;
 import javasabr.mqtt.model.subscriber.SingleSubscriber;
 import javasabr.mqtt.model.subscriber.SubscribeTopicFilter;
+import javasabr.mqtt.model.subscriber.Subscriber;
 import javasabr.mqtt.model.topic.TopicFilter;
 import javasabr.mqtt.model.topic.TopicName;
 import javasabr.mqtt.network.MqttClient;
@@ -20,6 +21,8 @@ public interface SubscriptionService {
 
   boolean isValid(TopicName topicName);
 
+  MqttClient resolveClient(Subscriber subscriber);
+
   default Array<SingleSubscriber> findSubscribers(TopicName topicName) {
     return findSubscribersTo(MutableArray.ofType(SingleSubscriber.class), topicName);
   }
diff --git a/service/src/main/java/javasabr/mqtt/service/impl/DefaultPublishDeliveringService.java b/service/src/main/java/javasabr/mqtt/service/impl/DefaultPublishDeliveringService.java
@@ -0,0 +1,56 @@
+package javasabr.mqtt.service.impl;
+
+import java.util.Collection;
+import javasabr.mqtt.model.QoS;
+import javasabr.mqtt.model.subscriber.SingleSubscriber;
+import javasabr.mqtt.network.packet.in.PublishInPacket;
+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;
+import org.jspecify.annotations.Nullable;
+
+@CustomLog
+@FieldDefaults(level = AccessLevel.PRIVATE, makeFinal = true)
+public class DefaultPublishDeliveringService implements PublishDeliveringService {
+
+  @Nullable
+  MqttPublishOutMessageHandler[] publishOutMessageHandlers;
+
+  public DefaultPublishDeliveringService(
+      Collection<? extends MqttPublishOutMessageHandler> knownPublishOutHandlers) {
+
+    int maxIndex = knownPublishOutHandlers
+        .stream()
+        .map(MqttPublishOutMessageHandler::qos)
+        .mapToInt(QoS::index)
+        .max()
+        .orElse(0);
+
+    var publishOutHandlers = new MqttPublishOutMessageHandler[maxIndex + 1];
+
+    for (MqttPublishOutMessageHandler knownPublishOutHandler : knownPublishOutHandlers) {
+      QoS qos = knownPublishOutHandler.qos();
+      if (publishOutHandlers[qos.index()] != null) {
+        throw new IllegalArgumentException(
+            "Found duplicate MqttPublishOutMessageHandler:[" + knownPublishOutHandler + "]");
+      }
+      publishOutHandlers[qos.index()] = knownPublishOutHandler;
+    }
+
+    this.publishOutMessageHandlers = publishOutHandlers;
+  }
+
+  @Override
+  public PublishHandlingResult startDelivering(PublishInPacket publish, SingleSubscriber subscriber) {
+    try {
+      //noinspection DataFlowIssue
+      return publishOutMessageHandlers[publish.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/publish/handler/MqttPublishOutMessageHandler.java b/service/src/main/java/javasabr/mqtt/service/publish/handler/MqttPublishOutMessageHandler.java
@@ -1,6 +1,5 @@
 package javasabr.mqtt.service.publish.handler;
 
-import javasabr.mqtt.model.ActionResult;
 import javasabr.mqtt.model.QoS;
 import javasabr.mqtt.model.subscriber.SingleSubscriber;
 import javasabr.mqtt.network.packet.in.PublishInPacket;
@@ -9,5 +8,5 @@ public interface MqttPublishOutMessageHandler {
 
   QoS qos();
 
-  ActionResult handle(PublishInPacket packet, SingleSubscriber subscriber);
+  PublishHandlingResult handle(PublishInPacket packet, SingleSubscriber subscriber);
 }
diff --git a/service/src/main/java/javasabr/mqtt/service/publish/handler/PublishHandlingResult.java b/service/src/main/java/javasabr/mqtt/service/publish/handler/PublishHandlingResult.java
@@ -1,4 +1,4 @@
-package javasabr.mqtt.service.message.handler;
+package javasabr.mqtt.service.publish.handler;
 
 import javasabr.mqtt.model.reason.code.PublishAckReasonCode;
 import javasabr.mqtt.model.reason.code.PublishReceivedReasonCode;
@@ -32,7 +32,13 @@ public enum PublishHandlingResult {
   PAYLOAD_FORMAT_INVALID(
       true,
       PublishAckReasonCode.PAYLOAD_FORMAT_INVALID,
-      PublishReceivedReasonCode.PAYLOAD_FORMAT_INVALID);
+      PublishReceivedReasonCode.PAYLOAD_FORMAT_INVALID),
+
+  // CUSTOM
+  NOT_EXPECTED_CLIENT(
+      true,
+      PublishAckReasonCode.UNSPECIFIED_ERROR,
+      PublishReceivedReasonCode.UNSPECIFIED_ERROR);
 
   boolean error;
   PublishAckReasonCode ackReasonCode;
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
@@ -6,13 +6,15 @@
 import javasabr.mqtt.network.packet.in.PublishInPacket;
 import javasabr.mqtt.service.PublishDeliveringService;
 import javasabr.mqtt.service.SubscriptionService;
-import javasabr.mqtt.service.message.handler.PublishHandlingResult;
 import javasabr.mqtt.service.publish.handler.MqttPublishInMessageHandler;
+import javasabr.mqtt.service.publish.handler.PublishHandlingResult;
 import javasabr.rlib.collections.array.Array;
 import lombok.AccessLevel;
+import lombok.CustomLog;
 import lombok.RequiredArgsConstructor;
 import lombok.experimental.FieldDefaults;
 
+@CustomLog
 @RequiredArgsConstructor
 @FieldDefaults(level = AccessLevel.PROTECTED, makeFinal = true)
 public abstract class AbstractMqttPublishInMessageHandler<C extends MqttClient>
@@ -24,6 +26,10 @@ public abstract class AbstractMqttPublishInMessageHandler<C extends MqttClient>
 
   @Override
   public void handle(MqttClient client, PublishInPacket packet) {
+    if (!expectedClient.isInstance(client)) {
+      log.warning(client, "Not expected client:[%s]"::formatted);
+      return;
+    }
     handleImpl(expectedClient.cast(client), packet);
   }
 
diff --git a/service/src/main/java/javasabr/mqtt/service/publish/handler/impl/AbstractMqttPublishOutMessageHandler.java b/service/src/main/java/javasabr/mqtt/service/publish/handler/impl/AbstractMqttPublishOutMessageHandler.java
@@ -0,0 +1,57 @@
+package javasabr.mqtt.service.publish.handler.impl;
+
+import javasabr.mqtt.model.MqttProperties;
+import javasabr.mqtt.model.subscriber.SingleSubscriber;
+import javasabr.mqtt.network.MqttClient;
+import javasabr.mqtt.network.packet.in.PublishInPacket;
+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;
+import lombok.experimental.FieldDefaults;
+
+@CustomLog
+@RequiredArgsConstructor
+@FieldDefaults(level = AccessLevel.PROTECTED, makeFinal = true)
+public abstract class AbstractMqttPublishOutMessageHandler<C extends MqttClient>
+    implements MqttPublishOutMessageHandler {
+
+  Class<C> expectedClient;
+  SubscriptionService subscriptionService;
+
+  @Override
+  public PublishHandlingResult handle(PublishInPacket packet, SingleSubscriber subscriber) {
+    MqttClient mqttClient = subscriptionService.resolveClient(subscriber);
+    if (!expectedClient.isInstance(mqttClient)) {
+      log.warning(mqttClient, "Accepted not expected client:[%s]"::formatted);
+      return PublishHandlingResult.NOT_EXPECTED_CLIENT;
+    }
+    return handleImpl(packet, expectedClient.cast(mqttClient));
+  }
+
+  protected abstract PublishHandlingResult handleImpl(PublishInPacket packet, C client) ;
+
+  protected void startDelivering(
+      MqttClient client,
+      PublishInPacket packet,
+      int messageId,
+      boolean duplicate) {
+    var packetOutFactory = client.packetOutFactory();
+    client.send(packetOutFactory.newPublish(
+        messageId,
+        qos(),
+        packet.isRetained(),
+        duplicate,
+        packet
+            .getTopicName()
+            .toString(),
+        MqttProperties.TOPIC_ALIAS_NOT_SET,
+        packet.getPayload(),
+        packet.isPayloadFormatIndicator(),
+        packet.getResponseTopic(),
+        packet.getCorrelationData(),
+        packet.userProperties()));
+  }
+}
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
@@ -0,0 +1,59 @@
+package javasabr.mqtt.service.publish.handler.impl;
+
+import javasabr.mqtt.network.MqttClient;
+import javasabr.mqtt.network.MqttSession;
+import javasabr.mqtt.network.MqttSession.PendingMessageHandler;
+import javasabr.mqtt.network.client.ExternalMqttClient;
+import javasabr.mqtt.network.packet.HasPacketId;
+import javasabr.mqtt.network.packet.in.PublishInPacket;
+import javasabr.mqtt.service.SubscriptionService;
+import javasabr.mqtt.service.publish.handler.PublishHandlingResult;
+import lombok.AccessLevel;
+import lombok.experimental.FieldDefaults;
+
+@FieldDefaults(level = AccessLevel.PROTECTED, makeFinal = true)
+public abstract class PersistedMqttPublishOutMessageHandler extends AbstractMqttPublishOutMessageHandler<ExternalMqttClient> {
+
+  PendingMessageHandler pendingMessageHandler;
+
+  protected PersistedMqttPublishOutMessageHandler(SubscriptionService subscriptionService) {
+    super(ExternalMqttClient.class, subscriptionService);
+    this.pendingMessageHandler = new PendingMessageHandler() {
+      @Override
+      public boolean handleResponse(MqttClient client, HasPacketId response) {
+        return handleReceivedResponse(client, response);
+      }
+      @Override
+      public void resend(MqttClient client, PublishInPacket packet, int packetId) {
+        tryToDeliverAgain(client, packet, packetId);
+      }
+    };
+  }
+
+  @Override
+  protected PublishHandlingResult handleImpl(PublishInPacket packet, ExternalMqttClient client) {
+
+    MqttSession session = client.session();
+    if (session == null) {
+      return PublishHandlingResult.SKIPPED;
+    }
+    // generate new uniq packet id per client
+    var packetId = session.nextPacketId();
+
+    // register waiting async response
+    session.registerOutPublish(packet, pendingMessageHandler, packetId);
+
+    // send publish
+    startDelivering(client, packet, packetId, false);
+
+    return PublishHandlingResult.SUCCESS;
+  }
+
+  protected boolean handleReceivedResponse(MqttClient client, HasPacketId response) {
+    return false;
+  }
+
+  protected void tryToDeliverAgain(MqttClient client, PublishInPacket packet, int messageId) {
+    startDelivering(client, packet, messageId, true);
+  }
+}
diff --git a/service/src/main/java/javasabr/mqtt/service/publish/handler/impl/Qos0MqttPublishInMessageHandler.java b/service/src/main/java/javasabr/mqtt/service/publish/handler/impl/Qos0MqttPublishInMessageHandler.java
@@ -5,9 +5,9 @@
 import javasabr.mqtt.service.PublishDeliveringService;
 import javasabr.mqtt.service.SubscriptionService;
 
-public class Qos0PublishInMessageHandler extends AbstractMqttPublishInMessageHandler<ExternalMqttClient> {
+public class Qos0MqttPublishInMessageHandler extends AbstractMqttPublishInMessageHandler<ExternalMqttClient> {
 
-  public Qos0PublishInMessageHandler(
+  public Qos0MqttPublishInMessageHandler(
       SubscriptionService subscriptionService,
       PublishDeliveringService publishDeliveringService) {
     super(ExternalMqttClient.class, subscriptionService, publishDeliveringService);
diff --git a/service/src/main/java/javasabr/mqtt/service/publish/handler/impl/Qos0MqttPublishOutMessageHandler.java b/service/src/main/java/javasabr/mqtt/service/publish/handler/impl/Qos0MqttPublishOutMessageHandler.java
@@ -0,0 +1,25 @@
+package javasabr.mqtt.service.publish.handler.impl;
+
+import javasabr.mqtt.model.QoS;
+import javasabr.mqtt.network.client.ExternalMqttClient;
+import javasabr.mqtt.network.packet.in.PublishInPacket;
+import javasabr.mqtt.service.SubscriptionService;
+import javasabr.mqtt.service.publish.handler.PublishHandlingResult;
+
+public class Qos0MqttPublishOutMessageHandler extends AbstractMqttPublishOutMessageHandler<ExternalMqttClient> {
+
+  public Qos0MqttPublishOutMessageHandler(SubscriptionService subscriptionService) {
+    super(ExternalMqttClient.class, subscriptionService);
+  }
+
+  @Override
+  public QoS qos() {
+    return QoS.AT_MOST_ONCE;
+  }
+
+  @Override
+  protected PublishHandlingResult handleImpl(PublishInPacket packet, ExternalMqttClient client) {
+    startDelivering(client, packet, packet.getPacketId(), 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
@@ -6,11 +6,11 @@
 import javasabr.mqtt.network.client.ExternalMqttClient;
 import javasabr.mqtt.service.PublishDeliveringService;
 import javasabr.mqtt.service.SubscriptionService;
-import javasabr.mqtt.service.message.handler.PublishHandlingResult;
+import javasabr.mqtt.service.publish.handler.PublishHandlingResult;
 
-public class Qos1PublishInMessageHandler extends Qos0PublishInMessageHandler {
+public class Qos1MqttPublishInMessageHandler extends Qos0MqttPublishInMessageHandler {
 
-  public Qos1PublishInMessageHandler(
+  public Qos1MqttPublishInMessageHandler(
       SubscriptionService subscriptionService,
       PublishDeliveringService publishDeliveringService) {
     super(subscriptionService, publishDeliveringService);
diff --git a/service/src/main/java/javasabr/mqtt/service/publish/handler/impl/Qos1MqttPublishOutMessageHandler.java b/service/src/main/java/javasabr/mqtt/service/publish/handler/impl/Qos1MqttPublishOutMessageHandler.java
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
diff --git a/service/src/main/java/javasabr/mqtt/service/publish/handler/impl/Qos2MqttPublishOutMessageHandler.java b/service/src/main/java/javasabr/mqtt/service/publish/handler/impl/Qos2MqttPublishOutMessageHandler.java
