Initial implementation of shared subscriptions

ChangeLog.txt

@@ -1,4 +1,6 @@
 Version 0.18-SNAPSHOT:
+   [feature] shared subscriptions:
+     - Initial implementation of shared subscription subscribe and publish part. (#796)
    [fix] Implements requirements on reserved topics (starts with $). Implements the matching rules and avoid to proceed with processing on client's publishes on those topics (#793)
    [feature] Handle will delay interval and MQTT5's Will optional properties (#770)
    [fix] Handle empty collector batches in PostOffice (#777)

broker/pom.xml

@@ -19,7 +19,7 @@
              https://github.com/netty/netty/blob/netty-4.1.100.Final/pom.xml#L650 -->
         <netty.tcnative.version>2.0.61.Final</netty.tcnative.version>
         <paho.version>1.2.5</paho.version>
-        <hivemqclient.version>1.3.0</hivemqclient.version>
+        <hivemqclient.version>1.3.3</hivemqclient.version>
         <h2.version>2.1.212</h2.version>
     </properties>
 

broker/src/main/java/io/moquette/broker/Authorizator.java

@@ -25,6 +25,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.function.Function;
 
 import static io.moquette.broker.Utils.messageId;
 import static io.netty.handler.codec.mqtt.MqttQoS.FAILURE;
@@ -39,6 +40,18 @@ final class Authorizator {
         this.policy = policy;
     }
 
+
+    List<MqttTopicSubscription> verifyAlsoSharedTopicsReadAccess(String clientID, String username, MqttSubscribeMessage msg) {
+        return verifyTopicsReadAccessWithTopicExtractor(clientID, username, msg, Authorizator::extractShareTopic);
+    }
+
+    private static Topic extractShareTopic(String s) {
+        if (SharedSubscriptionUtils.isSharedSubscription(s)) {
+            return Topic.asTopic(SharedSubscriptionUtils.extractFilterFromShared(s));
+        }
+        return Topic.asTopic(s);
+    }
+
     /**
      * @param clientID
      *            the clientID
@@ -49,33 +62,43 @@ final class Authorizator {
      * @return the list of verified topics for the given subscribe message.
      */
     List<MqttTopicSubscription> verifyTopicsReadAccess(String clientID, String username, MqttSubscribeMessage msg) {
+        return verifyTopicsReadAccessWithTopicExtractor(clientID, username, msg, Topic::asTopic);
+    }
+
+    private List<MqttTopicSubscription> verifyTopicsReadAccessWithTopicExtractor(String clientID, String username,
+                                                 MqttSubscribeMessage msg, Function<String, Topic> topicExtractor) {
         List<MqttTopicSubscription> ackTopics = new ArrayList<>();
 
         final int messageId = messageId(msg);
         for (MqttTopicSubscription req : msg.payload().topicSubscriptions()) {
-            Topic topic = new Topic(req.topicName());
-            if (!policy.canRead(topic, username, clientID)) {
-                // send SUBACK with 0x80, the user hasn't credentials to read the topic
-                LOG.warn("Client does not have read permissions on the topic username: {}, messageId: {}, " +
-                         "topic: {}", username, messageId, topic);
-                ackTopics.add(new MqttTopicSubscription(topic.toString(), FAILURE));
-            } else {
-                MqttQoS qos;
-                if (topic.isValid()) {
-                    LOG.debug("Client will be subscribed to the topic username: {}, messageId: {}, topic: {}",
-                        username, messageId, topic);
-                    qos = req.qualityOfService();
-                } else {
-                    LOG.warn("Topic filter is not valid username: {}, messageId: {}, topic: {}",
-                        username, messageId, topic);
-                    qos = FAILURE;
-                }
-                ackTopics.add(new MqttTopicSubscription(topic.toString(), qos));
-            }
+            Topic topic = topicExtractor.apply(req.topicName());
+            final MqttQoS qos = getQoSCheckingAlsoPermissionsOnTopic(clientID, username, messageId, topic,
+                req.qualityOfService());
+            ackTopics.add(new MqttTopicSubscription(req.topicName(), qos));
         }
         return ackTopics;
     }
 
+    private MqttQoS getQoSCheckingAlsoPermissionsOnTopic(String clientID, String username, int messageId,
+                                                         Topic topic, MqttQoS requestedQoS) {
+        if (policy.canRead(topic, username, clientID)) {
+            if (topic.isValid()) {
+                LOG.debug("Client will be subscribed to the topic username: {}, messageId: {}, topic: {}",
+                    username, messageId, topic);
+                return requestedQoS;
+            }
+
+            LOG.warn("Topic filter is not valid username: {}, messageId: {}, topic: {}",
+                username, messageId, topic);
+            return FAILURE;
+        }
+
+        // send SUBACK with 0x80, the user hasn't credentials to read the topic
+        LOG.warn("Client does not have read permissions on the topic username: {}, messageId: {}, " +
+                 "topic: {}", username, messageId, topic);
+        return FAILURE;
+    }
+
     /**
      * Ask the authorization policy if the topic can be used in a publish.
      *

broker/src/main/java/io/moquette/broker/MQTTConnection.java

@@ -690,10 +690,10 @@ void sendPublish(MqttPublishMessage publishMsg) {
         final String topicName = publishMsg.variableHeader().topicName();
         MqttQoS qos = publishMsg.fixedHeader().qosLevel();
         if (LOG.isTraceEnabled()) {
-            LOG.trace("Sending PUBLISH({}) message. MessageId={}, topic={}, payload={}", qos, packetId, topicName,
-                DebugUtils.payload2Str(publishMsg.payload()));
+            LOG.trace("Sending PUBLISH({}) message. MessageId={}, topic={}, payload={} to {}", qos, packetId, topicName,
+                DebugUtils.payload2Str(publishMsg.payload()), getClientId());
         } else {
-            LOG.debug("Sending PUBLISH({}) message. MessageId={}, topic={}", qos, packetId, topicName);
+            LOG.debug("Sending PUBLISH({}) message. MessageId={}, topic={} to {}", qos, packetId, topicName, getClientId());
         }
         sendIfWritableElseDrop(publishMsg);
     }
@@ -703,7 +703,7 @@ void sendIfWritableElseDrop(MqttMessage msg) {
             LOG.debug("OUT {}", msg.fixedHeader().messageType());
         }
         if (channel.isWritable()) {
-
+            LOG.debug("Sending message {} on the wire to {}", msg.fixedHeader().messageType(), getClientId());
             // Sending to external, retain a duplicate. Just retain is not
             // enough, since the receiver must have full control.
             Object retainedDup = msg;
@@ -718,6 +718,8 @@ void sendIfWritableElseDrop(MqttMessage msg) {
                 channelFuture = channel.write(retainedDup);
             }
             channelFuture.addListener(FIRE_EXCEPTION_ON_FAILURE);
+        } else {
+            LOG.debug("Dropping message {} from the wire, msg: {}", msg.fixedHeader().messageType(), msg);
         }
     }
 

broker/src/main/java/io/moquette/broker/PostOffice.java

@@ -16,6 +16,7 @@
 package io.moquette.broker;
 
 import io.moquette.broker.scheduler.ScheduledExpirationService;
+import io.moquette.broker.subscriptions.ShareName;
 import io.moquette.interception.BrokerInterceptor;
 import io.moquette.broker.subscriptions.ISubscriptionsDirectory;
 import io.moquette.broker.subscriptions.Subscription;
@@ -32,7 +33,6 @@
 import io.netty.handler.codec.mqtt.MqttSubscribeMessage;
 import io.netty.handler.codec.mqtt.MqttTopicSubscription;
 import io.netty.util.ReferenceCountUtil;
-import org.apache.commons.codec.binary.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -43,6 +43,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CompletableFuture;
@@ -273,38 +274,77 @@ public void wipeExistingScheduledWill(String clientId) {
         sessionRepository.deleteWill(clientId);
     }
 
+    // Used for internal purposes of subscribeClientToTopics method
+    private static final class SharedSubscriptionData {
+        final ShareName name;
+        final Topic topicFilter;
+        final MqttQoS requestedQoS;
+
+        private SharedSubscriptionData(ShareName name, Topic topicFilter, MqttQoS requestedQoS) {
+            Objects.requireNonNull(name);
+            Objects.requireNonNull(topicFilter);
+            Objects.requireNonNull(requestedQoS);
+            this.name = name;
+            this.topicFilter = topicFilter;
+            this.requestedQoS = requestedQoS;
+        }
+
+        static SharedSubscriptionData fromMqttSubscription(MqttTopicSubscription sub) {
+            return new SharedSubscriptionData(new ShareName(SharedSubscriptionUtils.extractShareName(sub.topicName())),
+                Topic.asTopic(SharedSubscriptionUtils.extractFilterFromShared(sub.topicName())), sub.qualityOfService());
+        }
+    }
+
     public void subscribeClientToTopics(MqttSubscribeMessage msg, String clientID, String username,
                                         MQTTConnection mqttConnection) {
         // verify which topics of the subscribe ongoing has read access permission
         int messageID = messageId(msg);
         final Session session = sessionRegistry.retrieve(clientID);
 
+        final List<SharedSubscriptionData> sharedSubscriptions;
+
         if (mqttConnection.isProtocolVersion5()) {
-            for (MqttTopicSubscription topicFilter : msg.payload().topicSubscriptions()) {
-                if (isSharedSubscription(topicFilter.topicName())) {
-                    final String shareName = extractShareName(topicFilter.topicName());
-                    if (!validateShareName(shareName)) {
-                        // this is a malformed packet, MQTT-4.13.1-1, disconnect it
-                        LOG.info("{} used an invalid shared subscription name {}, disconnecting", clientID, shareName);
-                        session.disconnectFromBroker();
-                        return;
-                    }
-                }
+            sharedSubscriptions = msg.payload().topicSubscriptions().stream()
+                .filter(sub -> SharedSubscriptionUtils.isSharedSubscription(sub.topicName()))
+                .map(SharedSubscriptionData::fromMqttSubscription)
+                .collect(Collectors.toList());
+
+            Optional<SharedSubscriptionData> invalidSharedSubscription = sharedSubscriptions.stream()
+                .filter(subData -> !SharedSubscriptionUtils.validateShareName(subData.name.toString()))
+                .findFirst();
+            if (invalidSharedSubscription.isPresent()) {
+                // this is a malformed packet, MQTT-4.13.1-1, disconnect it
+                LOG.info("{} used an invalid shared subscription name {}, disconnecting", clientID, invalidSharedSubscription.get().name);
+                session.disconnectFromBroker();
+                return;
             }
+        } else {
+            sharedSubscriptions = Collections.emptyList();
+        }
+
+        List<MqttTopicSubscription> ackTopics;
+        if (mqttConnection.isProtocolVersion5()) {
+            ackTopics = authorizator.verifyAlsoSharedTopicsReadAccess(clientID, username, msg);
+        } else {
+            ackTopics = authorizator.verifyTopicsReadAccess(clientID, username, msg);
         }
-        List<MqttTopicSubscription> ackTopics = authorizator.verifyTopicsReadAccess(clientID, username, msg);
         MqttSubAckMessage ackMessage = doAckMessageFromValidateFilters(ackTopics, messageID);
 
-        // store topics subscriptions in session
+        // store topics of non-shared subscriptions in session
         List<Subscription> newSubscriptions = ackTopics.stream()
-            .filter(req -> req.qualityOfService() != FAILURE)
-            .map(req -> {
-                final Topic topic = new Topic(req.topicName());
-                return new Subscription(clientID, topic, req.qualityOfService());
+            .filter(sub -> sub.qualityOfService() != FAILURE)
+            .filter(sub -> !SharedSubscriptionUtils.isSharedSubscription(sub.topicName()))
+            .map(sub -> {
+                final Topic topic = new Topic(sub.topicName());
+                return new Subscription(clientID, topic, sub.qualityOfService());
             }).collect(Collectors.toList());
 
         for (Subscription subscription : newSubscriptions) {
-            subscriptions.add(subscription);
+            subscriptions.add(subscription.getClientId(), subscription.getTopicFilter(), subscription.getRequestedQos());
+        }
+
+        for (SharedSubscriptionData sharedSubData : sharedSubscriptions) {
+            subscriptions.addShared(clientID, sharedSubData.name, sharedSubData.topicFilter, sharedSubData.requestedQoS);
         }
 
         // add the subscriptions to Session
@@ -320,33 +360,6 @@ public void subscribeClientToTopics(MqttSubscribeMessage msg, String clientID, S
         }
     }
 
-    /**
-     * @return the share name in the topic filter of format $share/{shareName}/{topicFilter}
-     * */
-    // VisibleForTesting
-    protected static String extractShareName(String sharedTopicFilter) {
-        int afterShare = "$share/".length();
-        int endOfShareName = sharedTopicFilter.indexOf('/', afterShare);
-        return sharedTopicFilter.substring(afterShare, endOfShareName);
-    }
-
-    /**
-     * @return true if shareName is well formed, is at least one characted and doesn't contain wildcard matchers
-     * */
-    private boolean validateShareName(String shareName) {
-        // MQTT-4.8.2-1 MQTT-4.8.2-2, must be longer than 1 char and do not contain + or #
-        Objects.requireNonNull(shareName);
-        return shareName.length() > 0 && !shareName.contains("+") && !shareName.contains("#");
-    }
-
-    /**
-     * @return true if topic filter is shared format
-     * */
-    private static boolean isSharedSubscription(String topicFilter) {
-        Objects.requireNonNull(topicFilter, "topicFilter can't be null");
-        return topicFilter.startsWith("$share/");
-    }
-
     private void publishRetainedMessagesForSubscriptions(String clientID, List<Subscription> newSubscriptions) {
         Session targetSession = this.sessionRegistry.retrieve(clientID);
         for (Subscription subscription : newSubscriptions) {

broker/src/main/java/io/moquette/broker/Server.java

@@ -169,7 +169,7 @@ public void startServer(IConfig config) throws IOException {
      * @throws IOException in case of any IO Error.
      */
     public void startServer(IConfig config, List<? extends InterceptHandler> handlers) throws IOException {
-        LOG.debug("Starting moquette integration using IConfig instance and intercept handlers");
+        LOG.debug("Starting Moquette integration using IConfig instance and intercept handlers");
         startServer(config, handlers, null, null, null);
     }
 

broker/src/main/java/io/moquette/broker/SharedSubscriptionUtils.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright (c) 2012-2023 The original author or authors
+ * ------------------------------------------------------
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Apache License v2.0 which accompanies this distribution.
+ *
+ * The Eclipse Public License is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ *
+ * The Apache License v2.0 is available at
+ * http://www.opensource.org/licenses/apache2.0.php
+ *
+ * You may elect to redistribute this code under either of these licenses.
+ */
+package io.moquette.broker;
+
+import java.util.Objects;
+
+/**
+ * Utility class that collects common utils methods for shared subscription topic parsing
+ * */
+class SharedSubscriptionUtils {
+
+    /**
+     * @return the share name in the topic filter of format $share/{shareName}/{topicFilter}
+     * */
+    // VisibleForTesting
+    protected static String extractShareName(String sharedTopicFilter) {
+        int afterShare = "$share/".length();
+        int endOfShareName = sharedTopicFilter.indexOf('/', afterShare);
+        return sharedTopicFilter.substring(afterShare, endOfShareName);
+    }
+
+    /**
+     * @return the filter part from full topic filter of format $share/{shareName}/{topicFilter}
+     * */
+    // VisibleForTesting
+    protected static String extractFilterFromShared(String fullSharedTopicFilter) {
+        int afterShare = "$share/".length();
+        int endOfShareName = fullSharedTopicFilter.indexOf('/', afterShare);
+        return fullSharedTopicFilter.substring(endOfShareName + 1);
+    }
+
+    /**
+     * @return true if topic filter is shared format
+     * */
+    protected static boolean isSharedSubscription(String topicFilter) {
+        Objects.requireNonNull(topicFilter, "topicFilter can't be null");
+        return topicFilter.startsWith("$share/");
+    }
+
+    /**
+     * @return true if shareName is well-formed, is at least one characted and doesn't contain wildcard matchers
+     * */
+    protected static boolean validateShareName(String shareName) {
+        // MQTT-4.8.2-1 MQTT-4.8.2-2, must be longer than 1 char and do not contain + or #
+        Objects.requireNonNull(shareName);
+        return shareName.length() > 0 && !shareName.contains("+") && !shareName.contains("#");
+    }
+}