BJ_ZTEITS-Developers / parking-cloud-yudao

Commit acdeb6e5c8f032e1c6312c42020963359506ebea

Authored by haohao
1 parent 84c46ad6

refactor(iot): 移除 IoT 设备影子表及相关 SQL，重构影子协议工具类 

- 删除 `iot_device_shadow` 表的 SQL 定义及相关菜单权限插入语句。
- 重构 `IotDeviceShadowController`，使用新的 `IotDeviceShadowProtocolUtils` 处理影子状态的时间戳。
- 更新 Redis DAO 和服务实现，确保影子数据的缓存逻辑与新工具类兼容。
- 增加影子透传编解码逻辑，支持 JSON 格式的设备透传。
- 补充单元测试，确保新逻辑的正确性与稳定性。
Inline Side-by-side
Showing 18 changed files with 318 additions and 152 deletions
sql/mysql/iot_device_shadow.sql deleted
View file @84c46ad
1   --- IoT 设备影子表
2   -CREATE TABLE IF NOT EXISTS `iot_device_shadow`
3   -(
4   - `id` bigint NOT NULL AUTO_INCREMENT COMMENT '编号',
5   - `device_id` bigint NOT NULL COMMENT '设备编号',
6   - `product_id` bigint NOT NULL COMMENT '产品编号',
7   - `product_key` varchar(64) NOT NULL COMMENT '产品标识',
8   - `device_name` varchar(255) NOT NULL COMMENT '设备名称',
9   - `reported` json DEFAULT NULL COMMENT '上报状态',
10   - `desired` json DEFAULT NULL COMMENT '期望状态',
11   - `metadata` json DEFAULT NULL COMMENT '元数据',
12   - `version` bigint NOT NULL DEFAULT 0 COMMENT '版本号',
13   - `shadow_time` bigint NOT NULL DEFAULT 0 COMMENT '文档最后更新时间',
14   - `status` tinyint NOT NULL DEFAULT 0 COMMENT '状态',
15   - `creator` varchar(64) DEFAULT '' COMMENT '创建者',
16   - `create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
17   - `updater` varchar(64) DEFAULT '' COMMENT '更新者',
18   - `update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
19   - `deleted` bit(1) NOT NULL DEFAULT b'0' COMMENT '是否删除',
20   - `tenant_id` bigint NOT NULL DEFAULT 0 COMMENT '租户编号',
21   - PRIMARY KEY (`id`),
22   - UNIQUE KEY `uk_device_id` (`device_id`)
23   -) ENGINE = InnoDB
24   - DEFAULT CHARSET = utf8mb4 COMMENT ='IoT 设备影子';
25   -
26   --- 菜单权限
27   -INSERT INTO `system_menu` (`id`, `name`, `permission`, `type`, `sort`, `parent_id`, `path`, `icon`, `component`,
28   - `component_name`, `status`, `visible`, `keep_alive`, `always_show`, `creator`, `create_time`,
29   - `updater`, `update_time`, `deleted`)
30   -VALUES (4056, '设备影子查询', 'iot:device:shadow-query', 3, 13, 4008, '', '', '', '', 0, b'1', b'1', b'1', '1', NOW(),
31   - '1', NOW(), b'0');
32   -INSERT INTO `system_menu` (`id`, `name`, `permission`, `type`, `sort`, `parent_id`, `path`, `icon`, `component`,
33   - `component_name`, `status`, `visible`, `keep_alive`, `always_show`, `creator`, `create_time`,
34   - `updater`, `update_time`, `deleted`)
35   -VALUES (4057, '设备影子更新', 'iot:device:shadow-update', 3, 14, 4008, '', '', '', '', 0, b'1', b'1', b'1', '1', NOW(),
36   - '1', NOW(), b'0');
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/IotDeviceShadowController.java
  View file @acdeb6e
... ... @@ -3,11 +3,12 @@ package cn.iocoder.yudao.module.iot.controller.admin.device;
3 3 import cn.hutool.core.lang.Assert;
4 4 import cn.hutool.core.map.MapUtil;
5 5 import cn.iocoder.yudao.framework.common.pojo.CommonResult;
  6 +import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
6 7 import cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow.IotDeviceShadowPropertyRespVO;
7 8 import cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow.IotDeviceShadowRespVO;
8 9 import cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow.IotDeviceShadowUpdateDesiredReqVO;
9 10 import cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow.IotDeviceShadowUpdateStatusReqVO;
10   -import cn.iocoder.yudao.module.iot.core.enums.shadow.IotDeviceShadowMetadatas;
  11 +import cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils;
11 12 import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
12 13 import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceShadowDO;
13 14 import cn.iocoder.yudao.module.iot.dal.dataobject.thingmodel.IotThingModelDO;
... ... @@ -57,29 +58,26 @@ public class IotDeviceShadowController {
57 58 Map<String, Object> desired = shadow.getDesired() != null ? shadow.getDesired() : MapUtil.empty();
58 59 Map<String, Object> reported = shadow.getReported() != null ? shadow.getReported() : MapUtil.empty();
59 60 Map<String, Object> metadata = shadow.getMetadata() != null ? shadow.getMetadata() : MapUtil.empty();
  61 + // 拼接 VO
60 62 List<IotDeviceShadowPropertyRespVO> properties = convertList(thingModels, thingModel -> {
61 63 ThingModelProperty prop = thingModel.getProperty();
62 64 Assert.notNull(prop, "属性不能为空");
63 65 return new IotDeviceShadowPropertyRespVO()
64   - .setIdentifier(thingModel.getIdentifier())
65   - .setName(thingModel.getName())
66   - .setDataType(prop.getDataType())
67   - .setDataSpecs(prop.getDataSpecs())
68   - .setDataSpecsList(prop.getDataSpecsList())
  66 + .setIdentifier(thingModel.getIdentifier()).setName(thingModel.getName())
  67 + .setDataType(prop.getDataType()).setDataSpecs(prop.getDataSpecs()).setDataSpecsList(prop.getDataSpecsList())
69 68 .setDesiredValue(desired.get(thingModel.getIdentifier()))
70 69 .setReportedValue(reported.get(thingModel.getIdentifier()))
71   - .setDesiredTime(IotDeviceShadowMetadatas.getDesiredTimestamp(metadata, thingModel.getIdentifier()))
72   - .setReportedTime(IotDeviceShadowMetadatas.getReportedTimestamp(metadata, thingModel.getIdentifier()));
  70 + .setDesiredTime(IotDeviceShadowProtocolUtils.getDesiredTimestamp(metadata, thingModel.getIdentifier()))
  71 + .setReportedTime(IotDeviceShadowProtocolUtils.getReportedTimestamp(metadata, thingModel.getIdentifier()));
73 72 });
74 73  
75   - return success(new IotDeviceShadowRespVO()
76   - .setVersion(shadow.getVersion())
77   - .setShadowTime(shadow.getShadowTime())
78   - .setStatus(shadow.getStatus())
79   - .setProperties(properties)
80   - .setDesired(desired)
81   - .setReported(reported)
82   - .setMetadata(metadata));
  74 + IotDeviceShadowRespVO respVO = BeanUtils.toBean(shadow, IotDeviceShadowRespVO.class);
  75 + // toBean 会拷贝 DO 中同名标量；Map 字段须用下方已归一化的局部变量覆盖，避免 null 透传前端
  76 + respVO.setProperties(properties);
  77 + respVO.setDesired(desired);
  78 + respVO.setReported(reported);
  79 + respVO.setMetadata(metadata);
  80 + return success(respVO);
83 81 }
84 82  
85 83 @PutMapping("/update-desired")
... ...
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/RedisKeyConstants.java
  View file @acdeb6e
... ... @@ -21,6 +21,7 @@ public interface RedisKeyConstants {
21 21 /**
22 22 * 设备影子数据缓存，采用 STRING 结构
23 23 * <p>
  24 + * 影子缓存：DB 为准，Redis 仅读缓存（cache-aside），写后失效。
24 25 * KEY 格式：device_shadow:{deviceId}
25 26 * VALUE 数据类型：String(JSON) {@link cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceShadowDO}
26 27 */
... ...
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/device/DeviceShadowRedisDAO.java
  View file @acdeb6e
... ... @@ -10,6 +10,10 @@ import static cn.iocoder.yudao.module.iot.dal.redis.RedisKeyConstants.DEVICE_SHA
10 10  
11 11 /**
12 12 * {@link IotDeviceShadowDO} 的 Redis DAO
  13 + * <p>
  14 + * 影子缓存：DB 为准，Redis 仅作读缓存（cache-aside），写后失效。
  15 + *
  16 + * @author 芋道源码
13 17 */
14 18 @Repository
15 19 public class DeviceShadowRedisDAO {
... ...
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImpl.java
  View file @acdeb6e
... ... @@ -7,7 +7,7 @@ import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
7 7 import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
8 8 import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
9 9 import cn.iocoder.yudao.module.iot.core.enums.shadow.IotDeviceShadowErrorCodeEnum;
10   -import cn.iocoder.yudao.module.iot.core.enums.shadow.IotDeviceShadowMetadatas;
  10 +import cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils;
11 11 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
12 12 import cn.iocoder.yudao.module.iot.core.topic.shadow.IotDeviceShadowReplyDTO;
13 13 import cn.iocoder.yudao.module.iot.core.topic.shadow.IotDeviceShadowUpdateReqDTO;
... ... @@ -32,8 +32,8 @@ import java.util.Map;
32 32 import java.util.function.Function;
33 33  
34 34 import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
35   -import static cn.iocoder.yudao.module.iot.core.enums.shadow.IotDeviceShadowProtocolConstants.INNER_METHOD_GET;
36   -import static cn.iocoder.yudao.module.iot.core.enums.shadow.IotDeviceShadowProtocolConstants.INNER_METHOD_UPDATE;
  35 +import static cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils.INNER_METHOD_GET;
  36 +import static cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils.INNER_METHOD_UPDATE;
37 37 import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.DEVICE_SHADOW_NOT_ENABLED;
38 38 import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.DEVICE_SHADOW_VERSION_CONFLICT;
39 39  
... ... @@ -54,17 +54,19 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService {
54 54 private IotDeviceShadowMapper deviceShadowMapper;
55 55 @Resource
56 56 private DeviceShadowRedisDAO deviceShadowRedisDAO;
  57 +
57 58 @Resource
58 59 private IotDeviceService deviceService;
59 60 @Resource
60 61 private IotDevicePropertyService devicePropertyService;
  62 + // 解决 IotDeviceShadowServiceImpl 与 IotDeviceMessageService 循环依赖
61 63 @Resource
62 64 @Lazy
63 65 private IotDeviceMessageService deviceMessageService;
64 66  
65 67 @Override
66 68 public IotDeviceShadowDO getOrCreateShadow(IotDeviceDO device) {
67   - IotDeviceShadowDO shadow = getShadowFromCacheOrDb(device.getId());
  69 + IotDeviceShadowDO shadow = getDeviceShadowFromCacheOrDb(device.getId());
68 70 if (shadow != null) {
69 71 return shadow;
70 72 }
... ... @@ -204,7 +206,7 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService {
204 206 @Override
205 207 public void resetShadow(Long deviceId) {
206 208 deviceService.validateDeviceExists(deviceId);
207   - IotDeviceShadowDO shadow = getShadowFromCacheOrDb(deviceId);
  209 + IotDeviceShadowDO shadow = getDeviceShadowFromCacheOrDb(deviceId);
208 210 if (shadow == null) {
209 211 return;
210 212 }
... ... @@ -219,7 +221,7 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService {
219 221 @Transactional(rollbackFor = Exception.class)
220 222 public void updateStatus(Long deviceId, Integer status) {
221 223 IotDeviceDO device = deviceService.validateDeviceExists(deviceId);
222   - IotDeviceShadowDO shadow = getShadowFromCacheOrDb(deviceId);
  224 + IotDeviceShadowDO shadow = getDeviceShadowFromCacheOrDb(deviceId);
223 225 if (shadow == null) {
224 226 // 影子行不存在时直接按目标 status 懒建，避免后续 getOrCreateShadow 始终以 ENABLE 插入
225 227 IotDeviceShadowDO newShadow = buildEmptyShadow(device).status(status).build();
... ... @@ -251,7 +253,8 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService {
251 253 // ========== 内部方法 ==========
252 254  
253 255 private IotDeviceShadowDO buildReportedUpdate(IotDeviceShadowDO latest, IotDeviceShadowUpdateReqDTO req) {
254   - if (req.getVersion() != null && req.getVersion() < latest.getVersion()) {
  256 + // 乐观锁：带 version 时必须与当前完全一致（null=强制更新、-1=清空已在上层处理）
  257 + if (req.getVersion() != null && !req.getVersion().equals(latest.getVersion())) {
255 258 throw new ShadowUpdateRejectedException(latest.getVersion(), IotDeviceShadowErrorCodeEnum.VERSION_CONFLICT);
256 259 }
257 260 Map<String, Object> mergedReported = mergeMap(latest.getReported(), req.getState().getReported());
... ... @@ -260,7 +263,7 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService {
260 263 }
261 264 long now = DateUtil.currentSeconds();
262 265 Map<String, Object> metadata = latest.getMetadata() != null ? new HashMap<>(latest.getMetadata()) : new HashMap<>();
263   - IotDeviceShadowMetadatas.putReportedTimestamps(metadata, req.getState().getReported().keySet(), now);
  266 + IotDeviceShadowProtocolUtils.putReportedTimestamps(metadata, req.getState().getReported().keySet(), now);
264 267 return IotDeviceShadowDO.builder()
265 268 .reported(mergedReported)
266 269 .metadata(metadata)
... ... @@ -273,7 +276,7 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService {
273 276 long now = DateUtil.currentSeconds();
274 277 Map<String, Object> mergedDesired = mergeMap(latest.getDesired(), desired);
275 278 Map<String, Object> metadata = latest.getMetadata() != null ? new HashMap<>(latest.getMetadata()) : new HashMap<>();
276   - IotDeviceShadowMetadatas.putDesiredTimestamps(metadata, desired.keySet(), now);
  279 + IotDeviceShadowProtocolUtils.putDesiredTimestamps(metadata, desired.keySet(), now);
277 280 return IotDeviceShadowDO.builder()
278 281 .desired(mergedDesired)
279 282 .metadata(metadata)
... ... @@ -302,7 +305,7 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService {
302 305 for (int i = 0; i < OPTIMISTIC_LOCK_RETRY; i++) {
303 306 // 首次尝试走缓存，重试时直读 DB 获取最新 version
304 307 IotDeviceShadowDO latest = i == 0
305   - ? getShadowFromCacheOrDb(deviceId)
  308 + ? getDeviceShadowFromCacheOrDb(deviceId)
306 309 : deviceShadowMapper.selectByDeviceId(deviceId);
307 310 if (latest == null) {
308 311 throw ShadowCasExhaustedException.shadowNotFound();
... ... @@ -311,6 +314,7 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService {
311 314 IotDeviceShadowDO update = updateBuilder.apply(latest);
312 315 int rows = deviceShadowMapper.updateByDeviceIdAndVersion(update, deviceId, latest.getVersion());
313 316 if (rows > 0) {
  317 + // cache-aside：写后失效 Redis 读缓存
314 318 deviceShadowRedisDAO.delete(deviceId);
315 319 enrichCommittedShadow(update, latest);
316 320 return update;
... ... @@ -352,7 +356,7 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService {
352 356  
353 357 private Long resolveShadowVersionSafely(Long deviceId) {
354 358 try {
355   - IotDeviceShadowDO shadow = getShadowFromCacheOrDb(deviceId);
  359 + IotDeviceShadowDO shadow = getDeviceShadowFromCacheOrDb(deviceId);
356 360 return shadow != null ? shadow.getVersion() : 0L;
357 361 } catch (Exception ex) {
358 362 log.warn("[resolveShadowVersionSafely][device({}) 读取 version 失败，使用保底 0]", deviceId, ex);
... ... @@ -373,7 +377,10 @@ public class IotDeviceShadowServiceImpl implements IotDeviceShadowService {
373 377 .shadowTime(DateUtil.currentSeconds());
374 378 }
375 379  
376   - private IotDeviceShadowDO getShadowFromCacheOrDb(Long deviceId) {
  380 + /**
  381 + * cache-aside 读路径：Redis 命中则返回，否则查 DB 并回填缓存
  382 + */
  383 + private IotDeviceShadowDO getDeviceShadowFromCacheOrDb(Long deviceId) {
377 384 IotDeviceShadowDO shadow = deviceShadowRedisDAO.get(deviceId);
378 385 if (shadow != null) {
379 386 return shadow;
... ...
yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImplTest.java
  View file @acdeb6e
... ... @@ -54,6 +54,20 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
54 54 }
55 55  
56 56 @Test
  57 + public void testGetOrCreateShadow_whenNotExists_insertNewRow() {
  58 + // 影子不存在时懒插入；并发下 DuplicateKeyException 由 getOrCreateShadow 兜底
  59 + IotDeviceDO device = buildDevice();
  60 + when(deviceShadowRedisDAO.get(device.getId())).thenReturn(null);
  61 + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(null);
  62 +
  63 + service.getOrCreateShadow(device);
  64 +
  65 + ArgumentCaptor<IotDeviceShadowDO> captor = ArgumentCaptor.forClass(IotDeviceShadowDO.class);
  66 + verify(deviceShadowMapper).insert(captor.capture());
  67 + assertEquals(device.getId(), captor.getValue().getDeviceId());
  68 + }
  69 +
  70 + @Test
57 71 public void testHandleDeviceShadowUpdate_get_replyFullShadow() {
58 72 IotDeviceDO device = buildDevice();
59 73 IotDeviceShadowDO shadow = buildShadow(device, 2L);
... ... @@ -106,6 +120,59 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
106 120 }
107 121  
108 122 @Test
  123 + public void testHandleDeviceShadowUpdate_versionGreaterThanLatest_send409() {
  124 + IotDeviceDO device = buildDevice();
  125 + IotDeviceShadowDO shadow = buildShadow(device, 1L);
  126 + when(deviceShadowRedisDAO.get(device.getId())).thenReturn(null);
  127 + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(shadow);
  128 +
  129 + Map<String, Object> reported = new HashMap<>();
  130 + reported.put("Temperature", 25);
  131 + Map<String, Object> state = new HashMap<>();
  132 + state.put("reported", reported);
  133 + Map<String, Object> params = new HashMap<>();
  134 + params.put("method", "update");
  135 + params.put("state", state);
  136 + params.put("version", 3L);
  137 + IotDeviceMessage message = IotDeviceMessage.requestOf(device.getId(), 1L, "server-1",
  138 + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), params);
  139 +
  140 + service.handleDeviceShadowUpdate(device, message);
  141 +
  142 + ArgumentCaptor<IotDeviceMessage> captor = ArgumentCaptor.forClass(IotDeviceMessage.class);
  143 + verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1"));
  144 + IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams();
  145 + @SuppressWarnings("unchecked")
  146 + Map<String, Object> payload = (Map<String, Object>) reply.getPayload();
  147 + assertEquals(409, payload.get("code"));
  148 + verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any());
  149 + }
  150 +
  151 + @Test
  152 + public void testHandleDeviceShadowUpdate_versionNull_forceUpdateWithoutVersionCheck() {
  153 + IotDeviceDO device = buildDevice();
  154 + IotDeviceShadowDO shadow = buildShadow(device, 5L);
  155 + when(deviceShadowRedisDAO.get(device.getId())).thenReturn(null);
  156 + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(shadow);
  157 + when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(5L))).thenReturn(1);
  158 +
  159 + Map<String, Object> reported = new HashMap<>();
  160 + reported.put("Temperature", 25);
  161 + Map<String, Object> state = new HashMap<>();
  162 + state.put("reported", reported);
  163 + Map<String, Object> params = new HashMap<>();
  164 + params.put("method", "update");
  165 + params.put("state", state);
  166 + // 不传 version → 强制更新，跳过版本精确匹配
  167 + IotDeviceMessage message = IotDeviceMessage.requestOf(device.getId(), 1L, "server-1",
  168 + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), params);
  169 +
  170 + service.handleDeviceShadowUpdate(device, message);
  171 +
  172 + verify(deviceShadowMapper).updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(5L));
  173 + }
  174 +
  175 + @Test
109 176 public void testHandleDeviceShadowUpdate_versionConflict_send409() {
110 177 IotDeviceDO device = buildDevice();
111 178 IotDeviceShadowDO shadow = buildShadow(device, 5L);
... ... @@ -131,7 +198,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
131 198 assertEquals("update_reply", reply.getMethod());
132 199 @SuppressWarnings("unchecked")
133 200 Map<String, Object> payload = (Map<String, Object>) reply.getPayload();
134   - assertEquals("error", payload.get("status"));
  201 + assertEquals(409, payload.get("code"));
135 202 verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any());
136 203 }
137 204  
... ... @@ -244,9 +311,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
244 311 IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams();
245 312 @SuppressWarnings("unchecked")
246 313 Map<String, Object> payload = (Map<String, Object>) reply.getPayload();
247   - @SuppressWarnings("unchecked")
248   - Map<String, Object> content = (Map<String, Object>) payload.get("content");
249   - assertEquals(401, content.get("errorcode"));
  314 + assertEquals(401, payload.get("code"));
250 315 }
251 316  
252 317 @Test
... ... @@ -283,7 +348,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
283 348 Map<String, Object> params = new HashMap<>();
284 349 params.put("method", "update");
285 350 params.put("state", state);
286   - params.put("version", 2L);
  351 + // version=null：强制更新，CAS 重试时不受请求 version 约束
287 352 IotDeviceMessage message = IotDeviceMessage.requestOf(device.getId(), 1L, "server-1",
288 353 IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), params);
289 354  
... ... @@ -367,9 +432,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
367 432 assertEquals(0L, reply.getVersion());
368 433 @SuppressWarnings("unchecked")
369 434 Map<String, Object> payload = (Map<String, Object>) reply.getPayload();
370   - @SuppressWarnings("unchecked")
371   - Map<String, Object> content = (Map<String, Object>) payload.get("content");
372   - assertEquals(500, content.get("errorcode"));
  435 + assertEquals(500, payload.get("code"));
373 436 }
374 437  
375 438 @Test
... ... @@ -397,9 +460,7 @@ public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
397 460 IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams();
398 461 @SuppressWarnings("unchecked")
399 462 Map<String, Object> payload = (Map<String, Object>) reply.getPayload();
400   - @SuppressWarnings("unchecked")
401   - Map<String, Object> content = (Map<String, Object>) payload.get("content");
402   - assertEquals(500, content.get("errorcode"));
  463 + assertEquals(500, payload.get("code"));
403 464 verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any());
404 465 }
405 466  
... ...
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/shadow/IotDeviceShadowErrorCodeEnum.java
  View file @acdeb6e
... ... @@ -6,7 +6,8 @@ import lombok.Getter;
6 6 /**
7 7 * IoT 设备影子设备侧错误码枚举
8 8 * <p>
9   - * 用于回填 update_reply.payload.content，不走 CommonResult
  9 + * 设备侧 wire code，仅用于回填 {@code update_reply.payload.code/msg}，不走 {@link cn.iocoder.yudao.framework.common.pojo.CommonResult}。
  10 + * 服务端内部异常请使用 {@link cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants}（如 {@code DEVICE_SHADOW_NOT_ENABLED}）。
10 11 *
11 12 * @author 芋道源码
12 13 */
... ...
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/shadow/IotDeviceShadowProtocolConstants.java deleted
View file @84c46ad
1   -package cn.iocoder.yudao.module.iot.core.enums.shadow;
2   -
3   -/**
4   - * IoT 设备影子协议常量
5   - * <p>
6   - * 内层 method、state 键名等与设备侧透传 JSON 对齐
7   - *
8   - * @author 芋道源码
9   - */
10   -public final class IotDeviceShadowProtocolConstants {
11   -
12   - private IotDeviceShadowProtocolConstants() {
13   - }
14   -
15   - // ========== 内层 method ==========
16   -
17   - public static final String INNER_METHOD_GET = "get";
18   - public static final String INNER_METHOD_UPDATE = "update";
19   - public static final String INNER_METHOD_GET_REPLY = "get_reply";
20   - public static final String INNER_METHOD_UPDATE_REPLY = "update_reply";
21   -
22   - // ========== state / metadata 键 ==========
23   -
24   - public static final String STATE_DESIRED = "desired";
25   - public static final String STATE_REPORTED = "reported";
26   - public static final String METADATA_TIMESTAMP = "timestamp";
27   -
28   - // ========== payload 键 ==========
29   -
30   - public static final String PAYLOAD_STATUS = "status";
31   - public static final String PAYLOAD_STATE = "state";
32   - public static final String PAYLOAD_VERSION = "version";
33   - public static final String PAYLOAD_CONTENT = "content";
34   - public static final String PAYLOAD_ERROR_CODE = "errorcode";
35   - public static final String PAYLOAD_ERROR_MESSAGE = "errormessage";
36   -
37   - // ========== payload status 值 ==========
38   -
39   - public static final String STATUS_SUCCESS = "success";
40   - public static final String STATUS_ERROR = "error";
41   -
42   -}
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/shadow/IotDeviceShadowReplyDTO.java
  View file @acdeb6e
... ... @@ -8,7 +8,8 @@ import lombok.NoArgsConstructor;
8 8 import java.util.HashMap;
9 9 import java.util.Map;
10 10  
11   -import static cn.iocoder.yudao.module.iot.core.enums.shadow.IotDeviceShadowProtocolConstants.*;
  11 +import static cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants.SUCCESS;
  12 +import static cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils.*;
12 13  
13 14 /**
14 15 * IoT 设备影子下行 Reply DTO
... ... @@ -28,7 +29,7 @@ public class IotDeviceShadowReplyDTO {
28 29 private String method;
29 30  
30 31 /**
31   - * 回复载荷
  32 + * 回复载荷（对齐 CommonResult：code / msg / data）
32 33 */
33 34 private Object payload;
34 35  
... ... @@ -47,9 +48,9 @@ public class IotDeviceShadowReplyDTO {
47 48 Map<String, Object> state = new HashMap<>();
48 49 state.put(STATE_DESIRED, desired != null ? desired : new HashMap<>());
49 50 state.put(STATE_REPORTED, reported != null ? reported : new HashMap<>());
50   - Map<String, Object> payload = new HashMap<>();
51   - payload.put(PAYLOAD_STATUS, STATUS_SUCCESS);
52   - payload.put(PAYLOAD_STATE, state);
  51 + Map<String, Object> data = new HashMap<>();
  52 + data.put(PAYLOAD_STATE, state);
  53 + Map<String, Object> payload = buildSuccessPayload(data);
53 54 return new IotDeviceShadowReplyDTO()
54 55 .setMethod(INNER_METHOD_GET_REPLY)
55 56 .setPayload(payload)
... ... @@ -58,9 +59,9 @@ public class IotDeviceShadowReplyDTO {
58 59 }
59 60  
60 61 public static IotDeviceShadowReplyDTO updateReplyOf(Long version, Long timestamp) {
61   - Map<String, Object> payload = new HashMap<>();
62   - payload.put(PAYLOAD_STATUS, STATUS_SUCCESS);
63   - payload.put(PAYLOAD_VERSION, version);
  62 + Map<String, Object> data = new HashMap<>();
  63 + data.put(PAYLOAD_VERSION, version);
  64 + Map<String, Object> payload = buildSuccessPayload(data);
64 65 return new IotDeviceShadowReplyDTO()
65 66 .setMethod(INNER_METHOD_UPDATE_REPLY)
66 67 .setPayload(payload)
... ... @@ -68,14 +69,13 @@ public class IotDeviceShadowReplyDTO {
68 69 .setTimestamp(timestamp);
69 70 }
70 71  
71   - public static IotDeviceShadowReplyDTO errorOf(Long version, Integer errorcode, String errormessage, Long timestamp) {
72   - Map<String, Object> content = new HashMap<>();
73   - content.put(PAYLOAD_ERROR_CODE, errorcode);
74   - content.put(PAYLOAD_ERROR_MESSAGE, errormessage);
  72 + public static IotDeviceShadowReplyDTO errorOf(Long version, Integer errorCode, String errorMsg, Long timestamp) {
  73 + Map<String, Object> data = new HashMap<>();
  74 + data.put(PAYLOAD_VERSION, version);
75 75 Map<String, Object> payload = new HashMap<>();
76   - payload.put(PAYLOAD_STATUS, STATUS_ERROR);
77   - payload.put(PAYLOAD_VERSION, version);
78   - payload.put(PAYLOAD_CONTENT, content);
  76 + payload.put(PAYLOAD_CODE, errorCode);
  77 + payload.put(PAYLOAD_MSG, errorMsg);
  78 + payload.put(PAYLOAD_DATA, data);
79 79 return new IotDeviceShadowReplyDTO()
80 80 .setMethod(INNER_METHOD_UPDATE_REPLY)
81 81 .setPayload(payload)
... ... @@ -83,4 +83,12 @@ public class IotDeviceShadowReplyDTO {
83 83 .setTimestamp(timestamp);
84 84 }
85 85  
  86 + private static Map<String, Object> buildSuccessPayload(Map<String, Object> data) {
  87 + Map<String, Object> payload = new HashMap<>();
  88 + payload.put(PAYLOAD_CODE, SUCCESS.getCode());
  89 + payload.put(PAYLOAD_MSG, SUCCESS.getMsg());
  90 + payload.put(PAYLOAD_DATA, data);
  91 + return payload;
  92 + }
  93 +
86 94 }
... ...
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/shadow/IotDeviceShadowMetadatas.java renamed to yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceShadowProtocolUtils.java
  View file @acdeb6e
1   -package cn.iocoder.yudao.module.iot.core.enums.shadow;
  1 +package cn.iocoder.yudao.module.iot.core.util;
2 2  
3 3 import cn.hutool.core.map.MapUtil;
4 4  
... ... @@ -6,20 +6,39 @@ import java.util.HashMap;
6 6 import java.util.Map;
7 7 import java.util.Set;
8 8  
9   -import static cn.iocoder.yudao.module.iot.core.enums.shadow.IotDeviceShadowProtocolConstants.*;
10   -
11 9 /**
12   - * IoT 设备影子 metadata 读写工具
  10 + * IoT 设备影子协议工具
13 11 * <p>
14   - * 结构：{ desired: { identifier: { timestamp } }, reported: { ... } }
  12 + * 内层 method、state 键名、payload 字段等与设备侧透传 JSON 对齐；并提供 metadata 读写。
15 13 *
16 14 * @author 芋道源码
17 15 */
18   -public final class IotDeviceShadowMetadatas {
  16 +public final class IotDeviceShadowProtocolUtils {
19 17  
20   - private IotDeviceShadowMetadatas() {
  18 + private IotDeviceShadowProtocolUtils() {
21 19 }
22 20  
  21 + // ========== 内层 method ==========
  22 +
  23 + public static final String INNER_METHOD_GET = "get";
  24 + public static final String INNER_METHOD_UPDATE = "update";
  25 + public static final String INNER_METHOD_GET_REPLY = "get_reply";
  26 + public static final String INNER_METHOD_UPDATE_REPLY = "update_reply";
  27 +
  28 + // ========== state / metadata 键 ==========
  29 +
  30 + public static final String STATE_DESIRED = "desired";
  31 + public static final String STATE_REPORTED = "reported";
  32 + public static final String METADATA_TIMESTAMP = "timestamp";
  33 +
  34 + // ========== payload 键（对齐 CommonResult：code / msg / data）==========
  35 +
  36 + public static final String PAYLOAD_CODE = "code";
  37 + public static final String PAYLOAD_MSG = "msg";
  38 + public static final String PAYLOAD_DATA = "data";
  39 + public static final String PAYLOAD_STATE = "state";
  40 + public static final String PAYLOAD_VERSION = "version";
  41 +
23 42 /**
24 43 * 获取 desired 下某属性的更新时间戳
25 44 *
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/downstream/IotEmqxDownstreamHandler.java
  View file @acdeb6e
... ... @@ -55,6 +55,11 @@ public class IotEmqxDownstreamHandler {
55 55 // 2.2 构建载荷
56 56 byte[] payload = deviceMessageService.serializeDeviceMessage(message, deviceInfo.getProductKey(),
57 57 deviceInfo.getDeviceName());
  58 + if (payload == null) {
  59 + log.warn("[handle][影子下行跳过，设备 ID: {}，method: {}，非 JSON 协议设备不支持影子透传]",
  60 + message.getDeviceId(), message.getMethod());
  61 + return;
  62 + }
58 63  
59 64 // 3. 发布消息
60 65 protocol.publishMessage(topic, payload);
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttProtocol.java
  View file @acdeb6e
... ... @@ -17,6 +17,7 @@ import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.handler.upstream.IotMqt
17 17 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.handler.upstream.IotMqttRegisterHandler;
18 18 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.handler.upstream.IotMqttUpstreamHandler;
19 19 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.manager.IotMqttConnectionManager;
  20 +import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService;
20 21 import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
21 22 import cn.iocoder.yudao.module.iot.gateway.util.IotMqttTopicUtils;
22 23 import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
... ... @@ -101,9 +102,10 @@ public class IotMqttProtocol implements IotProtocol {
101 102 // 初始化 Handler
102 103 this.deviceMessageService = SpringUtil.getBean(IotDeviceMessageService.class);
103 104 IotDeviceCommonApi deviceApi = SpringUtil.getBean(IotDeviceCommonApi.class);
  105 + IotDeviceService deviceService = SpringUtil.getBean(IotDeviceService.class);
104 106 this.authHandler = new IotMqttAuthHandler(connectionManager, deviceMessageService, deviceApi, serverId);
105 107 this.registerHandler = new IotMqttRegisterHandler(connectionManager, deviceMessageService);
106   - this.upstreamHandler = new IotMqttUpstreamHandler(connectionManager, deviceMessageService, serverId);
  108 + this.upstreamHandler = new IotMqttUpstreamHandler(connectionManager, deviceMessageService, deviceService, serverId);
107 109 }
108 110  
109 111 @Override
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/downstream/IotMqttDownstreamHandler.java
  View file @acdeb6e
... ... @@ -45,7 +45,12 @@ public class IotMqttDownstreamHandler {
45 45 // 2.1 序列化消息
46 46 byte[] payload = deviceMessageService.serializeDeviceMessage(message, connectionInfo.getProductKey(),
47 47 connectionInfo.getDeviceName());
48   - Assert.isTrue(payload != null && payload.length > 0, "消息编码结果不能为空");
  48 + if (payload == null) {
  49 + log.warn("[handle][影子下行跳过，设备 ID: {}，method: {}，非 JSON 协议设备不支持影子透传]",
  50 + message.getDeviceId(), message.getMethod());
  51 + return;
  52 + }
  53 + Assert.isTrue(payload.length > 0, "消息编码结果不能为空");
49 54 // 2.2 构建主题
50 55 Assert.notBlank(message.getMethod(), "消息方法不能为空");
51 56 boolean isReply = IotDeviceMessageUtils.isReplyMessage(message);
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/upstream/IotMqttAbstractHandler.java
  View file @acdeb6e
... ... @@ -72,6 +72,11 @@ public abstract class IotMqttAbstractHandler {
72 72 try {
73 73 // 1.1 序列化消息（根据设备配置的序列化类型）
74 74 byte[] encodedData = deviceMessageService.serializeDeviceMessage(responseMessage, productKey, deviceName);
  75 + if (encodedData == null) {
  76 + // 影子透传在非 JSON 设备上 encode 返回 null，跳过发布避免 NPE
  77 + log.warn("[writeResponse][影子下行跳过，客户端 ID: {}，method: {}]", endpoint.clientIdentifier(), method);
  78 + return;
  79 + }
75 80 // 1.2 构建响应主题
76 81 String replyTopic = IotMqttTopicUtils.buildTopicByMethod(method, productKey, deviceName, true);
77 82  
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/upstream/IotMqttUpstreamHandler.java
  View file @acdeb6e
... ... @@ -2,10 +2,13 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.handler.upstream;
2 2  
3 3 import cn.hutool.core.lang.Assert;
4 4 import cn.hutool.core.util.ArrayUtil;
5   -import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
  5 +import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO;
6 6 import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
  7 +import cn.iocoder.yudao.module.iot.core.enums.IotSerializeTypeEnum;
7 8 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
8 9 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.manager.IotMqttConnectionManager;
  10 +import cn.iocoder.yudao.module.iot.gateway.serialize.IotShadowPassthroughCodec;
  11 +import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService;
9 12 import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
10 13 import cn.iocoder.yudao.module.iot.gateway.util.IotMqttTopicUtils;
11 14 import io.vertx.mqtt.MqttEndpoint;
... ... @@ -25,11 +28,14 @@ import java.util.stream.Collectors;
25 28 public class IotMqttUpstreamHandler extends IotMqttAbstractHandler {
26 29  
27 30 private final String serverId;
  31 + private final IotDeviceService deviceService;
28 32  
29 33 public IotMqttUpstreamHandler(IotMqttConnectionManager connectionManager,
30 34 IotDeviceMessageService deviceMessageService,
  35 + IotDeviceService deviceService,
31 36 String serverId) {
32 37 super(connectionManager, deviceMessageService);
  38 + this.deviceService = deviceService;
33 39 this.serverId = serverId;
34 40 }
35 41  
... ... @@ -69,8 +75,11 @@ public class IotMqttUpstreamHandler extends IotMqttAbstractHandler {
69 75 .collect(Collectors.joining("."));
70 76 IotDeviceMessage message;
71 77 if (IotDeviceMessageMethodEnum.isPassthrough(topicMethod)) {
72   - // 影子透传：从 topic 解析 yudao method，原始 body 作为 params
73   - Map<String, Object> params = JsonUtils.parseObject(payload, Map.class);
  78 + // 影子透传：按设备 serializeType 解码，原始 body 作为 params
  79 + IotDeviceRespDTO device = deviceService.getDeviceFromCache(productKey, deviceName);
  80 + IotSerializeTypeEnum serializeType = device != null
  81 + ? IotSerializeTypeEnum.of(device.getSerializeType()) : null;
  82 + Map<String, Object> params = IotShadowPassthroughCodec.decode(payload, serializeType);
74 83 if (params == null) {
75 84 log.warn("[handleBusinessRequest][影子透传 payload 解析失败，客户端 ID: {}，主题: {}]", clientId, topic);
76 85 return;
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/serialize/IotShadowPassthroughCodec.java 0 → 100644
  View file @acdeb6e
  1 +package cn.iocoder.yudao.module.iot.gateway.serialize;
  2 +
  3 +import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
  4 +import cn.iocoder.yudao.module.iot.core.enums.IotSerializeTypeEnum;
  5 +import lombok.extern.slf4j.Slf4j;
  6 +
  7 +import java.util.Map;
  8 +
  9 +/**
  10 + * IoT 设备影子透传编解码
  11 + * <p>
  12 + * 影子协议本质为 JSON 结构，透传编解码规则：
  13 + * <ul>
  14 + * <li>{@code serializeType == null} 或 {@link IotSerializeTypeEnum#JSON}：按 JSON 编解码（未配置时默认 JSON，兼容历史设备）</li>
  15 + * <li>其它显式类型（如 {@link IotSerializeTypeEnum#BINARY}）：warn 并丢弃，返回 {@code null}，不静默喂给 JSON 解析器</li>
  16 + * </ul>
  17 + * 下行发布方收到 {@code null} 时应跳过发送，避免 NPE；影子功能目前仅支持 JSON 协议设备。
  18 + *
  19 + * @author 芋道源码
  20 + */
  21 +@Slf4j
  22 +public final class IotShadowPassthroughCodec {
  23 +
  24 + private static final String UNSUPPORTED_MSG = "影子透传暂仅支持 JSON serializeType";
  25 +
  26 + private IotShadowPassthroughCodec() {
  27 + }
  28 +
  29 + /**
  30 + * 编码影子透传 params
  31 + *
  32 + * @param params 透传参数（Map 或 DTO）
  33 + * @param serializeType 设备序列化类型
  34 + * @return 线格式字节；非 JSON 类型返回 null
  35 + */
  36 + public static byte[] encode(Object params, IotSerializeTypeEnum serializeType) {
  37 + if (isJsonSerializeType(serializeType)) {
  38 + return JsonUtils.toJsonByte(params);
  39 + }
  40 + log.warn("[encode][{}，serializeType={}，已丢弃]", UNSUPPORTED_MSG, serializeType);
  41 + return null;
  42 + }
  43 +
  44 + /**
  45 + * 解码影子透传 payload
  46 + *
  47 + * @param bytes 线格式字节
  48 + * @param serializeType 设备序列化类型
  49 + * @return 透传 params；非 JSON 类型或解析失败返回 null
  50 + */
  51 + @SuppressWarnings("unchecked")
  52 + public static Map<String, Object> decode(byte[] bytes, IotSerializeTypeEnum serializeType) {
  53 + if (!isJsonSerializeType(serializeType)) {
  54 + log.warn("[decode][{}，serializeType={}，已丢弃]", UNSUPPORTED_MSG, serializeType);
  55 + return null;
  56 + }
  57 + return JsonUtils.parseObject(bytes, Map.class);
  58 + }
  59 +
  60 + private static boolean isJsonSerializeType(IotSerializeTypeEnum serializeType) {
  61 + // null 视为未配置，默认走 JSON（与产品/设备默认 serializeType 一致）
  62 + return serializeType == null || IotSerializeTypeEnum.JSON.equals(serializeType);
  63 + }
  64 +
  65 +}
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceMessageServiceImpl.java
  View file @acdeb6e
... ... @@ -2,7 +2,6 @@ package cn.iocoder.yudao.module.iot.gateway.service.device.message;
2 2  
3 3 import cn.hutool.core.lang.Assert;
4 4 import cn.hutool.core.util.StrUtil;
5   -import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
6 5 import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO;
7 6 import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
8 7 import cn.iocoder.yudao.module.iot.core.enums.IotSerializeTypeEnum;
... ... @@ -11,6 +10,7 @@ import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer;
11 10 import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
12 11 import cn.iocoder.yudao.module.iot.gateway.serialize.IotMessageSerializer;
13 12 import cn.iocoder.yudao.module.iot.gateway.serialize.IotMessageSerializerManager;
  13 +import cn.iocoder.yudao.module.iot.gateway.serialize.IotShadowPassthroughCodec;
14 14 import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService;
15 15 import jakarta.annotation.Resource;
16 16 import lombok.extern.slf4j.Slf4j;
... ... @@ -42,9 +42,13 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
42 42 @Override
43 43 public byte[] serializeDeviceMessage(IotDeviceMessage message,
44 44 String productKey, String deviceName) {
45   - // 影子透传：直接序列化 params 为原始线 JSON，忽略产品 serializeType
46 45 if (IotDeviceMessageMethodEnum.isPassthrough(message.getMethod())) {
47   - return JsonUtils.toJsonByte(message.getParams());
  46 + IotDeviceRespDTO device = deviceService.getDeviceFromCache(productKey, deviceName);
  47 + if (device == null) {
  48 + throw exception(DEVICE_NOT_EXISTS, productKey, deviceName);
  49 + }
  50 + IotSerializeTypeEnum serializeType = IotSerializeTypeEnum.of(device.getSerializeType());
  51 + return encodePassthrough(message, serializeType, productKey, deviceName);
48 52 }
49 53 // 1.1 获取设备信息
50 54 IotDeviceRespDTO device = deviceService.getDeviceFromCache(productKey, deviceName);
... ... @@ -62,9 +66,8 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
62 66 @Override
63 67 public byte[] serializeDeviceMessage(IotDeviceMessage message,
64 68 IotSerializeTypeEnum serializeType) {
65   - // 影子透传
66 69 if (IotDeviceMessageMethodEnum.isPassthrough(message.getMethod())) {
67   - return JsonUtils.toJsonByte(message.getParams());
  70 + return encodePassthrough(message, serializeType, null, null);
68 71 }
69 72 // 1. 获取序列化器
70 73 IotMessageSerializer serializer = messageSerializerManager.get(serializeType);
... ... @@ -135,4 +138,20 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
135 138 }
136 139 }
137 140  
  141 + /**
  142 + * 影子透传编码；非 JSON 设备返回 null，由下行发布方跳过发送
  143 + */
  144 + private byte[] encodePassthrough(IotDeviceMessage message, IotSerializeTypeEnum serializeType,
  145 + String productKey, String deviceName) {
  146 + byte[] bytes = IotShadowPassthroughCodec.encode(message.getParams(), serializeType);
  147 + if (bytes != null) {
  148 + return bytes;
  149 + }
  150 + if (productKey != null && deviceName != null) {
  151 + log.warn("[encodePassthrough][影子下行跳过：设备({}/{}) serializeType={} 不支持，目前仅 JSON 协议设备可用]",
  152 + productKey, deviceName, serializeType);
  153 + }
  154 + return null;
  155 + }
  156 +
138 157 }
... ...
yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceShadowPassthroughTest.java
  View file @acdeb6e
... ... @@ -3,8 +3,10 @@ package cn.iocoder.yudao.module.iot.gateway.service.device.message;
3 3 import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
4 4 import cn.iocoder.yudao.framework.test.core.ut.BaseMockitoUnitTest;
5 5 import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
  6 +import cn.iocoder.yudao.module.iot.core.enums.IotSerializeTypeEnum;
6 7 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
7 8 import cn.iocoder.yudao.module.iot.core.topic.shadow.IotDeviceShadowReplyDTO;
  9 +import cn.iocoder.yudao.module.iot.gateway.serialize.IotShadowPassthroughCodec;
8 10 import org.junit.jupiter.api.Test;
9 11 import org.mockito.InjectMocks;
10 12  
... ... @@ -12,6 +14,7 @@ import java.util.HashMap;
12 14 import java.util.Map;
13 15  
14 16 import static org.junit.jupiter.api.Assertions.assertEquals;
  17 +import static org.junit.jupiter.api.Assertions.assertNotNull;
15 18 import static org.junit.jupiter.api.Assertions.assertNull;
16 19  
17 20 /**
... ... @@ -31,7 +34,7 @@ public class IotDeviceShadowPassthroughTest extends BaseMockitoUnitTest {
31 34 IotDeviceMessage message = IotDeviceMessage.requestOf(
32 35 IotDeviceMessageMethodEnum.SHADOW_GET.getMethod(), payload);
33 36  
34   - byte[] bytes = deviceMessageService.serializeDeviceMessage(message, null);
  37 + byte[] bytes = deviceMessageService.serializeDeviceMessage(message, IotSerializeTypeEnum.JSON);
35 38  
36 39 @SuppressWarnings("unchecked")
37 40 Map<String, Object> result = JsonUtils.parseObject(bytes, Map.class);
... ... @@ -60,11 +63,43 @@ public class IotDeviceShadowPassthroughTest extends BaseMockitoUnitTest {
60 63 IotDeviceMessage message = IotDeviceMessage.requestOf(
61 64 IotDeviceMessageMethodEnum.SHADOW_GET.getMethod(), reply);
62 65  
63   - byte[] bytes = deviceMessageService.serializeDeviceMessage(message, null);
  66 + byte[] bytes = deviceMessageService.serializeDeviceMessage(message, IotSerializeTypeEnum.JSON);
64 67 @SuppressWarnings("unchecked")
65 68 Map<String, Object> result = JsonUtils.parseObject(bytes, Map.class);
66 69 assertEquals("update_reply", result.get("method"));
67 70 assertNull(result.get("id"));
68 71 }
69 72  
  73 + @Test
  74 + public void testPassthroughCodec_jsonSerializeType() {
  75 + Map<String, Object> params = Map.of("method", "get");
  76 + byte[] bytes = IotShadowPassthroughCodec.encode(params, IotSerializeTypeEnum.JSON);
  77 + Map<String, Object> decoded = IotShadowPassthroughCodec.decode(bytes, IotSerializeTypeEnum.JSON);
  78 + assertNotNull(decoded);
  79 + assertEquals("get", decoded.get("method"));
  80 + }
  81 +
  82 + @Test
  83 + public void testPassthroughCodec_nullSerializeTypeAsJson() {
  84 + Map<String, Object> params = Map.of("method", "get");
  85 + byte[] bytes = IotShadowPassthroughCodec.encode(params, null);
  86 + Map<String, Object> decoded = IotShadowPassthroughCodec.decode(bytes, null);
  87 + assertNotNull(decoded);
  88 + assertEquals("get", decoded.get("method"));
  89 + }
  90 +
  91 + @Test
  92 + public void testSerializeDeviceMessage_passthrough_binaryReturnsNull() {
  93 + IotDeviceMessage message = IotDeviceMessage.requestOf(
  94 + IotDeviceMessageMethodEnum.SHADOW_GET.getMethod(), Map.of("method", "get_reply"));
  95 + assertNull(deviceMessageService.serializeDeviceMessage(message, IotSerializeTypeEnum.BINARY));
  96 + }
  97 +
  98 + @Test
  99 + public void testPassthroughCodec_binaryDrop() {
  100 + Map<String, Object> params = Map.of("method", "update");
  101 + assertNull(IotShadowPassthroughCodec.encode(params, IotSerializeTypeEnum.BINARY));
  102 + assertNull(IotShadowPassthroughCodec.decode(new byte[]{0x01}, IotSerializeTypeEnum.BINARY));
  103 + }
  104 +
70 105 }
... ...