From 8865481b2e9d2357998b09196510fc174d662299 Mon Sep 17 00:00:00 2001 From: haohao <1036606149@qq.com> Date: Sun, 21 Jun 2026 11:36:04 +0800 Subject: [PATCH] refactor(iot): 设备影子改回标准 IotDeviceMessage 封套，移除透传链路 --- yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImpl.java | 135 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++---------------------------------------------------------------- yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImplTest.java | 34 +++++++++++++++++++--------------- yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotDeviceMessageMethodEnum.java | 15 +-------------- yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/shadow/IotDeviceShadowErrorCodeEnum.java | 2 +- yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/shadow/IotDeviceShadowReplyDTO.java | 83 ----------------------------------------------------------------------------------- yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/shadow/IotDeviceShadowUpdateReqDTO.java | 2 +- yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceShadowProtocolUtils.java | 36 +++++++++++++++++++----------------- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/downstream/IotEmqxDownstreamHandler.java | 5 ----- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/upstream/IotEmqxUpstreamHandler.java | 29 +++++++---------------------- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/downstream/IotMqttDownstreamHandler.java | 5 ----- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/upstream/IotMqttAbstractHandler.java | 5 ----- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/upstream/IotMqttUpstreamHandler.java | 30 +++++++----------------------- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/serialize/IotMessageSerializer.java | 22 ---------------------- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/serialize/json/IotJsonSerializer.java | 15 --------------- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceMessageService.java | 12 ------------ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceMessageServiceImpl.java | 88 ---------------------------------------------------------------------------------------- yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceShadowPassthroughTest.java | 152 -------------------------------------------------------------------------------------------------------------------------------------------------------- yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/util/IotMqttTopicUtilsTest.java | 30 ++++++++++++++++++++++++++++++ 18 files changed, 156 insertions(+), 544 deletions(-) delete mode 100644 yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/shadow/IotDeviceShadowReplyDTO.java delete mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceShadowPassthroughTest.java create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/util/IotMqttTopicUtilsTest.java diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImpl.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImpl.java index d09a88a..173eac4 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImpl.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImpl.java @@ -8,7 +8,6 @@ import cn.iocoder.yudao.framework.common.util.json.JsonUtils; import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum; import cn.iocoder.yudao.module.iot.core.enums.shadow.IotDeviceShadowErrorCodeEnum; import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; -import cn.iocoder.yudao.module.iot.core.topic.shadow.IotDeviceShadowReplyDTO; import cn.iocoder.yudao.module.iot.core.topic.shadow.IotDeviceShadowUpdateReqDTO; import cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils; import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO; @@ -31,8 +30,10 @@ import java.util.HashMap; import java.util.Map; import java.util.function.Function; +import static cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants.SUCCESS; import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception; -import static cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils.*; +import static cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils.INNER_METHOD_GET; +import static cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils.INNER_METHOD_UPDATE; import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.DEVICE_SHADOW_NOT_ENABLED; import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.DEVICE_SHADOW_VERSION_CONFLICT; @@ -88,29 +89,23 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { } catch (Exception ex) { log.error("[handleDeviceShadowUpdate][device({}) message({}) 处理异常]", device.getId(), message, ex); Long version = resolveShadowVersionSafely(device.getId()); - sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf( - extractInnerMethod(message.getParams()), version, - IotDeviceShadowErrorCodeEnum.SERVER_ERROR.getCode(), - IotDeviceShadowErrorCodeEnum.SERVER_ERROR.getMessage(), DateUtil.currentSeconds())); + sendShadowErrorReply(device, message.getServerId(), message.getRequestId(), + IotDeviceShadowErrorCodeEnum.SERVER_ERROR, version); } } private void handleDeviceShadowUpdate0(IotDeviceDO device, IotDeviceMessage message) { - // 1. 解析请求（影子 body 当前按 JSON 建模，非 JSON 设备在 gateway 透传层已拒绝） + // 1. 解析请求（影子 params 按 JSON 建模，设备须配置 serializeType=JSON） IotDeviceShadowUpdateReqDTO req = JsonUtils.convertObject(message.getParams(), IotDeviceShadowUpdateReqDTO.class); if (req == null) { - sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf( - extractInnerMethod(message.getParams()), 0L, - IotDeviceShadowErrorCodeEnum.JSON_FORMAT_ERROR.getCode(), - IotDeviceShadowErrorCodeEnum.JSON_FORMAT_ERROR.getMessage(), DateUtil.currentSeconds())); + sendShadowErrorReply(device, message.getServerId(), message.getRequestId(), + IotDeviceShadowErrorCodeEnum.JSON_FORMAT_ERROR, 0L); return; } if (StrUtil.isEmpty(req.getMethod())) { - sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf( - INNER_METHOD_UPDATE, resolveShadowVersionSafely(device.getId()), - IotDeviceShadowErrorCodeEnum.METHOD_MISSING.getCode(), - IotDeviceShadowErrorCodeEnum.METHOD_MISSING.getMessage(), DateUtil.currentSeconds())); + sendShadowErrorReply(device, message.getServerId(), message.getRequestId(), + IotDeviceShadowErrorCodeEnum.METHOD_MISSING, resolveShadowVersionSafely(device.getId())); return; } @@ -123,42 +118,35 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { // 3. 内层 method 分发 if (INNER_METHOD_GET.equals(req.getMethod())) { - sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.getReplyOf( - shadow.getDesired(), shadow.getReported(), shadow.getVersion(), shadow.getShadowTime())); + sendShadowGetReply(device, message.getServerId(), message.getRequestId(), + shadow.getDesired(), shadow.getReported(), shadow.getVersion(), shadow.getShadowTime()); return; } if (INNER_METHOD_UPDATE.equals(req.getMethod())) { - handleReportedUpdate(device, message.getServerId(), req); + handleReportedUpdate(device, message.getServerId(), message.getRequestId(), req); return; } - sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf( - req.getMethod(), shadow.getVersion(), - IotDeviceShadowErrorCodeEnum.INVALID_METHOD.getCode(), - IotDeviceShadowErrorCodeEnum.INVALID_METHOD.getMessage(), DateUtil.currentSeconds())); + sendShadowErrorReply(device, message.getServerId(), message.getRequestId(), + IotDeviceShadowErrorCodeEnum.INVALID_METHOD, shadow.getVersion()); } - private void handleReportedUpdate(IotDeviceDO device, String serverId, IotDeviceShadowUpdateReqDTO req) { + private void handleReportedUpdate(IotDeviceDO device, String serverId, String requestId, + IotDeviceShadowUpdateReqDTO req) { Long deviceId = device.getId(); if (req.getState() == null) { - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf( - INNER_METHOD_UPDATE, resolveShadowVersionSafely(deviceId), - IotDeviceShadowErrorCodeEnum.STATE_MISSING.getCode(), - IotDeviceShadowErrorCodeEnum.STATE_MISSING.getMessage(), DateUtil.currentSeconds())); + sendShadowErrorReply(device, serverId, requestId, + IotDeviceShadowErrorCodeEnum.STATE_MISSING, resolveShadowVersionSafely(deviceId)); return; } if (req.getState().getReported() == null) { - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf( - INNER_METHOD_UPDATE, resolveShadowVersionSafely(deviceId), - IotDeviceShadowErrorCodeEnum.REPORTED_MISSING.getCode(), - IotDeviceShadowErrorCodeEnum.REPORTED_MISSING.getMessage(), DateUtil.currentSeconds())); + sendShadowErrorReply(device, serverId, requestId, + IotDeviceShadowErrorCodeEnum.REPORTED_MISSING, resolveShadowVersionSafely(deviceId)); return; } if (CollUtil.isEmpty(req.getState().getReported())) { - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf( - INNER_METHOD_UPDATE, resolveShadowVersionSafely(deviceId), - IotDeviceShadowErrorCodeEnum.REPORTED_EMPTY.getCode(), - IotDeviceShadowErrorCodeEnum.REPORTED_EMPTY.getMessage(), DateUtil.currentSeconds())); + sendShadowErrorReply(device, serverId, requestId, + IotDeviceShadowErrorCodeEnum.REPORTED_EMPTY, resolveShadowVersionSafely(deviceId)); return; } @@ -166,24 +154,20 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { if (req.getVersion() != null && req.getVersion() == -1L) { try { resetShadowWithRetry(deviceId); - sendShadowReplySafely(device, serverId, - IotDeviceShadowReplyDTO.updateReplyOf(0L, DateUtil.currentSeconds())); + sendShadowUpdateReply(device, serverId, requestId, 0L, DateUtil.currentSeconds()); } catch (ShadowCasExhaustedException ex) { - handleCasExhaustedForDevice(device, serverId, ex); + handleCasExhaustedForDevice(device, serverId, requestId, ex); } return; } try { IotDeviceShadowDO committed = casUpdateShadow(deviceId, latest -> buildReportedUpdate(latest, req)); - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.updateReplyOf(committed.getVersion(), - committed.getShadowTime())); + sendShadowUpdateReply(device, serverId, requestId, committed.getVersion(), committed.getShadowTime()); } catch (ShadowUpdateRejectedException ex) { - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf( - INNER_METHOD_UPDATE, ex.getVersion(), - ex.getErrorCode().getCode(), ex.getErrorCode().getMessage(), DateUtil.currentSeconds())); + sendShadowErrorReply(device, serverId, requestId, ex.getErrorCode(), ex.getVersion()); } catch (ShadowCasExhaustedException ex) { - handleCasExhaustedForDevice(device, serverId, ex); + handleCasExhaustedForDevice(device, serverId, requestId, ex); } } @@ -203,9 +187,9 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { // 设备在线则主动推送全量影子（复用 CAS 返回值，避免二次查库） String serverId = devicePropertyService.getDeviceServerId(deviceId); if (StrUtil.isNotEmpty(serverId)) { - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.getReplyOf( + sendShadowGetReply(device, serverId, null, committed.getDesired(), committed.getReported(), - committed.getVersion(), committed.getShadowTime())); + committed.getVersion(), committed.getShadowTime()); } } catch (ShadowCasExhaustedException ex) { throw exception(DEVICE_SHADOW_VERSION_CONFLICT); @@ -337,7 +321,7 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { } /** - * 将 CAS 成功写入的局部字段与 latest 快照合并，供全量 get_reply 使用 + * 将 CAS 成功写入的局部字段与 latest 快照合并，供全量 get 回复使用 */ private static void enrichCommittedShadow(IotDeviceShadowDO committed, IotDeviceShadowDO latest) { if (committed.getDesired() == null) { @@ -351,22 +335,19 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { } } - private void handleCasExhaustedForDevice(IotDeviceDO device, String serverId, ShadowCasExhaustedException ex) { + private void handleCasExhaustedForDevice(IotDeviceDO device, String serverId, String requestId, + ShadowCasExhaustedException ex) { if (ex.isShadowNotFound()) { - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf( - INNER_METHOD_UPDATE, 0L, - IotDeviceShadowErrorCodeEnum.SERVER_ERROR.getCode(), - IotDeviceShadowErrorCodeEnum.SERVER_ERROR.getMessage(), DateUtil.currentSeconds())); + sendShadowErrorReply(device, serverId, requestId, + IotDeviceShadowErrorCodeEnum.SERVER_ERROR, 0L); return; } - sendVersionConflictReply(device, serverId, ex.getLastKnownVersion()); + sendVersionConflictReply(device, serverId, requestId, ex.getLastKnownVersion()); } - private void sendVersionConflictReply(IotDeviceDO device, String serverId, Long version) { - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf( - INNER_METHOD_UPDATE, version, - IotDeviceShadowErrorCodeEnum.VERSION_CONFLICT.getCode(), - IotDeviceShadowErrorCodeEnum.VERSION_CONFLICT.getMessage(), DateUtil.currentSeconds())); + private void sendVersionConflictReply(IotDeviceDO device, String serverId, String requestId, Long version) { + sendShadowErrorReply(device, serverId, requestId, + IotDeviceShadowErrorCodeEnum.VERSION_CONFLICT, version); } private Long resolveShadowVersionSafely(Long deviceId) { @@ -407,15 +388,41 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { return shadow; } + private void sendShadowGetReply(IotDeviceDO device, String serverId, String requestId, + Map desired, Map reported, Long version, Long timestamp) { + sendShadowReplySafely(device, serverId, requestId, SUCCESS.getCode(), SUCCESS.getMsg(), + IotDeviceShadowProtocolUtils.buildGetData(desired, reported, version, timestamp)); + } + + private void sendShadowUpdateReply(IotDeviceDO device, String serverId, String requestId, + Long version, Long timestamp) { + sendShadowReplySafely(device, serverId, requestId, SUCCESS.getCode(), SUCCESS.getMsg(), + IotDeviceShadowProtocolUtils.buildUpdateData(version, timestamp)); + } + + /** + * 错误回复：data 仍带 version 供设备 resync（有意偏离纯 CommonResult 的 error-data-null 约定） + */ + private void sendShadowErrorReply(IotDeviceDO device, String serverId, String requestId, + IotDeviceShadowErrorCodeEnum errorCode, Long version) { + sendShadowReplySafely(device, serverId, requestId, errorCode.getCode(), errorCode.getMessage(), + IotDeviceShadowProtocolUtils.buildUpdateData(version, DateUtil.currentSeconds())); + } + /** - * 下发影子回复，内部吞掉异常，保证上行处理链路不向上抛 + * 下发影子回复（标准 IotDeviceMessage 封套），内部吞掉异常，保证上行处理链路不向上抛 + *

+ * 回复 method 取上行的 {@link IotDeviceMessageMethodEnum#SHADOW_UPDATE}（与框架"回复用原请求 method"一致）： + * 带 code 的回复经 IotDeviceMessageUtils#isUpstreamMessage 取反 SHADOW_UPDATE.upstream(true) → 判定为下行， + * 正确投递到设备；下行 Topic 为 thing/shadow/update_reply。 */ - private void sendShadowReplySafely(IotDeviceDO device, String serverId, IotDeviceShadowReplyDTO replyDTO) { - IotDeviceMessage shadowGetMsg = IotDeviceMessage.requestOf( - device.getId(), device.getTenantId(), serverId, - IotDeviceMessageMethodEnum.SHADOW_GET.getMethod(), replyDTO); + private void sendShadowReplySafely(IotDeviceDO device, String serverId, String requestId, + Integer code, String msg, Map data) { + IotDeviceMessage shadowReplyMsg = IotDeviceMessage.replyOf( + requestId, IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), data, code, msg); + shadowReplyMsg.setDeviceId(device.getId()).setTenantId(device.getTenantId()).setServerId(serverId); try { - deviceMessageService.sendDeviceMessage(shadowGetMsg, device, serverId); + deviceMessageService.sendDeviceMessage(shadowReplyMsg, device, serverId); } catch (Exception ex) { log.error("[sendShadowReply][device({}) 下发影子回复失败]", device.getId(), ex); } diff --git a/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImplTest.java b/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImplTest.java index 71ed6f4..57779db 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImplTest.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImplTest.java @@ -5,7 +5,6 @@ import cn.iocoder.yudao.framework.common.exception.ServiceException; import cn.iocoder.yudao.framework.test.core.ut.BaseMockitoUnitTest; import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum; import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; -import cn.iocoder.yudao.module.iot.core.topic.shadow.IotDeviceShadowReplyDTO; import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO; import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceShadowDO; import cn.iocoder.yudao.module.iot.dal.mysql.device.IotDeviceShadowMapper; @@ -76,15 +75,18 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest { Map params = new HashMap<>(); params.put("method", "get"); - IotDeviceMessage message = IotDeviceMessage.requestOf(device.getId(), 1L, "server-1", + IotDeviceMessage message = IotDeviceMessage.requestOf("req-get-1", IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), params); + message.setDeviceId(device.getId()).setTenantId(1L).setServerId("server-1"); service.handleDeviceShadowUpdate(device, message); ArgumentCaptor captor = ArgumentCaptor.forClass(IotDeviceMessage.class); verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1")); - IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams(); - assertEquals("get_reply", reply.getMethod()); + IotDeviceMessage reply = captor.getValue(); + assertEquals(IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), reply.getMethod()); + assertEquals("req-get-1", reply.getRequestId()); + assertEquals(0, reply.getCode()); assertEquals(2L, getReplyVersion(reply)); } @@ -141,7 +143,8 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest { ArgumentCaptor captor = ArgumentCaptor.forClass(IotDeviceMessage.class); verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1")); - IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams(); + IotDeviceMessage reply = captor.getValue(); + assertEquals(IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), reply.getMethod()); assertEquals(409, reply.getCode()); verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any()); } @@ -192,8 +195,8 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest { ArgumentCaptor captor = ArgumentCaptor.forClass(IotDeviceMessage.class); verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1")); - IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams(); - assertEquals("update_reply", reply.getMethod()); + IotDeviceMessage reply = captor.getValue(); + assertEquals(IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), reply.getMethod()); assertEquals(409, reply.getCode()); verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any()); } @@ -262,8 +265,8 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest { ArgumentCaptor captor = ArgumentCaptor.forClass(IotDeviceMessage.class); verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1")); - IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams(); - assertEquals("get_reply", reply.getMethod()); + IotDeviceMessage reply = captor.getValue(); + assertEquals(IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), reply.getMethod()); assertEquals(2L, getReplyVersion(reply)); // getOrCreateShadow + CAS 各一次 SELECT，推送不再额外查库 verify(deviceShadowMapper, times(2)).selectByDeviceId(deviceId); @@ -299,7 +302,8 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest { ArgumentCaptor captor = ArgumentCaptor.forClass(IotDeviceMessage.class); verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1")); - IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams(); + IotDeviceMessage reply = captor.getValue(); + assertEquals(IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), reply.getMethod()); assertEquals(400, reply.getCode()); } @@ -319,7 +323,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest { ArgumentCaptor captor = ArgumentCaptor.forClass(IotDeviceMessage.class); verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1")); - IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams(); + IotDeviceMessage reply = captor.getValue(); assertEquals(401, reply.getCode()); } @@ -375,7 +379,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest { ArgumentCaptor replyCaptor = ArgumentCaptor.forClass(IotDeviceMessage.class); verify(deviceMessageService).sendDeviceMessage(replyCaptor.capture(), eq(device), eq("server-1")); - IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) replyCaptor.getValue().getParams(); + IotDeviceMessage reply = replyCaptor.getValue(); assertEquals(3L, getReplyVersion(reply)); } @@ -439,7 +443,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest { ArgumentCaptor captor = ArgumentCaptor.forClass(IotDeviceMessage.class); verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1")); - IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams(); + IotDeviceMessage reply = captor.getValue(); assertEquals(0L, getReplyVersion(reply)); assertEquals(500, reply.getCode()); } @@ -466,7 +470,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest { ArgumentCaptor captor = ArgumentCaptor.forClass(IotDeviceMessage.class); verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1")); - IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams(); + IotDeviceMessage reply = captor.getValue(); assertEquals(500, reply.getCode()); verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any()); } @@ -501,7 +505,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest { } @SuppressWarnings("unchecked") - private static Long getReplyVersion(IotDeviceShadowReplyDTO reply) { + private static Long getReplyVersion(IotDeviceMessage reply) { if (!(reply.getData() instanceof Map)) { return null; } diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotDeviceMessageMethodEnum.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotDeviceMessageMethodEnum.java index c57f87b..1e536ee 100644 --- a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotDeviceMessageMethodEnum.java +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotDeviceMessageMethodEnum.java @@ -70,7 +70,6 @@ public enum IotDeviceMessageMethodEnum implements ArrayValuable { // ========== 设备影子 ========== SHADOW_UPDATE("thing.shadow.update", "设备影子更新", true), - SHADOW_GET("thing.shadow.get", "设备影子获取", false), ; @@ -83,15 +82,7 @@ public enum IotDeviceMessageMethodEnum implements ArrayValuable { public static final Set REPLY_DISABLED = SetUtils.asSet( STATE_UPDATE.getMethod(), OTA_PROGRESS.getMethod(), // 参考阿里云，OTA 升级进度上报，不进行回复 - SHADOW_UPDATE.getMethod() // 设备影子由影子 Service 主动下发 thing.shadow.get - ); - - /** - * 透传方法集合：设备侧使用原始线格式，不套 IotDeviceMessage 封套 - */ - public static final Set PASSTHROUGH_METHODS = SetUtils.asSet( - SHADOW_UPDATE.getMethod(), - SHADOW_GET.getMethod() + SHADOW_UPDATE.getMethod() // 设备影子由影子 Service 主动下发回复（thing/shadow/update_reply） ); private final String method; @@ -114,8 +105,4 @@ public enum IotDeviceMessageMethodEnum implements ArrayValuable { return REPLY_DISABLED.contains(method); } - public static boolean isPassthrough(String method) { - return PASSTHROUGH_METHODS.contains(method); - } - } diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/shadow/IotDeviceShadowErrorCodeEnum.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/shadow/IotDeviceShadowErrorCodeEnum.java index 2a304e1..dcf85b7 100644 --- a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/shadow/IotDeviceShadowErrorCodeEnum.java +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/shadow/IotDeviceShadowErrorCodeEnum.java @@ -6,7 +6,7 @@ import lombok.Getter; /** * IoT 设备影子设备侧错误码枚举 *

- * 设备侧 wire code，仅用于回填 {@code update_reply.code/msg}，不走 + * 设备侧 wire code，仅用于回填影子下行回复（thing.shadow.update_reply）的 code/msg，不走 * {@link cn.iocoder.yudao.framework.common.pojo.CommonResult}。 * 服务端内部异常请使用 {@link cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants}（如 * {@code DEVICE_SHADOW_NOT_ENABLED}）。 diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/shadow/IotDeviceShadowReplyDTO.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/shadow/IotDeviceShadowReplyDTO.java deleted file mode 100644 index b0f7f8d..0000000 --- a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/shadow/IotDeviceShadowReplyDTO.java +++ /dev/null @@ -1,83 +0,0 @@ -package cn.iocoder.yudao.module.iot.core.topic.shadow; - -import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum; -import lombok.AllArgsConstructor; -import lombok.Data; -import lombok.NoArgsConstructor; - -import java.util.HashMap; -import java.util.Map; - -import static cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants.SUCCESS; -import static cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils.*; - -/** - * IoT 设备影子下行 Reply DTO - *

- * 扁平对齐 CommonResult + method，用于 {@link IotDeviceMessageMethodEnum#SHADOW_GET} - * 透传线 body。 - * 错误回复的 {@link #data} 仍携带 version 供设备 resync，有意偏离纯 CommonResult（error 时 data - * 通常为 null）。 - * - * @author 芋道源码 - */ -@Data -@NoArgsConstructor -@AllArgsConstructor -public class IotDeviceShadowReplyDTO { - - /** - * 内层方法：get_reply / update_reply - */ - private String method; - - /** - * 结果码，0 表示成功 - */ - private Integer code; - - /** - * 结果描述 - */ - private String msg; - - /** - * 业务数据：get_reply 为 {state, version, timestamp}；update_reply 为 {version, - * timestamp} - */ - private Object data; - - public static IotDeviceShadowReplyDTO getReplyOf(Map desired, Map reported, - Long version, Long timestamp) { - Map state = new HashMap<>(); - state.put(STATE_DESIRED, desired != null ? desired : new HashMap<>()); - state.put(STATE_REPORTED, reported != null ? reported : new HashMap<>()); - Map data = new HashMap<>(); - data.put(PAYLOAD_STATE, state); - data.put(PAYLOAD_VERSION, version); - data.put(METADATA_TIMESTAMP, timestamp); - return of(INNER_METHOD_GET_REPLY, SUCCESS.getCode(), SUCCESS.getMsg(), data); - } - - public static IotDeviceShadowReplyDTO updateReplyOf(Long version, Long timestamp) { - return of(INNER_METHOD_UPDATE_REPLY, SUCCESS.getCode(), SUCCESS.getMsg(), - dataWithVersionTimestamp(version, timestamp)); - } - - public static IotDeviceShadowReplyDTO errorOf(String innerMethod, Long version, Integer errorCode, - String errorMsg, Long timestamp) { - return of(toReplyMethod(innerMethod), errorCode, errorMsg, dataWithVersionTimestamp(version, timestamp)); - } - - private static IotDeviceShadowReplyDTO of(String method, Integer code, String msg, Map data) { - return new IotDeviceShadowReplyDTO().setMethod(method).setCode(code).setMsg(msg).setData(data); - } - - private static Map dataWithVersionTimestamp(Long version, Long timestamp) { - Map data = new HashMap<>(); - data.put(PAYLOAD_VERSION, version); - data.put(METADATA_TIMESTAMP, timestamp); - return data; - } - -} diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/shadow/IotDeviceShadowUpdateReqDTO.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/shadow/IotDeviceShadowUpdateReqDTO.java index 0464b5b..b75edca 100644 --- a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/shadow/IotDeviceShadowUpdateReqDTO.java +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/shadow/IotDeviceShadowUpdateReqDTO.java @@ -10,7 +10,7 @@ import java.util.Map; /** * IoT 设备影子上行 Request DTO *

- * 用于 {@link IotDeviceMessageMethodEnum#SHADOW_UPDATE} 消息的 params 参数（透传原始线 body） + * 用于 {@link IotDeviceMessageMethodEnum#SHADOW_UPDATE} 标准封套的 params 参数 * * @author 芋道源码 */ diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceShadowProtocolUtils.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceShadowProtocolUtils.java index e886479..9e9e9ad 100644 --- a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceShadowProtocolUtils.java +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceShadowProtocolUtils.java @@ -9,7 +9,7 @@ import java.util.Set; /** * IoT 设备影子协议工具 *

- * 内层 method、state 键名、payload 字段等与设备侧透传 JSON 对齐；并提供 metadata 读写。 + * 内层 method、state 键名、reply data 字段等与设备侧标准封套 params/data 对齐；并提供 metadata 读写。 * * @author 芋道源码 */ @@ -18,12 +18,10 @@ public final class IotDeviceShadowProtocolUtils { private IotDeviceShadowProtocolUtils() { } - // ========== 内层 method ========== + // ========== 内层 method（上行 params.method） ========== public static final String INNER_METHOD_GET = "get"; public static final String INNER_METHOD_UPDATE = "update"; - public static final String INNER_METHOD_GET_REPLY = "get_reply"; - public static final String INNER_METHOD_UPDATE_REPLY = "update_reply"; // ========== state / metadata 键 ========== @@ -37,24 +35,28 @@ public final class IotDeviceShadowProtocolUtils { public static final String PAYLOAD_VERSION = "version"; /** - * 由内层 method 推导 reply method（get → get_reply，其余 → update_reply） + * 构建 get 响应 data：{ state:{desired,reported}, version, timestamp } */ - public static String toReplyMethod(String innerMethod) { - if (INNER_METHOD_GET.equals(innerMethod)) { - return INNER_METHOD_GET_REPLY; - } - return INNER_METHOD_UPDATE_REPLY; + public static Map buildGetData(Map desired, Map reported, + Long version, Long timestamp) { + Map state = new HashMap<>(); + state.put(STATE_DESIRED, desired != null ? desired : new HashMap<>()); + state.put(STATE_REPORTED, reported != null ? reported : new HashMap<>()); + Map data = new HashMap<>(); + data.put(PAYLOAD_STATE, state); + data.put(PAYLOAD_VERSION, version); + data.put(METADATA_TIMESTAMP, timestamp); + return data; } /** - * 从透传 params 提取内层 method；无法解析时默认 update + * 构建 update 响应 / 错误 data：{ version, timestamp } */ - public static String extractInnerMethod(Object params) { - if (!(params instanceof Map)) { - return INNER_METHOD_UPDATE; - } - Object method = ((Map) params).get("method"); - return method != null ? String.valueOf(method) : INNER_METHOD_UPDATE; + public static Map buildUpdateData(Long version, Long timestamp) { + Map data = new HashMap<>(); + data.put(PAYLOAD_VERSION, version); + data.put(METADATA_TIMESTAMP, timestamp); + return data; } /** diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/downstream/IotEmqxDownstreamHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/downstream/IotEmqxDownstreamHandler.java index 6d03d35..db5ea12 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/downstream/IotEmqxDownstreamHandler.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/downstream/IotEmqxDownstreamHandler.java @@ -55,11 +55,6 @@ public class IotEmqxDownstreamHandler { // 2.2 构建载荷 byte[] payload = deviceMessageService.serializeDeviceMessage(message, deviceInfo.getProductKey(), deviceInfo.getDeviceName()); - if (payload == null) { - log.warn("[handle][影子下行跳过，设备 ID: {}，method: {}，非 JSON 协议设备不支持影子透传]", - message.getDeviceId(), message.getMethod()); - return; - } // 3. 发布消息 protocol.publishMessage(topic, payload); diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/upstream/IotEmqxUpstreamHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/upstream/IotEmqxUpstreamHandler.java index d691604..1587416 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/upstream/IotEmqxUpstreamHandler.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/upstream/IotEmqxUpstreamHandler.java @@ -3,16 +3,12 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.emqx.handler.upstream; import cn.hutool.core.util.ArrayUtil; import cn.hutool.core.util.StrUtil; import cn.hutool.extra.spring.SpringUtil; -import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum; import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService; import cn.iocoder.yudao.module.iot.gateway.util.IotMqttTopicUtils; import io.vertx.mqtt.messages.MqttPublishMessage; import lombok.extern.slf4j.Slf4j; -import java.util.Arrays; -import java.util.stream.Collectors; - /** * IoT 网关 EMQX 上行消息处理器 * @@ -47,25 +43,14 @@ public class IotEmqxUpstreamHandler { return; } - // 2.1 反序列化消息 - String topicMethod = Arrays.stream(topicParts, 4, topicParts.length) - .collect(Collectors.joining(".")); - IotDeviceMessage message; - if (IotDeviceMessageMethodEnum.isPassthrough(topicMethod)) { - message = deviceMessageService.deserializePassthroughMessage(payload, topicMethod, productKey, deviceName); - if (message == null) { - log.warn("[handle][topic({}) 影子透传 payload 解析失败]", topic); - return; - } - } else { - message = deviceMessageService.deserializeDeviceMessage(payload, productKey, deviceName); - if (message == null) { - log.warn("[handle][topic({}) payload({}) 消息解码失败]", topic, new String(payload)); - return; - } - // 2.2 标准化回复消息的 method（MQTT 协议中，设备回复消息的 method 会携带 _reply 后缀） - IotMqttTopicUtils.normalizeReplyMethod(message); + // 2.1 反序列化消息（标准 IotDeviceMessage 封套，method 来自 body） + IotDeviceMessage message = deviceMessageService.deserializeDeviceMessage(payload, productKey, deviceName); + if (message == null) { + log.warn("[handle][topic({}) payload({}) 消息解码失败]", topic, new String(payload)); + return; } + // 2.2 标准化回复消息的 method（MQTT 协议中，设备回复消息的 method 会携带 _reply 后缀） + IotMqttTopicUtils.normalizeReplyMethod(message); // 3. 发送消息到队列 deviceMessageService.sendDeviceMessage(message, productKey, deviceName, serverId); diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/downstream/IotMqttDownstreamHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/downstream/IotMqttDownstreamHandler.java index 4f96df4..f2ae99f 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/downstream/IotMqttDownstreamHandler.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/downstream/IotMqttDownstreamHandler.java @@ -45,11 +45,6 @@ public class IotMqttDownstreamHandler { // 2.1 序列化消息 byte[] payload = deviceMessageService.serializeDeviceMessage(message, connectionInfo.getProductKey(), connectionInfo.getDeviceName()); - if (payload == null) { - log.warn("[handle][影子下行跳过，设备 ID: {}，method: {}，非 JSON 协议设备不支持影子透传]", - message.getDeviceId(), message.getMethod()); - return; - } Assert.isTrue(payload.length > 0, "消息编码结果不能为空"); // 2.2 构建主题 Assert.notBlank(message.getMethod(), "消息方法不能为空"); diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/upstream/IotMqttAbstractHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/upstream/IotMqttAbstractHandler.java index 4d9da08..443dc10 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/upstream/IotMqttAbstractHandler.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/upstream/IotMqttAbstractHandler.java @@ -72,11 +72,6 @@ public abstract class IotMqttAbstractHandler { try { // 1.1 序列化消息（根据设备配置的序列化类型） byte[] encodedData = deviceMessageService.serializeDeviceMessage(responseMessage, productKey, deviceName); - if (encodedData == null) { - // 影子透传在非 JSON 设备上 encode 返回 null，跳过发布避免 NPE - log.warn("[writeResponse][影子下行跳过，客户端 ID: {}，method: {}]", endpoint.clientIdentifier(), method); - return; - } // 1.2 构建响应主题 String replyTopic = IotMqttTopicUtils.buildTopicByMethod(method, productKey, deviceName, true); diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/upstream/IotMqttUpstreamHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/upstream/IotMqttUpstreamHandler.java index 94ae2c9..d0da391 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/upstream/IotMqttUpstreamHandler.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/upstream/IotMqttUpstreamHandler.java @@ -2,7 +2,6 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.handler.upstream; import cn.hutool.core.lang.Assert; import cn.hutool.core.util.ArrayUtil; -import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum; import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.manager.IotMqttConnectionManager; import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService; @@ -10,10 +9,6 @@ import cn.iocoder.yudao.module.iot.gateway.util.IotMqttTopicUtils; import io.vertx.mqtt.MqttEndpoint; import lombok.extern.slf4j.Slf4j; -import java.util.Arrays; -import java.util.stream.Collectors; - - /** * IoT 网关 MQTT 上行消息处理器：处理业务消息（属性上报、事件上报等） * @@ -62,25 +57,14 @@ public class IotMqttUpstreamHandler extends IotMqttAbstractHandler { return; } - // 2.1 反序列化消息 - String topicMethod = Arrays.stream(topicParts, 4, topicParts.length) - .collect(Collectors.joining(".")); - IotDeviceMessage message; - if (IotDeviceMessageMethodEnum.isPassthrough(topicMethod)) { - message = deviceMessageService.deserializePassthroughMessage(payload, topicMethod, productKey, deviceName); - if (message == null) { - log.warn("[handleBusinessRequest][影子透传 payload 解析失败，客户端 ID: {}，主题: {}]", clientId, topic); - return; - } - } else { - message = deviceMessageService.deserializeDeviceMessage(payload, productKey, deviceName); - if (message == null) { - log.warn("[handleBusinessRequest][消息解码失败，客户端 ID: {}，主题: {}]", clientId, topic); - return; - } - // 2.2 标准化回复消息的 method（MQTT 协议中，设备回复消息的 method 会携带 _reply 后缀） - IotMqttTopicUtils.normalizeReplyMethod(message); + // 2.1 反序列化消息（标准 IotDeviceMessage 封套，method 来自 body） + IotDeviceMessage message = deviceMessageService.deserializeDeviceMessage(payload, productKey, deviceName); + if (message == null) { + log.warn("[handleBusinessRequest][消息解码失败，客户端 ID: {}，主题: {}]", clientId, topic); + return; } + // 2.2 标准化回复消息的 method（MQTT 协议中，设备回复消息的 method 会携带 _reply 后缀） + IotMqttTopicUtils.normalizeReplyMethod(message); // 3. 处理业务消息 deviceMessageService.sendDeviceMessage(message, productKey, deviceName, serverId); diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/serialize/IotMessageSerializer.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/serialize/IotMessageSerializer.java index 08713ea..095dae0 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/serialize/IotMessageSerializer.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/serialize/IotMessageSerializer.java @@ -3,8 +3,6 @@ package cn.iocoder.yudao.module.iot.gateway.serialize; import cn.iocoder.yudao.module.iot.core.enums.IotSerializeTypeEnum; import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; -import java.util.Map; - /** * IoT 设备消息序列化器接口 * @@ -37,24 +35,4 @@ public interface IotMessageSerializer { */ IotSerializeTypeEnum getType(); - /** - * 透传原始 body 编码（影子等不走 {@link IotDeviceMessage} 封套的场景） - * - * @param value 待编码对象（通常为 Map 或 DTO） - * @return 线格式字节 - */ - default byte[] encodeRaw(Object value) { - throw new UnsupportedOperationException("该序列化器不支持影子透传编码"); - } - - /** - * 透传原始 body 解码 - * - * @param bytes 线格式字节 - * @return 透传 params - */ - default Map decodeRaw(byte[] bytes) { - throw new UnsupportedOperationException("该序列化器不支持影子透传解码"); - } - } diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/serialize/json/IotJsonSerializer.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/serialize/json/IotJsonSerializer.java index 534131c..7fa6570 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/serialize/json/IotJsonSerializer.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/serialize/json/IotJsonSerializer.java @@ -6,8 +6,6 @@ import cn.iocoder.yudao.module.iot.core.enums.IotSerializeTypeEnum; import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; import cn.iocoder.yudao.module.iot.gateway.serialize.IotMessageSerializer; -import java.util.Map; - /** * JSON 格式的消息序列化器 * @@ -36,17 +34,4 @@ public class IotJsonSerializer implements IotMessageSerializer { return message; } - @Override - public byte[] encodeRaw(Object value) { - Assert.notNull(value, "待编码数据不能为空"); - return JsonUtils.toJsonByte(value); - } - - @Override - @SuppressWarnings("unchecked") - public Map decodeRaw(byte[] bytes) { - Assert.notNull(bytes, "待解码数据不能为空"); - return JsonUtils.parseObject(bytes, Map.class); - } - } diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceMessageService.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceMessageService.java index 85ce4b3..7d16a65 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceMessageService.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceMessageService.java @@ -62,16 +62,4 @@ public interface IotDeviceMessageService { void sendDeviceMessage(IotDeviceMessage message, String productKey, String deviceName, String serverId); - /** - * 透传上行组装：按设备 serializeType 解码 body，提取 id 作 requestId，构建消息 - * - * @param payload 原始 body - * @param method 透传 method（来自 topic 等，如 thing.shadow.update） - * @param productKey 产品 Key - * @param deviceName 设备名称 - * @return 设备消息；解码失败返回 null - */ - IotDeviceMessage deserializePassthroughMessage(byte[] payload, String method, - String productKey, String deviceName); - } diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceMessageServiceImpl.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceMessageServiceImpl.java index 97cf63a..5143288 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceMessageServiceImpl.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceMessageServiceImpl.java @@ -1,10 +1,8 @@ package cn.iocoder.yudao.module.iot.gateway.service.device.message; -import cn.hutool.core.convert.Convert; import cn.hutool.core.lang.Assert; import cn.hutool.core.util.StrUtil; import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO; -import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum; import cn.iocoder.yudao.module.iot.core.enums.IotSerializeTypeEnum; import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer; @@ -17,8 +15,6 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; import java.time.LocalDateTime; -import java.util.Map; -import java.util.function.Function; import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception; import static cn.iocoder.yudao.module.iot.gateway.enums.ErrorCodeConstants.DEVICE_NOT_EXISTS; @@ -44,14 +40,6 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { @Override public byte[] serializeDeviceMessage(IotDeviceMessage message, String productKey, String deviceName) { - if (IotDeviceMessageMethodEnum.isPassthrough(message.getMethod())) { - IotDeviceRespDTO device = deviceService.getDeviceFromCache(productKey, deviceName); - if (device == null) { - throw exception(DEVICE_NOT_EXISTS, productKey, deviceName); - } - IotSerializeTypeEnum serializeType = IotSerializeTypeEnum.of(device.getSerializeType()); - return encodePassthrough(message, serializeType, productKey, deviceName); - } // 1.1 获取设备信息 IotDeviceRespDTO device = deviceService.getDeviceFromCache(productKey, deviceName); if (device == null) { @@ -68,9 +56,6 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { @Override public byte[] serializeDeviceMessage(IotDeviceMessage message, IotSerializeTypeEnum serializeType) { - if (IotDeviceMessageMethodEnum.isPassthrough(message.getMethod())) { - return encodePassthrough(message, serializeType, null, null); - } // 1. 获取序列化器 IotMessageSerializer serializer = messageSerializerManager.get(serializeType); if (serializer == null) { @@ -110,23 +95,6 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { } @Override - public IotDeviceMessage deserializePassthroughMessage(byte[] payload, String method, - String productKey, String deviceName) { - IotDeviceRespDTO device = deviceService.getDeviceFromCache(productKey, deviceName); - if (device == null) { - throw exception(DEVICE_NOT_EXISTS, productKey, deviceName); - } - IotSerializeTypeEnum serializeType = IotSerializeTypeEnum.of(device.getSerializeType()); - Map params = decodePassthrough(payload, serializeType, productKey, deviceName); - if (params == null) { - // 非法/非对象 body：仍投递，由 biz 层回 JSON_FORMAT_ERROR(400) - return IotDeviceMessage.requestOf(null, method, null); - } - String requestId = Convert.toStr(params.get("id"), null); - return IotDeviceMessage.requestOf(requestId, method, params); - } - - @Override public void sendDeviceMessage(IotDeviceMessage message, String productKey, String deviceName, String serverId) { // 1. 获取设备信息 @@ -157,60 +125,4 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { } } - /** - * 影子透传编码；不支持的 serializeType 返回 null，由下行发布方跳过发送 - */ - private byte[] encodePassthrough(IotDeviceMessage message, IotSerializeTypeEnum serializeType, - String productKey, String deviceName) { - return passthroughRaw("encodePassthrough", serializeType, productKey, deviceName, false, - serializer -> serializer.encodeRaw(message.getParams())); - } - - /** - * 影子透传解码；解析失败或不支持的 serializeType 返回 null - */ - private Map decodePassthrough(byte[] bytes, IotSerializeTypeEnum serializeType, - String productKey, String deviceName) { - return passthroughRaw("decodePassthrough", serializeType, productKey, deviceName, true, - serializer -> serializer.decodeRaw(bytes)); - } - - private T passthroughRaw(String op, IotSerializeTypeEnum serializeType, - String productKey, String deviceName, boolean decode, - Function action) { - IotMessageSerializer serializer = getPassthroughSerializer(serializeType); - if (serializer == null) { - warnPassthroughUnsupported(op, serializeType, productKey, deviceName); - return null; - } - try { - return action.apply(serializer); - } catch (UnsupportedOperationException ex) { - warnPassthroughUnsupported(op, serializeType, productKey, deviceName); - return null; - } catch (RuntimeException ex) { - if (decode) { - log.warn("[{}][影子透传 payload 解析失败，productKey={}, deviceName={}]", - op, productKey, deviceName, ex); - return null; - } - throw ex; - } - } - - private IotMessageSerializer getPassthroughSerializer(IotSerializeTypeEnum serializeType) { - IotSerializeTypeEnum type = serializeType != null ? serializeType : IotSerializeTypeEnum.JSON; - return messageSerializerManager.get(type); - } - - private void warnPassthroughUnsupported(String op, IotSerializeTypeEnum serializeType, - String productKey, String deviceName) { - if (productKey != null && deviceName != null) { - log.warn("[{}][影子透传跳过：设备({}/{}) serializeType={} 不支持，目前仅 JSON 协议设备可用]", - op, productKey, deviceName, serializeType); - } else { - log.warn("[{}][影子透传跳过：serializeType={} 不支持，目前仅 JSON 协议设备可用]", op, serializeType); - } - } - } diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceShadowPassthroughTest.java b/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceShadowPassthroughTest.java deleted file mode 100644 index f7417ee..0000000 --- a/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceShadowPassthroughTest.java +++ /dev/null @@ -1,152 +0,0 @@ -package cn.iocoder.yudao.module.iot.gateway.service.device.message; - -import cn.iocoder.yudao.framework.common.util.json.JsonUtils; -import cn.iocoder.yudao.framework.test.core.ut.BaseMockitoUnitTest; -import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO; -import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum; -import cn.iocoder.yudao.module.iot.core.enums.IotSerializeTypeEnum; -import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; -import cn.iocoder.yudao.module.iot.core.topic.shadow.IotDeviceShadowReplyDTO; -import cn.iocoder.yudao.module.iot.gateway.serialize.IotMessageSerializerManager; -import cn.iocoder.yudao.module.iot.gateway.serialize.json.IotJsonSerializer; -import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; -import org.mockito.InjectMocks; -import org.mockito.Mock; -import org.mockito.Spy; - -import java.util.HashMap; -import java.util.Map; - -import static org.junit.jupiter.api.Assertions.*; -import static org.mockito.ArgumentMatchers.eq; -import static org.mockito.Mockito.when; - -/** - * 设备影子网关透传序列化单元测试 - */ -public class IotDeviceShadowPassthroughTest extends BaseMockitoUnitTest { - - @InjectMocks - private IotDeviceMessageServiceImpl deviceMessageService; - - @Mock - private IotDeviceService deviceService; - - @Spy - private IotMessageSerializerManager messageSerializerManager = new IotMessageSerializerManager(); - - @BeforeEach - public void setUpManager() { - // Spy 需手动触发构造逻辑 - if (messageSerializerManager == null) { - messageSerializerManager = new IotMessageSerializerManager(); - } - } - - @Test - public void testSerializeDeviceMessage_passthrough_flatReply() { - IotDeviceShadowReplyDTO reply = IotDeviceShadowReplyDTO.getReplyOf(new HashMap<>(), new HashMap<>(), 2L, 1469564576L); - IotDeviceMessage message = IotDeviceMessage.requestOf( - IotDeviceMessageMethodEnum.SHADOW_GET.getMethod(), reply); - - byte[] bytes = deviceMessageService.serializeDeviceMessage(message, IotSerializeTypeEnum.JSON); - - @SuppressWarnings("unchecked") - Map result = JsonUtils.parseObject(bytes, Map.class); - assertEquals("get_reply", result.get("method")); - assertEquals(0, ((Number) result.get("code")).intValue()); - assertNull(result.get("payload")); - assertNull(result.get("id")); - @SuppressWarnings("unchecked") - Map data = (Map) result.get("data"); - assertEquals(2, ((Number) data.get("version")).intValue()); - } - - @Test - public void testDeserializePassthroughMessage_fromTopicMethod() { - IotDeviceRespDTO device = new IotDeviceRespDTO().setId(1L).setSerializeType(IotSerializeTypeEnum.JSON.getType()); - when(deviceService.getDeviceFromCache(eq("pk"), eq("dn"))).thenReturn(device); - - byte[] payload = JsonUtils.toJsonByte(Map.of("method", "get", "id", "req-1")); - IotDeviceMessage message = deviceMessageService.deserializePassthroughMessage(payload, - IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), "pk", "dn"); - - assertNotNull(message); - assertEquals(IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), message.getMethod()); - assertEquals("req-1", message.getRequestId()); - assertEquals("get", ((Map) message.getParams()).get("method")); - } - - @Test - public void testSerializeUpdateReply_passthrough() { - IotDeviceShadowReplyDTO reply = IotDeviceShadowReplyDTO.updateReplyOf(3L, 1469564576L); - IotDeviceMessage message = IotDeviceMessage.requestOf( - IotDeviceMessageMethodEnum.SHADOW_GET.getMethod(), reply); - - byte[] bytes = deviceMessageService.serializeDeviceMessage(message, IotSerializeTypeEnum.JSON); - @SuppressWarnings("unchecked") - Map result = JsonUtils.parseObject(bytes, Map.class); - assertEquals("update_reply", result.get("method")); - assertEquals(0, ((Number) result.get("code")).intValue()); - assertNull(result.get("id")); - } - - @Test - public void testJsonSerializer_encodeRawDecodeRaw() { - IotJsonSerializer serializer = new IotJsonSerializer(); - Map params = Map.of("method", "get"); - byte[] bytes = serializer.encodeRaw(params); - Map decoded = serializer.decodeRaw(bytes); - assertNotNull(decoded); - assertEquals("get", decoded.get("method")); - } - - @Test - public void testSerializeDeviceMessage_passthrough_binaryReturnsNull() { - IotDeviceMessage message = IotDeviceMessage.requestOf( - IotDeviceMessageMethodEnum.SHADOW_GET.getMethod(), Map.of("method", "get_reply")); - assertNull(deviceMessageService.serializeDeviceMessage(message, IotSerializeTypeEnum.BINARY)); - } - - @Test - public void testBinarySerializer_rawNotSupported() { - cn.iocoder.yudao.module.iot.gateway.serialize.binary.IotBinarySerializer serializer = - new cn.iocoder.yudao.module.iot.gateway.serialize.binary.IotBinarySerializer(); - assertThrows(UnsupportedOperationException.class, () -> serializer.encodeRaw(Map.of("method", "update"))); - } - - @Test - public void testDeserializePassthroughMessage_numericId() { - IotDeviceRespDTO device = new IotDeviceRespDTO().setId(1L).setSerializeType(IotSerializeTypeEnum.JSON.getType()); - when(deviceService.getDeviceFromCache(eq("pk"), eq("dn"))).thenReturn(device); - - byte[] payload = JsonUtils.toJsonByte(Map.of("method", "get", "id", 123)); - IotDeviceMessage message = deviceMessageService.deserializePassthroughMessage(payload, - IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), "pk", "dn"); - - assertNotNull(message); - assertEquals("123", message.getRequestId()); - } - - @Test - public void testDeserializePassthroughMessage_invalidJsonStillDelivered() { - IotDeviceRespDTO device = new IotDeviceRespDTO().setId(1L).setSerializeType(IotSerializeTypeEnum.JSON.getType()); - when(deviceService.getDeviceFromCache(eq("pk"), eq("dn"))).thenReturn(device); - - IotDeviceMessage message = deviceMessageService.deserializePassthroughMessage("{invalid".getBytes(), - IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), "pk", "dn"); - - assertNotNull(message); - assertNull(message.getParams()); - } - - @Test - public void testErrorOf_getMethodUsesGetReply() { - IotDeviceShadowReplyDTO reply = IotDeviceShadowReplyDTO.errorOf("get", 1L, 400, "JSON 格式错误", 1000L); - assertEquals("get_reply", reply.getMethod()); - assertEquals(400, reply.getCode()); - } - -} diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/util/IotMqttTopicUtilsTest.java b/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/util/IotMqttTopicUtilsTest.java new file mode 100644 index 0000000..1c78d3f --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/util/IotMqttTopicUtilsTest.java @@ -0,0 +1,30 @@ +package cn.iocoder.yudao.module.iot.gateway.util; + +import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * {@link IotMqttTopicUtils} 单元测试 + */ +class IotMqttTopicUtilsTest { + + private static final String PRODUCT_KEY = "pk"; + private static final String DEVICE_NAME = "dn"; + + @Test + void testBuildTopicByMethod_shadowUpdateReplyAppendsReplySuffix() { + String topic = IotMqttTopicUtils.buildTopicByMethod( + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), PRODUCT_KEY, DEVICE_NAME, true); + assertEquals("/sys/pk/dn/thing/shadow/update_reply", topic); + } + + @Test + void testBuildTopicByMethod_serviceInvokeReplyAppendsReplySuffix() { + String topic = IotMqttTopicUtils.buildTopicByMethod( + IotDeviceMessageMethodEnum.SERVICE_INVOKE.getMethod(), PRODUCT_KEY, DEVICE_NAME, true); + assertEquals("/sys/pk/dn/thing/service/invoke_reply", topic); + } + +} -- libgit2 0.21.4