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 1b3baa856b fix(mqtt): fix duplicate startup and resource leaks on
config updates (#7539)
1b3baa856b is described below
commit 1b3baa856b804c3db2eb623db6da8ebd007945d1
Author: lymerin <[email protected]>
AuthorDate: Sun Oct 11 22:34:36 2026 +0800
fix(mqtt): fix duplicate startup and resource leaks on config updates
(#7539)
---
.../plugin/mqtt/handler/MqttPluginDataHandler.java | 49 ++--
.../mqtt/handler/MqttPluginDataHandlerTest.java | 322 ++++++++++++++++++---
.../shenyu/protocol/mqtt/MqttBootstrapServer.java | 48 ++-
.../protocol/mqtt/MqttServerConfiguration.java | 34 ++-
.../protocol/mqtt/MqttBootstrapServerTest.java | 140 ++++++++-
5 files changed, 492 insertions(+), 101 deletions(-)
diff --git
a/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-mqtt/src/main/java/org/apache/shenyu/plugin/mqtt/handler/MqttPluginDataHandler.java
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-mqtt/src/main/java/org/apache/shenyu/plugin/mqtt/handler/MqttPluginDataHandler.java
index 3cafba458b..9bac3c8f52 100644
---
a/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-mqtt/src/main/java/org/apache/shenyu/plugin/mqtt/handler/MqttPluginDataHandler.java
+++
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-mqtt/src/main/java/org/apache/shenyu/plugin/mqtt/handler/MqttPluginDataHandler.java
@@ -21,52 +21,43 @@ import org.apache.shenyu.common.dto.PluginData;
import org.apache.shenyu.common.enums.PluginEnum;
import org.apache.shenyu.common.utils.GsonUtils;
import org.apache.shenyu.plugin.base.handler.PluginDataHandler;
-import org.apache.shenyu.protocol.mqtt.BootstrapServer;
import org.apache.shenyu.protocol.mqtt.MqttBootstrapServer;
import org.apache.shenyu.protocol.mqtt.MqttServerConfiguration;
-import java.net.InetAddress;
-import java.net.Socket;
+import java.util.Objects;
/**
* The type Mqtt plugin data handler.
*/
public class MqttPluginDataHandler implements PluginDataHandler {
- private final BootstrapServer server = new MqttBootstrapServer();
+ private final MqttBootstrapServer server = new MqttBootstrapServer();
+
+ private MqttServerConfiguration runningConfiguration;
@Override
- public void handlerPlugin(final PluginData pluginData) {
+ public synchronized void handlerPlugin(final PluginData pluginData) {
+ if (!pluginData.getEnabled()) {
+ server.shutdown();
+ runningConfiguration = null;
+ return;
+ }
MqttServerConfiguration configuration =
GsonUtils.getInstance().fromJson(pluginData.getConfig(),
MqttServerConfiguration.class);
- configuration.afterPropertiesSet();
- if (pluginData.getEnabled()) {
- server.init();
- server.start();
- } else {
- if (isPortUsing(configuration.getPort())) {
- server.shutdown();
- }
+ if (configuration.hasSameRuntimeSettings(runningConfiguration) &&
server.isRunning()) {
+ return;
}
+ if (Objects.nonNull(runningConfiguration)) {
+ server.shutdown();
+ runningConfiguration = null;
+ }
+ configuration.afterPropertiesSet();
+ server.init();
+ server.start();
+ runningConfiguration = configuration;
}
@Override
public String pluginNamed() {
return PluginEnum.MQTT.getName();
}
-
- /**
- * Ture is use else false.
- *
- * @param port server port
- * @return boolean
- */
- private boolean isPortUsing(final int port) {
- boolean flag = false;
- try (Socket socket = new Socket(InetAddress.getByName("127.0.0.1"),
port)) {
- flag = true;
- } catch (Exception ignored) {
-
- }
- return flag;
- }
}
diff --git
a/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-mqtt/src/test/java/org/apache/shenyu/plugin/mqtt/handler/MqttPluginDataHandlerTest.java
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-mqtt/src/test/java/org/apache/shenyu/plugin/mqtt/handler/MqttPluginDataHandlerTest.java
index 1e66b443ed..bc827bb356 100644
---
a/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-mqtt/src/test/java/org/apache/shenyu/plugin/mqtt/handler/MqttPluginDataHandlerTest.java
+++
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-mqtt/src/test/java/org/apache/shenyu/plugin/mqtt/handler/MqttPluginDataHandlerTest.java
@@ -17,16 +17,46 @@
package org.apache.shenyu.plugin.mqtt.handler;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.EventLoopGroup;
+import io.netty.util.ResourceLeakDetector;
import org.apache.shenyu.common.dto.PluginData;
-import org.awaitility.Awaitility;
+import org.apache.shenyu.protocol.mqtt.MqttBootstrapServer;
+import org.apache.shenyu.protocol.mqtt.MqttContext;
+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.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
-import java.net.InetAddress;
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.lang.management.ManagementFactory;
+import java.lang.management.ThreadInfo;
+import java.lang.reflect.Field;
+import java.net.BindException;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.concurrent.Callable;
+import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
@@ -34,56 +64,266 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
*/
public class MqttPluginDataHandlerTest {
+ private static final String CONFIG =
"{\"port\":0,\"workerGroupThreadCount\":1}";
+
private MqttPluginDataHandler mqttPluginDataHandlerUnderTest;
+ private MqttBootstrapServer server;
+
@BeforeEach
public void setUp() {
mqttPluginDataHandlerUnderTest = new MqttPluginDataHandler();
+ server = getField(mqttPluginDataHandlerUnderTest, "server",
MqttBootstrapServer.class);
+ new MqttContext().setBossGroupThreadCount(1);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ server.shutdown();
+ new MqttContext().setBossGroupThreadCount(0);
+ ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.DISABLED);
+ }
+
+ @Test
+ public void testEnableAndDisableWithoutConfig() {
+ mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(true, CONFIG));
+ final ChannelFuture future = getField(server, "future",
ChannelFuture.class);
+ final EventLoopGroup bossGroup = getField(server, "bossGroup",
EventLoopGroup.class);
+ final EventLoopGroup workerGroup = getField(server, "workerGroup",
EventLoopGroup.class);
+ assertTrue(future.channel().isActive());
+
+ mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(false, null));
+
+ assertFalse(future.channel().isActive());
+ assertTrue(bossGroup.isTerminated());
+ assertTrue(workerGroup.isTerminated());
+ assertNull(getField(server, "future", ChannelFuture.class));
+ assertDoesNotThrow(() ->
mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(false, null)));
+ }
+
+ @Test
+ public void testEquivalentEnabledUpdatesKeepResources() {
+ mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(true, CONFIG));
+ final ChannelFuture future = getField(server, "future",
ChannelFuture.class);
+ final EventLoopGroup bossGroup = getField(server, "bossGroup",
EventLoopGroup.class);
+ final EventLoopGroup workerGroup = getField(server, "workerGroup",
EventLoopGroup.class);
+
+ mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(true, CONFIG));
+ // Defaults, JSON order, level case and the unused boss setting do not
change the effective configuration.
+ mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(true,
"{\"leakDetectorLevel\":\"disabled\",\"password\":\"shenyu\","
+ +
"\"userName\":\"shenyu\",\"maxPayloadSize\":65536,\"bossGroupThreadCount\":2,\"workerGroupThreadCount\":1,\"port\":0}"));
+
+ assertSame(future, getField(server, "future", ChannelFuture.class));
+ assertSame(bossGroup, getField(server, "bossGroup",
EventLoopGroup.class));
+ assertSame(workerGroup, getField(server, "workerGroup",
EventLoopGroup.class));
+ assertTrue(future.channel().isActive());
+ }
+
+ @ParameterizedTest
+ @MethodSource("configurationChanges")
+ public void testChangedConfigurationReplacesResources(final String
setting, final String value) throws Exception {
+ JsonObject initialConfig =
JsonParser.parseString(CONFIG).getAsJsonObject();
+ initialConfig.addProperty("port", unusedPort());
+ mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(true,
initialConfig.toString()));
+ final ChannelFuture oldFuture = getField(server, "future",
ChannelFuture.class);
+ final EventLoopGroup oldBoss = getField(server, "bossGroup",
EventLoopGroup.class);
+ final EventLoopGroup oldWorker = getField(server, "workerGroup",
EventLoopGroup.class);
+ String changedValue = "port".equals(setting) ?
Integer.toString(unusedPort()) : value;
+ JsonObject changedConfig = initialConfig.deepCopy();
+ changedConfig.add(setting, JsonParser.parseString(changedValue));
+
+ mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(true,
changedConfig.toString()));
+
+ assertFalse(oldFuture.channel().isActive());
+ assertTrue(oldBoss.isTerminated());
+ assertTrue(oldWorker.isTerminated());
+ final ChannelFuture replacement = getField(server, "future",
ChannelFuture.class);
+ assertNotSame(oldFuture, replacement);
+ assertTrue(replacement.channel().isActive());
+ }
+
+ @Test
+ public void testEncryptedPasswordUpdatesKeepResources() {
+ String encryptedConfig =
"{\"port\":0,\"workerGroupThreadCount\":1,\"password\":\"secret\",\"isEncryptPassword\":true,\"encryptMode\":\"MD5\"}";
+ mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(true,
encryptedConfig));
+ final ChannelFuture future = getField(server, "future",
ChannelFuture.class);
+ String effectivePassword = new MqttContext().getPassword();
+
+ mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(true,
encryptedConfig));
+ mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(true,
"{\"port\":0,\"workerGroupThreadCount\":1,\"password\":\"" + effectivePassword
+ "\"}"));
+
+ assertSame(future, getField(server, "future", ChannelFuture.class));
+ assertTrue(MqttContext.isValid("shenyu",
effectivePassword.getBytes(StandardCharsets.UTF_8)));
+ }
+
+ @Test
+ public void testChangedPortOnDisableClosesRunningListener() throws
Exception {
+ mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(true, CONFIG));
+ final ChannelFuture future = getField(server, "future",
ChannelFuture.class);
+
+ mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(false,
"{\"port\":" + unusedPort() + "}"));
+
+ assertFalse(future.channel().isActive());
+ assertNull(getField(server, "future", ChannelFuture.class));
+ }
+
+ @Test
+ public void testFailedReplacementCanRetryPreviousConfiguration() throws
Exception {
+ mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(true, CONFIG));
+ final ChannelFuture previous = getField(server, "future",
ChannelFuture.class);
+ try (ServerSocket occupied = new ServerSocket(0)) {
+ String failedConfig = "{\"port\":" + occupied.getLocalPort() +
",\"workerGroupThreadCount\":1}";
+ assertThrows(BindException.class, () ->
mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(true, failedConfig)));
+
+ assertFalse(previous.channel().isActive());
+ assertNull(getField(server, "future", ChannelFuture.class));
+ }
+
+ mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(true, CONFIG));
+
+ assertTrue(getField(server, "future",
ChannelFuture.class).channel().isActive());
}
@Test
- public void testEnableConfiguration() throws InterruptedException {
- final PluginData enablePluginData = new PluginData("pluginId",
"pluginName", "{\n"
- + " \"port\": 9500,"
- + " \"bossGroupThreadCount\": 1,"
- + " \"maxPayloadSize\": 65536,"
- + " \"workerGroupThreadCount\": 12,"
- + " \"userName\": \"shenyu\","
- + " \"password\": \"shenyu\","
- + " \"isEncryptPassword\": false,"
- + " \"encryptMode\": \"\","
- + " \"leakDetectorLevel\": \"DISABLED\""
- + "}", "0", true, null);
- mqttPluginDataHandlerUnderTest.handlerPlugin(enablePluginData);
- assertTrue(isPortUsing());
- final PluginData disablePluginData = new PluginData("pluginId",
"pluginName", "{\n"
- + " \"port\": 9500,"
- + " \"bossGroupThreadCount\": 1,"
- + " \"maxPayloadSize\": 65536,"
- + " \"workerGroupThreadCount\": 12,"
- + " \"userName\": \"shenyu\","
- + " \"password\": \"shenyu\","
- + " \"isEncryptPassword\": false,"
- + " \"encryptMode\": \"\","
- + " \"leakDetectorLevel\": \"DISABLED\""
- + "}", "0", false, null);
- mqttPluginDataHandlerUnderTest.handlerPlugin(disablePluginData);
-
- Awaitility.await()
- .atMost(5, TimeUnit.SECONDS)
- .untilAsserted(() -> assertFalse(isPortUsing()));
- }
-
- private boolean isPortUsing() {
- boolean flag = false;
+ public void testFailedStartCanRetryTheSameConfiguration() throws Exception
{
+ String config;
+ try (ServerSocket occupied = new ServerSocket(0)) {
+ config = "{\"port\":" + occupied.getLocalPort() +
",\"workerGroupThreadCount\":1}";
+ assertThrows(BindException.class, () ->
mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(true, config)));
+ }
+
+ mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(true, config));
+
+ assertTrue(getField(server, "future",
ChannelFuture.class).channel().isActive());
+ }
+
+ @Test
+ public void testDisableThenEnableTheSameConfiguration() {
+ assertDoesNotThrow(() ->
mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(false, CONFIG)));
+ mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(true, CONFIG));
+ final ChannelFuture previous = getField(server, "future",
ChannelFuture.class);
+ mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(false,
CONFIG));
+
+ mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(true, CONFIG));
+
+ final ChannelFuture current = getField(server, "future",
ChannelFuture.class);
+ assertNotSame(previous, current);
+ assertTrue(current.channel().isActive());
+ }
+
+ @Test
+ public void testRepeatedUpdatePreservesConnectionAndDisableClosesIt()
throws Exception {
+ mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(true, CONFIG));
+ final ChannelFuture future = getField(server, "future",
ChannelFuture.class);
+ int port = ((InetSocketAddress)
future.channel().localAddress()).getPort();
+ try (Socket client = new Socket("127.0.0.1", port)) {
+ client.setSoTimeout(5000);
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ DataOutputStream connect = new DataOutputStream(bytes);
+ connect.writeUTF("MQTT");
+ connect.writeByte(4);
+ connect.writeByte(0xC2);
+ connect.writeShort(60);
+ connect.writeUTF("lifecycle-test-client");
+ connect.writeUTF("shenyu");
+ connect.writeUTF("shenyu");
+ client.getOutputStream().write(0x10);
+ client.getOutputStream().write(bytes.size());
+ client.getOutputStream().write(bytes.toByteArray());
+ assertArrayEquals(new byte[]{0x20, 2, 0, 0},
client.getInputStream().readNBytes(4));
+
+ mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(true,
CONFIG));
+ client.getOutputStream().write(new byte[]{(byte) 0xC0, 0});
+ assertArrayEquals(new byte[]{(byte) 0xD0, 0},
client.getInputStream().readNBytes(2));
+
+ mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(false,
CONFIG));
+
+ assertEquals(-1, client.getInputStream().read());
+ }
+ }
+
+ @Test
+ public void testConcurrentEnabledUpdatesStartOnlyOnce() throws Exception {
+ Callable<ChannelFuture> update = () -> {
+ mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(true,
CONFIG));
+ return getField(server, "future", ChannelFuture.class);
+ };
+ final FutureTask<ChannelFuture> first = new FutureTask<>(update);
+ final FutureTask<ChannelFuture> second = new FutureTask<>(update);
+ Thread firstThread = new Thread(first);
+ Thread secondThread = new Thread(second);
try {
- InetAddress address = InetAddress.getByName("127.0.0.1");
- Socket socket = new Socket(address, 9500);
- flag = true;
- } catch (Exception ignored) {
+ // Pause the first update at the real server's startup monitor.
+ synchronized (server) {
+ firstThread.start();
+ awaitBlockedBy(firstThread, Thread.currentThread());
+ secondThread.start();
+ awaitBlockedBy(secondThread, firstThread);
+ }
+ assertSame(first.get(10, TimeUnit.SECONDS), second.get(10,
TimeUnit.SECONDS));
+ assertTrue(server.isRunning());
+ final EventLoopGroup bossGroup = getField(server, "bossGroup",
EventLoopGroup.class);
+ final EventLoopGroup workerGroup = getField(server, "workerGroup",
EventLoopGroup.class);
+ mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(false,
CONFIG));
+ assertTrue(bossGroup.isTerminated());
+ assertTrue(workerGroup.isTerminated());
+ } finally {
+ firstThread.interrupt();
+ secondThread.interrupt();
+ firstThread.join(TimeUnit.SECONDS.toMillis(10));
+ secondThread.join(TimeUnit.SECONDS.toMillis(10));
+ assertFalse(firstThread.isAlive());
+ assertFalse(secondThread.isAlive());
}
- return flag;
+ }
+
+ @Test
+ public void testSameConfigurationRestartsClosedListener() {
+ mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(true, CONFIG));
+ final ChannelFuture previous = getField(server, "future",
ChannelFuture.class);
+ final EventLoopGroup oldBoss = getField(server, "bossGroup",
EventLoopGroup.class);
+ final EventLoopGroup oldWorker = getField(server, "workerGroup",
EventLoopGroup.class);
+ previous.channel().close().syncUninterruptibly();
+ assertFalse(server.isRunning());
+ mqttPluginDataHandlerUnderTest.handlerPlugin(pluginData(true, CONFIG));
+
+ assertTrue(server.isRunning());
+ assertNotSame(previous, getField(server, "future",
ChannelFuture.class));
+ assertTrue(oldBoss.isTerminated());
+ assertTrue(oldWorker.isTerminated());
+ }
+
+ private void awaitBlockedBy(final Thread thread, final Thread owner) {
+ await().atMost(Duration.ofSeconds(5)).until(() -> {
+ ThreadInfo info =
ManagementFactory.getThreadMXBean().getThreadInfo(thread.getId());
+ return info.getThreadState() == Thread.State.BLOCKED &&
info.getLockOwnerId() == owner.getId();
+ });
+ }
+
+ private static Stream<Arguments> configurationChanges() {
+ return Stream.of(Arguments.of("port", "0"), Arguments.of("userName",
"\"changed\""), Arguments.of("password", "\"changed\""),
+ Arguments.of("maxPayloadSize", "1024"),
Arguments.of("workerGroupThreadCount", "2"), Arguments.of("leakDetectorLevel",
"\"SIMPLE\""));
+ }
+
+ private PluginData pluginData(final boolean enabled, final String config) {
+ return new PluginData("pluginId", "mqtt", config, "0", enabled, null);
+ }
+
+ private int unusedPort() throws Exception {
+ try (ServerSocket socket = new ServerSocket(0)) {
+ return socket.getLocalPort();
+ }
+ }
+
+ private <T> T getField(final Object target, final String name, final
Class<T> type) {
+ try {
+ Field field = target instanceof MqttBootstrapServer ?
MqttBootstrapServer.class.getDeclaredField(name) :
MqttPluginDataHandler.class.getDeclaredField(name);
+ field.setAccessible(true);
+ return type.cast(field.get(target));
+ } catch (ReflectiveOperationException e) {
+ throw new AssertionError(e);
+ }
}
}
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/MqttBootstrapServer.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/MqttBootstrapServer.java
index 3e21c24a20..fe417dee92 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/MqttBootstrapServer.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/MqttBootstrapServer.java
@@ -54,36 +54,58 @@ public class MqttBootstrapServer implements BootstrapServer
{
}
@Override
- public void start() {
- //// todo thread start mqtt server
-
ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.valueOf(ENV.getLeakDetectorLevel().toUpperCase(Locale.ROOT)));
- bossGroup = new NioEventLoopGroup(ENV.getBossGroupThreadCount());
- workerGroup = new NioEventLoopGroup(ENV.getWorkerGroupThreadCount());
- ServerBootstrap bootstrap = new ServerBootstrap();
- bootstrap.group(bossGroup, workerGroup)
- .channel(NioServerSocketChannel.class)
- .childHandler(new
MqttTransportServerInitializer(ENV.getMaxPayloadSize()));
+ public synchronized void start() {
+ if (isRunning()) {
+ return;
+ }
+ shutdown();
+ boolean started = false;
try {
- future = bootstrap.bind(ENV.getPort()).sync();
- //// todo log
+
ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.valueOf(ENV.getLeakDetectorLevel().toUpperCase(Locale.ROOT)));
+ bossGroup = new NioEventLoopGroup(ENV.getBossGroupThreadCount());
+ workerGroup = new
NioEventLoopGroup(ENV.getWorkerGroupThreadCount());
+ ServerBootstrap bootstrap = new ServerBootstrap();
+ bootstrap.group(bossGroup, workerGroup)
+ .channel(NioServerSocketChannel.class)
+ .childHandler(new
MqttTransportServerInitializer(ENV.getMaxPayloadSize()));
+ future = bootstrap.bind(ENV.getPort());
+ future.sync();
+ started = true;
} catch (InterruptedException e) {
- //// todo log
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException("Interrupted while starting MQTT
server", e);
+ } finally {
+ if (!started) {
+ shutdown();
+ }
}
}
@Override
- public void shutdown() {
+ public synchronized void shutdown() {
if (Objects.nonNull(future)) {
future.channel().close().syncUninterruptibly();
+ future = null;
}
if (Objects.nonNull(bossGroup)) {
bossGroup.shutdownGracefully().syncUninterruptibly();
+ bossGroup = null;
}
if (Objects.nonNull(workerGroup)) {
workerGroup.shutdownGracefully().syncUninterruptibly();
+ workerGroup = null;
}
}
+ /**
+ * Whether the MQTT listener is active.
+ *
+ * @return true when the listener is active
+ */
+ public synchronized boolean isRunning() {
+ return Objects.nonNull(future) && future.channel().isActive();
+ }
+
private void initRepositories() throws IllegalAccessException,
InstantiationException {
Reflections reflections = new Reflections(REPOSITORY_PACKAGE_NAME);
for (Class<? extends BaseRepository> clazz :
reflections.getSubTypesOf(BaseRepository.class)) {
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/MqttServerConfiguration.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/MqttServerConfiguration.java
index 5b210df88e..47738a1d75 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/MqttServerConfiguration.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/MqttServerConfiguration.java
@@ -19,6 +19,9 @@ package org.apache.shenyu.protocol.mqtt;
import org.apache.shenyu.protocol.mqtt.utils.EncryptUtil;
+import java.util.Locale;
+import java.util.Objects;
+
/**
* mqtt server configuration.
*/
@@ -47,11 +50,8 @@ public class MqttServerConfiguration {
*/
public void afterPropertiesSet() {
- if (isEncryptPassword()) {
- setPassword(encryptPassword());
- }
MqttContext context = new MqttContext();
- context.setPassword(getPassword());
+ context.setPassword(effectivePassword());
context.setPort(getPort());
context.setMaxPayloadSize(getMaxPayloadSize());
context.setUserName(getUserName());
@@ -59,8 +59,30 @@ public class MqttServerConfiguration {
context.setLeakDetectorLevel(getLeakDetectorLevel());
}
- private String encryptPassword() {
- return EncryptUtil.choose(getEncryptMode(), getPassword());
+ private String effectivePassword() {
+ return isEncryptPassword() ? EncryptUtil.choose(getEncryptMode(),
getPassword()) : getPassword();
+ }
+
+ /**
+ * Compare the settings applied to the MQTT context.
+ *
+ * @param other the configuration to compare
+ * @return true when both configurations have the same runtime settings
+ */
+ public boolean hasSameRuntimeSettings(final MqttServerConfiguration other)
{
+ if (Objects.isNull(other)) {
+ return false;
+ }
+ return port == other.port
+ && maxPayloadSize == other.maxPayloadSize
+ && workerGroupThreadCount == other.workerGroupThreadCount
+ && Objects.equals(userName, other.userName)
+ && Objects.equals(effectivePassword(),
other.effectivePassword())
+ &&
Objects.equals(normalizeLeakDetectorLevel(leakDetectorLevel),
normalizeLeakDetectorLevel(other.leakDetectorLevel));
+ }
+
+ private static String normalizeLeakDetectorLevel(final String level) {
+ return Objects.isNull(level) ? null : level.toUpperCase(Locale.ROOT);
}
/**
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/MqttBootstrapServerTest.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/MqttBootstrapServerTest.java
index 66e941c36e..fe8f1f717c 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/MqttBootstrapServerTest.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/MqttBootstrapServerTest.java
@@ -28,11 +28,25 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.lang.reflect.Field;
+import java.net.BindException;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
import java.time.Duration;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
@@ -40,8 +54,11 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
*/
public final class MqttBootstrapServerTest {
+ private MqttBootstrapServer server;
+
@BeforeEach
public void setUp() {
+ server = new MqttBootstrapServer();
MqttContext context = new MqttContext();
context.setPort(0);
context.setBossGroupThreadCount(1);
@@ -54,6 +71,7 @@ public final class MqttBootstrapServerTest {
@AfterEach
public void tearDown() {
+ server.shutdown();
MqttContext context = new MqttContext();
context.setPort(0);
context.setBossGroupThreadCount(0);
@@ -66,8 +84,6 @@ public final class MqttBootstrapServerTest {
@Test
public void initShouldRegisterAllRepositories() {
- MqttBootstrapServer server = new MqttBootstrapServer();
-
server.init();
assertNotNull(Singleton.INST.get(ChannelRepository.class));
@@ -76,26 +92,126 @@ public final class MqttBootstrapServerTest {
}
@Test
- public void startAndShutdownShouldReleaseChannelAndEventLoops() throws
Exception {
- MqttBootstrapServer server = new MqttBootstrapServer();
-
+ public void startAndShutdownShouldReleaseChannelAndEventLoops() {
server.start();
- ChannelFuture future = getField(server, "future", ChannelFuture.class);
+ final ChannelFuture future = getField(server, "future",
ChannelFuture.class);
+ final EventLoopGroup bossGroup = getField(server, "bossGroup",
EventLoopGroup.class);
+ final EventLoopGroup workerGroup = getField(server, "workerGroup",
EventLoopGroup.class);
assertTrue(future.channel().isActive());
server.shutdown();
assertFalse(future.channel().isActive());
- EventLoopGroup bossGroup = getField(server, "bossGroup",
EventLoopGroup.class);
- EventLoopGroup workerGroup = getField(server, "workerGroup",
EventLoopGroup.class);
await().atMost(Duration.ofSeconds(5)).until(bossGroup::isTerminated);
await().atMost(Duration.ofSeconds(5)).until(workerGroup::isTerminated);
+ assertNull(getField(server, "future", ChannelFuture.class));
+ assertNull(getField(server, "bossGroup", EventLoopGroup.class));
+ assertNull(getField(server, "workerGroup", EventLoopGroup.class));
+ }
+
+ @Test
+ public void repeatedStartShouldKeepTheSameResources() {
+ server.start();
+ final ChannelFuture future = getField(server, "future",
ChannelFuture.class);
+ final EventLoopGroup bossGroup = getField(server, "bossGroup",
EventLoopGroup.class);
+ final EventLoopGroup workerGroup = getField(server, "workerGroup",
EventLoopGroup.class);
+
+ // Reuse the bound port so a second bind would fail instead of
choosing another ephemeral port.
+ new MqttContext().setPort(((InetSocketAddress)
future.channel().localAddress()).getPort());
+ server.start();
+
+ assertSame(future, getField(server, "future", ChannelFuture.class));
+ assertSame(bossGroup, getField(server, "bossGroup",
EventLoopGroup.class));
+ assertSame(workerGroup, getField(server, "workerGroup",
EventLoopGroup.class));
+ assertTrue(future.channel().isActive());
+ }
+
+ @Test
+ public void failedBindShouldReleaseEventLoopsAndAllowRetry() throws
Exception {
+ final AtomicReference<EventLoopGroup> allocatedBoss = new
AtomicReference<>();
+ final AtomicReference<EventLoopGroup> allocatedWorker = new
AtomicReference<>();
+ server = new MqttBootstrapServer() {
+ @Override
+ public void shutdown() {
+ allocatedBoss.set(getField(this, "bossGroup",
EventLoopGroup.class));
+ allocatedWorker.set(getField(this, "workerGroup",
EventLoopGroup.class));
+ super.shutdown();
+ }
+ };
+ try (ServerSocket occupied = new ServerSocket(0)) {
+ new MqttContext().setPort(occupied.getLocalPort());
+
+ assertThrows(BindException.class, server::start);
+
+ assertTrue(allocatedBoss.get().isTerminated());
+ assertTrue(allocatedWorker.get().isTerminated());
+ assertNull(getField(server, "future", ChannelFuture.class));
+ assertNull(getField(server, "bossGroup", EventLoopGroup.class));
+ assertNull(getField(server, "workerGroup", EventLoopGroup.class));
+ }
+
+ new MqttContext().setPort(0);
+ server.start();
+ assertTrue(getField(server, "future",
ChannelFuture.class).channel().isActive());
+ }
+
+ @Test
+ public void concurrentStartsShouldKeepTheSameResources() throws Exception {
+ final CountDownLatch ready = new CountDownLatch(2);
+ final CountDownLatch start = new CountDownLatch(1);
+ Callable<ChannelFuture> task = () -> {
+ ready.countDown();
+ assertTrue(start.await(5, TimeUnit.SECONDS));
+ server.start();
+ return getField(server, "future", ChannelFuture.class);
+ };
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ try {
+ final Future<ChannelFuture> first = executor.submit(task);
+ final Future<ChannelFuture> second = executor.submit(task);
+ assertTrue(ready.await(5, TimeUnit.SECONDS));
+ start.countDown();
+
+ assertSame(first.get(10, TimeUnit.SECONDS), second.get(10,
TimeUnit.SECONDS));
+ assertTrue(server.isRunning());
+ final EventLoopGroup bossGroup = getField(server, "bossGroup",
EventLoopGroup.class);
+ final EventLoopGroup workerGroup = getField(server, "workerGroup",
EventLoopGroup.class);
+ server.shutdown();
+ assertTrue(bossGroup.isTerminated());
+ assertTrue(workerGroup.isTerminated());
+ } finally {
+ start.countDown();
+ executor.shutdownNow();
+ assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS));
+ }
+ }
+
+ @Test
+ public void startShouldReplaceClosedListenerAndReleaseOldEventLoops() {
+ server.start();
+ final ChannelFuture previous = getField(server, "future",
ChannelFuture.class);
+ final EventLoopGroup oldBoss = getField(server, "bossGroup",
EventLoopGroup.class);
+ final EventLoopGroup oldWorker = getField(server, "workerGroup",
EventLoopGroup.class);
+ new MqttContext().setPort(((InetSocketAddress)
previous.channel().localAddress()).getPort());
+ previous.channel().close().syncUninterruptibly();
+ assertFalse(server.isRunning());
+
+ server.start();
+
+ assertTrue(server.isRunning());
+ assertNotSame(previous, getField(server, "future",
ChannelFuture.class));
+ assertTrue(oldBoss.isTerminated());
+ assertTrue(oldWorker.isTerminated());
}
- private <T> T getField(final Object target, final String name, final
Class<T> type) throws Exception {
- Field field = MqttBootstrapServer.class.getDeclaredField(name);
- field.setAccessible(true);
- return type.cast(field.get(target));
+ private <T> T getField(final Object target, final String name, final
Class<T> type) {
+ try {
+ Field field = MqttBootstrapServer.class.getDeclaredField(name);
+ field.setAccessible(true);
+ return type.cast(field.get(target));
+ } catch (ReflectiveOperationException e) {
+ throw new AssertionError(e);
+ }
}
}
