review：【IoT 物联网】mqtt broker 协议

YunaiV
1 parent 74db72c9
Showing 5 changed files with 17 additions and 5 deletions
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttDownstreamSubscriber.java
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttHttpAuthProtocol.java
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttUpstreamProtocol.http
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttUpstreamProtocol.java
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttEventHandler.java
@@ -48,15 +48,15 @@ public class IotMqttDownstreamSubscriber implements IotMessageSubscriber&lt;IotDevi
     @Override
     public void onMessage(IotDeviceMessage message) {
         log.info("[onMessage][接收到下行消息：{}]", message);
-
         try {
             // 根据消息方法处理不同的下行消息
             String method = message.getMethod();
             if (method == null) {
-                log.warn("[onMessage][消息方法为空]");
+                log.warn("[onMessage][消息({})方法为空]", message);
                 return;
             }
+            // TODO @haohao：看看怎么融合下
             if (method.startsWith("thing.service.property.")) {
                 handlePropertyMessage(message);
             } else if (method.startsWith("thing.service.") && !method.contains("property") && !method.contains("config")
@@ -83,6 +83,7 @@ public class IotMqttDownstreamSubscriber implements IotMessageSubscriber&lt;IotDevi
         String method = message.getMethod();
         // 通过 deviceId 获取设备信息
+        // TODO @haohao：通过缓存拿；
         IotDeviceGetReqDTO reqDTO = new IotDeviceGetReqDTO();
         reqDTO.setId(message.getDeviceId());
         IotDeviceRespDTO deviceInfo = deviceApi.getDevice(reqDTO).getData();
@@ -12,6 +12,7 @@ import jakarta.annotation.PreDestroy;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+// TODO @haohao：可以融合到 IotMqttUpstreamProtocol 么？主要考虑，一个协议，一个 protocol；
 /**
  * IoT 网关 MQTT HTTP 认证协议：提供 HTTP 认证接口供 EMQX 调用
  *
@@ -82,6 +83,7 @@ public class IotMqttHttpAuthProtocol extends AbstractVerticle {
         }
     }
+    // TODO @haohao：这里不搞默认，必须填写哈；
     /**
      * 获取认证服务端口
      * 从配置中获取，默认使用 8090 端口
+### TODO 芋艿：这个文件是不是要留着；
 ### IoT MQTT HTTP 认证服务测试文件
 ### 用于测试独立的 MQTT HTTP 认证服务接口
 ### 服务由 IotMqttHttpAuthProtocol 启动，运行在配置的端口上（默认 8090）
@@ -101,7 +102,7 @@ Content-Type: application/json
 ### 假设配置了端口 8091：
 ### POST http://localhost:8091/mqtt/auth/authenticate
 ### Content-Type: application/json
-### 
+###
 ### {
 ###   "clientid": "test_product.test_device",
 ###   "username": "test_device&test_product",
@@ -136,4 +137,4 @@ Content-Type: application/json
 ###     url = "http://localhost:8090/mqtt/auth/disconnected" # 使用配置的端口
 ###     events = ["client.disconnected"]
 ###   }
-### ] 
 \ No newline at end of file
+### ]
 \ No newline at end of file
@@ -9,6 +9,7 @@ import cn.iocoder.yudao.module.iot.gateway.config.IotGatewayProperties;
 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttUpstreamRouter;
 import io.netty.handler.codec.mqtt.MqttQoS;
 import io.vertx.core.Vertx;
+import io.vertx.core.buffer.Buffer;
 import io.vertx.mqtt.MqttClient;
 import io.vertx.mqtt.MqttClientOptions;
 import jakarta.annotation.PostConstruct;
@@ -59,6 +60,7 @@ public class IotMqttUpstreamProtocol {
         // 创建 MQTT 客户端
         MqttClientOptions options = new MqttClientOptions()
+                // TODO @haohao：clientid 建议也配置文件；想通名字，会冲突哇？
                 .setClientId("yudao-iot-gateway-" + IdUtil.fastSimpleUUID())
                 .setUsername(emqxProperties.getMqttUsername())
                 .setPassword(emqxProperties.getMqttPassword())
@@ -121,6 +123,7 @@ public class IotMqttUpstreamProtocol {
      */
     private void connectMqtt() {
         // 检查必要的 MQTT 配置
+        // TODO @haohao：这些通过配置文件的 validate 去做哈，简化下；
         String host = emqxProperties.getMqttHost();
         Integer port = emqxProperties.getMqttPort();
         if (StrUtil.isBlank(host)) {
@@ -149,6 +152,7 @@ public class IotMqttUpstreamProtocol {
                     // 订阅主题
                     subscribeToTopics();
                 })
+                // TODO @haohao：这个要不要改成，必须连接成功？不做重试；不然启动也蛮危险的？
                 .exceptionally(error -> {
                     log.error("[connectMqtt][连接 MQTT Broker 失败]", error);
                     reconnectWithDelay();
@@ -178,12 +182,14 @@ public class IotMqttUpstreamProtocol {
      */
     private void subscribeToTopics() {
         List<String> topicList = emqxProperties.getMqttTopics();
+        // TODO @haohao：建议 topicList 直接 validate 校验
         if (CollUtil.isEmpty(topicList)) {
             log.warn("[subscribeToTopics][未配置 MQTT 主题，使用默认主题]");
             topicList = List.of("/sys/#"); // 默认订阅所有系统主题
         }
         for (String topic : topicList) {
+            // TODO @haohao：直接 validate 校验；嘿嘿，主要保证核心逻辑，简单点
             if (StrUtil.isBlank(topic)) {
                 log.warn("[subscribeToTopics][跳过空主题]");
                 continue;
@@ -224,7 +230,7 @@ public class IotMqttUpstreamProtocol {
             return;
         }
-        mqttClient.publish(topic, io.vertx.core.buffer.Buffer.buffer(payload), DEFAULT_QOS, false, false)
+        mqttClient.publish(topic, Buffer.buffer(payload), DEFAULT_QOS, false, false)
                 .onSuccess(v -> log.debug("[publishMessage][发送消息成功][topic: {}]", topic))
                 .onFailure(err -> log.error("[publishMessage][发送消息失败][topic: {}]", topic, err));
     }
@@ -238,6 +244,7 @@ public class IotMqttUpstreamProtocol {
         return IotDeviceMessageUtils.generateServerId(emqxProperties.getMqttPort());
     }
+    // TODO @haohao：这个要删除哇？
     /**
      * 获取 MQTT 客户端
      *
@@ -56,6 +56,7 @@ public class IotMqttEventHandler extends IotMqttAbstractHandler {
             log.info("[handle][处理设备事件上报成功][topic: {}]", topic);
             // 发送响应消息
+            // TODO @haohao：这里应该只 ack 哈；reply 在 biz 业务处理了。handleUpstreamDeviceMessage
             String method = "thing.event." + eventIdentifier + ".post";
             sendResponse(topic, JSONUtil.parseObj(payload), method);
         } catch (Exception e) {