This is an automated email from the ASF dual-hosted git repository.
xiaoyu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new 99dfb84 [ISSUE #2071] add mqtt plugin (#2753)
99dfb84 is described below
commit 99dfb8460923d09158f125ec7cf64d44b2807fc9
Author: Sinsy <[email protected]>
AuthorDate: Thu Jan 13 12:10:58 2022 +0800
[ISSUE #2071] add mqtt plugin (#2753)
* add mqtt plugin
* fix ci error
* fix error
* fix netty unsafe at jdk 11
* ignore ut
* change pom
* add sql
---
.../src/main/resources/sql-script/h2/schema.sql | 12 +++
.../src/main/resources/sql-script/mysql/schema.sql | 12 +++
.../src/main/resources/sql-script/pg/schema.sql | 19 ++++-
.../org/apache/shenyu/common/enums/PluginEnum.java | 5 ++
shenyu-plugin/pom.xml | 1 +
shenyu-plugin/shenyu-plugin-mqtt/pom.xml | 45 ++++++++++++
.../plugin/mqtt/handler/MqttPluginDataHandler.java | 52 +++++++++++++
.../mqtt/handler/MqttPluginDataHandlerTest.java | 85 ++++++++++++++++++++++
.../org/apache/shenyu/protocol/mqtt/Connect.java | 2 +-
.../shenyu/protocol/mqtt/MqttBootstrapServer.java | 9 +--
.../mqtt/{MqttEnv.java => MqttContext.java} | 23 +++---
.../protocol/mqtt/MqttServerConfiguration.java | 14 ++--
.../shenyu-spring-boot-starter-plugin/pom.xml | 1 +
.../shenyu-spring-boot-starter-plugin-mqtt/pom.xml | 48 ++++++++++++
.../plugin/mqtt/MqttPluginConfiguration.java | 40 ++++++++++
.../src/main/resources/META-INF/spring.factories | 19 +++++
.../plugin/mqtt/MqttPluginConfigurationTest.java | 43 +++++++++++
17 files changed, 399 insertions(+), 31 deletions(-)
diff --git a/shenyu-admin/src/main/resources/sql-script/h2/schema.sql
b/shenyu-admin/src/main/resources/sql-script/h2/schema.sql
index 78b204e..eb6e421 100644
--- a/shenyu-admin/src/main/resources/sql-script/h2/schema.sql
+++ b/shenyu-admin/src/main/resources/sql-script/h2/schema.sql
@@ -353,6 +353,7 @@ INSERT IGNORE INTO `plugin` (`id`, `name`, `role`, `sort`,
`enabled`, `date_crea
INSERT IGNORE INTO `plugin` (`id`, `name`, `role`, `sort`, `enabled`,
`date_created`, `date_updated`) VALUES ('25', 'cryptorResponse', 'Cryptor',
410, '1', '2021-08-06 13:55:30', '2021-08-13 16:03:40');;
INSERT IGNORE INTO `plugin` (`id`, `name`, `role`, `sort`, `config`,
`enabled`, `date_created`, `date_updated`) VALUES ('26', 'websocket', 'Proxy',
200, '{"multiSelectorHandle":"1"}', '1', '2021-08-27 13:55:30', '2021-08-27
16:03:40');;
INSERT IGNORE INTO `plugin` (`id`, `name`, `role`, `sort`, `enabled`,
`date_created`, `date_updated`) VALUES ('27', 'generalContext', 'Common', 125,
'1', '2021-11-24 21:38:48', '2021-11-24 11:47:01');;
+INSERT IGNORE INTO `plugin` (`id`, `name`, `role`, `sort`, `config`,
`enabled`, `date_created`, `date_updated`) VALUES ('28', 'mqtt', 'Proxy', 125,
'{"port": 9500,"bossGroupThreadCount": 1,"maxPayloadSize":
65536,"workerGroupThreadCount": 12,"userName": "shenyu","password":
"shenyu","isEncryptPassword": false,"encryptMode": "","leakDetectorLevel":
"DISABLED"}', '0', '2022-01-12 20:33:50', '2022-01-12 20:34:07');;
/*insert plugin_handle data for sentinel*/
INSERT IGNORE INTO plugin_handle
(`plugin_id`,`field`,`label`,`data_type`,`type`,`sort`,`ext_obj`) VALUES ('10'
,'flowRuleGrade','flowRuleGrade','3', 2, 8,
'{"required":"1","defaultValue":"1","rule":""}');;
@@ -512,6 +513,17 @@ INSERT IGNORE INTO plugin_handle
(`plugin_id`,`field`,`label`,`data_type`,`type`
/*insert plugin_handle data for plugin motan*/
INSERT IGNORE INTO plugin_handle
(`plugin_id`,`field`,`label`,`data_type`,`type`,`sort`,`ext_obj`) VALUES ('17',
'register', 'register', 2, 3, 0, null);;
+/*insert plugin_handle data for plugin mqtt*/
+INSERT IGNORE INTO plugin_handle
(`plugin_id`,`field`,`label`,`data_type`,`type`,`sort`,`ext_obj`) VALUES ('28',
'port', 'port', 1, 3, 1, NULL);;
+INSERT IGNORE INTO plugin_handle
(`plugin_id`,`field`,`label`,`data_type`,`type`,`sort`,`ext_obj`) VALUES ('28',
'bossGroupThreadCount', 'bossGroupThreadCount', 1, 3, 1, NULL);;
+INSERT IGNORE INTO plugin_handle
(`plugin_id`,`field`,`label`,`data_type`,`type`,`sort`,`ext_obj`) VALUES ('28',
'maxPayloadSize', 'maxPayloadSize', 1, 3, 1, NULL);;
+INSERT IGNORE INTO plugin_handle
(`plugin_id`,`field`,`label`,`data_type`,`type`,`sort`,`ext_obj`) VALUES ('28',
'workerGroupThreadCount', 'workerGroupThreadCount', 1, 3, 1, NULL);;
+INSERT IGNORE INTO plugin_handle
(`plugin_id`,`field`,`label`,`data_type`,`type`,`sort`,`ext_obj`) VALUES ('28',
'userName', 'userName', 2, 3, 1, NULL);;
+INSERT IGNORE INTO plugin_handle
(`plugin_id`,`field`,`label`,`data_type`,`type`,`sort`,`ext_obj`) VALUES ('28',
'password', 'password', 2, 3, 1, NULL);;
+INSERT IGNORE INTO plugin_handle
(`plugin_id`,`field`,`label`,`data_type`,`type`,`sort`,`ext_obj`) VALUES ('28',
'isEncryptPassword', 'isEncryptPassword', 2, 3, 1, NULL);;
+INSERT IGNORE INTO plugin_handle
(`plugin_id`,`field`,`label`,`data_type`,`type`,`sort`,`ext_obj`) VALUES ('28',
'encryptMode', 'encryptMode', 2, 3, 1, NULL);;
+INSERT IGNORE INTO plugin_handle
(`plugin_id`,`field`,`label`,`data_type`,`type`,`sort`,`ext_obj`) VALUES ('28',
'leakDetectorLevel', 'leakDetectorLevel', 2, 3, 1, NULL);;
+
/** insert resource for resource */
INSERT IGNORE INTO `resource` (`id`, `parent_id`, `title`, `name`, `url`,
`component`, `resource_type`, `sort`, `icon`, `is_leaf`, `is_route`, `perms`,
`status`, `date_created`, `date_updated`)
VALUES('1346775491550474240','','SHENYU.MENU.PLUGIN.LIST','plug','/plug','PluginList','0','0','dashboard','0','0','','1','2021-01-06
05:07:54','2021-01-07 18:34:11');;
diff --git a/shenyu-admin/src/main/resources/sql-script/mysql/schema.sql
b/shenyu-admin/src/main/resources/sql-script/mysql/schema.sql
index 2ee0796..647bfbe 100644
--- a/shenyu-admin/src/main/resources/sql-script/mysql/schema.sql
+++ b/shenyu-admin/src/main/resources/sql-script/mysql/schema.sql
@@ -366,6 +366,7 @@ INSERT IGNORE INTO `plugin` (`id`, `name`, `role`, `sort`,
`enabled`, `date_crea
INSERT IGNORE INTO `plugin` (`id`, `name`, `role`, `sort`, `enabled`,
`date_created`, `date_updated`) VALUES ('25', 'cryptorResponse', 'Cryptor',
410, '1', '2021-08-06 13:55:30', '2021-08-13 16:03:40');;
INSERT IGNORE INTO `plugin` (`id`, `name`, `role`, `sort`, `config`,
`enabled`, `date_created`, `date_updated`) VALUES ('26', 'websocket', 'Proxy',
200, '{"multiSelectorHandle":"1"}', '1', '2021-08-27 13:55:30', '2021-08-27
16:03:40');;
INSERT IGNORE INTO `plugin` (`id`, `name`, `role`, `sort`, `enabled`,
`date_created`, `date_updated`) VALUES ('27', 'generalContext', 'Common', 125,
'1', '2021-11-24 21:38:48', '2021-11-24 11:47:01');;
+INSERT IGNORE INTO `plugin` (`id`, `name`, `role`, `sort`, `config`,
`enabled`, `date_created`, `date_updated`) VALUES ('28', 'mqtt', 'Proxy', 125,
'{"port": 9500,"bossGroupThreadCount": 1,"maxPayloadSize":
65536,"workerGroupThreadCount": 12,"userName": "shenyu","password":
"shenyu","isEncryptPassword": false,"encryptMode": "","leakDetectorLevel":
"DISABLED"}', '0', '2022-01-12 20:33:50', '2022-01-12 20:34:07');;
/*insert plugin_handle data for sentinel*/
INSERT IGNORE INTO plugin_handle
(`plugin_id`,`field`,`label`,`data_type`,`type`,`sort`,`ext_obj`) VALUES ('10'
,'flowRuleGrade','flowRuleGrade','3', 2, 8,
'{"required":"1","defaultValue":"1","rule":""}');;
@@ -525,6 +526,17 @@ INSERT IGNORE INTO plugin_handle
(`plugin_id`,`field`,`label`,`data_type`,`type`
/*insert plugin_handle data for plugin motan*/
INSERT IGNORE INTO plugin_handle
(`plugin_id`,`field`,`label`,`data_type`,`type`,`sort`,`ext_obj`) VALUES ('17',
'register', 'register', 2, 3, 0, null);;
+/*insert plugin_handle data for plugin mqtt*/
+INSERT IGNORE INTO plugin_handle
(`plugin_id`,`field`,`label`,`data_type`,`type`,`sort`,`ext_obj`) VALUES ('28',
'port', 'port', 1, 3, 1, NULL);;
+INSERT IGNORE INTO plugin_handle
(`plugin_id`,`field`,`label`,`data_type`,`type`,`sort`,`ext_obj`) VALUES ('28',
'bossGroupThreadCount', 'bossGroupThreadCount', 1, 3, 1, NULL);;
+INSERT IGNORE INTO plugin_handle
(`plugin_id`,`field`,`label`,`data_type`,`type`,`sort`,`ext_obj`) VALUES ('28',
'maxPayloadSize', 'maxPayloadSize', 1, 3, 1, NULL);;
+INSERT IGNORE INTO plugin_handle
(`plugin_id`,`field`,`label`,`data_type`,`type`,`sort`,`ext_obj`) VALUES ('28',
'workerGroupThreadCount', 'workerGroupThreadCount', 1, 3, 1, NULL);;
+INSERT IGNORE INTO plugin_handle
(`plugin_id`,`field`,`label`,`data_type`,`type`,`sort`,`ext_obj`) VALUES ('28',
'userName', 'userName', 2, 3, 1, NULL);;
+INSERT IGNORE INTO plugin_handle
(`plugin_id`,`field`,`label`,`data_type`,`type`,`sort`,`ext_obj`) VALUES ('28',
'password', 'password', 2, 3, 1, NULL);;
+INSERT IGNORE INTO plugin_handle
(`plugin_id`,`field`,`label`,`data_type`,`type`,`sort`,`ext_obj`) VALUES ('28',
'isEncryptPassword', 'isEncryptPassword', 2, 3, 1, NULL);;
+INSERT IGNORE INTO plugin_handle
(`plugin_id`,`field`,`label`,`data_type`,`type`,`sort`,`ext_obj`) VALUES ('28',
'encryptMode', 'encryptMode', 2, 3, 1, NULL);;
+INSERT IGNORE INTO plugin_handle
(`plugin_id`,`field`,`label`,`data_type`,`type`,`sort`,`ext_obj`) VALUES ('28',
'leakDetectorLevel', 'leakDetectorLevel', 2, 3, 1, NULL);;
+
/** insert resource for resource */
INSERT IGNORE INTO `resource` (`id`, `parent_id`, `title`, `name`, `url`,
`component`, `resource_type`, `sort`, `icon`, `is_leaf`, `is_route`, `perms`,
`status`, `date_created`, `date_updated`)
VALUES('1346775491550474240','','SHENYU.MENU.PLUGIN.LIST','plug','/plug','PluginList','0','0','dashboard','0','0','','1','2021-01-06
05:07:54','2021-01-07 18:34:11');;
diff --git a/shenyu-admin/src/main/resources/sql-script/pg/schema.sql
b/shenyu-admin/src/main/resources/sql-script/pg/schema.sql
index 986918a..9aa15cc 100644
--- a/shenyu-admin/src/main/resources/sql-script/pg/schema.sql
+++ b/shenyu-admin/src/main/resources/sql-script/pg/schema.sql
@@ -394,7 +394,11 @@ ELSE
PERFORM public.dblink_exec('init_conn', 'INSERT INTO "plugin" VALUES
(''' || '8' || ''', ''' || 'springCloud' || ''', NULL, ''' || 'Proxy' || ''',
200, 0, ''' || '2018-06-25 13:47:57' || ''', ''' || '2018-06-25 13:47:57' ||
''');');
PERFORM public.dblink_exec('init_conn', 'INSERT INTO "plugin" VALUES
(''' || '9' || ''', ''' || 'hystrix' || ''', NULL, ''' || 'FaultTolerance' ||
''', 130, 0, ''' || '2020-01-15 10:19:10' || ''', ''' || '2020-01-15 10:19:10'
|| ''');');
PERFORM public.dblink_exec('init_conn', 'INSERT INTO "plugin" VALUES
(''' || '27' || ''', ''' || 'generalContext' || ''', NULL, ''' || 'Common' ||
''', 125, 0, ''' || '2021-11-24 21:38:48' || ''', ''' || '2021-11-24 21:38:48'
|| ''');');
- PERFORM public.dblink_exec('init_conn', 'COMMIT');
+ PERFORM public.dblink_exec('init_conn', 'INSERT INTO "plugin" VALUES ('''
|| '28' || ''', ''' || 'mqtt' || ''', ''' || 'Proxy' || ''', 125, 0, ''' ||
'2022-01-12 20:33:50' || ''', ''' || '2022-01-12 20:34:07' || ''');');
+ PERFORM public.dblink_exec('init_conn', 'INSERT INTO "plugin" VALUES ('''
|| '28' || ''', ''' || 'mqtt' || ''', ''' || '{"port":
9500,"bossGroupThreadCount": 1,"maxPayloadSize":
65536,"workerGroupThreadCount": 12,"userName": "shenyu","password":
"shenyu","isEncryptPassword": false,"encryptMode": "","leakDetectorLevel":
"DISABLED"}' || ''', ''' || 'Proxy' || ''', 310, 0, ''' || '2022-01-12
20:33:50' || ''', ''' || '2022-01-12 20:33:50' || ''');');
+
+
+PERFORM public.dblink_exec('init_conn', 'COMMIT');
END IF;
-- ----------------------------------------------------
@@ -637,7 +641,18 @@ ELSE
/*insert "plugin_handle" data for plugin motan*/
PERFORM public.dblink_exec('init_conn', 'INSERT INTO "plugin_handle" (
plugin_id , field , label , data_type , type , sort , ext_obj ) VALUES (''' ||
'17' || ''', ''' || 'register' || ''', ''' || 'register' || ''', 2, 3, 0,
null);');
- PERFORM public.dblink_exec('init_conn', 'COMMIT');
+ /*insert plugin_handle data for plugin mqtt*/
+ PERFORM public.dblink_exec('init_conn', 'INSERT INTO "plugin_handle" (
plugin_id , field , label , data_type , type , sort , ext_obj ) VALUES (''' ||
'28' || ''', ''' || 'port' || ''', ''' || 'port' || ''', 1, 3, 1, null);');
+ PERFORM public.dblink_exec('init_conn', 'INSERT INTO "plugin_handle" (
plugin_id , field , label , data_type , type , sort , ext_obj ) VALUES (''' ||
'28' || ''', ''' || 'bossGroupThreadCount' || ''', ''' ||
'bossGroupThreadCount' || ''', 1, 3, 1, null);');
+ PERFORM public.dblink_exec('init_conn', 'INSERT INTO "plugin_handle" (
plugin_id , field , label , data_type , type , sort , ext_obj ) VALUES (''' ||
'28' || ''', ''' || 'maxPayloadSize' || ''', ''' || 'maxPayloadSize' || ''', 1,
3, 1, null);');
+ PERFORM public.dblink_exec('init_conn', 'INSERT INTO "plugin_handle" (
plugin_id , field , label , data_type , type , sort , ext_obj ) VALUES (''' ||
'28' || ''', ''' || 'workerGroupThreadCount' || ''', ''' ||
'workerGroupThreadCount' || ''', 1, 3, 1, null);');
+ PERFORM public.dblink_exec('init_conn', 'INSERT INTO "plugin_handle" (
plugin_id , field , label , data_type , type , sort , ext_obj ) VALUES (''' ||
'28' || ''', ''' || 'userName' || ''', ''' || 'userName' || ''', 2, 3, 1,
null);');
+ PERFORM public.dblink_exec('init_conn', 'INSERT INTO "plugin_handle" (
plugin_id , field , label , data_type , type , sort , ext_obj ) VALUES (''' ||
'28' || ''', ''' || 'password' || ''', ''' || 'password' || ''', 2, 3, 1,
null);');
+ PERFORM public.dblink_exec('init_conn', 'INSERT INTO "plugin_handle" (
plugin_id , field , label , data_type , type , sort , ext_obj ) VALUES (''' ||
'28' || ''', ''' || 'isEncryptPassword' || ''', ''' || 'isEncryptPassword' ||
''', 2, 3, 1, null);');
+ PERFORM public.dblink_exec('init_conn', 'INSERT INTO "plugin_handle" (
plugin_id , field , label , data_type , type , sort , ext_obj ) VALUES (''' ||
'28' || ''', ''' || 'encryptMode' || ''', ''' || 'encryptMode' || ''', 2, 3, 1,
null);');
+ PERFORM public.dblink_exec('init_conn', 'INSERT INTO "plugin_handle" (
plugin_id , field , label , data_type , type , sort , ext_obj ) VALUES (''' ||
'28' || ''', ''' || 'leakDetectorLevel' || ''', ''' || 'leakDetectorLevel' ||
''', 2, 3, 1, null);');
+
+PERFORM public.dblink_exec('init_conn', 'COMMIT');
END IF;
diff --git
a/shenyu-common/src/main/java/org/apache/shenyu/common/enums/PluginEnum.java
b/shenyu-common/src/main/java/org/apache/shenyu/common/enums/PluginEnum.java
index ce25d6c..c9aed89 100644
--- a/shenyu-common/src/main/java/org/apache/shenyu/common/enums/PluginEnum.java
+++ b/shenyu-common/src/main/java/org/apache/shenyu/common/enums/PluginEnum.java
@@ -24,6 +24,11 @@ import java.util.List;
* PluginEnum.
*/
public enum PluginEnum {
+
+ /**
+ * Mqtt plugin enum.
+ */
+ MQTT(0, 0, "mqtt"),
/**
* Global plugin enum.
diff --git a/shenyu-plugin/pom.xml b/shenyu-plugin/pom.xml
index bbeb1d2..02fd94f 100644
--- a/shenyu-plugin/pom.xml
+++ b/shenyu-plugin/pom.xml
@@ -59,5 +59,6 @@
<module>shenyu-plugin-dubbo</module>
<module>shenyu-plugin-uri</module>
<module>shenyu-plugin-general-context</module>
+ <module>shenyu-plugin-mqtt</module>
</modules>
</project>
diff --git a/shenyu-plugin/shenyu-plugin-mqtt/pom.xml
b/shenyu-plugin/shenyu-plugin-mqtt/pom.xml
new file mode 100644
index 0000000..2cc020a
--- /dev/null
+++ b/shenyu-plugin/shenyu-plugin-mqtt/pom.xml
@@ -0,0 +1,45 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ 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.
+ -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0";
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <parent>
+ <artifactId>shenyu-plugin</artifactId>
+ <groupId>org.apache.shenyu</groupId>
+ <version>2.4.2-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>shenyu-plugin-mqtt</artifactId>
+
+ <dependencies>
+ <dependency>
+ <artifactId>shenyu-protocol-mqtt</artifactId>
+ <groupId>org.apache.shenyu</groupId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.shenyu</groupId>
+ <artifactId>shenyu-plugin-base</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ </dependencies>
+
+</project>
\ No newline at end of file
diff --git
a/shenyu-plugin/shenyu-plugin-mqtt/src/main/java/org/apache/shenyu/plugin/mqtt/handler/MqttPluginDataHandler.java
b/shenyu-plugin/shenyu-plugin-mqtt/src/main/java/org/apache/shenyu/plugin/mqtt/handler/MqttPluginDataHandler.java
new file mode 100644
index 0000000..4c1a2ca
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-mqtt/src/main/java/org/apache/shenyu/plugin/mqtt/handler/MqttPluginDataHandler.java
@@ -0,0 +1,52 @@
+/*
+ * 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.plugin.mqtt.handler;
+
+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;
+
+/**
+ * The type Mqtt plugin data handler.
+ */
+public class MqttPluginDataHandler implements PluginDataHandler {
+
+ private final BootstrapServer server = new MqttBootstrapServer();
+
+ @Override
+ public void handlerPlugin(final PluginData pluginData) {
+ if (pluginData.getEnabled()) {
+ MqttServerConfiguration configuration =
GsonUtils.getInstance().fromJson(pluginData.getConfig(),
MqttServerConfiguration.class);
+ configuration.afterPropertiesSet();
+ server.init();
+ server.start();
+ } else {
+ server.shutdown();
+ }
+ }
+
+ @Override
+ public String pluginNamed() {
+ return PluginEnum.MQTT.getName();
+ }
+
+}
diff --git
a/shenyu-plugin/shenyu-plugin-mqtt/src/test/java/org/apache/shenyu/plugin/mqtt/handler/MqttPluginDataHandlerTest.java
b/shenyu-plugin/shenyu-plugin-mqtt/src/test/java/org/apache/shenyu/plugin/mqtt/handler/MqttPluginDataHandlerTest.java
new file mode 100644
index 0000000..279185d
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-mqtt/src/test/java/org/apache/shenyu/plugin/mqtt/handler/MqttPluginDataHandlerTest.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.plugin.mqtt.handler;
+
+import org.apache.shenyu.common.dto.PluginData;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import java.net.InetAddress;
+import java.net.Socket;
+
+/**
+ * Test case for {@link MqttPluginDataHandler}.
+ */
+public class MqttPluginDataHandlerTest {
+
+ private MqttPluginDataHandler mqttPluginDataHandlerUnderTest;
+
+ @Before
+ public void setUp() {
+ mqttPluginDataHandlerUnderTest = new MqttPluginDataHandler();
+ }
+
+ @Test
+ @Ignore
+ public void testEnableConfiguration() {
+ 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);
+ mqttPluginDataHandlerUnderTest.handlerPlugin(enablePluginData);
+ Assert.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);
+ mqttPluginDataHandlerUnderTest.handlerPlugin(disablePluginData);
+ Assert.assertFalse(isPortUsing());
+ }
+
+ private boolean isPortUsing() {
+ boolean flag = false;
+ try {
+ InetAddress address = InetAddress.getByName("127.0.0.1");
+ Socket socket = new Socket(address, 9500);
+ flag = true;
+ } catch (Exception ignored) {
+
+ }
+ return flag;
+
+ }
+}
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 4613c11..3f5174e 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
@@ -43,7 +43,7 @@ public class Connect extends MessageType {
String userName = msg.payload().userName();
byte[] passwordInBytes = msg.payload().passwordInBytes();
- if (!MqttEnv.isValid(userName, passwordInBytes)) {
+ if (!MqttContext.isValid(userName, passwordInBytes)) {
ctx.writeAndFlush(wrong(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD));
return;
}
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 8b0808b..fa7fe7c 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
@@ -36,7 +36,7 @@ public class MqttBootstrapServer implements BootstrapServer {
private static final String REPOSITORY_PACKAGE_NAME =
"org.apache.shenyu.protocol.mqtt.repositories";
- private static final MqttEnv ENV = new MqttEnv();
+ private static final MqttContext ENV = new MqttContext();
private EventLoopGroup bossGroup;
@@ -73,14 +73,9 @@ public class MqttBootstrapServer implements BootstrapServer {
@Override
public void shutdown() {
- try {
- future.channel().closeFuture().sync();
- } catch (InterruptedException e) {
- //// todo log
- }
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
-
+ future.channel().close();
}
private void initRepositories() throws IllegalAccessException,
InstantiationException {
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/MqttEnv.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/MqttContext.java
similarity index 86%
rename from
shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/MqttEnv.java
rename to
shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/MqttContext.java
index 5a0b775..74e6b0d 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/MqttEnv.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/MqttContext.java
@@ -22,7 +22,7 @@ import org.apache.commons.lang3.StringUtils;
/**
* mqtt env.
*/
-public class MqttEnv {
+public class MqttContext {
private static int port;
@@ -38,11 +38,6 @@ public class MqttEnv {
private static String leakDetectorLevel;
- static {
- MqttServerConfiguration configuration = new MqttServerConfiguration();
- configuration.afterPropertiesSet();
- }
-
/**
* Whether userName and password are correct.
* @param userName userName
@@ -56,7 +51,7 @@ public class MqttEnv {
return false;
}
- return MqttEnv.userName.equals(userName) &&
MqttEnv.password.equals(password);
+ return MqttContext.userName.equals(userName) &&
MqttContext.password.equals(password);
}
/**
@@ -120,7 +115,7 @@ public class MqttEnv {
* @param port port
*/
public void setPort(final int port) {
- MqttEnv.port = port;
+ MqttContext.port = port;
}
/**
@@ -128,7 +123,7 @@ public class MqttEnv {
* @param bossGroupThreadCount bossGroupThreadCount
*/
public void setBossGroupThreadCount(final int bossGroupThreadCount) {
- MqttEnv.bossGroupThreadCount = bossGroupThreadCount;
+ MqttContext.bossGroupThreadCount = bossGroupThreadCount;
}
/**
@@ -136,7 +131,7 @@ public class MqttEnv {
* @param maxPayloadSize maxPayloadSize
*/
public void setMaxPayloadSize(final int maxPayloadSize) {
- MqttEnv.maxPayloadSize = maxPayloadSize;
+ MqttContext.maxPayloadSize = maxPayloadSize;
}
/**
@@ -144,7 +139,7 @@ public class MqttEnv {
* @param workerGroupThreadCount workerGroupThreadCount
*/
public void setWorkerGroupThreadCount(final int workerGroupThreadCount) {
- MqttEnv.workerGroupThreadCount = workerGroupThreadCount;
+ MqttContext.workerGroupThreadCount = workerGroupThreadCount;
}
/**
@@ -152,7 +147,7 @@ public class MqttEnv {
* @param userName userName
*/
public void setUserName(final String userName) {
- MqttEnv.userName = userName;
+ MqttContext.userName = userName;
}
/**
@@ -160,7 +155,7 @@ public class MqttEnv {
* @param password password.
*/
public void setPassword(final String password) {
- MqttEnv.password = password;
+ MqttContext.password = password;
}
/**
@@ -168,6 +163,6 @@ public class MqttEnv {
* @param leakDetectorLevel leakDetectorLevel
*/
public void setLeakDetectorLevel(final String leakDetectorLevel) {
- MqttEnv.leakDetectorLevel = leakDetectorLevel;
+ MqttContext.leakDetectorLevel = leakDetectorLevel;
}
}
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 392fbd0..5b210df 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
@@ -50,13 +50,13 @@ public class MqttServerConfiguration {
if (isEncryptPassword()) {
setPassword(encryptPassword());
}
- MqttEnv env = new MqttEnv();
- env.setPassword(getPassword());
- env.setPort(getPort());
- env.setMaxPayloadSize(getMaxPayloadSize());
- env.setUserName(getUserName());
- env.setWorkerGroupThreadCount(getWorkerGroupThreadCount());
- env.setLeakDetectorLevel(getLeakDetectorLevel());
+ MqttContext context = new MqttContext();
+ context.setPassword(getPassword());
+ context.setPort(getPort());
+ context.setMaxPayloadSize(getMaxPayloadSize());
+ context.setUserName(getUserName());
+ context.setWorkerGroupThreadCount(getWorkerGroupThreadCount());
+ context.setLeakDetectorLevel(getLeakDetectorLevel());
}
private String encryptPassword() {
diff --git
a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/pom.xml
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/pom.xml
index d3bd5a8..911ad61 100644
--- a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/pom.xml
+++ b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/pom.xml
@@ -57,5 +57,6 @@
<module>shenyu-spring-boot-starter-plugin-dubbo</module>
<module>shenyu-spring-boot-starter-plugin-uri</module>
<module>shenyu-spring-boot-starter-plugin-general-context</module>
+ <module>shenyu-spring-boot-starter-plugin-mqtt</module>
</modules>
</project>
diff --git
a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-mqtt/pom.xml
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-mqtt/pom.xml
new file mode 100644
index 0000000..03ff658
--- /dev/null
+++
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-mqtt/pom.xml
@@ -0,0 +1,48 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ 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.
+ -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0";
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <parent>
+ <artifactId>shenyu-spring-boot-starter-plugin</artifactId>
+ <groupId>org.apache.shenyu</groupId>
+ <version>2.4.2-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>shenyu-spring-boot-starter-plugin-mqtt</artifactId>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.shenyu</groupId>
+ <artifactId>shenyu-plugin-mqtt</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-test</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+</project>
\ No newline at end of file
diff --git
a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-mqtt/src/main/java/org/apache/shenyu/springboot/starter/plugin/mqtt/MqttPluginConfiguration.java
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-mqtt/src/main/java/org/apache/shenyu/springboot/starter/plugin/mqtt/MqttPluginConfiguration.java
new file mode 100644
index 0000000..bf0fb33
--- /dev/null
+++
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-mqtt/src/main/java/org/apache/shenyu/springboot/starter/plugin/mqtt/MqttPluginConfiguration.java
@@ -0,0 +1,40 @@
+/*
+ * 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.springboot.starter.plugin.mqtt;
+
+import org.apache.shenyu.plugin.base.handler.PluginDataHandler;
+import org.apache.shenyu.plugin.mqtt.handler.MqttPluginDataHandler;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * The type mqtt plugin configuration.
+ */
+@Configuration
+public class MqttPluginConfiguration {
+
+ /**
+ * Mqtt plugin data handler plugin data handler.
+ *
+ * @return the plugin data handler
+ */
+ @Bean
+ public PluginDataHandler mqttPluginDataHandler() {
+ return new MqttPluginDataHandler();
+ }
+}
diff --git
a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-mqtt/src/main/resources/META-INF/spring.factories
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-mqtt/src/main/resources/META-INF/spring.factories
new file mode 100644
index 0000000..e8c3127
--- /dev/null
+++
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-mqtt/src/main/resources/META-INF/spring.factories
@@ -0,0 +1,19 @@
+#
+# 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.
+#
+
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
+org.apache.shenyu.springboot.starter.plugin.mqtt.MqttPluginConfiguration
diff --git
a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-mqtt/src/test/java/org/apache/shenyu/springboot/starter/plugin/mqtt/MqttPluginConfigurationTest.java
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-mqtt/src/test/java/org/apache/shenyu/springboot/starter/plugin/mqtt/MqttPluginConfigurationTest.java
new file mode 100644
index 0000000..1d0cb1f
--- /dev/null
+++
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-mqtt/src/test/java/org/apache/shenyu/springboot/starter/plugin/mqtt/MqttPluginConfigurationTest.java
@@ -0,0 +1,43 @@
+/*
+ * 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.springboot.starter.plugin.mqtt;
+
+import org.apache.shenyu.plugin.base.handler.PluginDataHandler;
+import org.junit.Test;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+
+import static org.junit.Assert.assertNotNull;
+
+/**
+ * Test case for {@link MqttPluginConfiguration}.
+ */
+public class MqttPluginConfigurationTest {
+
+ @Test
+ public void testJwtPluginDataHandler() {
+ new ApplicationContextRunner()
+
.withConfiguration(AutoConfigurations.of(MqttPluginConfiguration.class))
+ .withBean(MqttPluginConfigurationTest.class)
+ .withPropertyValues("debug=true")
+ .run(context -> {
+ PluginDataHandler handler =
context.getBean("mqttPluginDataHandler", PluginDataHandler.class);
+ assertNotNull(handler);
+ });
+ }
+}
