diff --git a/sql/mysql/iot_device_shadow.sql b/sql/mysql/iot_device_shadow.sql deleted file mode 100644 index 0c899b1..0000000 --- a/sql/mysql/iot_device_shadow.sql +++ /dev/null @@ -1,36 +0,0 @@ --- IoT 设备影子表 -CREATE TABLE IF NOT EXISTS `iot_device_shadow` -( - `id` bigint NOT NULL AUTO_INCREMENT COMMENT '编号', - `device_id` bigint NOT NULL COMMENT '设备编号', - `product_id` bigint NOT NULL COMMENT '产品编号', - `product_key` varchar(64) NOT NULL COMMENT '产品标识', - `device_name` varchar(255) NOT NULL COMMENT '设备名称', - `reported` json DEFAULT NULL COMMENT '上报状态', - `desired` json DEFAULT NULL COMMENT '期望状态', - `metadata` json DEFAULT NULL COMMENT '元数据', - `version` bigint NOT NULL DEFAULT 0 COMMENT '版本号', - `shadow_time` bigint NOT NULL DEFAULT 0 COMMENT '文档最后更新时间', - `status` tinyint NOT NULL DEFAULT 0 COMMENT '状态', - `creator` varchar(64) DEFAULT '' COMMENT '创建者', - `create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', - `updater` varchar(64) DEFAULT '' COMMENT '更新者', - `update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', - `deleted` bit(1) NOT NULL DEFAULT b'0' COMMENT '是否删除', - `tenant_id` bigint NOT NULL DEFAULT 0 COMMENT '租户编号', - PRIMARY KEY (`id`), - UNIQUE KEY `uk_device_id` (`device_id`) -) ENGINE = InnoDB - DEFAULT CHARSET = utf8mb4 COMMENT ='IoT 设备影子'; - --- 菜单权限 -INSERT INTO `system_menu` (`id`, `name`, `permission`, `type`, `sort`, `parent_id`, `path`, `icon`, `component`, - `component_name`, `status`, `visible`, `keep_alive`, `always_show`, `creator`, `create_time`, - `updater`, `update_time`, `deleted`) -VALUES (4056, '设备影子查询', 'iot:device:shadow-query', 3, 13, 4008, '', '', '', '', 0, b'1', b'1', b'1', '1', NOW(), - '1', NOW(), b'0'); -INSERT INTO `system_menu` (`id`, `name`, `permission`, `type`, `sort`, `parent_id`, `path`, `icon`, `component`, - `component_name`, `status`, `visible`, `keep_alive`, `always_show`, `creator`, `create_time`, - `updater`, `update_time`, `deleted`) -VALUES (4057, '设备影子更新', 'iot:device:shadow-update', 3, 14, 4008, '', '', '', '', 0, b'1', b'1', b'1', '1', NOW(), - '1', NOW(), b'0'); diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/IotDeviceShadowController.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/IotDeviceShadowController.java index cca5228..5f1ab03 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/IotDeviceShadowController.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/IotDeviceShadowController.java @@ -3,11 +3,12 @@ package cn.iocoder.yudao.module.iot.controller.admin.device; import cn.hutool.core.lang.Assert; import cn.hutool.core.map.MapUtil; import cn.iocoder.yudao.framework.common.pojo.CommonResult; +import cn.iocoder.yudao.framework.common.util.object.BeanUtils; import cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow.IotDeviceShadowPropertyRespVO; import cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow.IotDeviceShadowRespVO; import cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow.IotDeviceShadowUpdateDesiredReqVO; import cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow.IotDeviceShadowUpdateStatusReqVO; -import cn.iocoder.yudao.module.iot.core.enums.shadow.IotDeviceShadowMetadatas; +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.dataobject.thingmodel.IotThingModelDO; @@ -57,29 +58,26 @@ public class IotDeviceShadowController { Map desired = shadow.getDesired() != null ? shadow.getDesired() : MapUtil.empty(); Map reported = shadow.getReported() != null ? shadow.getReported() : MapUtil.empty(); Map metadata = shadow.getMetadata() != null ? shadow.getMetadata() : MapUtil.empty(); + // 拼接 VO List properties = convertList(thingModels, thingModel -> { ThingModelProperty prop = thingModel.getProperty(); Assert.notNull(prop, "属性不能为空"); return new IotDeviceShadowPropertyRespVO() - .setIdentifier(thingModel.getIdentifier()) - .setName(thingModel.getName()) - .setDataType(prop.getDataType()) - .setDataSpecs(prop.getDataSpecs()) - .setDataSpecsList(prop.getDataSpecsList()) + .setIdentifier(thingModel.getIdentifier()).setName(thingModel.getName()) + .setDataType(prop.getDataType()).setDataSpecs(prop.getDataSpecs()).setDataSpecsList(prop.getDataSpecsList()) .setDesiredValue(desired.get(thingModel.getIdentifier())) .setReportedValue(reported.get(thingModel.getIdentifier())) - .setDesiredTime(IotDeviceShadowMetadatas.getDesiredTimestamp(metadata, thingModel.getIdentifier())) - .setReportedTime(IotDeviceShadowMetadatas.getReportedTimestamp(metadata, thingModel.getIdentifier())); + .setDesiredTime(IotDeviceShadowProtocolUtils.getDesiredTimestamp(metadata, thingModel.getIdentifier())) + .setReportedTime(IotDeviceShadowProtocolUtils.getReportedTimestamp(metadata, thingModel.getIdentifier())); }); - return success(new IotDeviceShadowRespVO() - .setVersion(shadow.getVersion()) - .setShadowTime(shadow.getShadowTime()) - .setStatus(shadow.getStatus()) - .setProperties(properties) - .setDesired(desired) - .setReported(reported) - .setMetadata(metadata)); + IotDeviceShadowRespVO respVO = BeanUtils.toBean(shadow, IotDeviceShadowRespVO.class); + // toBean 会拷贝 DO 中同名标量；Map 字段须用下方已归一化的局部变量覆盖，避免 null 透传前端 + respVO.setProperties(properties); + respVO.setDesired(desired); + respVO.setReported(reported); + respVO.setMetadata(metadata); + return success(respVO); } @PutMapping("/update-desired") diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/RedisKeyConstants.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/RedisKeyConstants.java index 405597c..0082e9d 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/RedisKeyConstants.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/RedisKeyConstants.java @@ -21,6 +21,7 @@ public interface RedisKeyConstants { /** * 设备影子数据缓存，采用 STRING 结构 *

+ * 影子缓存：DB 为准，Redis 仅读缓存（cache-aside），写后失效。 * KEY 格式：device_shadow:{deviceId} * VALUE 数据类型：String(JSON) {@link cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceShadowDO} */ diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/device/DeviceShadowRedisDAO.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/device/DeviceShadowRedisDAO.java index 11b8fcc..d84423b 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/device/DeviceShadowRedisDAO.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/device/DeviceShadowRedisDAO.java @@ -10,6 +10,10 @@ import static cn.iocoder.yudao.module.iot.dal.redis.RedisKeyConstants.DEVICE_SHA /** * {@link IotDeviceShadowDO} 的 Redis DAO + *

+ * 影子缓存：DB 为准，Redis 仅作读缓存（cache-aside），写后失效。 + * + * @author 芋道源码 */ @Repository public class DeviceShadowRedisDAO { 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 21bb240..4a2e70d 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,7 +7,7 @@ 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.enums.shadow.IotDeviceShadowMetadatas; +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; @@ -32,8 +32,8 @@ 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.enums.shadow.IotDeviceShadowProtocolConstants.INNER_METHOD_GET; -import static cn.iocoder.yudao.module.iot.core.enums.shadow.IotDeviceShadowProtocolConstants.INNER_METHOD_UPDATE; +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; @@ -54,17 +54,19 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { private IotDeviceShadowMapper deviceShadowMapper; @Resource private DeviceShadowRedisDAO deviceShadowRedisDAO; + @Resource private IotDeviceService deviceService; @Resource private IotDevicePropertyService devicePropertyService; + // 解决 IotDeviceShadowServiceImpl 与 IotDeviceMessageService 循环依赖 @Resource @Lazy private IotDeviceMessageService deviceMessageService; @Override public IotDeviceShadowDO getOrCreateShadow(IotDeviceDO device) { - IotDeviceShadowDO shadow = getShadowFromCacheOrDb(device.getId()); + IotDeviceShadowDO shadow = getDeviceShadowFromCacheOrDb(device.getId()); if (shadow != null) { return shadow; } @@ -204,7 +206,7 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { @Override public void resetShadow(Long deviceId) { deviceService.validateDeviceExists(deviceId); - IotDeviceShadowDO shadow = getShadowFromCacheOrDb(deviceId); + IotDeviceShadowDO shadow = getDeviceShadowFromCacheOrDb(deviceId); if (shadow == null) { return; } @@ -219,7 +221,7 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { @Transactional(rollbackFor = Exception.class) public void updateStatus(Long deviceId, Integer status) { IotDeviceDO device = deviceService.validateDeviceExists(deviceId); - IotDeviceShadowDO shadow = getShadowFromCacheOrDb(deviceId); + IotDeviceShadowDO shadow = getDeviceShadowFromCacheOrDb(deviceId); if (shadow == null) { // 影子行不存在时直接按目标 status 懒建，避免后续 getOrCreateShadow 始终以 ENABLE 插入 IotDeviceShadowDO newShadow = buildEmptyShadow(device).status(status).build(); @@ -251,7 +253,8 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { // ========== 内部方法 ========== private IotDeviceShadowDO buildReportedUpdate(IotDeviceShadowDO latest, IotDeviceShadowUpdateReqDTO req) { - if (req.getVersion() != null && req.getVersion() < latest.getVersion()) { + // 乐观锁：带 version 时必须与当前完全一致（null=强制更新、-1=清空已在上层处理） + if (req.getVersion() != null && !req.getVersion().equals(latest.getVersion())) { throw new ShadowUpdateRejectedException(latest.getVersion(), IotDeviceShadowErrorCodeEnum.VERSION_CONFLICT); } Map mergedReported = mergeMap(latest.getReported(), req.getState().getReported()); @@ -260,7 +263,7 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { } long now = DateUtil.currentSeconds(); Map metadata = latest.getMetadata() != null ? new HashMap<>(latest.getMetadata()) : new HashMap<>(); - IotDeviceShadowMetadatas.putReportedTimestamps(metadata, req.getState().getReported().keySet(), now); + IotDeviceShadowProtocolUtils.putReportedTimestamps(metadata, req.getState().getReported().keySet(), now); return IotDeviceShadowDO.builder() .reported(mergedReported) .metadata(metadata) @@ -273,7 +276,7 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { long now = DateUtil.currentSeconds(); Map mergedDesired = mergeMap(latest.getDesired(), desired); Map metadata = latest.getMetadata() != null ? new HashMap<>(latest.getMetadata()) : new HashMap<>(); - IotDeviceShadowMetadatas.putDesiredTimestamps(metadata, desired.keySet(), now); + IotDeviceShadowProtocolUtils.putDesiredTimestamps(metadata, desired.keySet(), now); return IotDeviceShadowDO.builder() .desired(mergedDesired) .metadata(metadata) @@ -302,7 +305,7 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { for (int i = 0; i < OPTIMISTIC_LOCK_RETRY; i++) { // 首次尝试走缓存，重试时直读 DB 获取最新 version IotDeviceShadowDO latest = i == 0 - ? getShadowFromCacheOrDb(deviceId) + ? getDeviceShadowFromCacheOrDb(deviceId) : deviceShadowMapper.selectByDeviceId(deviceId); if (latest == null) { throw ShadowCasExhaustedException.shadowNotFound(); @@ -311,6 +314,7 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { IotDeviceShadowDO update = updateBuilder.apply(latest); int rows = deviceShadowMapper.updateByDeviceIdAndVersion(update, deviceId, latest.getVersion()); if (rows > 0) { + // cache-aside：写后失效 Redis 读缓存 deviceShadowRedisDAO.delete(deviceId); enrichCommittedShadow(update, latest); return update; @@ -352,7 +356,7 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { private Long resolveShadowVersionSafely(Long deviceId) { try { - IotDeviceShadowDO shadow = getShadowFromCacheOrDb(deviceId); + IotDeviceShadowDO shadow = getDeviceShadowFromCacheOrDb(deviceId); return shadow != null ? shadow.getVersion() : 0L; } catch (Exception ex) { log.warn("[resolveShadowVersionSafely][device({}) 读取 version 失败，使用保底 0]", deviceId, ex); @@ -373,7 +377,10 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { .shadowTime(DateUtil.currentSeconds()); } - private IotDeviceShadowDO getShadowFromCacheOrDb(Long deviceId) { + /** + * cache-aside 读路径：Redis 命中则返回，否则查 DB 并回填缓存 + */ + private IotDeviceShadowDO getDeviceShadowFromCacheOrDb(Long deviceId) { IotDeviceShadowDO shadow = deviceShadowRedisDAO.get(deviceId); if (shadow != null) { return shadow; 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 fef46b9..b26012a 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 @@ -54,6 +54,20 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest { } @Test + public void testGetOrCreateShadow_whenNotExists_insertNewRow() { + // 影子不存在时懒插入；并发下 DuplicateKeyException 由 getOrCreateShadow 兜底 + IotDeviceDO device = buildDevice(); + when(deviceShadowRedisDAO.get(device.getId())).thenReturn(null); + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(null); + + service.getOrCreateShadow(device); + + ArgumentCaptor captor = ArgumentCaptor.forClass(IotDeviceShadowDO.class); + verify(deviceShadowMapper).insert(captor.capture()); + assertEquals(device.getId(), captor.getValue().getDeviceId()); + } + + @Test public void testHandleDeviceShadowUpdate_get_replyFullShadow() { IotDeviceDO device = buildDevice(); IotDeviceShadowDO shadow = buildShadow(device, 2L); @@ -106,6 +120,59 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest { } @Test + public void testHandleDeviceShadowUpdate_versionGreaterThanLatest_send409() { + IotDeviceDO device = buildDevice(); + IotDeviceShadowDO shadow = buildShadow(device, 1L); + when(deviceShadowRedisDAO.get(device.getId())).thenReturn(null); + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(shadow); + + Map reported = new HashMap<>(); + reported.put("Temperature", 25); + Map state = new HashMap<>(); + state.put("reported", reported); + Map params = new HashMap<>(); + params.put("method", "update"); + params.put("state", state); + params.put("version", 3L); + IotDeviceMessage message = IotDeviceMessage.requestOf(device.getId(), 1L, "server-1", + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), params); + + 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(); + @SuppressWarnings("unchecked") + Map payload = (Map) reply.getPayload(); + assertEquals(409, payload.get("code")); + verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any()); + } + + @Test + public void testHandleDeviceShadowUpdate_versionNull_forceUpdateWithoutVersionCheck() { + IotDeviceDO device = buildDevice(); + IotDeviceShadowDO shadow = buildShadow(device, 5L); + when(deviceShadowRedisDAO.get(device.getId())).thenReturn(null); + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(shadow); + when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(5L))).thenReturn(1); + + Map reported = new HashMap<>(); + reported.put("Temperature", 25); + Map state = new HashMap<>(); + state.put("reported", reported); + Map params = new HashMap<>(); + params.put("method", "update"); + params.put("state", state); + // 不传 version → 强制更新，跳过版本精确匹配 + IotDeviceMessage message = IotDeviceMessage.requestOf(device.getId(), 1L, "server-1", + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), params); + + service.handleDeviceShadowUpdate(device, message); + + verify(deviceShadowMapper).updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(5L)); + } + + @Test public void testHandleDeviceShadowUpdate_versionConflict_send409() { IotDeviceDO device = buildDevice(); IotDeviceShadowDO shadow = buildShadow(device, 5L); @@ -131,7 +198,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest { assertEquals("update_reply", reply.getMethod()); @SuppressWarnings("unchecked") Map payload = (Map) reply.getPayload(); - assertEquals("error", payload.get("status")); + assertEquals(409, payload.get("code")); verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any()); } @@ -244,9 +311,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest { IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams(); @SuppressWarnings("unchecked") Map payload = (Map) reply.getPayload(); - @SuppressWarnings("unchecked") - Map content = (Map) payload.get("content"); - assertEquals(401, content.get("errorcode")); + assertEquals(401, payload.get("code")); } @Test @@ -283,7 +348,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest { Map params = new HashMap<>(); params.put("method", "update"); params.put("state", state); - params.put("version", 2L); + // version=null：强制更新，CAS 重试时不受请求 version 约束 IotDeviceMessage message = IotDeviceMessage.requestOf(device.getId(), 1L, "server-1", IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), params); @@ -367,9 +432,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest { assertEquals(0L, reply.getVersion()); @SuppressWarnings("unchecked") Map payload = (Map) reply.getPayload(); - @SuppressWarnings("unchecked") - Map content = (Map) payload.get("content"); - assertEquals(500, content.get("errorcode")); + assertEquals(500, payload.get("code")); } @Test @@ -397,9 +460,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest { IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams(); @SuppressWarnings("unchecked") Map payload = (Map) reply.getPayload(); - @SuppressWarnings("unchecked") - Map content = (Map) payload.get("content"); - assertEquals(500, content.get("errorcode")); + assertEquals(500, payload.get("code")); verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any()); } 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 6087003..89a7174 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,8 @@ import lombok.Getter; /** * IoT 设备影子设备侧错误码枚举 *

- * 用于回填 update_reply.payload.content，不走 CommonResult + * 设备侧 wire code，仅用于回填 {@code update_reply.payload.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/enums/shadow/IotDeviceShadowProtocolConstants.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/shadow/IotDeviceShadowProtocolConstants.java deleted file mode 100644 index a6414fb..0000000 --- a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/shadow/IotDeviceShadowProtocolConstants.java +++ /dev/null @@ -1,42 +0,0 @@ -package cn.iocoder.yudao.module.iot.core.enums.shadow; - -/** - * IoT 设备影子协议常量 - *

- * 内层 method、state 键名等与设备侧透传 JSON 对齐 - * - * @author 芋道源码 - */ -public final class IotDeviceShadowProtocolConstants { - - private IotDeviceShadowProtocolConstants() { - } - - // ========== 内层 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 键 ========== - - public static final String STATE_DESIRED = "desired"; - public static final String STATE_REPORTED = "reported"; - public static final String METADATA_TIMESTAMP = "timestamp"; - - // ========== payload 键 ========== - - public static final String PAYLOAD_STATUS = "status"; - public static final String PAYLOAD_STATE = "state"; - public static final String PAYLOAD_VERSION = "version"; - public static final String PAYLOAD_CONTENT = "content"; - public static final String PAYLOAD_ERROR_CODE = "errorcode"; - public static final String PAYLOAD_ERROR_MESSAGE = "errormessage"; - - // ========== payload status 值 ========== - - public static final String STATUS_SUCCESS = "success"; - public static final String STATUS_ERROR = "error"; - -} 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 9f4816d..4e8b49b 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 @@ -8,7 +8,8 @@ import lombok.NoArgsConstructor; import java.util.HashMap; import java.util.Map; -import static cn.iocoder.yudao.module.iot.core.enums.shadow.IotDeviceShadowProtocolConstants.*; +import static cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants.SUCCESS; +import static cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils.*; /** * IoT 设备影子下行 Reply DTO @@ -28,7 +29,7 @@ public class IotDeviceShadowReplyDTO { private String method; /** - * 回复载荷 + * 回复载荷（对齐 CommonResult：code / msg / data） */ private Object payload; @@ -47,9 +48,9 @@ public class IotDeviceShadowReplyDTO { Map state = new HashMap<>(); state.put(STATE_DESIRED, desired != null ? desired : new HashMap<>()); state.put(STATE_REPORTED, reported != null ? reported : new HashMap<>()); - Map payload = new HashMap<>(); - payload.put(PAYLOAD_STATUS, STATUS_SUCCESS); - payload.put(PAYLOAD_STATE, state); + Map data = new HashMap<>(); + data.put(PAYLOAD_STATE, state); + Map payload = buildSuccessPayload(data); return new IotDeviceShadowReplyDTO() .setMethod(INNER_METHOD_GET_REPLY) .setPayload(payload) @@ -58,9 +59,9 @@ public class IotDeviceShadowReplyDTO { } public static IotDeviceShadowReplyDTO updateReplyOf(Long version, Long timestamp) { - Map payload = new HashMap<>(); - payload.put(PAYLOAD_STATUS, STATUS_SUCCESS); - payload.put(PAYLOAD_VERSION, version); + Map data = new HashMap<>(); + data.put(PAYLOAD_VERSION, version); + Map payload = buildSuccessPayload(data); return new IotDeviceShadowReplyDTO() .setMethod(INNER_METHOD_UPDATE_REPLY) .setPayload(payload) @@ -68,14 +69,13 @@ public class IotDeviceShadowReplyDTO { .setTimestamp(timestamp); } - public static IotDeviceShadowReplyDTO errorOf(Long version, Integer errorcode, String errormessage, Long timestamp) { - Map content = new HashMap<>(); - content.put(PAYLOAD_ERROR_CODE, errorcode); - content.put(PAYLOAD_ERROR_MESSAGE, errormessage); + 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_STATUS, STATUS_ERROR); - payload.put(PAYLOAD_VERSION, version); - payload.put(PAYLOAD_CONTENT, content); + 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) @@ -83,4 +83,12 @@ public class IotDeviceShadowReplyDTO { .setTimestamp(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; + } + } diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/shadow/IotDeviceShadowMetadatas.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceShadowProtocolUtils.java index 4790b35..1728d16 100644 --- a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/shadow/IotDeviceShadowMetadatas.java +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceShadowProtocolUtils.java @@ -1,4 +1,4 @@ -package cn.iocoder.yudao.module.iot.core.enums.shadow; +package cn.iocoder.yudao.module.iot.core.util; import cn.hutool.core.map.MapUtil; @@ -6,20 +6,39 @@ import java.util.HashMap; import java.util.Map; import java.util.Set; -import static cn.iocoder.yudao.module.iot.core.enums.shadow.IotDeviceShadowProtocolConstants.*; - /** - * IoT 设备影子 metadata 读写工具 + * IoT 设备影子协议工具 *

- * 结构：{ desired: { identifier: { timestamp } }, reported: { ... } } + * 内层 method、state 键名、payload 字段等与设备侧透传 JSON 对齐；并提供 metadata 读写。 * * @author 芋道源码 */ -public final class IotDeviceShadowMetadatas { +public final class IotDeviceShadowProtocolUtils { - private IotDeviceShadowMetadatas() { + private IotDeviceShadowProtocolUtils() { } + // ========== 内层 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 键 ========== + + public static final String STATE_DESIRED = "desired"; + public static final String STATE_REPORTED = "reported"; + public static final String METADATA_TIMESTAMP = "timestamp"; + + // ========== payload 键（对齐 CommonResult：code / msg / 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"; + /** * 获取 desired 下某属性的更新时间戳 * 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 db5ea12..6d03d35 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,6 +55,11 @@ 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/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 354cd1b..76f0906 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,6 +17,7 @@ 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; @@ -101,9 +102,10 @@ 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, serverId); + this.upstreamHandler = new IotMqttUpstreamHandler(connectionManager, deviceMessageService, deviceService, serverId); } @Override 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 18a5a41..4f96df4 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,7 +45,12 @@ public class IotMqttDownstreamHandler { // 2.1 序列化消息 byte[] payload = deviceMessageService.serializeDeviceMessage(message, connectionInfo.getProductKey(), connectionInfo.getDeviceName()); - Assert.isTrue(payload != null && payload.length > 0, "消息编码结果不能为空"); + 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(), "消息方法不能为空"); boolean isReply = IotDeviceMessageUtils.isReplyMessage(message); 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 443dc10..4d9da08 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,6 +72,11 @@ 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 2b5ec37..8e682f8 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,10 +2,13 @@ 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.framework.common.util.json.JsonUtils; +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; @@ -25,11 +28,14 @@ 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; } @@ -69,8 +75,11 @@ public class IotMqttUpstreamHandler extends IotMqttAbstractHandler { .collect(Collectors.joining(".")); IotDeviceMessage message; if (IotDeviceMessageMethodEnum.isPassthrough(topicMethod)) { - // 影子透传：从 topic 解析 yudao method，原始 body 作为 params - Map params = JsonUtils.parseObject(payload, Map.class); + // 影子透传：按设备 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) { log.warn("[handleBusinessRequest][影子透传 payload 解析失败，客户端 ID: {}，主题: {}]", clientId, topic); return; 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 new file mode 100644 index 0000000..a8f4f86 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/serialize/IotShadowPassthroughCodec.java @@ -0,0 +1,65 @@ +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/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 8ffe1a7..176b07c 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 @@ -2,7 +2,6 @@ package cn.iocoder.yudao.module.iot.gateway.service.device.message; import cn.hutool.core.lang.Assert; import cn.hutool.core.util.StrUtil; -import cn.iocoder.yudao.framework.common.util.json.JsonUtils; 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; @@ -11,6 +10,7 @@ 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; @@ -42,9 +42,13 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { @Override public byte[] serializeDeviceMessage(IotDeviceMessage message, String productKey, String deviceName) { - // 影子透传：直接序列化 params 为原始线 JSON，忽略产品 serializeType if (IotDeviceMessageMethodEnum.isPassthrough(message.getMethod())) { - return JsonUtils.toJsonByte(message.getParams()); + 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); @@ -62,9 +66,8 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { @Override public byte[] serializeDeviceMessage(IotDeviceMessage message, IotSerializeTypeEnum serializeType) { - // 影子透传 if (IotDeviceMessageMethodEnum.isPassthrough(message.getMethod())) { - return JsonUtils.toJsonByte(message.getParams()); + return encodePassthrough(message, serializeType, null, null); } // 1. 获取序列化器 IotMessageSerializer serializer = messageSerializerManager.get(serializeType); @@ -135,4 +138,20 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { } } + /** + * 影子透传编码；非 JSON 设备返回 null，由下行发布方跳过发送 + */ + private byte[] encodePassthrough(IotDeviceMessage message, IotSerializeTypeEnum serializeType, + String productKey, String deviceName) { + byte[] bytes = IotShadowPassthroughCodec.encode(message.getParams(), serializeType); + if (bytes != null) { + return bytes; + } + if (productKey != null && deviceName != null) { + log.warn("[encodePassthrough][影子下行跳过：设备({}/{}) serializeType={} 不支持，目前仅 JSON 协议设备可用]", + productKey, deviceName, 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 bdf7a75..691a9bc 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 @@ -3,8 +3,10 @@ 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.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 org.junit.jupiter.api.Test; import org.mockito.InjectMocks; @@ -12,6 +14,7 @@ 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; /** @@ -31,7 +34,7 @@ public class IotDeviceShadowPassthroughTest extends BaseMockitoUnitTest { IotDeviceMessage message = IotDeviceMessage.requestOf( IotDeviceMessageMethodEnum.SHADOW_GET.getMethod(), payload); - byte[] bytes = deviceMessageService.serializeDeviceMessage(message, null); + byte[] bytes = deviceMessageService.serializeDeviceMessage(message, IotSerializeTypeEnum.JSON); @SuppressWarnings("unchecked") Map result = JsonUtils.parseObject(bytes, Map.class); @@ -60,11 +63,43 @@ public class IotDeviceShadowPassthroughTest extends BaseMockitoUnitTest { IotDeviceMessage message = IotDeviceMessage.requestOf( IotDeviceMessageMethodEnum.SHADOW_GET.getMethod(), reply); - byte[] bytes = deviceMessageService.serializeDeviceMessage(message, null); + byte[] bytes = deviceMessageService.serializeDeviceMessage(message, IotSerializeTypeEnum.JSON); @SuppressWarnings("unchecked") Map result = JsonUtils.parseObject(bytes, Map.class); assertEquals("update_reply", result.get("method")); 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() { + Map params = Map.of("method", "get"); + byte[] bytes = IotShadowPassthroughCodec.encode(params, null); + Map decoded = IotShadowPassthroughCodec.decode(bytes, null); + 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 testPassthroughCodec_binaryDrop() { + Map params = Map.of("method", "update"); + assertNull(IotShadowPassthroughCodec.encode(params, IotSerializeTypeEnum.BINARY)); + assertNull(IotShadowPassthroughCodec.decode(new byte[]{0x01}, IotSerializeTypeEnum.BINARY)); + } + }