BJ_ZTEITS-Developers / parking-cloud-yudao

Commit 70135174e502bcd7576e8c4b704eeb52023987eb

Authored by YunaiV
1 parent 63d7bfe2

feat：【iot】MQTT 协议：1）增加 IotDirectDeviceMqttProtocolIntegrationTest 单测案例

Inline Side-by-side
Showing 1 changed file with 370 additions and 0 deletions
yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotDirectDeviceMqttProtocolIntegrationTest.java 0 → 100644
  View file @7013517
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt;
  2 +
  3 +import cn.hutool.core.map.MapUtil;
  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;
  7 +import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
  8 +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.util.IotDeviceAuthUtils;
  11 +import io.netty.handler.codec.mqtt.MqttQoS;
  12 +import io.vertx.core.Vertx;
  13 +import io.vertx.core.buffer.Buffer;
  14 +import io.vertx.mqtt.MqttClient;
  15 +import io.vertx.mqtt.MqttClientOptions;
  16 +import lombok.extern.slf4j.Slf4j;
  17 +import org.junit.jupiter.api.AfterAll;
  18 +import org.junit.jupiter.api.BeforeAll;
  19 +import org.junit.jupiter.api.Test;
  20 +
  21 +import java.util.concurrent.CountDownLatch;
  22 +import java.util.concurrent.TimeUnit;
  23 +
  24 +/**
  25 + * IoT 直连设备 MQTT 协议集成测试（手动测试）
  26 + *
  27 + * <p>测试场景：直连设备（IotProductDeviceTypeEnum 的 DIRECT 类型）通过 MQTT 协议直接连接平台
  28 + *
  29 + * <p>使用步骤：
  30 + * <ol>
  31 + * <li>启动 yudao-module-iot-gateway 服务（MQTT 端口 1883）</li>
  32 + * <li>运行以下测试方法：
  33 + * <ul>
  34 + * <li>{@link #testConnect()} - 设备连接认证</li>
  35 + * <li>{@link #testPropertyPost()} - 设备属性上报</li>
  36 + * <li>{@link #testEventPost()} - 设备事件上报</li>
  37 + * </ul>
  38 + * </li>
  39 + * </ol>
  40 + *
  41 + * <p>注意：MQTT 协议是有状态的长连接，认证在连接时通过 username/password 完成，
  42 + * 认证成功后同一连接上的后续请求无需再携带认证信息
  43 + *
  44 + * @author 芋道源码
  45 + */
  46 +@Slf4j
  47 +public class IotDirectDeviceMqttProtocolIntegrationTest {
  48 +
  49 + private static final String SERVER_HOST = "127.0.0.1";
  50 + private static final int SERVER_PORT = 1883;
  51 + private static final int TIMEOUT_SECONDS = 10;
  52 +
  53 + // ===================== 直连设备信息（根据实际情况修改，从 iot_device 表查询） =====================
  54 + private static final String PRODUCT_KEY = "4aymZgOTOOCrDKRT";
  55 + private static final String DEVICE_NAME = "small";
  56 + private static final String DEVICE_SECRET = "0baa4c2ecc104ae1a26b4070c218bdf3";
  57 +
  58 + // ===================== 全局共享 Vertx 实例 =====================
  59 + private static Vertx vertx;
  60 +
  61 + @BeforeAll
  62 + public static void setUp() {
  63 + vertx = Vertx.vertx();
  64 + }
  65 +
  66 + @AfterAll
  67 + public static void tearDown() {
  68 + if (vertx != null) {
  69 + vertx.close();
  70 + }
  71 + }
  72 +
  73 + // ===================== 连接认证测试 =====================
  74 +
  75 + /**
  76 + * 连接认证测试：设备通过 MQTT 协议连接平台
  77 + */
  78 + @Test
  79 + public void testConnect() throws Exception {
  80 + CountDownLatch latch = new CountDownLatch(1);
  81 +
  82 + // 1. 构建认证信息
  83 + IotDeviceAuthReqDTO authInfo = IotDeviceAuthUtils.getAuthInfo(PRODUCT_KEY, DEVICE_NAME, DEVICE_SECRET);
  84 + log.info("[testConnect][认证信息: clientId={}, username={}, password={}]",
  85 + authInfo.getClientId(), authInfo.getUsername(), authInfo.getPassword());
  86 +
  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);
  97 + client.connect(SERVER_PORT, SERVER_HOST)
  98 + .onComplete(ar -> {
  99 + if (ar.succeeded()) {
  100 + log.info("[testConnect][连接成功，客户端 ID: {}]", client.clientId());
  101 + // 断开连接
  102 + client.disconnect()
  103 + .onComplete(disconnectAr -> {
  104 + if (disconnectAr.succeeded()) {
  105 + log.info("[testConnect][断开连接成功]");
  106 + } else {
  107 + log.error("[testConnect][断开连接失败]", disconnectAr.cause());
  108 + }
  109 + latch.countDown();
  110 + });
  111 + } else {
  112 + log.error("[testConnect][连接失败]", ar.cause());
  113 + latch.countDown();
  114 + }
  115 + });
  116 +
  117 + // 4. 等待测试完成
  118 + boolean completed = latch.await(TIMEOUT_SECONDS, TimeUnit.SECONDS);
  119 + if (!completed) {
  120 + log.warn("[testConnect][测试超时]");
  121 + }
  122 + }
  123 +
  124 + // ===================== 直连设备属性上报测试 =====================
  125 +
  126 + /**
  127 + * 属性上报测试
  128 + */
  129 + @Test
  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 + }
  211 + }
  212 +
  213 + // ===================== 直连设备事件上报测试 =====================
  214 +
  215 + /**
  216 + * 事件上报测试
  217 + */
  218 + @Test
  219 + public void testEventPost() throws Exception {
  220 + CountDownLatch latch = new CountDownLatch(1);
  221 +
  222 + // 1. 构建认证信息
  223 + IotDeviceAuthReqDTO authInfo = IotDeviceAuthUtils.getAuthInfo(PRODUCT_KEY, DEVICE_NAME, DEVICE_SECRET);
  224 +
  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);
  232 +
  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 + });
  246 +
  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());
  287 + latch.countDown();
  288 + }
  289 + });
  290 + } else {
  291 + log.error("[testEventPost][连接失败]", connectAr.cause());
  292 + latch.countDown();
  293 + }
  294 + });
  295 +
  296 + // 7. 等待测试完成
  297 + boolean completed = latch.await(TIMEOUT_SECONDS, TimeUnit.SECONDS);
  298 + if (!completed) {
  299 + log.warn("[testEventPost][测试超时]");
  300 + }
  301 + }
  302 +
  303 + // ===================== 订阅下行消息测试 =====================
  304 +
  305 + /**
  306 + * 订阅下行消息测试：订阅服务端下发的消息
  307 + */
  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 客户端配置
  316 + MqttClientOptions options = new MqttClientOptions()
  317 + .setClientId(authInfo.getClientId())
  318 + .setUsername(authInfo.getUsername())
  319 + .setPassword(authInfo.getPassword())
  320 + .setCleanSession(true)
  321 + .setKeepAliveInterval(60);
  322 +
  323 + // 3. 创建 MQTT 客户端并连接
  324 + MqttClient client = MqttClient.create(vertx, options);
  325 + 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 + });
  335 +
  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 + });
  357 + } else {
  358 + log.error("[testSubscribe][连接失败]", connectAr.cause());
  359 + latch.countDown();
  360 + }
  361 + });
  362 +
  363 + // 6. 等待测试完成
  364 + boolean completed = latch.await(60, TimeUnit.SECONDS);
  365 + if (!completed) {
  366 + log.warn("[testSubscribe][测试超时]");
  367 + }
  368 + }
  369 +
  370 +}