BJ_ZTEITS-Developers / parking-cloud-yudao

Commit c105ecea96e0ba864ba6e3100d6697cd66c480d2

Authored by YunaiV
1 parent d9a08094

feat（iot）：【协议改造】websocket 初步改造（100%）

Inline Side-by-side
Showing 6 changed files with 155 additions and 428 deletions
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayProperties.java
  View file @c105ece
... ... @@ -291,7 +291,7 @@ public class IotGatewayProperties {
291 291 */
292 292 private Integer keepAliveTimeoutSeconds = 300;
293 293  
294   - // TODO @AI：所有跟 ssl 相关的参数，是不是可以统一？放到 protocol 层级？ProtocolInstanceProperties【优先级：低】暂时不用规划；
  294 + // NOTE：SSL 相关参数后续统一到 protocol 层级（优先级低）
295 295 /**
296 296 * 是否启用 SSL
297 297 */
... ... @@ -336,7 +336,7 @@ public class IotGatewayProperties {
336 336  
337 337 }
338 338  
339   - // TODO @AI：【暂时忽略】改成 ProtocolProperties
  339 + // NOTE：暂未统一为 ProtocolProperties，待协议改造完成再调整
340 340 /**
341 341 * 协议实例配置
342 342 */
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/handler/upstream/IotTcpUpstreamHandler.java
  View file @c105ece
1 1 package cn.iocoder.yudao.module.iot.gateway.protocol.tcp.handler.upstream;
2 2  
3 3 import cn.hutool.core.util.BooleanUtil;
4   -import cn.hutool.core.util.IdUtil;
5 4 import cn.hutool.extra.spring.SpringUtil;
6 5 import cn.iocoder.yudao.framework.common.exception.ServiceException;
7 6 import cn.iocoder.yudao.framework.common.pojo.CommonResult;
... ... @@ -79,35 +78,32 @@ public class IotTcpUpstreamHandler implements Handler<NetSocket> {
79 78 @Override
80 79 @SuppressWarnings("DuplicatedCode")
81 80 public void handle(NetSocket socket) {
82   - // TODO @AI：clientId 去掉；其它模块也看看，怎么去掉下看看；
83   - String clientId = IdUtil.simpleUUID();
84   - log.debug("[handle][设备连接，客户端 ID: {}，地址: {}]", clientId, socket.remoteAddress());
  81 + String remoteAddress = String.valueOf(socket.remoteAddress());
  82 + log.debug("[handle][设备连接，地址: {}]", remoteAddress);
85 83  
86 84 // 1. 设置异常和关闭处理器
87 85 socket.exceptionHandler(ex -> {
88   - log.warn("[handle][连接异常，客户端 ID: {}，地址: {}]", clientId, socket.remoteAddress());
  86 + log.warn("[handle][连接异常，地址: {}]", remoteAddress, ex);
89 87 socket.close();
90 88 });
91 89 socket.closeHandler(v -> {
92   - log.debug("[handle][连接关闭，客户端 ID: {}，地址: {}]", clientId, socket.remoteAddress());
  90 + log.debug("[handle][连接关闭，地址: {}]", remoteAddress);
93 91 cleanupConnection(socket);
94 92 });
95 93  
96 94 // 2.1 设置消息处理器
97   - // TODO @AI：去掉 clientId；
98 95 Handler<Buffer> messageHandler = buffer -> {
99 96 try {
100 97 processMessage(buffer, socket);
101 98 } catch (Exception e) {
102   - log.error("[handle][消息处理失败，客户端 ID: {}，地址: {}]",
103   - clientId, socket.remoteAddress(), e);
  99 + log.error("[handle][消息处理失败，地址: {}]", remoteAddress, e);
104 100 socket.close();
105 101 }
106 102 };
107 103 // 2.2 使用拆包器处理粘包/拆包
108 104 RecordParser parser = codec.createDecodeParser(messageHandler);
109 105 socket.handler(parser);
110   - log.debug("[handle][启用 {} 拆包器，客户端 ID: {}]", codec.getType(), clientId);
  106 + log.debug("[handle][启用 {} 拆包器，地址: {}]", codec.getType(), remoteAddress);
111 107 }
112 108  
113 109 /**
... ... @@ -135,23 +131,23 @@ public class IotTcpUpstreamHandler implements Handler&lt;NetSocket&gt; {
135 131 handleRegisterRequest(message, socket);
136 132 } else {
137 133 // 业务消息
138   - handleBusinessRequest(null, message, socket);
  134 + handleBusinessRequest(message, socket);
139 135 }
140 136 } catch (ServiceException e) {
141 137 // 业务异常，返回对应的错误码和错误信息
142   - log.warn("[processMessage][业务异常，客户端 ID: {}，错误: {}]", null, e.getMessage());
  138 + log.warn("[processMessage][业务异常，地址: {}，错误: {}]", socket.remoteAddress(), e.getMessage());
143 139 String requestId = message != null ? message.getRequestId() : null;
144 140 String method = message != null ? message.getMethod() : null;
145 141 sendErrorResponse(socket, requestId, method, e.getCode(), e.getMessage());
146 142 } catch (IllegalArgumentException e) {
147 143 // 参数校验失败，返回 400
148   - log.warn("[processMessage][参数校验失败，客户端 ID: {}，错误: {}]", null, e.getMessage());
  144 + log.warn("[processMessage][参数校验失败，地址: {}，错误: {}]", socket.remoteAddress(), e.getMessage());
149 145 String requestId = message != null ? message.getRequestId() : null;
150 146 String method = message != null ? message.getMethod() : null;
151 147 sendErrorResponse(socket, requestId, method, BAD_REQUEST.getCode(), e.getMessage());
152 148 } catch (Exception e) {
153 149 // 其他异常，返回 500，并重新抛出让上层关闭连接
154   - log.error("[processMessage][处理消息失败，客户端 ID: {}]", null, e);
  150 + log.error("[processMessage][处理消息失败，地址: {}]", socket.remoteAddress(), e);
155 151 String requestId = message != null ? message.getRequestId() : null;
156 152 String method = message != null ? message.getMethod() : null;
157 153 sendErrorResponse(socket, requestId, method,
... ... @@ -218,21 +214,21 @@ public class IotTcpUpstreamHandler implements Handler&lt;NetSocket&gt; {
218 214 // 3. 发送成功响应
219 215 sendSuccessResponse(socket, message.getRequestId(),
220 216 IotDeviceMessageMethodEnum.DEVICE_REGISTER.getMethod(), result.getData());
221   - log.info("[handleRegisterRequest][注册成功，客户端 ID: {}，设备名: {}]", null, params.getDeviceName());
  217 + log.info("[handleRegisterRequest][注册成功，地址: {}，设备名: {}]",
  218 + socket.remoteAddress(), params.getDeviceName());
222 219 }
223 220  
224 221 /**
225 222 * 处理业务请求
226 223 *
227   - * @param clientId 客户端 ID
228 224 * @param message 消息信息
229 225 * @param socket 网络连接
230 226 */
231   - private void handleBusinessRequest(String clientId, IotDeviceMessage message, NetSocket socket) {
  227 + private void handleBusinessRequest(IotDeviceMessage message, NetSocket socket) {
232 228 // 1. 获取认证信息并处理业务消息
233 229 IotTcpConnectionManager.ConnectionInfo connectionInfo = connectionManager.getConnectionInfo(socket);
234 230 if (connectionInfo == null) {
235   - log.error("[handleBusinessRequest][无法获取连接信息，客户端 ID: {}]", clientId);
  231 + log.error("[handleBusinessRequest][无法获取连接信息，地址: {}]", socket.remoteAddress());
236 232 sendErrorResponse(socket, message.getRequestId(), message.getMethod(),
237 233 UNAUTHORIZED.getCode(), "设备未认证，无法处理业务消息");
238 234 return;
... ... @@ -241,7 +237,7 @@ public class IotTcpUpstreamHandler implements Handler&lt;NetSocket&gt; {
241 237 // 2. 发送消息到消息总线
242 238 deviceMessageService.sendDeviceMessage(message, connectionInfo.getProductKey(),
243 239 connectionInfo.getDeviceName(), serverId);
244   - log.info("[handleBusinessRequest][发送消息到消息总线，客户端 ID: {}，消息: {}", clientId, message);
  240 + log.info("[handleBusinessRequest][发送消息到消息总线，地址: {}，消息: {}]", socket.remoteAddress(), message);
245 241 }
246 242  
247 243 /**
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/websocket/IotWebSocketProtocol.java
  View file @c105ece
... ... @@ -9,13 +9,11 @@ import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
9 9 import cn.iocoder.yudao.module.iot.gateway.config.IotGatewayProperties.ProtocolInstanceProperties;
10 10 import cn.iocoder.yudao.module.iot.gateway.protocol.IotProtocol;
11 11 import cn.iocoder.yudao.module.iot.gateway.protocol.websocket.handler.downstream.IotWebSocketDownstreamSubscriber;
12   -import cn.iocoder.yudao.module.iot.gateway.protocol.websocket.manager.IotWebSocketConnectionManager;
13 12 import cn.iocoder.yudao.module.iot.gateway.protocol.websocket.handler.downstream.IotWebSocketDownstreamHandler;
14 13 import cn.iocoder.yudao.module.iot.gateway.protocol.websocket.handler.upstream.IotWebSocketUpstreamHandler;
  14 +import cn.iocoder.yudao.module.iot.gateway.protocol.websocket.manager.IotWebSocketConnectionManager;
15 15 import cn.iocoder.yudao.module.iot.gateway.serialize.IotMessageSerializer;
16 16 import cn.iocoder.yudao.module.iot.gateway.serialize.IotMessageSerializerManager;
17   -import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService;
18   -import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
19 17 import io.vertx.core.Vertx;
20 18 import io.vertx.core.http.HttpServer;
21 19 import io.vertx.core.http.HttpServerOptions;
... ... @@ -24,9 +22,10 @@ import lombok.Getter;
24 22 import lombok.extern.slf4j.Slf4j;
25 23 import org.springframework.util.Assert;
26 24  
27   -// TODO @AI：注释调整下，参考 IotTcpProtocol
28 25 /**
29   - * IoT 网关 WebSocket 协议：接收设备上行消息
  26 + * IoT WebSocket 协议实现
  27 + * <p>
  28 + * 基于 Vert.x 实现 WebSocket 服务器，接收设备上行消息
30 29 *
31 30 * @author 芋道源码
32 31 */
... ... @@ -62,12 +61,6 @@ public class IotWebSocketProtocol implements IotProtocol {
62 61 */
63 62 private final IotWebSocketConnectionManager connectionManager;
64 63  
65   - // TODO @AI：可以不用这个变量，从 properties 里面获取
66   - /**
67   - * WebSocket 配置
68   - */
69   - private final IotWebSocketConfig wsConfig;
70   -
71 64 /**
72 65 * 下行消息订阅者
73 66 */
... ... @@ -78,14 +71,10 @@ public class IotWebSocketProtocol implements IotProtocol {
78 71 */
79 72 private final IotMessageSerializer serializer;
80 73  
81   - private final IotDeviceService deviceService;
82   - private final IotDeviceMessageService messageService;
83   -
84 74 public IotWebSocketProtocol(ProtocolInstanceProperties properties) {
85 75 Assert.notNull(properties, "协议实例配置不能为空");
86 76 Assert.notNull(properties.getWebsocket(), "WebSocket 协议配置（websocket）不能为空");
87 77 this.properties = properties;
88   - this.wsConfig = properties.getWebsocket();
89 78 this.serverId = IotDeviceMessageUtils.generateServerId(properties.getPort());
90 79  
91 80 // 初始化序列化器
... ... @@ -94,9 +83,7 @@ public class IotWebSocketProtocol implements IotProtocol {
94 83 IotMessageSerializerManager serializerManager = SpringUtil.getBean(IotMessageSerializerManager.class);
95 84 this.serializer = serializerManager.get(serializeType);
96 85  
97   - // 初始化基础依赖
98   - this.deviceService = SpringUtil.getBean(IotDeviceService.class);
99   - this.messageService = SpringUtil.getBean(IotDeviceMessageService.class);
  86 + // 初始化连接管理器
100 87 this.connectionManager = new IotWebSocketConnectionManager();
101 88  
102 89 // 初始化下行消息订阅者
... ... @@ -127,6 +114,7 @@ public class IotWebSocketProtocol implements IotProtocol {
127 114 this.vertx = Vertx.vertx();
128 115  
129 116 // 1.2 创建服务器选项
  117 + IotWebSocketConfig wsConfig = properties.getWebsocket();
130 118 HttpServerOptions options = new HttpServerOptions()
131 119 .setPort(properties.getPort())
132 120 .setIdleTimeout(wsConfig.getIdleTimeoutSeconds())
... ... @@ -150,8 +138,7 @@ public class IotWebSocketProtocol implements IotProtocol {
150 138 return;
151 139 }
152 140 // 创建上行处理器
153   - IotWebSocketUpstreamHandler handler = new IotWebSocketUpstreamHandler(this,
154   - messageService, deviceService, connectionManager, serializer);
  141 + IotWebSocketUpstreamHandler handler = new IotWebSocketUpstreamHandler(serverId, serializer, connectionManager);
155 142 handler.handle(socket);
156 143 });
157 144  
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/websocket/handler/downstream/IotWebSocketDownstreamHandler.java
  View file @c105ece
... ... @@ -2,7 +2,6 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.websocket.handler.downstrea
2 2  
3 3 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
4 4 import cn.hutool.core.util.StrUtil;
5   -import cn.iocoder.yudao.module.iot.core.enums.IotSerializeTypeEnum;
6 5 import cn.iocoder.yudao.module.iot.gateway.protocol.websocket.manager.IotWebSocketConnectionManager;
7 6 import cn.iocoder.yudao.module.iot.gateway.serialize.IotMessageSerializer;
8 7 import lombok.RequiredArgsConstructor;
... ... @@ -37,23 +36,17 @@ public class IotWebSocketDownstreamHandler {
37 36 return;
38 37 }
39 38  
40   - // 2. 编码消息并发送到设备
  39 + // 2. 序列化
41 40 byte[] bytes = serializer.serialize(message);
42   - // TODO @AI：参考别的模块的做法，直接发？类似 tcp 这种；
43   - boolean success;
44   - if (serializer.getType() == IotSerializeTypeEnum.BINARY) {
45   - success = connectionManager.sendToDevice(message.getDeviceId(), bytes);
46   - } else {
47   - String jsonMessage = StrUtil.utf8Str(bytes);
48   - success = connectionManager.sendToDevice(message.getDeviceId(), jsonMessage);
49   - }
50   - if (success) {
51   - log.info("[handle][下行消息发送成功，设备 ID: {}，方法: {}，消息 ID: {}，数据长度: {} 字节]",
52   - message.getDeviceId(), message.getMethod(), message.getId(), bytes.length);
53   - } else {
54   - log.error("[handle][下行消息发送失败，设备 ID: {}，方法: {}，消息 ID: {}]",
55   - message.getDeviceId(), message.getMethod(), message.getId());
  41 + String bytesContent = StrUtil.utf8Str(bytes);
  42 +
  43 + // 3. 发送到设备
  44 + boolean success = connectionManager.sendToDevice(connectionInfo.getDeviceId(), bytesContent);
  45 + if (!success) {
  46 + throw new RuntimeException("下行消息发送失败");
56 47 }
  48 + log.info("[handle][下行消息发送成功，设备 ID: {}，方法: {}，消息 ID: {}，数据长度: {} 字节]",
  49 + message.getDeviceId(), message.getMethod(), message.getId(), bytes.length);
57 50 } catch (Exception e) {
58 51 log.error("[handle][处理下行消息失败，设备 ID: {}，方法: {}，消息内容: {}]",
59 52 message.getDeviceId(), message.getMethod(), message, e);
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/websocket/handler/upstream/IotWebSocketUpstreamHandler.java
  View file @c105ece
1 1 package cn.iocoder.yudao.module.iot.gateway.protocol.websocket.handler.upstream;
2 2  
3   -import cn.hutool.core.map.MapUtil;
4 3 import cn.hutool.core.util.ArrayUtil;
5 4 import cn.hutool.core.util.BooleanUtil;
6   -import cn.hutool.core.util.IdUtil;
7 5 import cn.hutool.core.util.StrUtil;
8 6 import cn.hutool.extra.spring.SpringUtil;
  7 +import cn.iocoder.yudao.framework.common.exception.ServiceException;
9 8 import cn.iocoder.yudao.framework.common.pojo.CommonResult;
10 9 import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
11 10 import cn.iocoder.yudao.module.iot.core.biz.IotDeviceCommonApi;
12 11 import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceAuthReqDTO;
13 12 import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO;
14 13 import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
15   -import cn.iocoder.yudao.module.iot.core.enums.IotSerializeTypeEnum;
16 14 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
17 15 import cn.iocoder.yudao.module.iot.core.topic.IotDeviceIdentity;
18 16 import cn.iocoder.yudao.module.iot.core.topic.auth.IotDeviceRegisterReqDTO;
19 17 import cn.iocoder.yudao.module.iot.core.topic.auth.IotDeviceRegisterRespDTO;
20 18 import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils;
21   -import cn.iocoder.yudao.module.iot.gateway.protocol.websocket.IotWebSocketProtocol;
22 19 import cn.iocoder.yudao.module.iot.gateway.protocol.websocket.manager.IotWebSocketConnectionManager;
23 20 import cn.iocoder.yudao.module.iot.gateway.serialize.IotMessageSerializer;
24 21 import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService;
25 22 import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
26 23 import io.vertx.core.Handler;
27   -import io.vertx.core.buffer.Buffer;
28 24 import io.vertx.core.http.ServerWebSocket;
29 25 import lombok.extern.slf4j.Slf4j;
  26 +import org.springframework.util.Assert;
30 27  
31   -import java.util.Map;
  28 +import static cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants.*;
  29 +import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
  30 +import static cn.iocoder.yudao.module.iot.gateway.enums.ErrorCodeConstants.DEVICE_AUTH_FAIL;
32 31  
33 32  
34 33 /**
... ... @@ -52,256 +51,175 @@ public class IotWebSocketUpstreamHandler implements Handler&lt;ServerWebSocket&gt; {
52 51 */
53 52 private final IotWebSocketConnectionManager connectionManager;
54 53  
55   - // TODO @AI：是不是可以去掉？
56   - private final boolean binaryPayload;
57   -
58 54 private final IotDeviceMessageService deviceMessageService;
59 55 private final IotDeviceService deviceService;
60 56 private final IotDeviceCommonApi deviceApi;
61 57  
62   - // TODO @AI：参数、顺序参考 IotTcpUpstreamHandler
63   - public IotWebSocketUpstreamHandler(IotWebSocketProtocol protocol,
64   - IotDeviceMessageService deviceMessageService,
65   - IotDeviceService deviceService,
66   - IotWebSocketConnectionManager connectionManager,
67   - IotMessageSerializer serializer) {
68   - this.deviceMessageService = deviceMessageService;
69   - this.deviceService = deviceService;
70   - this.connectionManager = connectionManager;
  58 + public IotWebSocketUpstreamHandler(String serverId,
  59 + IotMessageSerializer serializer,
  60 + IotWebSocketConnectionManager connectionManager) {
  61 + this.serverId = serverId;
71 62 this.serializer = serializer;
72   - this.binaryPayload = serializer.getType() == IotSerializeTypeEnum.BINARY;
  63 + this.connectionManager = connectionManager;
  64 + this.deviceMessageService = SpringUtil.getBean(IotDeviceMessageService.class);
  65 + this.deviceService = SpringUtil.getBean(IotDeviceService.class);
73 66 this.deviceApi = SpringUtil.getBean(IotDeviceCommonApi.class);
74   - this.serverId = protocol.getServerId();
75   - // TODO @AI：通过 springutil；deviceService、deviceMessageService；
76 67 }
77 68  
78 69 @Override
79 70 @SuppressWarnings("DuplicatedCode")
80 71 public void handle(ServerWebSocket socket) {
81   - String clientId = IdUtil.simpleUUID();
82   - log.debug("[handle][设备连接，客户端 ID: {}，地址: {}]", clientId, socket.remoteAddress());
  72 + String remoteAddress = String.valueOf(socket.remoteAddress());
  73 + log.debug("[handle][设备连接，地址: {}]", remoteAddress);
83 74  
84 75 // 1. 设置异常和关闭处理器
85   - // TODO @AI：clientId 去掉；
86 76 socket.exceptionHandler(ex -> {
87   - log.warn("[handle][连接异常，客户端 ID: {}，地址: {}]", clientId, socket.remoteAddress());
  77 + log.warn("[handle][连接异常，地址: {}]", remoteAddress, ex);
88 78 socket.close();
89 79 });
90 80 socket.closeHandler(v -> {
91   - log.debug("[handle][连接关闭，客户端 ID: {}，地址: {}]", clientId, socket.remoteAddress());
  81 + log.debug("[handle][连接关闭，地址: {}]", remoteAddress);
92 82 cleanupConnection(socket);
93 83 });
94 84  
95   - // 2. 设置消息处理器（JSON 使用文本，BINARY 使用二进制）
96   - // TODO @AI：是不是 text、binary 保持统一？用一个 mesagehandler？
97   - if (binaryPayload) {
98   - socket.binaryMessageHandler(buffer -> {
99   - try {
100   - processMessage(clientId, buffer.getBytes(), socket);
101   - } catch (Exception e) {
102   - log.error("[handle][消息解码失败，断开连接，客户端 ID: {}，地址: {}，错误: {}]",
103   - clientId, socket.remoteAddress(), e.getMessage());
104   - cleanupConnection(socket);
105   - socket.close();
106   - }
107   - });
108   - socket.textMessageHandler(message -> {
109   - log.warn("[handle][收到文本帧但当前序列化为 BINARY，断开连接，客户端 ID: {}，地址: {}]",
110   - clientId, socket.remoteAddress());
111   - cleanupConnection(socket);
  85 + // 2. 设置消息处理器（仅支持文本帧）
  86 + socket.textMessageHandler(message -> {
  87 + try {
  88 + processMessage(StrUtil.utf8Bytes(message), socket);
  89 + } catch (Exception e) {
  90 + log.error("[handle][消息解码失败，断开连接，地址: {}，错误: {}]", remoteAddress, e.getMessage());
112 91 socket.close();
113   - });
114   - } else {
115   - socket.textMessageHandler(message -> {
116   - try {
117   - processMessage(clientId, StrUtil.utf8Bytes(message), socket);
118   - } catch (Exception e) {
119   - log.error("[handle][消息解码失败，断开连接，客户端 ID: {}，地址: {}，错误: {}]",
120   - clientId, socket.remoteAddress(), e.getMessage());
121   - // TODO @AI：是不是不用 cleanupConnection？closehandler 本身就吹了了；
122   - cleanupConnection(socket);
123   - socket.close();
124   - }
125   - });
126   - socket.binaryMessageHandler(buffer -> {
127   - try {
128   - processMessage(clientId, buffer.getBytes(), socket);
129   - } catch (Exception e) {
130   - log.error("[handle][消息解码失败，断开连接，客户端 ID: {}，地址: {}，错误: {}]",
131   - clientId, socket.remoteAddress(), e.getMessage());
132   - cleanupConnection(socket);
133   - socket.close();
134   - }
135   - });
136   - }
  92 + }
  93 + });
137 94 }
138 95  
139 96 /**
140 97 * 处理消息
141 98 *
142   - * @param clientId 客户端 ID
143 99 * @param payload 消息负载
144 100 * @param socket WebSocket 连接
145   - * @throws Exception 消息解码失败时抛出异常
146 101 */
147   - private void processMessage(String clientId, byte[] payload, ServerWebSocket socket) throws Exception {
148   - // 1.1 基础检查
149   - if (ArrayUtil.isEmpty(payload)) {
150   - return;
151   - }
152   - // 1.2 解码消息
153   - IotDeviceMessage deviceMessage;
  102 + private void processMessage(byte[] payload, ServerWebSocket socket) {
  103 + IotDeviceMessage message = null;
154 104 try {
155   - deviceMessage = serializer.deserialize(payload);
156   - if (deviceMessage == null) {
157   - throw new Exception("解码后消息为空");
  105 + // 1.1 基础检查
  106 + if (ArrayUtil.isEmpty(payload)) {
  107 + return;
158 108 }
159   - } catch (Exception e) {
160   - throw new Exception("消息解码失败: " + e.getMessage(), e);
161   - }
162   -
163   - // 2. 根据消息类型路由处理
164   - try {
165   - if (AUTH_METHOD.equals(deviceMessage.getMethod())) {
166   - // 认证请求
167   - handleAuthenticationRequest(clientId, deviceMessage, socket);
168   - } else if (IotDeviceMessageMethodEnum.DEVICE_REGISTER.getMethod().equals(deviceMessage.getMethod())) {
169   - // 设备动态注册请求
170   - handleRegisterRequest(clientId, deviceMessage, socket);
  109 + // 1.2 解码消息
  110 + message = serializer.deserialize(payload);
  111 + Assert.notNull(message, "消息反序列化失败");
  112 + Assert.hasText(message.getMethod(), "method 不能为空");
  113 +
  114 + // 2. 根据消息类型路由处理
  115 + if (AUTH_METHOD.equals(message.getMethod())) {
  116 + handleAuthenticationRequest(message, socket);
  117 + } else if (IotDeviceMessageMethodEnum.DEVICE_REGISTER.getMethod().equals(message.getMethod())) {
  118 + handleRegisterRequest(message, socket);
171 119 } else {
172   - // 业务消息
173   - handleBusinessRequest(clientId, deviceMessage, socket);
  120 + handleBusinessRequest(message, socket);
174 121 }
  122 + } catch (ServiceException e) {
  123 + log.warn("[processMessage][业务异常，错误: {}]", e.getMessage());
  124 + String requestId = message != null ? message.getRequestId() : null;
  125 + String method = message != null ? message.getMethod() : null;
  126 + sendErrorResponse(socket, requestId, method, e.getCode(), e.getMessage());
  127 + } catch (IllegalArgumentException e) {
  128 + log.warn("[processMessage][参数校验失败，错误: {}]", e.getMessage());
  129 + String requestId = message != null ? message.getRequestId() : null;
  130 + String method = message != null ? message.getMethod() : null;
  131 + sendErrorResponse(socket, requestId, method, BAD_REQUEST.getCode(), e.getMessage());
175 132 } catch (Exception e) {
176   - // TODO @AI：参考 IotTcpUpstreamHandler 处理；业务、参数、其它
177   - log.error("[processMessage][处理消息失败，客户端 ID: {}，消息方法: {}]",
178   - clientId, deviceMessage.getMethod(), e);
179   - // 发送错误响应，避免客户端一直等待
180   - try {
181   - sendErrorResponse(socket, deviceMessage.getRequestId(), "消息处理失败");
182   - } catch (Exception responseEx) {
183   - log.error("[processMessage][发送错误响应失败，客户端 ID: {}]", clientId, responseEx);
184   - }
  133 + log.error("[processMessage][处理消息失败]", e);
  134 + String requestId = message != null ? message.getRequestId() : null;
  135 + String method = message != null ? message.getMethod() : null;
  136 + sendErrorResponse(socket, requestId, method, INTERNAL_SERVER_ERROR.getCode(),
  137 + INTERNAL_SERVER_ERROR.getMsg());
  138 + throw e;
185 139 }
186 140 }
187 141  
188 142 /**
189 143 * 处理认证请求
190 144 *
191   - * @param clientId 客户端 ID
192 145 * @param message 消息信息
193 146 * @param socket WebSocket 连接
194 147 */
195   - private void handleAuthenticationRequest(String clientId, IotDeviceMessage message, ServerWebSocket socket) {
196   - try {
197   - // 1.1 解析认证参数
198   - // TODO @AI：参数解析；参考 tcp 对应的 handleAuthenticationRequest
199   - IotDeviceAuthReqDTO authParams = parseAuthParams(message.getParams());
200   - if (authParams == null) {
201   - log.warn("[handleAuthenticationRequest][认证参数解析失败，客户端 ID: {}]", clientId);
202   - sendErrorResponse(socket, message.getRequestId(), "认证参数不完整");
203   - return;
204   - }
205   - // 1.2 执行认证
206   - if (!validateDeviceAuth(authParams)) {
207   - log.warn("[handleAuthenticationRequest][认证失败，客户端 ID: {}，username: {}]",
208   - clientId, authParams.getUsername());
209   - sendErrorResponse(socket, message.getRequestId(), "认证失败");
210   - return;
211   - }
212   -
213   - // 2.1 解析设备信息
214   - IotDeviceIdentity deviceInfo = IotDeviceAuthUtils.parseUsername(authParams.getUsername());
215   - if (deviceInfo == null) {
216   - sendErrorResponse(socket, message.getRequestId(), "解析设备信息失败");
217   - return;
218   - }
219   - // 2.2 获取设备信息
220   - IotDeviceRespDTO device = deviceService.getDeviceFromCache(deviceInfo.getProductKey(),
221   - deviceInfo.getDeviceName());
222   - if (device == null) {
223   - sendErrorResponse(socket, message.getRequestId(), "设备不存在");
224   - return;
225   - }
226   -
227   - // 3.1 注册连接
228   - registerConnection(socket, device, clientId);
229   - // 3.2 发送上线消息
230   - sendOnlineMessage(device);
231   - // 3.3 发送成功响应
232   - sendSuccessResponse(socket, message.getRequestId(), "认证成功");
233   - log.info("[handleAuthenticationRequest][认证成功，设备 ID: {}，设备名: {}]",
234   - device.getId(), device.getDeviceName());
235   - } catch (Exception e) {
236   - log.error("[handleAuthenticationRequest][认证处理异常，客户端 ID: {}]", clientId, e);
237   - sendErrorResponse(socket, message.getRequestId(), "认证处理异常");
  148 + @SuppressWarnings("DuplicatedCode")
  149 + private void handleAuthenticationRequest(IotDeviceMessage message, ServerWebSocket socket) {
  150 + // 1. 解析认证参数
  151 + IotDeviceAuthReqDTO authParams = JsonUtils.convertObject(message.getParams(), IotDeviceAuthReqDTO.class);
  152 + Assert.notNull(authParams, "认证参数不能为空");
  153 + Assert.hasText(authParams.getUsername(), "username 不能为空");
  154 + Assert.hasText(authParams.getPassword(), "password 不能为空");
  155 +
  156 + // 2.1 执行认证
  157 + CommonResult<Boolean> authResult = deviceApi.authDevice(authParams);
  158 + authResult.checkError();
  159 + if (BooleanUtil.isFalse(authResult.getData())) {
  160 + throw exception(DEVICE_AUTH_FAIL);
238 161 }
  162 + // 2.2 解析设备信息
  163 + IotDeviceIdentity deviceInfo = IotDeviceAuthUtils.parseUsername(authParams.getUsername());
  164 + Assert.notNull(deviceInfo, "解析设备信息失败");
  165 + // 2.3 获取设备信息
  166 + IotDeviceRespDTO device = deviceService.getDeviceFromCache(deviceInfo.getProductKey(), deviceInfo.getDeviceName());
  167 + Assert.notNull(device, "设备不存在");
  168 +
  169 + // 3.1 注册连接
  170 + registerConnection(socket, device);
  171 + // 3.2 发送上线消息
  172 + sendOnlineMessage(device);
  173 + // 3.3 发送成功响应
  174 + sendSuccessResponse(socket, message.getRequestId(), AUTH_METHOD, "认证成功");
  175 + log.info("[handleAuthenticationRequest][认证成功，设备 ID: {}，设备名: {}]", device.getId(), device.getDeviceName());
239 176 }
240 177  
241 178 /**
242 179 * 处理设备动态注册请求（一型一密，不需要认证）
243 180 *
244   - * @param clientId 客户端 ID
245 181 * @param message 消息信息
246 182 * @param socket WebSocket 连接
247 183 * @see <a href="https://help.aliyun.com/zh/iot/user-guide/unique-certificate-per-product-verification">阿里云 - 一型一密</a>
248 184 */
249   - private void handleRegisterRequest(String clientId, IotDeviceMessage message, ServerWebSocket socket) {
250   - // TODO @AI：参数解析；参考 tcp 对应的 handleRegisterRequest
251   - try {
252   - // 1. 解析注册参数
253   - IotDeviceRegisterReqDTO params = parseRegisterParams(message.getParams());
254   - if (params == null
255   - || StrUtil.hasEmpty(params.getProductKey(), params.getDeviceName(), params.getProductSecret())) {
256   - log.warn("[handleRegisterRequest][注册参数解析失败，客户端 ID: {}]", clientId);
257   - sendErrorResponse(socket, message.getRequestId(), "注册参数不完整");
258   - return;
259   - }
260   -
261   - // 2. 调用动态注册
262   - CommonResult<IotDeviceRegisterRespDTO> result = deviceApi.registerDevice(params);
263   - if (result.isError()) {
264   - log.warn("[handleRegisterRequest][注册失败，客户端 ID: {}，错误: {}]", clientId, result.getMsg());
265   - sendErrorResponse(socket, message.getRequestId(), result.getMsg());
266   - return;
267   - }
268   -
269   - // 3. 发送成功响应（包含 deviceSecret）
270   - sendRegisterSuccessResponse(socket, message.getRequestId(), result.getData());
271   - log.info("[handleRegisterRequest][注册成功，客户端 ID: {}，设备名: {}]",
272   - clientId, params.getDeviceName());
273   - } catch (Exception e) {
274   - log.error("[handleRegisterRequest][注册处理异常，客户端 ID: {}]", clientId, e);
275   - sendErrorResponse(socket, message.getRequestId(), "注册处理异常");
276   - }
  185 + @SuppressWarnings("DuplicatedCode")
  186 + private void handleRegisterRequest(IotDeviceMessage message, ServerWebSocket socket) {
  187 + // 1. 解析注册参数
  188 + IotDeviceRegisterReqDTO params = JsonUtils.convertObject(message.getParams(), IotDeviceRegisterReqDTO.class);
  189 + Assert.notNull(params, "注册参数不能为空");
  190 + Assert.hasText(params.getProductKey(), "productKey 不能为空");
  191 + Assert.hasText(params.getDeviceName(), "deviceName 不能为空");
  192 +
  193 + // 2. 调用动态注册
  194 + CommonResult<IotDeviceRegisterRespDTO> result = deviceApi.registerDevice(params);
  195 + result.checkError();
  196 +
  197 + // 3. 发送成功响应（包含 deviceSecret）
  198 + sendSuccessResponse(socket, message.getRequestId(),
  199 + IotDeviceMessageMethodEnum.DEVICE_REGISTER.getMethod(), result.getData());
  200 + log.info("[handleRegisterRequest][注册成功，设备名: {}]", params.getDeviceName());
277 201 }
278 202  
279   - // TODO @AI：参考对应的 tcp 的 handleBusinessRequest
280 203 /**
281 204 * 处理业务请求
282 205 *
283   - * @param clientId 客户端 ID
284 206 * @param message 消息信息
285 207 * @param socket WebSocket 连接
286 208 */
287   - private void handleBusinessRequest(String clientId, IotDeviceMessage message, ServerWebSocket socket) {
288   - try {
289   - // 1. 获取认证信息并处理业务消息
290   - IotWebSocketConnectionManager.ConnectionInfo connectionInfo = connectionManager.getConnectionInfo(socket);
291   - if (connectionInfo == null) {
292   - log.warn("[handleBusinessRequest][连接未认证，拒绝处理业务消息，客户端 ID: {}]", clientId);
293   - sendErrorResponse(socket, message.getRequestId(), "连接未认证");
294   - return;
295   - }
296   -
297   - // 2. 发送消息到消息总线
298   - deviceMessageService.sendDeviceMessage(message, connectionInfo.getProductKey(),
299   - connectionInfo.getDeviceName(), serverId);
300   - log.info("[handleBusinessRequest][发送消息到消息总线，客户端 ID: {}，消息: {}",
301   - clientId, message.toString());
302   - } catch (Exception e) {
303   - log.error("[handleBusinessRequest][业务请求处理异常，客户端 ID: {}]", clientId, e);
  209 + private void handleBusinessRequest(IotDeviceMessage message, ServerWebSocket socket) {
  210 + // 1. 获取认证信息并处理业务消息
  211 + IotWebSocketConnectionManager.ConnectionInfo connectionInfo = connectionManager.getConnectionInfo(socket);
  212 + if (connectionInfo == null) {
  213 + log.warn("[handleBusinessRequest][连接未认证，拒绝处理业务消息]");
  214 + sendErrorResponse(socket, message.getRequestId(), message.getMethod(),
  215 + UNAUTHORIZED.getCode(), "设备未认证，无法处理业务消息");
  216 + return;
304 217 }
  218 +
  219 + // 2. 发送消息到消息总线
  220 + deviceMessageService.sendDeviceMessage(message, connectionInfo.getProductKey(),
  221 + connectionInfo.getDeviceName(), serverId);
  222 + log.info("[handleBusinessRequest][发送消息到消息总线，消息: {}]", message);
305 223 }
306 224  
307 225 /**
... ... @@ -309,9 +227,8 @@ public class IotWebSocketUpstreamHandler implements Handler&lt;ServerWebSocket&gt; {
309 227 *
310 228 * @param socket WebSocket 连接
311 229 * @param device 设备
312   - * @param clientId 客户端 ID
313 230 */
314   - private void registerConnection(ServerWebSocket socket, IotDeviceRespDTO device, String clientId) {
  231 + private void registerConnection(ServerWebSocket socket, IotDeviceRespDTO device) {
315 232 IotWebSocketConnectionManager.ConnectionInfo connectionInfo = new IotWebSocketConnectionManager.ConnectionInfo()
316 233 .setDeviceId(device.getId())
317 234 .setProductKey(device.getProductKey())
... ... @@ -362,152 +279,18 @@ public class IotWebSocketUpstreamHandler implements Handler&lt;ServerWebSocket&gt; {
362 279 * 发送响应消息
363 280 *
364 281 * @param socket WebSocket 连接
365   - * @param success 是否成功
366   - * @param message 消息
367 282 * @param requestId 请求 ID
  283 + * @param method 请求方法
  284 + * @param data 响应数据
368 285 */
369   - private void sendResponse(ServerWebSocket socket, boolean success, String message, String requestId) {
370   - try {
371   - Object responseData = MapUtil.builder()
372   - .put("success", success)
373   - .put("message", message)
374   - .build();
375   -
376   - int code = success ? 0 : 401;
377   - IotDeviceMessage responseMessage = IotDeviceMessage.replyOf(requestId, AUTH_METHOD, responseData, code, message);
378   -
379   - writeResponse(socket, responseMessage);
380   - } catch (Exception e) {
381   - log.error("[sendResponse][发送响应失败，requestId: {}]", requestId, e);
382   - }
  286 + private void sendSuccessResponse(ServerWebSocket socket, String requestId, String method, Object data) {
  287 + IotDeviceMessage responseMessage = IotDeviceMessage.replyOf(requestId, method, data, SUCCESS.getCode(), null);
  288 + writeResponse(socket, responseMessage);
383 289 }
384 290  
385   - /**
386   - * 验证设备认证信息
387   - *
388   - * @param authParams 认证参数
389   - * @return 是否认证成功
390   - */
391   - private boolean validateDeviceAuth(IotDeviceAuthReqDTO authParams) {
392   - try {
393   - CommonResult<Boolean> result = deviceApi.authDevice(new IotDeviceAuthReqDTO()
394   - .setClientId(authParams.getClientId()).setUsername(authParams.getUsername())
395   - .setPassword(authParams.getPassword()));
396   - result.checkError();
397   - return BooleanUtil.isTrue(result.getData());
398   - } catch (Exception e) {
399   - log.error("[validateDeviceAuth][设备认证异常，username: {}]", authParams.getUsername(), e);
400   - return false;
401   - }
402   - }
403   -
404   - /**
405   - * 发送错误响应
406   - *
407   - * @param socket WebSocket 连接
408   - * @param requestId 请求 ID
409   - * @param errorMessage 错误消息
410   - */
411   - private void sendErrorResponse(ServerWebSocket socket, String requestId, String errorMessage) {
412   - sendResponse(socket, false, errorMessage, requestId);
413   - }
414   -
415   - /**
416   - * 发送成功响应
417   - *
418   - * @param socket WebSocket 连接
419   - * @param requestId 请求 ID
420   - * @param message 消息
421   - */
422   - @SuppressWarnings("SameParameterValue")
423   - private void sendSuccessResponse(ServerWebSocket socket, String requestId, String message) {
424   - sendResponse(socket, true, message, requestId);
425   - }
426   -
427   - /**
428   - * 解析认证参数
429   - *
430   - * @param params 参数对象（通常为 Map 类型）
431   - * @return 认证参数 DTO，解析失败时返回 null
432   - */
433   - @SuppressWarnings({"unchecked", "DuplicatedCode"})
434   - private IotDeviceAuthReqDTO parseAuthParams(Object params) {
435   - if (params == null) {
436   - return null;
437   - }
438   - try {
439   - // 参数默认为 Map 类型，直接转换
440   - if (params instanceof Map) {
441   - Map<String, Object> paramMap = (Map<String, Object>) params;
442   - return new IotDeviceAuthReqDTO()
443   - .setClientId(MapUtil.getStr(paramMap, "clientId"))
444   - .setUsername(MapUtil.getStr(paramMap, "username"))
445   - .setPassword(MapUtil.getStr(paramMap, "password"));
446   - }
447   - // 如果已经是目标类型，直接返回
448   - if (params instanceof IotDeviceAuthReqDTO) {
449   - return (IotDeviceAuthReqDTO) params;
450   - }
451   -
452   - // 其他情况尝试 JSON 转换
453   - return JsonUtils.convertObject(params, IotDeviceAuthReqDTO.class);
454   - } catch (Exception e) {
455   - log.error("[parseAuthParams][解析认证参数({})失败]", params, e);
456   - return null;
457   - }
458   - }
459   -
460   - /**
461   - * 解析注册参数
462   - *
463   - * @param params 参数对象（通常为 Map 类型）
464   - * @return 注册参数 DTO，解析失败时返回 null
465   - */
466   - @SuppressWarnings("unchecked")
467   - private IotDeviceRegisterReqDTO parseRegisterParams(Object params) {
468   - if (params == null) {
469   - return null;
470   - }
471   - try {
472   - // 参数默认为 Map 类型，直接转换
473   - if (params instanceof Map) {
474   - Map<String, Object> paramMap = (Map<String, Object>) params;
475   - return new IotDeviceRegisterReqDTO()
476   - .setProductKey(MapUtil.getStr(paramMap, "productKey"))
477   - .setDeviceName(MapUtil.getStr(paramMap, "deviceName"))
478   - .setProductSecret(MapUtil.getStr(paramMap, "productSecret"));
479   - }
480   - // 如果已经是目标类型，直接返回
481   - if (params instanceof IotDeviceRegisterReqDTO) {
482   - return (IotDeviceRegisterReqDTO) params;
483   - }
484   -
485   - // 其他情况尝试 JSON 转换
486   - return JsonUtils.convertObject(params, IotDeviceRegisterReqDTO.class);
487   - } catch (Exception e) {
488   - log.error("[parseRegisterParams][解析注册参数({})失败]", params, e);
489   - return null;
490   - }
491   - }
492   -
493   - /**
494   - * 发送注册成功响应（包含 deviceSecret）
495   - *
496   - * @param socket WebSocket 连接
497   - * @param requestId 请求 ID
498   - * @param registerResp 注册响应
499   - */
500   - private void sendRegisterSuccessResponse(ServerWebSocket socket, String requestId,
501   - IotDeviceRegisterRespDTO registerResp) {
502   - try {
503   - // 1. 构建响应消息（参考 HTTP 返回格式，直接返回 IotDeviceRegisterRespDTO）
504   - IotDeviceMessage responseMessage = IotDeviceMessage.replyOf(requestId,
505   - IotDeviceMessageMethodEnum.DEVICE_REGISTER.getMethod(), registerResp, 0, null);
506   - // 2. 发送响应
507   - writeResponse(socket, responseMessage);
508   - } catch (Exception e) {
509   - log.error("[sendRegisterSuccessResponse][发送注册成功响应失败，requestId: {}]", requestId, e);
510   - }
  291 + private void sendErrorResponse(ServerWebSocket socket, String requestId, String method, Integer code, String msg) {
  292 + IotDeviceMessage responseMessage = IotDeviceMessage.replyOf(requestId, method, null, code, msg);
  293 + writeResponse(socket, responseMessage);
511 294 }
512 295  
513 296 /**
... ... @@ -515,11 +298,7 @@ public class IotWebSocketUpstreamHandler implements Handler&lt;ServerWebSocket&gt; {
515 298 */
516 299 private void writeResponse(ServerWebSocket socket, IotDeviceMessage responseMessage) {
517 300 byte[] payload = serializer.serialize(responseMessage);
518   - if (binaryPayload) {
519   - socket.writeBinaryMessage(Buffer.buffer(payload));
520   - } else {
521   - socket.writeTextMessage(StrUtil.utf8Str(payload));
522   - }
  301 + socket.writeTextMessage(StrUtil.utf8Str(payload));
523 302 }
524 303  
525 304 }
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/websocket/manager/IotWebSocketConnectionManager.java
  View file @c105ece
1 1 package cn.iocoder.yudao.module.iot.gateway.protocol.websocket.manager;
2 2  
3   -import io.vertx.core.buffer.Buffer;
4 3 import io.vertx.core.http.ServerWebSocket;
5 4 import lombok.Data;
6 5 import lombok.experimental.Accessors;
... ... @@ -115,33 +114,6 @@ public class IotWebSocketConnectionManager {
115 114 }
116 115 }
117 116  
118   - // TODO @AI：没必要这里加一个；
119   - /**
120   - * 发送消息到设备（二进制消息）
121   - *
122   - * @param deviceId 设备 ID
123   - * @param payload 二进制消息
124   - * @return 是否发送成功
125   - */
126   - public boolean sendToDevice(Long deviceId, byte[] payload) {
127   - ServerWebSocket socket = deviceSocketMap.get(deviceId);
128   - if (socket == null) {
129   - log.warn("[sendToDevice][设备未连接，设备 ID: {}]", deviceId);
130   - return false;
131   - }
132   -
133   - try {
134   - socket.writeBinaryMessage(Buffer.buffer(payload));
135   - log.debug("[sendToDevice][发送消息成功，设备 ID: {}，数据长度: {} 字节]", deviceId, payload.length);
136   - return true;
137   - } catch (Exception e) {
138   - log.error("[sendToDevice][发送消息失败，设备 ID: {}]", deviceId, e);
139   - // 发送失败时清理连接
140   - unregisterConnection(socket);
141   - return false;
142   - }
143   - }
144   -
145 117 /**
146 118 * 连接信息（包含认证信息）
147 119 */
... ...