[ISSUE apache#44] Support shared subscription

Mu-L · Jun 9, 2023 · 1d0b869 · 1d0b869
1 parent 501ade3
commit 1d0b869
Show file tree

Hide file tree

Showing 13 changed files with 264 additions and 3 deletions.
diff --git a/mqtt-common/src/main/java/org/apache/rocketmq/mqtt/common/facade/LmqQueueStore.java b/mqtt-common/src/main/java/org/apache/rocketmq/mqtt/common/facade/LmqQueueStore.java
@@ -20,6 +20,8 @@
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.message.MessageQueue;
 import org.apache.rocketmq.mqtt.common.model.Message;
 import org.apache.rocketmq.mqtt.common.model.PullResult;
 import org.apache.rocketmq.mqtt.common.model.Queue;
@@ -48,6 +50,19 @@ public interface LmqQueueStore {
      */
     CompletableFuture<PullResult> pullMessage(String firstTopic, Queue queue, QueueOffset queueOffset, long count);
 
+    /**
+     * pop messages
+     *
+     * @param consumerGroup
+     * @param firstTopic
+     * @param queue
+     * @param count
+     * @return
+     */
+    CompletableFuture<PullResult> popMessage(String consumerGroup, String firstTopic, Queue queue, long count);
+
+    void popAck(String lmqTopic, String consumerGroup, Message message);
+
     /**
      * pull last messages
      *

diff --git a/mqtt-common/src/main/java/org/apache/rocketmq/mqtt/common/model/Constants.java b/mqtt-common/src/main/java/org/apache/rocketmq/mqtt/common/model/Constants.java
@@ -23,6 +23,7 @@ public class Constants {
 
     public static final String PLUS_SIGN = "+";
     public static final String NUMBER_SIGN = "#";
+    public static final String DOLLAR_SIGN = "$";
     public static final String COLON = ":";
 
     public static final String P2P = "/p2p/";
@@ -55,5 +56,7 @@ public class Constants {
     public static final byte CTRL_2 = '\u0002';
 
     public static final String NOT_FOUND = "NOT_FOUND";
+    public static final String SHARED_PREFIX = DOLLAR_SIGN + "share";
+    public static final String EMPTY_SHARE_NAME = "";
 
 }
diff --git a/mqtt-common/src/main/java/org/apache/rocketmq/mqtt/common/model/PullResult.java b/mqtt-common/src/main/java/org/apache/rocketmq/mqtt/common/model/PullResult.java
@@ -24,6 +24,7 @@
 public class PullResult {
     public static final int PULL_SUCCESS = 301;
     public static final int PULL_OFFSET_MOVED = 302;
+    public static final int NO_NEW_MSG = 303;
     private int code = PULL_SUCCESS;
     private String remark;
     private List<Message> messageList;

diff --git a/mqtt-common/src/main/java/org/apache/rocketmq/mqtt/common/model/Queue.java b/mqtt-common/src/main/java/org/apache/rocketmq/mqtt/common/model/Queue.java
@@ -51,6 +51,10 @@ public boolean isP2p() {
         return TopicUtils.isP2pTopic(queueName);
     }
 
+    public boolean isShare() {
+        return TopicUtils.isSharedSubscription(queueName);
+    }
+
     public long getQueueId() {
         return queueId;
     }

diff --git a/mqtt-common/src/main/java/org/apache/rocketmq/mqtt/common/model/Subscription.java b/mqtt-common/src/main/java/org/apache/rocketmq/mqtt/common/model/Subscription.java
@@ -85,6 +85,17 @@ public int hashCode() {
         return topicFilter != null ? topicFilter.hashCode() : 0;
     }
 
+    public boolean isShare() {
+        return TopicUtils.isSharedSubscription(topicFilter);
+    }
+
+    public String getSharedName() {
+        if (!isShare()) {
+            return null;
+        }
+        return TopicUtils.getSharedName(topicFilter);
+    }
+
     public String getTopicFilter() {
         return topicFilter;
     }

diff --git a/mqtt-common/src/main/java/org/apache/rocketmq/mqtt/common/util/TopicUtils.java b/mqtt-common/src/main/java/org/apache/rocketmq/mqtt/common/util/TopicUtils.java
@@ -17,6 +17,8 @@
 
 package org.apache.rocketmq.mqtt.common.util;
 
+import java.util.Arrays;
+
 import org.apache.commons.lang3.StringUtils;
 import org.apache.rocketmq.mqtt.common.model.Constants;
 import org.apache.rocketmq.mqtt.common.model.MqttTopic;
@@ -116,6 +118,9 @@ public static MqttTopic decode(String topics) {
         if (topics.startsWith(Constants.MQTT_TOPIC_DELIMITER)) {
             topics = topics.substring(1);
         }
+        if (topics.startsWith(Constants.SHARED_PREFIX)) {
+            topics = TopicUtils.getSharedTopicFilter(topics);
+        }
         String topic;
         String secondTopic = null;
         int index = topics.indexOf(Constants.MQTT_TOPIC_DELIMITER, 1);
@@ -185,4 +190,24 @@ public static String wrapLmq(String firstTopic, String secondTopic) {
     public static String wrapP2pLmq(String clientId) {
         return normalizeTopic(Constants.P2P + clientId);
     }
+
+    // shared subscription topic filter format: $share/{ShareName}/{filter}
+    public static boolean isSharedSubscription(String topicFilter) {
+        if (StringUtils.isEmpty(topicFilter)) {
+            return false;
+        }
+        if (!topicFilter.startsWith(Constants.SHARED_PREFIX)) {
+            return false;
+        }
+        String[] arr = topicFilter.split(Constants.MQTT_TOPIC_DELIMITER);
+        return arr.length > 2;
+    }
+
+    public static String getSharedName(String topicFilter) {
+        return topicFilter.split(Constants.MQTT_TOPIC_DELIMITER)[1];
+    }
+
+    public static String getSharedTopicFilter(String topicFilter) {
+        return topicFilter.split(Constants.MQTT_TOPIC_DELIMITER, 3)[2];
+    }
 }
diff --git a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/session/Session.java b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/session/Session.java
@@ -290,6 +290,9 @@ public boolean addSendingMessages(Subscription subscription, Queue queue, List<M
         if (!subscriptions.containsKey(subscription.getTopicFilter())) {
             return false;
         }
+        if (subscription.isShare()) {
+            return true;
+        }
         if (!sendingMessages.containsKey(subscription)) {
             sendingMessages.putIfAbsent(subscription, new ConcurrentHashMap<>(16));
         }

diff --git a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/session/infly/PushAction.java b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/session/infly/PushAction.java
@@ -21,6 +21,8 @@
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.mqtt.common.facade.LmqQueueStore;
 import org.apache.rocketmq.mqtt.common.model.Message;
 import org.apache.rocketmq.mqtt.common.model.Queue;
 import org.apache.rocketmq.mqtt.common.model.Subscription;
@@ -56,6 +58,8 @@ public class PushAction {
     @Resource
     private ConnectConf connectConf;
 
+    @Resource
+    private LmqQueueStore lmqQueueStore;
 
     public void messageArrive(Session session, Subscription subscription, Queue queue) {
         if (session == null) {
@@ -163,6 +167,9 @@ public void write(Session session, Message message, int mqttId, int qos, Subscri
             if (subscription.isRetry()) {
                 message.setRetry(message.getRetry() + 1);
                 logger.warn("retryPush:{},{},{}", session.getClientId(), message.getMsgId(), message.getRetry());
+            } else if (subscription.isShare()) {
+                String lmqTopic = MixAll.LMQ_PREFIX + StringUtils.replace(message.getOriginTopic(), "/","%");
+                lmqQueueStore.popAck(lmqTopic, subscription.getSharedName(), message);
             }
         });
     }

diff --git a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/session/loop/QueueCache.java b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/session/loop/QueueCache.java
@@ -111,7 +111,7 @@ public void refreshCache(Pair<Queue, Session> pair) {
         if (queue == null) {
             return;
         }
-        if (queue.isP2p() || queue.isRetry()) {
+        if (queue.isP2p() || queue.isRetry() || queue.isShare()) {
             return;
         }
         addLoadEvent(queue, pair.getRight());
@@ -188,6 +188,13 @@ public PullResultStatus pullMessage(Session session, Subscription subscription,
             callbackResult(pullResult, callBackResult);
             return DONE;
         }
+
+        if (subscription.isShare()) {
+            CompletableFuture<PullResult> pullResult = lmqQueueStore.popMessage(subscription.getSharedName(), toFirstTopic(subscription), queue, count);
+            callbackResult(pullResult, callBackResult);
+            return DONE;
+        }
+
         CacheEntry cacheEntry = cache.getIfPresent(queue);
         if (cacheEntry == null) {
             StatUtil.addPv("NoPullCache", 1);

diff --git a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/session/loop/SessionLoopImpl.java b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/session/loop/SessionLoopImpl.java
@@ -34,6 +34,7 @@
 import org.apache.rocketmq.mqtt.common.model.Subscription;
 import org.apache.rocketmq.mqtt.common.model.WillMessage;
 import org.apache.rocketmq.mqtt.common.util.SpringUtils;
+import org.apache.rocketmq.mqtt.common.util.TopicUtils;
 import org.apache.rocketmq.mqtt.cs.channel.ChannelInfo;
 import org.apache.rocketmq.mqtt.cs.channel.ChannelManager;
 import org.apache.rocketmq.mqtt.cs.config.ConnectConf;
@@ -393,6 +394,7 @@ public void notifyPullMessage(Session session, Subscription subscription, Queue
                 throw new RuntimeException(
                         "invalid notifyPullMessage, subscription is null, but queue is not null," + session.getClientId());
             }
+            logger.info("session loop impl doing notifyPullMessage queueFresh.freshQueue({}, {}})", session, subscription);
             queueFresh.freshQueue(session, subscription);
             pullMessage(session, subscription, queue);
             return;
@@ -539,7 +541,6 @@ private void doPull(PullEvent pullEvent) {
                         pushAction.messageArrive(session, subscription, queue);
                     }
                 } else if (PullResult.PULL_OFFSET_MOVED == pullResult.getCode()) {
-                    queueOffset.setOffset(pullResult.getNextQueueOffset().getOffset());
                     queueOffset.setOffset(pullResult.getNextQueueOffset().getOffset());
                     session.markPersistOffsetFlag(true);
                     pullMessage(session, subscription, queue);

diff --git a/mqtt-ds/src/main/java/org/apache/rocketmq/mqtt/ds/notify/NotifyManager.java b/mqtt-ds/src/main/java/org/apache/rocketmq/mqtt/ds/notify/NotifyManager.java
@@ -117,6 +117,8 @@ public void init() throws MQClientException {
 
     private void refresh() throws MQClientException {
         Set<String> tmp = metaPersistManager.getAllFirstTopics();
+        logger.info("Notify Manager is refreshing, all first topic is " + tmp);
+
         if (tmp == null || tmp.isEmpty()) {
             return;
         }