feat: implement MQTT Last Will and Testament (LWT)

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

@@ -46,6 +46,39 @@
             <artifactId>reflections</artifactId>
             <version>0.9.11</version>
         </dependency>
+
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <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>
 
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <configuration>
+                    <argLine>--add-opens java.base/java.lang=ALL-UNNAMED
+                        --add-opens java.base/java.util=ALL-UNNAMED
+                        --add-opens java.base/java.net=ALL-UNNAMED
+                        -Dnet.bytebuddy.experimental=true</argLine>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
 </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;
 
@@ -67,6 +68,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

@@ -22,6 +22,10 @@
 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.WillRepository;
+
+import java.util.Objects;
 
 /**
  * mqtt transport handler.
@@ -38,6 +42,16 @@ public void channelRead(final ChannelHandlerContext ctx, final Object msg) throw
         }
     }
 
+    @Override
+    public void channelInactive(final ChannelHandlerContext ctx) throws Exception {
+        WillRepository.WillEntry will = Singleton.INST.get(WillRepository.class).get(ctx.channel());
+        if (Objects.nonNull(will)) {
+            Publish.publishWill(will);
+            Singleton.INST.get(WillRepository.class).remove(ctx.channel());
+        }
+        super.channelInactive(ctx);
+    }
+
     @Override
     public void operationComplete(final Future<? super Void> future) throws Exception {
 

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.handler.codec.mqtt.MqttMessageType.PUBACK;
@@ -124,4 +127,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

@@ -19,11 +19,15 @@
 
 import io.netty.channel.Channel;
 import io.netty.handler.codec.mqtt.MqttTopicSubscription;
+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;
@@ -91,4 +95,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);
+    }
+
 }