refactor(iot): 影子透传按 serializeType 序列化，reply 对齐 CommonResult

haohao
1 parent 666afcc3
Showing 5 changed files with 50 additions and 36 deletions
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImpl.java
yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImplTest.java
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/shadow/IotDeviceShadowErrorCodeEnum.java
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/shadow/IotDeviceShadowReplyDTO.java
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceMessageServiceImpl.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<String, Object> 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<String, Object> metadata = latest.getMetadata() != null ? new HashMap<>(latest.getMetadata()) : new HashMap<>();
+        Map<String, Object> 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<String, Object> desired) {
         long now = DateUtil.currentSeconds();
         Map<String, Object> mergedDesired = mergeMap(latest.getDesired(), desired);
-        Map<String, Object> metadata = latest.getMetadata() != null ? new HashMap<>(latest.getMetadata()) : new HashMap<>();
+        Map<String, Object> 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<IotDeviceShadowDO, IotDeviceShadowDO> updateBuilder) {
+    private IotDeviceShadowDO casUpdateShadow(Long deviceId,
+            Function<IotDeviceShadowDO, IotDeviceShadowDO> updateBuilder) {
         Long lastKnownVersion = 0L;
         for (int i = 0; i < OPTIMISTIC_LOCK_RETRY; i++) {
             // 首次尝试走缓存，重试时直读 DB 获取最新 version
@@ -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<String, Object> reported = new HashMap<>();
         reported.put("Temperature", 25);
@@ -364,7 +365,8 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
         service.handleDeviceShadowUpdate(device, message);
         ArgumentCaptor<IotDeviceShadowDO> 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"));
@@ -6,8 +6,10 @@ import lombok.Getter;
 /**
  * IoT 设备影子设备侧错误码枚举
  * <p>
- * 设备侧 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 芋道源码
  */
@@ -14,8 +14,10 @@ import static cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils
 /**
  * IoT 设备影子下行 Reply DTO
  * <p>
- * 扁平对齐 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<String, Object> desired, Map<String, Object> reported,
-                                                     Long version, Long timestamp) {
+            Long version, Long timestamp) {
         Map<String, Object> 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));
     }
@@ -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<String, Object> 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> T passthroughRaw(String op, IotSerializeTypeEnum serializeType,
-                                 String productKey, String deviceName, boolean decode,
-                                 Function<IotMessageSerializer, T> action) {
+            String productKey, String deviceName, boolean decode,
+            Function<IotMessageSerializer, T> 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);