[broker-30] Implement delivering retained messages

acl-groovy-dsl/src/main/resources/acl.gdsl

@@ -33,7 +33,7 @@ contributor(context(scope: scriptScope())) {
       && !enclosingCall("allOf") && !enclosingCall("anyOf")) {
     method name: 'topicFilter',
         type: 'javasabr.mqtt.service.acl.builder.SubscribeRuleBuilder',
-        params: [string: 'javasabr.mqtt.model.acl.matcher.ValueMatcher<String>...'],
+        params: [string: 'javasabr.mqtt.model.acl.matcher.TopicFilterMatcher...'],
         doc: 'Set of topic filters matching by rule'
     method name: 'match',
         type: 'javasabr.mqtt.model.acl.matcher.TopicFilterMatcher',

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

@@ -20,6 +20,7 @@
 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;
@@ -28,6 +29,7 @@
 import javasabr.mqtt.service.impl.DefaultMqttConnectionFactory;
 import javasabr.mqtt.service.impl.DefaultPublishDeliveringService;
 import javasabr.mqtt.service.impl.DefaultPublishReceivingService;
+import javasabr.mqtt.service.impl.DefaultRetainMessageService;
 import javasabr.mqtt.service.impl.DefaultTopicService;
 import javasabr.mqtt.service.impl.DisabledAuthorizationService;
 import javasabr.mqtt.service.impl.ExternalNetworkMqttUserFactory;
@@ -114,8 +116,13 @@ AuthorizationService authorizationService() {
   }
 
   @Bean
-  SubscriptionService subscriptionService() {
-    return new InMemorySubscriptionService();
+  SubscriptionService subscriptionService(RetainMessageService retainMessageService) {
+    return new InMemorySubscriptionService(retainMessageService);
+  }
+
+  @Bean
+  RetainMessageService retainMessageService(PublishDeliveringService publishDeliveringService) {
+    return new DefaultRetainMessageService(publishDeliveringService);
   }
 
   @Bean
@@ -210,7 +217,8 @@ MqttInMessageHandler publishMqttInMessageHandler(
     return new PublishMqttInMessageHandler(
         publishReceivingService,
         messageOutFactoryService,
-        topicService, authorizationService,
+        topicService,
+        authorizationService,
         fieldValidators);
   }
 
@@ -254,24 +262,18 @@ 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);
+  MqttPublishOutMessageHandler qos1MqttPublishOutMessageHandler(MessageOutFactoryService messageOutFactoryService) {
+    return new Qos1MqttPublishOutMessageHandler(messageOutFactoryService);
   }
 
   @Bean
-  MqttPublishOutMessageHandler qos2MqttPublishOutMessageHandler(
-      SubscriptionService subscriptionService,
-      MessageOutFactoryService messageOutFactoryService) {
-    return new Qos2MqttPublishOutMessageHandler(subscriptionService, messageOutFactoryService);
+  MqttPublishOutMessageHandler qos2MqttPublishOutMessageHandler(MessageOutFactoryService messageOutFactoryService) {
+    return new Qos2MqttPublishOutMessageHandler(messageOutFactoryService);
   }
 
   @Bean
@@ -284,33 +286,39 @@ PublishDeliveringService publishDeliveringService(
   MqttPublishInMessageHandler qos0MqttPublishInMessageHandler(
       SubscriptionService subscriptionService,
       PublishDeliveringService publishDeliveringService,
-      MessageOutFactoryService messageOutFactoryService) {
+      MessageOutFactoryService messageOutFactoryService,
+      RetainMessageService retainMessageService) {
     return new Qos0MqttPublishInMessageHandler(
         subscriptionService,
         publishDeliveringService,
-        messageOutFactoryService);
+        messageOutFactoryService,
+        retainMessageService);
   }
 
   @Bean
   MqttPublishInMessageHandler qos1MqttPublishInMessageHandler(
       SubscriptionService subscriptionService,
       PublishDeliveringService publishDeliveringService,
-      MessageOutFactoryService messageOutFactoryService) {
+      MessageOutFactoryService messageOutFactoryService,
+      RetainMessageService retainMessageService) {
     return new Qos1MqttPublishInMessageHandler(
         subscriptionService,
         publishDeliveringService,
-        messageOutFactoryService);
+        messageOutFactoryService,
+        retainMessageService);
   }
 
   @Bean
   MqttPublishInMessageHandler qos2MqttPublishInMessageHandler(
       SubscriptionService subscriptionService,
       PublishDeliveringService publishDeliveringService,
-      MessageOutFactoryService messageOutFactoryService) {
+      MessageOutFactoryService messageOutFactoryService,
+      RetainMessageService retainMessageService) {
     return new Qos2MqttPublishInMessageHandler(
         subscriptionService,
         publishDeliveringService,
-        messageOutFactoryService);
+        messageOutFactoryService,
+        retainMessageService);
   }
 
   @Bean

core-service/build.gradle

@@ -12,4 +12,5 @@ dependencies {
 
   testImplementation projects.testSupport
   testImplementation testFixtures(projects.network)
-}
+  testImplementation testFixtures(projects.model)
+}

core-service/src/main/java/javasabr/mqtt/service/RetainMessageService.java

@@ -0,0 +1,15 @@
+package javasabr.mqtt.service;
+
+import javasabr.mqtt.model.publishing.Publish;
+import javasabr.mqtt.model.subscriber.SingleSubscriber;
+import javasabr.mqtt.model.subscriber.Subscriber;
+import javasabr.mqtt.model.subscription.Subscription;
+import javasabr.mqtt.service.publish.handler.PublishHandlingResult;
+import javasabr.rlib.collections.array.Array;
+
+public interface RetainMessageService {
+
+  void retainMessage(Publish publish);
+
+  void deliverRetainedMessages(Subscriber subscriber);
+}

core-service/src/main/java/javasabr/mqtt/service/SubscriptionService.java

@@ -15,7 +15,7 @@
  * Subscription service
  */
 public interface SubscriptionService {
-  
+
   default Array<SingleSubscriber> findSubscribers(TopicName topicName) {
     return findSubscribersTo(MutableArray.ofType(SingleSubscriber.class), topicName);
   }

core-service/src/main/java/javasabr/mqtt/service/impl/DefaultRetainMessageService.java

@@ -0,0 +1,47 @@
+package javasabr.mqtt.service.impl;
+
+import javasabr.mqtt.model.publishing.Publish;
+import javasabr.mqtt.model.subscriber.SingleSubscriber;
+import javasabr.mqtt.model.subscriber.Subscriber;
+import javasabr.mqtt.model.subscription.Subscription;
+import javasabr.mqtt.model.topic.tree.ConcurrentRetainedMessageTree;
+import javasabr.mqtt.service.PublishDeliveringService;
+import javasabr.mqtt.service.RetainMessageService;
+import javasabr.mqtt.service.publish.handler.PublishHandlingResult;
+import javasabr.rlib.collections.array.Array;
+import javasabr.rlib.collections.array.MutableArray;
+import lombok.AccessLevel;
+import lombok.experimental.FieldDefaults;
+
+@FieldDefaults(level = AccessLevel.PRIVATE, makeFinal = true)
+public class DefaultRetainMessageService implements RetainMessageService {
+
+  PublishDeliveringService publishDeliveringService;
+  ConcurrentRetainedMessageTree retainedMessageTree;
+
+  public DefaultRetainMessageService(PublishDeliveringService publishDeliveringService) {
+    this.publishDeliveringService = publishDeliveringService;
+    this.retainedMessageTree = new ConcurrentRetainedMessageTree();
+  }
+
+  @Override
+  public void retainMessage(Publish publish) {
+    retainedMessageTree.retainMessage(publish);
+
+  }
+
+  @Override
+  public void deliverRetainedMessages(Subscriber subscriber) {
+    SingleSubscriber singleSubscriber = subscriber.resolveSingle();
+    Subscription subscription = singleSubscriber.subscription();
+    boolean retainAsPublished = subscription.retainAsPublished();
+    Array<Publish> retainedMessages = retainedMessageTree.getRetainedMessage(subscription.topicFilter());
+    MutableArray<PublishHandlingResult> result = MutableArray.ofType(PublishHandlingResult.class);
+    for (Publish message : retainedMessages) {
+      if (!retainAsPublished) {
+        message = message.withoutRetain();
+      }
+      publishDeliveringService.startDelivering(message, singleSubscriber);
+    }
+  }
+}

core-service/src/main/java/javasabr/mqtt/service/impl/InMemorySubscriptionService.java

@@ -1,27 +1,33 @@
 package javasabr.mqtt.service.impl;
 
+import static javasabr.mqtt.model.SubscribeRetainHandling.SEND;
+import static javasabr.mqtt.model.SubscribeRetainHandling.SEND_IF_SUBSCRIPTION_DOES_NOT_EXIST;
 import static javasabr.mqtt.model.reason.code.UnsubscribeAckReasonCode.NO_SUBSCRIPTION_EXISTED;
 import static javasabr.mqtt.model.reason.code.UnsubscribeAckReasonCode.SUCCESS;
 
 import javasabr.mqtt.model.MqttClientConnectionConfig;
 import javasabr.mqtt.model.MqttUser;
+import javasabr.mqtt.model.SubscribeRetainHandling;
 import javasabr.mqtt.model.reason.code.SubscribeAckReasonCode;
 import javasabr.mqtt.model.reason.code.UnsubscribeAckReasonCode;
 import javasabr.mqtt.model.session.ActiveSubscriptions;
 import javasabr.mqtt.model.session.MqttSession;
 import javasabr.mqtt.model.subscriber.SingleSubscriber;
+import javasabr.mqtt.model.subscriber.Subscriber;
 import javasabr.mqtt.model.subscriber.tree.ConcurrentSubscriberTree;
 import javasabr.mqtt.model.subscription.Subscription;
 import javasabr.mqtt.model.topic.SharedTopicFilter;
 import javasabr.mqtt.model.topic.TopicFilter;
 import javasabr.mqtt.model.topic.TopicName;
+import javasabr.mqtt.service.RetainMessageService;
 import javasabr.mqtt.service.SubscriptionService;
 import javasabr.rlib.collections.array.Array;
 import javasabr.rlib.collections.array.ArrayFactory;
 import javasabr.rlib.collections.array.MutableArray;
 import lombok.AccessLevel;
 import lombok.CustomLog;
 import lombok.experimental.FieldDefaults;
+import org.jspecify.annotations.Nullable;
 
 /**
  * In memory subscription service based on {@link ConcurrentSubscriberTree}
@@ -30,10 +36,12 @@
 @FieldDefaults(level = AccessLevel.PRIVATE, makeFinal = true)
 public class InMemorySubscriptionService implements SubscriptionService {
 
+  RetainMessageService retainMessageService;
   ConcurrentSubscriberTree subscriberTree;
 
-  public InMemorySubscriptionService() {
+  public InMemorySubscriptionService(RetainMessageService retainMessageService) {
     this.subscriberTree = new ConcurrentSubscriberTree();
+    this.retainMessageService = retainMessageService;
   }
 
   @Override
@@ -71,9 +79,13 @@ private SubscribeAckReasonCode addSubscription(MqttUser user, MqttSession sessio
       return SubscribeAckReasonCode.WILDCARD_SUBSCRIPTIONS_NOT_SUPPORTED;
     }
     ActiveSubscriptions activeSubscriptions = session.activeSubscriptions();
-    SingleSubscriber previous = subscriberTree.subscribe(user, subscription);
-    if (previous != null) {
-      activeSubscriptions.remove(previous.subscription());
+    SingleSubscriber newSubscriber = new SingleSubscriber(user, subscription);
+    SingleSubscriber previousSubscriber = subscriberTree.subscribe(newSubscriber);
+    if (previousSubscriber != null) {
+      activeSubscriptions.remove(previousSubscriber.subscription());
+    }
+    if (isRetainHandlingRequired(subscription, previousSubscriber)) {
+        retainMessageService.deliverRetainedMessages(newSubscriber);
     }
     activeSubscriptions.add(subscription);
     return subscription.qos().subscribeAckReasonCode();
@@ -125,7 +137,18 @@ public void restoreSubscriptions(MqttUser user, MqttSession session) {
         .activeSubscriptions()
         .subscriptions();
     for (Subscription subscription : subscriptions) {
-      subscriberTree.subscribe(user, subscription);
+      subscriberTree.subscribe(new SingleSubscriber(user, subscription));
+    }
+  }
+
+  private static boolean isRetainHandlingRequired(
+      Subscription newSubscription,
+      @Nullable Subscriber previousSubscriber) {
+    if (newSubscription.topicFilter().isShared() || !newSubscription.qos().isValid()) {
+      return false;
     }
+    SubscribeRetainHandling retainHandling = newSubscription.retainHandling();
+    return retainHandling == SEND || (retainHandling == SEND_IF_SUBSCRIPTION_DOES_NOT_EXIST
+                                          && previousSubscriber == null);
   }
 }

core-service/src/main/java/javasabr/mqtt/service/publish/handler/impl/AbstractMqttPublishInMessageHandler.java

@@ -10,6 +10,7 @@
 import javasabr.mqtt.network.user.NetworkMqttUser;
 import javasabr.mqtt.service.MessageOutFactoryService;
 import javasabr.mqtt.service.PublishDeliveringService;
+import javasabr.mqtt.service.RetainMessageService;
 import javasabr.mqtt.service.SubscriptionService;
 import javasabr.mqtt.service.publish.handler.MqttPublishInMessageHandler;
 import javasabr.mqtt.service.publish.handler.PublishHandlingResult;
@@ -29,6 +30,7 @@ public abstract class AbstractMqttPublishInMessageHandler<U extends NetworkMqttU
   SubscriptionService subscriptionService;
   PublishDeliveringService publishDeliveringService;
   MessageOutFactoryService messageOutFactoryService;
+  RetainMessageService retainMessageService;
 
   @Override
   public void handle(NetworkMqttUser user, Publish publish) {
@@ -52,6 +54,9 @@ protected boolean validateImpl(U user, NetworkMqttSession session, Publish publi
   }
 
   protected void handleImpl(U user, NetworkMqttSession session, Publish publish) {
+    if (publish.retained()) {
+      retainMessageService.retainMessage(publish);
+    }
     TopicName topicName = publish.topicName();
     Array<SingleSubscriber> subscribers = subscriptionService.findSubscribers(topicName);
     if (subscribers.isEmpty()) {

core-service/src/main/java/javasabr/mqtt/service/publish/handler/impl/AbstractMqttPublishOutMessageHandler.java

@@ -7,7 +7,6 @@
 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;
@@ -23,7 +22,6 @@ public abstract class AbstractMqttPublishOutMessageHandler<U extends NetworkMqtt
     implements MqttPublishOutMessageHandler {
 
   Class<U> expectedUserType;
-  SubscriptionService subscriptionService;
   MessageOutFactoryService messageOutFactoryService;
 
   @Override

core-service/src/main/java/javasabr/mqtt/service/publish/handler/impl/Qos0MqttPublishInMessageHandler.java

@@ -10,15 +10,22 @@
 import javasabr.mqtt.network.session.NetworkMqttSession;
 import javasabr.mqtt.service.MessageOutFactoryService;
 import javasabr.mqtt.service.PublishDeliveringService;
+import javasabr.mqtt.service.RetainMessageService;
 import javasabr.mqtt.service.SubscriptionService;
 
 public class Qos0MqttPublishInMessageHandler extends AbstractMqttPublishInMessageHandler<ExternalNetworkMqttUser> {
 
   public Qos0MqttPublishInMessageHandler(
       SubscriptionService subscriptionService,
       PublishDeliveringService publishDeliveringService,
-      MessageOutFactoryService messageOutFactoryService) {
-    super(ExternalNetworkMqttUser.class, subscriptionService, publishDeliveringService, messageOutFactoryService);
+      MessageOutFactoryService messageOutFactoryService,
+      RetainMessageService retainMessageService) {
+    super(
+        ExternalNetworkMqttUser.class,
+        subscriptionService,
+        publishDeliveringService,
+        messageOutFactoryService,
+        retainMessageService);
   }
 
   @Override

core-service/src/main/java/javasabr/mqtt/service/publish/handler/impl/Qos0MqttPublishOutMessageHandler.java

@@ -6,16 +6,12 @@
 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 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