BJ_ZTEITS-Developers / parking-cloud-yudao

Commit 81739186c971fdc508cf5cd115e5be49a26f38dd

Authored by 安浩浩
1 parent 348c1387

【功能优化】IoT：重构上行请求处理逻辑，合并属性和事件上报处理，简化代码结构，删除冗余处理器

Inline Side-by-side
Showing 9 changed files with 201 additions and 264 deletions
yudao-module-iot/yudao-module-iot-biz/pom.xml
  View file @8173918
@@ -64,17 +64,6 @@ @@ -64,17 +64,6 @@
64 <artifactId>yudao-spring-boot-starter-excel</artifactId> 64 <artifactId>yudao-spring-boot-starter-excel</artifactId>
65 </dependency> 65 </dependency>
66 66
67 - <!-- TODO @haohao：貌似不需要这个？ -->  
68 -<!-- <dependency>-->  
69 -<!-- <groupId>io.vertx</groupId>-->  
70 -<!-- <artifactId>vertx-web</artifactId>-->  
71 -<!-- </dependency>-->  
72 -  
73 - <!-- TODO @haohao：貌似 biz 模块，不需要 MQTT？ -->  
74 -<!-- <dependency>-->  
75 -<!-- <groupId>org.eclipse.paho</groupId> &lt;!&ndash; MQTT &ndash;&gt;-->  
76 -<!-- <artifactId>org.eclipse.paho.client.mqttv3</artifactId>-->  
77 -<!-- </dependency>-->  
78 <!-- 消息队列相关 --> 67 <!-- 消息队列相关 -->
79 <dependency> 68 <dependency>
80 <groupId>org.apache.rocketmq</groupId> 69 <groupId>org.apache.rocketmq</groupId>
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotKafkaMQDataBridgeExecute.java
  View file @8173918
@@ -10,7 +10,6 @@ import org.apache.kafka.common.serialization.StringSerializer; @@ -10,7 +10,6 @@ import org.apache.kafka.common.serialization.StringSerializer;
10 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; 10 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
11 import org.springframework.kafka.core.DefaultKafkaProducerFactory; 11 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
12 import org.springframework.kafka.core.KafkaTemplate; 12 import org.springframework.kafka.core.KafkaTemplate;
13 -import org.springframework.stereotype.Component;  
14 13
15 import java.time.Duration; 14 import java.time.Duration;
16 import java.time.LocalDateTime; 15 import java.time.LocalDateTime;
@@ -24,7 +23,7 @@ import java.util.concurrent.TimeUnit; @@ -24,7 +23,7 @@ import java.util.concurrent.TimeUnit;
24 * @author HUIHUI 23 * @author HUIHUI
25 */ 24 */
26 @ConditionalOnClass(name = "org.springframework.kafka.core.KafkaTemplate") 25 @ConditionalOnClass(name = "org.springframework.kafka.core.KafkaTemplate")
27 -@Component 26 +//@Component
28 @Slf4j 27 @Slf4j
29 public class IotKafkaMQDataBridgeExecute extends 28 public class IotKafkaMQDataBridgeExecute extends
30 AbstractCacheableDataBridgeExecute<IotDataBridgeKafkaMQConfig, KafkaTemplate<String, String>> { 29 AbstractCacheableDataBridgeExecute<IotDataBridgeKafkaMQConfig, KafkaTemplate<String, String>> {
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
  View file @8173918
@@ -7,10 +7,13 @@ import org.pf4j.spring.SpringPlugin; @@ -7,10 +7,13 @@ import org.pf4j.spring.SpringPlugin;
7 import org.springframework.context.ApplicationContext; 7 import org.springframework.context.ApplicationContext;
8 import org.springframework.context.annotation.AnnotationConfigApplicationContext; 8 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
9 9
10 -// TODO @芋艿：完善注释  
11 -  
12 /** 10 /**
13 - * 负责插件的启动和停止 11 + * EMQX 插件实现类
  12 + *
  13 + * 基于 PF4J 插件框架，实现 EMQX 消息中间件的集成
  14 + * 负责插件的生命周期管理，包括启动、停止和应用上下文的创建
  15 + *
  16 + * @author 芋道源码
14 */ 17 */
15 @Slf4j 18 @Slf4j
16 public class IotEmqxPlugin extends SpringPlugin { 19 public class IotEmqxPlugin extends SpringPlugin {
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
  View file @8173918
@@ -27,6 +27,7 @@ public class IotDeviceDownstreamHandlerImpl implements IotDeviceDownstreamHandle @@ -27,6 +27,7 @@ public class IotDeviceDownstreamHandlerImpl implements IotDeviceDownstreamHandle
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 // TODO @haohao：讨论，感觉 mqtt 和 http，可以做个相对统一的格式哈。
  30 + // 回复 都使用 Alink 格式，方便后续扩展。
30 // 设备服务调用 标准 JSON 31 // 设备服务调用 标准 JSON
31 // 请求Topic：/sys/${productKey}/${deviceName}/thing/service/${tsl.service.identifier} 32 // 请求Topic：/sys/${productKey}/${deviceName}/thing/service/${tsl.service.identifier}
32 // 响应Topic：/sys/${productKey}/${deviceName}/thing/service/${tsl.service.identifier}_reply 33 // 响应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/router/IotDeviceMqttMessageHandler.java
  View file @8173918
@@ -31,6 +31,7 @@ import java.util.Map; @@ -31,6 +31,7 @@ import java.util.Map;
31 public class IotDeviceMqttMessageHandler { 31 public class IotDeviceMqttMessageHandler {
32 32
33 // TODO @haohao：讨论，感觉 mqtt 和 http，可以做个相对统一的格式哈。 33 // TODO @haohao：讨论，感觉 mqtt 和 http，可以做个相对统一的格式哈。
  34 + // 回复 都使用 Alink 格式，方便后续扩展。
34 // 设备上报属性 标准 JSON 35 // 设备上报属性 标准 JSON
35 // 请求 Topic：/sys/${productKey}/${deviceName}/thing/event/property/post 36 // 请求 Topic：/sys/${productKey}/${deviceName}/thing/event/property/post
36 // 响应 Topic：/sys/${productKey}/${deviceName}/thing/event/property/post_reply 37 // 响应 Topic：/sys/${productKey}/${deviceName}/thing/event/property/post_reply
yudao-module-iot/yudao-module-iot-plugins/yudao-module-iot-plugin-http/src/main/java/cn/iocoder/yudao/module/iot/plugin/http/upstream/IotDeviceUpstreamServer.java
  View file @8173918
@@ -2,8 +2,7 @@ package cn.iocoder.yudao.module.iot.plugin.http.upstream; @@ -2,8 +2,7 @@ package cn.iocoder.yudao.module.iot.plugin.http.upstream;
2 2
3 import cn.iocoder.yudao.module.iot.api.device.IotDeviceUpstreamApi; 3 import cn.iocoder.yudao.module.iot.api.device.IotDeviceUpstreamApi;
4 import cn.iocoder.yudao.module.iot.plugin.http.config.IotPluginHttpProperties; 4 import cn.iocoder.yudao.module.iot.plugin.http.config.IotPluginHttpProperties;
5 -import cn.iocoder.yudao.module.iot.plugin.http.upstream.router.IotDeviceEventReportVertxHandler;  
6 -import cn.iocoder.yudao.module.iot.plugin.http.upstream.router.IotDevicePropertyReportVertxHandler; 5 +import cn.iocoder.yudao.module.iot.plugin.http.upstream.router.IotDeviceUpstreamVertxHandler;
7 import io.vertx.core.Vertx; 6 import io.vertx.core.Vertx;
8 import io.vertx.core.http.HttpServer; 7 import io.vertx.core.http.HttpServer;
9 import io.vertx.ext.web.Router; 8 import io.vertx.ext.web.Router;
@@ -32,10 +31,12 @@ public class IotDeviceUpstreamServer { @@ -32,10 +31,12 @@ public class IotDeviceUpstreamServer {
32 // 创建 Router 实例 31 // 创建 Router 实例
33 Router router = Router.router(vertx); 32 Router router = Router.router(vertx);
34 router.route().handler(BodyHandler.create()); // 处理 Body 33 router.route().handler(BodyHandler.create()); // 处理 Body
35 - router.post(IotDevicePropertyReportVertxHandler.PATH)  
36 - .handler(new IotDevicePropertyReportVertxHandler(deviceUpstreamApi));  
37 - router.post(IotDeviceEventReportVertxHandler.PATH)  
38 - .handler(new IotDeviceEventReportVertxHandler(deviceUpstreamApi)); 34 +
  35 + // 使用统一的 Handler 处理所有上行请求
  36 + IotDeviceUpstreamVertxHandler upstreamHandler = new IotDeviceUpstreamVertxHandler(deviceUpstreamApi);
  37 + router.post(IotDeviceUpstreamVertxHandler.PROPERTY_PATH).handler(upstreamHandler);
  38 + router.post(IotDeviceUpstreamVertxHandler.EVENT_PATH).handler(upstreamHandler);
  39 +
39 // 创建 HttpServer 实例 40 // 创建 HttpServer 实例
40 this.server = vertx.createHttpServer().requestHandler(router); 41 this.server = vertx.createHttpServer().requestHandler(router);
41 } 42 }
yudao-module-iot/yudao-module-iot-plugins/yudao-module-iot-plugin-http/src/main/java/cn/iocoder/yudao/module/iot/plugin/http/upstream/router/IotDeviceEventReportVertxHandler.java deleted
View file @348c138
1 -package cn.iocoder.yudao.module.iot.plugin.http.upstream.router;  
2 -  
3 -import cn.hutool.core.util.IdUtil;  
4 -import cn.hutool.core.util.ObjUtil;  
5 -import cn.iocoder.yudao.framework.common.pojo.CommonResult;  
6 -import cn.iocoder.yudao.module.iot.api.device.IotDeviceUpstreamApi;  
7 -import cn.iocoder.yudao.module.iot.api.device.dto.control.upstream.IotDeviceEventReportReqDTO;  
8 -import cn.iocoder.yudao.module.iot.api.device.dto.control.upstream.IotDeviceStateUpdateReqDTO;  
9 -import cn.iocoder.yudao.module.iot.enums.device.IotDeviceStateEnum;  
10 -import cn.iocoder.yudao.module.iot.plugin.common.pojo.IotStandardResponse;  
11 -import cn.iocoder.yudao.module.iot.plugin.common.util.IotPluginCommonUtils;  
12 -import io.vertx.core.Handler;  
13 -import io.vertx.core.json.JsonObject;  
14 -import io.vertx.ext.web.RoutingContext;  
15 -import lombok.RequiredArgsConstructor;  
16 -import lombok.extern.slf4j.Slf4j;  
17 -  
18 -import java.time.LocalDateTime;  
19 -import java.util.HashMap;  
20 -import java.util.Map;  
21 -  
22 -import static cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants.BAD_REQUEST;  
23 -import static cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants.INTERNAL_SERVER_ERROR;  
24 -  
25 -/**  
26 - * IoT 设备事件上报的 Vert.x Handler  
27 - */  
28 -@RequiredArgsConstructor  
29 -@Slf4j  
30 -public class IotDeviceEventReportVertxHandler implements Handler<RoutingContext> {  
31 -  
32 - public static final String PATH = "/sys/:productKey/:deviceName/thing/event/:identifier/post";  
33 - private static final String VERSION = "1.0";  
34 - private static final String EVENT_METHOD_PREFIX = "thing.event.";  
35 - private static final String EVENT_METHOD_SUFFIX = ".post";  
36 -  
37 - private final IotDeviceUpstreamApi deviceUpstreamApi;  
38 -  
39 - @Override  
40 - @SuppressWarnings("unchecked")  
41 - public void handle(RoutingContext routingContext) {  
42 - // 1. 解析参数  
43 - IotDeviceEventReportReqDTO reportReqDTO;  
44 - String identifier = null;  
45 - String requestId = IdUtil.fastSimpleUUID();  
46 - try {  
47 - String productKey = routingContext.pathParam("productKey");  
48 - String deviceName = routingContext.pathParam("deviceName");  
49 - identifier = routingContext.pathParam("identifier");  
50 - JsonObject body = routingContext.body().asJsonObject();  
51 - requestId = ObjUtil.defaultIfBlank(body.getString("id"), requestId);  
52 -  
53 - // 按照标准JSON格式处理事件参数  
54 - Map<String, Object> params;  
55 -  
56 - // 优先使用params字段，符合标准  
57 - if (body.getJsonObject("params") != null) {  
58 - params = body.getJsonObject("params").getMap();  
59 - } else {  
60 - // 兼容旧格式  
61 - params = new HashMap<>();  
62 - }  
63 -  
64 - reportReqDTO = ((IotDeviceEventReportReqDTO) new IotDeviceEventReportReqDTO().setRequestId(requestId)  
65 - .setProcessId(IotPluginCommonUtils.getProcessId()).setReportTime(LocalDateTime.now())  
66 - .setProductKey(productKey).setDeviceName(deviceName))  
67 - .setIdentifier(identifier).setParams(params);  
68 - } catch (Exception e) {  
69 - log.error("[handle][路径参数({}) 解析参数失败]", routingContext.pathParams(), e);  
70 - // 使用IotStandardResponse实体类返回错误  
71 - String method = identifier != null ? EVENT_METHOD_PREFIX + identifier + EVENT_METHOD_SUFFIX  
72 - : "thing.event.unknown.post";  
73 - IotStandardResponse errorResponse = IotStandardResponse.error(  
74 - requestId, method, BAD_REQUEST.getCode(), BAD_REQUEST.getMsg());  
75 - IotPluginCommonUtils.writeJsonResponse(routingContext, errorResponse);  
76 - return;  
77 - }  
78 -  
79 - try {  
80 - // 2. 设备上线  
81 - deviceUpstreamApi.updateDeviceState(  
82 - ((IotDeviceStateUpdateReqDTO) new IotDeviceStateUpdateReqDTO().setRequestId(IdUtil.fastSimpleUUID())  
83 - .setProcessId(IotPluginCommonUtils.getProcessId()).setReportTime(LocalDateTime.now())  
84 - .setProductKey(reportReqDTO.getProductKey()).setDeviceName(reportReqDTO.getDeviceName()))  
85 - .setState(IotDeviceStateEnum.ONLINE.getState()));  
86 -  
87 - // 3.1 事件上报  
88 - CommonResult<Boolean> result = deviceUpstreamApi.reportDeviceEvent(reportReqDTO);  
89 -  
90 - // 3.2 返回结果 - 使用IotStandardResponse实体类  
91 - String method = EVENT_METHOD_PREFIX + reportReqDTO.getIdentifier() + EVENT_METHOD_SUFFIX;  
92 - IotStandardResponse response;  
93 - if (result.isSuccess()) {  
94 - response = IotStandardResponse.success(reportReqDTO.getRequestId(), method, result.getData());  
95 - } else {  
96 - response = IotStandardResponse.error(  
97 - reportReqDTO.getRequestId(), method, result.getCode(), result.getMsg());  
98 - }  
99 - IotPluginCommonUtils.writeJsonResponse(routingContext, response);  
100 - } catch (Exception e) {  
101 - log.error("[handle][请求参数({}) 事件上报异常]", reportReqDTO, e);  
102 -  
103 - // 构建错误响应 - 使用IotStandardResponse实体类  
104 - String method = EVENT_METHOD_PREFIX + reportReqDTO.getIdentifier() + EVENT_METHOD_SUFFIX;  
105 - IotStandardResponse errorResponse = IotStandardResponse.error(  
106 - reportReqDTO.getRequestId(), method,  
107 - INTERNAL_SERVER_ERROR.getCode(), INTERNAL_SERVER_ERROR.getMsg());  
108 - IotPluginCommonUtils.writeJsonResponse(routingContext, errorResponse);  
109 - }  
110 - }  
111 -}  
yudao-module-iot/yudao-module-iot-plugins/yudao-module-iot-plugin-http/src/main/java/cn/iocoder/yudao/module/iot/plugin/http/upstream/router/IotDevicePropertyReportVertxHandler.java deleted
View file @348c138
1 -package cn.iocoder.yudao.module.iot.plugin.http.upstream.router;  
2 -  
3 -import cn.hutool.core.util.IdUtil;  
4 -import cn.hutool.core.util.ObjUtil;  
5 -import cn.iocoder.yudao.framework.common.pojo.CommonResult;  
6 -import cn.iocoder.yudao.module.iot.api.device.IotDeviceUpstreamApi;  
7 -import cn.iocoder.yudao.module.iot.api.device.dto.control.upstream.IotDevicePropertyReportReqDTO;  
8 -import cn.iocoder.yudao.module.iot.api.device.dto.control.upstream.IotDeviceStateUpdateReqDTO;  
9 -import cn.iocoder.yudao.module.iot.enums.device.IotDeviceStateEnum;  
10 -import cn.iocoder.yudao.module.iot.plugin.common.pojo.IotStandardResponse;  
11 -import cn.iocoder.yudao.module.iot.plugin.common.util.IotPluginCommonUtils;  
12 -import io.vertx.core.Handler;  
13 -import io.vertx.core.json.JsonObject;  
14 -import io.vertx.ext.web.RoutingContext;  
15 -import lombok.RequiredArgsConstructor;  
16 -import lombok.extern.slf4j.Slf4j;  
17 -  
18 -import java.time.LocalDateTime;  
19 -import java.util.HashMap;  
20 -import java.util.Map;  
21 -  
22 -import static cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants.BAD_REQUEST;  
23 -import static cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants.INTERNAL_SERVER_ERROR;  
24 -  
25 -// TODO @芋艿：【待定 005】要不要简化成，解析后，统一处理？只有一个 Handler！！！  
26 -/**  
27 - * IoT 设备属性上报的 Vert.x Handler  
28 - *  
29 - * @author haohao  
30 - */  
31 -@RequiredArgsConstructor  
32 -@Slf4j  
33 -public class IotDevicePropertyReportVertxHandler implements Handler<RoutingContext> {  
34 -  
35 - public static final String PATH = "/sys/:productKey/:deviceName/thing/event/property/post";  
36 - private static final String VERSION = "1.0";  
37 - private static final String METHOD = "thing.event.property.post";  
38 -  
39 - private final IotDeviceUpstreamApi deviceUpstreamApi;  
40 -  
41 - @Override  
42 - @SuppressWarnings("unchecked")  
43 - public void handle(RoutingContext routingContext) {  
44 - // 1. 解析参数  
45 - IotDevicePropertyReportReqDTO reportReqDTO;  
46 - String requestId = IdUtil.fastSimpleUUID();  
47 - try {  
48 - String productKey = routingContext.pathParam("productKey");  
49 - String deviceName = routingContext.pathParam("deviceName");  
50 - JsonObject body = routingContext.body().asJsonObject();  
51 - requestId = ObjUtil.defaultIfBlank(body.getString("id"), requestId);  
52 -  
53 - // 按照标准JSON格式处理属性数据  
54 - Map<String, Object> properties = new HashMap<>();  
55 -  
56 - // 优先使用params字段，符合标准  
57 - Map<String, Object> params = body.getJsonObject("params") != null ? body.getJsonObject("params").getMap()  
58 - : null;  
59 -  
60 - if (params != null) {  
61 - // 将标准格式的params转换为平台需要的properties格式  
62 - for (Map.Entry<String, Object> entry : params.entrySet()) {  
63 - String key = entry.getKey();  
64 - Object valueObj = entry.getValue();  
65 -  
66 - // 如果是复杂结构（包含value和time）  
67 - if (valueObj instanceof Map) {  
68 - Map<String, Object> valueMap = (Map<String, Object>) valueObj;  
69 - if (valueMap.containsKey("value")) {  
70 - properties.put(key, valueMap.get("value"));  
71 - } else {  
72 - properties.put(key, valueObj);  
73 - }  
74 - } else {  
75 - properties.put(key, valueObj);  
76 - }  
77 - }  
78 - } else {  
79 - // 兼容旧格式，直接使用properties字段  
80 - properties = body.getJsonObject("properties") != null ? body.getJsonObject("properties").getMap()  
81 - : new HashMap<>();  
82 - }  
83 -  
84 - reportReqDTO = ((IotDevicePropertyReportReqDTO) new IotDevicePropertyReportReqDTO().setRequestId(requestId)  
85 - .setProcessId(IotPluginCommonUtils.getProcessId()).setReportTime(LocalDateTime.now())  
86 - .setProductKey(productKey).setDeviceName(deviceName))  
87 - .setProperties(properties);  
88 - } catch (Exception e) {  
89 - log.error("[handle][路径参数({}) 解析参数失败]", routingContext.pathParams(), e);  
90 - // 使用IotStandardResponse实体类返回错误  
91 - IotStandardResponse errorResponse = IotStandardResponse.error(  
92 - requestId, METHOD, BAD_REQUEST.getCode(), BAD_REQUEST.getMsg());  
93 - IotPluginCommonUtils.writeJsonResponse(routingContext, errorResponse);  
94 - return;  
95 - }  
96 -  
97 - // TODO @芋艿：secret 校验。目前的想法：  
98 - // 方案一：请求的时候，带上 secret 参数，然后进行校验，减少请求的频次。不过可能要看下 mqtt 能不能复用！  
99 - // 方案二：本地有设备信息的缓存，异步刷新。这样可能 mqtt 的校验，和 http 校验都容易适配。  
100 -  
101 - try {  
102 - // 2. 设备上线  
103 - deviceUpstreamApi.updateDeviceState(  
104 - ((IotDeviceStateUpdateReqDTO) new IotDeviceStateUpdateReqDTO().setRequestId(IdUtil.fastSimpleUUID())  
105 - .setProcessId(IotPluginCommonUtils.getProcessId()).setReportTime(LocalDateTime.now())  
106 - .setProductKey(reportReqDTO.getProductKey()).setDeviceName(reportReqDTO.getDeviceName()))  
107 - .setState(IotDeviceStateEnum.ONLINE.getState()));  
108 -  
109 - // 3.1 属性上报  
110 - CommonResult<Boolean> result = deviceUpstreamApi.reportDeviceProperty(reportReqDTO);  
111 -  
112 - // 3.2 返回结果 - 使用IotStandardResponse实体类  
113 - IotStandardResponse response;  
114 - if (result.isSuccess()) {  
115 - response = IotStandardResponse.success(reportReqDTO.getRequestId(), METHOD, result.getData());  
116 - } else {  
117 - response = IotStandardResponse.error(  
118 - reportReqDTO.getRequestId(), METHOD, result.getCode(), result.getMsg());  
119 - }  
120 - IotPluginCommonUtils.writeJsonResponse(routingContext, response);  
121 - } catch (Exception e) {  
122 - log.error("[handle][请求参数({}) 属性上报异常]", reportReqDTO, e);  
123 -  
124 - // 构建错误响应 - 使用IotStandardResponse实体类  
125 - IotStandardResponse errorResponse = IotStandardResponse.error(  
126 - reportReqDTO.getRequestId(), METHOD,  
127 - INTERNAL_SERVER_ERROR.getCode(), INTERNAL_SERVER_ERROR.getMsg());  
128 - IotPluginCommonUtils.writeJsonResponse(routingContext, errorResponse);  
129 - }  
130 - }  
131 -}  
yudao-module-iot/yudao-module-iot-plugins/yudao-module-iot-plugin-http/src/main/java/cn/iocoder/yudao/module/iot/plugin/http/upstream/router/IotDeviceUpstreamVertxHandler.java 0 → 100644
  View file @8173918
  1 +package cn.iocoder.yudao.module.iot.plugin.http.upstream.router;
  2 +
  3 +import cn.hutool.core.util.IdUtil;
  4 +import cn.hutool.core.util.ObjUtil;
  5 +import cn.iocoder.yudao.framework.common.pojo.CommonResult;
  6 +import cn.iocoder.yudao.module.iot.api.device.IotDeviceUpstreamApi;
  7 +import cn.iocoder.yudao.module.iot.api.device.dto.control.upstream.IotDeviceEventReportReqDTO;
  8 +import cn.iocoder.yudao.module.iot.api.device.dto.control.upstream.IotDevicePropertyReportReqDTO;
  9 +import cn.iocoder.yudao.module.iot.api.device.dto.control.upstream.IotDeviceStateUpdateReqDTO;
  10 +import cn.iocoder.yudao.module.iot.enums.device.IotDeviceStateEnum;
  11 +import cn.iocoder.yudao.module.iot.plugin.common.pojo.IotStandardResponse;
  12 +import cn.iocoder.yudao.module.iot.plugin.common.util.IotPluginCommonUtils;
  13 +import io.vertx.core.Handler;
  14 +import io.vertx.core.json.JsonObject;
  15 +import io.vertx.ext.web.RoutingContext;
  16 +import lombok.RequiredArgsConstructor;
  17 +import lombok.extern.slf4j.Slf4j;
  18 +
  19 +import java.time.LocalDateTime;
  20 +import java.util.HashMap;
  21 +import java.util.Map;
  22 +
  23 +import static cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants.BAD_REQUEST;
  24 +import static cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants.INTERNAL_SERVER_ERROR;
  25 +
  26 +/**
  27 + * IoT 设备上行统一处理的 Vert.x Handler
  28 + * <p>
  29 + * 统一处理设备属性上报和事件上报的请求
  30 + *
  31 + * @author haohao
  32 + */
  33 +@RequiredArgsConstructor
  34 +@Slf4j
  35 +public class IotDeviceUpstreamVertxHandler implements Handler<RoutingContext> {
  36 +
  37 + // 属性上报路径
  38 + public static final String PROPERTY_PATH = "/sys/:productKey/:deviceName/thing/event/property/post";
  39 + // 事件上报路径
  40 + public static final String EVENT_PATH = "/sys/:productKey/:deviceName/thing/event/:identifier/post";
  41 +
  42 + private static final String PROPERTY_METHOD = "thing.event.property.post";
  43 + private static final String EVENT_METHOD_PREFIX = "thing.event.";
  44 + private static final String EVENT_METHOD_SUFFIX = ".post";
  45 +
  46 + private final IotDeviceUpstreamApi deviceUpstreamApi;
  47 +
  48 + @Override
  49 + public void handle(RoutingContext routingContext) {
  50 + String path = routingContext.request().path();
  51 + String requestId = IdUtil.fastSimpleUUID();
  52 +
  53 + try {
  54 + // 1. 解析通用参数
  55 + String productKey = routingContext.pathParam("productKey");
  56 + String deviceName = routingContext.pathParam("deviceName");
  57 + JsonObject body = routingContext.body().asJsonObject();
  58 + requestId = ObjUtil.defaultIfBlank(body.getString("id"), requestId);
  59 +
  60 + // 2. 根据路径模式处理不同类型的请求
  61 + CommonResult<Boolean> result;
  62 + String method;
  63 +
  64 + if (path.matches(".*/thing/event/property/post")) {
  65 + // 处理属性上报
  66 + IotDevicePropertyReportReqDTO reportReqDTO = parsePropertyReportRequest(productKey, deviceName, requestId, body);
  67 +
  68 + // 设备上线
  69 + updateDeviceState(reportReqDTO.getProductKey(), reportReqDTO.getDeviceName());
  70 +
  71 + // 属性上报
  72 + result = deviceUpstreamApi.reportDeviceProperty(reportReqDTO);
  73 + method = PROPERTY_METHOD;
  74 + } else if (path.matches(".*/thing/event/.+/post")) {
  75 + // 处理事件上报
  76 + String identifier = routingContext.pathParam("identifier");
  77 + IotDeviceEventReportReqDTO reportReqDTO = parseEventReportRequest(productKey, deviceName, identifier, requestId, body);
  78 +
  79 + // 设备上线
  80 + updateDeviceState(reportReqDTO.getProductKey(), reportReqDTO.getDeviceName());
  81 +
  82 + // 事件上报
  83 + result = deviceUpstreamApi.reportDeviceEvent(reportReqDTO);
  84 + method = EVENT_METHOD_PREFIX + identifier + EVENT_METHOD_SUFFIX;
  85 + } else {
  86 + // 不支持的请求路径
  87 + IotStandardResponse errorResponse = IotStandardResponse.error(requestId, "unknown", BAD_REQUEST.getCode(), "不支持的请求路径");
  88 + IotPluginCommonUtils.writeJsonResponse(routingContext, errorResponse);
  89 + return;
  90 + }
  91 +
  92 + // 3. 返回标准响应
  93 + IotStandardResponse response;
  94 + if (result.isSuccess()) {
  95 + response = IotStandardResponse.success(requestId, method, result.getData());
  96 + } else {
  97 + response = IotStandardResponse.error(requestId, method, result.getCode(), result.getMsg());
  98 + }
  99 + IotPluginCommonUtils.writeJsonResponse(routingContext, response);
  100 +
  101 + } catch (Exception e) {
  102 + log.error("[handle][处理上行请求异常] path={}", path, e);
  103 +
  104 + // 构建错误响应
  105 + String method = path.contains("/property/") ? PROPERTY_METHOD : EVENT_METHOD_PREFIX + (routingContext.pathParams().containsKey("identifier") ? routingContext.pathParam("identifier") : "unknown") + EVENT_METHOD_SUFFIX;
  106 +
  107 + IotStandardResponse errorResponse = IotStandardResponse.error(requestId, method, INTERNAL_SERVER_ERROR.getCode(), INTERNAL_SERVER_ERROR.getMsg());
  108 + IotPluginCommonUtils.writeJsonResponse(routingContext, errorResponse);
  109 + }
  110 + }
  111 +
  112 + /**
  113 + * 更新设备状态
  114 + *
  115 + * @param productKey 产品Key
  116 + * @param deviceName 设备名称
  117 + */
  118 + private void updateDeviceState(String productKey, String deviceName) {
  119 + deviceUpstreamApi.updateDeviceState(((IotDeviceStateUpdateReqDTO) new IotDeviceStateUpdateReqDTO().setRequestId(IdUtil.fastSimpleUUID()).setProcessId(IotPluginCommonUtils.getProcessId()).setReportTime(LocalDateTime.now()).setProductKey(productKey).setDeviceName(deviceName)).setState(IotDeviceStateEnum.ONLINE.getState()));
  120 + }
  121 +
  122 + /**
  123 + * 解析属性上报请求
  124 + *
  125 + * @param productKey 产品Key
  126 + * @param deviceName 设备名称
  127 + * @param requestId 请求ID
  128 + * @param body 请求体
  129 + * @return 属性上报请求DTO
  130 + */
  131 + @SuppressWarnings("unchecked")
  132 + private IotDevicePropertyReportReqDTO parsePropertyReportRequest(String productKey, String deviceName, String requestId, JsonObject body) {
  133 +
  134 + // 按照标准JSON格式处理属性数据
  135 + Map<String, Object> properties = new HashMap<>();
  136 +
  137 + // 优先使用params字段，符合标准
  138 + Map<String, Object> params = body.getJsonObject("params") != null ? body.getJsonObject("params").getMap() : null;
  139 +
  140 + if (params != null) {
  141 + // 将标准格式的params转换为平台需要的properties格式
  142 + for (Map.Entry<String, Object> entry : params.entrySet()) {
  143 + String key = entry.getKey();
  144 + Object valueObj = entry.getValue();
  145 +
  146 + // 如果是复杂结构（包含value和time）
  147 + if (valueObj instanceof Map) {
  148 + Map<String, Object> valueMap = (Map<String, Object>) valueObj;
  149 + properties.put(key, valueMap.getOrDefault("value", valueObj));
  150 + } else {
  151 + properties.put(key, valueObj);
  152 + }
  153 + }
  154 + }
  155 +
  156 + return ((IotDevicePropertyReportReqDTO) new IotDevicePropertyReportReqDTO().setRequestId(requestId).setProcessId(IotPluginCommonUtils.getProcessId()).setReportTime(LocalDateTime.now()).setProductKey(productKey).setDeviceName(deviceName)).setProperties(properties);
  157 + }
  158 +
  159 + /**
  160 + * 解析事件上报请求
  161 + *
  162 + * @param productKey 产品Key
  163 + * @param deviceName 设备名称
  164 + * @param identifier 事件标识符
  165 + * @param requestId 请求ID
  166 + * @param body 请求体
  167 + * @return 事件上报请求DTO
  168 + */
  169 + @SuppressWarnings("unchecked")
  170 + private IotDeviceEventReportReqDTO parseEventReportRequest(String productKey, String deviceName, String identifier, String requestId, JsonObject body) {
  171 +
  172 + // 按照标准JSON格式处理事件参数
  173 + Map<String, Object> params;
  174 +
  175 + // 优先使用params字段，符合标准
  176 + if (body.getJsonObject("params") != null) {
  177 + params = body.getJsonObject("params").getMap();
  178 + } else {
  179 + // 兼容旧格式
  180 + params = new HashMap<>();
  181 + }
  182 +
  183 + return ((IotDeviceEventReportReqDTO) new IotDeviceEventReportReqDTO().setRequestId(requestId).setProcessId(IotPluginCommonUtils.getProcessId()).setReportTime(LocalDateTime.now()).setProductKey(productKey).setDeviceName(deviceName)).setIdentifier(identifier).setParams(params);
  184 + }
  185 +}
0 \ No newline at end of file 186 \ No newline at end of file