BJ_ZTEITS-Developers / parking-cloud-yudao

Commit 63d7bfe2d26ef9d971746edbf469877f325d42b2

Authored by YunaiV
1 parent de1a53a5

feat：【iot】TCP 协议：1）增加 register 协议；2）增加 gateway 相关的单测

Inline Side-by-side
Showing 6 changed files with 784 additions and 2 deletions
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/codec/tcp/IotTcpBinaryDeviceMessageCodec.java
  View file @63d7bfe
@@ -13,7 +13,7 @@ import org.springframework.stereotype.Component; @@ -13,7 +13,7 @@ import org.springframework.stereotype.Component;
13 import java.nio.charset.StandardCharsets; 13 import java.nio.charset.StandardCharsets;
14 14
15 /** 15 /**
16 - * TCP 二进制格式 {@link IotDeviceMessage} 编解码器 16 + * TCP/UDP 二进制格式 {@link IotDeviceMessage} 编解码器
17 * <p> 17 * <p>
18 * 二进制协议格式（所有数值使用大端序）： 18 * 二进制协议格式（所有数值使用大端序）：
19 * 19 *
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/codec/tcp/IotTcpJsonDeviceMessageCodec.java
  View file @63d7bfe
@@ -11,7 +11,7 @@ import lombok.NoArgsConstructor; @@ -11,7 +11,7 @@ import lombok.NoArgsConstructor;
11 import org.springframework.stereotype.Component; 11 import org.springframework.stereotype.Component;
12 12
13 /** 13 /**
14 - * TCP JSON 格式 {@link IotDeviceMessage} 编解码器 14 + * TCP/UDP JSON 格式 {@link IotDeviceMessage} 编解码器
15 * 15 *
16 * 采用纯 JSON 格式传输，格式如下： 16 * 采用纯 JSON 格式传输，格式如下：
17 * { 17 * {
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/router/IotTcpUpstreamHandler.java
  View file @63d7bfe
@@ -10,7 +10,10 @@ import cn.iocoder.yudao.framework.common.util.json.JsonUtils; @@ -10,7 +10,10 @@ import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
10 import cn.iocoder.yudao.module.iot.core.biz.IotDeviceCommonApi; 10 import cn.iocoder.yudao.module.iot.core.biz.IotDeviceCommonApi;
11 import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceAuthReqDTO; 11 import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceAuthReqDTO;
12 import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO; 12 import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO;
  13 +import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
13 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; 14 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  15 +import cn.iocoder.yudao.module.iot.core.topic.auth.IotDeviceRegisterReqDTO;
  16 +import cn.iocoder.yudao.module.iot.core.topic.auth.IotDeviceRegisterRespDTO;
14 import cn.iocoder.yudao.module.iot.core.topic.IotDeviceIdentity; 17 import cn.iocoder.yudao.module.iot.core.topic.IotDeviceIdentity;
15 import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils; 18 import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils;
16 import cn.iocoder.yudao.module.iot.gateway.codec.tcp.IotTcpBinaryDeviceMessageCodec; 19 import cn.iocoder.yudao.module.iot.gateway.codec.tcp.IotTcpBinaryDeviceMessageCodec;
@@ -120,6 +123,9 @@ public class IotTcpUpstreamHandler implements Handler&lt;NetSocket&gt; { @@ -120,6 +123,9 @@ public class IotTcpUpstreamHandler implements Handler&lt;NetSocket&gt; {
120 if (AUTH_METHOD.equals(message.getMethod())) { 123 if (AUTH_METHOD.equals(message.getMethod())) {
121 // 认证请求 124 // 认证请求
122 handleAuthenticationRequest(clientId, message, codecType, socket); 125 handleAuthenticationRequest(clientId, message, codecType, socket);
  126 + } else if (IotDeviceMessageMethodEnum.DEVICE_REGISTER.getMethod().equals(message.getMethod())) {
  127 + // 设备动态注册请求
  128 + handleRegisterRequest(clientId, message, codecType, socket);
123 } else { 129 } else {
124 // 业务消息 130 // 业务消息
125 handleBusinessRequest(clientId, message, codecType, socket); 131 handleBusinessRequest(clientId, message, codecType, socket);
@@ -191,6 +197,44 @@ public class IotTcpUpstreamHandler implements Handler&lt;NetSocket&gt; { @@ -191,6 +197,44 @@ public class IotTcpUpstreamHandler implements Handler&lt;NetSocket&gt; {
191 } 197 }
192 198
193 /** 199 /**
  200 + * 处理设备动态注册请求（一型一密，不需要认证）
  201 + *
  202 + * @param clientId 客户端 ID
  203 + * @param message 消息信息
  204 + * @param codecType 消息编解码类型
  205 + * @param socket 网络连接
  206 + * @see <a href="https://help.aliyun.com/zh/iot/user-guide/unique-certificate-per-product-verification">阿里云 - 一型一密</a>
  207 + */
  208 + private void handleRegisterRequest(String clientId, IotDeviceMessage message, String codecType,
  209 + NetSocket socket) {
  210 + try {
  211 + // 1. 解析注册参数
  212 + IotDeviceRegisterReqDTO registerParams = parseRegisterParams(message.getParams());
  213 + if (registerParams == null) {
  214 + log.warn("[handleRegisterRequest][注册参数解析失败，客户端 ID: {}]", clientId);
  215 + sendErrorResponse(socket, message.getRequestId(), "注册参数不完整", codecType);
  216 + return;
  217 + }
  218 +
  219 + // 2. 调用动态注册
  220 + CommonResult<IotDeviceRegisterRespDTO> result = deviceApi.registerDevice(registerParams);
  221 + if (result.isError()) {
  222 + log.warn("[handleRegisterRequest][注册失败，客户端 ID: {}，错误: {}]", clientId, result.getMsg());
  223 + sendErrorResponse(socket, message.getRequestId(), result.getMsg(), codecType);
  224 + return;
  225 + }
  226 +
  227 + // 3. 发送成功响应（包含 deviceSecret）
  228 + sendRegisterSuccessResponse(socket, message.getRequestId(), result.getData(), codecType);
  229 + log.info("[handleRegisterRequest][注册成功，客户端 ID: {}，设备名: {}]",
  230 + clientId, registerParams.getDeviceName());
  231 + } catch (Exception e) {
  232 + log.error("[handleRegisterRequest][注册处理异常，客户端 ID: {}]", clientId, e);
  233 + sendErrorResponse(socket, message.getRequestId(), "注册处理异常", codecType);
  234 + }
  235 + }
  236 +
  237 + /**
194 * 处理业务请求 238 * 处理业务请求
195 * 239 *
196 * @param clientId 客户端 ID 240 * @param clientId 客户端 ID
@@ -405,4 +449,73 @@ public class IotTcpUpstreamHandler implements Handler&lt;NetSocket&gt; { @@ -405,4 +449,73 @@ public class IotTcpUpstreamHandler implements Handler&lt;NetSocket&gt; {
405 } 449 }
406 } 450 }
407 451
  452 + /**
  453 + * 解析注册参数
  454 + *
  455 + * @param params 参数对象（通常为 Map 类型）
  456 + * @return 注册参数 DTO，解析失败时返回 null
  457 + */
  458 + @SuppressWarnings("unchecked")
  459 + private IotDeviceRegisterReqDTO parseRegisterParams(Object params) {
  460 + if (params == null) {
  461 + return null;
  462 + }
  463 +
  464 + try {
  465 + // 参数默认为 Map 类型，直接转换
  466 + if (params instanceof java.util.Map) {
  467 + java.util.Map<String, Object> paramMap = (java.util.Map<String, Object>) params;
  468 + String productKey = MapUtil.getStr(paramMap, "productKey");
  469 + String deviceName = MapUtil.getStr(paramMap, "deviceName");
  470 + String productSecret = MapUtil.getStr(paramMap, "productSecret");
  471 + if (StrUtil.hasBlank(productKey, deviceName, productSecret)) {
  472 + return null;
  473 + }
  474 + return new IotDeviceRegisterReqDTO()
  475 + .setProductKey(productKey)
  476 + .setDeviceName(deviceName)
  477 + .setProductSecret(productSecret);
  478 + }
  479 +
  480 + // 如果已经是目标类型，直接返回
  481 + if (params instanceof IotDeviceRegisterReqDTO) {
  482 + return (IotDeviceRegisterReqDTO) params;
  483 + }
  484 +
  485 + // 其他情况尝试 JSON 转换
  486 + String jsonStr = JsonUtils.toJsonString(params);
  487 + return JsonUtils.parseObject(jsonStr, IotDeviceRegisterReqDTO.class);
  488 + } catch (Exception e) {
  489 + log.error("[parseRegisterParams][解析注册参数({})失败]", params, e);
  490 + return null;
  491 + }
  492 + }
  493 +
  494 + /**
  495 + * 发送注册成功响应（包含 deviceSecret）
  496 + *
  497 + * @param socket 网络连接
  498 + * @param requestId 请求 ID
  499 + * @param registerResp 注册响应
  500 + * @param codecType 消息编解码类型
  501 + */
  502 + private void sendRegisterSuccessResponse(NetSocket socket, String requestId,
  503 + IotDeviceRegisterRespDTO registerResp, String codecType) {
  504 + try {
  505 + // 构建响应数据
  506 + Object responseData = MapUtil.builder()
  507 + .put("success", true)
  508 + .put("deviceSecret", registerResp.getDeviceSecret())
  509 + .put("message", "注册成功")
  510 + .build();
  511 + IotDeviceMessage responseMessage = IotDeviceMessage.replyOf(requestId,
  512 + IotDeviceMessageMethodEnum.DEVICE_REGISTER.getMethod(), responseData, 0, "注册成功");
  513 +
  514 + byte[] encodedData = deviceMessageService.encodeDeviceMessage(responseMessage, codecType);
  515 + socket.write(Buffer.buffer(encodedData));
  516 + } catch (Exception e) {
  517 + log.error("[sendRegisterSuccessResponse][发送注册成功响应失败，requestId: {}]", requestId, e);
  518 + }
  519 + }
  520 +
408 } 521 }
409 \ No newline at end of file 522 \ No newline at end of file
yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/IotDirectDeviceTcpProtocolIntegrationTest.java
  View file @63d7bfe
@@ -6,6 +6,7 @@ import cn.hutool.core.util.IdUtil; @@ -6,6 +6,7 @@ import cn.hutool.core.util.IdUtil;
6 import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceAuthReqDTO; 6 import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceAuthReqDTO;
7 import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum; 7 import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
8 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; 8 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  9 +import cn.iocoder.yudao.module.iot.core.topic.auth.IotDeviceRegisterReqDTO;
9 import cn.iocoder.yudao.module.iot.core.topic.event.IotDeviceEventPostReqDTO; 10 import cn.iocoder.yudao.module.iot.core.topic.event.IotDeviceEventPostReqDTO;
10 import cn.iocoder.yudao.module.iot.core.topic.property.IotDevicePropertyPostReqDTO; 11 import cn.iocoder.yudao.module.iot.core.topic.property.IotDevicePropertyPostReqDTO;
11 import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils; 12 import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils;
@@ -37,6 +38,7 @@ import java.net.Socket; @@ -37,6 +38,7 @@ import java.net.Socket;
37 * <li>运行以下测试方法： 38 * <li>运行以下测试方法：
38 * <ul> 39 * <ul>
39 * <li>{@link #testAuth()} - 设备认证</li> 40 * <li>{@link #testAuth()} - 设备认证</li>
  41 + * <li>{@link #testDeviceRegister()} - 设备动态注册（一型一密）</li>
40 * <li>{@link #testPropertyPost()} - 设备属性上报</li> 42 * <li>{@link #testPropertyPost()} - 设备属性上报</li>
41 * <li>{@link #testEventPost()} - 设备事件上报</li> 43 * <li>{@link #testEventPost()} - 设备事件上报</li>
42 * </ul> 44 * </ul>
@@ -98,6 +100,46 @@ public class IotDirectDeviceTcpProtocolIntegrationTest { @@ -98,6 +100,46 @@ public class IotDirectDeviceTcpProtocolIntegrationTest {
98 } 100 }
99 } 101 }
100 102
  103 + // ===================== 动态注册测试 =====================
  104 +
  105 + /**
  106 + * 直连设备动态注册测试（一型一密）
  107 + * <p>
  108 + * 使用产品密钥（productSecret）验证身份，成功后返回设备密钥（deviceSecret）
  109 + * <p>
  110 + * 注意：此接口不需要认证
  111 + */
  112 + @Test
  113 + public void testDeviceRegister() throws Exception {
  114 + // 1.1 构建注册消息
  115 + IotDeviceRegisterReqDTO registerReqDTO = new IotDeviceRegisterReqDTO();
  116 + registerReqDTO.setProductKey(PRODUCT_KEY);
  117 + registerReqDTO.setDeviceName("test-tcp-" + System.currentTimeMillis());
  118 + registerReqDTO.setProductSecret("test-product-secret");
  119 + IotDeviceMessage request = IotDeviceMessage.of(IdUtil.fastSimpleUUID(),
  120 + IotDeviceMessageMethodEnum.DEVICE_REGISTER.getMethod(), registerReqDTO, null, null, null);
  121 + // 1.2 编码
  122 + byte[] payload = CODEC.encode(request);
  123 + log.info("[testDeviceRegister][Codec: {}, 请求消息: {}, 数据包长度: {} 字节]", CODEC.type(), request, payload.length);
  124 + if (CODEC instanceof IotTcpBinaryDeviceMessageCodec) {
  125 + log.info("[testDeviceRegister][二进制数据包(HEX): {}]", HexUtil.encodeHexStr(payload));
  126 + }
  127 +
  128 + // 2.1 发送请求
  129 + try (Socket socket = new Socket(SERVER_HOST, SERVER_PORT)) {
  130 + socket.setSoTimeout(TIMEOUT_MS);
  131 + byte[] responseBytes = sendAndReceive(socket, payload);
  132 + // 2.2 解码响应
  133 + if (responseBytes != null) {
  134 + IotDeviceMessage response = CODEC.decode(responseBytes);
  135 + log.info("[testDeviceRegister][响应消息: {}]", response);
  136 + log.info("[testDeviceRegister][成功后可使用返回的 deviceSecret 进行一机一密认证]");
  137 + } else {
  138 + log.warn("[testDeviceRegister][未收到响应]");
  139 + }
  140 + }
  141 + }
  142 +
101 // ===================== 直连设备属性上报测试 ===================== 143 // ===================== 直连设备属性上报测试 =====================
102 144
103 /** 145 /**
yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/IotGatewayDeviceTcpProtocolIntegrationTest.java 0 → 100644
  View file @63d7bfe
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.tcp;
  2 +
  3 +import cn.hutool.core.map.MapUtil;
  4 +import cn.hutool.core.util.HexUtil;
  5 +import cn.hutool.core.util.IdUtil;
  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.mq.message.IotDeviceMessage;
  9 +import cn.iocoder.yudao.module.iot.core.topic.IotDeviceIdentity;
  10 +import cn.iocoder.yudao.module.iot.core.topic.auth.IotSubDeviceRegisterReqDTO;
  11 +import cn.iocoder.yudao.module.iot.core.topic.property.IotDevicePropertyPackPostReqDTO;
  12 +import cn.iocoder.yudao.module.iot.core.topic.topo.IotDeviceTopoAddReqDTO;
  13 +import cn.iocoder.yudao.module.iot.core.topic.topo.IotDeviceTopoDeleteReqDTO;
  14 +import cn.iocoder.yudao.module.iot.core.topic.topo.IotDeviceTopoGetReqDTO;
  15 +import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils;
  16 +import cn.iocoder.yudao.module.iot.gateway.codec.IotDeviceMessageCodec;
  17 +import cn.iocoder.yudao.module.iot.gateway.codec.tcp.IotTcpBinaryDeviceMessageCodec;
  18 +import cn.iocoder.yudao.module.iot.gateway.codec.tcp.IotTcpJsonDeviceMessageCodec;
  19 +import lombok.extern.slf4j.Slf4j;
  20 +import org.junit.jupiter.api.Test;
  21 +
  22 +import java.io.InputStream;
  23 +import java.io.OutputStream;
  24 +import java.net.Socket;
  25 +import java.util.Collections;
  26 +import java.util.List;
  27 +import java.util.Map;
  28 +
  29 +/**
  30 + * IoT 网关设备 TCP 协议集成测试（手动测试）
  31 + *
  32 + * <p>测试场景：网关设备（IotProductDeviceTypeEnum 的 GATEWAY 类型）通过 TCP 协议管理子设备拓扑关系
  33 + *
  34 + * <p>支持两种编解码格式：
  35 + * <ul>
  36 + * <li>{@link IotTcpJsonDeviceMessageCodec} - JSON 格式</li>
  37 + * <li>{@link IotTcpBinaryDeviceMessageCodec} - 二进制格式</li>
  38 + * </ul>
  39 + *
  40 + * <p>使用步骤：
  41 + * <ol>
  42 + * <li>启动 yudao-module-iot-gateway 服务（TCP 端口 8091）</li>
  43 + * <li>修改 {@link #CODEC} 选择测试的编解码格式</li>
  44 + * <li>运行以下测试方法：
  45 + * <ul>
  46 + * <li>{@link #testAuth()} - 网关设备认证</li>
  47 + * <li>{@link #testTopoAdd()} - 添加子设备拓扑关系</li>
  48 + * <li>{@link #testTopoDelete()} - 删除子设备拓扑关系</li>
  49 + * <li>{@link #testTopoGet()} - 获取子设备拓扑关系</li>
  50 + * <li>{@link #testSubDeviceRegister()} - 子设备动态注册</li>
  51 + * <li>{@link #testPropertyPackPost()} - 批量上报属性（网关 + 子设备）</li>
  52 + * </ul>
  53 + * </li>
  54 + * </ol>
  55 + *
  56 + * <p>注意：TCP 协议是有状态的长连接，认证成功后同一连接上的后续请求无需再携带认证信息
  57 + *
  58 + * @author 芋道源码
  59 + */
  60 +@Slf4j
  61 +public class IotGatewayDeviceTcpProtocolIntegrationTest {
  62 +
  63 + private static final String SERVER_HOST = "127.0.0.1";
  64 + private static final int SERVER_PORT = 8091;
  65 + private static final int TIMEOUT_MS = 5000;
  66 +
  67 + // ===================== 编解码器选择（修改此处切换 JSON / Binary） =====================
  68 + private static final IotDeviceMessageCodec CODEC = new IotTcpJsonDeviceMessageCodec();
  69 +// private static final IotDeviceMessageCodec CODEC = new IotTcpBinaryDeviceMessageCodec();
  70 +
  71 + // ===================== 网关设备信息（根据实际情况修改，从 iot_device 表查询网关设备） =====================
  72 + private static final String GATEWAY_PRODUCT_KEY = "m6XcS1ZJ3TW8eC0v";
  73 + private static final String GATEWAY_DEVICE_NAME = "sub-ddd";
  74 + private static final String GATEWAY_DEVICE_SECRET = "b3d62c70f8a4495487ed1d35d61ac2b3";
  75 +
  76 + // ===================== 子设备信息（根据实际情况修改，从 iot_device 表查询子设备） =====================
  77 + private static final String SUB_DEVICE_PRODUCT_KEY = "jAufEMTF1W6wnPhn";
  78 + private static final String SUB_DEVICE_NAME = "chazuo-it";
  79 + private static final String SUB_DEVICE_SECRET = "d46ef9b28ab14238b9c00a3a668032af";
  80 +
  81 + // ===================== 认证测试 =====================
  82 +
  83 + /**
  84 + * 网关设备认证测试
  85 + */
  86 + @Test
  87 + public void testAuth() throws Exception {
  88 + // 1.1 构建认证消息
  89 + IotDeviceAuthReqDTO authInfo = IotDeviceAuthUtils.getAuthInfo(
  90 + GATEWAY_PRODUCT_KEY, GATEWAY_DEVICE_NAME, GATEWAY_DEVICE_SECRET);
  91 + IotDeviceAuthReqDTO authReqDTO = new IotDeviceAuthReqDTO()
  92 + .setClientId(authInfo.getClientId())
  93 + .setUsername(authInfo.getUsername())
  94 + .setPassword(authInfo.getPassword());
  95 + IotDeviceMessage request = IotDeviceMessage.of(IdUtil.fastSimpleUUID(), "auth", authReqDTO, null, null, null);
  96 + // 1.2 编码
  97 + byte[] payload = CODEC.encode(request);
  98 + log.info("[testAuth][Codec: {}, 请求消息: {}, 数据包长度: {} 字节]", CODEC.type(), request, payload.length);
  99 + if (CODEC instanceof IotTcpBinaryDeviceMessageCodec) {
  100 + log.info("[testAuth][二进制数据包(HEX): {}]", HexUtil.encodeHexStr(payload));
  101 + }
  102 +
  103 + // 2.1 发送请求
  104 + try (Socket socket = new Socket(SERVER_HOST, SERVER_PORT)) {
  105 + socket.setSoTimeout(TIMEOUT_MS);
  106 + byte[] responseBytes = sendAndReceive(socket, payload);
  107 + // 2.2 解码响应
  108 + if (responseBytes != null) {
  109 + IotDeviceMessage response = CODEC.decode(responseBytes);
  110 + log.info("[testAuth][响应消息: {}]", response);
  111 + } else {
  112 + log.warn("[testAuth][未收到响应]");
  113 + }
  114 + }
  115 + }
  116 +
  117 + // ===================== 拓扑管理测试 =====================
  118 +
  119 + /**
  120 + * 添加子设备拓扑关系测试
  121 + */
  122 + @Test
  123 + public void testTopoAdd() throws Exception {
  124 + try (Socket socket = new Socket(SERVER_HOST, SERVER_PORT)) {
  125 + socket.setSoTimeout(TIMEOUT_MS);
  126 +
  127 + // 1. 先进行认证
  128 + IotDeviceMessage authResponse = authenticate(socket);
  129 + log.info("[testTopoAdd][认证响应: {}]", authResponse);
  130 +
  131 + // 2.1 构建子设备认证信息
  132 + IotDeviceAuthReqDTO subAuthInfo = IotDeviceAuthUtils.getAuthInfo(
  133 + SUB_DEVICE_PRODUCT_KEY, SUB_DEVICE_NAME, SUB_DEVICE_SECRET);
  134 + IotDeviceAuthReqDTO subDeviceAuth = new IotDeviceAuthReqDTO()
  135 + .setClientId(subAuthInfo.getClientId())
  136 + .setUsername(subAuthInfo.getUsername())
  137 + .setPassword(subAuthInfo.getPassword());
  138 + // 2.2 构建请求参数
  139 + IotDeviceTopoAddReqDTO params = new IotDeviceTopoAddReqDTO();
  140 + params.setSubDevices(Collections.singletonList(subDeviceAuth));
  141 + IotDeviceMessage request = IotDeviceMessage.of(
  142 + IdUtil.fastSimpleUUID(),
  143 + IotDeviceMessageMethodEnum.TOPO_ADD.getMethod(),
  144 + params,
  145 + null, null, null);
  146 + // 2.3 编码
  147 + byte[] payload = CODEC.encode(request);
  148 + log.info("[testTopoAdd][Codec: {}, 请求消息: {}]", CODEC.type(), request);
  149 +
  150 + // 3.1 发送请求
  151 + byte[] responseBytes = sendAndReceive(socket, payload);
  152 + // 3.2 解码响应
  153 + if (responseBytes != null) {
  154 + IotDeviceMessage response = CODEC.decode(responseBytes);
  155 + log.info("[testTopoAdd][响应消息: {}]", response);
  156 + } else {
  157 + log.warn("[testTopoAdd][未收到响应]");
  158 + }
  159 + }
  160 + }
  161 +
  162 + /**
  163 + * 删除子设备拓扑关系测试
  164 + */
  165 + @Test
  166 + public void testTopoDelete() throws Exception {
  167 + try (Socket socket = new Socket(SERVER_HOST, SERVER_PORT)) {
  168 + socket.setSoTimeout(TIMEOUT_MS);
  169 +
  170 + // 1. 先进行认证
  171 + IotDeviceMessage authResponse = authenticate(socket);
  172 + log.info("[testTopoDelete][认证响应: {}]", authResponse);
  173 +
  174 + // 2.1 构建请求参数
  175 + IotDeviceTopoDeleteReqDTO params = new IotDeviceTopoDeleteReqDTO();
  176 + params.setSubDevices(Collections.singletonList(
  177 + new IotDeviceIdentity(SUB_DEVICE_PRODUCT_KEY, SUB_DEVICE_NAME)));
  178 + IotDeviceMessage request = IotDeviceMessage.of(
  179 + IdUtil.fastSimpleUUID(),
  180 + IotDeviceMessageMethodEnum.TOPO_DELETE.getMethod(),
  181 + params,
  182 + null, null, null);
  183 + // 2.2 编码
  184 + byte[] payload = CODEC.encode(request);
  185 + log.info("[testTopoDelete][Codec: {}, 请求消息: {}]", CODEC.type(), request);
  186 +
  187 + // 3.1 发送请求
  188 + byte[] responseBytes = sendAndReceive(socket, payload);
  189 + // 3.2 解码响应
  190 + if (responseBytes != null) {
  191 + IotDeviceMessage response = CODEC.decode(responseBytes);
  192 + log.info("[testTopoDelete][响应消息: {}]", response);
  193 + } else {
  194 + log.warn("[testTopoDelete][未收到响应]");
  195 + }
  196 + }
  197 + }
  198 +
  199 + /**
  200 + * 获取子设备拓扑关系测试
  201 + */
  202 + @Test
  203 + public void testTopoGet() throws Exception {
  204 + try (Socket socket = new Socket(SERVER_HOST, SERVER_PORT)) {
  205 + socket.setSoTimeout(TIMEOUT_MS);
  206 +
  207 + // 1. 先进行认证
  208 + IotDeviceMessage authResponse = authenticate(socket);
  209 + log.info("[testTopoGet][认证响应: {}]", authResponse);
  210 +
  211 + // 2.1 构建请求参数
  212 + IotDeviceTopoGetReqDTO params = new IotDeviceTopoGetReqDTO();
  213 + IotDeviceMessage request = IotDeviceMessage.of(
  214 + IdUtil.fastSimpleUUID(),
  215 + IotDeviceMessageMethodEnum.TOPO_GET.getMethod(),
  216 + params,
  217 + null, null, null);
  218 + // 2.2 编码
  219 + byte[] payload = CODEC.encode(request);
  220 + log.info("[testTopoGet][Codec: {}, 请求消息: {}]", CODEC.type(), request);
  221 +
  222 + // 3.1 发送请求
  223 + byte[] responseBytes = sendAndReceive(socket, payload);
  224 + // 3.2 解码响应
  225 + if (responseBytes != null) {
  226 + IotDeviceMessage response = CODEC.decode(responseBytes);
  227 + log.info("[testTopoGet][响应消息: {}]", response);
  228 + } else {
  229 + log.warn("[testTopoGet][未收到响应]");
  230 + }
  231 + }
  232 + }
  233 +
  234 + // ===================== 子设备注册测试 =====================
  235 +
  236 + /**
  237 + * 子设备动态注册测试
  238 + */
  239 + @Test
  240 + public void testSubDeviceRegister() throws Exception {
  241 + try (Socket socket = new Socket(SERVER_HOST, SERVER_PORT)) {
  242 + socket.setSoTimeout(TIMEOUT_MS);
  243 +
  244 + // 1. 先进行认证
  245 + IotDeviceMessage authResponse = authenticate(socket);
  246 + log.info("[testSubDeviceRegister][认证响应: {}]", authResponse);
  247 +
  248 + // 2.1 构建请求参数
  249 + IotSubDeviceRegisterReqDTO subDevice = new IotSubDeviceRegisterReqDTO();
  250 + subDevice.setProductKey(SUB_DEVICE_PRODUCT_KEY);
  251 + subDevice.setDeviceName("mougezishebei");
  252 + IotDeviceMessage request = IotDeviceMessage.of(
  253 + IdUtil.fastSimpleUUID(),
  254 + IotDeviceMessageMethodEnum.SUB_DEVICE_REGISTER.getMethod(),
  255 + Collections.singletonList(subDevice),
  256 + null, null, null);
  257 + // 2.2 编码
  258 + byte[] payload = CODEC.encode(request);
  259 + log.info("[testSubDeviceRegister][Codec: {}, 请求消息: {}]", CODEC.type(), request);
  260 +
  261 + // 3.1 发送请求
  262 + byte[] responseBytes = sendAndReceive(socket, payload);
  263 + // 3.2 解码响应
  264 + if (responseBytes != null) {
  265 + IotDeviceMessage response = CODEC.decode(responseBytes);
  266 + log.info("[testSubDeviceRegister][响应消息: {}]", response);
  267 + } else {
  268 + log.warn("[testSubDeviceRegister][未收到响应]");
  269 + }
  270 + }
  271 + }
  272 +
  273 + // ===================== 批量上报测试 =====================
  274 +
  275 + /**
  276 + * 批量上报属性测试（网关 + 子设备）
  277 + */
  278 + @Test
  279 + public void testPropertyPackPost() throws Exception {
  280 + try (Socket socket = new Socket(SERVER_HOST, SERVER_PORT)) {
  281 + socket.setSoTimeout(TIMEOUT_MS);
  282 +
  283 + // 1. 先进行认证
  284 + IotDeviceMessage authResponse = authenticate(socket);
  285 + log.info("[testPropertyPackPost][认证响应: {}]", authResponse);
  286 +
  287 + // 2.1 构建【网关设备】自身属性
  288 + Map<String, Object> gatewayProperties = MapUtil.<String, Object>builder()
  289 + .put("temperature", 25.5)
  290 + .build();
  291 + // 2.2 构建【网关设备】自身事件
  292 + IotDevicePropertyPackPostReqDTO.EventValue gatewayEvent = new IotDevicePropertyPackPostReqDTO.EventValue();
  293 + gatewayEvent.setValue(MapUtil.builder().put("message", "gateway started").build());
  294 + gatewayEvent.setTime(System.currentTimeMillis());
  295 + Map<String, IotDevicePropertyPackPostReqDTO.EventValue> gatewayEvents = MapUtil.<String, IotDevicePropertyPackPostReqDTO.EventValue>builder()
  296 + .put("statusReport", gatewayEvent)
  297 + .build();
  298 + // 2.3 构建【网关子设备】属性
  299 + Map<String, Object> subDeviceProperties = MapUtil.<String, Object>builder()
  300 + .put("power", 100)
  301 + .build();
  302 + // 2.4 构建【网关子设备】事件
  303 + IotDevicePropertyPackPostReqDTO.EventValue subDeviceEvent = new IotDevicePropertyPackPostReqDTO.EventValue();
  304 + subDeviceEvent.setValue(MapUtil.builder().put("errorCode", 0).build());
  305 + subDeviceEvent.setTime(System.currentTimeMillis());
  306 + Map<String, IotDevicePropertyPackPostReqDTO.EventValue> subDeviceEvents = MapUtil.<String, IotDevicePropertyPackPostReqDTO.EventValue>builder()
  307 + .put("healthCheck", subDeviceEvent)
  308 + .build();
  309 + // 2.5 构建子设备数据
  310 + IotDevicePropertyPackPostReqDTO.SubDeviceData subDeviceData = new IotDevicePropertyPackPostReqDTO.SubDeviceData();
  311 + subDeviceData.setIdentity(new IotDeviceIdentity(SUB_DEVICE_PRODUCT_KEY, SUB_DEVICE_NAME));
  312 + subDeviceData.setProperties(subDeviceProperties);
  313 + subDeviceData.setEvents(subDeviceEvents);
  314 + // 2.6 构建请求参数
  315 + IotDevicePropertyPackPostReqDTO params = new IotDevicePropertyPackPostReqDTO();
  316 + params.setProperties(gatewayProperties);
  317 + params.setEvents(gatewayEvents);
  318 + params.setSubDevices(List.of(subDeviceData));
  319 + IotDeviceMessage request = IotDeviceMessage.of(
  320 + IdUtil.fastSimpleUUID(),
  321 + IotDeviceMessageMethodEnum.PROPERTY_PACK_POST.getMethod(),
  322 + params,
  323 + null, null, null);
  324 + // 2.7 编码
  325 + byte[] payload = CODEC.encode(request);
  326 + log.info("[testPropertyPackPost][Codec: {}, 请求消息: {}]", CODEC.type(), request);
  327 +
  328 + // 3.1 发送请求
  329 + byte[] responseBytes = sendAndReceive(socket, payload);
  330 + // 3.2 解码响应
  331 + if (responseBytes != null) {
  332 + IotDeviceMessage response = CODEC.decode(responseBytes);
  333 + log.info("[testPropertyPackPost][响应消息: {}]", response);
  334 + } else {
  335 + log.warn("[testPropertyPackPost][未收到响应]");
  336 + }
  337 + }
  338 + }
  339 +
  340 + // ===================== 辅助方法 =====================
  341 +
  342 + /**
  343 + * 执行网关设备认证
  344 + */
  345 + private IotDeviceMessage authenticate(Socket socket) throws Exception {
  346 + IotDeviceAuthReqDTO authInfo = IotDeviceAuthUtils.getAuthInfo(
  347 + GATEWAY_PRODUCT_KEY, GATEWAY_DEVICE_NAME, GATEWAY_DEVICE_SECRET);
  348 + IotDeviceAuthReqDTO authReqDTO = new IotDeviceAuthReqDTO()
  349 + .setClientId(authInfo.getClientId())
  350 + .setUsername(authInfo.getUsername())
  351 + .setPassword(authInfo.getPassword());
  352 + IotDeviceMessage request = IotDeviceMessage.of(IdUtil.fastSimpleUUID(), "auth", authReqDTO, null, null, null);
  353 + byte[] payload = CODEC.encode(request);
  354 + byte[] responseBytes = sendAndReceive(socket, payload);
  355 + if (responseBytes != null) {
  356 + return CODEC.decode(responseBytes);
  357 + }
  358 + return null;
  359 + }
  360 +
  361 + /**
  362 + * 发送 TCP 请求并接收响应
  363 + */
  364 + private byte[] sendAndReceive(Socket socket, byte[] payload) throws Exception {
  365 + // 1. 发送请求
  366 + OutputStream out = socket.getOutputStream();
  367 + InputStream in = socket.getInputStream();
  368 + out.write(payload);
  369 + out.flush();
  370 +
  371 + // 2.1 等待一小段时间让服务器处理
  372 + Thread.sleep(100);
  373 + // 2.2 接收响应
  374 + byte[] buffer = new byte[4096];
  375 + try {
  376 + int length = in.read(buffer);
  377 + if (length > 0) {
  378 + byte[] response = new byte[length];
  379 + System.arraycopy(buffer, 0, response, 0, length);
  380 + return response;
  381 + }
  382 + return null;
  383 + } catch (java.net.SocketTimeoutException e) {
  384 + log.warn("[sendAndReceive][接收响应超时]");
  385 + return null;
  386 + }
  387 + }
  388 +
  389 +}
yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/IotGatewaySubDeviceTcpProtocolIntegrationTest.java 0 → 100644
  View file @63d7bfe
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.tcp;
  2 +
  3 +import cn.hutool.core.map.MapUtil;
  4 +import cn.hutool.core.util.HexUtil;
  5 +import cn.hutool.core.util.IdUtil;
  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.mq.message.IotDeviceMessage;
  9 +import cn.iocoder.yudao.module.iot.core.topic.event.IotDeviceEventPostReqDTO;
  10 +import cn.iocoder.yudao.module.iot.core.topic.property.IotDevicePropertyPostReqDTO;
  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.tcp.IotTcpBinaryDeviceMessageCodec;
  14 +import cn.iocoder.yudao.module.iot.gateway.codec.tcp.IotTcpJsonDeviceMessageCodec;
  15 +import lombok.extern.slf4j.Slf4j;
  16 +import org.junit.jupiter.api.Test;
  17 +
  18 +import java.io.InputStream;
  19 +import java.io.OutputStream;
  20 +import java.net.Socket;
  21 +
  22 +/**
  23 + * IoT 网关子设备 TCP 协议集成测试（手动测试）
  24 + *
  25 + * <p>测试场景：子设备（IotProductDeviceTypeEnum 的 SUB 类型）通过网关设备代理上报数据
  26 + *
  27 + * <p><b>重要说明：子设备无法直接连接平台，所有请求均由网关设备（Gateway）代为转发。</b>
  28 + *
  29 + * <p>支持两种编解码格式：
  30 + * <ul>
  31 + * <li>{@link IotTcpJsonDeviceMessageCodec} - JSON 格式</li>
  32 + * <li>{@link IotTcpBinaryDeviceMessageCodec} - 二进制格式</li>
  33 + * </ul>
  34 + *
  35 + * <p>使用步骤：
  36 + * <ol>
  37 + * <li>启动 yudao-module-iot-gateway 服务（TCP 端口 8091）</li>
  38 + * <li>确保子设备已通过 {@link IotGatewayDeviceTcpProtocolIntegrationTest#testTopoAdd()} 绑定到网关</li>
  39 + * <li>修改 {@link #CODEC} 选择测试的编解码格式</li>
  40 + * <li>运行以下测试方法：
  41 + * <ul>
  42 + * <li>{@link #testAuth()} - 子设备认证</li>
  43 + * <li>{@link #testPropertyPost()} - 子设备属性上报（由网关代理转发）</li>
  44 + * <li>{@link #testEventPost()} - 子设备事件上报（由网关代理转发）</li>
  45 + * </ul>
  46 + * </li>
  47 + * </ol>
  48 + *
  49 + * <p>注意：TCP 协议是有状态的长连接，认证成功后同一连接上的后续请求无需再携带认证信息
  50 + *
  51 + * @author 芋道源码
  52 + */
  53 +@Slf4j
  54 +public class IotGatewaySubDeviceTcpProtocolIntegrationTest {
  55 +
  56 + private static final String SERVER_HOST = "127.0.0.1";
  57 + private static final int SERVER_PORT = 8091;
  58 + private static final int TIMEOUT_MS = 5000;
  59 +
  60 + // ===================== 编解码器选择（修改此处切换 JSON / Binary） =====================
  61 + private static final IotDeviceMessageCodec CODEC = new IotTcpJsonDeviceMessageCodec();
  62 +// private static final IotDeviceMessageCodec CODEC = new IotTcpBinaryDeviceMessageCodec();
  63 +
  64 + // ===================== 网关子设备信息（根据实际情况修改，从 iot_device 表查询子设备） =====================
  65 + private static final String PRODUCT_KEY = "jAufEMTF1W6wnPhn";
  66 + private static final String DEVICE_NAME = "chazuo-it";
  67 + private static final String DEVICE_SECRET = "d46ef9b28ab14238b9c00a3a668032af";
  68 +
  69 + // ===================== 认证测试 =====================
  70 +
  71 + /**
  72 + * 子设备认证测试
  73 + */
  74 + @Test
  75 + public void testAuth() throws Exception {
  76 + // 1.1 构建认证消息
  77 + IotDeviceAuthReqDTO authInfo = IotDeviceAuthUtils.getAuthInfo(PRODUCT_KEY, DEVICE_NAME, DEVICE_SECRET);
  78 + IotDeviceAuthReqDTO authReqDTO = new IotDeviceAuthReqDTO()
  79 + .setClientId(authInfo.getClientId())
  80 + .setUsername(authInfo.getUsername())
  81 + .setPassword(authInfo.getPassword());
  82 + IotDeviceMessage request = IotDeviceMessage.of(IdUtil.fastSimpleUUID(), "auth", authReqDTO, null, null, null);
  83 + // 1.2 编码
  84 + byte[] payload = CODEC.encode(request);
  85 + log.info("[testAuth][Codec: {}, 请求消息: {}, 数据包长度: {} 字节]", CODEC.type(), request, payload.length);
  86 + if (CODEC instanceof IotTcpBinaryDeviceMessageCodec) {
  87 + log.info("[testAuth][二进制数据包(HEX): {}]", HexUtil.encodeHexStr(payload));
  88 + }
  89 +
  90 + // 2.1 发送请求
  91 + try (Socket socket = new Socket(SERVER_HOST, SERVER_PORT)) {
  92 + socket.setSoTimeout(TIMEOUT_MS);
  93 + byte[] responseBytes = sendAndReceive(socket, payload);
  94 + // 2.2 解码响应
  95 + if (responseBytes != null) {
  96 + IotDeviceMessage response = CODEC.decode(responseBytes);
  97 + log.info("[testAuth][响应消息: {}]", response);
  98 + } else {
  99 + log.warn("[testAuth][未收到响应]");
  100 + }
  101 + }
  102 + }
  103 +
  104 + // ===================== 子设备属性上报测试 =====================
  105 +
  106 + /**
  107 + * 子设备属性上报测试
  108 + */
  109 + @Test
  110 + public void testPropertyPost() throws Exception {
  111 + try (Socket socket = new Socket(SERVER_HOST, SERVER_PORT)) {
  112 + socket.setSoTimeout(TIMEOUT_MS);
  113 +
  114 + // 1. 先进行认证
  115 + IotDeviceMessage authResponse = authenticate(socket);
  116 + log.info("[testPropertyPost][认证响应: {}]", authResponse);
  117 +
  118 + // 2.1 构建属性上报消息
  119 + IotDeviceMessage request = IotDeviceMessage.of(
  120 + IdUtil.fastSimpleUUID(),
  121 + IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod(),
  122 + IotDevicePropertyPostReqDTO.of(MapUtil.<String, Object>builder()
  123 + .put("power", 100)
  124 + .put("status", "online")
  125 + .put("temperature", 36.5)
  126 + .build()),
  127 + null, null, null);
  128 + // 2.2 编码
  129 + byte[] payload = CODEC.encode(request);
  130 + log.info("[testPropertyPost][子设备属性上报 - 请求实际由 Gateway 代为转发]");
  131 + log.info("[testPropertyPost][Codec: {}, 请求消息: {}]", CODEC.type(), request);
  132 +
  133 + // 3.1 发送请求
  134 + byte[] responseBytes = sendAndReceive(socket, payload);
  135 + // 3.2 解码响应
  136 + if (responseBytes != null) {
  137 + IotDeviceMessage response = CODEC.decode(responseBytes);
  138 + log.info("[testPropertyPost][响应消息: {}]", response);
  139 + } else {
  140 + log.warn("[testPropertyPost][未收到响应]");
  141 + }
  142 + }
  143 + }
  144 +
  145 + // ===================== 子设备事件上报测试 =====================
  146 +
  147 + /**
  148 + * 子设备事件上报测试
  149 + */
  150 + @Test
  151 + public void testEventPost() throws Exception {
  152 + try (Socket socket = new Socket(SERVER_HOST, SERVER_PORT)) {
  153 + socket.setSoTimeout(TIMEOUT_MS);
  154 +
  155 + // 1. 先进行认证
  156 + IotDeviceMessage authResponse = authenticate(socket);
  157 + log.info("[testEventPost][认证响应: {}]", authResponse);
  158 +
  159 + // 2.1 构建事件上报消息
  160 + IotDeviceMessage request = IotDeviceMessage.of(
  161 + IdUtil.fastSimpleUUID(),
  162 + IotDeviceMessageMethodEnum.EVENT_POST.getMethod(),
  163 + IotDeviceEventPostReqDTO.of(
  164 + "alarm",
  165 + MapUtil.<String, Object>builder()
  166 + .put("level", "warning")
  167 + .put("message", "temperature too high")
  168 + .put("threshold", 40)
  169 + .put("current", 42)
  170 + .build(),
  171 + System.currentTimeMillis()),
  172 + null, null, null);
  173 + // 2.2 编码
  174 + byte[] payload = CODEC.encode(request);
  175 + log.info("[testEventPost][子设备事件上报 - 请求实际由 Gateway 代为转发]");
  176 + log.info("[testEventPost][Codec: {}, 请求消息: {}]", CODEC.type(), request);
  177 +
  178 + // 3.1 发送请求
  179 + byte[] responseBytes = sendAndReceive(socket, payload);
  180 + // 3.2 解码响应
  181 + if (responseBytes != null) {
  182 + IotDeviceMessage response = CODEC.decode(responseBytes);
  183 + log.info("[testEventPost][响应消息: {}]", response);
  184 + } else {
  185 + log.warn("[testEventPost][未收到响应]");
  186 + }
  187 + }
  188 + }
  189 +
  190 + // ===================== 辅助方法 =====================
  191 +
  192 + /**
  193 + * 执行子设备认证
  194 + */
  195 + private IotDeviceMessage authenticate(Socket socket) throws Exception {
  196 + IotDeviceAuthReqDTO authInfo = IotDeviceAuthUtils.getAuthInfo(PRODUCT_KEY, DEVICE_NAME, DEVICE_SECRET);
  197 + IotDeviceAuthReqDTO authReqDTO = new IotDeviceAuthReqDTO()
  198 + .setClientId(authInfo.getClientId())
  199 + .setUsername(authInfo.getUsername())
  200 + .setPassword(authInfo.getPassword());
  201 + IotDeviceMessage request = IotDeviceMessage.of(IdUtil.fastSimpleUUID(), "auth", authReqDTO, null, null, null);
  202 + byte[] payload = CODEC.encode(request);
  203 + byte[] responseBytes = sendAndReceive(socket, payload);
  204 + if (responseBytes != null) {
  205 + return CODEC.decode(responseBytes);
  206 + }
  207 + return null;
  208 + }
  209 +
  210 + /**
  211 + * 发送 TCP 请求并接收响应
  212 + */
  213 + private byte[] sendAndReceive(Socket socket, byte[] payload) throws Exception {
  214 + // 1. 发送请求
  215 + OutputStream out = socket.getOutputStream();
  216 + InputStream in = socket.getInputStream();
  217 + out.write(payload);
  218 + out.flush();
  219 +
  220 + // 2.1 等待一小段时间让服务器处理
  221 + Thread.sleep(100);
  222 + // 2.2 接收响应
  223 + byte[] buffer = new byte[4096];
  224 + try {
  225 + int length = in.read(buffer);
  226 + if (length > 0) {
  227 + byte[] response = new byte[length];
  228 + System.arraycopy(buffer, 0, response, 0, length);
  229 + return response;
  230 + }
  231 + return null;
  232 + } catch (java.net.SocketTimeoutException e) {
  233 + log.warn("[sendAndReceive][接收响应超时]");
  234 + return null;
  235 + }
  236 + }
  237 +
  238 +}