[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/build.gradle

@@ -24,4 +24,4 @@ tasks.withType(GroovyCompile).configureEach {
 configurations.each {
   it.exclude group: "org.slf4j", module: "slf4j-log4j12"
   it.exclude group: "org.springframework.boot", module: "spring-boot-starter-logging"
-}
+}

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

@@ -93,13 +93,14 @@ CredentialSource credentialSource(
   @Bean
   AuthenticationService authenticationService(
       CredentialSource credentialSource,
-      @Value("${authentication.allow.anonymous:false}") boolean allowAnonymousAuth) {
+      @Value("${authentication.allow.anonymous:false}")
+      boolean allowAnonymousAuth) {
     return new SimpleAuthenticationService(credentialSource, allowAnonymousAuth);
   }
 
   @Bean
-  SubscriptionService subscriptionService() {
-    return new InMemorySubscriptionService();
+  SubscriptionService subscriptionService(PublishDeliveringService publishDeliveringService) {
+    return new InMemorySubscriptionService(publishDeliveringService);
   }
 
   @Bean
@@ -153,10 +154,7 @@ MqttInMessageHandler publishMqttInMessageHandler(
       PublishReceivingService publishReceivingService,
       MessageOutFactoryService messageOutFactoryService,
       TopicService topicService) {
-    return new PublishMqttInMessageHandler(
-        publishReceivingService,
-        messageOutFactoryService,
-        topicService);
+    return new PublishMqttInMessageHandler(publishReceivingService, messageOutFactoryService, topicService);
   }
 
   @Bean
@@ -187,10 +185,7 @@ MqttInMessageHandler unsubscribeMqttInMessageHandler(
       SubscriptionService subscriptionService,
       MessageOutFactoryService messageOutFactoryService,
       TopicService topicService) {
-    return new UnsubscribeMqttInMessageHandler(
-        subscriptionService,
-        messageOutFactoryService,
-        topicService);
+    return new UnsubscribeMqttInMessageHandler(subscriptionService, messageOutFactoryService, topicService);
   }
 
   @Bean
@@ -199,24 +194,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
@@ -230,32 +219,23 @@ MqttPublishInMessageHandler qos0MqttPublishInMessageHandler(
       SubscriptionService subscriptionService,
       PublishDeliveringService publishDeliveringService,
       MessageOutFactoryService messageOutFactoryService) {
-    return new Qos0MqttPublishInMessageHandler(
-        subscriptionService,
-        publishDeliveringService,
-        messageOutFactoryService);
+    return new Qos0MqttPublishInMessageHandler(subscriptionService, publishDeliveringService, messageOutFactoryService);
   }
 
   @Bean
   MqttPublishInMessageHandler qos1MqttPublishInMessageHandler(
       SubscriptionService subscriptionService,
       PublishDeliveringService publishDeliveringService,
       MessageOutFactoryService messageOutFactoryService) {
-    return new Qos1MqttPublishInMessageHandler(
-        subscriptionService,
-        publishDeliveringService,
-        messageOutFactoryService);
+    return new Qos1MqttPublishInMessageHandler(subscriptionService, publishDeliveringService, messageOutFactoryService);
   }
 
   @Bean
   MqttPublishInMessageHandler qos2MqttPublishInMessageHandler(
       SubscriptionService subscriptionService,
       PublishDeliveringService publishDeliveringService,
       MessageOutFactoryService messageOutFactoryService) {
-    return new Qos2MqttPublishInMessageHandler(
-        subscriptionService,
-        publishDeliveringService,
-        messageOutFactoryService);
+    return new Qos2MqttPublishInMessageHandler(subscriptionService, publishDeliveringService, messageOutFactoryService);
   }
 
   @Bean
@@ -280,22 +260,10 @@ MqttServerConnectionConfig externalConnectionConfig(Environment env) {
             "mqtt.external.connection.max.message.size",
             int.class,
             MqttProperties.MAXIMUM_MESSAGE_SIZE_DEFAULT),
-        env.getProperty(
-            "mqtt.external.connection.max.string.length",
-            int.class,
-            MqttProperties.MAXIMUM_STRING_LENGTH),
-        env.getProperty(
-            "mqtt.external.connection.max.binary.size",
-            int.class,
-            MqttProperties.MAXIMUM_BINARY_SIZE),
-        env.getProperty(
-            "mqtt.external.connection.max.topic.levels",
-            int.class,
-            MqttProperties.MAXIMUM_TOPIC_LEVELS),
-        env.getProperty(
-            "mqtt.external.connection.min.keep.alive",
-            int.class,
-            MqttProperties.SERVER_KEEP_ALIVE_DEFAULT),
+        env.getProperty("mqtt.external.connection.max.string.length", int.class, MqttProperties.MAXIMUM_STRING_LENGTH),
+        env.getProperty("mqtt.external.connection.max.binary.size", int.class, MqttProperties.MAXIMUM_BINARY_SIZE),
+        env.getProperty("mqtt.external.connection.max.topic.levels", int.class, MqttProperties.MAXIMUM_TOPIC_LEVELS),
+        env.getProperty("mqtt.external.connection.min.keep.alive", int.class, MqttProperties.SERVER_KEEP_ALIVE_DEFAULT),
         env.getProperty(
             "mqtt.external.connection.receive.maximum",
             int.class,
@@ -360,7 +328,8 @@ NetworkMqttUserFactory externalClientFactory(NetworkMqttUserReleaseHandler exter
   MqttConnectionFactory externalConnectionFactory(
       MqttServerConnectionConfig externalServerConnectionConfig,
       NetworkMqttUserFactory mqttUserFactory,
-      @Value("${mqtt.external.connection.max.packets.by.read:100}") int maxPacketsByRead) {
+      @Value("${mqtt.external.connection.max.packets.by.read:100}")
+      int maxPacketsByRead) {
     return new DefaultMqttConnectionFactory(externalServerConnectionConfig, mqttUserFactory, maxPacketsByRead);
   }
 

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/PublishDeliveringService.java

@@ -3,8 +3,11 @@
 import javasabr.mqtt.model.publishing.Publish;
 import javasabr.mqtt.model.subscriber.SingleSubscriber;
 import javasabr.mqtt.service.publish.handler.PublishHandlingResult;
+import javasabr.rlib.collections.array.Array;
 
 public interface PublishDeliveringService {
 
   PublishHandlingResult startDelivering(Publish publish, SingleSubscriber subscriber);
+
+  Array<PublishHandlingResult> deliverRetainedMessages(SingleSubscriber 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/DefaultPublishDeliveringService.java

@@ -1,12 +1,17 @@
 package javasabr.mqtt.service.impl;
 
 import java.util.Collection;
+import java.util.function.Function;
 import javasabr.mqtt.model.QoS;
 import javasabr.mqtt.model.publishing.Publish;
 import javasabr.mqtt.model.subscriber.SingleSubscriber;
+import javasabr.mqtt.model.subscription.Subscription;
+import javasabr.mqtt.model.topic.tree.ConcurrentRetainedMessageTree;
 import javasabr.mqtt.service.PublishDeliveringService;
 import javasabr.mqtt.service.publish.handler.MqttPublishOutMessageHandler;
 import javasabr.mqtt.service.publish.handler.PublishHandlingResult;
+import javasabr.rlib.collections.array.Array;
+import javasabr.rlib.collections.array.MutableArray;
 import lombok.AccessLevel;
 import lombok.CustomLog;
 import lombok.experimental.FieldDefaults;
@@ -18,6 +23,7 @@ public class DefaultPublishDeliveringService implements PublishDeliveringService
 
   @Nullable
   MqttPublishOutMessageHandler[] publishOutMessageHandlers;
+  ConcurrentRetainedMessageTree retainedMessageTree;
 
   public DefaultPublishDeliveringService(
       Collection<? extends MqttPublishOutMessageHandler> knownPublishOutHandlers) {
@@ -39,13 +45,34 @@ public DefaultPublishDeliveringService(
       }
       handlers[qos.level()] = knownPublishOutHandler;
     }
-
+    this.retainedMessageTree = new ConcurrentRetainedMessageTree();
     this.publishOutMessageHandlers = handlers;
     log.info(publishOutMessageHandlers, DefaultPublishDeliveringService::buildServiceDescription);
   }
 
   @Override
   public PublishHandlingResult startDelivering(Publish publish, SingleSubscriber subscriber) {
+    if (publish.retained()) {
+      Subscription subscription = subscriber.subscription();
+      boolean retainAsPublished = subscription.retainAsPublished();
+      Function<Publish, Publish> transformer = retainAsPublished ? Function.identity() : Publish::withoutRetain;
+      retainedMessageTree.retainMessage(transformer.apply(publish));
+    }
+    return startDeliveringWithoutRetain(publish, subscriber);
+  }
+
+  @Override
+  public Array<PublishHandlingResult> deliverRetainedMessages(SingleSubscriber subscriber) {
+    Subscription subscription = subscriber.subscription();
+    Array<Publish> retainedMessages = retainedMessageTree.getRetainedMessage(subscription.topicFilter());
+    MutableArray<PublishHandlingResult> result = MutableArray.ofType(PublishHandlingResult.class);
+    for (Publish message : retainedMessages) {
+      result.add(startDeliveringWithoutRetain(message, subscriber));
+    }
+    return Array.copyOf(result);
+  }
+
+  private PublishHandlingResult startDeliveringWithoutRetain(Publish publish, SingleSubscriber subscriber) {
     try {
       //noinspection DataFlowIssue
       return publishOutMessageHandlers[subscriber.qos().level()].handle(publish, subscriber);

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

@@ -1,10 +1,13 @@
 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.QoS;
 import javasabr.mqtt.model.reason.code.SubscribeAckReasonCode;
 import javasabr.mqtt.model.reason.code.UnsubscribeAckReasonCode;
 import javasabr.mqtt.model.session.ActiveSubscriptions;
@@ -15,7 +18,9 @@
 import javasabr.mqtt.model.topic.SharedTopicFilter;
 import javasabr.mqtt.model.topic.TopicFilter;
 import javasabr.mqtt.model.topic.TopicName;
+import javasabr.mqtt.service.PublishDeliveringService;
 import javasabr.mqtt.service.SubscriptionService;
+import javasabr.mqtt.service.publish.handler.PublishHandlingResult;
 import javasabr.rlib.collections.array.Array;
 import javasabr.rlib.collections.array.ArrayFactory;
 import javasabr.rlib.collections.array.MutableArray;
@@ -30,10 +35,12 @@
 @FieldDefaults(level = AccessLevel.PRIVATE, makeFinal = true)
 public class InMemorySubscriptionService implements SubscriptionService {
 
+  PublishDeliveringService publishDeliveringService;
   ConcurrentSubscriberTree subscriberTree;
 
-  public InMemorySubscriptionService() {
+  public InMemorySubscriptionService(PublishDeliveringService publishDeliveringService) {
     this.subscriberTree = new ConcurrentSubscriberTree();
+    this.publishDeliveringService = publishDeliveringService;
   }
 
   @Override
@@ -75,8 +82,14 @@ private SubscribeAckReasonCode addSubscription(MqttUser user, MqttSession sessio
     if (previous != null) {
       activeSubscriptions.remove(previous.subscription());
     }
+    QoS subscriptionQoS = subscription.qos();
+    if (subscriptionQoS.ordinal() <= 2 && (subscription.retainHandling() == SEND ||
+        (subscription.retainHandling() == SEND_IF_SUBSCRIPTION_DOES_NOT_EXIST && previous == null))) {
+      sendRetainedMessages(user, subscription);
+    }
     activeSubscriptions.add(subscription);
-    return subscription.qos().subscribeAckReasonCode();
+    return subscriptionQoS
+        .subscribeAckReasonCode();
   }
 
   @Override
@@ -128,4 +141,24 @@ public void restoreSubscriptions(MqttUser user, MqttSession session) {
       subscriberTree.subscribe(user, subscription);
     }
   }
+
+  private void sendRetainedMessages(MqttUser user, Subscription subscription) {
+    int count = 0;
+    String clientId = user.clientId();
+    PublishHandlingResult errorResult = null;
+    SingleSubscriber singleSubscriber = new SingleSubscriber(user, subscription);
+    var results = publishDeliveringService.deliverRetainedMessages(singleSubscriber);
+    for (PublishHandlingResult result : results) {
+      if (result.error()) {
+        errorResult = result;
+      } else if (result == PublishHandlingResult.SUCCESS) {
+        count++;
+      }
+      if (errorResult != null) {
+        log.debug(clientId, errorResult, "[%s] Error occurred [%s] during sending retained messages"::formatted);
+      } else {
+        log.debug(clientId, count, "[%s] Delivering of [%s] retained message has been started"::formatted);
+      }
+    }
+  }
 }

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/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

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