builder().put("rice", 3).build(),
+ System.currentTimeMillis()),
+ null, null, null);
+
+ // 4. 发布消息并等待响应
+ String topic = String.format("/sys/%s/%s/thing/event/post", PRODUCT_KEY, DEVICE_NAME);
+ IotDeviceMessage response = publishAndWaitReply(client, topic, request);
+ log.info("[testEventPost][响应消息: {}]", response);
+
+ // 5. 断开连接
+ disconnect(client);
+ }
- // 1. 构建认证信息
- IotDeviceAuthReqDTO authInfo = IotDeviceAuthUtils.getAuthInfo(PRODUCT_KEY, DEVICE_NAME, DEVICE_SECRET);
+ // ===================== 设备动态注册测试（一型一密） =====================
- // 2. 创建 MQTT 客户端配置
- MqttClientOptions options = new MqttClientOptions()
- .setClientId(authInfo.getClientId())
- .setUsername(authInfo.getUsername())
- .setPassword(authInfo.getPassword())
- .setCleanSession(true)
- .setKeepAliveInterval(60);
+ /**
+ * 直连设备动态注册测试（一型一密）
+ *
+ * 使用产品密钥（productSecret）验证身份，成功后返回设备密钥（deviceSecret）
+ *
+ * 注意：此接口不需要认证
+ */
+ @Test
+ public void testDeviceRegister() throws Exception {
+ // 1. 连接并认证（使用已有设备连接）
+ MqttClient client = connectAndAuth();
+ log.info("[testDeviceRegister][连接认证成功]");
+
+ // 2.1 构建注册消息
+ IotDeviceRegisterReqDTO registerReqDTO = new IotDeviceRegisterReqDTO();
+ registerReqDTO.setProductKey(PRODUCT_KEY);
+ registerReqDTO.setDeviceName("test-mqtt-" + System.currentTimeMillis());
+ registerReqDTO.setProductSecret("test-product-secret");
+ IotDeviceMessage request = IotDeviceMessage.of(IdUtil.fastSimpleUUID(),
+ IotDeviceMessageMethodEnum.DEVICE_REGISTER.getMethod(), registerReqDTO, null, null, null);
+ // 2.2 订阅 _reply 主题
+ String replyTopic = String.format("/sys/%s/%s/thing/auth/register_reply",
+ registerReqDTO.getProductKey(), registerReqDTO.getDeviceName());
+ subscribeReply(client, replyTopic);
+
+ // 3. 发布消息并等待响应
+ String topic = String.format("/sys/%s/%s/thing/auth/register",
+ registerReqDTO.getProductKey(), registerReqDTO.getDeviceName());
+ IotDeviceMessage response = publishAndWaitReply(client, topic, request);
+ log.info("[testDeviceRegister][响应消息: {}]", response);
+ log.info("[testDeviceRegister][成功后可使用返回的 deviceSecret 进行一机一密认证]");
+
+ // 4. 断开连接
+ disconnect(client);
+ }
- // 3. 创建 MQTT 客户端并连接
- MqttClient client = MqttClient.create(vertx, options);
- // TODO @AI：可以像 tcp 里面一样，有个复用么？auth 流程；
- client.connect(SERVER_PORT, SERVER_HOST)
- .onComplete(connectAr -> {
- if (connectAr.succeeded()) {
- log.info("[testEventPost][连接成功]");
-
- // 4.1 设置消息处理器，接收 _reply 响应
- client.publishHandler(message -> {
- log.info("[testEventPost][收到响应: topic={}, payload={}]",
- message.topicName(), message.payload().toString());
- });
+ // ===================== 订阅下行消息测试 =====================
- // 4.2 订阅 _reply 主题
- String replyTopic = String.format("/sys/%s/%s/thing/event/post_reply", PRODUCT_KEY, DEVICE_NAME);
- client.subscribe(replyTopic, MqttQoS.AT_LEAST_ONCE.value())
- .onComplete(subscribeAr -> {
- if (subscribeAr.succeeded()) {
- log.info("[testEventPost][订阅响应主题成功: {}]", replyTopic);
-
- // 5. 构建事件上报消息（Alink 协议格式）
- String topic = String.format("/sys/%s/%s/thing/event/post", PRODUCT_KEY, DEVICE_NAME);
- String payload = JsonUtils.toJsonString(MapUtil.builder()
- .put("id", IdUtil.fastSimpleUUID())
- .put("version", "1.0")
- .put("method", IotDeviceMessageMethodEnum.EVENT_POST.getMethod())
- .put("params", IotDeviceEventPostReqDTO.of(
- "eat",
- MapUtil.builder().put("rice", 3).build(),
- System.currentTimeMillis()))
- .build());
- log.info("[testEventPost][发送消息: topic={}, payload={}]", topic, payload);
-
- // 6. 发布消息
- client.publish(topic, Buffer.buffer(payload), MqttQoS.AT_LEAST_ONCE, false, false)
- .onComplete(publishAr -> {
- if (publishAr.succeeded()) {
- log.info("[testEventPost][消息发布成功，messageId={}]", publishAr.result());
- } else {
- log.error("[testEventPost][消息发布失败]", publishAr.cause());
- }
-
- // 等待一会儿接收响应
- vertx.setTimer(2000, id -> {
- client.disconnect()
- .onComplete(disconnectAr -> {
- log.info("[testEventPost][断开连接]");
- latch.countDown();
- });
- });
- });
- } else {
- log.error("[testEventPost][订阅响应主题失败]", subscribeAr.cause());
+ /**
+ * 订阅下行消息测试：订阅服务端下发的消息
+ */
+ @Test
+ public void testSubscribe() throws Exception {
+ CountDownLatch latch = new CountDownLatch(1);
+
+ // 1. 连接并认证
+ MqttClient client = connectAndAuth();
+ log.info("[testSubscribe][连接认证成功]");
+
+ // 2. 设置消息处理器
+ client.publishHandler(message -> {
+ log.info("[testSubscribe][收到消息: topic={}, payload={}]",
+ message.topicName(), message.payload().toString());
+ });
+
+ // 3. 订阅下行主题
+ String topic = String.format("/sys/%s/%s/thing/service/#", PRODUCT_KEY, DEVICE_NAME);
+ log.info("[testSubscribe][订阅主题: {}]", topic);
+
+ client.subscribe(topic, MqttQoS.AT_LEAST_ONCE.value())
+ .onComplete(subscribeAr -> {
+ if (subscribeAr.succeeded()) {
+ log.info("[testSubscribe][订阅成功，等待下行消息... (30秒后自动断开)]");
+ // 保持连接 30 秒等待消息
+ vertx.setTimer(30000, id -> {
+ client.disconnect()
+ .onComplete(disconnectAr -> {
+ log.info("[testSubscribe][断开连接]");
latch.countDown();
- }
- });
+ });
+ });
} else {
- log.error("[testEventPost][连接失败]", connectAr.cause());
+ log.error("[testSubscribe][订阅失败]", subscribeAr.cause());
latch.countDown();
}
});
- // 7. 等待测试完成
- boolean completed = latch.await(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ // 4. 等待测试完成
+ boolean completed = latch.await(60, TimeUnit.SECONDS);
if (!completed) {
- log.warn("[testEventPost][测试超时]");
+ log.warn("[testSubscribe][测试超时]");
}
}
- // ===================== 订阅下行消息测试 =====================
+ // ===================== 辅助方法 =====================
/**
- * 订阅下行消息测试：订阅服务端下发的消息
+ * 创建 MQTT 客户端
+ *
+ * @param authInfo 认证信息
+ * @return MQTT 客户端
*/
- @Test
- public void testSubscribe() throws Exception {
- CountDownLatch latch = new CountDownLatch(1);
-
- // 1. 构建认证信息
- IotDeviceAuthReqDTO authInfo = IotDeviceAuthUtils.getAuthInfo(PRODUCT_KEY, DEVICE_NAME, DEVICE_SECRET);
-
- // 2. 创建 MQTT 客户端配置
+ private MqttClient connect(IotDeviceAuthReqDTO authInfo) {
MqttClientOptions options = new MqttClientOptions()
.setClientId(authInfo.getClientId())
.setUsername(authInfo.getUsername())
.setPassword(authInfo.getPassword())
.setCleanSession(true)
.setKeepAliveInterval(60);
+ return MqttClient.create(vertx, options);
+ }
+
+ /**
+ * 连接并认证设备
+ *
+ * @return 已认证的 MQTT 客户端
+ */
+ private MqttClient connectAndAuth() throws Exception {
+ // 1. 创建客户端并连接
+ IotDeviceAuthReqDTO authInfo = IotDeviceAuthUtils.getAuthInfo(PRODUCT_KEY, DEVICE_NAME, DEVICE_SECRET);
+ MqttClient client = connect(authInfo);
- // 3. 创建 MQTT 客户端并连接
- MqttClient client = MqttClient.create(vertx, options);
+ // 2.1 连接
+ CompletableFuture future = new CompletableFuture<>();
client.connect(SERVER_PORT, SERVER_HOST)
- .onComplete(connectAr -> {
- if (connectAr.succeeded()) {
- log.info("[testSubscribe][连接成功]");
-
- // 4. 设置消息处理器
- client.publishHandler(message -> {
- log.info("[testSubscribe][收到消息: topic={}, payload={}]",
- message.topicName(), message.payload().toString());
- });
+ .onComplete(ar -> {
+ if (ar.succeeded()) {
+ future.complete(client);
+ } else {
+ future.completeExceptionally(ar.cause());
+ }
+ });
+ // 2.2 等待连接结果
+ return future.get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ }
- // 5. 订阅下行主题
- String topic = String.format("/sys/%s/%s/thing/service/#", PRODUCT_KEY, DEVICE_NAME);
- log.info("[testSubscribe][订阅主题: {}]", topic);
-
- client.subscribe(topic, MqttQoS.AT_LEAST_ONCE.value())
- .onComplete(subscribeAr -> {
- if (subscribeAr.succeeded()) {
- log.info("[testSubscribe][订阅成功，等待下行消息... (按 Ctrl+C 结束)]");
- // 保持连接 30 秒等待消息
- vertx.setTimer(30000, id -> {
- client.disconnect()
- .onComplete(disconnectAr -> {
- log.info("[testSubscribe][断开连接]");
- latch.countDown();
- });
- });
- } else {
- log.error("[testSubscribe][订阅失败]", subscribeAr.cause());
- latch.countDown();
- }
- });
+ /**
+ * 订阅响应主题
+ *
+ * @param client MQTT 客户端
+ * @param replyTopic 响应主题
+ */
+ private void subscribeReply(MqttClient client, String replyTopic) throws Exception {
+ // 1. 订阅响应主题
+ CompletableFuture future = new CompletableFuture<>();
+ client.subscribe(replyTopic, MqttQoS.AT_LEAST_ONCE.value())
+ .onComplete(ar -> {
+ if (ar.succeeded()) {
+ log.info("[subscribeReply][订阅响应主题成功: {}]", replyTopic);
+ future.complete(null);
} else {
- log.error("[testSubscribe][连接失败]", connectAr.cause());
- latch.countDown();
+ future.completeExceptionally(ar.cause());
}
});
+ // 2. 等待订阅结果
+ future.get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ }
- // 6. 等待测试完成
- boolean completed = latch.await(60, TimeUnit.SECONDS);
- if (!completed) {
- log.warn("[testSubscribe][测试超时]");
+ /**
+ * 发布消息并等待响应
+ *
+ * @param client MQTT 客户端
+ * @param topic 发布主题
+ * @param request 请求消息
+ * @return 响应消息
+ */
+ private IotDeviceMessage publishAndWaitReply(MqttClient client, String topic, IotDeviceMessage request) {
+ // 1. 设置消息处理器，接收响应
+ CompletableFuture future = new CompletableFuture<>();
+ client.publishHandler(message -> {
+ log.info("[publishAndWaitReply][收到响应: topic={}, payload={}]",
+ message.topicName(), message.payload().toString());
+ IotDeviceMessage response = CODEC.decode(message.payload().getBytes());
+ future.complete(response);
+ });
+
+ // 2. 编码并发布消息
+ byte[] payload = CODEC.encode(request);
+ log.info("[publishAndWaitReply][Codec: {}, 发送消息: topic={}, payload={}]",
+ CODEC.type(), topic, new String(payload));
+
+ client.publish(topic, Buffer.buffer(payload), MqttQoS.AT_LEAST_ONCE, false, false)
+ .onComplete(ar -> {
+ if (ar.succeeded()) {
+ log.info("[publishAndWaitReply][消息发布成功，messageId={}]", ar.result());
+ } else {
+ log.error("[publishAndWaitReply][消息发布失败]", ar.cause());
+ future.completeExceptionally(ar.cause());
+ }
+ });
+
+ // 3. 等待响应（超时返回 null）
+ try {
+ return future.get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ log.warn("[publishAndWaitReply][等待响应超时或失败]");
+ return null;
}
}
+ /**
+ * 断开连接
+ *
+ * @param client MQTT 客户端
+ */
+ private void disconnect(MqttClient client) throws Exception {
+ // 1. 断开连接
+ CompletableFuture future = new CompletableFuture<>();
+ client.disconnect()
+ .onComplete(ar -> {
+ if (ar.succeeded()) {
+ log.info("[disconnect][断开连接成功]");
+ future.complete(null);
+ } else {
+ future.completeExceptionally(ar.cause());
+ }
+ });
+ // 2. 等待断开结果
+ future.get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ }
+
}