continue working on reorganazing network

Author: JavaSaBr

application/src/main/java/javasabr/mqtt/broker/application/config/MqttBrokerSpringConfig.java

@@ -9,15 +9,18 @@
 import javasabr.mqtt.network.MqttConnection;
 import javasabr.mqtt.network.MqttConnectionFactory;
 import javasabr.mqtt.network.handler.MqttClientReleaseHandler;
+import javasabr.mqtt.network.handler.PublishInHandler;
 import javasabr.mqtt.service.AuthenticationService;
 import javasabr.mqtt.service.ClientIdRegistry;
 import javasabr.mqtt.service.CredentialSource;
 import javasabr.mqtt.service.MqttConnectionService;
 import javasabr.mqtt.service.MqttSessionService;
+import javasabr.mqtt.service.PublishingService;
 import javasabr.mqtt.service.SubscriptionService;
 import javasabr.mqtt.service.handler.client.ExternalMqttClientReleaseHandler;
 import javasabr.mqtt.service.impl.DefaultMqttConnectionFactory;
 import javasabr.mqtt.service.impl.DefaultMqttConnectionService;
+import javasabr.mqtt.service.impl.DefaultPublishingService;
 import javasabr.mqtt.service.impl.ExternalMqttClientFactory;
 import javasabr.mqtt.service.impl.FileCredentialsSource;
 import javasabr.mqtt.service.impl.InMemoryClientIdRegistry;
@@ -26,6 +29,13 @@
 import javasabr.mqtt.service.impl.SimpleSubscriptionService;
 import javasabr.mqtt.service.message.handler.MqttInMessageHandler;
 import javasabr.mqtt.service.message.handler.impl.ConnectInMqttInMessageHandler;
+import javasabr.mqtt.service.message.handler.impl.DisconnectMqttInMessageHandler;
+import javasabr.mqtt.service.message.handler.impl.PendingResponseMqttInMessageHandler;
+import javasabr.mqtt.service.message.handler.impl.PublishAckMqttInMessageHandler;
+import javasabr.mqtt.service.message.handler.impl.PublishCompleteMqttInMessageHandler;
+import javasabr.mqtt.service.message.handler.impl.PublishMqttInMessageHandler;
+import javasabr.mqtt.service.message.handler.impl.PublishReceiveMqttInMessageHandler;
+import javasabr.mqtt.service.message.handler.impl.PublishReleaseMqttInMessageHandler;
 import javasabr.rlib.network.NetworkFactory;
 import javasabr.rlib.network.ServerNetworkConfig;
 import javasabr.rlib.network.server.ServerNetwork;
@@ -74,6 +84,11 @@ SubscriptionService subscriptionService() {
     return new SimpleSubscriptionService();
   }
 
+  @Bean
+  PublishingService publishingService() {
+    return new DefaultPublishingService(new PublishInHandler[0]);
+  }
+
   @Bean
   MqttInMessageHandler connectInMqttInMessageHandler(
       ClientIdRegistry clientIdRegistry,
@@ -87,6 +102,36 @@ MqttInMessageHandler connectInMqttInMessageHandler(
         subscriptionService);
   }
 
+  @Bean
+  MqttInMessageHandler publishAckMqttInMessageHandler() {
+    return new PublishAckMqttInMessageHandler();
+  }
+
+  @Bean
+  MqttInMessageHandler publishCompleteMqttInMessageHandler() {
+    return new PublishCompleteMqttInMessageHandler();
+  }
+
+  @Bean
+  MqttInMessageHandler publishMqttInMessageHandler(PublishingService publishingService) {
+    return new PublishMqttInMessageHandler(publishingService);
+  }
+
+  @Bean
+  MqttInMessageHandler publishReceiveMqttInMessageHandler() {
+    return new PublishReceiveMqttInMessageHandler();
+  }
+
+  @Bean
+  MqttInMessageHandler publishReleaseMqttInMessageHandler() {
+    return new PublishReleaseMqttInMessageHandler();
+  }
+
+  @Bean
+  MqttInMessageHandler disconnectMqttInMessageHandler() {
+    return new DisconnectMqttInMessageHandler();
+  }
+
   @Bean
   MqttConnectionService mqttConnectionService(Collection<? extends MqttInMessageHandler> inMessageHandlers) {
     return new DefaultMqttConnectionService(inMessageHandlers);

gradle/libs.versions.toml

@@ -1,6 +1,6 @@
 [versions]
 # https://gitlab.com/JavaSaBr/maven-repo/-/packages
-rlib = "10.0.alpha4"
+rlib = "10.0.alpha5"
 # https://mvnrepository.com/artifact/org.jetbrains/annotations
 jetbrains-annotations = "26.0.2"
 # https://mvnrepository.com/artifact/org.projectlombok/lombok

service/src/main/java/javasabr/mqtt/service/impl/DefaultMqttConnectionService.java

@@ -31,6 +31,7 @@ public DefaultMqttConnectionService(Collection<? extends MqttInMessageHandler> k
     }
 
     this.inMessageHandlers = inMessageHandlers;
+    log.info(knownInMessageHandlers.size(), "Registered [%s] MqttInMessageHandlers"::formatted);
   }
 
   @Override

service/src/main/java/javasabr/mqtt/service/message/handler/impl/PendingResponseMqttInMessageHandler.java

@@ -0,0 +1,23 @@
+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.HasPacketId;
+import javasabr.mqtt.network.packet.in.MqttReadablePacket;
+
+public abstract class PendingResponseMqttInMessageHandler<P extends MqttReadablePacket & HasPacketId>
+    extends AbstractMqttInMessageHandler<ExternalMqttClient, P> {
+
+  protected PendingResponseMqttInMessageHandler(Class<P> expectedNetworkPacket) {
+    super(ExternalMqttClient.class, expectedNetworkPacket);
+  }
+
+  @Override
+  protected void processReceived(MqttConnection connection, ExternalMqttClient client, P networkPacket) {
+    MqttSession session = client.session();
+    if (session != null) {
+      session.updateOutPendingPacket(client, networkPacket);
+    }
+  }
+}

service/src/main/java/javasabr/mqtt/service/message/handler/impl/PublishAckMqttInMessageHandler.java

@@ -0,0 +1,16 @@
+package javasabr.mqtt.service.message.handler.impl;
+
+import javasabr.mqtt.network.packet.MqttPacketType;
+import javasabr.mqtt.network.packet.in.PublishAckInPacket;
+
+public class PublishAckMqttInMessageHandler extends PendingResponseMqttInMessageHandler<PublishAckInPacket> {
+
+  public PublishAckMqttInMessageHandler() {
+    super(PublishAckInPacket.class);
+  }
+
+  @Override
+  public int packetType() {
+    return MqttPacketType.PUBLISH_ACK.typeIndex();
+  }
+}

service/src/main/java/javasabr/mqtt/service/message/handler/impl/PublishCompleteMqttInMessageHandler.java

@@ -0,0 +1,16 @@
+package javasabr.mqtt.service.message.handler.impl;
+
+import javasabr.mqtt.network.packet.MqttPacketType;
+import javasabr.mqtt.network.packet.in.PublishCompleteInPacket;
+
+public class PublishCompleteMqttInMessageHandler extends PendingResponseMqttInMessageHandler<PublishCompleteInPacket>  {
+
+  public PublishCompleteMqttInMessageHandler() {
+    super(PublishCompleteInPacket.class);
+  }
+
+  @Override
+  public int packetType() {
+    return MqttPacketType.PUBLISH_COMPLETED.typeIndex();
+  }
+}

service/src/main/java/javasabr/mqtt/service/message/handler/impl/PublishMqttInMessageHandler.java

@@ -0,0 +1,33 @@
+package javasabr.mqtt.service.message.handler.impl;
+
+import javasabr.mqtt.network.MqttConnection;
+import javasabr.mqtt.network.client.ExternalMqttClient;
+import javasabr.mqtt.network.packet.MqttPacketType;
+import javasabr.mqtt.network.packet.in.PublishInPacket;
+import javasabr.mqtt.service.PublishingService;
+import lombok.AccessLevel;
+import lombok.experimental.FieldDefaults;
+
+@FieldDefaults(level = AccessLevel.PRIVATE, makeFinal = true)
+public class PublishMqttInMessageHandler extends AbstractMqttInMessageHandler<ExternalMqttClient, PublishInPacket> {
+
+  PublishingService publishingService;
+
+  public PublishMqttInMessageHandler(PublishingService publishingService) {
+    super(ExternalMqttClient.class, PublishInPacket.class);
+    this.publishingService = publishingService;
+  }
+
+  @Override
+  protected void processReceived(
+      MqttConnection connection,
+      ExternalMqttClient client,
+      PublishInPacket networkPacket) {
+    publishingService.publish(client, networkPacket);
+  }
+
+  @Override
+  public int packetType() {
+    return MqttPacketType.PUBLISH.typeIndex();
+  }
+}

service/src/main/java/javasabr/mqtt/service/message/handler/impl/PublishReceiveMqttInMessageHandler.java

@@ -0,0 +1,16 @@
+package javasabr.mqtt.service.message.handler.impl;
+
+import javasabr.mqtt.network.packet.MqttPacketType;
+import javasabr.mqtt.network.packet.in.PublishReceivedInPacket;
+
+public class PublishReceiveMqttInMessageHandler extends PendingResponseMqttInMessageHandler<PublishReceivedInPacket> {
+
+  public PublishReceiveMqttInMessageHandler() {
+    super(PublishReceivedInPacket.class);
+  }
+
+  @Override
+  public int packetType() {
+    return MqttPacketType.PUBLISH_RECEIVED.typeIndex();
+  }
+}

service/src/main/java/javasabr/mqtt/service/message/handler/impl/PublishReleaseMqttInMessageHandler.java

@@ -0,0 +1,16 @@
+package javasabr.mqtt.service.message.handler.impl;
+
+import javasabr.mqtt.network.packet.MqttPacketType;
+import javasabr.mqtt.network.packet.in.PublishReleaseInPacket;
+
+public class PublishReleaseMqttInMessageHandler extends PendingResponseMqttInMessageHandler<PublishReleaseInPacket> {
+
+  public PublishReleaseMqttInMessageHandler() {
+    super(PublishReleaseInPacket.class);
+  }
+
+  @Override
+  public int packetType() {
+    return MqttPacketType.PUBLISH_RELEASED.typeIndex();
+  }
+}

service/src/main/java/javasabr/mqtt/service/message/handler/impl/package-info.java

@@ -0,0 +1,4 @@
+@NullMarked
+package javasabr.mqtt.service.message.handler.impl;
+
+import org.jspecify.annotations.NullMarked;