diff --git a/yudao-module-iot/yudao-module-iot-plugins/yudao-module-iot-plugin-emqx/src/main/java/cn/iocoder/yudao/module/iot/plugin/emqx/config/IotEmqxPlugin.java b/yudao-module-iot/yudao-module-iot-plugins/yudao-module-iot-plugin-emqx/src/main/java/cn/iocoder/yudao/module/iot/plugin/emqx/config/IotEmqxPlugin.java index 74a49c4f19..275c20eb1c 100644 --- a/yudao-module-iot/yudao-module-iot-plugins/yudao-module-iot-plugin-emqx/src/main/java/cn/iocoder/yudao/module/iot/plugin/emqx/config/IotEmqxPlugin.java +++ b/yudao-module-iot/yudao-module-iot-plugins/yudao-module-iot-plugin-emqx/src/main/java/cn/iocoder/yudao/module/iot/plugin/emqx/config/IotEmqxPlugin.java @@ -10,10 +10,9 @@ import org.springframework.context.annotation.AnnotationConfigApplicationContext /** * EMQX 插件实现类 * - * 基于 PF4J 插件框架,实现 EMQX 消息中间件的集成 - * 负责插件的生命周期管理,包括启动、停止和应用上下文的创建 + * 基于 PF4J 插件框架,实现 EMQX 消息中间件的集成:负责插件的生命周期管理,包括启动、停止和应用上下文的创建 * - * @author 芋道源码 + * @author haohao */ @Slf4j public class IotEmqxPlugin extends SpringPlugin { @@ -26,7 +25,6 @@ public class IotEmqxPlugin extends SpringPlugin { public void start() { log.info("[EmqxPlugin][EmqxPlugin 插件启动开始...]"); try { - log.info("[EmqxPlugin][EmqxPlugin 插件启动成功...]"); } catch (Exception e) { log.error("[EmqxPlugin][EmqxPlugin 插件开启动异常...]", e); @@ -52,6 +50,7 @@ public class IotEmqxPlugin extends SpringPlugin { // 继续使用插件自己的 ClassLoader 以加载插件内部的类 pluginContext.setClassLoader(getWrapper().getPluginClassLoader()); // 扫描当前插件的自动配置包 + // TODO @芋艿:是不是要配置下包 pluginContext.scan("cn.iocoder.yudao.module.iot.plugin.emqx.config"); pluginContext.refresh(); return pluginContext; diff --git a/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 b/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 index 977f0869c7..c1e64afb97 100644 --- a/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 +++ b/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 @@ -17,7 +17,6 @@ import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.MQTT_TOPIC_IL /** * EMQX 插件的 {@link IotDeviceDownstreamHandler} 实现类 - *
* * @author 芋道源码 */ @@ -26,8 +25,7 @@ public class IotDeviceDownstreamHandlerImpl implements IotDeviceDownstreamHandle private static final String SYS_TOPIC_PREFIX = "/sys/"; - // TODO @haohao:讨论,感觉 mqtt 和 http,可以做个相对统一的格式哈。 - // 回复 都使用 Alink 格式,方便后续扩展。 + // TODO @haohao:讨论,感觉 mqtt 和 http,可以做个相对统一的格式哈。;回复 都使用 Alink 格式,方便后续扩展。 // 设备服务调用 标准 JSON // 请求Topic:/sys/${productKey}/${deviceName}/thing/service/${tsl.service.identifier} // 响应Topic:/sys/${productKey}/${deviceName}/thing/service/${tsl.service.identifier}_reply @@ -62,11 +60,8 @@ public class IotDeviceDownstreamHandlerImpl implements IotDeviceDownstreamHandle try { // 构建请求主题 String topic = buildServiceTopic(reqDTO.getProductKey(), reqDTO.getDeviceName(), reqDTO.getIdentifier()); - - // 生成请求ID(如果没有提供) - String requestId = reqDTO.getRequestId() != null ? reqDTO.getRequestId() : generateRequestId(); - // 构建请求消息 + String requestId = reqDTO.getRequestId() != null ? reqDTO.getRequestId() : generateRequestId(); JSONObject request = buildServiceRequest(requestId, reqDTO.getIdentifier(), reqDTO.getParams()); // 发送消息 @@ -98,11 +93,8 @@ public class IotDeviceDownstreamHandlerImpl implements IotDeviceDownstreamHandle try { // 构建请求主题 String topic = buildPropertySetTopic(reqDTO.getProductKey(), reqDTO.getDeviceName()); - - // 生成请求ID(如果没有提供) - String requestId = reqDTO.getRequestId() != null ? reqDTO.getRequestId() : generateRequestId(); - // 构建请求消息 + String requestId = reqDTO.getRequestId() != null ? reqDTO.getRequestId() : generateRequestId(); JSONObject request = buildPropertySetRequest(requestId, reqDTO.getProperties()); // 发送消息 @@ -163,7 +155,7 @@ public class IotDeviceDownstreamHandlerImpl implements IotDeviceDownstreamHandle } /** - * 发布MQTT消息 + * 发布 MQTT 消息 */ private void publishMessage(String topic, JSONObject payload) { mqttClient.publish( diff --git a/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 b/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 index 472eb83f7f..fcb2286158 100644 --- a/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 +++ b/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 @@ -13,13 +13,12 @@ import lombok.extern.slf4j.Slf4j; import java.util.Collections; /** - * IoT Emqx 连接认证的 Vert.x Handler - * MQTT - * HTTP + * IoT EMQX 连接认证的 Vert.x Handler + * + * EMQX HTTP * * 注意:该处理器需要返回特定格式:{"result": "allow"} 或 {"result": "deny"}, - * 以符合 EMQX 认证插件的要求,因此不使用 IotStandardResponse 实体类。 + * 以符合 EMQX 认证插件的要求,因此不使用 IotStandardResponse 实体类 * * @author haohao */ diff --git a/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 b/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 index b92868582c..6cf8d84c5c 100644 --- a/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 +++ b/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 @@ -1,5 +1,6 @@ package cn.iocoder.yudao.module.iot.plugin.emqx.upstream.router; +import cn.hutool.core.util.StrUtil; import cn.hutool.json.JSONObject; import cn.hutool.json.JSONUtil; import cn.iocoder.yudao.framework.common.util.json.JsonUtils; @@ -21,26 +22,20 @@ import java.util.Map; /** * IoT 设备 MQTT 消息处理器 - *
- * 参考: - *
- * "...">
+ *
+ * 参考:"设备属性、事件、服务">
*/
@Slf4j
public class IotDeviceMqttMessageHandler {
- // TODO @haohao:讨论,感觉 mqtt 和 http,可以做个相对统一的格式哈。
- // 回复 都使用 Alink 格式,方便后续扩展。
+ // TODO @haohao:讨论,感觉 mqtt 和 http,可以做个相对统一的格式哈;回复 都使用 Alink 格式,方便后续扩展。
// 设备上报属性 标准 JSON
// 请求 Topic:/sys/${productKey}/${deviceName}/thing/event/property/post
// 响应 Topic:/sys/${productKey}/${deviceName}/thing/event/property/post_reply
// 设备上报事件 标准 JSON
- // 请求
- // Topic:/sys/${productKey}/${deviceName}/thing/event/${tsl.event.identifier}/post
- // 响应
- // Topic:/sys/${productKey}/${deviceName}/thing/event/${tsl.event.identifier}/post_reply
+ // 请求 Topic:/sys/${productKey}/${deviceName}/thing/event/${tsl.event.identifier}/post
+ // 响应 Topic:/sys/${productKey}/${deviceName}/thing/event/${tsl.event.identifier}/post_reply
private static final String SYS_TOPIC_PREFIX = "/sys/";
private static final String PROPERTY_POST_TOPIC = "/thing/event/property/post";
@@ -70,7 +65,7 @@ public class IotDeviceMqttMessageHandler {
log.info("[messageHandler][接收到消息][topic: {}][payload: {}]", topic, payload);
try {
- if (payload == null || payload.isEmpty()) {
+ if (StrUtil.isEmpty(payload)) {
log.warn("[messageHandler][消息内容为空][topic: {}]", topic);
return;
}
@@ -214,27 +209,20 @@ public class IotDeviceMqttMessageHandler {
* @param topic 原始主题
* @param jsonObject 原始消息JSON对象
* @param method 响应方法
- * @param customData 自定义数据,可为null
+ * @param customData 自定义数据,可为 null
*/
private void sendResponse(String topic, JSONObject jsonObject, String method, Object customData) {
String replyTopic = topic + REPLY_SUFFIX;
- // 使用IotStandardResponse实体类构建响应
+ // 响应结果
IotStandardResponse response = IotStandardResponse.success(
- jsonObject.getStr("id"),
- method,
- customData);
-
+ jsonObject.getStr("id"), method, customData);
try {
- mqttClient.publish(replyTopic,
- Buffer.buffer(JsonUtils.toJsonString(response)),
- MqttQoS.AT_LEAST_ONCE,
- false,
- false);
+ mqttClient.publish(replyTopic, Buffer.buffer(JsonUtils.toJsonString(response)),
+ MqttQoS.AT_LEAST_ONCE, false, false);
log.info("[sendResponse][发送响应消息成功][topic: {}]", replyTopic);
} catch (Exception e) {
- log.error("[sendResponse][发送响应消息失败][topic: {}][response: {}]",
- replyTopic, response, e);
+ log.error("[sendResponse][发送响应消息失败][topic: {}][response: {}]", replyTopic, response, e);
}
}
@@ -304,4 +292,5 @@ public class IotDeviceMqttMessageHandler {
return reportReqDTO;
}
+
}
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-plugins/yudao-module-iot-plugin-emqx/src/main/java/cn/iocoder/yudao/module/iot/plugin/emqx/upstream/router/IotDeviceWebhookVertxHandler.java b/yudao-module-iot/yudao-module-iot-plugins/yudao-module-iot-plugin-emqx/src/main/java/cn/iocoder/yudao/module/iot/plugin/emqx/upstream/router/IotDeviceWebhookVertxHandler.java
index 6f1e8a11b8..93fb01bc0a 100644
--- a/yudao-module-iot/yudao-module-iot-plugins/yudao-module-iot-plugin-emqx/src/main/java/cn/iocoder/yudao/module/iot/plugin/emqx/upstream/router/IotDeviceWebhookVertxHandler.java
+++ b/yudao-module-iot/yudao-module-iot-plugins/yudao-module-iot-plugin-emqx/src/main/java/cn/iocoder/yudao/module/iot/plugin/emqx/upstream/router/IotDeviceWebhookVertxHandler.java
@@ -16,13 +16,12 @@ import java.time.LocalDateTime;
import java.util.Collections;
/**
- * IoT Emqx Webhook 事件处理的 Vert.x Handler
+ * IoT EMQX Webhook 事件处理的 Vert.x Handler
*
- * EMQXWebhook
+ * EMQX Webhook
*
* 注意:该处理器需要返回特定格式:{"result": "success"} 或 {"result": "error"},
- * 以符合 EMQX Webhook 插件的要求,因此不使用 IotStandardResponse 实体类。
+ * 以符合 EMQX Webhook 插件的要求,因此不使用 IotStandardResponse 实体类。
*
* @author haohao
*/
@@ -137,7 +136,7 @@ public class IotDeviceWebhookVertxHandler implements Handler