BJ_ZTEITS-Developers / parking-cloud-yudao

Commit 8865481b2e9d2357998b09196510fc174d662299

Authored by haohao
1 parent 6e9b58cb

refactor(iot): 设备影子改回标准 IotDeviceMessage 封套，移除透传链路 

- 上行/下行统一走 Serializer 标准封套，删除 passthrough 与 encodeRaw/decodeRaw
- 删除 IotDeviceShadowReplyDTO，回复改用 replyOf + ProtocolUtils.buildGetData/buildUpdateData
- 下行回复走 thing/shadow/update_reply，移除 SHADOW_GET
- 同步调整 biz/gateway 测试
Inline Side-by-side
Showing 18 changed files with 156 additions and 544 deletions
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImpl.java
  View file @8865481
... ... @@ -8,7 +8,6 @@ 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 10 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
11   -import cn.iocoder.yudao.module.iot.core.topic.shadow.IotDeviceShadowReplyDTO;
12 11 import cn.iocoder.yudao.module.iot.core.topic.shadow.IotDeviceShadowUpdateReqDTO;
13 12 import cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils;
14 13 import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
... ... @@ -31,8 +30,10 @@ import java.util.HashMap;
31 30 import java.util.Map;
32 31 import java.util.function.Function;
33 32  
  33 +import static cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants.SUCCESS;
34 34 import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
35   -import static cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils.*;
  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;
36 37 import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.DEVICE_SHADOW_NOT_ENABLED;
37 38 import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.DEVICE_SHADOW_VERSION_CONFLICT;
38 39  
... ... @@ -88,29 +89,23 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService {
88 89 } catch (Exception ex) {
89 90 log.error("[handleDeviceShadowUpdate][device({}) message({}) 处理异常]", device.getId(), message, ex);
90 91 Long version = resolveShadowVersionSafely(device.getId());
91   - sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf(
92   - extractInnerMethod(message.getParams()), version,
93   - IotDeviceShadowErrorCodeEnum.SERVER_ERROR.getCode(),
94   - IotDeviceShadowErrorCodeEnum.SERVER_ERROR.getMessage(), DateUtil.currentSeconds()));
  92 + sendShadowErrorReply(device, message.getServerId(), message.getRequestId(),
  93 + IotDeviceShadowErrorCodeEnum.SERVER_ERROR, version);
95 94 }
96 95 }
97 96  
98 97 private void handleDeviceShadowUpdate0(IotDeviceDO device, IotDeviceMessage message) {
99   - // 1. 解析请求（影子 body 当前按 JSON 建模，非 JSON 设备在 gateway 透传层已拒绝）
  98 + // 1. 解析请求（影子 params 按 JSON 建模，设备须配置 serializeType=JSON）
100 99 IotDeviceShadowUpdateReqDTO req = JsonUtils.convertObject(message.getParams(),
101 100 IotDeviceShadowUpdateReqDTO.class);
102 101 if (req == null) {
103   - sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf(
104   - extractInnerMethod(message.getParams()), 0L,
105   - IotDeviceShadowErrorCodeEnum.JSON_FORMAT_ERROR.getCode(),
106   - IotDeviceShadowErrorCodeEnum.JSON_FORMAT_ERROR.getMessage(), DateUtil.currentSeconds()));
  102 + sendShadowErrorReply(device, message.getServerId(), message.getRequestId(),
  103 + IotDeviceShadowErrorCodeEnum.JSON_FORMAT_ERROR, 0L);
107 104 return;
108 105 }
109 106 if (StrUtil.isEmpty(req.getMethod())) {
110   - sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf(
111   - INNER_METHOD_UPDATE, resolveShadowVersionSafely(device.getId()),
112   - IotDeviceShadowErrorCodeEnum.METHOD_MISSING.getCode(),
113   - IotDeviceShadowErrorCodeEnum.METHOD_MISSING.getMessage(), DateUtil.currentSeconds()));
  107 + sendShadowErrorReply(device, message.getServerId(), message.getRequestId(),
  108 + IotDeviceShadowErrorCodeEnum.METHOD_MISSING, resolveShadowVersionSafely(device.getId()));
114 109 return;
115 110 }
116 111  
... ... @@ -123,42 +118,35 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService {
123 118  
124 119 // 3. 内层 method 分发
125 120 if (INNER_METHOD_GET.equals(req.getMethod())) {
126   - sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.getReplyOf(
127   - shadow.getDesired(), shadow.getReported(), shadow.getVersion(), shadow.getShadowTime()));
  121 + sendShadowGetReply(device, message.getServerId(), message.getRequestId(),
  122 + shadow.getDesired(), shadow.getReported(), shadow.getVersion(), shadow.getShadowTime());
128 123 return;
129 124 }
130 125 if (INNER_METHOD_UPDATE.equals(req.getMethod())) {
131   - handleReportedUpdate(device, message.getServerId(), req);
  126 + handleReportedUpdate(device, message.getServerId(), message.getRequestId(), req);
132 127 return;
133 128 }
134   - sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf(
135   - req.getMethod(), shadow.getVersion(),
136   - IotDeviceShadowErrorCodeEnum.INVALID_METHOD.getCode(),
137   - IotDeviceShadowErrorCodeEnum.INVALID_METHOD.getMessage(), DateUtil.currentSeconds()));
  129 + sendShadowErrorReply(device, message.getServerId(), message.getRequestId(),
  130 + IotDeviceShadowErrorCodeEnum.INVALID_METHOD, shadow.getVersion());
138 131 }
139 132  
140   - private void handleReportedUpdate(IotDeviceDO device, String serverId, IotDeviceShadowUpdateReqDTO req) {
  133 + private void handleReportedUpdate(IotDeviceDO device, String serverId, String requestId,
  134 + IotDeviceShadowUpdateReqDTO req) {
141 135 Long deviceId = device.getId();
142 136  
143 137 if (req.getState() == null) {
144   - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(
145   - INNER_METHOD_UPDATE, resolveShadowVersionSafely(deviceId),
146   - IotDeviceShadowErrorCodeEnum.STATE_MISSING.getCode(),
147   - IotDeviceShadowErrorCodeEnum.STATE_MISSING.getMessage(), DateUtil.currentSeconds()));
  138 + sendShadowErrorReply(device, serverId, requestId,
  139 + IotDeviceShadowErrorCodeEnum.STATE_MISSING, resolveShadowVersionSafely(deviceId));
148 140 return;
149 141 }
150 142 if (req.getState().getReported() == null) {
151   - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(
152   - INNER_METHOD_UPDATE, resolveShadowVersionSafely(deviceId),
153   - IotDeviceShadowErrorCodeEnum.REPORTED_MISSING.getCode(),
154   - IotDeviceShadowErrorCodeEnum.REPORTED_MISSING.getMessage(), DateUtil.currentSeconds()));
  143 + sendShadowErrorReply(device, serverId, requestId,
  144 + IotDeviceShadowErrorCodeEnum.REPORTED_MISSING, resolveShadowVersionSafely(deviceId));
155 145 return;
156 146 }
157 147 if (CollUtil.isEmpty(req.getState().getReported())) {
158   - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(
159   - INNER_METHOD_UPDATE, resolveShadowVersionSafely(deviceId),
160   - IotDeviceShadowErrorCodeEnum.REPORTED_EMPTY.getCode(),
161   - IotDeviceShadowErrorCodeEnum.REPORTED_EMPTY.getMessage(), DateUtil.currentSeconds()));
  148 + sendShadowErrorReply(device, serverId, requestId,
  149 + IotDeviceShadowErrorCodeEnum.REPORTED_EMPTY, resolveShadowVersionSafely(deviceId));
162 150 return;
163 151 }
164 152  
... ... @@ -166,24 +154,20 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService {
166 154 if (req.getVersion() != null && req.getVersion() == -1L) {
167 155 try {
168 156 resetShadowWithRetry(deviceId);
169   - sendShadowReplySafely(device, serverId,
170   - IotDeviceShadowReplyDTO.updateReplyOf(0L, DateUtil.currentSeconds()));
  157 + sendShadowUpdateReply(device, serverId, requestId, 0L, DateUtil.currentSeconds());
171 158 } catch (ShadowCasExhaustedException ex) {
172   - handleCasExhaustedForDevice(device, serverId, ex);
  159 + handleCasExhaustedForDevice(device, serverId, requestId, ex);
173 160 }
174 161 return;
175 162 }
176 163  
177 164 try {
178 165 IotDeviceShadowDO committed = casUpdateShadow(deviceId, latest -> buildReportedUpdate(latest, req));
179   - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.updateReplyOf(committed.getVersion(),
180   - committed.getShadowTime()));
  166 + sendShadowUpdateReply(device, serverId, requestId, committed.getVersion(), committed.getShadowTime());
181 167 } catch (ShadowUpdateRejectedException ex) {
182   - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(
183   - INNER_METHOD_UPDATE, ex.getVersion(),
184   - ex.getErrorCode().getCode(), ex.getErrorCode().getMessage(), DateUtil.currentSeconds()));
  168 + sendShadowErrorReply(device, serverId, requestId, ex.getErrorCode(), ex.getVersion());
185 169 } catch (ShadowCasExhaustedException ex) {
186   - handleCasExhaustedForDevice(device, serverId, ex);
  170 + handleCasExhaustedForDevice(device, serverId, requestId, ex);
187 171 }
188 172 }
189 173  
... ... @@ -203,9 +187,9 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService {
203 187 // 设备在线则主动推送全量影子（复用 CAS 返回值，避免二次查库）
204 188 String serverId = devicePropertyService.getDeviceServerId(deviceId);
205 189 if (StrUtil.isNotEmpty(serverId)) {
206   - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.getReplyOf(
  190 + sendShadowGetReply(device, serverId, null,
207 191 committed.getDesired(), committed.getReported(),
208   - committed.getVersion(), committed.getShadowTime()));
  192 + committed.getVersion(), committed.getShadowTime());
209 193 }
210 194 } catch (ShadowCasExhaustedException ex) {
211 195 throw exception(DEVICE_SHADOW_VERSION_CONFLICT);
... ... @@ -337,7 +321,7 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService {
337 321 }
338 322  
339 323 /**
340   - * 将 CAS 成功写入的局部字段与 latest 快照合并，供全量 get_reply 使用
  324 + * 将 CAS 成功写入的局部字段与 latest 快照合并，供全量 get 回复使用
341 325 */
342 326 private static void enrichCommittedShadow(IotDeviceShadowDO committed, IotDeviceShadowDO latest) {
343 327 if (committed.getDesired() == null) {
... ... @@ -351,22 +335,19 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService {
351 335 }
352 336 }
353 337  
354   - private void handleCasExhaustedForDevice(IotDeviceDO device, String serverId, ShadowCasExhaustedException ex) {
  338 + private void handleCasExhaustedForDevice(IotDeviceDO device, String serverId, String requestId,
  339 + ShadowCasExhaustedException ex) {
355 340 if (ex.isShadowNotFound()) {
356   - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(
357   - INNER_METHOD_UPDATE, 0L,
358   - IotDeviceShadowErrorCodeEnum.SERVER_ERROR.getCode(),
359   - IotDeviceShadowErrorCodeEnum.SERVER_ERROR.getMessage(), DateUtil.currentSeconds()));
  341 + sendShadowErrorReply(device, serverId, requestId,
  342 + IotDeviceShadowErrorCodeEnum.SERVER_ERROR, 0L);
360 343 return;
361 344 }
362   - sendVersionConflictReply(device, serverId, ex.getLastKnownVersion());
  345 + sendVersionConflictReply(device, serverId, requestId, ex.getLastKnownVersion());
363 346 }
364 347  
365   - private void sendVersionConflictReply(IotDeviceDO device, String serverId, Long version) {
366   - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(
367   - INNER_METHOD_UPDATE, version,
368   - IotDeviceShadowErrorCodeEnum.VERSION_CONFLICT.getCode(),
369   - IotDeviceShadowErrorCodeEnum.VERSION_CONFLICT.getMessage(), DateUtil.currentSeconds()));
  348 + private void sendVersionConflictReply(IotDeviceDO device, String serverId, String requestId, Long version) {
  349 + sendShadowErrorReply(device, serverId, requestId,
  350 + IotDeviceShadowErrorCodeEnum.VERSION_CONFLICT, version);
370 351 }
371 352  
372 353 private Long resolveShadowVersionSafely(Long deviceId) {
... ... @@ -407,15 +388,41 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService {
407 388 return shadow;
408 389 }
409 390  
  391 + private void sendShadowGetReply(IotDeviceDO device, String serverId, String requestId,
  392 + Map<String, Object> desired, Map<String, Object> reported, Long version, Long timestamp) {
  393 + sendShadowReplySafely(device, serverId, requestId, SUCCESS.getCode(), SUCCESS.getMsg(),
  394 + IotDeviceShadowProtocolUtils.buildGetData(desired, reported, version, timestamp));
  395 + }
  396 +
  397 + private void sendShadowUpdateReply(IotDeviceDO device, String serverId, String requestId,
  398 + Long version, Long timestamp) {
  399 + sendShadowReplySafely(device, serverId, requestId, SUCCESS.getCode(), SUCCESS.getMsg(),
  400 + IotDeviceShadowProtocolUtils.buildUpdateData(version, timestamp));
  401 + }
  402 +
  403 + /**
  404 + * 错误回复：data 仍带 version 供设备 resync（有意偏离纯 CommonResult 的 error-data-null 约定）
  405 + */
  406 + private void sendShadowErrorReply(IotDeviceDO device, String serverId, String requestId,
  407 + IotDeviceShadowErrorCodeEnum errorCode, Long version) {
  408 + sendShadowReplySafely(device, serverId, requestId, errorCode.getCode(), errorCode.getMessage(),
  409 + IotDeviceShadowProtocolUtils.buildUpdateData(version, DateUtil.currentSeconds()));
  410 + }
  411 +
410 412 /**
411   - * 下发影子回复，内部吞掉异常，保证上行处理链路不向上抛
  413 + * 下发影子回复（标准 IotDeviceMessage 封套），内部吞掉异常，保证上行处理链路不向上抛
  414 + * <p>
  415 + * 回复 method 取上行的 {@link IotDeviceMessageMethodEnum#SHADOW_UPDATE}（与框架"回复用原请求 method"一致）：
  416 + * 带 code 的回复经 IotDeviceMessageUtils#isUpstreamMessage 取反 SHADOW_UPDATE.upstream(true) → 判定为下行，
  417 + * 正确投递到设备；下行 Topic 为 thing/shadow/update_reply。
412 418 */
413   - private void sendShadowReplySafely(IotDeviceDO device, String serverId, IotDeviceShadowReplyDTO replyDTO) {
414   - IotDeviceMessage shadowGetMsg = IotDeviceMessage.requestOf(
415   - device.getId(), device.getTenantId(), serverId,
416   - IotDeviceMessageMethodEnum.SHADOW_GET.getMethod(), replyDTO);
  419 + private void sendShadowReplySafely(IotDeviceDO device, String serverId, String requestId,
  420 + Integer code, String msg, Map<String, Object> data) {
  421 + IotDeviceMessage shadowReplyMsg = IotDeviceMessage.replyOf(
  422 + requestId, IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), data, code, msg);
  423 + shadowReplyMsg.setDeviceId(device.getId()).setTenantId(device.getTenantId()).setServerId(serverId);
417 424 try {
418   - deviceMessageService.sendDeviceMessage(shadowGetMsg, device, serverId);
  425 + deviceMessageService.sendDeviceMessage(shadowReplyMsg, device, serverId);
419 426 } catch (Exception ex) {
420 427 log.error("[sendShadowReply][device({}) 下发影子回复失败]", device.getId(), ex);
421 428 }
... ...
yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImplTest.java
  View file @8865481
... ... @@ -5,7 +5,6 @@ import cn.iocoder.yudao.framework.common.exception.ServiceException;
5 5 import cn.iocoder.yudao.framework.test.core.ut.BaseMockitoUnitTest;
6 6 import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
7 7 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
8   -import cn.iocoder.yudao.module.iot.core.topic.shadow.IotDeviceShadowReplyDTO;
9 8 import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
10 9 import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceShadowDO;
11 10 import cn.iocoder.yudao.module.iot.dal.mysql.device.IotDeviceShadowMapper;
... ... @@ -76,15 +75,18 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
76 75  
77 76 Map<String, Object> params = new HashMap<>();
78 77 params.put("method", "get");
79   - IotDeviceMessage message = IotDeviceMessage.requestOf(device.getId(), 1L, "server-1",
  78 + IotDeviceMessage message = IotDeviceMessage.requestOf("req-get-1",
80 79 IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), params);
  80 + message.setDeviceId(device.getId()).setTenantId(1L).setServerId("server-1");
81 81  
82 82 service.handleDeviceShadowUpdate(device, message);
83 83  
84 84 ArgumentCaptor<IotDeviceMessage> captor = ArgumentCaptor.forClass(IotDeviceMessage.class);
85 85 verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1"));
86   - IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams();
87   - assertEquals("get_reply", reply.getMethod());
  86 + IotDeviceMessage reply = captor.getValue();
  87 + assertEquals(IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), reply.getMethod());
  88 + assertEquals("req-get-1", reply.getRequestId());
  89 + assertEquals(0, reply.getCode());
88 90 assertEquals(2L, getReplyVersion(reply));
89 91 }
90 92  
... ... @@ -141,7 +143,8 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
141 143  
142 144 ArgumentCaptor<IotDeviceMessage> captor = ArgumentCaptor.forClass(IotDeviceMessage.class);
143 145 verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1"));
144   - IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams();
  146 + IotDeviceMessage reply = captor.getValue();
  147 + assertEquals(IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), reply.getMethod());
145 148 assertEquals(409, reply.getCode());
146 149 verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any());
147 150 }
... ... @@ -192,8 +195,8 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
192 195  
193 196 ArgumentCaptor<IotDeviceMessage> captor = ArgumentCaptor.forClass(IotDeviceMessage.class);
194 197 verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1"));
195   - IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams();
196   - assertEquals("update_reply", reply.getMethod());
  198 + IotDeviceMessage reply = captor.getValue();
  199 + assertEquals(IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), reply.getMethod());
197 200 assertEquals(409, reply.getCode());
198 201 verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any());
199 202 }
... ... @@ -262,8 +265,8 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
262 265  
263 266 ArgumentCaptor<IotDeviceMessage> captor = ArgumentCaptor.forClass(IotDeviceMessage.class);
264 267 verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1"));
265   - IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams();
266   - assertEquals("get_reply", reply.getMethod());
  268 + IotDeviceMessage reply = captor.getValue();
  269 + assertEquals(IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), reply.getMethod());
267 270 assertEquals(2L, getReplyVersion(reply));
268 271 // getOrCreateShadow + CAS 各一次 SELECT，推送不再额外查库
269 272 verify(deviceShadowMapper, times(2)).selectByDeviceId(deviceId);
... ... @@ -299,7 +302,8 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
299 302  
300 303 ArgumentCaptor<IotDeviceMessage> captor = ArgumentCaptor.forClass(IotDeviceMessage.class);
301 304 verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1"));
302   - IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams();
  305 + IotDeviceMessage reply = captor.getValue();
  306 + assertEquals(IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), reply.getMethod());
303 307 assertEquals(400, reply.getCode());
304 308 }
305 309  
... ... @@ -319,7 +323,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
319 323  
320 324 ArgumentCaptor<IotDeviceMessage> captor = ArgumentCaptor.forClass(IotDeviceMessage.class);
321 325 verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1"));
322   - IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams();
  326 + IotDeviceMessage reply = captor.getValue();
323 327 assertEquals(401, reply.getCode());
324 328 }
325 329  
... ... @@ -375,7 +379,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
375 379  
376 380 ArgumentCaptor<IotDeviceMessage> replyCaptor = ArgumentCaptor.forClass(IotDeviceMessage.class);
377 381 verify(deviceMessageService).sendDeviceMessage(replyCaptor.capture(), eq(device), eq("server-1"));
378   - IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) replyCaptor.getValue().getParams();
  382 + IotDeviceMessage reply = replyCaptor.getValue();
379 383 assertEquals(3L, getReplyVersion(reply));
380 384 }
381 385  
... ... @@ -439,7 +443,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
439 443  
440 444 ArgumentCaptor<IotDeviceMessage> captor = ArgumentCaptor.forClass(IotDeviceMessage.class);
441 445 verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1"));
442   - IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams();
  446 + IotDeviceMessage reply = captor.getValue();
443 447 assertEquals(0L, getReplyVersion(reply));
444 448 assertEquals(500, reply.getCode());
445 449 }
... ... @@ -466,7 +470,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
466 470  
467 471 ArgumentCaptor<IotDeviceMessage> captor = ArgumentCaptor.forClass(IotDeviceMessage.class);
468 472 verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1"));
469   - IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams();
  473 + IotDeviceMessage reply = captor.getValue();
470 474 assertEquals(500, reply.getCode());
471 475 verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any());
472 476 }
... ... @@ -501,7 +505,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
501 505 }
502 506  
503 507 @SuppressWarnings("unchecked")
504   - private static Long getReplyVersion(IotDeviceShadowReplyDTO reply) {
  508 + private static Long getReplyVersion(IotDeviceMessage reply) {
505 509 if (!(reply.getData() instanceof Map)) {
506 510 return null;
507 511 }
... ...
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotDeviceMessageMethodEnum.java
  View file @8865481
... ... @@ -70,7 +70,6 @@ public enum IotDeviceMessageMethodEnum implements ArrayValuable&lt;String&gt; {
70 70 // ========== 设备影子 ==========
71 71  
72 72 SHADOW_UPDATE("thing.shadow.update", "设备影子更新", true),
73   - SHADOW_GET("thing.shadow.get", "设备影子获取", false),
74 73  
75 74 ;
76 75  
... ... @@ -83,15 +82,7 @@ public enum IotDeviceMessageMethodEnum implements ArrayValuable&lt;String&gt; {
83 82 public static final Set<String> REPLY_DISABLED = SetUtils.asSet(
84 83 STATE_UPDATE.getMethod(),
85 84 OTA_PROGRESS.getMethod(), // 参考阿里云，OTA 升级进度上报，不进行回复
86   - SHADOW_UPDATE.getMethod() // 设备影子由影子 Service 主动下发 thing.shadow.get
87   - );
88   -
89   - /**
90   - * 透传方法集合：设备侧使用原始线格式，不套 IotDeviceMessage 封套
91   - */
92   - public static final Set<String> PASSTHROUGH_METHODS = SetUtils.asSet(
93   - SHADOW_UPDATE.getMethod(),
94   - SHADOW_GET.getMethod()
  85 + SHADOW_UPDATE.getMethod() // 设备影子由影子 Service 主动下发回复（thing/shadow/update_reply）
95 86 );
96 87  
97 88 private final String method;
... ... @@ -114,8 +105,4 @@ public enum IotDeviceMessageMethodEnum implements ArrayValuable&lt;String&gt; {
114 105 return REPLY_DISABLED.contains(method);
115 106 }
116 107  
117   - public static boolean isPassthrough(String method) {
118   - return PASSTHROUGH_METHODS.contains(method);
119   - }
120   -
121 108 }
... ...
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/shadow/IotDeviceShadowErrorCodeEnum.java
  View file @8865481
... ... @@ -6,7 +6,7 @@ import lombok.Getter;
6 6 /**
7 7 * IoT 设备影子设备侧错误码枚举
8 8 * <p>
9   - * 设备侧 wire code，仅用于回填 {@code update_reply.code/msg}，不走
  9 + * 设备侧 wire code，仅用于回填影子下行回复（thing.shadow.update_reply）的 code/msg，不走
10 10 * {@link cn.iocoder.yudao.framework.common.pojo.CommonResult}。
11 11 * 服务端内部异常请使用 {@link cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants}（如
12 12 * {@code DEVICE_SHADOW_NOT_ENABLED}）。
... ...
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/shadow/IotDeviceShadowReplyDTO.java deleted
View file @6e9b58c
1   -package cn.iocoder.yudao.module.iot.core.topic.shadow;
2   -
3   -import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
4   -import lombok.AllArgsConstructor;
5   -import lombok.Data;
6   -import lombok.NoArgsConstructor;
7   -
8   -import java.util.HashMap;
9   -import java.util.Map;
10   -
11   -import static cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants.SUCCESS;
12   -import static cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils.*;
13   -
14   -/**
15   - * IoT 设备影子下行 Reply DTO
16   - * <p>
17   - * 扁平对齐 CommonResult + method，用于 {@link IotDeviceMessageMethodEnum#SHADOW_GET}
18   - * 透传线 body。
19   - * 错误回复的 {@link #data} 仍携带 version 供设备 resync，有意偏离纯 CommonResult（error 时 data
20   - * 通常为 null）。
21   - *
22   - * @author 芋道源码
23   - */
24   -@Data
25   -@NoArgsConstructor
26   -@AllArgsConstructor
27   -public class IotDeviceShadowReplyDTO {
28   -
29   - /**
30   - * 内层方法：get_reply / update_reply
31   - */
32   - private String method;
33   -
34   - /**
35   - * 结果码，0 表示成功
36   - */
37   - private Integer code;
38   -
39   - /**
40   - * 结果描述
41   - */
42   - private String msg;
43   -
44   - /**
45   - * 业务数据：get_reply 为 {state, version, timestamp}；update_reply 为 {version,
46   - * timestamp}
47   - */
48   - private Object data;
49   -
50   - public static IotDeviceShadowReplyDTO getReplyOf(Map<String, Object> desired, Map<String, Object> reported,
51   - Long version, Long timestamp) {
52   - Map<String, Object> state = new HashMap<>();
53   - state.put(STATE_DESIRED, desired != null ? desired : new HashMap<>());
54   - state.put(STATE_REPORTED, reported != null ? reported : new HashMap<>());
55   - Map<String, Object> data = new HashMap<>();
56   - data.put(PAYLOAD_STATE, state);
57   - data.put(PAYLOAD_VERSION, version);
58   - data.put(METADATA_TIMESTAMP, timestamp);
59   - return of(INNER_METHOD_GET_REPLY, SUCCESS.getCode(), SUCCESS.getMsg(), data);
60   - }
61   -
62   - public static IotDeviceShadowReplyDTO updateReplyOf(Long version, Long timestamp) {
63   - return of(INNER_METHOD_UPDATE_REPLY, SUCCESS.getCode(), SUCCESS.getMsg(),
64   - dataWithVersionTimestamp(version, timestamp));
65   - }
66   -
67   - public static IotDeviceShadowReplyDTO errorOf(String innerMethod, Long version, Integer errorCode,
68   - String errorMsg, Long timestamp) {
69   - return of(toReplyMethod(innerMethod), errorCode, errorMsg, dataWithVersionTimestamp(version, timestamp));
70   - }
71   -
72   - private static IotDeviceShadowReplyDTO of(String method, Integer code, String msg, Map<String, Object> data) {
73   - return new IotDeviceShadowReplyDTO().setMethod(method).setCode(code).setMsg(msg).setData(data);
74   - }
75   -
76   - private static Map<String, Object> dataWithVersionTimestamp(Long version, Long timestamp) {
77   - Map<String, Object> data = new HashMap<>();
78   - data.put(PAYLOAD_VERSION, version);
79   - data.put(METADATA_TIMESTAMP, timestamp);
80   - return data;
81   - }
82   -
83   -}
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/shadow/IotDeviceShadowUpdateReqDTO.java
  View file @8865481
... ... @@ -10,7 +10,7 @@ import java.util.Map;
10 10 /**
11 11 * IoT 设备影子上行 Request DTO
12 12 * <p>
13   - * 用于 {@link IotDeviceMessageMethodEnum#SHADOW_UPDATE} 消息的 params 参数（透传原始线 body）
  13 + * 用于 {@link IotDeviceMessageMethodEnum#SHADOW_UPDATE} 标准封套的 params 参数
14 14 *
15 15 * @author 芋道源码
16 16 */
... ...
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceShadowProtocolUtils.java
  View file @8865481
... ... @@ -9,7 +9,7 @@ import java.util.Set;
9 9 /**
10 10 * IoT 设备影子协议工具
11 11 * <p>
12   - * 内层 method、state 键名、payload 字段等与设备侧透传 JSON 对齐；并提供 metadata 读写。
  12 + * 内层 method、state 键名、reply data 字段等与设备侧标准封套 params/data 对齐；并提供 metadata 读写。
13 13 *
14 14 * @author 芋道源码
15 15 */
... ... @@ -18,12 +18,10 @@ public final class IotDeviceShadowProtocolUtils {
18 18 private IotDeviceShadowProtocolUtils() {
19 19 }
20 20  
21   - // ========== 内层 method ==========
  21 + // ========== 内层 method（上行 params.method） ==========
22 22  
23 23 public static final String INNER_METHOD_GET = "get";
24 24 public static final String INNER_METHOD_UPDATE = "update";
25   - public static final String INNER_METHOD_GET_REPLY = "get_reply";
26   - public static final String INNER_METHOD_UPDATE_REPLY = "update_reply";
27 25  
28 26 // ========== state / metadata 键 ==========
29 27  
... ... @@ -37,24 +35,28 @@ public final class IotDeviceShadowProtocolUtils {
37 35 public static final String PAYLOAD_VERSION = "version";
38 36  
39 37 /**
40   - * 由内层 method 推导 reply method（get → get_reply，其余 → update_reply）
  38 + * 构建 get 响应 data：{ state:{desired,reported}, version, timestamp }
41 39 */
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;
  40 + public static Map<String, Object> buildGetData(Map<String, Object> desired, Map<String, Object> reported,
  41 + Long version, Long timestamp) {
  42 + Map<String, Object> state = new HashMap<>();
  43 + state.put(STATE_DESIRED, desired != null ? desired : new HashMap<>());
  44 + state.put(STATE_REPORTED, reported != null ? reported : new HashMap<>());
  45 + Map<String, Object> data = new HashMap<>();
  46 + data.put(PAYLOAD_STATE, state);
  47 + data.put(PAYLOAD_VERSION, version);
  48 + data.put(METADATA_TIMESTAMP, timestamp);
  49 + return data;
47 50 }
48 51  
49 52 /**
50   - * 从透传 params 提取内层 method；无法解析时默认 update
  53 + * 构建 update 响应 / 错误 data：{ version, timestamp }
51 54 */
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;
  55 + public static Map<String, Object> buildUpdateData(Long version, Long timestamp) {
  56 + Map<String, Object> data = new HashMap<>();
  57 + data.put(PAYLOAD_VERSION, version);
  58 + data.put(METADATA_TIMESTAMP, timestamp);
  59 + return data;
58 60 }
59 61  
60 62 /**
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/downstream/IotEmqxDownstreamHandler.java
  View file @8865481
... ... @@ -55,11 +55,6 @@ public class IotEmqxDownstreamHandler {
55 55 // 2.2 构建载荷
56 56 byte[] payload = deviceMessageService.serializeDeviceMessage(message, deviceInfo.getProductKey(),
57 57 deviceInfo.getDeviceName());
58   - if (payload == null) {
59   - log.warn("[handle][影子下行跳过，设备 ID: {}，method: {}，非 JSON 协议设备不支持影子透传]",
60   - message.getDeviceId(), message.getMethod());
61   - return;
62   - }
63 58  
64 59 // 3. 发布消息
65 60 protocol.publishMessage(topic, payload);
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/upstream/IotEmqxUpstreamHandler.java
  View file @8865481
... ... @@ -3,16 +3,12 @@ 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;
7 6 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
8 7 import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
9 8 import cn.iocoder.yudao.module.iot.gateway.util.IotMqttTopicUtils;
10 9 import io.vertx.mqtt.messages.MqttPublishMessage;
11 10 import lombok.extern.slf4j.Slf4j;
12 11  
13   -import java.util.Arrays;
14   -import java.util.stream.Collectors;
15   -
16 12 /**
17 13 * IoT 网关 EMQX 上行消息处理器
18 14 *
... ... @@ -47,25 +43,14 @@ public class IotEmqxUpstreamHandler {
47 43 return;
48 44 }
49 45  
50   - // 2.1 反序列化消息
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);
  46 + // 2.1 反序列化消息（标准 IotDeviceMessage 封套，method 来自 body）
  47 + IotDeviceMessage message = deviceMessageService.deserializeDeviceMessage(payload, productKey, deviceName);
  48 + if (message == null) {
  49 + log.warn("[handle][topic({}) payload({}) 消息解码失败]", topic, new String(payload));
  50 + return;
68 51 }
  52 + // 2.2 标准化回复消息的 method（MQTT 协议中，设备回复消息的 method 会携带 _reply 后缀）
  53 + IotMqttTopicUtils.normalizeReplyMethod(message);
69 54  
70 55 // 3. 发送消息到队列
71 56 deviceMessageService.sendDeviceMessage(message, productKey, deviceName, serverId);
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/downstream/IotMqttDownstreamHandler.java
  View file @8865481
... ... @@ -45,11 +45,6 @@ public class IotMqttDownstreamHandler {
45 45 // 2.1 序列化消息
46 46 byte[] payload = deviceMessageService.serializeDeviceMessage(message, connectionInfo.getProductKey(),
47 47 connectionInfo.getDeviceName());
48   - if (payload == null) {
49   - log.warn("[handle][影子下行跳过，设备 ID: {}，method: {}，非 JSON 协议设备不支持影子透传]",
50   - message.getDeviceId(), message.getMethod());
51   - return;
52   - }
53 48 Assert.isTrue(payload.length > 0, "消息编码结果不能为空");
54 49 // 2.2 构建主题
55 50 Assert.notBlank(message.getMethod(), "消息方法不能为空");
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/upstream/IotMqttAbstractHandler.java
  View file @8865481
... ... @@ -72,11 +72,6 @@ public abstract class IotMqttAbstractHandler {
72 72 try {
73 73 // 1.1 序列化消息（根据设备配置的序列化类型）
74 74 byte[] encodedData = deviceMessageService.serializeDeviceMessage(responseMessage, productKey, deviceName);
75   - if (encodedData == null) {
76   - // 影子透传在非 JSON 设备上 encode 返回 null，跳过发布避免 NPE
77   - log.warn("[writeResponse][影子下行跳过，客户端 ID: {}，method: {}]", endpoint.clientIdentifier(), method);
78   - return;
79   - }
80 75 // 1.2 构建响应主题
81 76 String replyTopic = IotMqttTopicUtils.buildTopicByMethod(method, productKey, deviceName, true);
82 77  
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/upstream/IotMqttUpstreamHandler.java
  View file @8865481
... ... @@ -2,7 +2,6 @@ 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.enums.IotDeviceMessageMethodEnum;
6 5 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
7 6 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.manager.IotMqttConnectionManager;
8 7 import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
... ... @@ -10,10 +9,6 @@ import cn.iocoder.yudao.module.iot.gateway.util.IotMqttTopicUtils;
10 9 import io.vertx.mqtt.MqttEndpoint;
11 10 import lombok.extern.slf4j.Slf4j;
12 11  
13   -import java.util.Arrays;
14   -import java.util.stream.Collectors;
15   -
16   -
17 12 /**
18 13 * IoT 网关 MQTT 上行消息处理器：处理业务消息（属性上报、事件上报等）
19 14 *
... ... @@ -62,25 +57,14 @@ public class IotMqttUpstreamHandler extends IotMqttAbstractHandler {
62 57 return;
63 58 }
64 59  
65   - // 2.1 反序列化消息
66   - String topicMethod = Arrays.stream(topicParts, 4, topicParts.length)
67   - .collect(Collectors.joining("."));
68   - IotDeviceMessage message;
69   - if (IotDeviceMessageMethodEnum.isPassthrough(topicMethod)) {
70   - message = deviceMessageService.deserializePassthroughMessage(payload, topicMethod, productKey, deviceName);
71   - if (message == null) {
72   - log.warn("[handleBusinessRequest][影子透传 payload 解析失败，客户端 ID: {}，主题: {}]", clientId, topic);
73   - return;
74   - }
75   - } else {
76   - message = deviceMessageService.deserializeDeviceMessage(payload, productKey, deviceName);
77   - if (message == null) {
78   - log.warn("[handleBusinessRequest][消息解码失败，客户端 ID: {}，主题: {}]", clientId, topic);
79   - return;
80   - }
81   - // 2.2 标准化回复消息的 method（MQTT 协议中，设备回复消息的 method 会携带 _reply 后缀）
82   - IotMqttTopicUtils.normalizeReplyMethod(message);
  60 + // 2.1 反序列化消息（标准 IotDeviceMessage 封套，method 来自 body）
  61 + IotDeviceMessage message = deviceMessageService.deserializeDeviceMessage(payload, productKey, deviceName);
  62 + if (message == null) {
  63 + log.warn("[handleBusinessRequest][消息解码失败，客户端 ID: {}，主题: {}]", clientId, topic);
  64 + return;
83 65 }
  66 + // 2.2 标准化回复消息的 method（MQTT 协议中，设备回复消息的 method 会携带 _reply 后缀）
  67 + IotMqttTopicUtils.normalizeReplyMethod(message);
84 68  
85 69 // 3. 处理业务消息
86 70 deviceMessageService.sendDeviceMessage(message, productKey, deviceName, serverId);
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/serialize/IotMessageSerializer.java
  View file @8865481
... ... @@ -3,8 +3,6 @@ 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   -
8 6 /**
9 7 * IoT 设备消息序列化器接口
10 8 *
... ... @@ -37,24 +35,4 @@ public interface IotMessageSerializer {
37 35 */
38 36 IotSerializeTypeEnum getType();
39 37  
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   -
60 38 }
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/serialize/json/IotJsonSerializer.java
  View file @8865481
... ... @@ -6,8 +6,6 @@ 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   -
11 9 /**
12 10 * JSON 格式的消息序列化器
13 11 *
... ... @@ -36,17 +34,4 @@ public class IotJsonSerializer implements IotMessageSerializer {
36 34 return message;
37 35 }
38 36  
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   -
52 37 }
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceMessageService.java
  View file @8865481
... ... @@ -62,16 +62,4 @@ 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   -
77 65 }
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceMessageServiceImpl.java
  View file @8865481
1 1 package cn.iocoder.yudao.module.iot.gateway.service.device.message;
2 2  
3   -import cn.hutool.core.convert.Convert;
4 3 import cn.hutool.core.lang.Assert;
5 4 import cn.hutool.core.util.StrUtil;
6 5 import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO;
7   -import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
8 6 import cn.iocoder.yudao.module.iot.core.enums.IotSerializeTypeEnum;
9 7 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
10 8 import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer;
... ... @@ -17,8 +15,6 @@ import lombok.extern.slf4j.Slf4j;
17 15 import org.springframework.stereotype.Service;
18 16  
19 17 import java.time.LocalDateTime;
20   -import java.util.Map;
21   -import java.util.function.Function;
22 18  
23 19 import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
24 20 import static cn.iocoder.yudao.module.iot.gateway.enums.ErrorCodeConstants.DEVICE_NOT_EXISTS;
... ... @@ -44,14 +40,6 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
44 40 @Override
45 41 public byte[] serializeDeviceMessage(IotDeviceMessage message,
46 42 String productKey, String deviceName) {
47   - if (IotDeviceMessageMethodEnum.isPassthrough(message.getMethod())) {
48   - IotDeviceRespDTO device = deviceService.getDeviceFromCache(productKey, deviceName);
49   - if (device == null) {
50   - throw exception(DEVICE_NOT_EXISTS, productKey, deviceName);
51   - }
52   - IotSerializeTypeEnum serializeType = IotSerializeTypeEnum.of(device.getSerializeType());
53   - return encodePassthrough(message, serializeType, productKey, deviceName);
54   - }
55 43 // 1.1 获取设备信息
56 44 IotDeviceRespDTO device = deviceService.getDeviceFromCache(productKey, deviceName);
57 45 if (device == null) {
... ... @@ -68,9 +56,6 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
68 56 @Override
69 57 public byte[] serializeDeviceMessage(IotDeviceMessage message,
70 58 IotSerializeTypeEnum serializeType) {
71   - if (IotDeviceMessageMethodEnum.isPassthrough(message.getMethod())) {
72   - return encodePassthrough(message, serializeType, null, null);
73   - }
74 59 // 1. 获取序列化器
75 60 IotMessageSerializer serializer = messageSerializerManager.get(serializeType);
76 61 if (serializer == null) {
... ... @@ -110,23 +95,6 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
110 95 }
111 96  
112 97 @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
130 98 public void sendDeviceMessage(IotDeviceMessage message,
131 99 String productKey, String deviceName, String serverId) {
132 100 // 1. 获取设备信息
... ... @@ -157,60 +125,4 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
157 125 }
158 126 }
159 127  
160   - /**
161   - * 影子透传编码；不支持的 serializeType 返回 null，由下行发布方跳过发送
162   - */
163   - private byte[] encodePassthrough(IotDeviceMessage message, IotSerializeTypeEnum serializeType,
164   - String productKey, String deviceName) {
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;
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) {
208   - if (productKey != null && deviceName != null) {
209   - log.warn("[{}][影子透传跳过：设备({}/{}) serializeType={} 不支持，目前仅 JSON 协议设备可用]",
210   - op, productKey, deviceName, serializeType);
211   - } else {
212   - log.warn("[{}][影子透传跳过：serializeType={} 不支持，目前仅 JSON 协议设备可用]", op, serializeType);
213   - }
214   - }
215   -
216 128 }
... ...
yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceShadowPassthroughTest.java deleted
View file @6e9b58c
1   -package cn.iocoder.yudao.module.iot.gateway.service.device.message;
2   -
3   -import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
4   -import cn.iocoder.yudao.framework.test.core.ut.BaseMockitoUnitTest;
5   -import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO;
6   -import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
7   -import cn.iocoder.yudao.module.iot.core.enums.IotSerializeTypeEnum;
8   -import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
9   -import cn.iocoder.yudao.module.iot.core.topic.shadow.IotDeviceShadowReplyDTO;
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;
14   -import org.junit.jupiter.api.Test;
15   -import org.mockito.InjectMocks;
16   -import org.mockito.Mock;
17   -import org.mockito.Spy;
18   -
19   -import java.util.HashMap;
20   -import java.util.Map;
21   -
22   -import static org.junit.jupiter.api.Assertions.*;
23   -import static org.mockito.ArgumentMatchers.eq;
24   -import static org.mockito.Mockito.when;
25   -
26   -/**
27   - * 设备影子网关透传序列化单元测试
28   - */
29   -public class IotDeviceShadowPassthroughTest extends BaseMockitoUnitTest {
30   -
31   - @InjectMocks
32   - private IotDeviceMessageServiceImpl deviceMessageService;
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   -
48   - @Test
49   - public void testSerializeDeviceMessage_passthrough_flatReply() {
50   - IotDeviceShadowReplyDTO reply = IotDeviceShadowReplyDTO.getReplyOf(new HashMap<>(), new HashMap<>(), 2L, 1469564576L);
51   - IotDeviceMessage message = IotDeviceMessage.requestOf(
52   - IotDeviceMessageMethodEnum.SHADOW_GET.getMethod(), reply);
53   -
54   - byte[] bytes = deviceMessageService.serializeDeviceMessage(message, IotSerializeTypeEnum.JSON);
55   -
56   - @SuppressWarnings("unchecked")
57   - Map<String, Object> result = JsonUtils.parseObject(bytes, Map.class);
58   - assertEquals("get_reply", result.get("method"));
59   - assertEquals(0, ((Number) result.get("code")).intValue());
60   - assertNull(result.get("payload"));
61   - assertNull(result.get("id"));
62   - @SuppressWarnings("unchecked")
63   - Map<String, Object> data = (Map<String, Object>) result.get("data");
64   - assertEquals(2, ((Number) data.get("version")).intValue());
65   - }
66   -
67   - @Test
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");
75   -
76   - assertNotNull(message);
77   - assertEquals(IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), message.getMethod());
78   - assertEquals("req-1", message.getRequestId());
79   - assertEquals("get", ((Map<?, ?>) message.getParams()).get("method"));
80   - }
81   -
82   - @Test
83   - public void testSerializeUpdateReply_passthrough() {
84   - IotDeviceShadowReplyDTO reply = IotDeviceShadowReplyDTO.updateReplyOf(3L, 1469564576L);
85   - IotDeviceMessage message = IotDeviceMessage.requestOf(
86   - IotDeviceMessageMethodEnum.SHADOW_GET.getMethod(), reply);
87   -
88   - byte[] bytes = deviceMessageService.serializeDeviceMessage(message, IotSerializeTypeEnum.JSON);
89   - @SuppressWarnings("unchecked")
90   - Map<String, Object> result = JsonUtils.parseObject(bytes, Map.class);
91   - assertEquals("update_reply", result.get("method"));
92   - assertEquals(0, ((Number) result.get("code")).intValue());
93   - assertNull(result.get("id"));
94   - }
95   -
96   - @Test
97   - public void testJsonSerializer_encodeRawDecodeRaw() {
98   - IotJsonSerializer serializer = new IotJsonSerializer();
99   - Map<String, Object> params = Map.of("method", "get");
100   - byte[] bytes = serializer.encodeRaw(params);
101   - Map<String, Object> decoded = serializer.decodeRaw(bytes);
102   - assertNotNull(decoded);
103   - assertEquals("get", decoded.get("method"));
104   - }
105   -
106   - @Test
107   - public void testSerializeDeviceMessage_passthrough_binaryReturnsNull() {
108   - IotDeviceMessage message = IotDeviceMessage.requestOf(
109   - IotDeviceMessageMethodEnum.SHADOW_GET.getMethod(), Map.of("method", "get_reply"));
110   - assertNull(deviceMessageService.serializeDeviceMessage(message, IotSerializeTypeEnum.BINARY));
111   - }
112   -
113   - @Test
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());
150   - }
151   -
152   -}
yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/util/IotMqttTopicUtilsTest.java 0 → 100644
  View file @8865481
  1 +package cn.iocoder.yudao.module.iot.gateway.util;
  2 +
  3 +import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
  4 +import org.junit.jupiter.api.Test;
  5 +
  6 +import static org.junit.jupiter.api.Assertions.assertEquals;
  7 +
  8 +/**
  9 + * {@link IotMqttTopicUtils} 单元测试
  10 + */
  11 +class IotMqttTopicUtilsTest {
  12 +
  13 + private static final String PRODUCT_KEY = "pk";
  14 + private static final String DEVICE_NAME = "dn";
  15 +
  16 + @Test
  17 + void testBuildTopicByMethod_shadowUpdateReplyAppendsReplySuffix() {
  18 + String topic = IotMqttTopicUtils.buildTopicByMethod(
  19 + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), PRODUCT_KEY, DEVICE_NAME, true);
  20 + assertEquals("/sys/pk/dn/thing/shadow/update_reply", topic);
  21 + }
  22 +
  23 + @Test
  24 + void testBuildTopicByMethod_serviceInvokeReplyAppendsReplySuffix() {
  25 + String topic = IotMqttTopicUtils.buildTopicByMethod(
  26 + IotDeviceMessageMethodEnum.SERVICE_INVOKE.getMethod(), PRODUCT_KEY, DEVICE_NAME, true);
  27 + assertEquals("/sys/pk/dn/thing/service/invoke_reply", topic);
  28 + }
  29 +
  30 +}
... ...