From 9124e5be37744fcc1062ea23df747640d17258e1 Mon Sep 17 00:00:00 2001
From: xiwa <xw2sy@163.com>
Date: Sat, 24 Feb 2024 23:32:58 +0800
Subject: [PATCH] =?UTF-8?q?feat:mqtt=E6=8F=92=E4=BB=B6=E5=A2=9E=E5=8A=A0?=
 =?UTF-8?q?=E5=AD=90=E8=AE=BE=E5=A4=87=E6=B3=A8=E5=86=8C?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../plugins/http/service/HttpPlugin.java      |  2 +-
 .../plugins/mqtt/service/MqttPlugin.java      |  2 +-
 .../plugins/mqtt/service/MqttVerticle.java    | 83 ++++++++++---------
 pom.xml                                       |  2 +-
 4 files changed, 45 insertions(+), 44 deletions(-)

diff --git a/http-plugin/src/main/java/cc/iotkit/plugins/http/service/HttpPlugin.java b/http-plugin/src/main/java/cc/iotkit/plugins/http/service/HttpPlugin.java
index 824902e..99d2ede 100755
--- a/http-plugin/src/main/java/cc/iotkit/plugins/http/service/HttpPlugin.java
+++ b/http-plugin/src/main/java/cc/iotkit/plugins/http/service/HttpPlugin.java
@@ -73,7 +73,7 @@ public class HttpPlugin implements PluginCloseListener {
                 CountDownLatch wait = new CountDownLatch(1);
                 Future<Void> future = vertx.undeploy(deployedId);
                 future.onSuccess(unused -> {
-                    log.info("tcp plugin stopped success");
+                    log.info("http plugin stopped success");
                     wait.countDown();
                 });
                 future.onFailure(h -> {
diff --git a/mqtt-plugin/src/main/java/cc/iotkit/plugins/mqtt/service/MqttPlugin.java b/mqtt-plugin/src/main/java/cc/iotkit/plugins/mqtt/service/MqttPlugin.java
index c0d1bcf..306b4bf 100755
--- a/mqtt-plugin/src/main/java/cc/iotkit/plugins/mqtt/service/MqttPlugin.java
+++ b/mqtt-plugin/src/main/java/cc/iotkit/plugins/mqtt/service/MqttPlugin.java
@@ -72,7 +72,7 @@ public class MqttPlugin implements PluginCloseListener, IPlugin {
                 CountDownLatch wait = new CountDownLatch(1);
                 Future<Void> future = vertx.undeploy(deployedId);
                 future.onSuccess(unused -> {
-                    log.info("tcp plugin stopped success");
+                    log.info("mqtt plugin stopped success");
                     wait.countDown();
                 });
                 future.onFailure(h -> {
diff --git a/mqtt-plugin/src/main/java/cc/iotkit/plugins/mqtt/service/MqttVerticle.java b/mqtt-plugin/src/main/java/cc/iotkit/plugins/mqtt/service/MqttVerticle.java
index 24e3516..6feefc1 100755
--- a/mqtt-plugin/src/main/java/cc/iotkit/plugins/mqtt/service/MqttVerticle.java
+++ b/mqtt-plugin/src/main/java/cc/iotkit/plugins/mqtt/service/MqttVerticle.java
@@ -132,6 +132,7 @@ public class MqttVerticle extends AbstractVerticle implements Handler<MqttEndpoi
 
         String productKey = parts[0];
         String deviceName = parts[1];
+        String gwModel = parts[2];
         if (!auth.getUsername().equals(deviceName)) {
             log.error("username:{}不正确", deviceName);
             endpoint.reject(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USERNAME_OR_PASSWORD);
@@ -157,10 +158,11 @@ public class MqttVerticle extends AbstractVerticle implements Handler<MqttEndpoi
                 pluginInfo.getPluginId(),
                 fillAction(
                         DeviceRegister.builder()
-                                .model(parts[2])
+                                .productKey(productKey)
+                                .deviceName(deviceName)
+                                .model(gwModel)
                                 .version("1.0")
                                 .build()
-                        , productKey, deviceName
                 )
         );
         if (result.getCode() != 0) {
@@ -185,15 +187,7 @@ public class MqttVerticle extends AbstractVerticle implements Handler<MqttEndpoi
                 return;
             }
             //下线
-            thingService.post(
-                    pluginInfo.getPluginId(),
-                    fillAction(
-                            DeviceStateChange.builder()
-                                    .state(DeviceState.OFFLINE)
-                                    .build()
-                            , productKey, deviceName
-                    )
-            );
+            offline(productKey, deviceName);
             DEVICE_ONLINE.clear();
             //删除设备与连接关系
             endpointMap.remove(deviceName);
@@ -202,6 +196,8 @@ public class MqttVerticle extends AbstractVerticle implements Handler<MqttEndpoi
             if (!MQTT_CONNECT_POOL.get(clientId)) {
                 return;
             }
+            //下线
+            offline(productKey, deviceName);
             //删除设备与连接关系
             endpointMap.remove(deviceName);
             MQTT_CONNECT_POOL.put(clientId, false);
@@ -230,15 +226,7 @@ public class MqttVerticle extends AbstractVerticle implements Handler<MqttEndpoi
                 //删除设备对应连接
                 endpointMap.remove(device.getDeviceName());
                 //下线
-                thingService.post(
-                        pluginInfo.getPluginId(),
-                        fillAction(
-                                DeviceStateChange.builder()
-                                        .state(DeviceState.OFFLINE)
-                                        .build()
-                                , device.getProductKey(), device.getDeviceName()
-                        )
-                );
+                offline(device.getProductKey(), device.getDeviceName());
                 DEVICE_ONLINE.remove(device.getDeviceName());
             }
 
@@ -291,13 +279,20 @@ public class MqttVerticle extends AbstractVerticle implements Handler<MqttEndpoi
                     ActionResult regResult = thingService.post(
                             pluginInfo.getPluginId(),
                             fillAction(
-                                    DeviceRegister.builder()
-                                            .productKey(subPk)
-                                            .deviceName(subDn)
-                                            .model(params.getString("model"))
+                                    SubDeviceRegister.builder()
+                                            .productKey(productKey)
+                                            .deviceName(deviceName)
+                                            .model(gwModel)
                                             .version("1.0")
+                                            .subs(List.of(
+                                                    DeviceRegister.builder()
+                                                            .productKey(subPk)
+                                                            .deviceName(subDn)
+                                                            .model(params.getString("model"))
+                                                            .version("1.0")
+                                                            .build()
+                                            ))
                                             .build()
-                                    , subPk, subDn
                             )
                     );
                     if (regResult.getCode() == 0) {
@@ -357,9 +352,10 @@ public class MqttVerticle extends AbstractVerticle implements Handler<MqttEndpoi
         thingService.post(
                 pluginInfo.getPluginId(),
                 fillAction(DeviceStateChange.builder()
-                                .state(DeviceState.ONLINE)
-                                .build()
-                        , pk, dn
+                        .productKey(pk)
+                        .deviceName(dn)
+                        .state(DeviceState.ONLINE)
+                        .build()
                 )
         );
         DEVICE_ONLINE.put(dn, true);
@@ -385,10 +381,8 @@ public class MqttVerticle extends AbstractVerticle implements Handler<MqttEndpoi
         endpoint.publish(topic + "_reply", JsonObject.mapFrom(payloadReply).toBuffer(), MqttQoS.AT_LEAST_ONCE, false, false);
     }
 
-    private IDeviceAction fillAction(IDeviceAction action, String productKey, String deviceName) {
+    private IDeviceAction fillAction(IDeviceAction action) {
         action.setId(UniqueIdUtil.newRequestId());
-        action.setProductKey(productKey);
-        action.setDeviceName(deviceName);
         action.setTime(System.currentTimeMillis());
         return action;
     }
@@ -403,18 +397,25 @@ public class MqttVerticle extends AbstractVerticle implements Handler<MqttEndpoi
             }
 
             //下线
-            thingService.post(
-                    pluginInfo.getPluginId(),
-                    fillAction(
-                            DeviceStateChange.builder()
-                                    .state(DeviceState.OFFLINE)
-                                    .build(),
-                            parts[0],
-                            parts[1]
-                    )
-            );
+            offline(parts[0], parts[1]);
             DEVICE_ONLINE.clear();
         }
+        if(mqttServer!=null) {
+            mqttServer.close();
+        }
+    }
+
+    private void offline(String productKey, String deviceName) {
+        thingService.post(
+                pluginInfo.getPluginId(),
+                fillAction(
+                        DeviceStateChange.builder()
+                                .productKey(productKey)
+                                .deviceName(deviceName)
+                                .state(DeviceState.OFFLINE)
+                                .build()
+                )
+        );
     }
 
     public void publish(String deviceName, String topic, String msg) {
diff --git a/pom.xml b/pom.xml
index 8c106b9..c3922d4 100755
--- a/pom.xml
+++ b/pom.xml
@@ -37,7 +37,7 @@
         <dependency>
             <groupId>cc.iotkit</groupId>
             <artifactId>iot-plugin-core</artifactId>
-            <version>1.0.2</version>
+            <version>1.0.3</version>
         </dependency>
 
         <dependency>