This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new 2777620105 feat: implement MQTT Last Will and Testament (LWT) (#6902)
2777620105 is described below
commit 2777620105c434d42e23f3f3ce709a01568e0f36
Author: wy471x <[email protected]>
AuthorDate: Tue Oct 20 12:39:57 2026 +0800
feat: implement MQTT Last Will and Testament (LWT) (#6902)
* feat: implement MQTT Last Will and Testament (LWT)
Parse and store will fields (topic, message, QoS, retain) in WillRepository
on CONNECT. Publish will message to subscribers on ungraceful disconnect
via channelInactive hook. Clear will on graceful DISCONNECT. Fix DISCONNECT
message dispatch in MqttFactory that was previously dropped.
Co-Authored-By: Claude Opus 4.7 <[email protected]>
* Potential fix for pull request finding
Co-authored-by: Copilot Autofix powered by AI
<[email protected]>
* feat: route will delivery through wildcard topic matching
Publish.publishWill used an exact SubscribeRepository.get() lookup, so
clients subscribed to wildcard filters (e.g. status/#) never received
wills published to concrete topics like status/client-001. Port the
TopicMatcher and SubscribeRepository.getChannelsByTopic from #6906 and
route will delivery through it, consistent with normal publish routing.
Co-Authored-By: Claude Opus 4.7 <[email protected]>
* fix(mqtt): make SubscribeRepository updates synchronous and race-free
- allocate the per-topic channel list with computeIfAbsent and apply
add/remove on the calling thread, so concurrent subscribers of the same new
topic no longer overwrite each other's list and a subscription is visible as
soon as add returns; get(List) no longer throws when a topic has no
subscribers; drop the unused logger
- SubscribeRepositoryTest: replace the Mockito mock and the
awaitility/common-pool polling with deterministic assertions, and add a
concurrency regression test for eight subscribers of the same new topic
- ConnectTest: drive Connect with real MQTT messages instead of mocks, keep
the credentials fixed for the whole class, cover the identifier-rejected and
bad-credentials branches and assert a rejected CONNECT stores no will
- MqttFactoryTest: DISCONNECT is dispatched to Disconnect, so assert the
channel is closed and the channel/will repositories are cleared, and register
the WillRepository that Disconnect looks up via Singleton
- MqttTransportHandler: drop the will before publishing it and notify the
pipeline exactly once on channelInactive
* fix(mqtt): resolve merge conflicts in SubscribeRepository and its tests
- adapt getChannelsByTopic to the QoS-aware Map<Channel, MqttQoS> storage
- merge both sides of MqttTransportHandlerTest into a single lifecycle
- merge SubscribeRepositoryTest onto the async QoS API and drop duplicates
- remove duplicated imports and unused imports left by the merge
* fix(mqtt): cap will qos at the subscriber granted qos
getChannelsByTopic now returns matching channels mapped to the maximum
granted qos across exact, wildcard and overlapping filters, and
publishWill delivers each will with min(will qos, granted qos) like the
normal publish path, so a qos 2 will is no longer sent at qos 2 to a
qos 0 subscriber.
* fix(mqtt): apply retained-store semantics to the last will (#6852)
* test(mqtt): register WillRepository in DisconnectDispatchTest (#6852)
The graceful-DISCONNECT path now clears the stored will, so the test must
publish a WillRepository into the singleton just like the other MQTT
repositories it depends on.
---------
Co-authored-by: Claude Opus 4.7 <[email protected]>
Co-authored-by: xiaoyu <[email protected]>
Co-authored-by: Copilot Autofix powered by AI
<[email protected]>
Co-authored-by: aias00 <[email protected]>
---
shenyu-protocol/shenyu-protocol-mqtt/pom.xml | 12 ++
.../org/apache/shenyu/protocol/mqtt/Connect.java | 11 ++
.../apache/shenyu/protocol/mqtt/Disconnect.java | 4 +-
.../shenyu/protocol/mqtt/MqttTransportHandler.java | 16 ++
.../org/apache/shenyu/protocol/mqtt/Publish.java | 40 ++++
.../apache/shenyu/protocol/mqtt/TopicMatcher.java | 103 ++++++++++
.../mqtt/repositories/SubscribeRepository.java | 32 +++
.../protocol/mqtt/repositories/WillRepository.java | 85 ++++++++
.../apache/shenyu/protocol/mqtt/ConnectTest.java | 216 +++++++++++++++++----
.../protocol/mqtt/DisconnectDispatchTest.java | 2 +
.../shenyu/protocol/mqtt/DisconnectTest.java | 96 +++++++++
.../shenyu/protocol/mqtt/MqttFactoryTest.java | 21 +-
.../protocol/mqtt/MqttTransportHandlerTest.java | 145 +++++++++++---
.../shenyu/protocol/mqtt/PublishWillTest.java | 194 ++++++++++++++++++
.../shenyu/protocol/mqtt/TopicMatcherTest.java | 100 ++++++++++
.../mqtt/repositories/SubscribeRepositoryTest.java | 100 +++++++++-
.../mqtt/repositories/WillRepositoryTest.java | 112 +++++++++++
17 files changed, 1224 insertions(+), 65 deletions(-)
diff --git a/shenyu-protocol/shenyu-protocol-mqtt/pom.xml
b/shenyu-protocol/shenyu-protocol-mqtt/pom.xml
index cf8e31fb56..6ce05fbb8b 100644
--- a/shenyu-protocol/shenyu-protocol-mqtt/pom.xml
+++ b/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>
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Connect.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Connect.java
index 157aa33412..f073aa5edd 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Connect.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Connect.java
@@ -29,6 +29,7 @@ import
org.apache.shenyu.protocol.mqtt.repositories.ChannelRepository;
import org.apache.shenyu.protocol.mqtt.repositories.MqttSession;
import org.apache.shenyu.protocol.mqtt.repositories.SessionRepository;
import org.apache.shenyu.protocol.mqtt.repositories.SubscribeRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.WillRepository;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -80,6 +81,16 @@ public class Connect extends MessageType {
// 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);
+ }
+
boolean cleanSession = msg.variableHeader().isCleanSession();
SessionRepository sessionRepository =
Singleton.INST.get(SessionRepository.class);
MqttSession session = sessionRepository.get(clientId);
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Disconnect.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Disconnect.java
index b71a849656..4e3e919016 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Disconnect.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Disconnect.java
@@ -24,6 +24,7 @@ import
org.apache.shenyu.protocol.mqtt.repositories.ChannelRepository;
import org.apache.shenyu.protocol.mqtt.repositories.MqttSession;
import org.apache.shenyu.protocol.mqtt.repositories.SessionRepository;
import org.apache.shenyu.protocol.mqtt.repositories.SubscribeRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.WillRepository;
import org.apache.shenyu.protocol.mqtt.utils.MqttPacketIdGenerator;
import java.util.Objects;
@@ -42,7 +43,8 @@ public class Disconnect extends MessageType {
@Override
public void disconnect(final ChannelHandlerContext ctx) {
- //// todo Last words
+ // MQTT-3.1.2-8: a graceful DISCONNECT discards the stored will.
+ Singleton.INST.get(WillRepository.class).remove(ctx.channel());
cleanChannel(ctx.channel());
ctx.close();
}
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/MqttTransportHandler.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/MqttTransportHandler.java
index b177c0ed35..0b97aa6bc0 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/MqttTransportHandler.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/MqttTransportHandler.java
@@ -17,6 +17,7 @@
package org.apache.shenyu.protocol.mqtt;
+import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
@@ -24,6 +25,10 @@ import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.util.ReferenceCountUtil;
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.
@@ -46,6 +51,17 @@ public class MqttTransportHandler extends
ChannelInboundHandlerAdapter implement
@Override
public void channelInactive(final ChannelHandlerContext ctx) throws
Exception {
+ final Channel channel = ctx.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; remove it before publishing
so a repeated
+ // channelInactive cannot deliver it twice.
+ willRepository.remove(channel);
+ Publish.publishWill(will);
+ }
+ // release channel state before notifying the rest of the pipeline
exactly once.
Disconnect.cleanChannel(ctx.channel());
ctx.fireChannelInactive();
}
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Publish.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Publish.java
index 76f28c08a7..e5ad6fc169 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Publish.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Publish.java
@@ -33,9 +33,11 @@ import io.netty.util.ReferenceCountUtil;
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.apache.shenyu.protocol.mqtt.repositories.WillRepository;
import org.apache.shenyu.protocol.mqtt.utils.MqttPacketIdGenerator;
import java.util.Map;
+import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import static io.netty.channel.ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE;
@@ -134,4 +136,42 @@ public class Publish extends MessageType {
private static MqttQoS minQoS(final MqttQoS publishQoS, final MqttQoS
grantedQoS) {
return publishQoS.value() <= grantedQoS.value() ? publishQoS :
grantedQoS;
}
+
+ /**
+ * 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;
+ }
+ // MQTT-3.1.2-14/16: a retained will is stored like any other retained
publish,
+ // and an empty retained will payload clears the stored value.
+ if (will.isRetain()) {
+ if (will.getMessage().length > 0) {
+ Singleton.INST.get(TopicRepository.class).add(will.getTopic(),
will.getMessage());
+ } else {
+
Singleton.INST.get(TopicRepository.class).remove(will.getTopic());
+ }
+ }
+ // MQTT-3.1.2-14/16: a retained will is stored like any other retained
publish,
+ // and an empty retained will payload clears the stored value.
+ final Map<Channel, MqttQoS> subscribers =
Singleton.INST.get(SubscribeRepository.class).getChannelsByTopic(will.getTopic());
+ final MqttQoS willQos = MqttQoS.valueOf(will.getQos());
+ subscribers.entrySet().parallelStream().forEach(entry -> {
+ Channel channel = entry.getKey();
+ if (channel.isActive()) {
+ MqttQoS qos = minQoS(willQos, entry.getValue());
+ int packetId = MqttQoS.AT_MOST_ONCE == qos
+ ? 0
+ :
java.util.concurrent.ThreadLocalRandom.current().nextInt(1, 65536);
+ MqttFixedHeader mqttFixedHeader = new
MqttFixedHeader(MqttMessageType.PUBLISH, false, qos, will.isRetain(), 0);
+ MqttPublishVariableHeader mqttPublishVariableHeader = new
MqttPublishVariableHeader(will.getTopic(), packetId);
+ MqttPublishMessage mqttPublishMessage = new
MqttPublishMessage(mqttFixedHeader, mqttPublishVariableHeader,
+ Unpooled.wrappedBuffer(will.getMessage()));
+ channel.writeAndFlush(mqttPublishMessage);
+ }
+ });
+ }
}
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/TopicMatcher.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/TopicMatcher.java
new file mode 100644
index 0000000000..850b3addd3
--- /dev/null
+++
b/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;
+ }
+}
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/repositories/SubscribeRepository.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/repositories/SubscribeRepository.java
index 2bd69d50c1..7b96da491c 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/repositories/SubscribeRepository.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/repositories/SubscribeRepository.java
@@ -20,6 +20,7 @@ package org.apache.shenyu.protocol.mqtt.repositories;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.handler.codec.mqtt.MqttTopicSubscription;
+import org.apache.shenyu.protocol.mqtt.TopicMatcher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -108,4 +109,35 @@ public class SubscribeRepository implements
BaseRepository<List<String>, Map<Cha
return qos1.value() >= qos2.value() ? qos1 : qos2;
}
+ /**
+ * Get the channels whose subscription filter matches the published topic,
+ * mapped to the maximum qos granted across all their matching filters.
+ * Supports MQTT wildcards: + (single-level) and # (multi-level).
+ *
+ * @param topic the published topic name
+ * @return matching channels with their maximum granted qos
+ */
+ public Map<Channel, MqttQoS> getChannelsByTopic(final String topic) {
+ // MQTT requires at most one delivery per publish per client, so merge
the
+ // granted qos when overlapping filters (e.g. sport/# and #) both
match.
+ Map<Channel, MqttQoS> result = new ConcurrentHashMap<>();
+
+ // fast path: exact subscription, no wildcard scan needed
+ Map<Channel, MqttQoS> exactMatch = TOPIC_CHANNEL_FACTORY.get(topic);
+ if (Objects.nonNull(exactMatch)) {
+ result.putAll(exactMatch);
+ }
+
+ for (Map.Entry<String, Map<Channel, MqttQoS>> 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)) {
+ entry.getValue().forEach((channel, qos) ->
result.merge(channel, qos, SubscribeRepository::maxQoS));
+ }
+ }
+ return result;
+ }
+
}
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/repositories/WillRepository.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/repositories/WillRepository.java
new file mode 100644
index 0000000000..9a57439ae4
--- /dev/null
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/repositories/WillRepository.java
@@ -0,0 +1,85 @@
+/*
+ * 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.repositories;
+
+import io.netty.channel.Channel;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Stores Last Will and Testament for connected clients.
+ * Will is set on CONNECT and cleared on graceful DISCONNECT.
+ * On ungraceful disconnect (channelInactive with will present), the will is
published.
+ */
+public class WillRepository implements BaseRepository<Channel,
WillRepository.WillEntry> {
+
+ private static final Map<Channel, WillEntry> WILL_FACTORY = new
ConcurrentHashMap<>();
+
+ @Override
+ public void add(final Channel channel, final WillEntry willEntry) {
+ WILL_FACTORY.put(channel, willEntry);
+ }
+
+ @Override
+ public void remove(final Channel channel) {
+ WILL_FACTORY.remove(channel);
+ }
+
+ @Override
+ public WillEntry get(final Channel channel) {
+ return WILL_FACTORY.get(channel);
+ }
+
+ /**
+ * Holds the will message fields from a CONNECT payload.
+ */
+ public static class WillEntry {
+
+ private final String topic;
+
+ private final byte[] message;
+
+ private final int qos;
+
+ private final boolean retain;
+
+ public WillEntry(final String topic, final byte[] message, final int
qos, final boolean retain) {
+ this.topic = topic;
+ this.message = message;
+ this.qos = qos;
+ this.retain = retain;
+ }
+
+ public String getTopic() {
+ return topic;
+ }
+
+ public byte[] getMessage() {
+ return message;
+ }
+
+ public int getQos() {
+ return qos;
+ }
+
+ public boolean isRetain() {
+ return retain;
+ }
+ }
+}
\ No newline at end of file
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/ConnectTest.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/ConnectTest.java
index 2b8acc8e33..011b759892 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/ConnectTest.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/ConnectTest.java
@@ -30,16 +30,23 @@ import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.handler.codec.mqtt.MqttVersion;
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.apache.shenyu.protocol.mqtt.repositories.WillRepository.WillEntry;
import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.nio.charset.StandardCharsets;
-import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
import static
io.netty.handler.codec.mqtt.MqttConnectReturnCode.CONNECTION_ACCEPTED;
+import static
io.netty.handler.codec.mqtt.MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD;
+import static
io.netty.handler.codec.mqtt.MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED;
import static
io.netty.handler.codec.mqtt.MqttConnectReturnCode.CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION;
-import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -51,26 +58,55 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
*/
public final class ConnectTest {
- private static final String CLIENT_ID = "test-client";
-
private static final String USER_NAME = "test-user";
private static final String PASSWORD = "test-password";
- private static ChannelRepository channelRepository;
+ private static final String CLIENT_ID = "test-client";
+
+ private static final String WILL_TOPIC = "status/client-001";
+
+ private final List<EmbeddedChannel> channels = new ArrayList<>();
+
+ private ChannelRepository channelRepository;
+
+ private WillRepository willRepository;
+
+ private Connect connect;
@BeforeAll
- static void setUp() {
+ static void setUpCredentials() {
+ MqttContext mqttContext = new MqttContext();
+ mqttContext.setUserName(USER_NAME);
+ mqttContext.setPassword(PASSWORD);
+ }
+
+ @AfterAll
+ static void clearCredentials() {
+ MqttContext mqttContext = new MqttContext();
+ mqttContext.setUserName(null);
+ mqttContext.setPassword(null);
+ }
+
+ @BeforeEach
+ public void setUp() {
+ connect = new Connect();
channelRepository = new ChannelRepository();
+ willRepository = new WillRepository();
Singleton.INST.single(ChannelRepository.class, channelRepository);
- new MqttContext().setUserName(USER_NAME);
- new MqttContext().setPassword(PASSWORD);
+ Singleton.INST.single(WillRepository.class, willRepository);
}
- @AfterAll
- static void tearDown() {
- new MqttContext().setUserName(null);
- new MqttContext().setPassword(null);
+ @AfterEach
+ public void tearDown() {
+ for (EmbeddedChannel channel : channels) {
+ channelRepository.remove(channel);
+ willRepository.remove(channel);
+ channel.finishAndReleaseAll();
+ }
+ channels.clear();
+ Singleton.INST.single(ChannelRepository.class, new
ChannelRepository());
+ Singleton.INST.single(WillRepository.class, new WillRepository());
}
@Test
@@ -90,55 +126,169 @@ public final class ConnectTest {
@Test
public void unsupportedProtocolVersionIsRejected() {
- EmbeddedChannel channel = new EmbeddedChannel(new
ChannelInboundHandlerAdapter());
- ChannelHandlerContext ctx = channel.pipeline().lastContext();
+ EmbeddedChannel channel = newChannel();
- new Connect().connect(ctx, connectMessage("MQTT", 6));
+ connect.connect(context(channel), connectMessage("MQTT", 6));
MqttConnAckMessage ackMessage = channel.readOutbound();
assertNotNull(ackMessage);
- assertEquals(CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION,
- ackMessage.variableHeader().connectReturnCode());
- channel.runPendingTasks();
- assertFalse(channel.isActive());
+ assertEquals(CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION,
ackMessage.variableHeader().connectReturnCode());
+ assertFalse(ackMessage.variableHeader().isSessionPresent());
+ assertChannelClosed(channel);
+ assertNull(channelRepository.get(channel));
+ }
+
+ @Test
+ public void emptyClientIdIsRejected() {
+ EmbeddedChannel channel = newChannel();
+
+ connect.connect(context(channel),
buildConnectMessage(MqttVersion.MQTT_3_1_1.protocolName(),
+ MqttVersion.MQTT_3_1_1.protocolLevel(), "", PASSWORD, false,
0, false, null, null));
+
+ MqttConnAckMessage ackMessage = channel.readOutbound();
+ assertNotNull(ackMessage);
+ assertEquals(CONNECTION_REFUSED_IDENTIFIER_REJECTED,
ackMessage.variableHeader().connectReturnCode());
+ assertChannelClosed(channel);
+ assertNull(channelRepository.get(channel));
+ }
+
+ @Test
+ public void invalidCredentialsAreRejected() {
+ EmbeddedChannel channel = newChannel();
+
+ connect.connect(context(channel),
buildConnectMessage(MqttVersion.MQTT_3_1_1.protocolName(),
+ MqttVersion.MQTT_3_1_1.protocolLevel(), CLIENT_ID,
"invalid-password", false, 0, false, null, null));
+
+ MqttConnAckMessage ackMessage = channel.readOutbound();
+ assertNotNull(ackMessage);
+ assertEquals(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD,
ackMessage.variableHeader().connectReturnCode());
+ assertChannelClosed(channel);
assertNull(channelRepository.get(channel));
}
@Test
public void duplicateConnectIsRejected() {
- EmbeddedChannel channel = new EmbeddedChannel(new
ChannelInboundHandlerAdapter());
- ChannelHandlerContext ctx = channel.pipeline().lastContext();
+ EmbeddedChannel channel = newChannel();
+ ChannelHandlerContext ctx = context(channel);
- new Connect().connect(ctx,
connectMessage(MqttVersion.MQTT_3_1_1.protocolName(),
MqttVersion.MQTT_3_1_1.protocolLevel()));
+ connect.connect(ctx, connectMessage(MqttVersion.MQTT_3_1_1));
assertNotNull(channel.readOutbound());
- new Connect().connect(ctx,
connectMessage(MqttVersion.MQTT_3_1_1.protocolName(),
MqttVersion.MQTT_3_1_1.protocolLevel()));
+ connect.connect(ctx, connectMessage(MqttVersion.MQTT_3_1_1));
- channel.runPendingTasks();
- assertFalse(channel.isActive());
+ assertChannelClosed(channel);
assertNull(channel.readOutbound());
}
+ @Test
+ public void testStoresWillOnConnect() {
+ EmbeddedChannel channel = newChannel();
+ byte[] willMessage = "client disconnected
unexpectedly".getBytes(StandardCharsets.UTF_8);
+
+ connect.connect(context(channel), willConnectMessage(1, true,
WILL_TOPIC, willMessage));
+
+ WillEntry will = willRepository.get(channel);
+ assertNotNull(will);
+ assertEquals(WILL_TOPIC, will.getTopic());
+ assertArrayEquals(willMessage, will.getMessage());
+ assertEquals(1, will.getQos());
+ assertTrue(will.isRetain());
+ }
+
+ @Test
+ public void testDoesNotStoreWillWhenWillFlagIsFalse() {
+ EmbeddedChannel channel = newChannel();
+
+ connect.connect(context(channel),
connectMessage(MqttVersion.MQTT_3_1_1));
+
+ assertNull(willRepository.get(channel));
+ }
+
+ @Test
+ public void testWillQosZero() {
+ EmbeddedChannel channel = newChannel();
+ byte[] willMessage = "qos0 will".getBytes(StandardCharsets.UTF_8);
+
+ connect.connect(context(channel), willConnectMessage(0, false,
"topic/qos0", willMessage));
+
+ WillEntry will = willRepository.get(channel);
+ assertNotNull(will);
+ assertEquals(0, will.getQos());
+ assertFalse(will.isRetain());
+ }
+
+ @Test
+ public void testWillRetainTrue() {
+ EmbeddedChannel channel = newChannel();
+ byte[] willMessage = "retained will".getBytes(StandardCharsets.UTF_8);
+
+ connect.connect(context(channel), willConnectMessage(2, true,
"topic/retained", willMessage));
+
+ WillEntry will = willRepository.get(channel);
+ assertNotNull(will);
+ assertEquals(2, will.getQos());
+ assertTrue(will.isRetain());
+ }
+
+ @Test
+ public void willIsNotStoredWhenConnectIsRejected() {
+ EmbeddedChannel channel = newChannel();
+
+ connect.connect(context(channel), buildConnectMessage("MQTT", 6,
CLIENT_ID, PASSWORD,
+ true, 1, true, WILL_TOPIC, "retained
will".getBytes(StandardCharsets.UTF_8)));
+
+ assertNull(willRepository.get(channel));
+ }
+
private void connectIsAccepted(final MqttVersion version) {
- EmbeddedChannel channel = new EmbeddedChannel(new
ChannelInboundHandlerAdapter());
- ChannelHandlerContext ctx = channel.pipeline().lastContext();
+ EmbeddedChannel channel = newChannel();
- new Connect().connect(ctx, connectMessage(version.protocolName(),
version.protocolLevel()));
+ connect.connect(context(channel), connectMessage(version));
MqttConnAckMessage ackMessage = channel.readOutbound();
assertNotNull(ackMessage);
assertEquals(CONNECTION_ACCEPTED,
ackMessage.variableHeader().connectReturnCode());
assertTrue(ackMessage.variableHeader().isSessionPresent());
- await().atMost(Duration.ofSeconds(5))
- .until(() -> CLIENT_ID.equals(channelRepository.get(channel)));
+ assertEquals(CLIENT_ID, channelRepository.get(channel));
+ }
+
+ private void assertChannelClosed(final EmbeddedChannel channel) {
+ channel.runPendingTasks();
+ assertFalse(channel.isActive());
+ }
+
+ private EmbeddedChannel newChannel() {
+ EmbeddedChannel channel = new EmbeddedChannel(new
ChannelInboundHandlerAdapter());
+ channels.add(channel);
+ return channel;
+ }
+
+ private ChannelHandlerContext context(final EmbeddedChannel channel) {
+ return channel.pipeline().lastContext();
+ }
+
+ private MqttConnectMessage connectMessage(final MqttVersion version) {
+ return connectMessage(version.protocolName(), version.protocolLevel());
}
private MqttConnectMessage connectMessage(final String protocolName, final
int protocolLevel) {
+ return buildConnectMessage(protocolName, protocolLevel, CLIENT_ID,
PASSWORD, false, 0, false, null, null);
+ }
+
+ private MqttConnectMessage willConnectMessage(final int willQos, final
boolean willRetain,
+ final String willTopic, final byte[] willMessage) {
+ return buildConnectMessage(MqttVersion.MQTT_3_1_1.protocolName(),
MqttVersion.MQTT_3_1_1.protocolLevel(),
+ CLIENT_ID, PASSWORD, true, willQos, willRetain, willTopic,
willMessage);
+ }
+
+ private MqttConnectMessage buildConnectMessage(final String protocolName,
final int protocolLevel,
+ final String clientId, final String password, final boolean
willFlag, final int willQos,
+ final boolean willRetain, final String willTopic, final byte[]
willMessage) {
MqttFixedHeader fixedHeader = new
MqttFixedHeader(MqttMessageType.CONNECT, false, MqttQoS.AT_MOST_ONCE, false, 0);
MqttConnectVariableHeader variableHeader = new
MqttConnectVariableHeader(protocolName, protocolLevel,
- true, true, false, 0, false, false, 60);
- MqttConnectPayload payload = new MqttConnectPayload(CLIENT_ID, null,
null,
- USER_NAME, PASSWORD.getBytes(StandardCharsets.UTF_8));
+ true, true, willRetain, willQos, willFlag, false, 60);
+ MqttConnectPayload payload = new MqttConnectPayload(clientId,
willTopic, willMessage,
+ USER_NAME, password.getBytes(StandardCharsets.UTF_8));
return new MqttConnectMessage(fixedHeader, variableHeader, payload);
}
}
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/DisconnectDispatchTest.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/DisconnectDispatchTest.java
index 6509ddd686..d136d51d15 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/DisconnectDispatchTest.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/DisconnectDispatchTest.java
@@ -26,6 +26,7 @@ import
org.apache.shenyu.protocol.mqtt.repositories.ChannelRepository;
import org.apache.shenyu.protocol.mqtt.repositories.MqttSession;
import org.apache.shenyu.protocol.mqtt.repositories.SessionRepository;
import org.apache.shenyu.protocol.mqtt.repositories.SubscribeRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.WillRepository;
import org.awaitility.Awaitility;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -53,6 +54,7 @@ public class DisconnectDispatchTest {
Singleton.INST.single(ChannelRepository.class, new
ChannelRepository());
Singleton.INST.single(SessionRepository.class, new
SessionRepository());
Singleton.INST.single(SubscribeRepository.class, new
SubscribeRepository());
+ Singleton.INST.single(WillRepository.class, new WillRepository());
ctx = mock(ChannelHandlerContext.class);
channel = mock(Channel.class);
when(ctx.channel()).thenReturn(channel);
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/DisconnectTest.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/DisconnectTest.java
new file mode 100644
index 0000000000..d6ad2b5efd
--- /dev/null
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/DisconnectTest.java
@@ -0,0 +1,96 @@
+/*
+ * 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.Channel;
+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.SessionRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.SubscribeRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.WillRepository;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.nullValue;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+public class DisconnectTest {
+
+ @Mock
+ private ChannelHandlerContext ctx;
+
+ @Mock
+ private Channel channel;
+
+ private Disconnect disconnect;
+
+ private WillRepository willRepository;
+
+ private ChannelRepository channelRepository;
+
+ @BeforeEach
+ public void setUp() {
+ disconnect = new Disconnect();
+ willRepository = new WillRepository();
+ channelRepository = new ChannelRepository();
+ Singleton.INST.single(WillRepository.class, willRepository);
+ Singleton.INST.single(ChannelRepository.class, channelRepository);
+ // cleanChannel releases the stored session and subscriptions as well.
+ Singleton.INST.single(SessionRepository.class, new
SessionRepository());
+ Singleton.INST.single(SubscribeRepository.class, new
SubscribeRepository());
+ when(ctx.channel()).thenReturn(channel);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ Singleton.INST.single(WillRepository.class, new WillRepository());
+ Singleton.INST.single(ChannelRepository.class, new
ChannelRepository());
+ Singleton.INST.single(SessionRepository.class, new
SessionRepository());
+ Singleton.INST.single(SubscribeRepository.class, new
SubscribeRepository());
+ }
+
+ @Test
+ public void testDisconnectClearsWill() {
+ WillRepository.WillEntry will = new
WillRepository.WillEntry("topic/will", "goodbye".getBytes(), 0, false);
+ willRepository.add(channel, will);
+
+ disconnect.disconnect(ctx);
+
+ assertThat(willRepository.get(channel), nullValue());
+ }
+
+ @Test
+ public void testDisconnectWithoutWill() {
+ disconnect.disconnect(ctx);
+ assertThat(willRepository.get(channel), nullValue());
+ }
+
+ @Test
+ public void testDisconnectRemovesChannel() {
+ channelRepository.add(channel, "client-1");
+ disconnect.disconnect(ctx);
+ assertThat(channelRepository.get(channel), nullValue());
+ }
+}
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/MqttFactoryTest.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/MqttFactoryTest.java
index cecdd764d2..f28ff11057 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/MqttFactoryTest.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/MqttFactoryTest.java
@@ -42,6 +42,8 @@ import org.apache.shenyu.common.utils.Singleton;
import org.apache.shenyu.protocol.mqtt.repositories.ChannelRepository;
import org.apache.shenyu.protocol.mqtt.repositories.SessionRepository;
import org.apache.shenyu.protocol.mqtt.repositories.SubscribeRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.WillRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.WillRepository.WillEntry;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -51,6 +53,7 @@ import java.util.Collections;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
@@ -64,9 +67,16 @@ public final class MqttFactoryTest {
private static final String PASSWORD = "factory-password";
+ private ChannelRepository channelRepository;
+
+ private WillRepository willRepository;
+
@BeforeEach
public void setUp() {
- Singleton.INST.single(ChannelRepository.class, new
ChannelRepository());
+ channelRepository = new ChannelRepository();
+ willRepository = new WillRepository();
+ Singleton.INST.single(ChannelRepository.class, channelRepository);
+ Singleton.INST.single(WillRepository.class, willRepository);
Singleton.INST.single(SessionRepository.class, new
SessionRepository());
Singleton.INST.single(SubscribeRepository.class, new
SubscribeRepository());
new MqttContext().setUserName(USER_NAME);
@@ -75,6 +85,8 @@ public final class MqttFactoryTest {
@AfterEach
public void tearDown() {
+ Singleton.INST.single(ChannelRepository.class, new
ChannelRepository());
+ Singleton.INST.single(WillRepository.class, new WillRepository());
new MqttContext().setUserName(null);
new MqttContext().setPassword(null);
}
@@ -153,13 +165,18 @@ public final class MqttFactoryTest {
}
@Test
- public void disconnectShouldCloseChannel() {
+ public void disconnectShouldBeDispatchedAndClearConnectionState() {
EmbeddedChannel channel = new EmbeddedChannel(new
ChannelInboundHandlerAdapter());
ChannelHandlerContext ctx = channel.pipeline().lastContext();
+ channelRepository.add(channel, CLIENT_ID);
+ willRepository.add(channel, new WillEntry("status/will",
"goodbye".getBytes(StandardCharsets.UTF_8), 1, true));
new MqttFactory(new
MqttMessage(fixedHeader(MqttMessageType.DISCONNECT)), ctx).connect();
+ channel.runPendingTasks();
assertFalse(channel.isActive());
+ assertNull(channelRepository.get(channel));
+ assertNull(willRepository.get(channel));
channel.finishAndReleaseAll();
}
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/MqttTransportHandlerTest.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/MqttTransportHandlerTest.java
index d2aeaf37c6..053414f956 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/MqttTransportHandlerTest.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/MqttTransportHandlerTest.java
@@ -19,6 +19,9 @@ package org.apache.shenyu.protocol.mqtt;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.embedded.EmbeddedChannel;
import io.netty.handler.codec.mqtt.MqttConnectMessage;
import io.netty.handler.codec.mqtt.MqttConnectPayload;
@@ -35,29 +38,41 @@ import org.apache.shenyu.common.utils.Singleton;
import org.apache.shenyu.protocol.mqtt.repositories.ChannelRepository;
import org.apache.shenyu.protocol.mqtt.repositories.SessionRepository;
import org.apache.shenyu.protocol.mqtt.repositories.SubscribeRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.WillRepository;
import org.apache.shenyu.protocol.mqtt.utils.MqttPacketIdGenerator;
import org.awaitility.core.ThrowingRunnable;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Collections;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
/**
* Test cases for {@link MqttTransportHandler}.
*/
+@ExtendWith(MockitoExtension.class)
public final class MqttTransportHandlerTest {
private static final String TOPIC = "test/topic";
+ private static final String WILL_TOPIC = "status/offline";
+
private static final String CLIENT_ID = "test-client";
private static final String USER_NAME = "test-user";
@@ -76,15 +91,30 @@ public final class MqttTransportHandlerTest {
private static final SubscribeRepository SUBSCRIBE_REPOSITORY = new
SubscribeRepository();
+ @Mock
+ private ChannelHandlerContext ctx;
+
+ @Mock
+ private Channel channel;
+
+ private MqttTransportHandler handler;
+
+ private WillRepository willRepository;
+
private EmbeddedChannel registeredChannel;
@BeforeEach
public void setUp() {
+ handler = new MqttTransportHandler();
+ willRepository = new WillRepository();
+ Singleton.INST.single(WillRepository.class, willRepository);
Singleton.INST.single(ChannelRepository.class, CHANNEL_REPOSITORY);
Singleton.INST.single(SessionRepository.class, new
SessionRepository());
Singleton.INST.single(SubscribeRepository.class, SUBSCRIBE_REPOSITORY);
new MqttContext().setUserName(USER_NAME);
new MqttContext().setPassword(PASSWORD);
+ // only the will tests drive the handler through a mocked context
+ lenient().when(ctx.channel()).thenReturn(channel);
registeredChannel = new EmbeddedChannel();
CHANNEL_REPOSITORY.add(registeredChannel, CLIENT_ID);
@@ -98,75 +128,80 @@ public final class MqttTransportHandlerTest {
@AfterEach
public void tearDown() {
+ willRepository.remove(channel);
+ SUBSCRIBE_REPOSITORY.remove(channel);
Singleton.INST.get(SessionRepository.class).remove(CLIENT_ID);
MqttPacketIdGenerator.remove(registeredChannel);
CHANNEL_REPOSITORY.remove(registeredChannel);
SUBSCRIBE_REPOSITORY.remove(registeredChannel);
- awaitAssert(() ->
assertFalse(SUBSCRIBE_REPOSITORY.get(TOPIC).containsKey(registeredChannel)));
+ awaitAssert(() -> {
+
assertFalse(SUBSCRIBE_REPOSITORY.get(TOPIC).containsKey(registeredChannel));
+ assertTrue(SUBSCRIBE_REPOSITORY.get(WILL_TOPIC).isEmpty());
+ });
registeredChannel.finishAndReleaseAll();
-
new MqttContext().setUserName(null);
new MqttContext().setPassword(null);
}
@Test
public void channelReadReleasesInboundPublishMessage() {
- EmbeddedChannel channel = new EmbeddedChannel(new
MqttTransportHandler());
+ EmbeddedChannel publisherChannel = new EmbeddedChannel(new
MqttTransportHandler());
MqttFixedHeader fixedHeader = new
MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_MOST_ONCE, false, 0);
MqttPublishVariableHeader variableHeader = new
MqttPublishVariableHeader(TOPIC, 1);
MqttPublishMessage message = new MqttPublishMessage(fixedHeader,
variableHeader,
Unpooled.copiedBuffer("hello", CharsetUtil.UTF_8));
ByteBuf payload = message.payload();
- channel.writeInbound(message);
+ publisherChannel.writeInbound(message);
assertEquals(0, payload.refCnt());
- channel.finishAndReleaseAll();
+ publisherChannel.finishAndReleaseAll();
}
@Test
public void duplicateConnectCleansUpChannelRepository() {
- EmbeddedChannel channel = new EmbeddedChannel(new
MqttTransportHandler());
+ EmbeddedChannel sessionChannel = new EmbeddedChannel(new
MqttTransportHandler());
- channel.writeInbound(connectMessage());
- assertEquals(CLIENT_ID, CHANNEL_REPOSITORY.get(channel));
+ sessionChannel.writeInbound(connectMessage());
+ assertEquals(CLIENT_ID, CHANNEL_REPOSITORY.get(sessionChannel));
- channel.writeInbound(connectMessage());
- channel.runPendingTasks();
+ sessionChannel.writeInbound(connectMessage());
+ sessionChannel.runPendingTasks();
- assertFalse(channel.isActive());
- assertNull(CHANNEL_REPOSITORY.get(channel));
- channel.finishAndReleaseAll();
+ assertFalse(sessionChannel.isActive());
+ assertNull(CHANNEL_REPOSITORY.get(sessionChannel));
+ sessionChannel.finishAndReleaseAll();
}
@Test
public void abruptChannelCloseCleansUpChannelRepository() {
- EmbeddedChannel channel = new EmbeddedChannel(new
MqttTransportHandler());
+ EmbeddedChannel sessionChannel = new EmbeddedChannel(new
MqttTransportHandler());
- channel.writeInbound(connectMessage());
- assertEquals(CLIENT_ID, CHANNEL_REPOSITORY.get(channel));
+ sessionChannel.writeInbound(connectMessage());
+ assertEquals(CLIENT_ID, CHANNEL_REPOSITORY.get(sessionChannel));
- channel.close();
- channel.runPendingTasks();
+ sessionChannel.close();
+ sessionChannel.runPendingTasks();
- assertFalse(channel.isActive());
- assertNull(CHANNEL_REPOSITORY.get(channel));
- channel.finishAndReleaseAll();
+ assertFalse(sessionChannel.isActive());
+ assertNull(CHANNEL_REPOSITORY.get(sessionChannel));
+ sessionChannel.finishAndReleaseAll();
}
@Test
public void nonMqttMessageClosesConnectedChannel() {
- EmbeddedChannel channel = new EmbeddedChannel(new
MqttTransportHandler());
+ EmbeddedChannel sessionChannel = new EmbeddedChannel(new
MqttTransportHandler());
- channel.writeInbound(connectMessage());
- assertEquals(CLIENT_ID, CHANNEL_REPOSITORY.get(channel));
+ sessionChannel.writeInbound(connectMessage());
+ assertEquals(CLIENT_ID, CHANNEL_REPOSITORY.get(sessionChannel));
- channel.writeInbound("not-a-mqtt-message");
+ sessionChannel.writeInbound("not-a-mqtt-message");
+ sessionChannel.runPendingTasks();
- assertFalse(channel.isActive());
- assertNull(CHANNEL_REPOSITORY.get(channel));
- channel.finishAndReleaseAll();
+ assertFalse(sessionChannel.isActive());
+ assertNull(CHANNEL_REPOSITORY.get(sessionChannel));
+ sessionChannel.finishAndReleaseAll();
}
@Test
@@ -180,6 +215,60 @@ public final class MqttTransportHandlerTest {
assertEquals(1, MqttPacketIdGenerator.next(registeredChannel));
}
+ @Test
+ public void channelInactiveIsPropagatedOnlyOnce() {
+ AtomicInteger fired = new AtomicInteger();
+ EmbeddedChannel sessionChannel = new EmbeddedChannel(new
MqttTransportHandler(), new ChannelInboundHandlerAdapter() {
+ @Override
+ public void channelInactive(final ChannelHandlerContext context)
throws Exception {
+ fired.incrementAndGet();
+ super.channelInactive(context);
+ }
+ });
+
+ sessionChannel.close();
+ sessionChannel.runPendingTasks();
+
+ assertEquals(1, fired.get());
+ sessionChannel.finishAndReleaseAll();
+ }
+
+ @Test
+ public void testChannelInactiveFiresWillAndRemovesIt() throws Exception {
+ willRepository.add(channel, new WillRepository.WillEntry(WILL_TOPIC,
"sudden disconnect".getBytes(), 1, true));
+ SUBSCRIBE_REPOSITORY.add(channel,
+ Collections.singletonList(new
MqttTopicSubscription(WILL_TOPIC, MqttQoS.AT_LEAST_ONCE)));
+ awaitAssert(() ->
assertTrue(SUBSCRIBE_REPOSITORY.get(WILL_TOPIC).containsKey(channel)));
+ when(channel.isActive()).thenReturn(true);
+
+ handler.channelInactive(ctx);
+
+ ArgumentCaptor<MqttPublishMessage> published =
ArgumentCaptor.forClass(MqttPublishMessage.class);
+ verify(channel).writeAndFlush(published.capture());
+ assertEquals(WILL_TOPIC,
published.getValue().variableHeader().topicName());
+ assertEquals(MqttQoS.AT_LEAST_ONCE,
published.getValue().fixedHeader().qosLevel());
+ assertTrue(published.getValue().fixedHeader().isRetain());
+ assertNull(willRepository.get(channel));
+ }
+
+ @Test
+ public void testChannelInactiveDoesNothingWhenNoWill() throws Exception {
+ handler.channelInactive(ctx);
+
+ assertNull(willRepository.get(channel));
+ }
+
+ @Test
+ public void testChannelInactiveAfterDisconnectClearsWill() throws
Exception {
+ willRepository.add(channel, new
WillRepository.WillEntry("status/clean", "graceful close".getBytes(), 0,
false));
+
+ // a graceful disconnect removes the will first, so channelInactive
must not publish it
+ willRepository.remove(channel);
+ handler.channelInactive(ctx);
+
+ assertNull(willRepository.get(channel));
+ }
+
private MqttConnectMessage connectMessage() {
MqttFixedHeader fixedHeader = new
MqttFixedHeader(MqttMessageType.CONNECT, false, MqttQoS.AT_MOST_ONCE, false, 0);
MqttConnectVariableHeader variableHeader = new
MqttConnectVariableHeader(
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/PublishWillTest.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/PublishWillTest.java
new file mode 100644
index 0000000000..9c3a5251e6
--- /dev/null
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/PublishWillTest.java
@@ -0,0 +1,194 @@
+/*
+ * 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.Channel;
+import io.netty.handler.codec.mqtt.MqttPublishMessage;
+import io.netty.handler.codec.mqtt.MqttQoS;
+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.apache.shenyu.protocol.mqtt.repositories.WillRepository;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+public class PublishWillTest {
+
+ @Mock
+ private SubscribeRepository subscribeRepository;
+
+ @Mock
+ private Channel subscriberChannel;
+
+ @Mock
+ private Channel otherChannel;
+
+ @BeforeEach
+ public void setUp() {
+ Singleton.INST.single(SubscribeRepository.class, subscribeRepository);
+ Singleton.INST.single(TopicRepository.class, new TopicRepository());
+ }
+
+ @AfterEach
+ public void tearDown() {
+ Singleton.INST.single(SubscribeRepository.class, new
SubscribeRepository());
+ Singleton.INST.get(TopicRepository.class).remove("status/retained");
+ Singleton.INST.get(TopicRepository.class).remove("status/cleared");
+ }
+
+ @Test
+ public void testRetainedWillIsStoredForLaterSubscribers() {
+
when(subscribeRepository.getChannelsByTopic("status/retained")).thenReturn(Collections.emptyMap());
+
+ byte[] message = "gone".getBytes(StandardCharsets.UTF_8);
+ Publish.publishWill(new WillRepository.WillEntry("status/retained",
message, 0, true));
+
+ await().atMost(Duration.ofSeconds(5)).untilAsserted(() ->
+ assertArrayEquals(message,
Singleton.INST.get(TopicRepository.class).get("status/retained")));
+ }
+
+ @Test
+ public void testEmptyRetainedWillClearsStoredValue() {
+
when(subscribeRepository.getChannelsByTopic("status/cleared")).thenReturn(Collections.emptyMap());
+ TopicRepository topicRepository =
Singleton.INST.get(TopicRepository.class);
+ topicRepository.add("status/cleared",
"previous".getBytes(StandardCharsets.UTF_8));
+ await().atMost(Duration.ofSeconds(5)).untilAsserted(() ->
+ assertArrayEquals("previous".getBytes(StandardCharsets.UTF_8),
topicRepository.get("status/cleared")));
+
+ Publish.publishWill(new WillRepository.WillEntry("status/cleared", new
byte[0], 0, true));
+
+ assertNull(topicRepository.get("status/cleared"));
+ }
+
+ @Test
+ public void testPublishWillToActiveSubscriber() {
+ when(subscriberChannel.isActive()).thenReturn(true);
+ when(subscribeRepository.getChannelsByTopic("status/offline"))
+ .thenReturn(Collections.singletonMap(subscriberChannel,
MqttQoS.EXACTLY_ONCE));
+
+ byte[] message = "client lost".getBytes();
+ WillRepository.WillEntry will = new
WillRepository.WillEntry("status/offline", message, 1, true);
+ Publish.publishWill(will);
+
+ ArgumentCaptor<Object> captor = ArgumentCaptor.forClass(Object.class);
+ verify(subscriberChannel).writeAndFlush(captor.capture());
+ MqttPublishMessage published = (MqttPublishMessage) captor.getValue();
+ assertEquals("status/offline", published.variableHeader().topicName());
+ assertEquals(1, published.fixedHeader().qosLevel().value());
+ assertTrue(published.fixedHeader().isRetain());
+ }
+
+ @Test
+ public void testPublishWillSkipsInactiveChannel() {
+ when(subscriberChannel.isActive()).thenReturn(false);
+ when(subscribeRepository.getChannelsByTopic("status/inactive"))
+ .thenReturn(Collections.singletonMap(subscriberChannel,
MqttQoS.AT_LEAST_ONCE));
+
+ WillRepository.WillEntry will = new
WillRepository.WillEntry("status/inactive", "msg".getBytes(), 0, false);
+ Publish.publishWill(will);
+
+ verify(subscriberChannel, never()).writeAndFlush(any());
+ }
+
+ @Test
+ public void testPublishWillToEmptySubscribers() {
+
when(subscribeRepository.getChannelsByTopic("topic/none")).thenReturn(Collections.emptyMap());
+
+ WillRepository.WillEntry will = new
WillRepository.WillEntry("topic/none", "msg".getBytes(), 2, false);
+ Publish.publishWill(will);
+ }
+
+ @Test
+ public void testPublishWillQosAndRetain() {
+ when(subscriberChannel.isActive()).thenReturn(true);
+ when(subscribeRepository.getChannelsByTopic("qos/retain"))
+ .thenReturn(Collections.singletonMap(subscriberChannel,
MqttQoS.EXACTLY_ONCE));
+
+ WillRepository.WillEntry will = new
WillRepository.WillEntry("qos/retain", "data".getBytes(), 0, false);
+ Publish.publishWill(will);
+
+ ArgumentCaptor<Object> captor = ArgumentCaptor.forClass(Object.class);
+ verify(subscriberChannel).writeAndFlush(captor.capture());
+ MqttPublishMessage published = (MqttPublishMessage) captor.getValue();
+ assertEquals(0, published.fixedHeader().qosLevel().value());
+ assertFalse(published.fixedHeader().isRetain());
+ }
+
+ @Test
+ public void testPublishWillToWildcardSubscriber() {
+ when(subscriberChannel.isActive()).thenReturn(true);
+ when(subscribeRepository.getChannelsByTopic("status/client-001"))
+ .thenReturn(Collections.singletonMap(subscriberChannel,
MqttQoS.AT_MOST_ONCE));
+
+ WillRepository.WillEntry will = new
WillRepository.WillEntry("status/client-001", "gone".getBytes(), 0, false);
+ Publish.publishWill(will);
+
+ ArgumentCaptor<Object> captor = ArgumentCaptor.forClass(Object.class);
+ verify(subscriberChannel).writeAndFlush(captor.capture());
+ MqttPublishMessage published = (MqttPublishMessage) captor.getValue();
+ assertEquals("status/client-001",
published.variableHeader().topicName());
+ }
+
+ @Test
+ public void testPublishWillCapsQosAtSubscriberGrantedQos() {
+ when(subscriberChannel.isActive()).thenReturn(true);
+ when(otherChannel.isActive()).thenReturn(true);
+ Map<Channel, MqttQoS> subscribers = new HashMap<>();
+ subscribers.put(subscriberChannel, MqttQoS.AT_MOST_ONCE);
+ subscribers.put(otherChannel, MqttQoS.EXACTLY_ONCE);
+
when(subscribeRepository.getChannelsByTopic("status/qos-cap")).thenReturn(subscribers);
+
+ WillRepository.WillEntry will = new
WillRepository.WillEntry("status/qos-cap", "client lost".getBytes(), 2, false);
+ Publish.publishWill(will);
+
+ ArgumentCaptor<Object> captor = ArgumentCaptor.forClass(Object.class);
+ verify(subscriberChannel).writeAndFlush(captor.capture());
+ MqttPublishMessage toQos0Subscriber = (MqttPublishMessage)
captor.getValue();
+ assertEquals(MqttQoS.AT_MOST_ONCE,
toQos0Subscriber.fixedHeader().qosLevel());
+ assertEquals(0, toQos0Subscriber.variableHeader().packetId());
+
+ ArgumentCaptor<Object> otherCaptor =
ArgumentCaptor.forClass(Object.class);
+ verify(otherChannel).writeAndFlush(otherCaptor.capture());
+ MqttPublishMessage toQos2Subscriber = (MqttPublishMessage)
otherCaptor.getValue();
+ assertEquals(MqttQoS.EXACTLY_ONCE,
toQos2Subscriber.fixedHeader().qosLevel());
+ assertTrue(toQos2Subscriber.variableHeader().packetId() > 0);
+ }
+}
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/TopicMatcherTest.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/TopicMatcherTest.java
new file mode 100644
index 0000000000..561dc65b1f
--- /dev/null
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/TopicMatcherTest.java
@@ -0,0 +1,100 @@
+/*
+ * 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 org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TopicMatcherTest {
+
+ @Test
+ void testExactMatch() {
+ assertTrue(TopicMatcher.matches("sport/tennis/player1",
"sport/tennis/player1"));
+ assertFalse(TopicMatcher.matches("sport/tennis/player1",
"sport/tennis/player2"));
+ }
+
+ @Test
+ void testSingleLevelWildcard() {
+ assertTrue(TopicMatcher.matches("sport/+/player1",
"sport/tennis/player1"));
+ assertTrue(TopicMatcher.matches("sport/+/player1",
"sport/football/player1"));
+ assertTrue(TopicMatcher.matches("+/tennis/+", "sport/tennis/player1"));
+ assertFalse(TopicMatcher.matches("sport/+/player1",
"sport/tennis/stadium/player1"));
+ assertFalse(TopicMatcher.matches("sport/+", "sport"));
+ }
+
+ @Test
+ void testMultiLevelWildcard() {
+ assertTrue(TopicMatcher.matches("#", "sport/tennis/player1"));
+ assertTrue(TopicMatcher.matches("#", "sport"));
+ assertTrue(TopicMatcher.matches("sport/#", "sport"));
+ assertTrue(TopicMatcher.matches("sport/#", "sport/tennis"));
+ assertTrue(TopicMatcher.matches("sport/#", "sport/tennis/player1"));
+ assertTrue(TopicMatcher.matches("sport/#",
"sport/tennis/player1/ranking"));
+ assertFalse(TopicMatcher.matches("sport#", "sport/tennis"));
+ assertFalse(TopicMatcher.matches("sport/tennis#",
"sport/tennis/player1"));
+ }
+
+ @Test
+ void testMixedWildcards() {
+ assertTrue(TopicMatcher.matches("+/tennis/#", "sport/tennis/player1"));
+ assertTrue(TopicMatcher.matches("+/tennis/#",
"sport/tennis/player1/ranking"));
+ assertTrue(TopicMatcher.matches("sport/+/#", "sport/tennis/player1"));
+ assertFalse(TopicMatcher.matches("+/tennis/#",
"sport/football/player1"));
+ }
+
+ @Test
+ void testDollarTopicNotMatchedByLeadingWildcard() {
+ assertFalse(TopicMatcher.matches("#", "$SYS/broker/version"));
+ assertFalse(TopicMatcher.matches("+", "$SYS"));
+ assertFalse(TopicMatcher.matches("+/b", "$SYS/b"));
+ assertTrue(TopicMatcher.matches("$SYS/#", "$SYS/broker/version"));
+ assertTrue(TopicMatcher.matches("$SYS/+", "$SYS/broker"));
+ }
+
+ @Test
+ void testNullInput() {
+ assertFalse(TopicMatcher.matches(null, "topic"));
+ assertFalse(TopicMatcher.matches("topic", null));
+ }
+
+ @Test
+ void testValidFilter() {
+ assertTrue(TopicMatcher.isValidFilter("sport/tennis/player1"));
+ assertTrue(TopicMatcher.isValidFilter("sport/+/player1"));
+ assertTrue(TopicMatcher.isValidFilter("+"));
+ assertTrue(TopicMatcher.isValidFilter("#"));
+ assertTrue(TopicMatcher.isValidFilter("sport/#"));
+ assertTrue(TopicMatcher.isValidFilter("+/tennis/#"));
+ }
+
+ @Test
+ void testInvalidFilter() {
+ assertFalse(TopicMatcher.isValidFilter("sport#"));
+ assertFalse(TopicMatcher.isValidFilter("sport/tennis#"));
+ assertFalse(TopicMatcher.isValidFilter("#/sport"));
+ assertFalse(TopicMatcher.isValidFilter("sport/#/ranking"));
+ assertFalse(TopicMatcher.isValidFilter("sport+"));
+ assertFalse(TopicMatcher.isValidFilter("+tennis"));
+ assertFalse(TopicMatcher.isValidFilter("sport/+tennis"));
+ assertFalse(TopicMatcher.isValidFilter(""));
+ assertFalse(TopicMatcher.isValidFilter(null));
+ assertFalse(TopicMatcher.isValidFilter("sport/" + (char) 0));
+ }
+}
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/repositories/SubscribeRepositoryTest.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/repositories/SubscribeRepositoryTest.java
index 77e674f387..9a3117ad65 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/repositories/SubscribeRepositoryTest.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/repositories/SubscribeRepositoryTest.java
@@ -27,11 +27,13 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.time.Duration;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;
@@ -53,12 +55,28 @@ public final class SubscribeRepositoryTest {
private static final String OTHER_TOPIC = "test/other-topic";
- private static final List<String> ALL_TOPICS = Arrays.asList(ABSENT_TOPIC,
TOPIC, OTHER_TOPIC);
+ private static final String EXACT_TOPIC = "sport/tennis";
+
+ private static final String CHILD_TOPIC = "sport/tennis/player1";
+
+ private static final String SINGLE_LEVEL_FILTER = "sport/+/player1";
+
+ private static final String MULTI_LEVEL_FILTER = "sport/#";
+
+ private static final String MATCH_ALL_FILTER = "#";
+
+ private static final String CONCURRENT_TOPIC = "sport/concurrent";
+
+ private static final List<String> ALL_TOPICS = Arrays.asList(
+ ABSENT_TOPIC, TOPIC, OTHER_TOPIC,
+ EXACT_TOPIC, SINGLE_LEVEL_FILTER, MULTI_LEVEL_FILTER,
MATCH_ALL_FILTER, CONCURRENT_TOPIC);
private static final Duration TIMEOUT = Duration.ofSeconds(5);
private static final Duration POLL_INTERVAL = Duration.ofMillis(10);
+ private static final long JOIN_TIMEOUT_MILLIS = 5000L;
+
private SubscribeRepository repository;
private Channel channel;
@@ -197,6 +215,86 @@ public final class SubscribeRepositoryTest {
assertEquals(MqttQoS.AT_MOST_ONCE, repository.get(TOPIC).get(channel));
}
+ @Test
+ public void testGetChannelsByTopicExactMatch() {
+ repository.add(channel, Collections.singletonList(new
MqttTopicSubscription(EXACT_TOPIC, MqttQoS.AT_MOST_ONCE)));
+ awaitAssert(() ->
assertTrue(repository.getChannelsByTopic(EXACT_TOPIC).containsKey(channel)));
+ assertTrue(repository.getChannelsByTopic(CHILD_TOPIC).isEmpty());
+ }
+
+ @Test
+ public void testGetChannelsByTopicWildcardMatch() {
+ repository.add(channel, Collections.singletonList(new
MqttTopicSubscription(SINGLE_LEVEL_FILTER, MqttQoS.AT_MOST_ONCE)));
+ awaitAssert(() ->
assertTrue(repository.getChannelsByTopic(CHILD_TOPIC).containsKey(channel)));
+ assertTrue(repository.getChannelsByTopic(EXACT_TOPIC).isEmpty());
+ }
+
+ @Test
+ public void testGetChannelsByTopicDeduplicatesOverlappingSubscriptions() {
+ repository.add(channel, Arrays.asList(
+ new MqttTopicSubscription(MATCH_ALL_FILTER,
MqttQoS.AT_MOST_ONCE),
+ new MqttTopicSubscription(MULTI_LEVEL_FILTER,
MqttQoS.EXACTLY_ONCE)));
+ awaitAssert(() -> {
+ assertTrue(repository.get(MATCH_ALL_FILTER).containsKey(channel));
+
assertTrue(repository.get(MULTI_LEVEL_FILTER).containsKey(channel));
+ });
+
+ // MQTT requires at most one delivery per publish per client, at the
maximum qos of the matching filters
+ Map<Channel, MqttQoS> matched =
repository.getChannelsByTopic(EXACT_TOPIC);
+ assertEquals(1, matched.size());
+ assertEquals(MqttQoS.EXACTLY_ONCE, matched.get(channel));
+ }
+
+ @Test
+ public void testGetChannelsByTopicMultipleSubscribers() {
+ repository.add(channel, Collections.singletonList(new
MqttTopicSubscription(EXACT_TOPIC, MqttQoS.AT_MOST_ONCE)));
+ repository.add(otherChannel, Collections.singletonList(new
MqttTopicSubscription(EXACT_TOPIC, MqttQoS.AT_MOST_ONCE)));
+ awaitAssert(() -> assertEquals(2,
repository.getChannelsByTopic(EXACT_TOPIC).size()));
+
assertTrue(repository.getChannelsByTopic(EXACT_TOPIC).keySet().containsAll(Arrays.asList(channel,
otherChannel)));
+ }
+
+ /**
+ * Regression guard: subscribing to a brand new topic must keep every
client
+ * when several clients send SUBSCRIBE at the same time.
+ *
+ * @throws InterruptedException if a subscribing thread is interrupted
+ */
+ @Test
+ public void concurrentSubscribersOfTheSameNewTopicAreAllRegistered()
throws InterruptedException {
+ int subscriberCount = 8;
+ CountDownLatch startGate = new CountDownLatch(1);
+ List<Channel> subscribers = new ArrayList<>();
+ List<Thread> subscribingThreads = new ArrayList<>();
+ for (int i = 0; i < subscriberCount; i++) {
+ Channel subscriber = mock(Channel.class);
+ subscribers.add(subscriber);
+ Thread thread = new Thread(() -> subscribeAfter(startGate,
subscriber, CONCURRENT_TOPIC));
+ thread.start();
+ subscribingThreads.add(thread);
+ }
+
+ startGate.countDown();
+ for (Thread thread : subscribingThreads) {
+ thread.join(JOIN_TIMEOUT_MILLIS);
+ assertFalse(thread.isAlive(), "subscribing thread did not finish
in time");
+ }
+
+ awaitAssert(() -> assertEquals(subscriberCount,
repository.get(CONCURRENT_TOPIC).size()));
+ Map<Channel, MqttQoS> matched =
repository.getChannelsByTopic(CONCURRENT_TOPIC);
+ assertEquals(subscriberCount, matched.size());
+ assertTrue(matched.keySet().containsAll(subscribers));
+ }
+
+ private void subscribeAfter(final CountDownLatch startGate, final Channel
subscriber, final String topic) {
+ try {
+ startGate.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ repository.add(subscriber, Collections.singletonList(new
MqttTopicSubscription(topic, MqttQoS.AT_MOST_ONCE)));
+ }
+
/**
* The repository mutates its state asynchronously on the common pool,
* so assertions have to be retried until the mutation becomes visible.
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/repositories/WillRepositoryTest.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/repositories/WillRepositoryTest.java
new file mode 100644
index 0000000000..5692311a3c
--- /dev/null
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/repositories/WillRepositoryTest.java
@@ -0,0 +1,112 @@
+/*
+ * 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.repositories;
+
+import io.netty.channel.embedded.EmbeddedChannel;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.notNullValue;
+import static org.hamcrest.Matchers.nullValue;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class WillRepositoryTest {
+
+ private WillRepository willRepository;
+
+ private EmbeddedChannel channel;
+
+ @BeforeEach
+ public void setUp() {
+ willRepository = new WillRepository();
+ channel = new EmbeddedChannel();
+ }
+
+ @AfterEach
+ public void tearDown() {
+ channel.close();
+ }
+
+ @Test
+ public void testAddAndGetWillEntry() {
+ byte[] message = "offline".getBytes();
+ WillRepository.WillEntry entry = new
WillRepository.WillEntry("topic/test", message, 1, true);
+ willRepository.add(channel, entry);
+
+ WillRepository.WillEntry result = willRepository.get(channel);
+ assertThat(result, notNullValue());
+ assertEquals("topic/test", result.getTopic());
+ assertArrayEquals(message, result.getMessage());
+ assertEquals(1, result.getQos());
+ assertTrue(result.isRetain());
+ }
+
+ @Test
+ public void testRemoveWillEntry() {
+ WillRepository.WillEntry entry = new
WillRepository.WillEntry("topic/test", "bye".getBytes(), 0, false);
+ willRepository.add(channel, entry);
+ assertThat(willRepository.get(channel), notNullValue());
+
+ willRepository.remove(channel);
+ assertThat(willRepository.get(channel), nullValue());
+ }
+
+ @Test
+ public void testGetReturnsNullForUnknownChannel() {
+ assertThat(willRepository.get(channel), nullValue());
+ }
+
+ @Test
+ public void testWillEntryFields() {
+ byte[] message = "disconnected".getBytes();
+ WillRepository.WillEntry entry = new
WillRepository.WillEntry("alerts/disconnect", message, 2, false);
+
+ assertEquals("alerts/disconnect", entry.getTopic());
+ assertArrayEquals("disconnected".getBytes(), entry.getMessage());
+ assertEquals(2, entry.getQos());
+ assertFalse(entry.isRetain());
+ }
+
+ @Test
+ public void testReplaceWillEntryOnReconnect() {
+ WillRepository.WillEntry first = new
WillRepository.WillEntry("topic/a", "msg1".getBytes(), 0, false);
+ WillRepository.WillEntry second = new
WillRepository.WillEntry("topic/b", "msg2".getBytes(), 1, true);
+ willRepository.add(channel, first);
+ willRepository.add(channel, second);
+
+ WillRepository.WillEntry result = willRepository.get(channel);
+ assertEquals("topic/b", result.getTopic());
+ assertArrayEquals("msg2".getBytes(), result.getMessage());
+ }
+
+ @Test
+ public void testRemoveNonexistentChannel() {
+ EmbeddedChannel otherChannel = new EmbeddedChannel();
+ try {
+ willRepository.remove(otherChannel);
+ assertThat(willRepository.get(channel), nullValue());
+ } finally {
+ otherChannel.close();
+ }
+ }
+}
