feature: [158] Implement global mqtt message storage service

.github/copilot-instructions.md

@@ -77,6 +77,9 @@ The repository is organized into the following modules:
 ./gradlew :network:test
 ./gradlew :core-service:test
 
+# Run a single Spock spec while iterating on one area
+./gradlew :core-service:test --tests 'javasabr.mqtt.service.session.impl.InMemoryMqttSessionServiceTest'
+
 # Tests use JUnit Platform with Groovy/Spock
 # Tests run with maxParallelForks=2, forkEvery=100
 ```
@@ -166,6 +169,9 @@ The repository is organized into the following modules:
 ### Testing Conventions
 - **Framework**: Spock (Groovy-based BDD framework)
 - **Test location**: `src/test/groovy/` directories
+- **Base specs**: Use `UnitSpecification` for isolated unit tests and `IntegrationServiceSpecification` for service-level tests that need shared broker fixtures or helper services
+- **Async helpers**: Use `fromAsync(...)` and `waitForAsync(...)` from `BaseSpecification` to unwrap `Mono` and `CompletionStage` results in tests
+- **Lifecycle cleanup**: If a test creates a service with its own lifecycle or background thread (for example `InMemoryMqttSessionService`), close it in a Spock `cleanup:` block
 - **Test fixtures**: Available in network and model modules (testFixtures source set)
 - **Parallel execution**: Tests run with 2 parallel forks, forking every 100 tests
 
@@ -231,8 +237,11 @@ The codebase contains TODO comments in several classes related to MQTT protocol
 ### When Writing Tests
 1. Use Spock framework (Groovy syntax)
 2. Place in `src/test/groovy/` maintaining package structure
-3. Test fixtures can be used from network and model modules
-4. Tests automatically run with preview features enabled
+3. Prefer `UnitSpecification` for low-level unit tests and `IntegrationServiceSpecification` when existing shared services, mocked connections, or broker-oriented fixtures are useful
+4. For async service APIs returning `Mono` or `CompletionStage`, use `fromAsync(...)` for returned values and `waitForAsync(...)` when only completion matters
+5. Close manually created services or clients in `cleanup:` when they own resources or background work
+6. Test fixtures can be used from network and model modules
+7. Tests automatically run with preview features enabled
 
 ### When Modifying Build Configuration
 - Root `build.gradle`: Only for repository-wide settings and custom tasks

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

@@ -17,22 +17,16 @@
 import javasabr.mqtt.service.ClientIdRegistry;
 import javasabr.mqtt.service.ConnectionService;
 import javasabr.mqtt.service.MessageOutFactoryService;
-import javasabr.mqtt.service.PublishDeliveringService;
-import javasabr.mqtt.service.PublishReceivingService;
-import javasabr.mqtt.service.RetainMessageService;
 import javasabr.mqtt.service.SubscriptionService;
 import javasabr.mqtt.service.TopicService;
 import javasabr.mqtt.service.handler.client.ExternalNetworkMqttUserReleaseHandler;
 import javasabr.mqtt.service.impl.DefaultConnectionService;
 import javasabr.mqtt.service.impl.DefaultMessageOutFactoryService;
 import javasabr.mqtt.service.impl.DefaultMqttConnectionFactory;
-import javasabr.mqtt.service.impl.DefaultPublishDeliveringService;
-import javasabr.mqtt.service.impl.DefaultPublishReceivingService;
 import javasabr.mqtt.service.impl.DefaultTopicService;
 import javasabr.mqtt.service.impl.DisabledAuthorizationService;
 import javasabr.mqtt.service.impl.ExternalNetworkMqttUserFactory;
 import javasabr.mqtt.service.impl.InMemoryClientIdRegistry;
-import javasabr.mqtt.service.impl.InMemoryRetainMessageService;
 import javasabr.mqtt.service.impl.InMemorySubscriptionService;
 import javasabr.mqtt.service.message.handler.MqttInMessageHandler;
 import javasabr.mqtt.service.message.handler.impl.ConnectInMqttInMessageHandler;
@@ -47,14 +41,22 @@
 import javasabr.mqtt.service.message.out.factory.Mqtt311MessageOutFactory;
 import javasabr.mqtt.service.message.out.factory.Mqtt5MessageOutFactory;
 import javasabr.mqtt.service.message.out.factory.MqttMessageOutFactory;
-import javasabr.mqtt.service.publish.handler.MqttPublishInMessageHandler;
-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.mqtt.service.publish.IncomingPublishRouter;
+import javasabr.mqtt.service.publish.PublishDataStorage;
+import javasabr.mqtt.service.publish.PublishDispatcher;
+import javasabr.mqtt.service.publish.RetainPublishService;
+import javasabr.mqtt.service.publish.impl.DefaultIncomingPublishRouter;
+import javasabr.mqtt.service.publish.impl.DefaultPublishDispatcher;
+import javasabr.mqtt.service.publish.impl.InMemoryPublishDataStorage;
+import javasabr.mqtt.service.publish.impl.InMemoryRetainPublishService;
+import javasabr.mqtt.service.publish.processor.IncomingPublishProcessor;
+import javasabr.mqtt.service.publish.processor.Qos0IncomingPublishProcessor;
+import javasabr.mqtt.service.publish.processor.Qos1IncomingPublishProcessor;
+import javasabr.mqtt.service.publish.processor.Qos2IncomingPublishProcessor;
+import javasabr.mqtt.service.publish.sender.Qos0SubscriberPublishSender;
+import javasabr.mqtt.service.publish.sender.Qos1SubscriberPublishSender;
+import javasabr.mqtt.service.publish.sender.Qos2SubscriberPublishSender;
+import javasabr.mqtt.service.publish.sender.SubscriberPublishSender;
 import javasabr.mqtt.service.session.MqttSessionService;
 import javasabr.mqtt.service.session.impl.InMemoryMqttSessionService;
 import javasabr.rlib.network.NetworkFactory;
@@ -122,10 +124,15 @@ AuthorizationService authorizationService() {
   SubscriptionService subscriptionService(AuthorizationService authorizationService) {
     return new InMemorySubscriptionService(authorizationService);
   }
+
+  @Bean
+  PublishDataStorage publishDataStorage() {
+    return new InMemoryPublishDataStorage();
+  }
 
   @Bean
-  RetainMessageService retainMessageService() {
-    return new InMemoryRetainMessageService();
+  RetainPublishService retainMessageService() {
+    return new InMemoryRetainPublishService();
   }
 
   @Bean
@@ -176,15 +183,17 @@ MqttInMessageHandler publishCompleteMqttInMessageHandler(MessageOutFactoryServic
 
   @Bean
   MqttInMessageHandler publishMqttInMessageHandler(
-      PublishReceivingService publishReceivingService,
+      IncomingPublishRouter incomingPublishRouter,
       MessageOutFactoryService messageOutFactoryService,
       TopicService topicService,
-      AuthorizationService authorizationService) {
+      AuthorizationService authorizationService,
+      PublishDataStorage publishDataStorage) {
     return new PublishMqttInMessageHandler(
-        publishReceivingService,
+        incomingPublishRouter,
         messageOutFactoryService,
         topicService,
-        authorizationService);
+        authorizationService,
+        publishDataStorage);
   }
 
   @Bean
@@ -207,14 +216,14 @@ MqttInMessageHandler subscribeMqttInMessageHandler(
       SubscriptionService subscriptionService,
       MessageOutFactoryService messageOutFactoryService,
       TopicService topicService,
-      RetainMessageService retainMessageService,
-      PublishDeliveringService publishDeliveringService) {
+      RetainPublishService retainPublishService,
+      PublishDispatcher publishDispatcher) {
     return new SubscribeMqttInMessageHandler(
         subscriptionService,
         messageOutFactoryService,
         topicService,
-        retainMessageService,
-        publishDeliveringService);
+        retainPublishService, 
+        publishDispatcher);
   }
 
   @Bean
@@ -234,69 +243,69 @@ ConnectionService externalMqttConnectionService(Collection<? extends MqttInMessa
   }
 
   @Bean
-  MqttPublishOutMessageHandler qos0MqttPublishOutMessageHandler(MessageOutFactoryService messageOutFactoryService) {
-    return new Qos0MqttPublishOutMessageHandler(messageOutFactoryService);
+  SubscriberPublishSender qos0SubscriberPublishSender(MessageOutFactoryService messageOutFactoryService) {
+    return new Qos0SubscriberPublishSender(messageOutFactoryService);
   }
 
   @Bean
-  MqttPublishOutMessageHandler qos1MqttPublishOutMessageHandler(MessageOutFactoryService messageOutFactoryService) {
-    return new Qos1MqttPublishOutMessageHandler(messageOutFactoryService);
+  SubscriberPublishSender qos1SubscriberPublishSender(MessageOutFactoryService messageOutFactoryService) {
+    return new Qos1SubscriberPublishSender(messageOutFactoryService);
   }
 
   @Bean
-  MqttPublishOutMessageHandler qos2MqttPublishOutMessageHandler(MessageOutFactoryService messageOutFactoryService) {
-    return new Qos2MqttPublishOutMessageHandler(messageOutFactoryService);
+  SubscriberPublishSender qos2SubscriberPublishSender(MessageOutFactoryService messageOutFactoryService) {
+    return new Qos2SubscriberPublishSender(messageOutFactoryService);
   }
 
   @Bean
-  PublishDeliveringService publishDeliveringService(
-      Collection<? extends MqttPublishOutMessageHandler> knownPublishOutHandlers) {
-    return new DefaultPublishDeliveringService(knownPublishOutHandlers);
+  PublishDispatcher publishDispatcher(
+      Collection<? extends SubscriberPublishSender> knownSubscriberPublishSenders) {
+    return new DefaultPublishDispatcher(knownSubscriberPublishSenders);
   }
 
   @Bean
-  MqttPublishInMessageHandler qos0MqttPublishInMessageHandler(
+  IncomingPublishProcessor qos0IncomingPublishProcessor(
       SubscriptionService subscriptionService,
-      PublishDeliveringService publishDeliveringService,
+      PublishDispatcher publishDispatcher,
       MessageOutFactoryService messageOutFactoryService,
-      RetainMessageService retainMessageService) {
-    return new Qos0MqttPublishInMessageHandler(
+      RetainPublishService retainPublishService) {
+    return new Qos0IncomingPublishProcessor(
         subscriptionService,
-        publishDeliveringService,
-        messageOutFactoryService,
-        retainMessageService);
+        publishDispatcher,
+        messageOutFactoryService, 
+        retainPublishService);
   }
 
   @Bean
-  MqttPublishInMessageHandler qos1MqttPublishInMessageHandler(
+  IncomingPublishProcessor qos1IncomingPublishProcessor(
       SubscriptionService subscriptionService,
-      PublishDeliveringService publishDeliveringService,
+      PublishDispatcher publishDispatcher,
       MessageOutFactoryService messageOutFactoryService,
-      RetainMessageService retainMessageService) {
-    return new Qos1MqttPublishInMessageHandler(
+      RetainPublishService retainPublishService) {
+    return new Qos1IncomingPublishProcessor(
         subscriptionService,
-        publishDeliveringService,
+        publishDispatcher,
         messageOutFactoryService,
-        retainMessageService);
+        retainPublishService);
   }
 
   @Bean
-  MqttPublishInMessageHandler qos2MqttPublishInMessageHandler(
+  IncomingPublishProcessor qos2IncomingPublishProcessor(
       SubscriptionService subscriptionService,
-      PublishDeliveringService publishDeliveringService,
+      PublishDispatcher publishDispatcher,
       MessageOutFactoryService messageOutFactoryService,
-      RetainMessageService retainMessageService) {
-    return new Qos2MqttPublishInMessageHandler(
-        subscriptionService,
-        publishDeliveringService,
+      RetainPublishService retainPublishService) {
+    return new Qos2IncomingPublishProcessor(
+        subscriptionService, 
+        publishDispatcher,
         messageOutFactoryService,
-        retainMessageService);
+        retainPublishService);
   }
 
   @Bean
-  PublishReceivingService publishReceivingService(
-      Collection<? extends MqttPublishInMessageHandler> knownPublishInHandlers) {
-    return new DefaultPublishReceivingService(knownPublishInHandlers);
+  IncomingPublishRouter IncomingPublishRouter(
+      Collection<? extends IncomingPublishProcessor> knownIncomingPublishProcessors) {
+    return new DefaultIncomingPublishRouter(knownIncomingPublishProcessors);
   }
 
   @Bean

application/src/test/resources/log4j2-test.xml

@@ -15,13 +15,13 @@
     <Logger name="javasabr.mqtt.model.network.impl.DefaultMqttSession" level="DEBUG" additivity="false">
       <AppenderRef ref="BrokerConsoleTest"/>
     </Logger>
-    <Logger name="javasabr.mqtt.service.impl.DefaultPublishReceivingService" level="DEBUG" additivity="false">
+    <Logger name="javasabr.mqtt.service.publish.impl.DefaultIncomingPublishRouter" level="DEBUG" additivity="false">
       <AppenderRef ref="BrokerConsoleTest"/>
     </Logger>
-    <Logger name="javasabr.mqtt.service.publish.handler.impl.AbstractMqttPublishInMessageHandler" level="DEBUG" additivity="false">
+    <Logger name="javasabr.mqtt.service.publish.processor.AbstractIncomingPublishProcessor" level="DEBUG" additivity="false">
       <AppenderRef ref="BrokerConsoleTest"/>
     </Logger>
-    <Logger name="javasabr.mqtt.service.publish.handler.impl.Qos1MqttPublishInMessageHandler" level="DEBUG" additivity="false">
+    <Logger name="javasabr.mqtt.service.publish.processor.Qos1IncomingPublishProcessor" level="DEBUG" additivity="false">
       <AppenderRef ref="BrokerConsoleTest"/>
     </Logger>
     <Logger name="javasabr.mqtt.service.session.impl.InMemoryMqttSessionService" level="DEBUG" additivity="false">