feat: implement MQTT Last Will and Testament (LWT)

shenyu-protocol/shenyu-protocol-mqtt/pom.xml

@@ -52,6 +52,18 @@
             <artifactId>junit-jupiter</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-junit-jupiter</artifactId>
+            <version>${mockito.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <version>${mockito.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
 </project>

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

@@ -26,6 +26,7 @@
 import org.apache.commons.lang3.StringUtils;
 import org.apache.shenyu.common.utils.Singleton;
 import org.apache.shenyu.protocol.mqtt.repositories.ChannelRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.WillRepository;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -74,6 +75,17 @@ public void connect(final ChannelHandlerContext ctx, final MqttConnectMessage ms
 
         // record connect
         Singleton.INST.get(ChannelRepository.class).add(ctx.channel(), clientId);
+
+        // store will if present
+        if (msg.variableHeader().isWillFlag()) {
+            WillRepository.WillEntry will = new WillRepository.WillEntry(
+                    msg.payload().willTopic(),
+                    msg.payload().willMessageInBytes(),
+                    msg.variableHeader().willQos(),
+                    msg.variableHeader().isWillRetain());
+            Singleton.INST.get(WillRepository.class).add(ctx.channel(), will);
+        }
+
         MqttConnAckMessage ackMessage = MqttMessageBuilders.connAck()
                 .returnCode(MqttConnectReturnCode.CONNECTION_ACCEPTED)
                 .sessionPresent(true)

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

@@ -21,6 +21,7 @@
 import io.netty.channel.ChannelHandlerContext;
 import org.apache.shenyu.common.utils.Singleton;
 import org.apache.shenyu.protocol.mqtt.repositories.ChannelRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.WillRepository;
 
 /**
  * The DISCONNECT message is sent from the client to the server to indicate
@@ -36,8 +37,7 @@ public class Disconnect extends MessageType {
 
     @Override
     public void disconnect(final ChannelHandlerContext ctx) {
-        //// todo Last words
-        //// todo Clean session
+        Singleton.INST.get(WillRepository.class).remove(ctx.channel());
         cleanChannel(ctx.channel());
         ctx.close();
     }

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

@@ -65,8 +65,10 @@ public void connect() {
             case PINGREQ:
                 messageType.pingReq(ctx);
                 break;
-            case PUBACK:
             case DISCONNECT:
+                messageType.disconnect(ctx);
+                break;
+            case PUBACK:
             default:
                 break;
         }

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

@@ -17,13 +17,17 @@
 
 package org.apache.shenyu.protocol.mqtt;
 
+import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.handler.codec.mqtt.MqttMessage;
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.GenericFutureListener;
 import org.apache.shenyu.common.utils.Singleton;
 import org.apache.shenyu.protocol.mqtt.repositories.ChannelRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.WillRepository;
+
+import java.util.Objects;
 
 /**
  * mqtt transport handler.
@@ -42,8 +46,19 @@ public void channelRead(final ChannelHandlerContext ctx, final Object msg) throw
 
     @Override
     public void channelInactive(final ChannelHandlerContext ctx) throws Exception {
-        Singleton.INST.get(ChannelRepository.class).remove(ctx.channel());
-        ctx.fireChannelInactive();
+        final Channel channel = ctx.channel();
+        Singleton.INST.get(ChannelRepository.class).remove(channel);
+
+        final WillRepository willRepository = Singleton.INST.get(WillRepository.class);
+        final WillRepository.WillEntry will = willRepository.get(channel);
+        if (Objects.nonNull(will)) {
+            // a will is published at most once, and the repository keeps a strong reference
+            // to the channel, so it must be removed even if publishing fails.
+            willRepository.remove(channel);
+            Publish.publishWill(will);
+        }
+        // local state is consistent now, notify the rest of the pipeline exactly once.
+        super.channelInactive(ctx);
     }
 
     @Override

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

@@ -32,7 +32,10 @@
 import org.apache.shenyu.protocol.mqtt.repositories.SubscribeRepository;
 import org.apache.shenyu.protocol.mqtt.repositories.TopicRepository;
 
+import org.apache.shenyu.protocol.mqtt.repositories.WillRepository;
+
 import java.util.List;
+import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 
 import static io.netty.channel.ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE;
@@ -132,4 +135,29 @@ private void send(final String topic, final ByteBuf payload, final int packetId)
             }
         });
     }
+
+    /**
+     * Publish a Last Will message to all subscribers of the will topic.
+     *
+     * @param will the will entry containing topic, message, qos, and retain flag
+     */
+    static void publishWill(final WillRepository.WillEntry will) {
+        if (Objects.isNull(will) || Objects.isNull(will.getTopic()) || Objects.isNull(will.getMessage())) {
+            return;
+        }
+        final List<Channel> channels = Singleton.INST.get(SubscribeRepository.class).getChannelsByTopic(will.getTopic());
+        final MqttQoS willQos = MqttQoS.valueOf(will.getQos());
+        final int packetId = willQos == MqttQoS.AT_MOST_ONCE
+                ? 0
+                : java.util.concurrent.ThreadLocalRandom.current().nextInt(1, 65536);
+        channels.parallelStream().forEach(channel -> {
+            if (channel.isActive()) {
+                MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, willQos, will.isRetain(), 0);
+                MqttPublishVariableHeader mqttPublishVariableHeader = new MqttPublishVariableHeader(will.getTopic(), packetId);
+                MqttPublishMessage mqttPublishMessage = new MqttPublishMessage(mqttFixedHeader, mqttPublishVariableHeader,
+                        Unpooled.wrappedBuffer(will.getMessage()));
+                channel.writeAndFlush(mqttPublishMessage);
+            }
+        });
+    }
 }

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,33 +20,34 @@
 import io.netty.channel.Channel;
 import io.netty.handler.codec.mqtt.MqttTopicSubscription;
 import org.apache.commons.collections4.CollectionUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.shenyu.protocol.mqtt.TopicMatcher;
 
+import java.util.ArrayList;
+import java.util.Collections;
+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;
 import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.CopyOnWriteArraySet;
 
 /**
  * Topic and channel association.
+ *
+ * <p>Subscription updates are applied synchronously on the calling (event loop) thread and every
+ * topic holds a copy-on-write list of channels, so a subscription is visible to publish as soon as
+ * {@code add} returns and concurrent subscribers of the same topic never overwrite each other.
  */
 public class SubscribeRepository implements BaseRepository<List<String>, List<Channel>> {
 
-    private static final Logger LOG = LoggerFactory.getLogger(SubscribeRepository.class);
-
     private static final Map<String, List<Channel>> TOPIC_CHANNEL_FACTORY = new ConcurrentHashMap<>();
 
     @Override
     public void add(final List<String> topics, final List<Channel> channels) {
-        CompletableFuture.runAsync(() -> topics.parallelStream().forEach(s -> {
-            List<Channel> list = get(s);
-            list.addAll(channels);
-            TOPIC_CHANNEL_FACTORY.put(s, list);
-        }));
+        topics.forEach(topic -> TOPIC_CHANNEL_FACTORY
+                .computeIfAbsent(topic, key -> new CopyOnWriteArrayList<>())
+                .addAll(channels));
     }
 
     /**
@@ -55,16 +56,14 @@ public void add(final List<String> topics, final List<Channel> channels) {
      * @param mqttTopicSubscription mqtt subscription info
      */
     public void add(final Channel channel, final List<MqttTopicSubscription> mqttTopicSubscription) {
-        CompletableFuture.runAsync(() -> mqttTopicSubscription.parallelStream().forEach(s -> {
-            List<Channel> channels = get(s.topicName());
-            channels.add(channel);
-            TOPIC_CHANNEL_FACTORY.put(s.topicName(), channels);
-        }));
+        mqttTopicSubscription.forEach(subscription -> TOPIC_CHANNEL_FACTORY
+                .computeIfAbsent(subscription.topicName(), key -> new CopyOnWriteArrayList<>())
+                .add(channel));
     }
 
     @Override
     public void remove(final List<String> topics) {
-        CompletableFuture.runAsync(() -> topics.parallelStream().forEach(TOPIC_CHANNEL_FACTORY::remove));
+        topics.forEach(TOPIC_CHANNEL_FACTORY::remove);
     }
 
     /**
@@ -73,19 +72,19 @@ public void remove(final List<String> topics) {
      * @param channel channel
      */
     public void remove(final List<String> topics, final Channel channel) {
-        CompletableFuture.runAsync(() -> topics.parallelStream().forEach(topic -> {
+        topics.forEach(topic -> {
             List<Channel> channels = TOPIC_CHANNEL_FACTORY.get(topic);
             if (CollectionUtils.isNotEmpty(channels)) {
                 channels.remove(channel);
             }
-        }));
+        });
     }
 
     @Override
     public List<Channel> get(final List<String> topics) {
-        Set<Channel> channels = new CopyOnWriteArraySet<>();
-        topics.parallelStream().forEach(s -> channels.addAll(TOPIC_CHANNEL_FACTORY.get(s)));
-        return new CopyOnWriteArrayList<>(channels);
+        Set<Channel> channels = new LinkedHashSet<>();
+        topics.forEach(topic -> channels.addAll(TOPIC_CHANNEL_FACTORY.getOrDefault(topic, Collections.emptyList())));
+        return new ArrayList<>(channels);
     }
 
     /**
@@ -97,4 +96,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);
+    }
+
 }