BJ_ZTEITS-Developers / parking-cloud-yudao

Commit 99bcd252a336a9e349b44b61dc2489f3820f4344

Authored by YunaiV
1 parent 70135174

feat：【iot】MQTT 协议：1）增加 register 接口 

feat：【iot】TCP/UDP 协议：统一 register 返回数据的格式
Inline Side-by-side
Showing 4 changed files with 478 additions and 240 deletions
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttUpstreamHandler.java
  View file @99bcd25
1 package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router; 1 package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router;
2 2
  3 +import cn.hutool.core.map.MapUtil;
  4 +import cn.hutool.core.util.ArrayUtil;
3 import cn.hutool.core.util.BooleanUtil; 5 import cn.hutool.core.util.BooleanUtil;
4 import cn.hutool.core.util.StrUtil; 6 import cn.hutool.core.util.StrUtil;
5 import cn.hutool.extra.spring.SpringUtil; 7 import cn.hutool.extra.spring.SpringUtil;
6 import cn.iocoder.yudao.framework.common.pojo.CommonResult; 8 import cn.iocoder.yudao.framework.common.pojo.CommonResult;
  9 +import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
7 import cn.iocoder.yudao.module.iot.core.biz.IotDeviceCommonApi; 10 import cn.iocoder.yudao.module.iot.core.biz.IotDeviceCommonApi;
8 import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceAuthReqDTO; 11 import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceAuthReqDTO;
9 import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceGetReqDTO; 12 import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceGetReqDTO;
10 import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO; 13 import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO;
  14 +import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
11 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; 15 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
12 import cn.iocoder.yudao.module.iot.core.topic.IotDeviceIdentity; 16 import cn.iocoder.yudao.module.iot.core.topic.IotDeviceIdentity;
  17 +import cn.iocoder.yudao.module.iot.core.topic.auth.IotDeviceRegisterReqDTO;
  18 +import cn.iocoder.yudao.module.iot.core.topic.auth.IotDeviceRegisterRespDTO;
13 import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils; 19 import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils;
14 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol; 20 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol;
15 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.manager.IotMqttConnectionManager; 21 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.manager.IotMqttConnectionManager;
16 import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService; 22 import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
  23 +import cn.iocoder.yudao.module.iot.gateway.util.IotMqttTopicUtils;
17 import io.netty.handler.codec.mqtt.MqttConnectReturnCode; 24 import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
18 import io.netty.handler.codec.mqtt.MqttQoS; 25 import io.netty.handler.codec.mqtt.MqttQoS;
19 import io.vertx.mqtt.MqttEndpoint; 26 import io.vertx.mqtt.MqttEndpoint;
@@ -21,6 +28,7 @@ import io.vertx.mqtt.MqttTopicSubscription; @@ -21,6 +28,7 @@ import io.vertx.mqtt.MqttTopicSubscription;
21 import lombok.extern.slf4j.Slf4j; 28 import lombok.extern.slf4j.Slf4j;
22 29
23 import java.util.List; 30 import java.util.List;
  31 +import java.util.Map;
24 32
25 /** 33 /**
26 * MQTT 上行消息处理器 34 * MQTT 上行消息处理器
@@ -30,6 +38,16 @@ import java.util.List; @@ -30,6 +38,16 @@ import java.util.List;
30 @Slf4j 38 @Slf4j
31 public class IotMqttUpstreamHandler { 39 public class IotMqttUpstreamHandler {
32 40
  41 + /**
  42 + * 默认编解码类型（MQTT 使用 Alink 协议）
  43 + */
  44 + private static final String DEFAULT_CODEC_TYPE = "Alink";
  45 +
  46 + /**
  47 + * register 请求的 topic 后缀
  48 + */
  49 + private static final String REGISTER_TOPIC_SUFFIX = "/thing/auth/register";
  50 +
33 private final IotDeviceMessageService deviceMessageService; 51 private final IotDeviceMessageService deviceMessageService;
34 52
35 private final IotMqttConnectionManager connectionManager; 53 private final IotMqttConnectionManager connectionManager;
@@ -85,20 +103,28 @@ public class IotMqttUpstreamHandler { @@ -85,20 +103,28 @@ public class IotMqttUpstreamHandler {
85 }); 103 });
86 104
87 // 4. 设置消息处理器 105 // 4. 设置消息处理器
88 - endpoint.publishHandler(message -> { 106 + endpoint.publishHandler(mqttMessage -> {
89 try { 107 try {
90 - processMessage(clientId, message.topicName(), message.payload().getBytes()); 108 + // 4.1 根据 topic 判断是否为 register 请求
  109 + String topic = mqttMessage.topicName();
  110 + byte[] payload = mqttMessage.payload().getBytes();
  111 + if (topic.endsWith(REGISTER_TOPIC_SUFFIX)) {
  112 + // register 请求：使用默认编解码器处理（设备可能未注册）
  113 + processRegisterMessage(clientId, topic, payload, endpoint);
  114 + } else {
  115 + // 业务请求：正常处理
  116 + processMessage(clientId, topic, payload);
  117 + }
91 118
92 - // 根据 QoS 级别发送相应的确认消息  
93 - if (message.qosLevel() == MqttQoS.AT_LEAST_ONCE) { 119 + // 4.2 根据 QoS 级别发送相应的确认消息
  120 + if (mqttMessage.qosLevel() == MqttQoS.AT_LEAST_ONCE) {
94 // QoS 1: 发送 PUBACK 确认 121 // QoS 1: 发送 PUBACK 确认
95 - endpoint.publishAcknowledge(message.messageId());  
96 - } else if (message.qosLevel() == MqttQoS.EXACTLY_ONCE) { 122 + endpoint.publishAcknowledge(mqttMessage.messageId());
  123 + } else if (mqttMessage.qosLevel() == MqttQoS.EXACTLY_ONCE) {
97 // QoS 2: 发送 PUBREC 确认 124 // QoS 2: 发送 PUBREC 确认
98 - endpoint.publishReceived(message.messageId()); 125 + endpoint.publishReceived(mqttMessage.messageId());
99 } 126 }
100 // QoS 0 无需确认 127 // QoS 0 无需确认
101 -  
102 } catch (Exception e) { 128 } catch (Exception e) {
103 log.error("[handle][消息解码失败，断开连接，客户端 ID: {}，地址: {}，错误: {}]", 129 log.error("[handle][消息解码失败，断开连接，客户端 ID: {}，地址: {}，错误: {}]",
104 clientId, connectionManager.getEndpointAddress(endpoint), e.getMessage()); 130 clientId, connectionManager.getEndpointAddress(endpoint), e.getMessage());
@@ -161,10 +187,9 @@ public class IotMqttUpstreamHandler { @@ -161,10 +187,9 @@ public class IotMqttUpstreamHandler {
161 return; 187 return;
162 } 188 }
163 189
  190 + // 3. 解码消息（使用从 topic 解析的 productKey 和 deviceName）
164 String productKey = topicParts[2]; 191 String productKey = topicParts[2];
165 String deviceName = topicParts[3]; 192 String deviceName = topicParts[3];
166 -  
167 - // 3. 解码消息（使用从 topic 解析的 productKey 和 deviceName）  
168 try { 193 try {
169 IotDeviceMessage message = deviceMessageService.decodeDeviceMessage(payload, productKey, deviceName); 194 IotDeviceMessage message = deviceMessageService.decodeDeviceMessage(payload, productKey, deviceName);
170 if (message == null) { 195 if (message == null) {
@@ -172,10 +197,9 @@ public class IotMqttUpstreamHandler { @@ -172,10 +197,9 @@ public class IotMqttUpstreamHandler {
172 return; 197 return;
173 } 198 }
174 199
  200 + // 4. 处理业务消息（认证已在连接时完成）
175 log.info("[processMessage][收到设备消息，设备: {}.{}, 方法: {}]", 201 log.info("[processMessage][收到设备消息，设备: {}.{}, 方法: {}]",
176 productKey, deviceName, message.getMethod()); 202 productKey, deviceName, message.getMethod());
177 -  
178 - // 4. 处理业务消息（认证已在连接时完成）  
179 handleBusinessRequest(message, productKey, deviceName); 203 handleBusinessRequest(message, productKey, deviceName);
180 } catch (Exception e) { 204 } catch (Exception e) {
181 log.error("[processMessage][消息处理异常，客户端 ID: {}，主题: {}，错误: {}]", 205 log.error("[processMessage][消息处理异常，客户端 ID: {}，主题: {}，错误: {}]",
@@ -247,6 +271,195 @@ public class IotMqttUpstreamHandler { @@ -247,6 +271,195 @@ public class IotMqttUpstreamHandler {
247 } 271 }
248 272
249 /** 273 /**
  274 + * 处理 register 消息（设备动态注册，使用默认编解码器）
  275 + *
  276 + * @param clientId 客户端 ID
  277 + * @param topic 主题
  278 + * @param payload 消息内容
  279 + * @param endpoint MQTT 连接端点
  280 + */
  281 + private void processRegisterMessage(String clientId, String topic, byte[] payload, MqttEndpoint endpoint) {
  282 + // 1.1 基础检查
  283 + if (ArrayUtil.isEmpty(payload)) {
  284 + return;
  285 + }
  286 + // 1.2 解析主题，获取 productKey 和 deviceName
  287 + String[] topicParts = topic.split("/");
  288 + if (topicParts.length < 4 || StrUtil.hasBlank(topicParts[2], topicParts[3])) {
  289 + log.warn("[processRegisterMessage][topic({}) 格式不正确]", topic);
  290 + return;
  291 + }
  292 + String productKey = topicParts[2];
  293 + String deviceName = topicParts[3];
  294 +
  295 + // 2. 使用默认编解码器解码消息（设备可能未注册，无法获取 codecType）
  296 + IotDeviceMessage message;
  297 + try {
  298 + message = deviceMessageService.decodeDeviceMessage(payload, DEFAULT_CODEC_TYPE);
  299 + if (message == null) {
  300 + log.warn("[processRegisterMessage][消息解码失败，客户端 ID: {}，主题: {}]", clientId, topic);
  301 + return;
  302 + }
  303 + } catch (Exception e) {
  304 + log.error("[processRegisterMessage][消息解码异常，客户端 ID: {}，主题: {}，错误: {}]",
  305 + clientId, topic, e.getMessage(), e);
  306 + return;
  307 + }
  308 +
  309 + // 3. 处理设备动态注册请求
  310 + log.info("[processRegisterMessage][收到设备注册消息，设备: {}.{}, 方法: {}]",
  311 + productKey, deviceName, message.getMethod());
  312 + try {
  313 + handleRegisterRequest(message, productKey, deviceName, endpoint);
  314 + } catch (Exception e) {
  315 + log.error("[processRegisterMessage][消息处理异常，客户端 ID: {}，主题: {}，错误: {}]",
  316 + clientId, topic, e.getMessage(), e);
  317 + }
  318 + }
  319 +
  320 + /**
  321 + * 处理设备动态注册请求（一型一密，不需要 deviceSecret）
  322 + *
  323 + * @param message 消息信息
  324 + * @param productKey 产品 Key
  325 + * @param deviceName 设备名称
  326 + * @param endpoint MQTT 连接端点
  327 + * @see <a href="https://help.aliyun.com/zh/iot/user-guide/unique-certificate-per-product-verification">阿里云 - 一型一密</a>
  328 + */
  329 + private void handleRegisterRequest(IotDeviceMessage message, String productKey, String deviceName, MqttEndpoint endpoint) {
  330 + String clientId = endpoint.clientIdentifier();
  331 + try {
  332 + // 1. 解析注册参数
  333 + IotDeviceRegisterReqDTO registerParams = parseRegisterParams(message.getParams());
  334 + if (registerParams == null) {
  335 + log.warn("[handleRegisterRequest][注册参数解析失败，客户端 ID: {}]", clientId);
  336 + sendRegisterErrorResponse(endpoint, productKey, deviceName, message.getRequestId(), "注册参数不完整");
  337 + return;
  338 + }
  339 +
  340 + // 2. 调用动态注册 API
  341 + CommonResult<IotDeviceRegisterRespDTO> result = deviceApi.registerDevice(registerParams);
  342 + if (result.isError()) {
  343 + log.warn("[handleRegisterRequest][注册失败，客户端 ID: {}，错误: {}]", clientId, result.getMsg());
  344 + sendRegisterErrorResponse(endpoint, productKey, deviceName, message.getRequestId(), result.getMsg());
  345 + return;
  346 + }
  347 +
  348 + // 3. 发送成功响应（包含 deviceSecret）
  349 + sendRegisterSuccessResponse(endpoint, productKey, deviceName, message.getRequestId(), result.getData());
  350 + log.info("[handleRegisterRequest][注册成功，设备名: {}，客户端 ID: {}]",
  351 + registerParams.getDeviceName(), clientId);
  352 + } catch (Exception e) {
  353 + log.error("[handleRegisterRequest][注册处理异常，客户端 ID: {}]", clientId, e);
  354 + sendRegisterErrorResponse(endpoint, productKey, deviceName, message.getRequestId(), "注册处理异常");
  355 + }
  356 + }
  357 +
  358 + /**
  359 + * 解析注册参数
  360 + *
  361 + * @param params 参数对象（通常为 Map 类型）
  362 + * @return 注册参数 DTO，解析失败时返回 null
  363 + */
  364 + @SuppressWarnings("unchecked")
  365 + private IotDeviceRegisterReqDTO parseRegisterParams(Object params) {
  366 + if (params == null) {
  367 + return null;
  368 + }
  369 +
  370 + try {
  371 + // 参数默认为 Map 类型，直接转换
  372 + if (params instanceof Map) {
  373 + Map<String, Object> paramMap = (Map<String, Object>) params;
  374 + String productKey = MapUtil.getStr(paramMap, "productKey");
  375 + String deviceName = MapUtil.getStr(paramMap, "deviceName");
  376 + String productSecret = MapUtil.getStr(paramMap, "productSecret");
  377 + if (StrUtil.hasBlank(productKey, deviceName, productSecret)) {
  378 + return null;
  379 + }
  380 + return new IotDeviceRegisterReqDTO()
  381 + .setProductKey(productKey)
  382 + .setDeviceName(deviceName)
  383 + .setProductSecret(productSecret);
  384 + }
  385 +
  386 + // 如果已经是目标类型，直接返回
  387 + if (params instanceof IotDeviceRegisterReqDTO) {
  388 + return (IotDeviceRegisterReqDTO) params;
  389 + }
  390 +
  391 + // 其他情况尝试 JSON 转换
  392 + String jsonStr = JsonUtils.toJsonString(params);
  393 + return JsonUtils.parseObject(jsonStr, IotDeviceRegisterReqDTO.class);
  394 + } catch (Exception e) {
  395 + log.error("[parseRegisterParams][解析注册参数({})失败]", params, e);
  396 + return null;
  397 + }
  398 + }
  399 +
  400 + /**
  401 + * 发送注册成功响应（包含 deviceSecret）
  402 + *
  403 + * @param endpoint MQTT 连接端点
  404 + * @param productKey 产品 Key
  405 + * @param deviceName 设备名称
  406 + * @param requestId 请求 ID
  407 + * @param registerResp 注册响应
  408 + */
  409 + private void sendRegisterSuccessResponse(MqttEndpoint endpoint, String productKey, String deviceName,
  410 + String requestId, IotDeviceRegisterRespDTO registerResp) {
  411 + try {
  412 + // 1. 构建响应消息（参考 HTTP 返回格式，直接返回 IotDeviceRegisterRespDTO）
  413 + IotDeviceMessage responseMessage = IotDeviceMessage.replyOf(requestId,
  414 + IotDeviceMessageMethodEnum.DEVICE_REGISTER.getMethod(), registerResp, 0, null);
  415 +
  416 + // 2. 编码消息（使用默认编解码器，因为设备可能还未注册）
  417 + byte[] encodedData = deviceMessageService.encodeDeviceMessage(responseMessage, DEFAULT_CODEC_TYPE);
  418 +
  419 + // 3. 构建响应主题并发送（格式：/sys/{productKey}/{deviceName}/thing/auth/register_reply）
  420 + String replyTopic = IotMqttTopicUtils.buildTopicByMethod(
  421 + IotDeviceMessageMethodEnum.DEVICE_REGISTER.getMethod(), productKey, deviceName, true);
  422 + endpoint.publish(replyTopic, io.vertx.core.buffer.Buffer.buffer(encodedData),
  423 + MqttQoS.AT_LEAST_ONCE, false, false);
  424 + log.debug("[sendRegisterSuccessResponse][发送注册成功响应，主题: {}]", replyTopic);
  425 + } catch (Exception e) {
  426 + log.error("[sendRegisterSuccessResponse][发送注册成功响应异常，客户端 ID: {}]",
  427 + endpoint.clientIdentifier(), e);
  428 + }
  429 + }
  430 +
  431 + /**
  432 + * 发送注册错误响应
  433 + *
  434 + * @param endpoint MQTT 连接端点
  435 + * @param productKey 产品 Key
  436 + * @param deviceName 设备名称
  437 + * @param requestId 请求 ID
  438 + * @param errorMessage 错误消息
  439 + */
  440 + private void sendRegisterErrorResponse(MqttEndpoint endpoint, String productKey, String deviceName,
  441 + String requestId, String errorMessage) {
  442 + try {
  443 + // 1. 构建响应消息
  444 + IotDeviceMessage responseMessage = IotDeviceMessage.replyOf(requestId,
  445 + IotDeviceMessageMethodEnum.DEVICE_REGISTER.getMethod(), null, 400, errorMessage);
  446 +
  447 + // 2. 编码消息（使用默认编解码器，因为设备可能还未注册）
  448 + byte[] encodedData = deviceMessageService.encodeDeviceMessage(responseMessage, DEFAULT_CODEC_TYPE);
  449 +
  450 + // 3. 构建响应主题并发送（格式：/sys/{productKey}/{deviceName}/thing/auth/register_reply）
  451 + String replyTopic = IotMqttTopicUtils.buildTopicByMethod(
  452 + IotDeviceMessageMethodEnum.DEVICE_REGISTER.getMethod(), productKey, deviceName, true);
  453 + endpoint.publish(replyTopic, io.vertx.core.buffer.Buffer.buffer(encodedData),
  454 + MqttQoS.AT_LEAST_ONCE, false, false);
  455 + log.debug("[sendRegisterErrorResponse][发送注册错误响应，主题: {}]", replyTopic);
  456 + } catch (Exception e) {
  457 + log.error("[sendRegisterErrorResponse][发送注册错误响应异常，客户端 ID: {}]",
  458 + endpoint.clientIdentifier(), e);
  459 + }
  460 + }
  461 +
  462 + /**
250 * 处理业务请求 463 * 处理业务请求
251 */ 464 */
252 private void handleBusinessRequest(IotDeviceMessage message, String productKey, String deviceName) { 465 private void handleBusinessRequest(IotDeviceMessage message, String productKey, String deviceName) {
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/router/IotTcpUpstreamHandler.java
  View file @99bcd25
@@ -502,15 +502,10 @@ public class IotTcpUpstreamHandler implements Handler&lt;NetSocket&gt; { @@ -502,15 +502,10 @@ public class IotTcpUpstreamHandler implements Handler&lt;NetSocket&gt; {
502 private void sendRegisterSuccessResponse(NetSocket socket, String requestId, 502 private void sendRegisterSuccessResponse(NetSocket socket, String requestId,
503 IotDeviceRegisterRespDTO registerResp, String codecType) { 503 IotDeviceRegisterRespDTO registerResp, String codecType) {
504 try { 504 try {
505 - // 构建响应数据  
506 - Object responseData = MapUtil.builder()  
507 - .put("success", true)  
508 - .put("deviceSecret", registerResp.getDeviceSecret())  
509 - .put("message", "注册成功")  
510 - .build(); 505 + // 1. 构建响应消息（参考 HTTP 返回格式，直接返回 IotDeviceRegisterRespDTO）
511 IotDeviceMessage responseMessage = IotDeviceMessage.replyOf(requestId, 506 IotDeviceMessage responseMessage = IotDeviceMessage.replyOf(requestId,
512 - IotDeviceMessageMethodEnum.DEVICE_REGISTER.getMethod(), responseData, 0, "注册成功");  
513 - 507 + IotDeviceMessageMethodEnum.DEVICE_REGISTER.getMethod(), registerResp, 0, null);
  508 + // 2. 发送响应
514 byte[] encodedData = deviceMessageService.encodeDeviceMessage(responseMessage, codecType); 509 byte[] encodedData = deviceMessageService.encodeDeviceMessage(responseMessage, codecType);
515 socket.write(Buffer.buffer(encodedData)); 510 socket.write(Buffer.buffer(encodedData));
516 } catch (Exception e) { 511 } catch (Exception e) {
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/udp/router/IotUdpUpstreamHandler.java
  View file @99bcd25
@@ -405,15 +405,10 @@ public class IotUdpUpstreamHandler { @@ -405,15 +405,10 @@ public class IotUdpUpstreamHandler {
405 String requestId, IotDeviceRegisterRespDTO registerResp, 405 String requestId, IotDeviceRegisterRespDTO registerResp,
406 String codecType) { 406 String codecType) {
407 try { 407 try {
408 - // 构建响应数据  
409 - Object responseData = MapUtil.builder()  
410 - .put("success", true)  
411 - .put("deviceSecret", registerResp.getDeviceSecret())  
412 - .put("message", "注册成功")  
413 - .build(); 408 + // 1. 构建响应消息（参考 HTTP 返回格式，直接返回 IotDeviceRegisterRespDTO）
414 IotDeviceMessage responseMessage = IotDeviceMessage.replyOf(requestId, 409 IotDeviceMessage responseMessage = IotDeviceMessage.replyOf(requestId,
415 - IotDeviceMessageMethodEnum.DEVICE_REGISTER.getMethod(), responseData, 0, "注册成功");  
416 - // 发送响应 410 + IotDeviceMessageMethodEnum.DEVICE_REGISTER.getMethod(), registerResp, 0, null);
  411 + // 2. 发送响应
417 byte[] encodedData = deviceMessageService.encodeDeviceMessage(responseMessage, codecType); 412 byte[] encodedData = deviceMessageService.encodeDeviceMessage(responseMessage, codecType);
418 socket.send(Buffer.buffer(encodedData), address.getPort(), address.getHostString(), result -> { 413 socket.send(Buffer.buffer(encodedData), address.getPort(), address.getHostString(), result -> {
419 if (result.failed()) { 414 if (result.failed()) {
yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotDirectDeviceMqttProtocolIntegrationTest.java
  View file @99bcd25
@@ -2,12 +2,15 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt; @@ -2,12 +2,15 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt;
2 2
3 import cn.hutool.core.map.MapUtil; 3 import cn.hutool.core.map.MapUtil;
4 import cn.hutool.core.util.IdUtil; 4 import cn.hutool.core.util.IdUtil;
5 -import cn.iocoder.yudao.framework.common.util.json.JsonUtils;  
6 import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceAuthReqDTO; 5 import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceAuthReqDTO;
7 import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum; 6 import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
  7 +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  8 +import cn.iocoder.yudao.module.iot.core.topic.auth.IotDeviceRegisterReqDTO;
8 import cn.iocoder.yudao.module.iot.core.topic.event.IotDeviceEventPostReqDTO; 9 import cn.iocoder.yudao.module.iot.core.topic.event.IotDeviceEventPostReqDTO;
9 import cn.iocoder.yudao.module.iot.core.topic.property.IotDevicePropertyPostReqDTO; 10 import cn.iocoder.yudao.module.iot.core.topic.property.IotDevicePropertyPostReqDTO;
10 import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils; 11 import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils;
  12 +import cn.iocoder.yudao.module.iot.gateway.codec.IotDeviceMessageCodec;
  13 +import cn.iocoder.yudao.module.iot.gateway.codec.alink.IotAlinkDeviceMessageCodec;
11 import io.netty.handler.codec.mqtt.MqttQoS; 14 import io.netty.handler.codec.mqtt.MqttQoS;
12 import io.vertx.core.Vertx; 15 import io.vertx.core.Vertx;
13 import io.vertx.core.buffer.Buffer; 16 import io.vertx.core.buffer.Buffer;
@@ -18,6 +21,7 @@ import org.junit.jupiter.api.AfterAll; @@ -18,6 +21,7 @@ import org.junit.jupiter.api.AfterAll;
18 import org.junit.jupiter.api.BeforeAll; 21 import org.junit.jupiter.api.BeforeAll;
19 import org.junit.jupiter.api.Test; 22 import org.junit.jupiter.api.Test;
20 23
  24 +import java.util.concurrent.CompletableFuture;
21 import java.util.concurrent.CountDownLatch; 25 import java.util.concurrent.CountDownLatch;
22 import java.util.concurrent.TimeUnit; 26 import java.util.concurrent.TimeUnit;
23 27
@@ -31,9 +35,10 @@ import java.util.concurrent.TimeUnit; @@ -31,9 +35,10 @@ import java.util.concurrent.TimeUnit;
31 * <li>启动 yudao-module-iot-gateway 服务（MQTT 端口 1883）</li> 35 * <li>启动 yudao-module-iot-gateway 服务（MQTT 端口 1883）</li>
32 * <li>运行以下测试方法： 36 * <li>运行以下测试方法：
33 * <ul> 37 * <ul>
34 - * <li>{@link #testConnect()} - 设备连接认证</li> 38 + * <li>{@link #testAuth()} - 设备连接认证</li>
35 * <li>{@link #testPropertyPost()} - 设备属性上报</li> 39 * <li>{@link #testPropertyPost()} - 设备属性上报</li>
36 * <li>{@link #testEventPost()} - 设备事件上报</li> 40 * <li>{@link #testEventPost()} - 设备事件上报</li>
  41 + * <li>{@link #testSubscribe()} - 订阅下行消息</li>
37 * </ul> 42 * </ul>
38 * </li> 43 * </li>
39 * </ol> 44 * </ol>
@@ -50,6 +55,9 @@ public class IotDirectDeviceMqttProtocolIntegrationTest { @@ -50,6 +55,9 @@ public class IotDirectDeviceMqttProtocolIntegrationTest {
50 private static final int SERVER_PORT = 1883; 55 private static final int SERVER_PORT = 1883;
51 private static final int TIMEOUT_SECONDS = 10; 56 private static final int TIMEOUT_SECONDS = 10;
52 57
  58 + // ===================== 编解码器（MQTT 使用 Alink 协议） =====================
  59 + private static final IotDeviceMessageCodec CODEC = new IotAlinkDeviceMessageCodec();
  60 +
53 // ===================== 直连设备信息（根据实际情况修改，从 iot_device 表查询） ===================== 61 // ===================== 直连设备信息（根据实际情况修改，从 iot_device 表查询） =====================
54 private static final String PRODUCT_KEY = "4aymZgOTOOCrDKRT"; 62 private static final String PRODUCT_KEY = "4aymZgOTOOCrDKRT";
55 private static final String DEVICE_NAME = "small"; 63 private static final String DEVICE_NAME = "small";
@@ -73,10 +81,10 @@ public class IotDirectDeviceMqttProtocolIntegrationTest { @@ -73,10 +81,10 @@ public class IotDirectDeviceMqttProtocolIntegrationTest {
73 // ===================== 连接认证测试 ===================== 81 // ===================== 连接认证测试 =====================
74 82
75 /** 83 /**
76 - * 连接认证测试：设备通过 MQTT 协议连接平台 84 + * 认证测试：获取设备 Token
77 */ 85 */
78 @Test 86 @Test
79 - public void testConnect() throws Exception { 87 + public void testAuth() throws Exception {
80 CountDownLatch latch = new CountDownLatch(1); 88 CountDownLatch latch = new CountDownLatch(1);
81 89
82 // 1. 构建认证信息 90 // 1. 构建认证信息
@@ -84,16 +92,8 @@ public class IotDirectDeviceMqttProtocolIntegrationTest { @@ -84,16 +92,8 @@ public class IotDirectDeviceMqttProtocolIntegrationTest {
84 log.info("[testConnect][认证信息: clientId={}, username={}, password={}]", 92 log.info("[testConnect][认证信息: clientId={}, username={}, password={}]",
85 authInfo.getClientId(), authInfo.getUsername(), authInfo.getPassword()); 93 authInfo.getClientId(), authInfo.getUsername(), authInfo.getPassword());
86 94
87 - // 2. 创建 MQTT 客户端配置  
88 - MqttClientOptions options = new MqttClientOptions()  
89 - .setClientId(authInfo.getClientId())  
90 - .setUsername(authInfo.getUsername())  
91 - .setPassword(authInfo.getPassword())  
92 - .setCleanSession(true)  
93 - .setKeepAliveInterval(60);  
94 -  
95 - // 3. 创建 MQTT 客户端并连接  
96 - MqttClient client = MqttClient.create(vertx, options); 95 + // 2. 创建客户端并连接
  96 + MqttClient client = connect(authInfo);
97 client.connect(SERVER_PORT, SERVER_HOST) 97 client.connect(SERVER_PORT, SERVER_HOST)
98 .onComplete(ar -> { 98 .onComplete(ar -> {
99 if (ar.succeeded()) { 99 if (ar.succeeded()) {
@@ -114,7 +114,7 @@ public class IotDirectDeviceMqttProtocolIntegrationTest { @@ -114,7 +114,7 @@ public class IotDirectDeviceMqttProtocolIntegrationTest {
114 } 114 }
115 }); 115 });
116 116
117 - // 4. 等待测试完成 117 + // 3. 等待测试完成
118 boolean completed = latch.await(TIMEOUT_SECONDS, TimeUnit.SECONDS); 118 boolean completed = latch.await(TIMEOUT_SECONDS, TimeUnit.SECONDS);
119 if (!completed) { 119 if (!completed) {
120 log.warn("[testConnect][测试超时]"); 120 log.warn("[testConnect][测试超时]");
@@ -128,86 +128,31 @@ public class IotDirectDeviceMqttProtocolIntegrationTest { @@ -128,86 +128,31 @@ public class IotDirectDeviceMqttProtocolIntegrationTest {
128 */ 128 */
129 @Test 129 @Test
130 public void testPropertyPost() throws Exception { 130 public void testPropertyPost() throws Exception {
131 - CountDownLatch latch = new CountDownLatch(1);  
132 -  
133 - // 1. 构建认证信息  
134 - IotDeviceAuthReqDTO authInfo = IotDeviceAuthUtils.getAuthInfo(PRODUCT_KEY, DEVICE_NAME, DEVICE_SECRET);  
135 -  
136 - // 2. 创建 MQTT 客户端配置  
137 - MqttClientOptions options = new MqttClientOptions()  
138 - .setClientId(authInfo.getClientId())  
139 - .setUsername(authInfo.getUsername())  
140 - .setPassword(authInfo.getPassword())  
141 - .setCleanSession(true)  
142 - .setKeepAliveInterval(60);  
143 -  
144 - // 3. 创建 MQTT 客户端并连接  
145 - MqttClient client = MqttClient.create(vertx, options);  
146 - client.connect(SERVER_PORT, SERVER_HOST)  
147 - .onComplete(connectAr -> {  
148 - if (connectAr.succeeded()) {  
149 - log.info("[testPropertyPost][连接成功]");  
150 -  
151 - // 4.1 设置消息处理器，接收 _reply 响应  
152 - client.publishHandler(message -> {  
153 - log.info("[testPropertyPost][收到响应: topic={}, payload={}]",  
154 - message.topicName(), message.payload().toString());  
155 - });  
156 -  
157 - // 4.2 订阅 _reply 主题  
158 - String replyTopic = String.format("/sys/%s/%s/thing/property/post_reply", PRODUCT_KEY, DEVICE_NAME);  
159 - client.subscribe(replyTopic, MqttQoS.AT_LEAST_ONCE.value())  
160 - .onComplete(subscribeAr -> {  
161 - if (subscribeAr.succeeded()) {  
162 - log.info("[testPropertyPost][订阅响应主题成功: {}]", replyTopic);  
163 -  
164 - // 5. 构建属性上报消息（Alink 协议格式）  
165 - String topic = String.format("/sys/%s/%s/thing/property/post", PRODUCT_KEY, DEVICE_NAME);  
166 - String payload = JsonUtils.toJsonString(MapUtil.builder()  
167 - .put("id", IdUtil.fastSimpleUUID())  
168 - .put("version", "1.0")  
169 - .put("method", IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod())  
170 - .put("params", IotDevicePropertyPostReqDTO.of(MapUtil.<String, Object>builder()  
171 - .put("width", 1)  
172 - .put("height", "2")  
173 - .build()))  
174 - .build());  
175 - log.info("[testPropertyPost][发送消息: topic={}, payload={}]", topic, payload);  
176 -  
177 - // 6. 发布消息  
178 - client.publish(topic, Buffer.buffer(payload), MqttQoS.AT_LEAST_ONCE, false, false)  
179 - .onComplete(publishAr -> {  
180 - if (publishAr.succeeded()) {  
181 - log.info("[testPropertyPost][消息发布成功，messageId={}]", publishAr.result());  
182 - } else {  
183 - log.error("[testPropertyPost][消息发布失败]", publishAr.cause());  
184 - }  
185 -  
186 - // 等待一会儿接收响应  
187 - vertx.setTimer(2000, id -> {  
188 - client.disconnect()  
189 - .onComplete(disconnectAr -> {  
190 - log.info("[testPropertyPost][断开连接]");  
191 - latch.countDown();  
192 - });  
193 - });  
194 - });  
195 - } else {  
196 - log.error("[testPropertyPost][订阅响应主题失败]", subscribeAr.cause());  
197 - latch.countDown();  
198 - }  
199 - });  
200 - } else {  
201 - log.error("[testPropertyPost][连接失败]", connectAr.cause());  
202 - latch.countDown();  
203 - }  
204 - });  
205 -  
206 - // 7. 等待测试完成  
207 - boolean completed = latch.await(TIMEOUT_SECONDS, TimeUnit.SECONDS);  
208 - if (!completed) {  
209 - log.warn("[testPropertyPost][测试超时]");  
210 - } 131 + // 1. 连接并认证
  132 + MqttClient client = connectAndAuth();
  133 + log.info("[testPropertyPost][连接认证成功]");
  134 +
  135 + // 2. 订阅 _reply 主题
  136 + String replyTopic = String.format("/sys/%s/%s/thing/property/post_reply", PRODUCT_KEY, DEVICE_NAME);
  137 + subscribeReply(client, replyTopic);
  138 +
  139 + // 3. 构建属性上报消息
  140 + IotDeviceMessage request = IotDeviceMessage.of(
  141 + IdUtil.fastSimpleUUID(),
  142 + IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod(),
  143 + IotDevicePropertyPostReqDTO.of(MapUtil.<String, Object>builder()
  144 + .put("width", 1)
  145 + .put("height", "2")
  146 + .build()),
  147 + null, null, null);
  148 +
  149 + // 4. 发布消息并等待响应
  150 + String topic = String.format("/sys/%s/%s/thing/property/post", PRODUCT_KEY, DEVICE_NAME);
  151 + IotDeviceMessage response = publishAndWaitReply(client, topic, request);
  152 + log.info("[testPropertyPost][响应消息: {}]", response);
  153 +
  154 + // 5. 断开连接
  155 + disconnect(client);
211 } 156 }
212 157
213 // ===================== 直连设备事件上报测试 ===================== 158 // ===================== 直连设备事件上报测试 =====================
@@ -217,154 +162,244 @@ public class IotDirectDeviceMqttProtocolIntegrationTest { @@ -217,154 +162,244 @@ public class IotDirectDeviceMqttProtocolIntegrationTest {
217 */ 162 */
218 @Test 163 @Test
219 public void testEventPost() throws Exception { 164 public void testEventPost() throws Exception {
220 - CountDownLatch latch = new CountDownLatch(1); 165 + // 1. 连接并认证
  166 + MqttClient client = connectAndAuth();
  167 + log.info("[testEventPost][连接认证成功]");
  168 +
  169 + // 2. 订阅 _reply 主题
  170 + String replyTopic = String.format("/sys/%s/%s/thing/event/post_reply", PRODUCT_KEY, DEVICE_NAME);
  171 + subscribeReply(client, replyTopic);
  172 +
  173 + // 3. 构建事件上报消息
  174 + IotDeviceMessage request = IotDeviceMessage.of(
  175 + IdUtil.fastSimpleUUID(),
  176 + IotDeviceMessageMethodEnum.EVENT_POST.getMethod(),
  177 + IotDeviceEventPostReqDTO.of(
  178 + "eat",
  179 + MapUtil.<String, Object>builder().put("rice", 3).build(),
  180 + System.currentTimeMillis()),
  181 + null, null, null);
  182 +
  183 + // 4. 发布消息并等待响应
  184 + String topic = String.format("/sys/%s/%s/thing/event/post", PRODUCT_KEY, DEVICE_NAME);
  185 + IotDeviceMessage response = publishAndWaitReply(client, topic, request);
  186 + log.info("[testEventPost][响应消息: {}]", response);
  187 +
  188 + // 5. 断开连接
  189 + disconnect(client);
  190 + }
221 191
222 - // 1. 构建认证信息  
223 - IotDeviceAuthReqDTO authInfo = IotDeviceAuthUtils.getAuthInfo(PRODUCT_KEY, DEVICE_NAME, DEVICE_SECRET); 192 + // ===================== 设备动态注册测试（一型一密） =====================
224 193
225 - // 2. 创建 MQTT 客户端配置  
226 - MqttClientOptions options = new MqttClientOptions()  
227 - .setClientId(authInfo.getClientId())  
228 - .setUsername(authInfo.getUsername())  
229 - .setPassword(authInfo.getPassword())  
230 - .setCleanSession(true)  
231 - .setKeepAliveInterval(60); 194 + /**
  195 + * 直连设备动态注册测试（一型一密）
  196 + * <p>
  197 + * 使用产品密钥（productSecret）验证身份，成功后返回设备密钥（deviceSecret）
  198 + * <p>
  199 + * 注意：此接口不需要认证
  200 + */
  201 + @Test
  202 + public void testDeviceRegister() throws Exception {
  203 + // 1. 连接并认证（使用已有设备连接）
  204 + MqttClient client = connectAndAuth();
  205 + log.info("[testDeviceRegister][连接认证成功]");
  206 +
  207 + // 2.1 构建注册消息
  208 + IotDeviceRegisterReqDTO registerReqDTO = new IotDeviceRegisterReqDTO();
  209 + registerReqDTO.setProductKey(PRODUCT_KEY);
  210 + registerReqDTO.setDeviceName("test-mqtt-" + System.currentTimeMillis());
  211 + registerReqDTO.setProductSecret("test-product-secret");
  212 + IotDeviceMessage request = IotDeviceMessage.of(IdUtil.fastSimpleUUID(),
  213 + IotDeviceMessageMethodEnum.DEVICE_REGISTER.getMethod(), registerReqDTO, null, null, null);
  214 + // 2.2 订阅 _reply 主题
  215 + String replyTopic = String.format("/sys/%s/%s/thing/auth/register_reply",
  216 + registerReqDTO.getProductKey(), registerReqDTO.getDeviceName());
  217 + subscribeReply(client, replyTopic);
  218 +
  219 + // 3. 发布消息并等待响应
  220 + String topic = String.format("/sys/%s/%s/thing/auth/register",
  221 + registerReqDTO.getProductKey(), registerReqDTO.getDeviceName());
  222 + IotDeviceMessage response = publishAndWaitReply(client, topic, request);
  223 + log.info("[testDeviceRegister][响应消息: {}]", response);
  224 + log.info("[testDeviceRegister][成功后可使用返回的 deviceSecret 进行一机一密认证]");
  225 +
  226 + // 4. 断开连接
  227 + disconnect(client);
  228 + }
232 229
233 - // 3. 创建 MQTT 客户端并连接  
234 - MqttClient client = MqttClient.create(vertx, options);  
235 - // TODO @AI：可以像 tcp 里面一样，有个复用么？auth 流程；  
236 - client.connect(SERVER_PORT, SERVER_HOST)  
237 - .onComplete(connectAr -> {  
238 - if (connectAr.succeeded()) {  
239 - log.info("[testEventPost][连接成功]");  
240 -  
241 - // 4.1 设置消息处理器，接收 _reply 响应  
242 - client.publishHandler(message -> {  
243 - log.info("[testEventPost][收到响应: topic={}, payload={}]",  
244 - message.topicName(), message.payload().toString());  
245 - }); 230 + // ===================== 订阅下行消息测试 =====================
246 231
247 - // 4.2 订阅 _reply 主题  
248 - String replyTopic = String.format("/sys/%s/%s/thing/event/post_reply", PRODUCT_KEY, DEVICE_NAME);  
249 - client.subscribe(replyTopic, MqttQoS.AT_LEAST_ONCE.value())  
250 - .onComplete(subscribeAr -> {  
251 - if (subscribeAr.succeeded()) {  
252 - log.info("[testEventPost][订阅响应主题成功: {}]", replyTopic);  
253 -  
254 - // 5. 构建事件上报消息（Alink 协议格式）  
255 - String topic = String.format("/sys/%s/%s/thing/event/post", PRODUCT_KEY, DEVICE_NAME);  
256 - String payload = JsonUtils.toJsonString(MapUtil.builder()  
257 - .put("id", IdUtil.fastSimpleUUID())  
258 - .put("version", "1.0")  
259 - .put("method", IotDeviceMessageMethodEnum.EVENT_POST.getMethod())  
260 - .put("params", IotDeviceEventPostReqDTO.of(  
261 - "eat",  
262 - MapUtil.<String, Object>builder().put("rice", 3).build(),  
263 - System.currentTimeMillis()))  
264 - .build());  
265 - log.info("[testEventPost][发送消息: topic={}, payload={}]", topic, payload);  
266 -  
267 - // 6. 发布消息  
268 - client.publish(topic, Buffer.buffer(payload), MqttQoS.AT_LEAST_ONCE, false, false)  
269 - .onComplete(publishAr -> {  
270 - if (publishAr.succeeded()) {  
271 - log.info("[testEventPost][消息发布成功，messageId={}]", publishAr.result());  
272 - } else {  
273 - log.error("[testEventPost][消息发布失败]", publishAr.cause());  
274 - }  
275 -  
276 - // 等待一会儿接收响应  
277 - vertx.setTimer(2000, id -> {  
278 - client.disconnect()  
279 - .onComplete(disconnectAr -> {  
280 - log.info("[testEventPost][断开连接]");  
281 - latch.countDown();  
282 - });  
283 - });  
284 - });  
285 - } else {  
286 - log.error("[testEventPost][订阅响应主题失败]", subscribeAr.cause()); 232 + /**
  233 + * 订阅下行消息测试：订阅服务端下发的消息
  234 + */
  235 + @Test
  236 + public void testSubscribe() throws Exception {
  237 + CountDownLatch latch = new CountDownLatch(1);
  238 +
  239 + // 1. 连接并认证
  240 + MqttClient client = connectAndAuth();
  241 + log.info("[testSubscribe][连接认证成功]");
  242 +
  243 + // 2. 设置消息处理器
  244 + client.publishHandler(message -> {
  245 + log.info("[testSubscribe][收到消息: topic={}, payload={}]",
  246 + message.topicName(), message.payload().toString());
  247 + });
  248 +
  249 + // 3. 订阅下行主题
  250 + String topic = String.format("/sys/%s/%s/thing/service/#", PRODUCT_KEY, DEVICE_NAME);
  251 + log.info("[testSubscribe][订阅主题: {}]", topic);
  252 +
  253 + client.subscribe(topic, MqttQoS.AT_LEAST_ONCE.value())
  254 + .onComplete(subscribeAr -> {
  255 + if (subscribeAr.succeeded()) {
  256 + log.info("[testSubscribe][订阅成功，等待下行消息... (30秒后自动断开)]");
  257 + // 保持连接 30 秒等待消息
  258 + vertx.setTimer(30000, id -> {
  259 + client.disconnect()
  260 + .onComplete(disconnectAr -> {
  261 + log.info("[testSubscribe][断开连接]");
287 latch.countDown(); 262 latch.countDown();
288 - }  
289 - }); 263 + });
  264 + });
290 } else { 265 } else {
291 - log.error("[testEventPost][连接失败]", connectAr.cause()); 266 + log.error("[testSubscribe][订阅失败]", subscribeAr.cause());
292 latch.countDown(); 267 latch.countDown();
293 } 268 }
294 }); 269 });
295 270
296 - // 7. 等待测试完成  
297 - boolean completed = latch.await(TIMEOUT_SECONDS, TimeUnit.SECONDS); 271 + // 4. 等待测试完成
  272 + boolean completed = latch.await(60, TimeUnit.SECONDS);
298 if (!completed) { 273 if (!completed) {
299 - log.warn("[testEventPost][测试超时]"); 274 + log.warn("[testSubscribe][测试超时]");
300 } 275 }
301 } 276 }
302 277
303 - // ===================== 订阅下行消息测试 ===================== 278 + // ===================== 辅助方法 =====================
304 279
305 /** 280 /**
306 - * 订阅下行消息测试：订阅服务端下发的消息 281 + * 创建 MQTT 客户端
  282 + *
  283 + * @param authInfo 认证信息
  284 + * @return MQTT 客户端
307 */ 285 */
308 - @Test  
309 - public void testSubscribe() throws Exception {  
310 - CountDownLatch latch = new CountDownLatch(1);  
311 -  
312 - // 1. 构建认证信息  
313 - IotDeviceAuthReqDTO authInfo = IotDeviceAuthUtils.getAuthInfo(PRODUCT_KEY, DEVICE_NAME, DEVICE_SECRET);  
314 -  
315 - // 2. 创建 MQTT 客户端配置 286 + private MqttClient connect(IotDeviceAuthReqDTO authInfo) {
316 MqttClientOptions options = new MqttClientOptions() 287 MqttClientOptions options = new MqttClientOptions()
317 .setClientId(authInfo.getClientId()) 288 .setClientId(authInfo.getClientId())
318 .setUsername(authInfo.getUsername()) 289 .setUsername(authInfo.getUsername())
319 .setPassword(authInfo.getPassword()) 290 .setPassword(authInfo.getPassword())
320 .setCleanSession(true) 291 .setCleanSession(true)
321 .setKeepAliveInterval(60); 292 .setKeepAliveInterval(60);
  293 + return MqttClient.create(vertx, options);
  294 + }
  295 +
  296 + /**
  297 + * 连接并认证设备
  298 + *
  299 + * @return 已认证的 MQTT 客户端
  300 + */
  301 + private MqttClient connectAndAuth() throws Exception {
  302 + // 1. 创建客户端并连接
  303 + IotDeviceAuthReqDTO authInfo = IotDeviceAuthUtils.getAuthInfo(PRODUCT_KEY, DEVICE_NAME, DEVICE_SECRET);
  304 + MqttClient client = connect(authInfo);
322 305
323 - // 3. 创建 MQTT 客户端并连接  
324 - MqttClient client = MqttClient.create(vertx, options); 306 + // 2.1 连接
  307 + CompletableFuture<MqttClient> future = new CompletableFuture<>();
325 client.connect(SERVER_PORT, SERVER_HOST) 308 client.connect(SERVER_PORT, SERVER_HOST)
326 - .onComplete(connectAr -> {  
327 - if (connectAr.succeeded()) {  
328 - log.info("[testSubscribe][连接成功]");  
329 -  
330 - // 4. 设置消息处理器  
331 - client.publishHandler(message -> {  
332 - log.info("[testSubscribe][收到消息: topic={}, payload={}]",  
333 - message.topicName(), message.payload().toString());  
334 - }); 309 + .onComplete(ar -> {
  310 + if (ar.succeeded()) {
  311 + future.complete(client);
  312 + } else {
  313 + future.completeExceptionally(ar.cause());
  314 + }
  315 + });
  316 + // 2.2 等待连接结果
  317 + return future.get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
  318 + }
335 319
336 - // 5. 订阅下行主题  
337 - String topic = String.format("/sys/%s/%s/thing/service/#", PRODUCT_KEY, DEVICE_NAME);  
338 - log.info("[testSubscribe][订阅主题: {}]", topic);  
339 -  
340 - client.subscribe(topic, MqttQoS.AT_LEAST_ONCE.value())  
341 - .onComplete(subscribeAr -> {  
342 - if (subscribeAr.succeeded()) {  
343 - log.info("[testSubscribe][订阅成功，等待下行消息... (按 Ctrl+C 结束)]");  
344 - // 保持连接 30 秒等待消息  
345 - vertx.setTimer(30000, id -> {  
346 - client.disconnect()  
347 - .onComplete(disconnectAr -> {  
348 - log.info("[testSubscribe][断开连接]");  
349 - latch.countDown();  
350 - });  
351 - });  
352 - } else {  
353 - log.error("[testSubscribe][订阅失败]", subscribeAr.cause());  
354 - latch.countDown();  
355 - }  
356 - }); 320 + /**
  321 + * 订阅响应主题
  322 + *
  323 + * @param client MQTT 客户端
  324 + * @param replyTopic 响应主题
  325 + */
  326 + private void subscribeReply(MqttClient client, String replyTopic) throws Exception {
  327 + // 1. 订阅响应主题
  328 + CompletableFuture<Void> future = new CompletableFuture<>();
  329 + client.subscribe(replyTopic, MqttQoS.AT_LEAST_ONCE.value())
  330 + .onComplete(ar -> {
  331 + if (ar.succeeded()) {
  332 + log.info("[subscribeReply][订阅响应主题成功: {}]", replyTopic);
  333 + future.complete(null);
357 } else { 334 } else {
358 - log.error("[testSubscribe][连接失败]", connectAr.cause());  
359 - latch.countDown(); 335 + future.completeExceptionally(ar.cause());
360 } 336 }
361 }); 337 });
  338 + // 2. 等待订阅结果
  339 + future.get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
  340 + }
362 341
363 - // 6. 等待测试完成  
364 - boolean completed = latch.await(60, TimeUnit.SECONDS);  
365 - if (!completed) {  
366 - log.warn("[testSubscribe][测试超时]"); 342 + /**
  343 + * 发布消息并等待响应
  344 + *
  345 + * @param client MQTT 客户端
  346 + * @param topic 发布主题
  347 + * @param request 请求消息
  348 + * @return 响应消息
  349 + */
  350 + private IotDeviceMessage publishAndWaitReply(MqttClient client, String topic, IotDeviceMessage request) {
  351 + // 1. 设置消息处理器，接收响应
  352 + CompletableFuture<IotDeviceMessage> future = new CompletableFuture<>();
  353 + client.publishHandler(message -> {
  354 + log.info("[publishAndWaitReply][收到响应: topic={}, payload={}]",
  355 + message.topicName(), message.payload().toString());
  356 + IotDeviceMessage response = CODEC.decode(message.payload().getBytes());
  357 + future.complete(response);
  358 + });
  359 +
  360 + // 2. 编码并发布消息
  361 + byte[] payload = CODEC.encode(request);
  362 + log.info("[publishAndWaitReply][Codec: {}, 发送消息: topic={}, payload={}]",
  363 + CODEC.type(), topic, new String(payload));
  364 +
  365 + client.publish(topic, Buffer.buffer(payload), MqttQoS.AT_LEAST_ONCE, false, false)
  366 + .onComplete(ar -> {
  367 + if (ar.succeeded()) {
  368 + log.info("[publishAndWaitReply][消息发布成功，messageId={}]", ar.result());
  369 + } else {
  370 + log.error("[publishAndWaitReply][消息发布失败]", ar.cause());
  371 + future.completeExceptionally(ar.cause());
  372 + }
  373 + });
  374 +
  375 + // 3. 等待响应（超时返回 null）
  376 + try {
  377 + return future.get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
  378 + } catch (Exception e) {
  379 + log.warn("[publishAndWaitReply][等待响应超时或失败]");
  380 + return null;
367 } 381 }
368 } 382 }
369 383
  384 + /**
  385 + * 断开连接
  386 + *
  387 + * @param client MQTT 客户端
  388 + */
  389 + private void disconnect(MqttClient client) throws Exception {
  390 + // 1. 断开连接
  391 + CompletableFuture<Void> future = new CompletableFuture<>();
  392 + client.disconnect()
  393 + .onComplete(ar -> {
  394 + if (ar.succeeded()) {
  395 + log.info("[disconnect][断开连接成功]");
  396 + future.complete(null);
  397 + } else {
  398 + future.completeExceptionally(ar.cause());
  399 + }
  400 + });
  401 + // 2. 等待断开结果
  402 + future.get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
  403 + }
  404 +
370 } 405 }