From 666afcc3e7a2866459eb35840e8d52445234905e Mon Sep 17 00:00:00 2001 From: haohao <1036606149@qq.com> Date: Sat, 20 Jun 2026 22:46:13 +0800 Subject: [PATCH] refactor(iot): 影子透传按设备协议序列化，reply 对齐 CommonResult --- yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImpl.java | 37 +++++++++++++++++++++++-------------- yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImplTest.java | 52 +++++++++++++++++++++++++++++++++------------------- yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/shadow/IotDeviceShadowErrorCodeEnum.java | 3 +-- yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/shadow/IotDeviceShadowReplyDTO.java | 63 ++++++++++++++++++++++++--------------------------------------- yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/shadow/IotDeviceShadowUpdateReqDTO.java | 5 ----- yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceShadowProtocolUtils.java | 26 ++++++++++++++++++++++---- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/upstream/IotEmqxUpstreamHandler.java | 27 +++++++++++++++++++++------ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttProtocol.java | 4 +--- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/upstream/IotMqttUpstreamHandler.java | 18 ++---------------- 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/IotShadowPassthroughCodec.java | 65 ----------------------------------------------------------------- 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 | 75 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++-------- yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceShadowPassthroughTest.java | 117 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++----------------------------------- 15 files changed, 325 insertions(+), 216 deletions(-) delete mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/serialize/IotShadowPassthroughCodec.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 4a2e70d..9ea63cb 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 @@ -7,10 +7,10 @@ import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum; 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.util.IotDeviceShadowProtocolUtils; 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; import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceShadowDO; import cn.iocoder.yudao.module.iot.dal.mysql.device.IotDeviceShadowMapper; @@ -32,8 +32,7 @@ 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.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.core.util.IotDeviceShadowProtocolUtils.*; 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; @@ -89,23 +88,26 @@ 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(version, + sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf( + extractInnerMethod(message.getParams()), version, IotDeviceShadowErrorCodeEnum.SERVER_ERROR.getCode(), IotDeviceShadowErrorCodeEnum.SERVER_ERROR.getMessage(), DateUtil.currentSeconds())); } } private void handleDeviceShadowUpdate0(IotDeviceDO device, IotDeviceMessage message) { - // 1. 解析请求 + // 1. 解析请求（影子 body 当前按 JSON 建模，非 JSON 设备在 gateway 透传层已拒绝） IotDeviceShadowUpdateReqDTO req = JsonUtils.convertObject(message.getParams(), IotDeviceShadowUpdateReqDTO.class); if (req == null) { - sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf(0L, + sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf( + extractInnerMethod(message.getParams()), 0L, IotDeviceShadowErrorCodeEnum.JSON_FORMAT_ERROR.getCode(), IotDeviceShadowErrorCodeEnum.JSON_FORMAT_ERROR.getMessage(), DateUtil.currentSeconds())); return; } if (StrUtil.isEmpty(req.getMethod())) { - sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf(resolveShadowVersionSafely(device.getId()), + sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf( + INNER_METHOD_UPDATE, resolveShadowVersionSafely(device.getId()), IotDeviceShadowErrorCodeEnum.METHOD_MISSING.getCode(), IotDeviceShadowErrorCodeEnum.METHOD_MISSING.getMessage(), DateUtil.currentSeconds())); return; @@ -128,7 +130,8 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { handleReportedUpdate(device, message.getServerId(), req); return; } - sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf(shadow.getVersion(), + sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf( + req.getMethod(), shadow.getVersion(), IotDeviceShadowErrorCodeEnum.INVALID_METHOD.getCode(), IotDeviceShadowErrorCodeEnum.INVALID_METHOD.getMessage(), DateUtil.currentSeconds())); } @@ -137,19 +140,22 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { Long deviceId = device.getId(); if (req.getState() == null) { - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(resolveShadowVersionSafely(deviceId), + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf( + INNER_METHOD_UPDATE, resolveShadowVersionSafely(deviceId), IotDeviceShadowErrorCodeEnum.STATE_MISSING.getCode(), IotDeviceShadowErrorCodeEnum.STATE_MISSING.getMessage(), DateUtil.currentSeconds())); return; } if (req.getState().getReported() == null) { - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(resolveShadowVersionSafely(deviceId), + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf( + INNER_METHOD_UPDATE, resolveShadowVersionSafely(deviceId), IotDeviceShadowErrorCodeEnum.REPORTED_MISSING.getCode(), IotDeviceShadowErrorCodeEnum.REPORTED_MISSING.getMessage(), DateUtil.currentSeconds())); return; } if (CollUtil.isEmpty(req.getState().getReported())) { - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(resolveShadowVersionSafely(deviceId), + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf( + INNER_METHOD_UPDATE, resolveShadowVersionSafely(deviceId), IotDeviceShadowErrorCodeEnum.REPORTED_EMPTY.getCode(), IotDeviceShadowErrorCodeEnum.REPORTED_EMPTY.getMessage(), DateUtil.currentSeconds())); return; @@ -171,7 +177,8 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.updateReplyOf(committed.getVersion(), committed.getShadowTime())); } catch (ShadowUpdateRejectedException ex) { - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(ex.getVersion(), + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf( + INNER_METHOD_UPDATE, ex.getVersion(), ex.getErrorCode().getCode(), ex.getErrorCode().getMessage(), DateUtil.currentSeconds())); } catch (ShadowCasExhaustedException ex) { handleCasExhaustedForDevice(device, serverId, ex); @@ -340,7 +347,8 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { private void handleCasExhaustedForDevice(IotDeviceDO device, String serverId, ShadowCasExhaustedException ex) { if (ex.isShadowNotFound()) { - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(0L, + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf( + INNER_METHOD_UPDATE, 0L, IotDeviceShadowErrorCodeEnum.SERVER_ERROR.getCode(), IotDeviceShadowErrorCodeEnum.SERVER_ERROR.getMessage(), DateUtil.currentSeconds())); return; @@ -349,7 +357,8 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { } private void sendVersionConflictReply(IotDeviceDO device, String serverId, Long version) { - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(version, + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf( + INNER_METHOD_UPDATE, version, IotDeviceShadowErrorCodeEnum.VERSION_CONFLICT.getCode(), IotDeviceShadowErrorCodeEnum.VERSION_CONFLICT.getMessage(), DateUtil.currentSeconds())); } 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 b26012a..b2a4504 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 @@ -85,7 +85,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest { verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1")); IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams(); assertEquals("get_reply", reply.getMethod()); - assertEquals(2L, reply.getVersion()); + assertEquals(2L, getReplyVersion(reply)); } @Test @@ -142,9 +142,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(); - @SuppressWarnings("unchecked") - Map payload = (Map) reply.getPayload(); - assertEquals(409, payload.get("code")); + assertEquals(409, reply.getCode()); verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any()); } @@ -196,9 +194,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest { verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1")); IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams(); assertEquals("update_reply", reply.getMethod()); - @SuppressWarnings("unchecked") - Map payload = (Map) reply.getPayload(); - assertEquals(409, payload.get("code")); + assertEquals(409, reply.getCode()); verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any()); } @@ -268,7 +264,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest { verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1")); IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams(); assertEquals("get_reply", reply.getMethod()); - assertEquals(2L, reply.getVersion()); + assertEquals(2L, getReplyVersion(reply)); // getOrCreateShadow + CAS 各一次 SELECT，推送不再额外查库 verify(deviceShadowMapper, times(2)).selectByDeviceId(deviceId); verify(deviceShadowRedisDAO).delete(deviceId); @@ -293,6 +289,21 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest { } @Test + public void testHandleDeviceShadowUpdate_invalidParams_send400() { + IotDeviceDO device = buildDevice(); + + IotDeviceMessage message = IotDeviceMessage.requestOf(device.getId(), 1L, "server-1", + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), null); + + 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(400, reply.getCode()); + } + + @Test public void testHandleDeviceShadowUpdate_methodMissing_send401() { IotDeviceDO device = buildDevice(); IotDeviceShadowDO shadow = buildShadow(device, 3L); @@ -309,9 +320,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(); - @SuppressWarnings("unchecked") - Map payload = (Map) reply.getPayload(); - assertEquals(401, payload.get("code")); + assertEquals(401, reply.getCode()); } @Test @@ -365,7 +374,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(); - assertEquals(3L, reply.getVersion()); + assertEquals(3L, getReplyVersion(reply)); } @Test @@ -429,10 +438,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(0L, reply.getVersion()); - @SuppressWarnings("unchecked") - Map payload = (Map) reply.getPayload(); - assertEquals(500, payload.get("code")); + assertEquals(0L, getReplyVersion(reply)); + assertEquals(500, reply.getCode()); } @Test @@ -458,9 +465,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(); - @SuppressWarnings("unchecked") - Map payload = (Map) reply.getPayload(); - assertEquals(500, payload.get("code")); + assertEquals(500, reply.getCode()); verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any()); } @@ -493,4 +498,13 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest { .build(); } + @SuppressWarnings("unchecked") + private static Long getReplyVersion(IotDeviceShadowReplyDTO reply) { + if (!(reply.getData() instanceof Map)) { + return null; + } + Object version = ((Map) reply.getData()).get("version"); + return version instanceof Number ? ((Number) version).longValue() : null; + } + } 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 89a7174..284e806 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.payload.code/msg}，不走 {@link cn.iocoder.yudao.framework.common.pojo.CommonResult}。 + * 设备侧 wire code，仅用于回填 {@code 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}）。 * * @author 芋道源码 @@ -21,7 +21,6 @@ public enum IotDeviceShadowErrorCodeEnum { REPORTED_MISSING(404, "缺少 reported"), REPORTED_EMPTY(405, "reported 为空"), INVALID_METHOD(406, "无效内层 method"), - SHADOW_EMPTY(407, "影子内容为空"), PROPERTY_COUNT_EXCEEDED(408, "影子属性数超过限制"), VERSION_CONFLICT(409, "影子版本冲突"), SERVER_ERROR(500, "服务端异常"); 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 index 4e8b49b..5599019 100644 --- 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 @@ -14,7 +14,8 @@ import static cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils /** * IoT 设备影子下行 Reply DTO *

- * 用于 {@link IotDeviceMessageMethodEnum#SHADOW_GET} 消息的 params 参数（透传原始线 body） + * 扁平对齐 CommonResult + method，用于 {@link IotDeviceMessageMethodEnum#SHADOW_GET} 透传线 body。 + * 错误回复的 {@link #data} 仍携带 version 供设备 resync，有意偏离纯 CommonResult（error 时 data 通常为 null）。 * * @author 芋道源码 */ @@ -29,19 +30,19 @@ public class IotDeviceShadowReplyDTO { private String method; /** - * 回复载荷（对齐 CommonResult：code / msg / data） + * 结果码，0 表示成功 */ - private Object payload; + private Integer code; /** - * 影子版本号 + * 结果描述 */ - private Long version; + private String msg; /** - * 时间戳（epoch 秒） + * 业务数据：get_reply 为 {state, version, timestamp}；update_reply 为 {version, timestamp} */ - private Long timestamp; + private Object data; public static IotDeviceShadowReplyDTO getReplyOf(Map desired, Map reported, Long version, Long timestamp) { @@ -50,45 +51,29 @@ public class IotDeviceShadowReplyDTO { state.put(STATE_REPORTED, reported != null ? reported : new HashMap<>()); Map data = new HashMap<>(); data.put(PAYLOAD_STATE, state); - Map payload = buildSuccessPayload(data); - return new IotDeviceShadowReplyDTO() - .setMethod(INNER_METHOD_GET_REPLY) - .setPayload(payload) - .setVersion(version) - .setTimestamp(timestamp); + 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) { - Map data = new HashMap<>(); - data.put(PAYLOAD_VERSION, version); - Map payload = buildSuccessPayload(data); - return new IotDeviceShadowReplyDTO() - .setMethod(INNER_METHOD_UPDATE_REPLY) - .setPayload(payload) - .setVersion(version) - .setTimestamp(timestamp); + return of(INNER_METHOD_UPDATE_REPLY, SUCCESS.getCode(), SUCCESS.getMsg(), dataWithVersionTimestamp(version, timestamp)); } - public static IotDeviceShadowReplyDTO errorOf(Long version, Integer errorCode, String errorMsg, Long timestamp) { - Map data = new HashMap<>(); - data.put(PAYLOAD_VERSION, version); - Map payload = new HashMap<>(); - payload.put(PAYLOAD_CODE, errorCode); - payload.put(PAYLOAD_MSG, errorMsg); - payload.put(PAYLOAD_DATA, data); - return new IotDeviceShadowReplyDTO() - .setMethod(INNER_METHOD_UPDATE_REPLY) - .setPayload(payload) - .setVersion(version) - .setTimestamp(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 Map buildSuccessPayload(Map data) { - Map payload = new HashMap<>(); - payload.put(PAYLOAD_CODE, SUCCESS.getCode()); - payload.put(PAYLOAD_MSG, SUCCESS.getMsg()); - payload.put(PAYLOAD_DATA, data); - return payload; + 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 ba49773..0464b5b 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 @@ -44,11 +44,6 @@ public class IotDeviceShadowUpdateReqDTO { */ private Map reported; - /** - * 期望状态 - */ - private Map desired; - } } 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 1728d16..e886479 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 @@ -31,15 +31,33 @@ public final class IotDeviceShadowProtocolUtils { public static final String STATE_REPORTED = "reported"; public static final String METADATA_TIMESTAMP = "timestamp"; - // ========== payload 键（对齐 CommonResult：code / msg / data）========== + // ========== reply data 键 ========== - public static final String PAYLOAD_CODE = "code"; - public static final String PAYLOAD_MSG = "msg"; - public static final String PAYLOAD_DATA = "data"; public static final String PAYLOAD_STATE = "state"; public static final String PAYLOAD_VERSION = "version"; /** + * 由内层 method 推导 reply method（get → get_reply，其余 → update_reply） + */ + public static String toReplyMethod(String innerMethod) { + if (INNER_METHOD_GET.equals(innerMethod)) { + return INNER_METHOD_GET_REPLY; + } + return INNER_METHOD_UPDATE_REPLY; + } + + /** + * 从透传 params 提取内层 method；无法解析时默认 update + */ + 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; + } + + /** * 获取 desired 下某属性的更新时间戳 * * @param metadata 元数据 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 45c9481..d691604 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,12 +3,16 @@ 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 上行消息处理器 * @@ -44,13 +48,24 @@ public class IotEmqxUpstreamHandler { } // 2.1 反序列化消息 - IotDeviceMessage message = deviceMessageService.deserializeDeviceMessage(payload, productKey, deviceName); - if (message == null) { - log.warn("[handle][topic({}) payload({}) 消息解码失败]", topic, new String(payload)); - return; + 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.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/IotMqttProtocol.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttProtocol.java index 76f0906..354cd1b 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttProtocol.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttProtocol.java @@ -17,7 +17,6 @@ import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.handler.upstream.IotMqt import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.handler.upstream.IotMqttRegisterHandler; import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.handler.upstream.IotMqttUpstreamHandler; import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.manager.IotMqttConnectionManager; -import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService; import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService; import cn.iocoder.yudao.module.iot.gateway.util.IotMqttTopicUtils; import io.netty.handler.codec.mqtt.MqttConnectReturnCode; @@ -102,10 +101,9 @@ public class IotMqttProtocol implements IotProtocol { // 初始化 Handler this.deviceMessageService = SpringUtil.getBean(IotDeviceMessageService.class); IotDeviceCommonApi deviceApi = SpringUtil.getBean(IotDeviceCommonApi.class); - IotDeviceService deviceService = SpringUtil.getBean(IotDeviceService.class); this.authHandler = new IotMqttAuthHandler(connectionManager, deviceMessageService, deviceApi, serverId); this.registerHandler = new IotMqttRegisterHandler(connectionManager, deviceMessageService); - this.upstreamHandler = new IotMqttUpstreamHandler(connectionManager, deviceMessageService, deviceService, serverId); + this.upstreamHandler = new IotMqttUpstreamHandler(connectionManager, deviceMessageService, serverId); } @Override 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 8e682f8..94ae2c9 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,20 +2,15 @@ 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.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.gateway.protocol.mqtt.manager.IotMqttConnectionManager; -import cn.iocoder.yudao.module.iot.gateway.serialize.IotShadowPassthroughCodec; -import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService; import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService; 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.Map; import java.util.stream.Collectors; @@ -28,14 +23,11 @@ import java.util.stream.Collectors; public class IotMqttUpstreamHandler extends IotMqttAbstractHandler { private final String serverId; - private final IotDeviceService deviceService; public IotMqttUpstreamHandler(IotMqttConnectionManager connectionManager, IotDeviceMessageService deviceMessageService, - IotDeviceService deviceService, String serverId) { super(connectionManager, deviceMessageService); - this.deviceService = deviceService; this.serverId = serverId; } @@ -75,17 +67,11 @@ public class IotMqttUpstreamHandler extends IotMqttAbstractHandler { .collect(Collectors.joining(".")); IotDeviceMessage message; if (IotDeviceMessageMethodEnum.isPassthrough(topicMethod)) { - // 影子透传：按设备 serializeType 解码，原始 body 作为 params - IotDeviceRespDTO device = deviceService.getDeviceFromCache(productKey, deviceName); - IotSerializeTypeEnum serializeType = device != null - ? IotSerializeTypeEnum.of(device.getSerializeType()) : null; - Map params = IotShadowPassthroughCodec.decode(payload, serializeType); - if (params == null) { + message = deviceMessageService.deserializePassthroughMessage(payload, topicMethod, productKey, deviceName); + if (message == null) { log.warn("[handleBusinessRequest][影子透传 payload 解析失败，客户端 ID: {}，主题: {}]", clientId, topic); return; } - String requestId = (String) params.get("id"); - message = IotDeviceMessage.requestOf(requestId, topicMethod, params); } else { message = deviceMessageService.deserializeDeviceMessage(payload, productKey, deviceName); if (message == null) { 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 095dae0..08713ea 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,6 +3,8 @@ 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 设备消息序列化器接口 * @@ -35,4 +37,24 @@ 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/IotShadowPassthroughCodec.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/serialize/IotShadowPassthroughCodec.java deleted file mode 100644 index a8f4f86..0000000 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/serialize/IotShadowPassthroughCodec.java +++ /dev/null @@ -1,65 +0,0 @@ -package cn.iocoder.yudao.module.iot.gateway.serialize; - -import cn.iocoder.yudao.framework.common.util.json.JsonUtils; -import cn.iocoder.yudao.module.iot.core.enums.IotSerializeTypeEnum; -import lombok.extern.slf4j.Slf4j; - -import java.util.Map; - -/** - * IoT 设备影子透传编解码 - *

- * 影子协议本质为 JSON 结构，透传编解码规则： - *

    - *
  • {@code serializeType == null} 或 {@link IotSerializeTypeEnum#JSON}：按 JSON 编解码（未配置时默认 JSON，兼容历史设备）
    • - *
  • 其它显式类型（如 {@link IotSerializeTypeEnum#BINARY}）：warn 并丢弃，返回 {@code null}，不静默喂给 JSON 解析器
    • - *
- * 下行发布方收到 {@code null} 时应跳过发送，避免 NPE；影子功能目前仅支持 JSON 协议设备。 - * - * @author 芋道源码 - */ -@Slf4j -public final class IotShadowPassthroughCodec { - - private static final String UNSUPPORTED_MSG = "影子透传暂仅支持 JSON serializeType"; - - private IotShadowPassthroughCodec() { - } - - /** - * 编码影子透传 params - * - * @param params 透传参数（Map 或 DTO） - * @param serializeType 设备序列化类型 - * @return 线格式字节；非 JSON 类型返回 null - */ - public static byte[] encode(Object params, IotSerializeTypeEnum serializeType) { - if (isJsonSerializeType(serializeType)) { - return JsonUtils.toJsonByte(params); - } - log.warn("[encode][{}，serializeType={}，已丢弃]", UNSUPPORTED_MSG, serializeType); - return null; - } - - /** - * 解码影子透传 payload - * - * @param bytes 线格式字节 - * @param serializeType 设备序列化类型 - * @return 透传 params；非 JSON 类型或解析失败返回 null - */ - @SuppressWarnings("unchecked") - public static Map decode(byte[] bytes, IotSerializeTypeEnum serializeType) { - if (!isJsonSerializeType(serializeType)) { - log.warn("[decode][{}，serializeType={}，已丢弃]", UNSUPPORTED_MSG, serializeType); - return null; - } - return JsonUtils.parseObject(bytes, Map.class); - } - - private static boolean isJsonSerializeType(IotSerializeTypeEnum serializeType) { - // null 视为未配置，默认走 JSON（与产品/设备默认 serializeType 一致） - return serializeType == null || IotSerializeTypeEnum.JSON.equals(serializeType); - } - -} 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 7fa6570..534131c 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,6 +6,8 @@ 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 格式的消息序列化器 * @@ -34,4 +36,17 @@ 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 7d16a65..85ce4b3 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,4 +62,16 @@ 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 176b07c..86ae2d7 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,5 +1,6 @@ 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; @@ -10,13 +11,14 @@ import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer; import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils; import cn.iocoder.yudao.module.iot.gateway.serialize.IotMessageSerializer; import cn.iocoder.yudao.module.iot.gateway.serialize.IotMessageSerializerManager; -import cn.iocoder.yudao.module.iot.gateway.serialize.IotShadowPassthroughCodec; import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService; import jakarta.annotation.Resource; 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; @@ -108,6 +110,23 @@ 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. 获取设备信息 @@ -139,19 +158,59 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { } /** - * 影子透传编码；非 JSON 设备返回 null，由下行发布方跳过发送 + * 影子透传编码；不支持的 serializeType 返回 null，由下行发布方跳过发送 */ private byte[] encodePassthrough(IotDeviceMessage message, IotSerializeTypeEnum serializeType, String productKey, String deviceName) { - byte[] bytes = IotShadowPassthroughCodec.encode(message.getParams(), serializeType); - if (bytes != null) { - return bytes; + 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("[encodePassthrough][影子下行跳过：设备({}/{}) serializeType={} 不支持，目前仅 JSON 协议设备可用]", - productKey, deviceName, serializeType); + log.warn("[{}][影子透传跳过：设备({}/{}) serializeType={} 不支持，目前仅 JSON 协议设备可用]", + op, productKey, deviceName, serializeType); + } else { + log.warn("[{}][影子透传跳过：serializeType={} 不支持，目前仅 JSON 协议设备可用]", op, serializeType); } - return null; } } 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 index 691a9bc..f7417ee 100644 --- 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 @@ -2,20 +2,26 @@ 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.IotShadowPassthroughCodec; +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.assertEquals; -import static org.junit.jupiter.api.Assertions.assertNotNull; -import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.when; /** * 设备影子网关透传序列化单元测试 @@ -25,35 +31,51 @@ 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_noEnvelope() { - Map payload = new HashMap<>(); - payload.put("method", "get_reply"); - payload.put("version", 2L); - payload.put("timestamp", 1469564576L); + public void testSerializeDeviceMessage_passthrough_flatReply() { + IotDeviceShadowReplyDTO reply = IotDeviceShadowReplyDTO.getReplyOf(new HashMap<>(), new HashMap<>(), 2L, 1469564576L); IotDeviceMessage message = IotDeviceMessage.requestOf( - IotDeviceMessageMethodEnum.SHADOW_GET.getMethod(), payload); + 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(2, ((Number) result.get("version")).intValue()); + assertEquals(0, ((Number) result.get("code")).intValue()); + assertNull(result.get("payload")); assertNull(result.get("id")); - assertNull(result.get("code")); - assertNull(result.get("requestId")); + @SuppressWarnings("unchecked") + Map data = (Map) result.get("data"); + assertEquals(2, ((Number) data.get("version")).intValue()); } @Test - public void testDeserializeUpstreamShadow_fromTopicMethod() { - // 模拟上行透传：原始 body 不含 thing.shadow.update 封套 method - Map body = new HashMap<>(); - body.put("method", "get"); - IotDeviceMessage message = IotDeviceMessage.requestOf("req-1", - IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), body); + 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")); } @@ -67,23 +89,16 @@ public class IotDeviceShadowPassthroughTest extends BaseMockitoUnitTest { @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 testPassthroughCodec_jsonSerializeType() { - Map params = Map.of("method", "get"); - byte[] bytes = IotShadowPassthroughCodec.encode(params, IotSerializeTypeEnum.JSON); - Map decoded = IotShadowPassthroughCodec.decode(bytes, IotSerializeTypeEnum.JSON); - assertNotNull(decoded); - assertEquals("get", decoded.get("method")); - } - - @Test - public void testPassthroughCodec_nullSerializeTypeAsJson() { + public void testJsonSerializer_encodeRawDecodeRaw() { + IotJsonSerializer serializer = new IotJsonSerializer(); Map params = Map.of("method", "get"); - byte[] bytes = IotShadowPassthroughCodec.encode(params, null); - Map decoded = IotShadowPassthroughCodec.decode(bytes, null); + byte[] bytes = serializer.encodeRaw(params); + Map decoded = serializer.decodeRaw(bytes); assertNotNull(decoded); assertEquals("get", decoded.get("method")); } @@ -96,10 +111,42 @@ public class IotDeviceShadowPassthroughTest extends BaseMockitoUnitTest { } @Test - public void testPassthroughCodec_binaryDrop() { - Map params = Map.of("method", "update"); - assertNull(IotShadowPassthroughCodec.encode(params, IotSerializeTypeEnum.BINARY)); - assertNull(IotShadowPassthroughCodec.decode(new byte[]{0x01}, IotSerializeTypeEnum.BINARY)); + 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()); } } -- libgit2 0.21.4