feat: implement MQTT wildcard subscription matching

shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Publish.java

@@ -121,7 +121,7 @@ private String byteBufToString(final ByteBuf byteBuf) {
     }
 
     private void send(final String topic, final ByteBuf payload, final int packetId) {
-        List<Channel> channels = Singleton.INST.get(SubscribeRepository.class).get(topic);
+        List<Channel> channels = Singleton.INST.get(SubscribeRepository.class).getChannelsByTopic(topic);
         //// todo thread pool
         channels.parallelStream().forEach(channel -> {
             if (channel.isActive()) {

shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Subscribe.java

@@ -61,36 +61,40 @@ public void subscribe(final ChannelHandlerContext ctx, final MqttSubscribeMessag
         int packetId = msg.variableHeader().messageId();
 
         //// todo Regular match
-        List<String> ackTopics = mqttTopicSubscriptions
+        List<MqttTopicSubscription> validSubscriptions = mqttTopicSubscriptions
                 .stream()
                 .filter(topicSub -> topicSub.qualityOfService() != FAILURE)
-                .map(MqttTopicSubscription::topicName)
+                .filter(topicSub -> TopicMatcher.isValidFilter(topicSub.topicName()))
                 .collect(Collectors.toList());
 
-        Singleton.INST.get(SubscribeRepository.class).add(ctx.channel(), mqttTopicSubscriptions);
+        Singleton.INST.get(SubscribeRepository.class).add(ctx.channel(), validSubscriptions);
 
-        for (String ackTopic : ackTopics) {
-            String message = Singleton.INST.get(TopicRepository.class).get(ackTopic);
+        for (MqttTopicSubscription subscription : validSubscriptions) {
+            String message = Singleton.INST.get(TopicRepository.class).get(subscription.topicName());
             if (StringUtils.isNotEmpty(message)) {
-                sendSubMessage(ackTopic, message, packetId, channel);
+                sendSubMessage(subscription.topicName(), message, packetId, channel);
             }
         }
 
-        sendSubAckMessage(packetId, ackTopics, channel);
+        sendSubAckMessage(packetId, mqttTopicSubscriptions, channel);
     }
 
     /**
      * call back request of message.
      * @param packetId packetId
-     * @param ackTopics ackTopics
+     * @param subscriptions subscriptions
      * @param channel channel
      */
-    private void sendSubAckMessage(final int packetId, final List<String> ackTopics, final Channel channel) {
+    private void sendSubAckMessage(final int packetId, final List<MqttTopicSubscription> subscriptions, final Channel channel) {
 
         List<Integer> qos = new ArrayList<>();
-        for (int i = 0; i < ackTopics.size(); i++) {
-            // default qos 0
-            qos.add(MqttQoS.AT_MOST_ONCE.value());
+        for (MqttTopicSubscription subscription : subscriptions) {
+            // invalid topic filters are rejected with 0x80, otherwise default qos 0
+            if (TopicMatcher.isValidFilter(subscription.topicName())) {
+                qos.add(MqttQoS.AT_MOST_ONCE.value());
+            } else {
+                qos.add(MqttQoS.FAILURE.value());
+            }
         }
 
         MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.SUBACK, false, AT_MOST_ONCE,

shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/TopicMatcher.java

@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shenyu.protocol.mqtt;
+
+import java.util.Objects;
+
+/**
+ * MQTT topic filter matching per MQTT-4.7.
+ *
+ * <p>+ matches exactly one topic level.</p>
+ *
+ * <p># matches any number of subsequent levels (must appear at the end of the filter).</p>
+ */
+public final class TopicMatcher {
+
+    private TopicMatcher() {
+    }
+
+    /**
+     * Check whether a topic filter matches a topic name.
+     *
+     * @param filter the subscription topic filter (may contain + and # wildcards)
+     * @param topic  the published topic name (no wildcards)
+     * @return true if the filter matches the topic
+     */
+    public static boolean matches(final String filter, final String topic) {
+        if (Objects.isNull(filter) || Objects.isNull(topic)) {
+            return false;
+        }
+
+        // $ topics must not be matched by wildcards at the first level
+        if (topic.startsWith("$") && filter.length() > 0 && (filter.charAt(0) == '+' || filter.charAt(0) == '#')) {
+            return false;
+        }
+
+        String[] filterLevels = filter.split("/", -1);
+        String[] topicLevels = topic.split("/", -1);
+
+        int filterLen = filterLevels.length;
+        int topicLen = topicLevels.length;
+
+        for (int i = 0; i < filterLen; i++) {
+            String f = filterLevels[i];
+
+            if ("#".equals(f)) {
+                // MQTT-4.7.1-2: # matches any number of levels including the parent level
+                return i == filterLen - 1;
+            }
+
+            if (i >= topicLen) {
+                return false;
+            }
+
+            if (!"+".equals(f) && !f.equals(topicLevels[i])) {
+                return false;
+            }
+        }
+
+        return filterLen == topicLen;
+    }
+
+    /**
+     * Validate a topic filter per MQTT-4.7.1: wildcards must occupy an entire
+     * level, and # must be the last level. Filters must not be empty or
+     * contain the null character.
+     *
+     * @param filter the subscription topic filter
+     * @return true if the filter is valid
+     */
+    public static boolean isValidFilter(final String filter) {
+        if (Objects.isNull(filter) || filter.isEmpty() || filter.indexOf((char) 0) >= 0) {
+            return false;
+        }
+        String[] levels = filter.split("/", -1);
+        for (int i = 0; i < levels.length; i++) {
+            String level = levels[i];
+            if (level.indexOf('+') >= 0 || level.indexOf('#') >= 0) {
+                if (level.length() > 1) {
+                    return false;
+                }
+                if ("#".equals(level) && i != levels.length - 1) {
+                    return false;
+                }
+            }
+        }
+        return true;
+    }
+}

shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/repositories/SubscribeRepository.java

@@ -20,11 +20,15 @@
 import io.netty.channel.Channel;
 import io.netty.handler.codec.mqtt.MqttTopicSubscription;
 import org.apache.commons.collections4.CollectionUtils;
+import org.apache.shenyu.protocol.mqtt.TopicMatcher;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.ArrayList;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
@@ -97,4 +101,34 @@ public List<Channel> get(final String topic) {
         return TOPIC_CHANNEL_FACTORY.getOrDefault(topic, new CopyOnWriteArrayList<>());
     }
 
+    /**
+     * Get channels whose subscription filter matches the published topic.
+     * Supports MQTT wildcards: + (single-level) and # (multi-level).
+     *
+     * @param topic the published topic name
+     * @return channels subscribed to matching topic filters
+     */
+    public List<Channel> getChannelsByTopic(final String topic) {
+        // MQTT requires at most one delivery per publish per client, so dedupe
+        // channels when overlapping filters (e.g. sport/# and #) both match.
+        Set<Channel> result = new LinkedHashSet<>();
+
+        // fast path: exact subscription, no wildcard scan needed
+        List<Channel> exactMatch = TOPIC_CHANNEL_FACTORY.get(topic);
+        if (Objects.nonNull(exactMatch)) {
+            result.addAll(exactMatch);
+        }
+
+        for (Map.Entry<String, List<Channel>> entry : TOPIC_CHANNEL_FACTORY.entrySet()) {
+            String filter = entry.getKey();
+            if (filter.equals(topic) || filter.indexOf('+') < 0 && filter.indexOf('#') < 0) {
+                continue;
+            }
+            if (TopicMatcher.matches(filter, topic)) {
+                result.addAll(entry.getValue());
+            }
+        }
+        return new ArrayList<>(result);
+    }
+
 }

shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/SubscribeTest.java

@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shenyu.protocol.mqtt;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.handler.codec.mqtt.MqttMessageBuilders;
+import io.netty.handler.codec.mqtt.MqttQoS;
+import io.netty.handler.codec.mqtt.MqttSubAckMessage;
+import io.netty.handler.codec.mqtt.MqttSubscribeMessage;
+import org.apache.shenyu.common.utils.Singleton;
+import org.apache.shenyu.protocol.mqtt.repositories.SubscribeRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.TopicRepository;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.function.BooleanSupplier;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+class SubscribeTest {
+
+    @Test
+    void testSubscribeRejectsInvalidTopicFilter() throws InterruptedException {
+        EmbeddedChannel channel = new EmbeddedChannel();
+        channel.pipeline().addLast(new ChannelInboundHandlerAdapter());
+        ChannelHandlerContext ctx = channel.pipeline().firstContext();
+        MqttSubscribeMessage msg = MqttMessageBuilders.subscribe()
+                .messageId(1)
+                .addSubscription(MqttQoS.AT_MOST_ONCE, "sport/#")
+                .addSubscription(MqttQoS.AT_MOST_ONCE, "bad#filter")
+                .build();
+
+        SubscribeRepository repository = new SubscribeRepository();
+        Singleton.INST.single(SubscribeRepository.class, repository);
+        Singleton.INST.single(TopicRepository.class, new TopicRepository());
+
+        new Subscribe().subscribe(ctx, msg);
+        awaitUntil(() -> repository.get("sport/#").contains(channel));
+        assertTrue(repository.get("bad#filter").isEmpty());
+
+        MqttSubAckMessage subAck = channel.readOutbound();
+        assertNotNull(subAck);
+        List<Integer> granted = subAck.payload().grantedQoSLevels();
+        assertEquals(2, granted.size());
+        assertEquals(0, granted.get(0).intValue());
+        assertEquals(0x80, granted.get(1).intValue());
+
+        repository.remove(Collections.singletonList("sport/#"), channel);
+        awaitUntil(() -> repository.get("sport/#").isEmpty());
+        channel.finishAndReleaseAll();
+    }
+
+    private void awaitUntil(final BooleanSupplier condition) throws InterruptedException {
+        long deadline = System.currentTimeMillis() + 5000;
+        while (!condition.getAsBoolean()) {
+            if (System.currentTimeMillis() >= deadline) {
+                fail("condition not met within timeout");
+            }
+            Thread.sleep(10);
+        }
+    }
+}