Commit 36dd18d41faa4a34a197821e73af20116ba97da1

Authored by YunaiV
1 parent 006ef40c

【代码评审】IoT:MQTT 插件

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/util/MqttSignUtils.java
@@ -59,9 +59,11 @@ public class MqttSignUtils { @@ -59,9 +59,11 @@ public class MqttSignUtils {
59 @Getter 59 @Getter
60 @AllArgsConstructor 60 @AllArgsConstructor
61 public static class MqttSignResult { 61 public static class MqttSignResult {
  62 +
62 private final String clientId; 63 private final String clientId;
63 private final String username; 64 private final String username;
64 private final String password; 65 private final String password;
  66 +
65 } 67 }
66 68
67 } 69 }
68 \ No newline at end of file 70 \ No newline at end of file
yudao-module-iot/yudao-module-iot-plugins/yudao-module-iot-plugin-emqx/src/main/java/cn/iocoder/yudao/module/iot/plugin/emqx/config/IotPluginEmqxProperties.java
@@ -36,6 +36,7 @@ public class IotPluginEmqxProperties { @@ -36,6 +36,7 @@ public class IotPluginEmqxProperties {
36 */ 36 */
37 private boolean mqttSsl; 37 private boolean mqttSsl;
38 38
  39 + // TODO @haohao:这个是不是改成数组?
39 /** 40 /**
40 * 订阅的主题 41 * 订阅的主题
41 */ 42 */
yudao-module-iot/yudao-module-iot-plugins/yudao-module-iot-plugin-emqx/src/main/java/cn/iocoder/yudao/module/iot/plugin/emqx/downstream/IotDeviceDownstreamHandlerImpl.java
@@ -26,6 +26,7 @@ public class IotDeviceDownstreamHandlerImpl implements IotDeviceDownstreamHandle @@ -26,6 +26,7 @@ public class IotDeviceDownstreamHandlerImpl implements IotDeviceDownstreamHandle
26 26
27 private static final String SYS_TOPIC_PREFIX = "/sys/"; 27 private static final String SYS_TOPIC_PREFIX = "/sys/";
28 28
  29 + // TODO @haohao:讨论,感觉 mqtt 和 http,可以做个相对统一的格式哈。
29 // 设备服务调用 标准 JSON 30 // 设备服务调用 标准 JSON
30 // 请求Topic:/sys/${productKey}/${deviceName}/thing/service/${tsl.service.identifier} 31 // 请求Topic:/sys/${productKey}/${deviceName}/thing/service/${tsl.service.identifier}
31 // 响应Topic:/sys/${productKey}/${deviceName}/thing/service/${tsl.service.identifier}_reply 32 // 响应Topic:/sys/${productKey}/${deviceName}/thing/service/${tsl.service.identifier}_reply
yudao-module-iot/yudao-module-iot-plugins/yudao-module-iot-plugin-emqx/src/main/java/cn/iocoder/yudao/module/iot/plugin/emqx/upstream/IotDeviceUpstreamServer.java
@@ -26,17 +26,33 @@ import java.util.concurrent.TimeUnit; @@ -26,17 +26,33 @@ import java.util.concurrent.TimeUnit;
26 @Slf4j 26 @Slf4j
27 public class IotDeviceUpstreamServer { 27 public class IotDeviceUpstreamServer {
28 28
29 - private static final int RECONNECT_DELAY_MS = 5000; // 重连延迟时间(毫秒)  
30 - private static final int CONNECTION_TIMEOUT_MS = 10000; // 连接超时时间(毫秒)  
31 - private static final String TOPIC_SEPARATOR = ","; // 主题分隔符  
32 - private static final MqttQoS DEFAULT_QOS = MqttQoS.AT_LEAST_ONCE; // 默认QoS级别 29 + /**
  30 + * 重连延迟时间(毫秒)
  31 + */
  32 + private static final int RECONNECT_DELAY_MS = 5000;
  33 + /**
  34 + * 连接超时时间(毫秒)
  35 + */
  36 + private static final int CONNECTION_TIMEOUT_MS = 10000;
  37 + /**
  38 + * 主题分隔符
  39 + */
  40 + private static final String TOPIC_SEPARATOR = ",";
  41 + /**
  42 + * 默认 QoS 级别
  43 + */
  44 + private static final MqttQoS DEFAULT_QOS = MqttQoS.AT_LEAST_ONCE;
33 45
34 private final Vertx vertx; 46 private final Vertx vertx;
35 private final HttpServer server; 47 private final HttpServer server;
36 private final MqttClient client; 48 private final MqttClient client;
37 private final IotPluginEmqxProperties emqxProperties; 49 private final IotPluginEmqxProperties emqxProperties;
38 private final IotDeviceMqttMessageHandler mqttMessageHandler; 50 private final IotDeviceMqttMessageHandler mqttMessageHandler;
39 - private volatile boolean isRunning = false; // 服务运行状态标志 51 +
  52 + /**
  53 + * 服务运行状态标志
  54 + */
  55 + private volatile boolean isRunning = false;
40 56
41 public IotDeviceUpstreamServer(IotPluginEmqxProperties emqxProperties, 57 public IotDeviceUpstreamServer(IotPluginEmqxProperties emqxProperties,
42 IotDeviceUpstreamApi deviceUpstreamApi, 58 IotDeviceUpstreamApi deviceUpstreamApi,
@@ -50,6 +66,7 @@ public class IotDeviceUpstreamServer { @@ -50,6 +66,7 @@ public class IotDeviceUpstreamServer {
50 Router router = Router.router(vertx); 66 Router router = Router.router(vertx);
51 router.route().handler(BodyHandler.create()); // 处理 Body 67 router.route().handler(BodyHandler.create()); // 处理 Body
52 router.post(IotDeviceAuthVertxHandler.PATH) 68 router.post(IotDeviceAuthVertxHandler.PATH)
  69 + // TODO @haohao:疑问,mqtt 的认证,需要通过 http 呀?
53 .handler(new IotDeviceAuthVertxHandler(deviceUpstreamApi)); 70 .handler(new IotDeviceAuthVertxHandler(deviceUpstreamApi));
54 // 创建 HttpServer 实例 71 // 创建 HttpServer 实例
55 this.server = vertx.createHttpServer().requestHandler(router); 72 this.server = vertx.createHttpServer().requestHandler(router);
yudao-module-iot/yudao-module-iot-plugins/yudao-module-iot-plugin-emqx/src/main/java/cn/iocoder/yudao/module/iot/plugin/emqx/upstream/router/IotDeviceAuthVertxHandler.java
@@ -27,7 +27,7 @@ public class IotDeviceAuthVertxHandler implements Handler<RoutingContext> { @@ -27,7 +27,7 @@ public class IotDeviceAuthVertxHandler implements Handler<RoutingContext> {
27 @Override 27 @Override
28 @SuppressWarnings("unchecked") 28 @SuppressWarnings("unchecked")
29 public void handle(RoutingContext routingContext) { 29 public void handle(RoutingContext routingContext) {
30 - 30 + // TODO @haohao:try catch 兜底异常
31 JsonObject json = routingContext.body().asJsonObject(); 31 JsonObject json = routingContext.body().asJsonObject();
32 String clientId = json.getString("clientid"); 32 String clientId = json.getString("clientid");
33 String username = json.getString("username"); 33 String username = json.getString("username");
@@ -40,9 +40,11 @@ public class IotDeviceAuthVertxHandler implements Handler<RoutingContext> { @@ -40,9 +40,11 @@ public class IotDeviceAuthVertxHandler implements Handler<RoutingContext> {
40 denyAccess(routingContext); 40 denyAccess(routingContext);
41 return; 41 return;
42 } 42 }
  43 + // TODO @haohao:貌似可以考虑封装一个 writeJson ,里面有个参数是 data,然后里面去 JsonUtils.toJsonString(data)
43 IotPluginCommonUtils.writeJson(routingContext, "{\"result\": \"allow\"}"); 44 IotPluginCommonUtils.writeJson(routingContext, "{\"result\": \"allow\"}");
44 } 45 }
45 46
  47 + // TODO @haohao:下面两个简单方法,貌似可以考虑不抽小方法哈。
46 private void denyAccess(RoutingContext routingContext) { 48 private void denyAccess(RoutingContext routingContext) {
47 IotPluginCommonUtils.writeJson(routingContext, "{\"result\": \"deny\"}"); 49 IotPluginCommonUtils.writeJson(routingContext, "{\"result\": \"deny\"}");
48 } 50 }
yudao-module-iot/yudao-module-iot-plugins/yudao-module-iot-plugin-emqx/src/main/java/cn/iocoder/yudao/module/iot/plugin/emqx/upstream/router/IotDeviceMqttMessageHandler.java
@@ -26,12 +26,14 @@ import java.util.Arrays; @@ -26,12 +26,14 @@ import java.util.Arrays;
26 @Slf4j 26 @Slf4j
27 public class IotDeviceMqttMessageHandler { 27 public class IotDeviceMqttMessageHandler {
28 28
  29 + // TODO @haohao:讨论,感觉 mqtt 和 http,可以做个相对统一的格式哈。
29 // 设备上报属性 标准 JSON 30 // 设备上报属性 标准 JSON
30 - // 请求Topic:/sys/${productKey}/${deviceName}/thing/event/property/post  
31 - // 响应Topic:/sys/${productKey}/${deviceName}/thing/event/property/post_reply 31 + // 请求 Topic:/sys/${productKey}/${deviceName}/thing/event/property/post
  32 + // 响应 Topic:/sys/${productKey}/${deviceName}/thing/event/property/post_reply
  33 +
32 // 设备上报事件 标准 JSON 34 // 设备上报事件 标准 JSON
33 - // 请求Topic:/sys/${productKey}/${deviceName}/thing/event/${tsl.event.identifier}/post  
34 - // 响应Topic:/sys/${productKey}/${deviceName}/thing/event/${tsl.event.identifier}/post_reply 35 + // 请求 Topic:/sys/${productKey}/${deviceName}/thing/event/${tsl.event.identifier}/post
  36 + // 响应 Topic:/sys/${productKey}/${deviceName}/thing/event/${tsl.event.identifier}/post_reply
35 37
36 private static final String SYS_TOPIC_PREFIX = "/sys/"; 38 private static final String SYS_TOPIC_PREFIX = "/sys/";
37 private static final String PROPERTY_POST_TOPIC = "/thing/event/property/post"; 39 private static final String PROPERTY_POST_TOPIC = "/thing/event/property/post";