BJ_ZTEITS-Developers / parking-cloud-yudao

Commit 666afcc3e7a2866459eb35840e8d52445234905e

Authored by haohao
1 parent 233a2f7d

refactor(iot): 影子透传按设备协议序列化，reply 对齐 CommonResult 

- Gateway 透传编解码改为 IotMessageSerializer.encodeRaw/decodeRaw，移除 IotShadowPassthroughCodec
- 下行 reply 扁平为 { method, code, msg, data }，version/timestamp 收进 data（Breaking）
- 扩展 IotDeviceShadowProtocolUtils，biz/gateway 测试同步适配
Inline Side-by-side
Showing 15 changed files with 325 additions and 216 deletions
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImpl.java
  View file @666afcc
... ... @@ -7,10 +7,10 @@ import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
7 7 import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
8 8 import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
9 9 import cn.iocoder.yudao.module.iot.core.enums.shadow.IotDeviceShadowErrorCodeEnum;
10   -import cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils;
11 10 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
12 11 import cn.iocoder.yudao.module.iot.core.topic.shadow.IotDeviceShadowReplyDTO;
13 12 import cn.iocoder.yudao.module.iot.core.topic.shadow.IotDeviceShadowUpdateReqDTO;
  13 +import cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils;
14 14 import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
15 15 import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceShadowDO;
16 16 import cn.iocoder.yudao.module.iot.dal.mysql.device.IotDeviceShadowMapper;
... ... @@ -32,8 +32,7 @@ import java.util.Map;
32 32 import java.util.function.Function;
33 33  
34 34 import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
35   -import static cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils.INNER_METHOD_GET;
36   -import static cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils.INNER_METHOD_UPDATE;
  35 +import static cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils.*;
37 36 import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.DEVICE_SHADOW_NOT_ENABLED;
38 37 import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.DEVICE_SHADOW_VERSION_CONFLICT;
39 38  
... ... @@ -89,23 +88,26 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService {
89 88 } catch (Exception ex) {
90 89 log.error("[handleDeviceShadowUpdate][device({}) message({}) 处理异常]", device.getId(), message, ex);
91 90 Long version = resolveShadowVersionSafely(device.getId());
92   - sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf(version,
  91 + sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf(
  92 + extractInnerMethod(message.getParams()), version,
93 93 IotDeviceShadowErrorCodeEnum.SERVER_ERROR.getCode(),
94 94 IotDeviceShadowErrorCodeEnum.SERVER_ERROR.getMessage(), DateUtil.currentSeconds()));
95 95 }
96 96 }
97 97  
98 98 private void handleDeviceShadowUpdate0(IotDeviceDO device, IotDeviceMessage message) {
99   - // 1. 解析请求
  99 + // 1. 解析请求（影子 body 当前按 JSON 建模，非 JSON 设备在 gateway 透传层已拒绝）
100 100 IotDeviceShadowUpdateReqDTO req = JsonUtils.convertObject(message.getParams(), IotDeviceShadowUpdateReqDTO.class);
101 101 if (req == null) {
102   - sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf(0L,
  102 + sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf(
  103 + extractInnerMethod(message.getParams()), 0L,
103 104 IotDeviceShadowErrorCodeEnum.JSON_FORMAT_ERROR.getCode(),
104 105 IotDeviceShadowErrorCodeEnum.JSON_FORMAT_ERROR.getMessage(), DateUtil.currentSeconds()));
105 106 return;
106 107 }
107 108 if (StrUtil.isEmpty(req.getMethod())) {
108   - sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf(resolveShadowVersionSafely(device.getId()),
  109 + sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf(
  110 + INNER_METHOD_UPDATE, resolveShadowVersionSafely(device.getId()),
109 111 IotDeviceShadowErrorCodeEnum.METHOD_MISSING.getCode(),
110 112 IotDeviceShadowErrorCodeEnum.METHOD_MISSING.getMessage(), DateUtil.currentSeconds()));
111 113 return;
... ... @@ -128,7 +130,8 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService {
128 130 handleReportedUpdate(device, message.getServerId(), req);
129 131 return;
130 132 }
131   - sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf(shadow.getVersion(),
  133 + sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf(
  134 + req.getMethod(), shadow.getVersion(),
132 135 IotDeviceShadowErrorCodeEnum.INVALID_METHOD.getCode(),
133 136 IotDeviceShadowErrorCodeEnum.INVALID_METHOD.getMessage(), DateUtil.currentSeconds()));
134 137 }
... ... @@ -137,19 +140,22 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService {
137 140 Long deviceId = device.getId();
138 141  
139 142 if (req.getState() == null) {
140   - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(resolveShadowVersionSafely(deviceId),
  143 + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(
  144 + INNER_METHOD_UPDATE, resolveShadowVersionSafely(deviceId),
141 145 IotDeviceShadowErrorCodeEnum.STATE_MISSING.getCode(),
142 146 IotDeviceShadowErrorCodeEnum.STATE_MISSING.getMessage(), DateUtil.currentSeconds()));
143 147 return;
144 148 }
145 149 if (req.getState().getReported() == null) {
146   - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(resolveShadowVersionSafely(deviceId),
  150 + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(
  151 + INNER_METHOD_UPDATE, resolveShadowVersionSafely(deviceId),
147 152 IotDeviceShadowErrorCodeEnum.REPORTED_MISSING.getCode(),
148 153 IotDeviceShadowErrorCodeEnum.REPORTED_MISSING.getMessage(), DateUtil.currentSeconds()));
149 154 return;
150 155 }
151 156 if (CollUtil.isEmpty(req.getState().getReported())) {
152   - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(resolveShadowVersionSafely(deviceId),
  157 + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(
  158 + INNER_METHOD_UPDATE, resolveShadowVersionSafely(deviceId),
153 159 IotDeviceShadowErrorCodeEnum.REPORTED_EMPTY.getCode(),
154 160 IotDeviceShadowErrorCodeEnum.REPORTED_EMPTY.getMessage(), DateUtil.currentSeconds()));
155 161 return;
... ... @@ -171,7 +177,8 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService {
171 177 sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.updateReplyOf(committed.getVersion(),
172 178 committed.getShadowTime()));
173 179 } catch (ShadowUpdateRejectedException ex) {
174   - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(ex.getVersion(),
  180 + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(
  181 + INNER_METHOD_UPDATE, ex.getVersion(),
175 182 ex.getErrorCode().getCode(), ex.getErrorCode().getMessage(), DateUtil.currentSeconds()));
176 183 } catch (ShadowCasExhaustedException ex) {
177 184 handleCasExhaustedForDevice(device, serverId, ex);
... ... @@ -340,7 +347,8 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService {
340 347  
341 348 private void handleCasExhaustedForDevice(IotDeviceDO device, String serverId, ShadowCasExhaustedException ex) {
342 349 if (ex.isShadowNotFound()) {
343   - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(0L,
  350 + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(
  351 + INNER_METHOD_UPDATE, 0L,
344 352 IotDeviceShadowErrorCodeEnum.SERVER_ERROR.getCode(),
345 353 IotDeviceShadowErrorCodeEnum.SERVER_ERROR.getMessage(), DateUtil.currentSeconds()));
346 354 return;
... ... @@ -349,7 +357,8 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService {
349 357 }
350 358  
351 359 private void sendVersionConflictReply(IotDeviceDO device, String serverId, Long version) {
352   - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(version,
  360 + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(
  361 + INNER_METHOD_UPDATE, version,
353 362 IotDeviceShadowErrorCodeEnum.VERSION_CONFLICT.getCode(),
354 363 IotDeviceShadowErrorCodeEnum.VERSION_CONFLICT.getMessage(), DateUtil.currentSeconds()));
355 364 }
... ...
yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImplTest.java
  View file @666afcc
... ... @@ -85,7 +85,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
85 85 verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1"));
86 86 IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams();
87 87 assertEquals("get_reply", reply.getMethod());
88   - assertEquals(2L, reply.getVersion());
  88 + assertEquals(2L, getReplyVersion(reply));
89 89 }
90 90  
91 91 @Test
... ... @@ -142,9 +142,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
142 142 ArgumentCaptor<IotDeviceMessage> captor = ArgumentCaptor.forClass(IotDeviceMessage.class);
143 143 verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1"));
144 144 IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams();
145   - @SuppressWarnings("unchecked")
146   - Map<String, Object> payload = (Map<String, Object>) reply.getPayload();
147   - assertEquals(409, payload.get("code"));
  145 + assertEquals(409, reply.getCode());
148 146 verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any());
149 147 }
150 148  
... ... @@ -196,9 +194,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
196 194 verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1"));
197 195 IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams();
198 196 assertEquals("update_reply", reply.getMethod());
199   - @SuppressWarnings("unchecked")
200   - Map<String, Object> payload = (Map<String, Object>) reply.getPayload();
201   - assertEquals(409, payload.get("code"));
  197 + assertEquals(409, reply.getCode());
202 198 verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any());
203 199 }
204 200  
... ... @@ -268,7 +264,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
268 264 verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1"));
269 265 IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams();
270 266 assertEquals("get_reply", reply.getMethod());
271   - assertEquals(2L, reply.getVersion());
  267 + assertEquals(2L, getReplyVersion(reply));
272 268 // getOrCreateShadow + CAS 各一次 SELECT，推送不再额外查库
273 269 verify(deviceShadowMapper, times(2)).selectByDeviceId(deviceId);
274 270 verify(deviceShadowRedisDAO).delete(deviceId);
... ... @@ -293,6 +289,21 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
293 289 }
294 290  
295 291 @Test
  292 + public void testHandleDeviceShadowUpdate_invalidParams_send400() {
  293 + IotDeviceDO device = buildDevice();
  294 +
  295 + IotDeviceMessage message = IotDeviceMessage.requestOf(device.getId(), 1L, "server-1",
  296 + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), null);
  297 +
  298 + service.handleDeviceShadowUpdate(device, message);
  299 +
  300 + ArgumentCaptor<IotDeviceMessage> captor = ArgumentCaptor.forClass(IotDeviceMessage.class);
  301 + verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1"));
  302 + IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams();
  303 + assertEquals(400, reply.getCode());
  304 + }
  305 +
  306 + @Test
296 307 public void testHandleDeviceShadowUpdate_methodMissing_send401() {
297 308 IotDeviceDO device = buildDevice();
298 309 IotDeviceShadowDO shadow = buildShadow(device, 3L);
... ... @@ -309,9 +320,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
309 320 ArgumentCaptor<IotDeviceMessage> captor = ArgumentCaptor.forClass(IotDeviceMessage.class);
310 321 verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1"));
311 322 IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams();
312   - @SuppressWarnings("unchecked")
313   - Map<String, Object> payload = (Map<String, Object>) reply.getPayload();
314   - assertEquals(401, payload.get("code"));
  323 + assertEquals(401, reply.getCode());
315 324 }
316 325  
317 326 @Test
... ... @@ -365,7 +374,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
365 374 ArgumentCaptor<IotDeviceMessage> replyCaptor = ArgumentCaptor.forClass(IotDeviceMessage.class);
366 375 verify(deviceMessageService).sendDeviceMessage(replyCaptor.capture(), eq(device), eq("server-1"));
367 376 IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) replyCaptor.getValue().getParams();
368   - assertEquals(3L, reply.getVersion());
  377 + assertEquals(3L, getReplyVersion(reply));
369 378 }
370 379  
371 380 @Test
... ... @@ -429,10 +438,8 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
429 438 ArgumentCaptor<IotDeviceMessage> captor = ArgumentCaptor.forClass(IotDeviceMessage.class);
430 439 verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1"));
431 440 IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams();
432   - assertEquals(0L, reply.getVersion());
433   - @SuppressWarnings("unchecked")
434   - Map<String, Object> payload = (Map<String, Object>) reply.getPayload();
435   - assertEquals(500, payload.get("code"));
  441 + assertEquals(0L, getReplyVersion(reply));
  442 + assertEquals(500, reply.getCode());
436 443 }
437 444  
438 445 @Test
... ... @@ -458,9 +465,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
458 465 ArgumentCaptor<IotDeviceMessage> captor = ArgumentCaptor.forClass(IotDeviceMessage.class);
459 466 verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1"));
460 467 IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams();
461   - @SuppressWarnings("unchecked")
462   - Map<String, Object> payload = (Map<String, Object>) reply.getPayload();
463   - assertEquals(500, payload.get("code"));
  468 + assertEquals(500, reply.getCode());
464 469 verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any());
465 470 }
466 471  
... ... @@ -493,4 +498,13 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
493 498 .build();
494 499 }
495 500  
  501 + @SuppressWarnings("unchecked")
  502 + private static Long getReplyVersion(IotDeviceShadowReplyDTO reply) {
  503 + if (!(reply.getData() instanceof Map)) {
  504 + return null;
  505 + }
  506 + Object version = ((Map<String, Object>) reply.getData()).get("version");
  507 + return version instanceof Number ? ((Number) version).longValue() : null;
  508 + }
  509 +
496 510 }
... ...
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/shadow/IotDeviceShadowErrorCodeEnum.java
  View file @666afcc
... ... @@ -6,7 +6,7 @@ import lombok.Getter;
6 6 /**
7 7 * IoT 设备影子设备侧错误码枚举
8 8 * <p>
9   - * 设备侧 wire code，仅用于回填 {@code update_reply.payload.code/msg}，不走 {@link cn.iocoder.yudao.framework.common.pojo.CommonResult}。
  9 + * 设备侧 wire code，仅用于回填 {@code update_reply.code/msg}，不走 {@link cn.iocoder.yudao.framework.common.pojo.CommonResult}。
10 10 * 服务端内部异常请使用 {@link cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants}（如 {@code DEVICE_SHADOW_NOT_ENABLED}）。
11 11 *
12 12 * @author 芋道源码
... ... @@ -21,7 +21,6 @@ public enum IotDeviceShadowErrorCodeEnum {
21 21 REPORTED_MISSING(404, "缺少 reported"),
22 22 REPORTED_EMPTY(405, "reported 为空"),
23 23 INVALID_METHOD(406, "无效内层 method"),
24   - SHADOW_EMPTY(407, "影子内容为空"),
25 24 PROPERTY_COUNT_EXCEEDED(408, "影子属性数超过限制"),
26 25 VERSION_CONFLICT(409, "影子版本冲突"),
27 26 SERVER_ERROR(500, "服务端异常");
... ...
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/shadow/IotDeviceShadowReplyDTO.java
  View file @666afcc
... ... @@ -14,7 +14,8 @@ import static cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils
14 14 /**
15 15 * IoT 设备影子下行 Reply DTO
16 16 * <p>
17   - * 用于 {@link IotDeviceMessageMethodEnum#SHADOW_GET} 消息的 params 参数（透传原始线 body）
  17 + * 扁平对齐 CommonResult + method，用于 {@link IotDeviceMessageMethodEnum#SHADOW_GET} 透传线 body。
  18 + * 错误回复的 {@link #data} 仍携带 version 供设备 resync，有意偏离纯 CommonResult（error 时 data 通常为 null）。
18 19 *
19 20 * @author 芋道源码
20 21 */
... ... @@ -29,19 +30,19 @@ public class IotDeviceShadowReplyDTO {
29 30 private String method;
30 31  
31 32 /**
32   - * 回复载荷（对齐 CommonResult：code / msg / data）
  33 + * 结果码，0 表示成功
33 34 */
34   - private Object payload;
  35 + private Integer code;
35 36  
36 37 /**
37   - * 影子版本号
  38 + * 结果描述
38 39 */
39   - private Long version;
  40 + private String msg;
40 41  
41 42 /**
42   - * 时间戳（epoch 秒）
  43 + * 业务数据：get_reply 为 {state, version, timestamp}；update_reply 为 {version, timestamp}
43 44 */
44   - private Long timestamp;
  45 + private Object data;
45 46  
46 47 public static IotDeviceShadowReplyDTO getReplyOf(Map<String, Object> desired, Map<String, Object> reported,
47 48 Long version, Long timestamp) {
... ... @@ -50,45 +51,29 @@ public class IotDeviceShadowReplyDTO {
50 51 state.put(STATE_REPORTED, reported != null ? reported : new HashMap<>());
51 52 Map<String, Object> data = new HashMap<>();
52 53 data.put(PAYLOAD_STATE, state);
53   - Map<String, Object> payload = buildSuccessPayload(data);
54   - return new IotDeviceShadowReplyDTO()
55   - .setMethod(INNER_METHOD_GET_REPLY)
56   - .setPayload(payload)
57   - .setVersion(version)
58   - .setTimestamp(timestamp);
  54 + data.put(PAYLOAD_VERSION, version);
  55 + data.put(METADATA_TIMESTAMP, timestamp);
  56 + return of(INNER_METHOD_GET_REPLY, SUCCESS.getCode(), SUCCESS.getMsg(), data);
59 57 }
60 58  
61 59 public static IotDeviceShadowReplyDTO updateReplyOf(Long version, Long timestamp) {
62   - Map<String, Object> data = new HashMap<>();
63   - data.put(PAYLOAD_VERSION, version);
64   - Map<String, Object> payload = buildSuccessPayload(data);
65   - return new IotDeviceShadowReplyDTO()
66   - .setMethod(INNER_METHOD_UPDATE_REPLY)
67   - .setPayload(payload)
68   - .setVersion(version)
69   - .setTimestamp(timestamp);
  60 + return of(INNER_METHOD_UPDATE_REPLY, SUCCESS.getCode(), SUCCESS.getMsg(), dataWithVersionTimestamp(version, timestamp));
70 61 }
71 62  
72   - public static IotDeviceShadowReplyDTO errorOf(Long version, Integer errorCode, String errorMsg, Long timestamp) {
73   - Map<String, Object> data = new HashMap<>();
74   - data.put(PAYLOAD_VERSION, version);
75   - Map<String, Object> payload = new HashMap<>();
76   - payload.put(PAYLOAD_CODE, errorCode);
77   - payload.put(PAYLOAD_MSG, errorMsg);
78   - payload.put(PAYLOAD_DATA, data);
79   - return new IotDeviceShadowReplyDTO()
80   - .setMethod(INNER_METHOD_UPDATE_REPLY)
81   - .setPayload(payload)
82   - .setVersion(version)
83   - .setTimestamp(timestamp);
  63 + public static IotDeviceShadowReplyDTO errorOf(String innerMethod, Long version, Integer errorCode,
  64 + String errorMsg, Long timestamp) {
  65 + return of(toReplyMethod(innerMethod), errorCode, errorMsg, dataWithVersionTimestamp(version, timestamp));
84 66 }
85 67  
86   - private static Map<String, Object> buildSuccessPayload(Map<String, Object> data) {
87   - Map<String, Object> payload = new HashMap<>();
88   - payload.put(PAYLOAD_CODE, SUCCESS.getCode());
89   - payload.put(PAYLOAD_MSG, SUCCESS.getMsg());
90   - payload.put(PAYLOAD_DATA, data);
91   - return payload;
  68 + private static IotDeviceShadowReplyDTO of(String method, Integer code, String msg, Map<String, Object> data) {
  69 + return new IotDeviceShadowReplyDTO().setMethod(method).setCode(code).setMsg(msg).setData(data);
  70 + }
  71 +
  72 + private static Map<String, Object> dataWithVersionTimestamp(Long version, Long timestamp) {
  73 + Map<String, Object> data = new HashMap<>();
  74 + data.put(PAYLOAD_VERSION, version);
  75 + data.put(METADATA_TIMESTAMP, timestamp);
  76 + return data;
92 77 }
93 78  
94 79 }
... ...
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/shadow/IotDeviceShadowUpdateReqDTO.java
  View file @666afcc
... ... @@ -44,11 +44,6 @@ public class IotDeviceShadowUpdateReqDTO {
44 44 */
45 45 private Map<String, Object> reported;
46 46  
47   - /**
48   - * 期望状态
49   - */
50   - private Map<String, Object> desired;
51   -
52 47 }
53 48  
54 49 }
... ...
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceShadowProtocolUtils.java
  View file @666afcc
... ... @@ -31,15 +31,33 @@ public final class IotDeviceShadowProtocolUtils {
31 31 public static final String STATE_REPORTED = "reported";
32 32 public static final String METADATA_TIMESTAMP = "timestamp";
33 33  
34   - // ========== payload 键（对齐 CommonResult：code / msg / data）==========
  34 + // ========== reply data 键 ==========
35 35  
36   - public static final String PAYLOAD_CODE = "code";
37   - public static final String PAYLOAD_MSG = "msg";
38   - public static final String PAYLOAD_DATA = "data";
39 36 public static final String PAYLOAD_STATE = "state";
40 37 public static final String PAYLOAD_VERSION = "version";
41 38  
42 39 /**
  40 + * 由内层 method 推导 reply method（get → get_reply，其余 → update_reply）
  41 + */
  42 + public static String toReplyMethod(String innerMethod) {
  43 + if (INNER_METHOD_GET.equals(innerMethod)) {
  44 + return INNER_METHOD_GET_REPLY;
  45 + }
  46 + return INNER_METHOD_UPDATE_REPLY;
  47 + }
  48 +
  49 + /**
  50 + * 从透传 params 提取内层 method；无法解析时默认 update
  51 + */
  52 + public static String extractInnerMethod(Object params) {
  53 + if (!(params instanceof Map)) {
  54 + return INNER_METHOD_UPDATE;
  55 + }
  56 + Object method = ((Map<?, ?>) params).get("method");
  57 + return method != null ? String.valueOf(method) : INNER_METHOD_UPDATE;
  58 + }
  59 +
  60 + /**
43 61 * 获取 desired 下某属性的更新时间戳
44 62 *
45 63 * @param metadata 元数据
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/upstream/IotEmqxUpstreamHandler.java
  View file @666afcc
... ... @@ -3,12 +3,16 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.emqx.handler.upstream;
3 3 import cn.hutool.core.util.ArrayUtil;
4 4 import cn.hutool.core.util.StrUtil;
5 5 import cn.hutool.extra.spring.SpringUtil;
  6 +import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
6 7 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
7 8 import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
8 9 import cn.iocoder.yudao.module.iot.gateway.util.IotMqttTopicUtils;
9 10 import io.vertx.mqtt.messages.MqttPublishMessage;
10 11 import lombok.extern.slf4j.Slf4j;
11 12  
  13 +import java.util.Arrays;
  14 +import java.util.stream.Collectors;
  15 +
12 16 /**
13 17 * IoT 网关 EMQX 上行消息处理器
14 18 *
... ... @@ -44,13 +48,24 @@ public class IotEmqxUpstreamHandler {
44 48 }
45 49  
46 50 // 2.1 反序列化消息
47   - IotDeviceMessage message = deviceMessageService.deserializeDeviceMessage(payload, productKey, deviceName);
48   - if (message == null) {
49   - log.warn("[handle][topic({}) payload({}) 消息解码失败]", topic, new String(payload));
50   - return;
  51 + String topicMethod = Arrays.stream(topicParts, 4, topicParts.length)
  52 + .collect(Collectors.joining("."));
  53 + IotDeviceMessage message;
  54 + if (IotDeviceMessageMethodEnum.isPassthrough(topicMethod)) {
  55 + message = deviceMessageService.deserializePassthroughMessage(payload, topicMethod, productKey, deviceName);
  56 + if (message == null) {
  57 + log.warn("[handle][topic({}) 影子透传 payload 解析失败]", topic);
  58 + return;
  59 + }
  60 + } else {
  61 + message = deviceMessageService.deserializeDeviceMessage(payload, productKey, deviceName);
  62 + if (message == null) {
  63 + log.warn("[handle][topic({}) payload({}) 消息解码失败]", topic, new String(payload));
  64 + return;
  65 + }
  66 + // 2.2 标准化回复消息的 method（MQTT 协议中，设备回复消息的 method 会携带 _reply 后缀）
  67 + IotMqttTopicUtils.normalizeReplyMethod(message);
51 68 }
52   - // 2.2 标准化回复消息的 method（MQTT 协议中，设备回复消息的 method 会携带 _reply 后缀）
53   - IotMqttTopicUtils.normalizeReplyMethod(message);
54 69  
55 70 // 3. 发送消息到队列
56 71 deviceMessageService.sendDeviceMessage(message, productKey, deviceName, serverId);
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttProtocol.java
  View file @666afcc
... ... @@ -17,7 +17,6 @@ import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.handler.upstream.IotMqt
17 17 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.handler.upstream.IotMqttRegisterHandler;
18 18 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.handler.upstream.IotMqttUpstreamHandler;
19 19 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.manager.IotMqttConnectionManager;
20   -import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService;
21 20 import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
22 21 import cn.iocoder.yudao.module.iot.gateway.util.IotMqttTopicUtils;
23 22 import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
... ... @@ -102,10 +101,9 @@ public class IotMqttProtocol implements IotProtocol {
102 101 // 初始化 Handler
103 102 this.deviceMessageService = SpringUtil.getBean(IotDeviceMessageService.class);
104 103 IotDeviceCommonApi deviceApi = SpringUtil.getBean(IotDeviceCommonApi.class);
105   - IotDeviceService deviceService = SpringUtil.getBean(IotDeviceService.class);
106 104 this.authHandler = new IotMqttAuthHandler(connectionManager, deviceMessageService, deviceApi, serverId);
107 105 this.registerHandler = new IotMqttRegisterHandler(connectionManager, deviceMessageService);
108   - this.upstreamHandler = new IotMqttUpstreamHandler(connectionManager, deviceMessageService, deviceService, serverId);
  106 + this.upstreamHandler = new IotMqttUpstreamHandler(connectionManager, deviceMessageService, serverId);
109 107 }
110 108  
111 109 @Override
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/upstream/IotMqttUpstreamHandler.java
  View file @666afcc
... ... @@ -2,20 +2,15 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.handler.upstream;
2 2  
3 3 import cn.hutool.core.lang.Assert;
4 4 import cn.hutool.core.util.ArrayUtil;
5   -import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO;
6 5 import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
7   -import cn.iocoder.yudao.module.iot.core.enums.IotSerializeTypeEnum;
8 6 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
9 7 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.manager.IotMqttConnectionManager;
10   -import cn.iocoder.yudao.module.iot.gateway.serialize.IotShadowPassthroughCodec;
11   -import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService;
12 8 import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
13 9 import cn.iocoder.yudao.module.iot.gateway.util.IotMqttTopicUtils;
14 10 import io.vertx.mqtt.MqttEndpoint;
15 11 import lombok.extern.slf4j.Slf4j;
16 12  
17 13 import java.util.Arrays;
18   -import java.util.Map;
19 14 import java.util.stream.Collectors;
20 15  
21 16  
... ... @@ -28,14 +23,11 @@ import java.util.stream.Collectors;
28 23 public class IotMqttUpstreamHandler extends IotMqttAbstractHandler {
29 24  
30 25 private final String serverId;
31   - private final IotDeviceService deviceService;
32 26  
33 27 public IotMqttUpstreamHandler(IotMqttConnectionManager connectionManager,
34 28 IotDeviceMessageService deviceMessageService,
35   - IotDeviceService deviceService,
36 29 String serverId) {
37 30 super(connectionManager, deviceMessageService);
38   - this.deviceService = deviceService;
39 31 this.serverId = serverId;
40 32 }
41 33  
... ... @@ -75,17 +67,11 @@ public class IotMqttUpstreamHandler extends IotMqttAbstractHandler {
75 67 .collect(Collectors.joining("."));
76 68 IotDeviceMessage message;
77 69 if (IotDeviceMessageMethodEnum.isPassthrough(topicMethod)) {
78   - // 影子透传：按设备 serializeType 解码，原始 body 作为 params
79   - IotDeviceRespDTO device = deviceService.getDeviceFromCache(productKey, deviceName);
80   - IotSerializeTypeEnum serializeType = device != null
81   - ? IotSerializeTypeEnum.of(device.getSerializeType()) : null;
82   - Map<String, Object> params = IotShadowPassthroughCodec.decode(payload, serializeType);
83   - if (params == null) {
  70 + message = deviceMessageService.deserializePassthroughMessage(payload, topicMethod, productKey, deviceName);
  71 + if (message == null) {
84 72 log.warn("[handleBusinessRequest][影子透传 payload 解析失败，客户端 ID: {}，主题: {}]", clientId, topic);
85 73 return;
86 74 }
87   - String requestId = (String) params.get("id");
88   - message = IotDeviceMessage.requestOf(requestId, topicMethod, params);
89 75 } else {
90 76 message = deviceMessageService.deserializeDeviceMessage(payload, productKey, deviceName);
91 77 if (message == null) {
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/serialize/IotMessageSerializer.java
  View file @666afcc
... ... @@ -3,6 +3,8 @@ package cn.iocoder.yudao.module.iot.gateway.serialize;
3 3 import cn.iocoder.yudao.module.iot.core.enums.IotSerializeTypeEnum;
4 4 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
5 5  
  6 +import java.util.Map;
  7 +
6 8 /**
7 9 * IoT 设备消息序列化器接口
8 10 *
... ... @@ -35,4 +37,24 @@ public interface IotMessageSerializer {
35 37 */
36 38 IotSerializeTypeEnum getType();
37 39  
  40 + /**
  41 + * 透传原始 body 编码（影子等不走 {@link IotDeviceMessage} 封套的场景）
  42 + *
  43 + * @param value 待编码对象（通常为 Map 或 DTO）
  44 + * @return 线格式字节
  45 + */
  46 + default byte[] encodeRaw(Object value) {
  47 + throw new UnsupportedOperationException("该序列化器不支持影子透传编码");
  48 + }
  49 +
  50 + /**
  51 + * 透传原始 body 解码
  52 + *
  53 + * @param bytes 线格式字节
  54 + * @return 透传 params
  55 + */
  56 + default Map<String, Object> decodeRaw(byte[] bytes) {
  57 + throw new UnsupportedOperationException("该序列化器不支持影子透传解码");
  58 + }
  59 +
38 60 }
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/serialize/IotShadowPassthroughCodec.java deleted
View file @233a2f7
1   -package cn.iocoder.yudao.module.iot.gateway.serialize;
2   -
3   -import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
4   -import cn.iocoder.yudao.module.iot.core.enums.IotSerializeTypeEnum;
5   -import lombok.extern.slf4j.Slf4j;
6   -
7   -import java.util.Map;
8   -
9   -/**
10   - * IoT 设备影子透传编解码
11   - * <p>
12   - * 影子协议本质为 JSON 结构，透传编解码规则：
13   - * <ul>
14   - * <li>{@code serializeType == null} 或 {@link IotSerializeTypeEnum#JSON}：按 JSON 编解码（未配置时默认 JSON，兼容历史设备）</li>
15   - * <li>其它显式类型（如 {@link IotSerializeTypeEnum#BINARY}）：warn 并丢弃，返回 {@code null}，不静默喂给 JSON 解析器</li>
16   - * </ul>
17   - * 下行发布方收到 {@code null} 时应跳过发送，避免 NPE；影子功能目前仅支持 JSON 协议设备。
18   - *
19   - * @author 芋道源码
20   - */
21   -@Slf4j
22   -public final class IotShadowPassthroughCodec {
23   -
24   - private static final String UNSUPPORTED_MSG = "影子透传暂仅支持 JSON serializeType";
25   -
26   - private IotShadowPassthroughCodec() {
27   - }
28   -
29   - /**
30   - * 编码影子透传 params
31   - *
32   - * @param params 透传参数（Map 或 DTO）
33   - * @param serializeType 设备序列化类型
34   - * @return 线格式字节；非 JSON 类型返回 null
35   - */
36   - public static byte[] encode(Object params, IotSerializeTypeEnum serializeType) {
37   - if (isJsonSerializeType(serializeType)) {
38   - return JsonUtils.toJsonByte(params);
39   - }
40   - log.warn("[encode][{}，serializeType={}，已丢弃]", UNSUPPORTED_MSG, serializeType);
41   - return null;
42   - }
43   -
44   - /**
45   - * 解码影子透传 payload
46   - *
47   - * @param bytes 线格式字节
48   - * @param serializeType 设备序列化类型
49   - * @return 透传 params；非 JSON 类型或解析失败返回 null
50   - */
51   - @SuppressWarnings("unchecked")
52   - public static Map<String, Object> decode(byte[] bytes, IotSerializeTypeEnum serializeType) {
53   - if (!isJsonSerializeType(serializeType)) {
54   - log.warn("[decode][{}，serializeType={}，已丢弃]", UNSUPPORTED_MSG, serializeType);
55   - return null;
56   - }
57   - return JsonUtils.parseObject(bytes, Map.class);
58   - }
59   -
60   - private static boolean isJsonSerializeType(IotSerializeTypeEnum serializeType) {
61   - // null 视为未配置，默认走 JSON（与产品/设备默认 serializeType 一致）
62   - return serializeType == null || IotSerializeTypeEnum.JSON.equals(serializeType);
63   - }
64   -
65   -}
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/serialize/json/IotJsonSerializer.java
  View file @666afcc
... ... @@ -6,6 +6,8 @@ import cn.iocoder.yudao.module.iot.core.enums.IotSerializeTypeEnum;
6 6 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
7 7 import cn.iocoder.yudao.module.iot.gateway.serialize.IotMessageSerializer;
8 8  
  9 +import java.util.Map;
  10 +
9 11 /**
10 12 * JSON 格式的消息序列化器
11 13 *
... ... @@ -34,4 +36,17 @@ public class IotJsonSerializer implements IotMessageSerializer {
34 36 return message;
35 37 }
36 38  
  39 + @Override
  40 + public byte[] encodeRaw(Object value) {
  41 + Assert.notNull(value, "待编码数据不能为空");
  42 + return JsonUtils.toJsonByte(value);
  43 + }
  44 +
  45 + @Override
  46 + @SuppressWarnings("unchecked")
  47 + public Map<String, Object> decodeRaw(byte[] bytes) {
  48 + Assert.notNull(bytes, "待解码数据不能为空");
  49 + return JsonUtils.parseObject(bytes, Map.class);
  50 + }
  51 +
37 52 }
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceMessageService.java
  View file @666afcc
... ... @@ -62,4 +62,16 @@ public interface IotDeviceMessageService {
62 62 void sendDeviceMessage(IotDeviceMessage message,
63 63 String productKey, String deviceName, String serverId);
64 64  
  65 + /**
  66 + * 透传上行组装：按设备 serializeType 解码 body，提取 id 作 requestId，构建消息
  67 + *
  68 + * @param payload 原始 body
  69 + * @param method 透传 method（来自 topic 等，如 thing.shadow.update）
  70 + * @param productKey 产品 Key
  71 + * @param deviceName 设备名称
  72 + * @return 设备消息；解码失败返回 null
  73 + */
  74 + IotDeviceMessage deserializePassthroughMessage(byte[] payload, String method,
  75 + String productKey, String deviceName);
  76 +
65 77 }
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceMessageServiceImpl.java
  View file @666afcc
1 1 package cn.iocoder.yudao.module.iot.gateway.service.device.message;
2 2  
  3 +import cn.hutool.core.convert.Convert;
3 4 import cn.hutool.core.lang.Assert;
4 5 import cn.hutool.core.util.StrUtil;
5 6 import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO;
... ... @@ -10,13 +11,14 @@ import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer;
10 11 import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
11 12 import cn.iocoder.yudao.module.iot.gateway.serialize.IotMessageSerializer;
12 13 import cn.iocoder.yudao.module.iot.gateway.serialize.IotMessageSerializerManager;
13   -import cn.iocoder.yudao.module.iot.gateway.serialize.IotShadowPassthroughCodec;
14 14 import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService;
15 15 import jakarta.annotation.Resource;
16 16 import lombok.extern.slf4j.Slf4j;
17 17 import org.springframework.stereotype.Service;
18 18  
19 19 import java.time.LocalDateTime;
  20 +import java.util.Map;
  21 +import java.util.function.Function;
20 22  
21 23 import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
22 24 import static cn.iocoder.yudao.module.iot.gateway.enums.ErrorCodeConstants.DEVICE_NOT_EXISTS;
... ... @@ -108,6 +110,23 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
108 110 }
109 111  
110 112 @Override
  113 + public IotDeviceMessage deserializePassthroughMessage(byte[] payload, String method,
  114 + String productKey, String deviceName) {
  115 + IotDeviceRespDTO device = deviceService.getDeviceFromCache(productKey, deviceName);
  116 + if (device == null) {
  117 + throw exception(DEVICE_NOT_EXISTS, productKey, deviceName);
  118 + }
  119 + IotSerializeTypeEnum serializeType = IotSerializeTypeEnum.of(device.getSerializeType());
  120 + Map<String, Object> params = decodePassthrough(payload, serializeType, productKey, deviceName);
  121 + if (params == null) {
  122 + // 非法/非对象 body：仍投递，由 biz 层回 JSON_FORMAT_ERROR(400)
  123 + return IotDeviceMessage.requestOf(null, method, null);
  124 + }
  125 + String requestId = Convert.toStr(params.get("id"), null);
  126 + return IotDeviceMessage.requestOf(requestId, method, params);
  127 + }
  128 +
  129 + @Override
111 130 public void sendDeviceMessage(IotDeviceMessage message,
112 131 String productKey, String deviceName, String serverId) {
113 132 // 1. 获取设备信息
... ... @@ -139,19 +158,59 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
139 158 }
140 159  
141 160 /**
142   - * 影子透传编码；非 JSON 设备返回 null，由下行发布方跳过发送
  161 + * 影子透传编码；不支持的 serializeType 返回 null，由下行发布方跳过发送
143 162 */
144 163 private byte[] encodePassthrough(IotDeviceMessage message, IotSerializeTypeEnum serializeType,
145 164 String productKey, String deviceName) {
146   - byte[] bytes = IotShadowPassthroughCodec.encode(message.getParams(), serializeType);
147   - if (bytes != null) {
148   - return bytes;
  165 + return passthroughRaw("encodePassthrough", serializeType, productKey, deviceName, false,
  166 + serializer -> serializer.encodeRaw(message.getParams()));
  167 + }
  168 +
  169 + /**
  170 + * 影子透传解码；解析失败或不支持的 serializeType 返回 null
  171 + */
  172 + private Map<String, Object> decodePassthrough(byte[] bytes, IotSerializeTypeEnum serializeType,
  173 + String productKey, String deviceName) {
  174 + return passthroughRaw("decodePassthrough", serializeType, productKey, deviceName, true,
  175 + serializer -> serializer.decodeRaw(bytes));
  176 + }
  177 +
  178 + private <T> T passthroughRaw(String op, IotSerializeTypeEnum serializeType,
  179 + String productKey, String deviceName, boolean decode,
  180 + Function<IotMessageSerializer, T> action) {
  181 + IotMessageSerializer serializer = getPassthroughSerializer(serializeType);
  182 + if (serializer == null) {
  183 + warnPassthroughUnsupported(op, serializeType, productKey, deviceName);
  184 + return null;
149 185 }
  186 + try {
  187 + return action.apply(serializer);
  188 + } catch (UnsupportedOperationException ex) {
  189 + warnPassthroughUnsupported(op, serializeType, productKey, deviceName);
  190 + return null;
  191 + } catch (RuntimeException ex) {
  192 + if (decode) {
  193 + log.warn("[{}][影子透传 payload 解析失败，productKey={}, deviceName={}]",
  194 + op, productKey, deviceName, ex);
  195 + return null;
  196 + }
  197 + throw ex;
  198 + }
  199 + }
  200 +
  201 + private IotMessageSerializer getPassthroughSerializer(IotSerializeTypeEnum serializeType) {
  202 + IotSerializeTypeEnum type = serializeType != null ? serializeType : IotSerializeTypeEnum.JSON;
  203 + return messageSerializerManager.get(type);
  204 + }
  205 +
  206 + private void warnPassthroughUnsupported(String op, IotSerializeTypeEnum serializeType,
  207 + String productKey, String deviceName) {
150 208 if (productKey != null && deviceName != null) {
151   - log.warn("[encodePassthrough][影子下行跳过：设备({}/{}) serializeType={} 不支持，目前仅 JSON 协议设备可用]",
152   - productKey, deviceName, serializeType);
  209 + log.warn("[{}][影子透传跳过：设备({}/{}) serializeType={} 不支持，目前仅 JSON 协议设备可用]",
  210 + op, productKey, deviceName, serializeType);
  211 + } else {
  212 + log.warn("[{}][影子透传跳过：serializeType={} 不支持，目前仅 JSON 协议设备可用]", op, serializeType);
153 213 }
154   - return null;
155 214 }
156 215  
157 216 }
... ...
yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceShadowPassthroughTest.java
  View file @666afcc
... ... @@ -2,20 +2,26 @@ package cn.iocoder.yudao.module.iot.gateway.service.device.message;
2 2  
3 3 import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
4 4 import cn.iocoder.yudao.framework.test.core.ut.BaseMockitoUnitTest;
  5 +import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO;
5 6 import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
6 7 import cn.iocoder.yudao.module.iot.core.enums.IotSerializeTypeEnum;
7 8 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
8 9 import cn.iocoder.yudao.module.iot.core.topic.shadow.IotDeviceShadowReplyDTO;
9   -import cn.iocoder.yudao.module.iot.gateway.serialize.IotShadowPassthroughCodec;
  10 +import cn.iocoder.yudao.module.iot.gateway.serialize.IotMessageSerializerManager;
  11 +import cn.iocoder.yudao.module.iot.gateway.serialize.json.IotJsonSerializer;
  12 +import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService;
  13 +import org.junit.jupiter.api.BeforeEach;
10 14 import org.junit.jupiter.api.Test;
11 15 import org.mockito.InjectMocks;
  16 +import org.mockito.Mock;
  17 +import org.mockito.Spy;
12 18  
13 19 import java.util.HashMap;
14 20 import java.util.Map;
15 21  
16   -import static org.junit.jupiter.api.Assertions.assertEquals;
17   -import static org.junit.jupiter.api.Assertions.assertNotNull;
18   -import static org.junit.jupiter.api.Assertions.assertNull;
  22 +import static org.junit.jupiter.api.Assertions.*;
  23 +import static org.mockito.ArgumentMatchers.eq;
  24 +import static org.mockito.Mockito.when;
19 25  
20 26 /**
21 27 * 设备影子网关透传序列化单元测试
... ... @@ -25,35 +31,51 @@ public class IotDeviceShadowPassthroughTest extends BaseMockitoUnitTest {
25 31 @InjectMocks
26 32 private IotDeviceMessageServiceImpl deviceMessageService;
27 33  
  34 + @Mock
  35 + private IotDeviceService deviceService;
  36 +
  37 + @Spy
  38 + private IotMessageSerializerManager messageSerializerManager = new IotMessageSerializerManager();
  39 +
  40 + @BeforeEach
  41 + public void setUpManager() {
  42 + // Spy 需手动触发构造逻辑
  43 + if (messageSerializerManager == null) {
  44 + messageSerializerManager = new IotMessageSerializerManager();
  45 + }
  46 + }
  47 +
28 48 @Test
29   - public void testSerializeDeviceMessage_passthrough_noEnvelope() {
30   - Map<String, Object> payload = new HashMap<>();
31   - payload.put("method", "get_reply");
32   - payload.put("version", 2L);
33   - payload.put("timestamp", 1469564576L);
  49 + public void testSerializeDeviceMessage_passthrough_flatReply() {
  50 + IotDeviceShadowReplyDTO reply = IotDeviceShadowReplyDTO.getReplyOf(new HashMap<>(), new HashMap<>(), 2L, 1469564576L);
34 51 IotDeviceMessage message = IotDeviceMessage.requestOf(
35   - IotDeviceMessageMethodEnum.SHADOW_GET.getMethod(), payload);
  52 + IotDeviceMessageMethodEnum.SHADOW_GET.getMethod(), reply);
36 53  
37 54 byte[] bytes = deviceMessageService.serializeDeviceMessage(message, IotSerializeTypeEnum.JSON);
38 55  
39 56 @SuppressWarnings("unchecked")
40 57 Map<String, Object> result = JsonUtils.parseObject(bytes, Map.class);
41 58 assertEquals("get_reply", result.get("method"));
42   - assertEquals(2, ((Number) result.get("version")).intValue());
  59 + assertEquals(0, ((Number) result.get("code")).intValue());
  60 + assertNull(result.get("payload"));
43 61 assertNull(result.get("id"));
44   - assertNull(result.get("code"));
45   - assertNull(result.get("requestId"));
  62 + @SuppressWarnings("unchecked")
  63 + Map<String, Object> data = (Map<String, Object>) result.get("data");
  64 + assertEquals(2, ((Number) data.get("version")).intValue());
46 65 }
47 66  
48 67 @Test
49   - public void testDeserializeUpstreamShadow_fromTopicMethod() {
50   - // 模拟上行透传：原始 body 不含 thing.shadow.update 封套 method
51   - Map<String, Object> body = new HashMap<>();
52   - body.put("method", "get");
53   - IotDeviceMessage message = IotDeviceMessage.requestOf("req-1",
54   - IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), body);
  68 + public void testDeserializePassthroughMessage_fromTopicMethod() {
  69 + IotDeviceRespDTO device = new IotDeviceRespDTO().setId(1L).setSerializeType(IotSerializeTypeEnum.JSON.getType());
  70 + when(deviceService.getDeviceFromCache(eq("pk"), eq("dn"))).thenReturn(device);
  71 +
  72 + byte[] payload = JsonUtils.toJsonByte(Map.of("method", "get", "id", "req-1"));
  73 + IotDeviceMessage message = deviceMessageService.deserializePassthroughMessage(payload,
  74 + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), "pk", "dn");
55 75  
  76 + assertNotNull(message);
56 77 assertEquals(IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), message.getMethod());
  78 + assertEquals("req-1", message.getRequestId());
57 79 assertEquals("get", ((Map<?, ?>) message.getParams()).get("method"));
58 80 }
59 81  
... ... @@ -67,23 +89,16 @@ public class IotDeviceShadowPassthroughTest extends BaseMockitoUnitTest {
67 89 @SuppressWarnings("unchecked")
68 90 Map<String, Object> result = JsonUtils.parseObject(bytes, Map.class);
69 91 assertEquals("update_reply", result.get("method"));
  92 + assertEquals(0, ((Number) result.get("code")).intValue());
70 93 assertNull(result.get("id"));
71 94 }
72 95  
73 96 @Test
74   - public void testPassthroughCodec_jsonSerializeType() {
75   - Map<String, Object> params = Map.of("method", "get");
76   - byte[] bytes = IotShadowPassthroughCodec.encode(params, IotSerializeTypeEnum.JSON);
77   - Map<String, Object> decoded = IotShadowPassthroughCodec.decode(bytes, IotSerializeTypeEnum.JSON);
78   - assertNotNull(decoded);
79   - assertEquals("get", decoded.get("method"));
80   - }
81   -
82   - @Test
83   - public void testPassthroughCodec_nullSerializeTypeAsJson() {
  97 + public void testJsonSerializer_encodeRawDecodeRaw() {
  98 + IotJsonSerializer serializer = new IotJsonSerializer();
84 99 Map<String, Object> params = Map.of("method", "get");
85   - byte[] bytes = IotShadowPassthroughCodec.encode(params, null);
86   - Map<String, Object> decoded = IotShadowPassthroughCodec.decode(bytes, null);
  100 + byte[] bytes = serializer.encodeRaw(params);
  101 + Map<String, Object> decoded = serializer.decodeRaw(bytes);
87 102 assertNotNull(decoded);
88 103 assertEquals("get", decoded.get("method"));
89 104 }
... ... @@ -96,10 +111,42 @@ public class IotDeviceShadowPassthroughTest extends BaseMockitoUnitTest {
96 111 }
97 112  
98 113 @Test
99   - public void testPassthroughCodec_binaryDrop() {
100   - Map<String, Object> params = Map.of("method", "update");
101   - assertNull(IotShadowPassthroughCodec.encode(params, IotSerializeTypeEnum.BINARY));
102   - assertNull(IotShadowPassthroughCodec.decode(new byte[]{0x01}, IotSerializeTypeEnum.BINARY));
  114 + public void testBinarySerializer_rawNotSupported() {
  115 + cn.iocoder.yudao.module.iot.gateway.serialize.binary.IotBinarySerializer serializer =
  116 + new cn.iocoder.yudao.module.iot.gateway.serialize.binary.IotBinarySerializer();
  117 + assertThrows(UnsupportedOperationException.class, () -> serializer.encodeRaw(Map.of("method", "update")));
  118 + }
  119 +
  120 + @Test
  121 + public void testDeserializePassthroughMessage_numericId() {
  122 + IotDeviceRespDTO device = new IotDeviceRespDTO().setId(1L).setSerializeType(IotSerializeTypeEnum.JSON.getType());
  123 + when(deviceService.getDeviceFromCache(eq("pk"), eq("dn"))).thenReturn(device);
  124 +
  125 + byte[] payload = JsonUtils.toJsonByte(Map.of("method", "get", "id", 123));
  126 + IotDeviceMessage message = deviceMessageService.deserializePassthroughMessage(payload,
  127 + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), "pk", "dn");
  128 +
  129 + assertNotNull(message);
  130 + assertEquals("123", message.getRequestId());
  131 + }
  132 +
  133 + @Test
  134 + public void testDeserializePassthroughMessage_invalidJsonStillDelivered() {
  135 + IotDeviceRespDTO device = new IotDeviceRespDTO().setId(1L).setSerializeType(IotSerializeTypeEnum.JSON.getType());
  136 + when(deviceService.getDeviceFromCache(eq("pk"), eq("dn"))).thenReturn(device);
  137 +
  138 + IotDeviceMessage message = deviceMessageService.deserializePassthroughMessage("{invalid".getBytes(),
  139 + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), "pk", "dn");
  140 +
  141 + assertNotNull(message);
  142 + assertNull(message.getParams());
  143 + }
  144 +
  145 + @Test
  146 + public void testErrorOf_getMethodUsesGetReply() {
  147 + IotDeviceShadowReplyDTO reply = IotDeviceShadowReplyDTO.errorOf("get", 1L, 400, "JSON 格式错误", 1000L);
  148 + assertEquals("get_reply", reply.getMethod());
  149 + assertEquals(400, reply.getCode());
103 150 }
104 151  
105 152 }
... ...