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 171165b947 fix: set MQTT sessionPresent from cleanSession flag and
stored session state (#6889) (#6894)
171165b947 is described below
commit 171165b947477d5a08faf4a9dcc1459261fde8f2
Author: lll-peanut <[email protected]>
AuthorDate: Thu Oct 8 15:24:53 2026 +0800
fix: set MQTT sessionPresent from cleanSession flag and stored session
state (#6889) (#6894)
* fix: set MQTT sessionPresent from cleanSession flag and stored session
state (#6889)
* test: add unit tests for MQTT session present handling (#6889)
* fix: preserve MQTT session state on reconnect
Store MQTT session subscriptions with their QoS and resume them through the
QoS-aware subscribe repository path.
Dispatch DISCONNECT in MqttFactory so clean-session state is removed on
graceful disconnect, and unregister disconnected channels from subscriptions.
Use computeIfAbsent in SubscribeRepository to avoid concurrent
subscribe/resume overwrites.
* chore: trigger CI
* test: stabilize mqtt protocol tests
* ci: retry mqtt build
* ci: rerun integration tests
---------
Co-authored-by: aias00 <[email protected]>
---
.../org/apache/shenyu/protocol/mqtt/Connect.java | 28 ++-
.../apache/shenyu/protocol/mqtt/Disconnect.java | 22 +-
.../apache/shenyu/protocol/mqtt/MqttFactory.java | 3 +
.../shenyu/protocol/mqtt/MqttTransportHandler.java | 12 +-
.../org/apache/shenyu/protocol/mqtt/Subscribe.java | 10 +
.../apache/shenyu/protocol/mqtt/Unsubscribe.java | 9 +
.../protocol/mqtt/repositories/MqttSession.java | 103 +++++++++
.../mqtt/repositories/SessionRepository.java | 45 ++++
.../mqtt/repositories/SubscribeRepository.java | 19 +-
.../protocol/mqtt/ConnectSessionPresentTest.java | 177 +++++++++++++++
.../protocol/mqtt/DisconnectDispatchTest.java | 117 ++++++++++
.../shenyu/protocol/mqtt/MqttFactoryTest.java | 8 +-
.../protocol/mqtt/MqttSessionDisconnectTest.java | 247 +++++++++++++++++++++
.../protocol/mqtt/MqttTransportHandlerTest.java | 3 +
.../apache/shenyu/protocol/mqtt/PublishTest.java | 3 +
.../shenyu/protocol/mqtt/SubscribeQosTest.java | 124 +++++++++++
16 files changed, 904 insertions(+), 26 deletions(-)
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 b23b07a661..157aa33412 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
@@ -26,9 +26,14 @@ import io.netty.handler.codec.mqtt.MqttVersion;
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.MqttSession;
+import org.apache.shenyu.protocol.mqtt.repositories.SessionRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.SubscribeRepository;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.Objects;
+
import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE;
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;
@@ -74,9 +79,30 @@ public class Connect extends MessageType {
// record connect
Singleton.INST.get(ChannelRepository.class).add(ctx.channel(),
clientId);
+
+ boolean cleanSession = msg.variableHeader().isCleanSession();
+ SessionRepository sessionRepository =
Singleton.INST.get(SessionRepository.class);
+ MqttSession session = sessionRepository.get(clientId);
+ // MQTT-3.2.2-6/7: session present is only true when the client
connects
+ // with cleanSession=0 and the server holds a stored session for the
clientId.
+ boolean sessionPresent = !cleanSession && Objects.nonNull(session);
+ if (cleanSession) {
+ // A clean-session connect must discard any previously stored
session.
+ session = new MqttSession(clientId, true);
+ sessionRepository.add(clientId, session);
+ } else if (Objects.isNull(session)) {
+ session = new MqttSession(clientId, false);
+ sessionRepository.add(clientId, session);
+ }
+ if (sessionPresent && !session.getTopics().isEmpty()) {
+ // Resume the stored subscriptions (with their QoS) for the new
channel.
+ Singleton.INST.get(SubscribeRepository.class)
+ .add(ctx.channel(), session.getTopicSubscriptions());
+ }
+
MqttConnAckMessage ackMessage = MqttMessageBuilders.connAck()
.returnCode(MqttConnectReturnCode.CONNECTION_ACCEPTED)
- .sessionPresent(true)
+ .sessionPresent(sessionPresent)
.build();
ctx.writeAndFlush(ackMessage);
setConnected(ctx.channel(), true);
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 d802247732..b71a849656 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
@@ -21,8 +21,13 @@ 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.MqttSession;
+import org.apache.shenyu.protocol.mqtt.repositories.SessionRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.SubscribeRepository;
import org.apache.shenyu.protocol.mqtt.utils.MqttPacketIdGenerator;
+import java.util.Objects;
+
/**
* The DISCONNECT message is sent from the client to the server to indicate
* that it is about to close its TCP/IP connection.
@@ -38,13 +43,24 @@ public class Disconnect extends MessageType {
@Override
public void disconnect(final ChannelHandlerContext ctx) {
//// todo Last words
- //// todo Clean session
cleanChannel(ctx.channel());
ctx.close();
}
- private void cleanChannel(final Channel channel) {
- //// todo ttl
+ /**
+ * Discard clean session state and release resources for a disconnected
channel.
+ * @param channel disconnected channel
+ */
+ static void cleanChannel(final Channel channel) {
+ String clientId =
Singleton.INST.get(ChannelRepository.class).get(channel);
+ if (Objects.nonNull(clientId)) {
+ MqttSession session =
Singleton.INST.get(SessionRepository.class).get(clientId);
+ if (Objects.nonNull(session) && session.isCleanSession()) {
+ // A clean-session disconnect discards all stored session
state.
+ Singleton.INST.get(SessionRepository.class).remove(clientId);
+ }
+ }
+ Singleton.INST.get(SubscribeRepository.class).remove(channel);
Singleton.INST.get(ChannelRepository.class).remove(channel);
MqttPacketIdGenerator.remove(channel);
}
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/MqttFactory.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/MqttFactory.java
index 59954045b5..0cc421aa11 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/MqttFactory.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/MqttFactory.java
@@ -69,7 +69,10 @@ public class MqttFactory {
messageType.pubRel(ctx, msg);
break;
case PUBACK:
+ break;
case DISCONNECT:
+ messageType.disconnect(ctx);
+ break;
default:
break;
}
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 c1d1181747..b177c0ed35 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,7 +17,6 @@
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;
@@ -25,10 +24,6 @@ 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.ChannelRepository;
-import org.apache.shenyu.protocol.mqtt.repositories.SubscribeRepository;
-import org.apache.shenyu.protocol.mqtt.utils.MqttPacketIdGenerator;
/**
* mqtt transport handler.
@@ -51,16 +46,13 @@ public class MqttTransportHandler extends
ChannelInboundHandlerAdapter implement
@Override
public void channelInactive(final ChannelHandlerContext ctx) throws
Exception {
- Singleton.INST.get(ChannelRepository.class).remove(ctx.channel());
+ Disconnect.cleanChannel(ctx.channel());
ctx.fireChannelInactive();
}
@Override
public void operationComplete(final Future<? super Void> future) throws
Exception {
- Channel channel = ((ChannelFuture) future).channel();
- Singleton.INST.get(ChannelRepository.class).remove(channel);
- Singleton.INST.get(SubscribeRepository.class).remove(channel);
- MqttPacketIdGenerator.remove(channel);
+ Disconnect.cleanChannel(((ChannelFuture) future).channel());
}
}
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Subscribe.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Subscribe.java
index 1a9975cfd1..246c3ca4a2 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Subscribe.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Subscribe.java
@@ -32,11 +32,15 @@ import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.util.CharsetUtil;
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.MqttSession;
+import org.apache.shenyu.protocol.mqtt.repositories.SessionRepository;
import org.apache.shenyu.protocol.mqtt.repositories.SubscribeRepository;
import org.apache.shenyu.protocol.mqtt.repositories.TopicRepository;
import java.util.ArrayList;
import java.util.List;
+import java.util.Objects;
import java.util.stream.Collectors;
import static io.netty.channel.ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE;
@@ -69,6 +73,12 @@ public class Subscribe extends MessageType {
Singleton.INST.get(SubscribeRepository.class).add(ctx.channel(),
mqttTopicSubscriptions);
+ String clientId =
Singleton.INST.get(ChannelRepository.class).get(ctx.channel());
+ MqttSession session =
Singleton.INST.get(SessionRepository.class).get(clientId);
+ if (Objects.nonNull(session)) {
+ mqttTopicSubscriptions.forEach(subscription ->
session.addTopic(subscription.topicName(), subscription.qualityOfService()));
+ }
+
for (String ackTopic : ackTopics) {
String message =
Singleton.INST.get(TopicRepository.class).get(ackTopic);
if (StringUtils.isNotEmpty(message)) {
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Unsubscribe.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Unsubscribe.java
index 154b61af72..584b6260e5 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Unsubscribe.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Unsubscribe.java
@@ -25,9 +25,13 @@ import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.handler.codec.mqtt.MqttUnsubscribeMessage;
import io.netty.handler.codec.mqtt.MqttUnsubAckMessage;
import org.apache.shenyu.common.utils.Singleton;
+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 java.util.List;
+import java.util.Objects;
import static io.netty.channel.ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE;
import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from;
@@ -46,6 +50,11 @@ public class Unsubscribe extends MessageType {
List<String> topics = msg.payload().topics();
Channel channel = ctx.channel();
Singleton.INST.get(SubscribeRepository.class).remove(topics, channel);
+ String clientId =
Singleton.INST.get(ChannelRepository.class).get(channel);
+ MqttSession session =
Singleton.INST.get(SessionRepository.class).get(clientId);
+ if (Objects.nonNull(session)) {
+ topics.forEach(session::removeTopic);
+ }
int packetId = msg.variableHeader().messageId();
MqttFixedHeader mqttFixedHeader = new
MqttFixedHeader(MqttMessageType.UNSUBACK, false, MqttQoS.AT_MOST_ONCE, false,
0);
MqttUnsubAckMessage mqttUnsubAckMessage = new
MqttUnsubAckMessage(mqttFixedHeader, from(packetId));
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/repositories/MqttSession.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/repositories/MqttSession.java
new file mode 100644
index 0000000000..cedb97f7bd
--- /dev/null
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/repositories/MqttSession.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.repositories;
+
+import io.netty.handler.codec.mqtt.MqttQoS;
+import io.netty.handler.codec.mqtt.MqttTopicSubscription;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+/**
+ * MQTT session state stored for a client.
+ */
+public class MqttSession {
+
+ private final String clientId;
+
+ private final boolean cleanSession;
+
+ private final Map<String, MqttQoS> topics = new ConcurrentHashMap<>();
+
+ /**
+ * MqttSession constructor.
+ * @param clientId clientId
+ * @param cleanSession cleanSession
+ */
+ public MqttSession(final String clientId, final boolean cleanSession) {
+ this.clientId = clientId;
+ this.cleanSession = cleanSession;
+ }
+
+ /**
+ * get clientId.
+ * @return clientId
+ */
+ public String getClientId() {
+ return clientId;
+ }
+
+ /**
+ * get cleanSession.
+ * @return cleanSession
+ */
+ public boolean isCleanSession() {
+ return cleanSession;
+ }
+
+ /**
+ * add topic with its QoS, replacing any existing subscription for the
same topic.
+ * @param topic topic
+ * @param qos qos
+ * @return true if the topic was newly added
+ */
+ public boolean addTopic(final String topic, final MqttQoS qos) {
+ return Objects.isNull(topics.put(topic, qos));
+ }
+
+ /**
+ * remove topic.
+ * @param topic topic
+ * @return true if the topic was removed
+ */
+ public boolean removeTopic(final String topic) {
+ return Objects.nonNull(topics.remove(topic));
+ }
+
+ /**
+ * get topics.
+ * @return topics
+ */
+ public Set<String> getTopics() {
+ return topics.keySet();
+ }
+
+ /**
+ * get topic subscriptions with their QoS for session resume.
+ * @return topic subscriptions
+ */
+ public List<MqttTopicSubscription> getTopicSubscriptions() {
+ return topics.entrySet().stream()
+ .map(entry -> new MqttTopicSubscription(entry.getKey(),
entry.getValue()))
+ .collect(Collectors.toList());
+ }
+}
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/repositories/SessionRepository.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/repositories/SessionRepository.java
new file mode 100644
index 0000000000..510117968e
--- /dev/null
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/repositories/SessionRepository.java
@@ -0,0 +1,45 @@
+/*
+ * 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 java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Session repository, keyed by clientId.
+ */
+public class SessionRepository implements BaseRepository<String, MqttSession> {
+
+ private static final Map<String, MqttSession> SESSION_FACTORY = new
ConcurrentHashMap<>();
+
+ @Override
+ public void add(final String clientId, final MqttSession session) {
+ SESSION_FACTORY.put(clientId, session);
+ }
+
+ @Override
+ public void remove(final String clientId) {
+ SESSION_FACTORY.remove(clientId);
+ }
+
+ @Override
+ public MqttSession get(final String clientId) {
+ return SESSION_FACTORY.get(clientId);
+ }
+
+}
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 befd4a5c8e..2bd69d50c1 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
@@ -27,7 +27,6 @@ import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
-import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
/**
@@ -41,10 +40,10 @@ public class SubscribeRepository implements
BaseRepository<List<String>, Map<Cha
@Override
public void add(final List<String> topics, final Map<Channel, MqttQoS>
channelQos) {
- CompletableFuture.runAsync(() -> topics.parallelStream().forEach(topic
->
+ topics.forEach(topic ->
channelQos.forEach((channel, qos) -> TOPIC_CHANNEL_FACTORY
.computeIfAbsent(topic, key -> new
ConcurrentHashMap<>())
- .merge(channel, qos, SubscribeRepository::maxQoS))));
+ .merge(channel, qos, SubscribeRepository::maxQoS)));
}
/**
@@ -53,16 +52,17 @@ public class SubscribeRepository implements
BaseRepository<List<String>, Map<Cha
* @param mqttTopicSubscription mqtt subscription info
*/
public void add(final Channel channel, final List<MqttTopicSubscription>
mqttTopicSubscription) {
- CompletableFuture.runAsync(() -> mqttTopicSubscription.parallelStream()
+ // Complete registration before returning so a later disconnect cannot
overtake it.
+ mqttTopicSubscription.stream()
.filter(s -> s.qualityOfService() != MqttQoS.FAILURE)
.forEach(s -> TOPIC_CHANNEL_FACTORY
.computeIfAbsent(s.topicName(), key -> new
ConcurrentHashMap<>())
- .merge(channel, s.qualityOfService(),
SubscribeRepository::maxQoS)));
+ .merge(channel, s.qualityOfService(),
SubscribeRepository::maxQoS));
}
@Override
public void remove(final List<String> topics) {
- CompletableFuture.runAsync(() ->
topics.parallelStream().forEach(TOPIC_CHANNEL_FACTORY::remove));
+ topics.forEach(TOPIC_CHANNEL_FACTORY::remove);
}
/**
@@ -71,12 +71,12 @@ public class SubscribeRepository implements
BaseRepository<List<String>, Map<Cha
* @param channel channel
*/
public void remove(final List<String> topics, final Channel channel) {
- CompletableFuture.runAsync(() -> topics.parallelStream().forEach(topic
-> {
+ topics.forEach(topic -> {
Map<Channel, MqttQoS> subscribers =
TOPIC_CHANNEL_FACTORY.get(topic);
if (Objects.nonNull(subscribers)) {
subscribers.remove(channel);
}
- }));
+ });
}
/**
@@ -84,8 +84,7 @@ public class SubscribeRepository implements
BaseRepository<List<String>, Map<Cha
* @param channel channel
*/
public void remove(final Channel channel) {
- CompletableFuture.runAsync(() ->
TOPIC_CHANNEL_FACTORY.values().parallelStream()
- .forEach(subscribers -> subscribers.remove(channel)));
+ TOPIC_CHANNEL_FACTORY.values().forEach(subscribers ->
subscribers.remove(channel));
}
@Override
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/ConnectSessionPresentTest.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/ConnectSessionPresentTest.java
new file mode 100644
index 0000000000..c2b57536ba
--- /dev/null
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/ConnectSessionPresentTest.java
@@ -0,0 +1,177 @@
+/*
+ * 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.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.mqtt.MqttConnAckMessage;
+import io.netty.handler.codec.mqtt.MqttConnectMessage;
+import io.netty.handler.codec.mqtt.MqttConnectPayload;
+import io.netty.handler.codec.mqtt.MqttConnectVariableHeader;
+import io.netty.handler.codec.mqtt.MqttFixedHeader;
+import io.netty.handler.codec.mqtt.MqttMessageType;
+import io.netty.handler.codec.mqtt.MqttMessageBuilders;
+import io.netty.handler.codec.mqtt.MqttQoS;
+import io.netty.handler.codec.mqtt.MqttTopicSubscription;
+import io.netty.handler.codec.mqtt.MqttVersion;
+import io.netty.util.Attribute;
+import io.netty.util.AttributeKey;
+import org.apache.shenyu.common.utils.Singleton;
+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.awaitility.Awaitility;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import static
io.netty.handler.codec.mqtt.MqttConnectReturnCode.CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION;
+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.clearInvocations;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Test case for {@link Connect} session present behavior.
+ */
+public class ConnectSessionPresentTest {
+
+ private Connect connect;
+
+ private ChannelHandlerContext ctx;
+
+ private Channel channel;
+
+ private Attribute<Boolean> connectedAttribute;
+
+ @BeforeEach
+ public void setUp() {
+ Singleton.INST.single(ChannelRepository.class, new
ChannelRepository());
+ Singleton.INST.single(SessionRepository.class, new
SessionRepository());
+ Singleton.INST.single(SubscribeRepository.class, new
SubscribeRepository());
+ MqttContext mqttContext = new MqttContext();
+ mqttContext.setUserName("shenyu");
+ mqttContext.setPassword("shenyu");
+ connect = new Connect();
+ ctx = mock(ChannelHandlerContext.class);
+ channel = mock(Channel.class);
+ connectedAttribute = mock(Attribute.class);
+ ChannelFuture channelFuture = mock(ChannelFuture.class);
+ when(ctx.channel()).thenReturn(channel);
+
when(channel.attr(any(AttributeKey.class))).thenReturn(connectedAttribute);
+ when(connectedAttribute.get()).thenReturn(false);
+ when(ctx.writeAndFlush(any())).thenReturn(channelFuture);
+ when(ctx.close()).thenReturn(channelFuture);
+ }
+
+ @Test
+ public void firstConnectWithCleanSessionFalseShouldNotPresentSession() {
+ MqttConnAckMessage ack = connectWithCleanSession("client-first",
false);
+ assertFalse(ack.variableHeader().isSessionPresent());
+ }
+
+ @Test
+ public void reconnectWithCleanSessionFalseShouldPresentSession() {
+ String clientId = "client-reconnect";
+ assertFalse(connectWithCleanSession(clientId,
false).variableHeader().isSessionPresent());
+ assertTrue(connectWithCleanSession(clientId,
false).variableHeader().isSessionPresent());
+ }
+
+ @Test
+ public void connectWithCleanSessionTrueShouldNotPresentSession() {
+ MqttConnAckMessage ack = connectWithCleanSession("client-clean", true);
+ assertFalse(ack.variableHeader().isSessionPresent());
+ }
+
+ @Test
+ public void cleanSessionConnectShouldDiscardPreviousSession() {
+ String clientId = "client-discard";
+ assertFalse(connectWithCleanSession(clientId,
false).variableHeader().isSessionPresent());
+ assertFalse(connectWithCleanSession(clientId,
true).variableHeader().isSessionPresent());
+ MqttSession session =
Singleton.INST.get(SessionRepository.class).get(clientId);
+ assertTrue(session.isCleanSession());
+ assertTrue(session.getTopics().isEmpty());
+ }
+
+ @Test
+ public void resumeSessionShouldReRegisterSubscriptionsToNewChannel() {
+ String clientId = "client-resume";
+ MqttSession storedSession = new MqttSession(clientId, false);
+ storedSession.addTopic("test/topic", MqttQoS.AT_LEAST_ONCE);
+ Singleton.INST.get(SessionRepository.class).add(clientId,
storedSession);
+
+ MqttConnAckMessage ack = connectWithCleanSession(clientId, false);
+ assertTrue(ack.variableHeader().isSessionPresent());
+
+ MqttSession session =
Singleton.INST.get(SessionRepository.class).get(clientId);
+ assertTrue(session.getTopics().contains("test/topic"));
+ List<MqttTopicSubscription> subscriptions =
session.getTopicSubscriptions();
+ assertEquals(1, subscriptions.size());
+ assertEquals("test/topic", subscriptions.get(0).topicName());
+ assertEquals(MqttQoS.AT_LEAST_ONCE,
subscriptions.get(0).qualityOfService());
+
+ SubscribeRepository subscribeRepository =
Singleton.INST.get(SubscribeRepository.class);
+ Awaitility.await().atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(() ->
assertTrue(subscribeRepository.get("test/topic").containsKey(channel)));
+ }
+
+ @Test
+ public void
unsupportedProtocolVersionShouldNotAuthenticateOrStoreSession() {
+ MqttConnectMessage msg = new MqttConnectMessage(
+ new MqttFixedHeader(MqttMessageType.CONNECT, false,
MqttQoS.AT_MOST_ONCE, false, 0),
+ new MqttConnectVariableHeader("MQTT", 6, true, true, false, 0,
false, false, 60),
+ new MqttConnectPayload("client-mqtt-6", null, null, "shenyu",
+ "shenyu".getBytes(StandardCharsets.UTF_8)));
+
+ connect.connect(ctx, msg);
+
+ ArgumentCaptor<Object> captor = ArgumentCaptor.forClass(Object.class);
+ verify(ctx).writeAndFlush(captor.capture());
+ MqttConnAckMessage ack = (MqttConnAckMessage) captor.getValue();
+ assertEquals(CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION,
ack.variableHeader().connectReturnCode());
+ verify(ctx).channel();
+ verify(ctx).close();
+
assertNull(Singleton.INST.get(SessionRepository.class).get("client-mqtt-6"));
+ }
+
+ private MqttConnAckMessage connectWithCleanSession(final String clientId,
final boolean cleanSession) {
+ MqttConnectMessage msg = MqttMessageBuilders.connect()
+ .clientId(clientId)
+ .cleanSession(cleanSession)
+ .protocolVersion(MqttVersion.MQTT_3_1)
+ .username("shenyu")
+ .password("shenyu".getBytes(StandardCharsets.UTF_8))
+ .build();
+ clearInvocations(ctx);
+ connect.connect(ctx, msg);
+ ArgumentCaptor<Object> captor = ArgumentCaptor.forClass(Object.class);
+ verify(ctx).writeAndFlush(captor.capture());
+ return (MqttConnAckMessage) captor.getValue();
+ }
+}
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
new file mode 100644
index 0000000000..6509ddd686
--- /dev/null
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/DisconnectDispatchTest.java
@@ -0,0 +1,117 @@
+/*
+ * 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 io.netty.handler.codec.mqtt.MqttMessageBuilders;
+import io.netty.handler.codec.mqtt.MqttQoS;
+import org.apache.shenyu.common.utils.Singleton;
+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.awaitility.Awaitility;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+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.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Test case for {@link MqttFactory} DISCONNECT dispatch and clean-session
cleanup.
+ */
+public class DisconnectDispatchTest {
+
+ private ChannelHandlerContext ctx;
+
+ private Channel channel;
+
+ @BeforeEach
+ public void setUp() {
+ Singleton.INST.single(ChannelRepository.class, new
ChannelRepository());
+ Singleton.INST.single(SessionRepository.class, new
SessionRepository());
+ Singleton.INST.single(SubscribeRepository.class, new
SubscribeRepository());
+ ctx = mock(ChannelHandlerContext.class);
+ channel = mock(Channel.class);
+ when(ctx.channel()).thenReturn(channel);
+ }
+
+ @Test
+ public void cleanSessionDisconnectShouldRemoveStoredSession() {
+ String clientId = "client-clean";
+ registerChannel(clientId);
+ Singleton.INST.get(SessionRepository.class).add(clientId, new
MqttSession(clientId, true));
+
+ new MqttFactory(MqttMessageBuilders.disconnect().build(),
ctx).connect();
+
+ assertNull(Singleton.INST.get(SessionRepository.class).get(clientId));
+ }
+
+ @Test
+ public void
persistentSessionDisconnectShouldKeepStoredSessionAndUnregisterChannel() {
+ String clientId = "client-persistent";
+ String topic = "test/persistent";
+ registerChannel(clientId);
+ MqttSession session = new MqttSession(clientId, false);
+ session.addTopic(topic, MqttQoS.AT_MOST_ONCE);
+ Singleton.INST.get(SessionRepository.class).add(clientId, session);
+ Singleton.INST.get(SubscribeRepository.class).add(channel,
session.getTopicSubscriptions());
+ awaitSubscribed(topic);
+
+ new MqttFactory(MqttMessageBuilders.disconnect().build(),
ctx).connect();
+
+
assertNotNull(Singleton.INST.get(SessionRepository.class).get(clientId));
+ Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() ->
+
assertFalse(Singleton.INST.get(SubscribeRepository.class).get(topic).containsKey(channel)));
+ }
+
+ @Test
+ public void cleanSessionDisconnectShouldUnregisterSubscribedChannel() {
+ String clientId = "client-clean-sub";
+ String topic = "test/clean-sub";
+ registerChannel(clientId);
+ MqttSession session = new MqttSession(clientId, true);
+ session.addTopic(topic, MqttQoS.AT_MOST_ONCE);
+ Singleton.INST.get(SessionRepository.class).add(clientId, session);
+ Singleton.INST.get(SubscribeRepository.class).add(channel,
session.getTopicSubscriptions());
+ awaitSubscribed(topic);
+
+ new MqttFactory(MqttMessageBuilders.disconnect().build(),
ctx).connect();
+
+ Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() ->
+
assertFalse(Singleton.INST.get(SubscribeRepository.class).get(topic).containsKey(channel)));
+ }
+
+ private void registerChannel(final String clientId) {
+ Singleton.INST.get(ChannelRepository.class).add(channel, clientId);
+ Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() ->
+ assertEquals(clientId,
Singleton.INST.get(ChannelRepository.class).get(channel)));
+ }
+
+ private void awaitSubscribed(final String topic) {
+ Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() ->
+ assertEquals(1,
Singleton.INST.get(SubscribeRepository.class).get(topic).size()));
+ }
+}
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 deb0550121..cecdd764d2 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
@@ -40,6 +40,8 @@ import io.netty.handler.codec.mqtt.MqttUnsubscribePayload;
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.SessionRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.SubscribeRepository;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -65,6 +67,8 @@ public final class MqttFactoryTest {
@BeforeEach
public void setUp() {
Singleton.INST.single(ChannelRepository.class, new
ChannelRepository());
+ Singleton.INST.single(SessionRepository.class, new
SessionRepository());
+ Singleton.INST.single(SubscribeRepository.class, new
SubscribeRepository());
new MqttContext().setUserName(USER_NAME);
new MqttContext().setPassword(PASSWORD);
}
@@ -149,13 +153,13 @@ public final class MqttFactoryTest {
}
@Test
- public void disconnectShouldFallThroughToNoOp() {
+ public void disconnectShouldCloseChannel() {
EmbeddedChannel channel = new EmbeddedChannel(new
ChannelInboundHandlerAdapter());
ChannelHandlerContext ctx = channel.pipeline().lastContext();
new MqttFactory(new
MqttMessage(fixedHeader(MqttMessageType.DISCONNECT)), ctx).connect();
- assertTrue(channel.isActive());
+ assertFalse(channel.isActive());
channel.finishAndReleaseAll();
}
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/MqttSessionDisconnectTest.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/MqttSessionDisconnectTest.java
new file mode 100644
index 0000000000..e42a6aaf5c
--- /dev/null
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/MqttSessionDisconnectTest.java
@@ -0,0 +1,247 @@
+/*
+ * 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.embedded.EmbeddedChannel;
+import io.netty.handler.codec.mqtt.MqttConnAckMessage;
+import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
+import io.netty.handler.codec.mqtt.MqttMessage;
+import io.netty.handler.codec.mqtt.MqttMessageBuilders;
+import io.netty.handler.codec.mqtt.MqttMessageType;
+import io.netty.handler.codec.mqtt.MqttQoS;
+import io.netty.handler.codec.mqtt.MqttTopicSubscription;
+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.MqttSession;
+import org.apache.shenyu.protocol.mqtt.repositories.SessionRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.SubscribeRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.TopicRepository;
+import org.apache.shenyu.protocol.mqtt.utils.MqttPacketIdGenerator;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+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.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Regression tests for session cleanup on graceful and abnormal disconnects.
+ */
+public final class MqttSessionDisconnectTest {
+
+ private static final String CLIENT_ID = "session-disconnect-client";
+
+ private static final String TOPIC = "test/session-disconnect";
+
+ private static final String OTHER_TOPIC = "test/session-disconnect-other";
+
+ private static final String USER_NAME = "test-user";
+
+ private static final String PASSWORD = "test-password";
+
+ private final List<EmbeddedChannel> channels = new ArrayList<>();
+
+ private final ChannelRepository channelRepository = new
ChannelRepository();
+
+ private final SessionRepository sessionRepository = new
SessionRepository();
+
+ private final SubscribeRepository subscribeRepository = new
SubscribeRepository();
+
+ @BeforeEach
+ public void setUp() {
+ Singleton.INST.single(ChannelRepository.class, channelRepository);
+ Singleton.INST.single(SessionRepository.class, sessionRepository);
+ Singleton.INST.single(SubscribeRepository.class, subscribeRepository);
+ Singleton.INST.single(TopicRepository.class, new TopicRepository());
+ sessionRepository.remove(CLIENT_ID);
+ new MqttContext().setUserName(USER_NAME);
+ new MqttContext().setPassword(PASSWORD);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ channels.forEach(channel -> {
+ channel.finishAndReleaseAll();
+ channelRepository.remove(channel);
+ subscribeRepository.remove(channel);
+ MqttPacketIdGenerator.remove(channel);
+ });
+ await().atMost(Duration.ofSeconds(5)).untilAsserted(() ->
channels.forEach(channel -> {
+ assertFalse(subscribeRepository.get(TOPIC).containsKey(channel));
+
assertFalse(subscribeRepository.get(OTHER_TOPIC).containsKey(channel));
+ }));
+ sessionRepository.remove(CLIENT_ID);
+ new MqttContext().setUserName(null);
+ new MqttContext().setPassword(null);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void cleanSessionDisconnectDiscardsStateBeforeReconnect(final
boolean graceful) {
+ EmbeddedChannel channel = connect(true, false);
+ subscribe(channel);
+ assertTrue(sessionRepository.get(CLIENT_ID).isCleanSession());
+ assertEquals(2, sessionRepository.get(CLIENT_ID).getTopics().size());
+ assertEquals(1, MqttPacketIdGenerator.next(channel));
+
+ disconnect(channel, graceful);
+
+ assertFalse(channel.isActive());
+ assertNull(channelRepository.get(channel));
+ assertNull(sessionRepository.get(CLIENT_ID));
+ assertUnsubscribed(channel);
+ assertEquals(1, MqttPacketIdGenerator.next(channel));
+
+ EmbeddedChannel reconnected = connect(false, false);
+ assertTrue(sessionRepository.get(CLIENT_ID).getTopics().isEmpty());
+ assertUnsubscribed(reconnected);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void
persistentSessionDisconnectPreservesSubscriptionsForReconnect(final boolean
graceful) {
+ EmbeddedChannel channel = connect(false, false);
+ subscribe(channel);
+ final MqttSession session = sessionRepository.get(CLIENT_ID);
+
+ disconnect(channel, graceful);
+
+ assertFalse(channel.isActive());
+ assertNull(channelRepository.get(channel));
+ assertSame(session, sessionRepository.get(CLIENT_ID));
+ assertEquals(2, session.getTopics().size());
+ assertUnsubscribed(channel);
+
+ EmbeddedChannel reconnected = connect(false, true);
+ await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> {
+ assertEquals(MqttQoS.AT_LEAST_ONCE,
subscribeRepository.get(TOPIC).get(reconnected));
+ assertEquals(MqttQoS.EXACTLY_ONCE,
subscribeRepository.get(OTHER_TOPIC).get(reconnected));
+ });
+ assertUnsubscribed(channel);
+ }
+
+ @Test
+ public void closeFutureDiscardsCleanSessionBeforeChannelInactive() {
+ EmbeddedChannel channel = connect(true, false);
+ channel.closeFuture().addListener(new MqttTransportHandler());
+ subscribe(channel);
+
+ channel.close();
+
+ assertNull(sessionRepository.get(CLIENT_ID));
+ assertNull(channelRepository.get(channel));
+ assertUnsubscribed(channel);
+ }
+
+ @Test
+ public void
abruptCloseImmediatelyAfterSubscribeRemovesOnlyDisconnectedChannel() {
+ EmbeddedChannel otherChannel = newChannel();
+ subscribeRepository.add(otherChannel,
+ Collections.singletonList(new MqttTopicSubscription(TOPIC,
MqttQoS.AT_MOST_ONCE)));
+ EmbeddedChannel channel = connect(true, false);
+ channel.writeInbound(MqttMessageBuilders.subscribe()
+ .messageId(1)
+ .addSubscription(MqttQoS.AT_LEAST_ONCE, TOPIC)
+ .build());
+
+ channel.close();
+
+ assertNull(sessionRepository.get(CLIENT_ID));
+ assertFalse(subscribeRepository.get(TOPIC).containsKey(channel));
+ assertEquals(MqttQoS.AT_MOST_ONCE,
subscribeRepository.get(TOPIC).get(otherChannel));
+ }
+
+ @Test
+ public void channelWithoutStoredSessionCanClose() {
+ EmbeddedChannel channel = newChannel();
+ channelRepository.add(channel, CLIENT_ID);
+
+ channel.close();
+
+ assertFalse(channel.isActive());
+ assertNull(channelRepository.get(channel));
+ assertNull(sessionRepository.get(CLIENT_ID));
+ }
+
+ private EmbeddedChannel connect(final boolean cleanSession, final boolean
sessionPresent) {
+ EmbeddedChannel channel = newChannel();
+ channel.writeInbound(MqttMessageBuilders.connect()
+ .clientId(CLIENT_ID)
+ .cleanSession(cleanSession)
+ .protocolVersion(MqttVersion.MQTT_3_1_1)
+ .username(USER_NAME)
+ .password(PASSWORD.getBytes(StandardCharsets.UTF_8))
+ .build());
+ MqttConnAckMessage ack = channel.readOutbound();
+ assertNotNull(ack);
+ assertEquals(MqttConnectReturnCode.CONNECTION_ACCEPTED,
ack.variableHeader().connectReturnCode());
+ assertEquals(sessionPresent, ack.variableHeader().isSessionPresent());
+ return channel;
+ }
+
+ private EmbeddedChannel newChannel() {
+ EmbeddedChannel channel = new EmbeddedChannel(new
MqttTransportHandler());
+ channels.add(channel);
+ return channel;
+ }
+
+ private void subscribe(final EmbeddedChannel channel) {
+ channel.writeInbound(MqttMessageBuilders.subscribe()
+ .messageId(1)
+ .addSubscription(MqttQoS.AT_LEAST_ONCE, TOPIC)
+ .addSubscription(MqttQoS.EXACTLY_ONCE, OTHER_TOPIC)
+ .build());
+ MqttMessage ack = channel.readOutbound();
+ assertNotNull(ack);
+ assertEquals(MqttMessageType.SUBACK, ack.fixedHeader().messageType());
+ await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> {
+ assertEquals(MqttQoS.AT_LEAST_ONCE,
subscribeRepository.get(TOPIC).get(channel));
+ assertEquals(MqttQoS.EXACTLY_ONCE,
subscribeRepository.get(OTHER_TOPIC).get(channel));
+ });
+ }
+
+ private void disconnect(final EmbeddedChannel channel, final boolean
graceful) {
+ if (graceful) {
+ channel.writeInbound(MqttMessageBuilders.disconnect().build());
+ } else {
+ channel.close();
+ }
+ channel.runPendingTasks();
+ }
+
+ private void assertUnsubscribed(final EmbeddedChannel channel) {
+ await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> {
+ assertFalse(subscribeRepository.get(TOPIC).containsKey(channel));
+
assertFalse(subscribeRepository.get(OTHER_TOPIC).containsKey(channel));
+ });
+ }
+}
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 58c2466579..d2aeaf37c6 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
@@ -33,6 +33,7 @@ import io.netty.handler.codec.mqtt.MqttVersion;
import io.netty.util.CharsetUtil;
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.utils.MqttPacketIdGenerator;
import org.awaitility.core.ThrowingRunnable;
@@ -80,6 +81,7 @@ public final class MqttTransportHandlerTest {
@BeforeEach
public void setUp() {
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);
@@ -96,6 +98,7 @@ public final class MqttTransportHandlerTest {
@AfterEach
public void tearDown() {
+ Singleton.INST.get(SessionRepository.class).remove(CLIENT_ID);
MqttPacketIdGenerator.remove(registeredChannel);
CHANNEL_REPOSITORY.remove(registeredChannel);
SUBSCRIBE_REPOSITORY.remove(registeredChannel);
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/PublishTest.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/PublishTest.java
index a840b5dc0f..72fe5bc5eb 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/PublishTest.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/PublishTest.java
@@ -41,6 +41,7 @@ import io.netty.util.ReferenceCountUtil;
import org.apache.shenyu.common.utils.Singleton;
import org.apache.shenyu.protocol.mqtt.repositories.ChannelRepository;
import org.apache.shenyu.protocol.mqtt.repositories.SubscribeRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.SessionRepository;
import org.apache.shenyu.protocol.mqtt.repositories.TopicRepository;
import org.apache.shenyu.protocol.mqtt.utils.MqttPacketIdGenerator;
import org.awaitility.core.ThrowingRunnable;
@@ -122,6 +123,7 @@ public final class PublishTest {
Singleton.INST.single(SubscribeRepository.class, SUBSCRIBE_REPOSITORY);
Singleton.INST.single(TopicRepository.class, TOPIC_REPOSITORY);
Singleton.INST.single(ChannelRepository.class, CHANNEL_REPOSITORY);
+ Singleton.INST.single(SessionRepository.class, new
SessionRepository());
new MqttContext().setUserName(USER_NAME);
new MqttContext().setPassword(PASSWORD);
@@ -400,6 +402,7 @@ public final class PublishTest {
}
private void clearSharedState() {
+ Singleton.INST.get(SessionRepository.class).remove(CLIENT_ID);
ALL_TOPICS.forEach(TOPIC_REPOSITORY::remove);
SUBSCRIBE_REPOSITORY.remove(ALL_TOPICS);
awaitAssert(() -> ALL_TOPICS.forEach(topic ->
assertTrue(SUBSCRIBE_REPOSITORY.get(topic).isEmpty())));
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/SubscribeQosTest.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/SubscribeQosTest.java
new file mode 100644
index 0000000000..66dd613b32
--- /dev/null
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/SubscribeQosTest.java
@@ -0,0 +1,124 @@
+/*
+ * 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.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.mqtt.MqttMessageBuilders;
+import io.netty.handler.codec.mqtt.MqttQoS;
+import io.netty.handler.codec.mqtt.MqttSubscribeMessage;
+import io.netty.handler.codec.mqtt.MqttTopicSubscription;
+import io.netty.util.Attribute;
+import io.netty.util.AttributeKey;
+import org.apache.shenyu.common.utils.Singleton;
+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.TopicRepository;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Test case for {@link Subscribe} QoS retention in {@link MqttSession}.
+ */
+public class SubscribeQosTest {
+
+ private ChannelHandlerContext ctx;
+
+ private Channel channel;
+
+ private Attribute<Boolean> connectedAttribute;
+
+ @BeforeEach
+ public void setUp() {
+ Singleton.INST.single(ChannelRepository.class, new
ChannelRepository());
+ Singleton.INST.single(SessionRepository.class, new
SessionRepository());
+ Singleton.INST.single(SubscribeRepository.class, new
SubscribeRepository());
+ Singleton.INST.single(TopicRepository.class, new TopicRepository());
+ ctx = mock(ChannelHandlerContext.class);
+ channel = mock(Channel.class);
+ connectedAttribute = mock(Attribute.class);
+ when(ctx.channel()).thenReturn(channel);
+
when(channel.attr(any(AttributeKey.class))).thenReturn(connectedAttribute);
+ when(connectedAttribute.get()).thenReturn(true);
+ when(ctx.writeAndFlush(any())).thenReturn(mock(ChannelFuture.class));
+
when(channel.writeAndFlush(any())).thenReturn(mock(ChannelFuture.class));
+ }
+
+ @Test
+ public void subscribeShouldStoreQosInSession() {
+ String clientId = "client-qos";
+ Singleton.INST.get(ChannelRepository.class).add(channel, clientId);
+ MqttSession session = new MqttSession(clientId, false);
+ Singleton.INST.get(SessionRepository.class).add(clientId, session);
+ Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() ->
+ assertEquals(clientId,
Singleton.INST.get(ChannelRepository.class).get(channel)));
+
+ MqttSubscribeMessage msg = MqttMessageBuilders.subscribe()
+ .messageId(1)
+ .addSubscription(MqttQoS.AT_LEAST_ONCE, "test/qos")
+ .addSubscription(MqttQoS.EXACTLY_ONCE, "test/qos2")
+ .build();
+
+ new Subscribe().subscribe(ctx, msg);
+
+ MqttTopicSubscription qos1 = subscription(session, "test/qos");
+ MqttTopicSubscription qos2 = subscription(session, "test/qos2");
+ assertEquals(MqttQoS.AT_LEAST_ONCE, qos1.qualityOfService());
+ assertEquals(MqttQoS.EXACTLY_ONCE, qos2.qualityOfService());
+ }
+
+ @Test
+ public void resubscribeShouldReplacePreviousQos() {
+ String clientId = "client-resub";
+ Singleton.INST.get(ChannelRepository.class).add(channel, clientId);
+ MqttSession session = new MqttSession(clientId, false);
+ Singleton.INST.get(SessionRepository.class).add(clientId, session);
+ Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() ->
+ assertEquals(clientId,
Singleton.INST.get(ChannelRepository.class).get(channel)));
+
+ new Subscribe().subscribe(ctx, MqttMessageBuilders.subscribe()
+ .messageId(1)
+ .addSubscription(MqttQoS.EXACTLY_ONCE, "test/qos")
+ .build());
+ new Subscribe().subscribe(ctx, MqttMessageBuilders.subscribe()
+ .messageId(2)
+ .addSubscription(MqttQoS.AT_MOST_ONCE, "test/qos")
+ .build());
+
+ MqttTopicSubscription subscription = subscription(session, "test/qos");
+ assertEquals(MqttQoS.AT_MOST_ONCE, subscription.qualityOfService());
+ }
+
+ private MqttTopicSubscription subscription(final MqttSession session,
final String topic) {
+ return session.getTopicSubscriptions().stream()
+ .filter(subscription -> topic.equals(subscription.topicName()))
+ .findFirst()
+ .orElseThrow(AssertionError::new);
+ }
+}
