From 6e9b58cbcd4442af5b0a8343d6ead3e86788ddce Mon Sep 17 00:00:00 2001 From: haohao <1036606149@qq.com> Date: Sat, 20 Jun 2026 22:46:55 +0800 Subject: [PATCH] refactor(iot): 影子透传按 serializeType 序列化，reply 对齐 CommonResult --- yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImpl.java | 18 ++++++++++++------ yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImplTest.java | 24 +++++++++++++----------- yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/shadow/IotDeviceShadowErrorCodeEnum.java | 6 ++++-- yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/shadow/IotDeviceShadowReplyDTO.java | 16 ++++++++++------ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceMessageServiceImpl.java | 22 +++++++++++----------- 5 files changed, 50 insertions(+), 36 deletions(-) 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 9ea63cb..d09a88a 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 @@ -97,7 +97,8 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { private void handleDeviceShadowUpdate0(IotDeviceDO device, IotDeviceMessage message) { // 1. 解析请求（影子 body 当前按 JSON 建模，非 JSON 设备在 gateway 透传层已拒绝） - IotDeviceShadowUpdateReqDTO req = JsonUtils.convertObject(message.getParams(), IotDeviceShadowUpdateReqDTO.class); + IotDeviceShadowUpdateReqDTO req = JsonUtils.convertObject(message.getParams(), + IotDeviceShadowUpdateReqDTO.class); if (req == null) { sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf( extractInnerMethod(message.getParams()), 0L, @@ -165,7 +166,8 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { if (req.getVersion() != null && req.getVersion() == -1L) { try { resetShadowWithRetry(deviceId); - sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.updateReplyOf(0L, DateUtil.currentSeconds())); + sendShadowReplySafely(device, serverId, + IotDeviceShadowReplyDTO.updateReplyOf(0L, DateUtil.currentSeconds())); } catch (ShadowCasExhaustedException ex) { handleCasExhaustedForDevice(device, serverId, ex); } @@ -266,10 +268,12 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { } Map mergedReported = mergeMap(latest.getReported(), req.getState().getReported()); if (mergedReported.size() > MAX_PROPERTY_COUNT) { - throw new ShadowUpdateRejectedException(latest.getVersion(), IotDeviceShadowErrorCodeEnum.PROPERTY_COUNT_EXCEEDED); + throw new ShadowUpdateRejectedException(latest.getVersion(), + IotDeviceShadowErrorCodeEnum.PROPERTY_COUNT_EXCEEDED); } long now = DateUtil.currentSeconds(); - Map metadata = latest.getMetadata() != null ? new HashMap<>(latest.getMetadata()) : new HashMap<>(); + Map metadata = latest.getMetadata() != null ? new HashMap<>(latest.getMetadata()) + : new HashMap<>(); IotDeviceShadowProtocolUtils.putReportedTimestamps(metadata, req.getState().getReported().keySet(), now); return IotDeviceShadowDO.builder() .reported(mergedReported) @@ -282,7 +286,8 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { private IotDeviceShadowDO buildDesiredUpdate(IotDeviceShadowDO latest, Map desired) { long now = DateUtil.currentSeconds(); Map mergedDesired = mergeMap(latest.getDesired(), desired); - Map metadata = latest.getMetadata() != null ? new HashMap<>(latest.getMetadata()) : new HashMap<>(); + Map metadata = latest.getMetadata() != null ? new HashMap<>(latest.getMetadata()) + : new HashMap<>(); IotDeviceShadowProtocolUtils.putDesiredTimestamps(metadata, desired.keySet(), now); return IotDeviceShadowDO.builder() .desired(mergedDesired) @@ -307,7 +312,8 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { * * @return 写入成功后的 update DO（回填未更新字段，便于构建全量回复） */ - private IotDeviceShadowDO casUpdateShadow(Long deviceId, Function updateBuilder) { + private IotDeviceShadowDO casUpdateShadow(Long deviceId, + Function updateBuilder) { Long lastKnownVersion = 0L; for (int i = 0; i < OPTIMISTIC_LOCK_RETRY; i++) { // 首次尝试走缓存，重试时直读 DB 获取最新 version 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 b2a4504..71ed6f4 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 @@ -339,16 +339,17 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest { // 前两次 select 返回 v=1 的快照，之后返回 v=2 的最新快照 return selectCount.incrementAndGet() <= 2 ? staleShadow : latestShadow; }); - when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), anyLong())).thenAnswer(invocation -> { - Long expectVersion = invocation.getArgument(2); - if (expectVersion == 1L) { - return 0; - } - if (expectVersion == 2L) { - return 1; - } - return 0; - }); + when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), anyLong())) + .thenAnswer(invocation -> { + Long expectVersion = invocation.getArgument(2); + if (expectVersion == 1L) { + return 0; + } + if (expectVersion == 2L) { + return 1; + } + return 0; + }); Map reported = new HashMap<>(); reported.put("Temperature", 25); @@ -364,7 +365,8 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest { service.handleDeviceShadowUpdate(device, message); ArgumentCaptor updateCaptor = ArgumentCaptor.forClass(IotDeviceShadowDO.class); - verify(deviceShadowMapper, atLeastOnce()).updateByDeviceIdAndVersion(updateCaptor.capture(), eq(device.getId()), any()); + verify(deviceShadowMapper, atLeastOnce()).updateByDeviceIdAndVersion(updateCaptor.capture(), eq(device.getId()), + any()); IotDeviceShadowDO finalUpdate = updateCaptor.getAllValues().get(updateCaptor.getAllValues().size() - 1); // 重试时应基于 v=2 的最新 reported 合并，保留 Humidity=50 assertEquals(25, finalUpdate.getReported().get("Temperature")); 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 284e806..2a304e1 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,8 +6,10 @@ import lombok.Getter; /** * IoT 设备影子设备侧错误码枚举 *

- * 设备侧 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}）。 + * 设备侧 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 芋道源码 */ 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 5599019..b0f7f8d 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,8 +14,10 @@ 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）。 + * 扁平对齐 CommonResult + method，用于 {@link IotDeviceMessageMethodEnum#SHADOW_GET} + * 透传线 body。 + * 错误回复的 {@link #data} 仍携带 version 供设备 resync，有意偏离纯 CommonResult（error 时 data + * 通常为 null）。 * * @author 芋道源码 */ @@ -40,12 +42,13 @@ public class IotDeviceShadowReplyDTO { private String msg; /** - * 业务数据：get_reply 为 {state, version, timestamp}；update_reply 为 {version, timestamp} + * 业务数据：get_reply 为 {state, version, timestamp}；update_reply 为 {version, + * timestamp} */ private Object data; public static IotDeviceShadowReplyDTO getReplyOf(Map desired, Map reported, - Long version, Long timestamp) { + 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<>()); @@ -57,11 +60,12 @@ public class IotDeviceShadowReplyDTO { } public static IotDeviceShadowReplyDTO updateReplyOf(Long version, Long timestamp) { - return of(INNER_METHOD_UPDATE_REPLY, SUCCESS.getCode(), SUCCESS.getMsg(), dataWithVersionTimestamp(version, 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) { + String errorMsg, Long timestamp) { return of(toReplyMethod(innerMethod), errorCode, errorMsg, dataWithVersionTimestamp(version, timestamp)); } 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 86ae2d7..97cf63a 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 @@ -43,7 +43,7 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { @Override public byte[] serializeDeviceMessage(IotDeviceMessage message, - String productKey, String deviceName) { + String productKey, String deviceName) { if (IotDeviceMessageMethodEnum.isPassthrough(message.getMethod())) { IotDeviceRespDTO device = deviceService.getDeviceFromCache(productKey, deviceName); if (device == null) { @@ -67,7 +67,7 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { @Override public byte[] serializeDeviceMessage(IotDeviceMessage message, - IotSerializeTypeEnum serializeType) { + IotSerializeTypeEnum serializeType) { if (IotDeviceMessageMethodEnum.isPassthrough(message.getMethod())) { return encodePassthrough(message, serializeType, null, null); } @@ -83,7 +83,7 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { @Override public IotDeviceMessage deserializeDeviceMessage(byte[] bytes, - String productKey, String deviceName) { + String productKey, String deviceName) { // 1.1 获取设备信息 IotDeviceRespDTO device = deviceService.getDeviceFromCache(productKey, deviceName); if (device == null) { @@ -111,7 +111,7 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { @Override public IotDeviceMessage deserializePassthroughMessage(byte[] payload, String method, - String productKey, String deviceName) { + String productKey, String deviceName) { IotDeviceRespDTO device = deviceService.getDeviceFromCache(productKey, deviceName); if (device == null) { throw exception(DEVICE_NOT_EXISTS, productKey, deviceName); @@ -128,7 +128,7 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { @Override public void sendDeviceMessage(IotDeviceMessage message, - String productKey, String deviceName, String serverId) { + String productKey, String deviceName, String serverId) { // 1. 获取设备信息 IotDeviceRespDTO device = deviceService.getDeviceFromCache(productKey, deviceName); if (device == null) { @@ -148,7 +148,7 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { * @param serverId 设备连接的 serverId */ private void appendDeviceMessage(IotDeviceMessage message, - IotDeviceRespDTO device, String serverId) { + IotDeviceRespDTO device, String serverId) { message.setId(IotDeviceMessageUtils.generateMessageId()).setReportTime(LocalDateTime.now()) .setDeviceId(device.getId()).setTenantId(device.getTenantId()).setServerId(serverId); // 特殊：如果设备没有指定 requestId，则使用 messageId @@ -161,7 +161,7 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { * 影子透传编码；不支持的 serializeType 返回 null，由下行发布方跳过发送 */ private byte[] encodePassthrough(IotDeviceMessage message, IotSerializeTypeEnum serializeType, - String productKey, String deviceName) { + String productKey, String deviceName) { return passthroughRaw("encodePassthrough", serializeType, productKey, deviceName, false, serializer -> serializer.encodeRaw(message.getParams())); } @@ -170,14 +170,14 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { * 影子透传解码；解析失败或不支持的 serializeType 返回 null */ private Map decodePassthrough(byte[] bytes, IotSerializeTypeEnum serializeType, - String productKey, String deviceName) { + 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) { + String productKey, String deviceName, boolean decode, + Function action) { IotMessageSerializer serializer = getPassthroughSerializer(serializeType); if (serializer == null) { warnPassthroughUnsupported(op, serializeType, productKey, deviceName); @@ -204,7 +204,7 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { } private void warnPassthroughUnsupported(String op, IotSerializeTypeEnum serializeType, - String productKey, String deviceName) { + String productKey, String deviceName) { if (productKey != null && deviceName != null) { log.warn("[{}][影子透传跳过：设备({}/{}) serializeType={} 不支持，目前仅 JSON 协议设备可用]", op, productKey, deviceName, serializeType); -- libgit2 0.21.4