BJ_ZTEITS-Developers / parking-cloud-yudao

Commit 84c46ad6df2021c2564084d67ab2078191db7193

Authored by haohao
1 parent d97b85bb

feat(iot): 新增设备影子能力，支持 MQTT 透传与后台管理 

实现 desired/reported 状态存储与同步，提供查询、期望下发、开关控制与重置接口；
网关支持 thing.shadow.update/get 透传协议，并补充单元测试与菜单权限 SQL。
Inline Side-by-side
Showing 28 changed files with 1864 additions and 10 deletions
sql/mysql/iot_device_shadow.sql 0 → 100644
  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 0 → 100644
  View file @84c46ad
  1 +package cn.iocoder.yudao.module.iot.controller.admin.device;
  2 +
  3 +import cn.hutool.core.lang.Assert;
  4 +import cn.hutool.core.map.MapUtil;
  5 +import cn.iocoder.yudao.framework.common.pojo.CommonResult;
  6 +import cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow.IotDeviceShadowPropertyRespVO;
  7 +import cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow.IotDeviceShadowRespVO;
  8 +import cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow.IotDeviceShadowUpdateDesiredReqVO;
  9 +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.dal.dataobject.device.IotDeviceDO;
  12 +import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceShadowDO;
  13 +import cn.iocoder.yudao.module.iot.dal.dataobject.thingmodel.IotThingModelDO;
  14 +import cn.iocoder.yudao.module.iot.dal.dataobject.thingmodel.model.ThingModelProperty;
  15 +import cn.iocoder.yudao.module.iot.enums.thingmodel.IotThingModelTypeEnum;
  16 +import cn.iocoder.yudao.module.iot.service.device.IotDeviceService;
  17 +import cn.iocoder.yudao.module.iot.service.device.shadow.IotDeviceShadowService;
  18 +import cn.iocoder.yudao.module.iot.service.thingmodel.IotThingModelService;
  19 +import io.swagger.v3.oas.annotations.Operation;
  20 +import io.swagger.v3.oas.annotations.Parameter;
  21 +import io.swagger.v3.oas.annotations.tags.Tag;
  22 +import jakarta.annotation.Resource;
  23 +import jakarta.validation.Valid;
  24 +import org.springframework.security.access.prepost.PreAuthorize;
  25 +import org.springframework.validation.annotation.Validated;
  26 +import org.springframework.web.bind.annotation.*;
  27 +
  28 +import java.util.List;
  29 +import java.util.Map;
  30 +
  31 +import static cn.iocoder.yudao.framework.common.pojo.CommonResult.success;
  32 +import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertList;
  33 +
  34 +@Tag(name = "管理后台 - IoT 设备影子")
  35 +@RestController
  36 +@RequestMapping("/iot/device/shadow")
  37 +@Validated
  38 +public class IotDeviceShadowController {
  39 +
  40 + @Resource
  41 + private IotDeviceShadowService deviceShadowService;
  42 + @Resource
  43 + private IotDeviceService deviceService;
  44 + @Resource
  45 + private IotThingModelService thingModelService;
  46 +
  47 + @GetMapping("/get")
  48 + @Operation(summary = "获取设备影子")
  49 + @Parameter(name = "deviceId", description = "设备编号", required = true)
  50 + @PreAuthorize("@ss.hasPermission('iot:device:shadow-query')")
  51 + public CommonResult<IotDeviceShadowRespVO> getDeviceShadow(@RequestParam("deviceId") Long deviceId) {
  52 + IotDeviceDO device = deviceService.validateDeviceExists(deviceId);
  53 + IotDeviceShadowDO shadow = deviceShadowService.getOrCreateShadow(device);
  54 +
  55 + List<IotThingModelDO> thingModels = thingModelService.getThingModelListByProductIdAndType(
  56 + device.getProductId(), IotThingModelTypeEnum.PROPERTY.getType());
  57 + Map<String, Object> desired = shadow.getDesired() != null ? shadow.getDesired() : MapUtil.empty();
  58 + Map<String, Object> reported = shadow.getReported() != null ? shadow.getReported() : MapUtil.empty();
  59 + Map<String, Object> metadata = shadow.getMetadata() != null ? shadow.getMetadata() : MapUtil.empty();
  60 + List<IotDeviceShadowPropertyRespVO> properties = convertList(thingModels, thingModel -> {
  61 + ThingModelProperty prop = thingModel.getProperty();
  62 + Assert.notNull(prop, "属性不能为空");
  63 + return new IotDeviceShadowPropertyRespVO()
  64 + .setIdentifier(thingModel.getIdentifier())
  65 + .setName(thingModel.getName())
  66 + .setDataType(prop.getDataType())
  67 + .setDataSpecs(prop.getDataSpecs())
  68 + .setDataSpecsList(prop.getDataSpecsList())
  69 + .setDesiredValue(desired.get(thingModel.getIdentifier()))
  70 + .setReportedValue(reported.get(thingModel.getIdentifier()))
  71 + .setDesiredTime(IotDeviceShadowMetadatas.getDesiredTimestamp(metadata, thingModel.getIdentifier()))
  72 + .setReportedTime(IotDeviceShadowMetadatas.getReportedTimestamp(metadata, thingModel.getIdentifier()));
  73 + });
  74 +
  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));
  83 + }
  84 +
  85 + @PutMapping("/update-desired")
  86 + @Operation(summary = "更新设备影子期望状态")
  87 + @PreAuthorize("@ss.hasPermission('iot:device:shadow-update')")
  88 + public CommonResult<Boolean> updateDesiredState(@Valid @RequestBody IotDeviceShadowUpdateDesiredReqVO reqVO) {
  89 + deviceShadowService.updateDesiredState(reqVO.getDeviceId(), reqVO.getDesired());
  90 + return success(true);
  91 + }
  92 +
  93 + @PutMapping("/update-status")
  94 + @Operation(summary = "更新设备影子开关状态")
  95 + @PreAuthorize("@ss.hasPermission('iot:device:shadow-update')")
  96 + public CommonResult<Boolean> updateShadowStatus(@Valid @RequestBody IotDeviceShadowUpdateStatusReqVO reqVO) {
  97 + deviceShadowService.updateStatus(reqVO.getDeviceId(), reqVO.getStatus());
  98 + return success(true);
  99 + }
  100 +
  101 + @DeleteMapping("/reset")
  102 + @Operation(summary = "重置设备影子")
  103 + @Parameter(name = "deviceId", description = "设备编号", required = true)
  104 + @PreAuthorize("@ss.hasPermission('iot:device:shadow-update')")
  105 + public CommonResult<Boolean> resetShadow(@RequestParam("deviceId") Long deviceId) {
  106 + deviceShadowService.resetShadow(deviceId);
  107 + return success(true);
  108 + }
  109 +
  110 +}
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/vo/shadow/IotDeviceShadowPropertyRespVO.java 0 → 100644
  View file @84c46ad
  1 +package cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow;
  2 +
  3 +import cn.iocoder.yudao.module.iot.dal.dataobject.thingmodel.model.dataType.ThingModelDataSpecs;
  4 +import io.swagger.v3.oas.annotations.media.Schema;
  5 +import lombok.Data;
  6 +
  7 +import java.util.List;
  8 +
  9 +@Schema(description = "管理后台 - IoT 设备影子属性对比 Response VO")
  10 +@Data
  11 +public class IotDeviceShadowPropertyRespVO {
  12 +
  13 + @Schema(description = "标识符", example = "Temperature")
  14 + private String identifier;
  15 +
  16 + @Schema(description = "属性名称", example = "温度")
  17 + private String name;
  18 +
  19 + @Schema(description = "数据类型", example = "int")
  20 + private String dataType;
  21 +
  22 + @Schema(description = "数据定义")
  23 + private ThingModelDataSpecs dataSpecs;
  24 +
  25 + @Schema(description = "数据定义列表")
  26 + private List<ThingModelDataSpecs> dataSpecsList;
  27 +
  28 + @Schema(description = "期望值")
  29 + private Object desiredValue;
  30 +
  31 + @Schema(description = "上报值")
  32 + private Object reportedValue;
  33 +
  34 + @Schema(description = "期望更新时间（epoch 秒）")
  35 + private Long desiredTime;
  36 +
  37 + @Schema(description = "上报更新时间（epoch 秒）")
  38 + private Long reportedTime;
  39 +
  40 +}
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/vo/shadow/IotDeviceShadowRespVO.java 0 → 100644
  View file @84c46ad
  1 +package cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow;
  2 +
  3 +import io.swagger.v3.oas.annotations.media.Schema;
  4 +import lombok.Data;
  5 +
  6 +import java.util.List;
  7 +import java.util.Map;
  8 +
  9 +@Schema(description = "管理后台 - IoT 设备影子 Response VO")
  10 +@Data
  11 +public class IotDeviceShadowRespVO {
  12 +
  13 + @Schema(description = "版本号", example = "3")
  14 + private Long version;
  15 +
  16 + @Schema(description = "最后更新时间（epoch 秒）", example = "1469564576")
  17 + private Long shadowTime;
  18 +
  19 + @Schema(description = "是否启用", example = "0")
  20 + private Integer status;
  21 +
  22 + @Schema(description = "属性对比列表")
  23 + private List<IotDeviceShadowPropertyRespVO> properties;
  24 +
  25 + @Schema(description = "期望状态")
  26 + private Map<String, Object> desired;
  27 +
  28 + @Schema(description = "上报状态")
  29 + private Map<String, Object> reported;
  30 +
  31 + @Schema(description = "元数据")
  32 + private Map<String, Object> metadata;
  33 +
  34 +}
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/vo/shadow/IotDeviceShadowUpdateDesiredReqVO.java 0 → 100644
  View file @84c46ad
  1 +package cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow;
  2 +
  3 +import io.swagger.v3.oas.annotations.media.Schema;
  4 +import jakarta.validation.constraints.NotNull;
  5 +import lombok.Data;
  6 +
  7 +import java.util.Map;
  8 +
  9 +@Schema(description = "管理后台 - IoT 设备影子更新期望状态 Request VO")
  10 +@Data
  11 +public class IotDeviceShadowUpdateDesiredReqVO {
  12 +
  13 + @Schema(description = "设备编号", requiredMode = Schema.RequiredMode.REQUIRED, example = "1")
  14 + @NotNull(message = "设备编号不能为空")
  15 + private Long deviceId;
  16 +
  17 + @Schema(description = "期望状态", requiredMode = Schema.RequiredMode.REQUIRED)
  18 + @NotNull(message = "期望状态不能为空")
  19 + private Map<String, Object> desired;
  20 +
  21 +}
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/vo/shadow/IotDeviceShadowUpdateStatusReqVO.java 0 → 100644
  View file @84c46ad
  1 +package cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow;
  2 +
  3 +import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
  4 +import cn.iocoder.yudao.framework.common.validation.InEnum;
  5 +import io.swagger.v3.oas.annotations.media.Schema;
  6 +import jakarta.validation.constraints.NotNull;
  7 +import lombok.Data;
  8 +
  9 +@Schema(description = "管理后台 - IoT 设备影子更新状态 Request VO")
  10 +@Data
  11 +public class IotDeviceShadowUpdateStatusReqVO {
  12 +
  13 + @Schema(description = "设备编号", requiredMode = Schema.RequiredMode.REQUIRED, example = "1")
  14 + @NotNull(message = "设备编号不能为空")
  15 + private Long deviceId;
  16 +
  17 + @Schema(description = "状态", requiredMode = Schema.RequiredMode.REQUIRED, example = "0")
  18 + @NotNull(message = "状态不能为空")
  19 + @InEnum(CommonStatusEnum.class)
  20 + private Integer status;
  21 +
  22 +}
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/device/IotDeviceShadowDO.java 0 → 100644
  View file @84c46ad
  1 +package cn.iocoder.yudao.module.iot.dal.dataobject.device;
  2 +
  3 +import cn.iocoder.yudao.framework.tenant.core.db.TenantBaseDO;
  4 +import cn.iocoder.yudao.module.iot.dal.dataobject.product.IotProductDO;
  5 +import com.baomidou.mybatisplus.annotation.KeySequence;
  6 +import com.baomidou.mybatisplus.annotation.TableField;
  7 +import com.baomidou.mybatisplus.annotation.TableId;
  8 +import com.baomidou.mybatisplus.annotation.TableName;
  9 +import com.baomidou.mybatisplus.extension.handlers.JacksonTypeHandler;
  10 +import lombok.AllArgsConstructor;
  11 +import lombok.Builder;
  12 +import lombok.Data;
  13 +import lombok.NoArgsConstructor;
  14 +
  15 +import java.util.Map;
  16 +
  17 +/**
  18 + * IoT 设备影子 DO
  19 + *
  20 + * @author 芋道源码
  21 + */
  22 +@TableName(value = "iot_device_shadow", autoResultMap = true)
  23 +@KeySequence("iot_device_shadow_seq")
  24 +@Data
  25 +@Builder
  26 +@NoArgsConstructor
  27 +@AllArgsConstructor
  28 +public class IotDeviceShadowDO extends TenantBaseDO {
  29 +
  30 + /**
  31 + * 主键
  32 + */
  33 + @TableId
  34 + private Long id;
  35 + /**
  36 + * 设备编号
  37 + */
  38 + private Long deviceId;
  39 + /**
  40 + * 产品编号
  41 + * <p>
  42 + * 关联 {@link IotProductDO#getId()}
  43 + */
  44 + private Long productId;
  45 + /**
  46 + * 产品标识
  47 + */
  48 + private String productKey;
  49 + /**
  50 + * 设备名称
  51 + */
  52 + private String deviceName;
  53 + /**
  54 + * 上报状态
  55 + */
  56 + @TableField(typeHandler = JacksonTypeHandler.class)
  57 + private Map<String, Object> reported;
  58 + /**
  59 + * 期望状态
  60 + */
  61 + @TableField(typeHandler = JacksonTypeHandler.class)
  62 + private Map<String, Object> desired;
  63 + /**
  64 + * 元数据：{ desired: { key: { timestamp } }, reported: { ... } }
  65 + */
  66 + @TableField(typeHandler = JacksonTypeHandler.class)
  67 + private Map<String, Object> metadata;
  68 + /**
  69 + * 版本号（乐观锁）
  70 + */
  71 + private Long version;
  72 + /**
  73 + * 文档最后更新时间（epoch 秒）
  74 + */
  75 + private Long shadowTime;
  76 + /**
  77 + * 状态
  78 + * <p>
  79 + * 枚举 {@link cn.iocoder.yudao.framework.common.enums.CommonStatusEnum}
  80 + */
  81 + private Integer status;
  82 +
  83 +}
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/device/IotDeviceShadowMapper.java 0 → 100644
  View file @84c46ad
  1 +package cn.iocoder.yudao.module.iot.dal.mysql.device;
  2 +
  3 +import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX;
  4 +import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceShadowDO;
  5 +import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
  6 +import org.apache.ibatis.annotations.Mapper;
  7 +
  8 +/**
  9 + * IoT 设备影子 Mapper
  10 + *
  11 + * @author 芋道源码
  12 + */
  13 +@Mapper
  14 +public interface IotDeviceShadowMapper extends BaseMapperX<IotDeviceShadowDO> {
  15 +
  16 + default IotDeviceShadowDO selectByDeviceId(Long deviceId) {
  17 + return selectOne(IotDeviceShadowDO::getDeviceId, deviceId);
  18 + }
  19 +
  20 + default int updateByDeviceIdAndVersion(IotDeviceShadowDO update, Long deviceId, Long expectVersion) {
  21 + return update(update, new LambdaUpdateWrapper<IotDeviceShadowDO>()
  22 + .eq(IotDeviceShadowDO::getDeviceId, deviceId)
  23 + .eq(IotDeviceShadowDO::getVersion, expectVersion));
  24 + }
  25 +
  26 + default int updateStatusByDeviceId(Long deviceId, Integer status) {
  27 + return update(new LambdaUpdateWrapper<IotDeviceShadowDO>()
  28 + .set(IotDeviceShadowDO::getStatus, status)
  29 + .eq(IotDeviceShadowDO::getDeviceId, deviceId));
  30 + }
  31 +
  32 + default int deleteByDeviceId(Long deviceId) {
  33 + return delete(IotDeviceShadowDO::getDeviceId, deviceId);
  34 + }
  35 +
  36 +}
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/RedisKeyConstants.java
  View file @84c46ad
@@ -19,6 +19,14 @@ public interface RedisKeyConstants { @@ -19,6 +19,14 @@ public interface RedisKeyConstants {
19 String DEVICE_PROPERTY = "iot:device_property:%d"; 19 String DEVICE_PROPERTY = "iot:device_property:%d";
20 20
21 /** 21 /**
  22 + * 设备影子数据缓存，采用 STRING 结构
  23 + * <p>
  24 + * KEY 格式：device_shadow:{deviceId}
  25 + * VALUE 数据类型：String(JSON) {@link cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceShadowDO}
  26 + */
  27 + String DEVICE_SHADOW = "iot:device_shadow:%d";
  28 +
  29 + /**
22 * 设备的最后上报时间，采用 ZSET 结构 30 * 设备的最后上报时间，采用 ZSET 结构
23 * 31 *
24 * KEY 格式：{deviceId} 32 * KEY 格式：{deviceId}
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/device/DeviceShadowRedisDAO.java 0 → 100644
  View file @84c46ad
  1 +package cn.iocoder.yudao.module.iot.dal.redis.device;
  2 +
  3 +import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
  4 +import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceShadowDO;
  5 +import jakarta.annotation.Resource;
  6 +import org.springframework.data.redis.core.StringRedisTemplate;
  7 +import org.springframework.stereotype.Repository;
  8 +
  9 +import static cn.iocoder.yudao.module.iot.dal.redis.RedisKeyConstants.DEVICE_SHADOW;
  10 +
  11 +/**
  12 + * {@link IotDeviceShadowDO} 的 Redis DAO
  13 + */
  14 +@Repository
  15 +public class DeviceShadowRedisDAO {
  16 +
  17 + @Resource
  18 + private StringRedisTemplate stringRedisTemplate;
  19 +
  20 + public IotDeviceShadowDO get(Long deviceId) {
  21 + String redisKey = formatKey(deviceId);
  22 + String json = stringRedisTemplate.opsForValue().get(redisKey);
  23 + return JsonUtils.parseObject(json, IotDeviceShadowDO.class);
  24 + }
  25 +
  26 + public void set(IotDeviceShadowDO shadow) {
  27 + String redisKey = formatKey(shadow.getDeviceId());
  28 + stringRedisTemplate.opsForValue().set(redisKey, JsonUtils.toJsonString(shadow));
  29 + }
  30 +
  31 + public void delete(Long deviceId) {
  32 + stringRedisTemplate.delete(formatKey(deviceId));
  33 + }
  34 +
  35 + private static String formatKey(Long deviceId) {
  36 + return String.format(DEVICE_SHADOW, deviceId);
  37 + }
  38 +
  39 +}
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/ErrorCodeConstants.java
  View file @84c46ad
@@ -62,6 +62,10 @@ public interface ErrorCodeConstants { @@ -62,6 +62,10 @@ public interface ErrorCodeConstants {
62 ErrorCode DEVICE_MODBUS_POINT_NOT_EXISTS = new ErrorCode(1_050_007_000, "设备 Modbus 点位配置不存在"); 62 ErrorCode DEVICE_MODBUS_POINT_NOT_EXISTS = new ErrorCode(1_050_007_000, "设备 Modbus 点位配置不存在");
63 ErrorCode DEVICE_MODBUS_POINT_EXISTS = new ErrorCode(1_050_007_001, "设备 Modbus 点位配置已存在"); 63 ErrorCode DEVICE_MODBUS_POINT_EXISTS = new ErrorCode(1_050_007_001, "设备 Modbus 点位配置已存在");
64 64
  65 + // ========== 设备影子 1-050-007-100 ==========
  66 + ErrorCode DEVICE_SHADOW_NOT_ENABLED = new ErrorCode(1_050_007_100, "设备影子未启用");
  67 + ErrorCode DEVICE_SHADOW_VERSION_CONFLICT = new ErrorCode(1_050_007_101, "设备影子版本冲突");
  68 +
65 // ========== OTA 固件相关 1-050-008-000 ========== 69 // ========== OTA 固件相关 1-050-008-000 ==========
66 70
67 ErrorCode OTA_FIRMWARE_NOT_EXISTS = new ErrorCode(1_050_008_000, "固件信息不存在"); 71 ErrorCode OTA_FIRMWARE_NOT_EXISTS = new ErrorCode(1_050_008_000, "固件信息不存在");
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/IotDeviceServiceImpl.java
  View file @84c46ad
@@ -37,6 +37,7 @@ import cn.iocoder.yudao.module.iot.dal.mysql.device.IotDeviceMapper; @@ -37,6 +37,7 @@ import cn.iocoder.yudao.module.iot.dal.mysql.device.IotDeviceMapper;
37 import cn.iocoder.yudao.module.iot.dal.redis.RedisKeyConstants; 37 import cn.iocoder.yudao.module.iot.dal.redis.RedisKeyConstants;
38 import cn.iocoder.yudao.module.iot.enums.product.IotProductDeviceTypeEnum; 38 import cn.iocoder.yudao.module.iot.enums.product.IotProductDeviceTypeEnum;
39 import cn.iocoder.yudao.module.iot.service.device.message.IotDeviceMessageService; 39 import cn.iocoder.yudao.module.iot.service.device.message.IotDeviceMessageService;
  40 +import cn.iocoder.yudao.module.iot.service.device.shadow.IotDeviceShadowService;
40 import cn.iocoder.yudao.module.iot.service.product.IotProductService; 41 import cn.iocoder.yudao.module.iot.service.product.IotProductService;
41 import jakarta.annotation.Resource; 42 import jakarta.annotation.Resource;
42 import jakarta.validation.ConstraintViolationException; 43 import jakarta.validation.ConstraintViolationException;
@@ -81,6 +82,9 @@ public class IotDeviceServiceImpl implements IotDeviceService { @@ -81,6 +82,9 @@ public class IotDeviceServiceImpl implements IotDeviceService {
81 @Resource 82 @Resource
82 @Lazy // 延迟加载，解决循环依赖 83 @Lazy // 延迟加载，解决循环依赖
83 private IotDeviceMessageService deviceMessageService; 84 private IotDeviceMessageService deviceMessageService;
  85 + @Resource
  86 + @Lazy // 延迟加载，解决循环依赖
  87 + private IotDeviceShadowService deviceShadowService;
84 88
85 private IotDeviceServiceImpl getSelf() { 89 private IotDeviceServiceImpl getSelf() {
86 return SpringUtil.getBean(getClass()); 90 return SpringUtil.getBean(getClass());
@@ -209,7 +213,10 @@ public class IotDeviceServiceImpl implements IotDeviceService { @@ -209,7 +213,10 @@ public class IotDeviceServiceImpl implements IotDeviceService {
209 // 2. 删除设备 213 // 2. 删除设备
210 deviceMapper.deleteById(id); 214 deviceMapper.deleteById(id);
211 215
212 - // 3. 清空对应缓存 216 + // 3. 级联删除影子
  217 + deviceShadowService.deleteByDeviceId(id);
  218 +
  219 + // 4. 清空对应缓存
213 deleteDeviceCache(device); 220 deleteDeviceCache(device);
214 } 221 }
215 222
@@ -235,7 +242,10 @@ public class IotDeviceServiceImpl implements IotDeviceService { @@ -235,7 +242,10 @@ public class IotDeviceServiceImpl implements IotDeviceService {
235 // 2. 删除设备 242 // 2. 删除设备
236 deviceMapper.deleteByIds(ids); 243 deviceMapper.deleteByIds(ids);
237 244
238 - // 3. 清空对应缓存 245 + // 3. 级联删除影子
  246 + ids.forEach(deviceShadowService::deleteByDeviceId);
  247 +
  248 + // 4. 清空对应缓存
239 deleteDeviceCache(devices); 249 deleteDeviceCache(devices);
240 } 250 }
241 251
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/message/IotDeviceMessageService.java
  View file @84c46ad
@@ -37,6 +37,16 @@ public interface IotDeviceMessageService { @@ -37,6 +37,16 @@ public interface IotDeviceMessageService {
37 IotDeviceMessage sendDeviceMessage(IotDeviceMessage message, IotDeviceDO device); 37 IotDeviceMessage sendDeviceMessage(IotDeviceMessage message, IotDeviceDO device);
38 38
39 /** 39 /**
  40 + * 发送设备消息（指定 serverId，用于上行回复等场景）
  41 + *
  42 + * @param message 消息
  43 + * @param device 设备
  44 + * @param serverId 网关 serverId
  45 + * @return 设备消息
  46 + */
  47 + IotDeviceMessage sendDeviceMessage(IotDeviceMessage message, IotDeviceDO device, String serverId);
  48 +
  49 + /**
40 * 发送设备消息 50 * 发送设备消息
41 * 51 *
42 * @param message 消息（“codec（编解码）字段” 部分字段） 52 * @param message 消息（“codec（编解码）字段” 部分字段）
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/message/IotDeviceMessageServiceImpl.java
  View file @84c46ad
@@ -28,6 +28,7 @@ import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceMessageDO; @@ -28,6 +28,7 @@ import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceMessageDO;
28 import cn.iocoder.yudao.module.iot.dal.tdengine.IotDeviceMessageMapper; 28 import cn.iocoder.yudao.module.iot.dal.tdengine.IotDeviceMessageMapper;
29 import cn.iocoder.yudao.module.iot.service.device.IotDeviceService; 29 import cn.iocoder.yudao.module.iot.service.device.IotDeviceService;
30 import cn.iocoder.yudao.module.iot.service.device.property.IotDevicePropertyService; 30 import cn.iocoder.yudao.module.iot.service.device.property.IotDevicePropertyService;
  31 +import cn.iocoder.yudao.module.iot.service.device.shadow.IotDeviceShadowService;
31 import cn.iocoder.yudao.module.iot.service.ota.IotOtaTaskRecordService; 32 import cn.iocoder.yudao.module.iot.service.ota.IotOtaTaskRecordService;
32 import com.baomidou.mybatisplus.core.metadata.IPage; 33 import com.baomidou.mybatisplus.core.metadata.IPage;
33 import com.baomidou.mybatisplus.extension.plugins.pagination.Page; 34 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
@@ -65,6 +66,9 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { @@ -65,6 +66,9 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
65 @Resource 66 @Resource
66 @Lazy // 延迟加载，避免循环依赖 67 @Lazy // 延迟加载，避免循环依赖
67 private IotOtaTaskRecordService otaTaskRecordService; 68 private IotOtaTaskRecordService otaTaskRecordService;
  69 + @Resource
  70 + @Lazy // 延迟加载，避免循环依赖
  71 + private IotDeviceShadowService deviceShadowService;
68 72
69 @Resource 73 @Resource
70 private IotDeviceMessageMapper deviceMessageMapper; 74 private IotDeviceMessageMapper deviceMessageMapper;
@@ -120,7 +124,8 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { @@ -120,7 +124,8 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
120 return sendDeviceMessage(message, device, null); 124 return sendDeviceMessage(message, device, null);
121 } 125 }
122 126
123 - private IotDeviceMessage sendDeviceMessage(IotDeviceMessage message, IotDeviceDO device, String serverId) { 127 + @Override
  128 + public IotDeviceMessage sendDeviceMessage(IotDeviceMessage message, IotDeviceDO device, String serverId) {
124 // 1. 补充信息 129 // 1. 补充信息
125 appendDeviceMessage(message, device); 130 appendDeviceMessage(message, device);
126 131
@@ -229,6 +234,12 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { @@ -229,6 +234,12 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
229 return null; 234 return null;
230 } 235 }
231 236
  237 + // 设备影子更新
  238 + if (Objects.equal(message.getMethod(), IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod())) {
  239 + deviceShadowService.handleDeviceShadowUpdate(device, message);
  240 + return null;
  241 + }
  242 +
232 // 添加拓扑关系 243 // 添加拓扑关系
233 if (Objects.equal(message.getMethod(), IotDeviceMessageMethodEnum.TOPO_ADD.getMethod())) { 244 if (Objects.equal(message.getMethod(), IotDeviceMessageMethodEnum.TOPO_ADD.getMethod())) {
234 return deviceService.handleTopoAddMessage(message, device); 245 return deviceService.handleTopoAddMessage(message, device);
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowService.java 0 → 100644
  View file @84c46ad
  1 +package cn.iocoder.yudao.module.iot.service.device.shadow;
  2 +
  3 +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  4 +import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
  5 +import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceShadowDO;
  6 +
  7 +import java.util.Map;
  8 +
  9 +/**
  10 + * IoT 设备影子 Service 接口
  11 + *
  12 + * @author 芋道源码
  13 + */
  14 +public interface IotDeviceShadowService {
  15 +
  16 + /**
  17 + * 获取或创建设备影子
  18 + *
  19 + * @param device 设备
  20 + * @return 设备影子
  21 + */
  22 + IotDeviceShadowDO getOrCreateShadow(IotDeviceDO device);
  23 +
  24 + /**
  25 + * 处理设备影子上行更新
  26 + *
  27 + * @param device 设备
  28 + * @param message 上行消息
  29 + */
  30 + void handleDeviceShadowUpdate(IotDeviceDO device, IotDeviceMessage message);
  31 +
  32 + /**
  33 + * 更新期望状态
  34 + *
  35 + * @param deviceId 设备编号
  36 + * @param desired 期望状态
  37 + */
  38 + void updateDesiredState(Long deviceId, Map<String, Object> desired);
  39 +
  40 + /**
  41 + * 重置影子
  42 + *
  43 + * @param deviceId 设备编号
  44 + */
  45 + void resetShadow(Long deviceId);
  46 +
  47 + /**
  48 + * 更新影子开关状态
  49 + *
  50 + * @param deviceId 设备编号
  51 + * @param status 状态
  52 + */
  53 + void updateStatus(Long deviceId, Integer status);
  54 +
  55 + /**
  56 + * 删除设备影子（设备删除时级联）
  57 + *
  58 + * @param deviceId 设备编号
  59 + */
  60 + void deleteByDeviceId(Long deviceId);
  61 +
  62 +}
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImpl.java 0 → 100644
  View file @84c46ad
  1 +package cn.iocoder.yudao.module.iot.service.device.shadow;
  2 +
  3 +import cn.hutool.core.collection.CollUtil;
  4 +import cn.hutool.core.date.DateUtil;
  5 +import cn.hutool.core.util.StrUtil;
  6 +import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
  7 +import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
  8 +import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
  9 +import cn.iocoder.yudao.module.iot.core.enums.shadow.IotDeviceShadowErrorCodeEnum;
  10 +import cn.iocoder.yudao.module.iot.core.enums.shadow.IotDeviceShadowMetadatas;
  11 +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  12 +import cn.iocoder.yudao.module.iot.core.topic.shadow.IotDeviceShadowReplyDTO;
  13 +import cn.iocoder.yudao.module.iot.core.topic.shadow.IotDeviceShadowUpdateReqDTO;
  14 +import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
  15 +import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceShadowDO;
  16 +import cn.iocoder.yudao.module.iot.dal.mysql.device.IotDeviceShadowMapper;
  17 +import cn.iocoder.yudao.module.iot.dal.redis.device.DeviceShadowRedisDAO;
  18 +import cn.iocoder.yudao.module.iot.service.device.IotDeviceService;
  19 +import cn.iocoder.yudao.module.iot.service.device.message.IotDeviceMessageService;
  20 +import cn.iocoder.yudao.module.iot.service.device.property.IotDevicePropertyService;
  21 +import jakarta.annotation.Resource;
  22 +import lombok.Getter;
  23 +import lombok.extern.slf4j.Slf4j;
  24 +import org.springframework.context.annotation.Lazy;
  25 +import org.springframework.dao.DuplicateKeyException;
  26 +import org.springframework.stereotype.Service;
  27 +import org.springframework.transaction.annotation.Transactional;
  28 +import org.springframework.validation.annotation.Validated;
  29 +
  30 +import java.util.HashMap;
  31 +import java.util.Map;
  32 +import java.util.function.Function;
  33 +
  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;
  37 +import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.DEVICE_SHADOW_NOT_ENABLED;
  38 +import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.DEVICE_SHADOW_VERSION_CONFLICT;
  39 +
  40 +/**
  41 + * IoT 设备影子 Service 实现类
  42 + *
  43 + * @author 芋道源码
  44 + */
  45 +@Service
  46 +@Validated
  47 +@Slf4j
  48 +public class IotDeviceShadowServiceImpl implements IotDeviceShadowService {
  49 +
  50 + private static final int MAX_PROPERTY_COUNT = 128;
  51 + private static final int OPTIMISTIC_LOCK_RETRY = 3;
  52 +
  53 + @Resource
  54 + private IotDeviceShadowMapper deviceShadowMapper;
  55 + @Resource
  56 + private DeviceShadowRedisDAO deviceShadowRedisDAO;
  57 + @Resource
  58 + private IotDeviceService deviceService;
  59 + @Resource
  60 + private IotDevicePropertyService devicePropertyService;
  61 + @Resource
  62 + @Lazy
  63 + private IotDeviceMessageService deviceMessageService;
  64 +
  65 + @Override
  66 + public IotDeviceShadowDO getOrCreateShadow(IotDeviceDO device) {
  67 + IotDeviceShadowDO shadow = getShadowFromCacheOrDb(device.getId());
  68 + if (shadow != null) {
  69 + return shadow;
  70 + }
  71 + // 懒插入空文档
  72 + IotDeviceShadowDO newShadow = buildEmptyShadow(device)
  73 + .status(CommonStatusEnum.ENABLE.getStatus())
  74 + .build();
  75 + try {
  76 + deviceShadowMapper.insert(newShadow);
  77 + return newShadow;
  78 + } catch (DuplicateKeyException ex) {
  79 + return deviceShadowMapper.selectByDeviceId(device.getId());
  80 + }
  81 + }
  82 +
  83 + @Override
  84 + public void handleDeviceShadowUpdate(IotDeviceDO device, IotDeviceMessage message) {
  85 + try {
  86 + handleDeviceShadowUpdate0(device, message);
  87 + } catch (Exception ex) {
  88 + log.error("[handleDeviceShadowUpdate][device({}) message({}) 处理异常]", device.getId(), message, ex);
  89 + Long version = resolveShadowVersionSafely(device.getId());
  90 + sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf(version,
  91 + IotDeviceShadowErrorCodeEnum.SERVER_ERROR.getCode(),
  92 + IotDeviceShadowErrorCodeEnum.SERVER_ERROR.getMessage(), DateUtil.currentSeconds()));
  93 + }
  94 + }
  95 +
  96 + private void handleDeviceShadowUpdate0(IotDeviceDO device, IotDeviceMessage message) {
  97 + // 1. 解析请求
  98 + IotDeviceShadowUpdateReqDTO req = JsonUtils.convertObject(message.getParams(), IotDeviceShadowUpdateReqDTO.class);
  99 + if (req == null) {
  100 + sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf(0L,
  101 + IotDeviceShadowErrorCodeEnum.JSON_FORMAT_ERROR.getCode(),
  102 + IotDeviceShadowErrorCodeEnum.JSON_FORMAT_ERROR.getMessage(), DateUtil.currentSeconds()));
  103 + return;
  104 + }
  105 + if (StrUtil.isEmpty(req.getMethod())) {
  106 + sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf(resolveShadowVersionSafely(device.getId()),
  107 + IotDeviceShadowErrorCodeEnum.METHOD_MISSING.getCode(),
  108 + IotDeviceShadowErrorCodeEnum.METHOD_MISSING.getMessage(), DateUtil.currentSeconds()));
  109 + return;
  110 + }
  111 +
  112 + // 2. 获取影子并校验开关
  113 + IotDeviceShadowDO shadow = getOrCreateShadow(device);
  114 + if (CommonStatusEnum.isDisable(shadow.getStatus())) {
  115 + log.info("[handleDeviceShadowUpdate0][device({}) 影子未启用，忽略上行]", device.getId());
  116 + return;
  117 + }
  118 +
  119 + // 3. 内层 method 分发
  120 + if (INNER_METHOD_GET.equals(req.getMethod())) {
  121 + sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.getReplyOf(
  122 + shadow.getDesired(), shadow.getReported(), shadow.getVersion(), shadow.getShadowTime()));
  123 + return;
  124 + }
  125 + if (INNER_METHOD_UPDATE.equals(req.getMethod())) {
  126 + handleReportedUpdate(device, message.getServerId(), req);
  127 + return;
  128 + }
  129 + sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf(shadow.getVersion(),
  130 + IotDeviceShadowErrorCodeEnum.INVALID_METHOD.getCode(),
  131 + IotDeviceShadowErrorCodeEnum.INVALID_METHOD.getMessage(), DateUtil.currentSeconds()));
  132 + }
  133 +
  134 + private void handleReportedUpdate(IotDeviceDO device, String serverId, IotDeviceShadowUpdateReqDTO req) {
  135 + Long deviceId = device.getId();
  136 +
  137 + if (req.getState() == null) {
  138 + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(resolveShadowVersionSafely(deviceId),
  139 + IotDeviceShadowErrorCodeEnum.STATE_MISSING.getCode(),
  140 + IotDeviceShadowErrorCodeEnum.STATE_MISSING.getMessage(), DateUtil.currentSeconds()));
  141 + return;
  142 + }
  143 + if (req.getState().getReported() == null) {
  144 + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(resolveShadowVersionSafely(deviceId),
  145 + IotDeviceShadowErrorCodeEnum.REPORTED_MISSING.getCode(),
  146 + IotDeviceShadowErrorCodeEnum.REPORTED_MISSING.getMessage(), DateUtil.currentSeconds()));
  147 + return;
  148 + }
  149 + if (CollUtil.isEmpty(req.getState().getReported())) {
  150 + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(resolveShadowVersionSafely(deviceId),
  151 + IotDeviceShadowErrorCodeEnum.REPORTED_EMPTY.getCode(),
  152 + IotDeviceShadowErrorCodeEnum.REPORTED_EMPTY.getMessage(), DateUtil.currentSeconds()));
  153 + return;
  154 + }
  155 +
  156 + // version == -1：清空影子
  157 + if (req.getVersion() != null && req.getVersion() == -1L) {
  158 + try {
  159 + resetShadowWithRetry(deviceId);
  160 + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.updateReplyOf(0L, DateUtil.currentSeconds()));
  161 + } catch (ShadowCasExhaustedException ex) {
  162 + handleCasExhaustedForDevice(device, serverId, ex);
  163 + }
  164 + return;
  165 + }
  166 +
  167 + try {
  168 + IotDeviceShadowDO committed = casUpdateShadow(deviceId, latest -> buildReportedUpdate(latest, req));
  169 + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.updateReplyOf(committed.getVersion(),
  170 + committed.getShadowTime()));
  171 + } catch (ShadowUpdateRejectedException ex) {
  172 + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(ex.getVersion(),
  173 + ex.getErrorCode().getCode(), ex.getErrorCode().getMessage(), DateUtil.currentSeconds()));
  174 + } catch (ShadowCasExhaustedException ex) {
  175 + handleCasExhaustedForDevice(device, serverId, ex);
  176 + }
  177 + }
  178 +
  179 + @Override
  180 + public void updateDesiredState(Long deviceId, Map<String, Object> desired) {
  181 + IotDeviceDO device = deviceService.validateDeviceExists(deviceId);
  182 + IotDeviceShadowDO shadow = getOrCreateShadow(device);
  183 + if (CommonStatusEnum.isDisable(shadow.getStatus())) {
  184 + throw exception(DEVICE_SHADOW_NOT_ENABLED);
  185 + }
  186 + if (CollUtil.isEmpty(desired)) {
  187 + return;
  188 + }
  189 +
  190 + try {
  191 + IotDeviceShadowDO committed = casUpdateShadow(deviceId, latest -> buildDesiredUpdate(latest, desired));
  192 + // 设备在线则主动推送全量影子（复用 CAS 返回值，避免二次查库）
  193 + String serverId = devicePropertyService.getDeviceServerId(deviceId);
  194 + if (StrUtil.isNotEmpty(serverId)) {
  195 + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.getReplyOf(
  196 + committed.getDesired(), committed.getReported(),
  197 + committed.getVersion(), committed.getShadowTime()));
  198 + }
  199 + } catch (ShadowCasExhaustedException ex) {
  200 + throw exception(DEVICE_SHADOW_VERSION_CONFLICT);
  201 + }
  202 + }
  203 +
  204 + @Override
  205 + public void resetShadow(Long deviceId) {
  206 + deviceService.validateDeviceExists(deviceId);
  207 + IotDeviceShadowDO shadow = getShadowFromCacheOrDb(deviceId);
  208 + if (shadow == null) {
  209 + return;
  210 + }
  211 + try {
  212 + resetShadowWithRetry(deviceId);
  213 + } catch (ShadowCasExhaustedException ex) {
  214 + throw exception(DEVICE_SHADOW_VERSION_CONFLICT);
  215 + }
  216 + }
  217 +
  218 + @Override
  219 + @Transactional(rollbackFor = Exception.class)
  220 + public void updateStatus(Long deviceId, Integer status) {
  221 + IotDeviceDO device = deviceService.validateDeviceExists(deviceId);
  222 + IotDeviceShadowDO shadow = getShadowFromCacheOrDb(deviceId);
  223 + if (shadow == null) {
  224 + // 影子行不存在时直接按目标 status 懒建，避免后续 getOrCreateShadow 始终以 ENABLE 插入
  225 + IotDeviceShadowDO newShadow = buildEmptyShadow(device).status(status).build();
  226 + try {
  227 + deviceShadowMapper.insert(newShadow);
  228 + } catch (DuplicateKeyException ex) {
  229 + int rows = deviceShadowMapper.updateStatusByDeviceId(deviceId, status);
  230 + if (rows == 0) {
  231 + throw exception(DEVICE_SHADOW_VERSION_CONFLICT);
  232 + }
  233 + deviceShadowRedisDAO.delete(deviceId);
  234 + }
  235 + } else {
  236 + int rows = deviceShadowMapper.updateStatusByDeviceId(deviceId, status);
  237 + if (rows == 0) {
  238 + throw exception(DEVICE_SHADOW_VERSION_CONFLICT);
  239 + }
  240 + deviceShadowRedisDAO.delete(deviceId);
  241 + }
  242 + }
  243 +
  244 + @Override
  245 + @Transactional(rollbackFor = Exception.class)
  246 + public void deleteByDeviceId(Long deviceId) {
  247 + deviceShadowMapper.deleteByDeviceId(deviceId);
  248 + deviceShadowRedisDAO.delete(deviceId);
  249 + }
  250 +
  251 + // ========== 内部方法 ==========
  252 +
  253 + private IotDeviceShadowDO buildReportedUpdate(IotDeviceShadowDO latest, IotDeviceShadowUpdateReqDTO req) {
  254 + if (req.getVersion() != null && req.getVersion() < latest.getVersion()) {
  255 + throw new ShadowUpdateRejectedException(latest.getVersion(), IotDeviceShadowErrorCodeEnum.VERSION_CONFLICT);
  256 + }
  257 + Map<String, Object> mergedReported = mergeMap(latest.getReported(), req.getState().getReported());
  258 + if (mergedReported.size() > MAX_PROPERTY_COUNT) {
  259 + throw new ShadowUpdateRejectedException(latest.getVersion(), IotDeviceShadowErrorCodeEnum.PROPERTY_COUNT_EXCEEDED);
  260 + }
  261 + long now = DateUtil.currentSeconds();
  262 + Map<String, Object> metadata = latest.getMetadata() != null ? new HashMap<>(latest.getMetadata()) : new HashMap<>();
  263 + IotDeviceShadowMetadatas.putReportedTimestamps(metadata, req.getState().getReported().keySet(), now);
  264 + return IotDeviceShadowDO.builder()
  265 + .reported(mergedReported)
  266 + .metadata(metadata)
  267 + .version(latest.getVersion() + 1)
  268 + .shadowTime(now)
  269 + .build();
  270 + }
  271 +
  272 + private IotDeviceShadowDO buildDesiredUpdate(IotDeviceShadowDO latest, Map<String, Object> desired) {
  273 + long now = DateUtil.currentSeconds();
  274 + Map<String, Object> mergedDesired = mergeMap(latest.getDesired(), desired);
  275 + Map<String, Object> metadata = latest.getMetadata() != null ? new HashMap<>(latest.getMetadata()) : new HashMap<>();
  276 + IotDeviceShadowMetadatas.putDesiredTimestamps(metadata, desired.keySet(), now);
  277 + return IotDeviceShadowDO.builder()
  278 + .desired(mergedDesired)
  279 + .metadata(metadata)
  280 + .version(latest.getVersion() + 1)
  281 + .shadowTime(now)
  282 + .build();
  283 + }
  284 +
  285 + private void resetShadowWithRetry(Long deviceId) {
  286 + casUpdateShadow(deviceId, latest -> IotDeviceShadowDO.builder()
  287 + .desired(new HashMap<>())
  288 + .reported(new HashMap<>())
  289 + .metadata(new HashMap<>())
  290 + .version(0L)
  291 + .shadowTime(DateUtil.currentSeconds())
  292 + .build());
  293 + }
  294 +
  295 + /**
  296 + * 乐观锁 CAS 更新：每次重试基于最新影子重新构建 update，避免 lost-update
  297 + *
  298 + * @return 写入成功后的 update DO（回填未更新字段，便于构建全量回复）
  299 + */
  300 + private IotDeviceShadowDO casUpdateShadow(Long deviceId, Function<IotDeviceShadowDO, IotDeviceShadowDO> updateBuilder) {
  301 + Long lastKnownVersion = 0L;
  302 + for (int i = 0; i < OPTIMISTIC_LOCK_RETRY; i++) {
  303 + // 首次尝试走缓存，重试时直读 DB 获取最新 version
  304 + IotDeviceShadowDO latest = i == 0
  305 + ? getShadowFromCacheOrDb(deviceId)
  306 + : deviceShadowMapper.selectByDeviceId(deviceId);
  307 + if (latest == null) {
  308 + throw ShadowCasExhaustedException.shadowNotFound();
  309 + }
  310 + lastKnownVersion = latest.getVersion();
  311 + IotDeviceShadowDO update = updateBuilder.apply(latest);
  312 + int rows = deviceShadowMapper.updateByDeviceIdAndVersion(update, deviceId, latest.getVersion());
  313 + if (rows > 0) {
  314 + deviceShadowRedisDAO.delete(deviceId);
  315 + enrichCommittedShadow(update, latest);
  316 + return update;
  317 + }
  318 + }
  319 + throw ShadowCasExhaustedException.versionConflict(lastKnownVersion);
  320 + }
  321 +
  322 + /**
  323 + * 将 CAS 成功写入的局部字段与 latest 快照合并，供全量 get_reply 使用
  324 + */
  325 + private static void enrichCommittedShadow(IotDeviceShadowDO committed, IotDeviceShadowDO latest) {
  326 + if (committed.getDesired() == null) {
  327 + committed.setDesired(latest.getDesired());
  328 + }
  329 + if (committed.getReported() == null) {
  330 + committed.setReported(latest.getReported());
  331 + }
  332 + if (committed.getMetadata() == null) {
  333 + committed.setMetadata(latest.getMetadata());
  334 + }
  335 + }
  336 +
  337 + private void handleCasExhaustedForDevice(IotDeviceDO device, String serverId, ShadowCasExhaustedException ex) {
  338 + if (ex.isShadowNotFound()) {
  339 + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(0L,
  340 + IotDeviceShadowErrorCodeEnum.SERVER_ERROR.getCode(),
  341 + IotDeviceShadowErrorCodeEnum.SERVER_ERROR.getMessage(), DateUtil.currentSeconds()));
  342 + return;
  343 + }
  344 + sendVersionConflictReply(device, serverId, ex.getLastKnownVersion());
  345 + }
  346 +
  347 + private void sendVersionConflictReply(IotDeviceDO device, String serverId, Long version) {
  348 + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(version,
  349 + IotDeviceShadowErrorCodeEnum.VERSION_CONFLICT.getCode(),
  350 + IotDeviceShadowErrorCodeEnum.VERSION_CONFLICT.getMessage(), DateUtil.currentSeconds()));
  351 + }
  352 +
  353 + private Long resolveShadowVersionSafely(Long deviceId) {
  354 + try {
  355 + IotDeviceShadowDO shadow = getShadowFromCacheOrDb(deviceId);
  356 + return shadow != null ? shadow.getVersion() : 0L;
  357 + } catch (Exception ex) {
  358 + log.warn("[resolveShadowVersionSafely][device({}) 读取 version 失败，使用保底 0]", deviceId, ex);
  359 + return 0L;
  360 + }
  361 + }
  362 +
  363 + private IotDeviceShadowDO.IotDeviceShadowDOBuilder buildEmptyShadow(IotDeviceDO device) {
  364 + return IotDeviceShadowDO.builder()
  365 + .deviceId(device.getId())
  366 + .productId(device.getProductId())
  367 + .productKey(device.getProductKey())
  368 + .deviceName(device.getDeviceName())
  369 + .reported(new HashMap<>())
  370 + .desired(new HashMap<>())
  371 + .metadata(new HashMap<>())
  372 + .version(0L)
  373 + .shadowTime(DateUtil.currentSeconds());
  374 + }
  375 +
  376 + private IotDeviceShadowDO getShadowFromCacheOrDb(Long deviceId) {
  377 + IotDeviceShadowDO shadow = deviceShadowRedisDAO.get(deviceId);
  378 + if (shadow != null) {
  379 + return shadow;
  380 + }
  381 + shadow = deviceShadowMapper.selectByDeviceId(deviceId);
  382 + if (shadow != null) {
  383 + deviceShadowRedisDAO.set(shadow);
  384 + }
  385 + return shadow;
  386 + }
  387 +
  388 + /**
  389 + * 下发影子回复，内部吞掉异常，保证上行处理链路不向上抛
  390 + */
  391 + private void sendShadowReplySafely(IotDeviceDO device, String serverId, IotDeviceShadowReplyDTO replyDTO) {
  392 + IotDeviceMessage shadowGetMsg = IotDeviceMessage.requestOf(
  393 + device.getId(), device.getTenantId(), serverId,
  394 + IotDeviceMessageMethodEnum.SHADOW_GET.getMethod(), replyDTO);
  395 + try {
  396 + deviceMessageService.sendDeviceMessage(shadowGetMsg, device, serverId);
  397 + } catch (Exception ex) {
  398 + log.error("[sendShadowReply][device({}) 下发影子回复失败]", device.getId(), ex);
  399 + }
  400 + }
  401 +
  402 + private static Map<String, Object> mergeMap(Map<String, Object> oldMap, Map<String, Object> newMap) {
  403 + Map<String, Object> result = oldMap != null ? new HashMap<>(oldMap) : new HashMap<>();
  404 + if (CollUtil.isNotEmpty(newMap)) {
  405 + result.putAll(newMap);
  406 + }
  407 + return result;
  408 + }
  409 +
  410 + @Getter
  411 + private static class ShadowUpdateRejectedException extends RuntimeException {
  412 +
  413 + private final Long version;
  414 + private final IotDeviceShadowErrorCodeEnum errorCode;
  415 +
  416 + private ShadowUpdateRejectedException(Long version, IotDeviceShadowErrorCodeEnum errorCode) {
  417 + super(errorCode.getMessage());
  418 + this.version = version;
  419 + this.errorCode = errorCode;
  420 + }
  421 +
  422 + }
  423 +
  424 + @Getter
  425 + private static class ShadowCasExhaustedException extends RuntimeException {
  426 +
  427 + private final Long lastKnownVersion;
  428 + private final boolean shadowNotFound;
  429 +
  430 + private ShadowCasExhaustedException(Long lastKnownVersion, boolean shadowNotFound) {
  431 + this.lastKnownVersion = lastKnownVersion != null ? lastKnownVersion : 0L;
  432 + this.shadowNotFound = shadowNotFound;
  433 + }
  434 +
  435 + static ShadowCasExhaustedException shadowNotFound() {
  436 + return new ShadowCasExhaustedException(0L, true);
  437 + }
  438 +
  439 + static ShadowCasExhaustedException versionConflict(Long lastKnownVersion) {
  440 + return new ShadowCasExhaustedException(lastKnownVersion, false);
  441 + }
  442 +
  443 + }
  444 +
  445 +}
yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImplTest.java 0 → 100644
  View file @84c46ad
  1 +package cn.iocoder.yudao.module.iot.service.device.shadow;
  2 +
  3 +import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
  4 +import cn.iocoder.yudao.framework.common.exception.ServiceException;
  5 +import cn.iocoder.yudao.framework.test.core.ut.BaseMockitoUnitTest;
  6 +import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
  7 +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  8 +import cn.iocoder.yudao.module.iot.core.topic.shadow.IotDeviceShadowReplyDTO;
  9 +import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
  10 +import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceShadowDO;
  11 +import cn.iocoder.yudao.module.iot.dal.mysql.device.IotDeviceShadowMapper;
  12 +import cn.iocoder.yudao.module.iot.dal.redis.device.DeviceShadowRedisDAO;
  13 +import cn.iocoder.yudao.module.iot.service.device.IotDeviceService;
  14 +import cn.iocoder.yudao.module.iot.service.device.message.IotDeviceMessageService;
  15 +import cn.iocoder.yudao.module.iot.service.device.property.IotDevicePropertyService;
  16 +import org.junit.jupiter.api.BeforeEach;
  17 +import org.junit.jupiter.api.Test;
  18 +import org.mockito.ArgumentCaptor;
  19 +import org.mockito.InjectMocks;
  20 +import org.mockito.Mock;
  21 +import org.springframework.dao.DuplicateKeyException;
  22 +
  23 +import java.util.HashMap;
  24 +import java.util.Map;
  25 +import java.util.concurrent.atomic.AtomicInteger;
  26 +
  27 +import static org.junit.jupiter.api.Assertions.*;
  28 +import static org.mockito.ArgumentMatchers.*;
  29 +import static org.mockito.Mockito.*;
  30 +
  31 +/**
  32 + * {@link IotDeviceShadowServiceImpl} 的单元测试
  33 + */
  34 +public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
  35 +
  36 + @InjectMocks
  37 + private IotDeviceShadowServiceImpl service;
  38 +
  39 + @Mock
  40 + private IotDeviceShadowMapper deviceShadowMapper;
  41 + @Mock
  42 + private DeviceShadowRedisDAO deviceShadowRedisDAO;
  43 + @Mock
  44 + private IotDeviceService deviceService;
  45 + @Mock
  46 + private IotDevicePropertyService devicePropertyService;
  47 + @Mock
  48 + private IotDeviceMessageService deviceMessageService;
  49 +
  50 + @BeforeEach
  51 + public void setUp() {
  52 + reset(deviceShadowMapper, deviceShadowRedisDAO, deviceService,
  53 + devicePropertyService, deviceMessageService);
  54 + }
  55 +
  56 + @Test
  57 + public void testHandleDeviceShadowUpdate_get_replyFullShadow() {
  58 + IotDeviceDO device = buildDevice();
  59 + IotDeviceShadowDO shadow = buildShadow(device, 2L);
  60 + when(deviceShadowRedisDAO.get(device.getId())).thenReturn(null);
  61 + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(shadow);
  62 +
  63 + Map<String, Object> params = new HashMap<>();
  64 + params.put("method", "get");
  65 + IotDeviceMessage message = IotDeviceMessage.requestOf(device.getId(), 1L, "server-1",
  66 + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), params);
  67 +
  68 + service.handleDeviceShadowUpdate(device, message);
  69 +
  70 + ArgumentCaptor<IotDeviceMessage> captor = ArgumentCaptor.forClass(IotDeviceMessage.class);
  71 + verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1"));
  72 + IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams();
  73 + assertEquals("get_reply", reply.getMethod());
  74 + assertEquals(2L, reply.getVersion());
  75 + }
  76 +
  77 + @Test
  78 + public void testHandleDeviceShadowUpdate_update_mergeReportedAndVersionIncrement() {
  79 + IotDeviceDO device = buildDevice();
  80 + IotDeviceShadowDO shadow = buildShadow(device, 1L);
  81 + shadow.getReported().put("Temperature", 20);
  82 + when(deviceShadowRedisDAO.get(device.getId())).thenReturn(null);
  83 + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(shadow);
  84 + when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(1L))).thenReturn(1);
  85 +
  86 + Map<String, Object> reported = new HashMap<>();
  87 + reported.put("Temperature", 25);
  88 + reported.put("Humidity", 60);
  89 + Map<String, Object> state = new HashMap<>();
  90 + state.put("reported", reported);
  91 + Map<String, Object> params = new HashMap<>();
  92 + params.put("method", "update");
  93 + params.put("state", state);
  94 + params.put("version", 1L);
  95 + IotDeviceMessage message = IotDeviceMessage.requestOf(device.getId(), 1L, "server-1",
  96 + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), params);
  97 +
  98 + service.handleDeviceShadowUpdate(device, message);
  99 +
  100 + ArgumentCaptor<IotDeviceShadowDO> updateCaptor = ArgumentCaptor.forClass(IotDeviceShadowDO.class);
  101 + verify(deviceShadowMapper).updateByDeviceIdAndVersion(updateCaptor.capture(), eq(device.getId()), eq(1L));
  102 + assertEquals(25, updateCaptor.getValue().getReported().get("Temperature"));
  103 + assertEquals(60, updateCaptor.getValue().getReported().get("Humidity"));
  104 + assertEquals(2L, updateCaptor.getValue().getVersion());
  105 + verify(deviceShadowRedisDAO).delete(device.getId());
  106 + }
  107 +
  108 + @Test
  109 + public void testHandleDeviceShadowUpdate_versionConflict_send409() {
  110 + IotDeviceDO device = buildDevice();
  111 + IotDeviceShadowDO shadow = buildShadow(device, 5L);
  112 + when(deviceShadowRedisDAO.get(device.getId())).thenReturn(null);
  113 + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(shadow);
  114 +
  115 + Map<String, Object> reported = new HashMap<>();
  116 + reported.put("Temperature", 25);
  117 + Map<String, Object> state = new HashMap<>();
  118 + state.put("reported", reported);
  119 + Map<String, Object> params = new HashMap<>();
  120 + params.put("method", "update");
  121 + params.put("state", state);
  122 + params.put("version", 3L);
  123 + IotDeviceMessage message = IotDeviceMessage.requestOf(device.getId(), 1L, "server-1",
  124 + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), params);
  125 +
  126 + service.handleDeviceShadowUpdate(device, message);
  127 +
  128 + ArgumentCaptor<IotDeviceMessage> captor = ArgumentCaptor.forClass(IotDeviceMessage.class);
  129 + verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1"));
  130 + IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams();
  131 + assertEquals("update_reply", reply.getMethod());
  132 + @SuppressWarnings("unchecked")
  133 + Map<String, Object> payload = (Map<String, Object>) reply.getPayload();
  134 + assertEquals("error", payload.get("status"));
  135 + verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any());
  136 + }
  137 +
  138 + @Test
  139 + public void testHandleDeviceShadowUpdate_versionMinusOne_resetShadow() {
  140 + IotDeviceDO device = buildDevice();
  141 + IotDeviceShadowDO shadow = buildShadow(device, 3L);
  142 + shadow.getDesired().put("PowerSwitch", 1);
  143 + shadow.getReported().put("Temperature", 20);
  144 + when(deviceShadowRedisDAO.get(device.getId())).thenReturn(null);
  145 + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(shadow);
  146 + when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(3L))).thenReturn(1);
  147 +
  148 + Map<String, Object> reported = new HashMap<>();
  149 + reported.put("Temperature", 20);
  150 + Map<String, Object> state = new HashMap<>();
  151 + state.put("reported", reported);
  152 + Map<String, Object> params = new HashMap<>();
  153 + params.put("method", "update");
  154 + params.put("state", state);
  155 + params.put("version", -1L);
  156 + IotDeviceMessage message = IotDeviceMessage.requestOf(device.getId(), 1L, "server-1",
  157 + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), params);
  158 +
  159 + service.handleDeviceShadowUpdate(device, message);
  160 +
  161 + ArgumentCaptor<IotDeviceShadowDO> updateCaptor = ArgumentCaptor.forClass(IotDeviceShadowDO.class);
  162 + verify(deviceShadowMapper).updateByDeviceIdAndVersion(updateCaptor.capture(), eq(device.getId()), eq(3L));
  163 + assertTrue(updateCaptor.getValue().getDesired().isEmpty());
  164 + assertTrue(updateCaptor.getValue().getReported().isEmpty());
  165 + assertEquals(0L, updateCaptor.getValue().getVersion());
  166 + }
  167 +
  168 + @Test
  169 + public void testHandleDeviceShadowUpdate_disabled_ignore() {
  170 + IotDeviceDO device = buildDevice();
  171 + IotDeviceShadowDO shadow = buildShadow(device, 0L, CommonStatusEnum.DISABLE.getStatus());
  172 + when(deviceShadowRedisDAO.get(device.getId())).thenReturn(null);
  173 + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(shadow);
  174 +
  175 + Map<String, Object> params = new HashMap<>();
  176 + params.put("method", "get");
  177 + IotDeviceMessage message = IotDeviceMessage.requestOf(device.getId(), 1L, "server-1",
  178 + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), params);
  179 +
  180 + service.handleDeviceShadowUpdate(device, message);
  181 +
  182 + verify(deviceMessageService, never()).sendDeviceMessage(any(), any(), any());
  183 + }
  184 +
  185 + @Test
  186 + public void testUpdateDesiredState_online_pushShadow() {
  187 + Long deviceId = 2L;
  188 + IotDeviceDO device = buildDevice();
  189 + IotDeviceShadowDO shadow = buildShadow(device, 1L);
  190 + when(deviceService.validateDeviceExists(deviceId)).thenReturn(device);
  191 + when(deviceShadowRedisDAO.get(deviceId)).thenReturn(null);
  192 + when(deviceShadowMapper.selectByDeviceId(deviceId)).thenReturn(shadow);
  193 + when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(deviceId), eq(1L))).thenReturn(1);
  194 + when(devicePropertyService.getDeviceServerId(deviceId)).thenReturn("server-1");
  195 +
  196 + Map<String, Object> desired = new HashMap<>();
  197 + desired.put("PowerSwitch", 1);
  198 + service.updateDesiredState(deviceId, desired);
  199 +
  200 + ArgumentCaptor<IotDeviceMessage> captor = ArgumentCaptor.forClass(IotDeviceMessage.class);
  201 + verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1"));
  202 + IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams();
  203 + assertEquals("get_reply", reply.getMethod());
  204 + assertEquals(2L, reply.getVersion());
  205 + // getOrCreateShadow + CAS 各一次 SELECT，推送不再额外查库
  206 + verify(deviceShadowMapper, times(2)).selectByDeviceId(deviceId);
  207 + verify(deviceShadowRedisDAO).delete(deviceId);
  208 + }
  209 +
  210 + @Test
  211 + public void testUpdateDesiredState_offline_noPush() {
  212 + Long deviceId = 2L;
  213 + IotDeviceDO device = buildDevice();
  214 + IotDeviceShadowDO shadow = buildShadow(device, 0L);
  215 + when(deviceService.validateDeviceExists(deviceId)).thenReturn(device);
  216 + when(deviceShadowRedisDAO.get(deviceId)).thenReturn(null);
  217 + when(deviceShadowMapper.selectByDeviceId(deviceId)).thenReturn(shadow);
  218 + when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(deviceId), eq(0L))).thenReturn(1);
  219 + when(devicePropertyService.getDeviceServerId(deviceId)).thenReturn(null);
  220 +
  221 + Map<String, Object> desired = new HashMap<>();
  222 + desired.put("PowerSwitch", 1);
  223 + service.updateDesiredState(deviceId, desired);
  224 +
  225 + verify(deviceMessageService, never()).sendDeviceMessage(any(), any(), any());
  226 + }
  227 +
  228 + @Test
  229 + public void testHandleDeviceShadowUpdate_methodMissing_send401() {
  230 + IotDeviceDO device = buildDevice();
  231 + IotDeviceShadowDO shadow = buildShadow(device, 3L);
  232 + when(deviceShadowRedisDAO.get(device.getId())).thenReturn(null);
  233 + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(shadow);
  234 +
  235 + Map<String, Object> params = new HashMap<>();
  236 + params.put("state", Map.of("reported", Map.of("Temperature", 25)));
  237 + IotDeviceMessage message = IotDeviceMessage.requestOf(device.getId(), 1L, "server-1",
  238 + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), params);
  239 +
  240 + service.handleDeviceShadowUpdate(device, message);
  241 +
  242 + ArgumentCaptor<IotDeviceMessage> captor = ArgumentCaptor.forClass(IotDeviceMessage.class);
  243 + verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1"));
  244 + IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams();
  245 + @SuppressWarnings("unchecked")
  246 + 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"));
  250 + }
  251 +
  252 + @Test
  253 + public void testHandleDeviceShadowUpdate_update_retryReMergeFromLatest() {
  254 + IotDeviceDO device = buildDevice();
  255 + IotDeviceShadowDO staleShadow = buildShadow(device, 1L);
  256 + staleShadow.getReported().put("Temperature", 20);
  257 +
  258 + IotDeviceShadowDO latestShadow = buildShadow(device, 2L);
  259 + latestShadow.getReported().put("Temperature", 30);
  260 + latestShadow.getReported().put("Humidity", 50);
  261 +
  262 + when(deviceShadowRedisDAO.get(device.getId())).thenReturn(null);
  263 + AtomicInteger selectCount = new AtomicInteger(0);
  264 + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenAnswer(invocation -> {
  265 + // 前两次 select 返回 v=1 的快照，之后返回 v=2 的最新快照
  266 + return selectCount.incrementAndGet() <= 2 ? staleShadow : latestShadow;
  267 + });
  268 + when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), anyLong())).thenAnswer(invocation -> {
  269 + Long expectVersion = invocation.getArgument(2);
  270 + if (expectVersion == 1L) {
  271 + return 0;
  272 + }
  273 + if (expectVersion == 2L) {
  274 + return 1;
  275 + }
  276 + return 0;
  277 + });
  278 +
  279 + Map<String, Object> reported = new HashMap<>();
  280 + reported.put("Temperature", 25);
  281 + Map<String, Object> state = new HashMap<>();
  282 + state.put("reported", reported);
  283 + Map<String, Object> params = new HashMap<>();
  284 + params.put("method", "update");
  285 + params.put("state", state);
  286 + params.put("version", 2L);
  287 + IotDeviceMessage message = IotDeviceMessage.requestOf(device.getId(), 1L, "server-1",
  288 + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), params);
  289 +
  290 + service.handleDeviceShadowUpdate(device, message);
  291 +
  292 + ArgumentCaptor<IotDeviceShadowDO> updateCaptor = ArgumentCaptor.forClass(IotDeviceShadowDO.class);
  293 + verify(deviceShadowMapper, atLeastOnce()).updateByDeviceIdAndVersion(updateCaptor.capture(), eq(device.getId()), any());
  294 + IotDeviceShadowDO finalUpdate = updateCaptor.getAllValues().get(updateCaptor.getAllValues().size() - 1);
  295 + // 重试时应基于 v=2 的最新 reported 合并，保留 Humidity=50
  296 + assertEquals(25, finalUpdate.getReported().get("Temperature"));
  297 + assertEquals(50, finalUpdate.getReported().get("Humidity"));
  298 + assertEquals(3L, finalUpdate.getVersion());
  299 +
  300 + ArgumentCaptor<IotDeviceMessage> replyCaptor = ArgumentCaptor.forClass(IotDeviceMessage.class);
  301 + verify(deviceMessageService).sendDeviceMessage(replyCaptor.capture(), eq(device), eq("server-1"));
  302 + IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) replyCaptor.getValue().getParams();
  303 + assertEquals(3L, reply.getVersion());
  304 + }
  305 +
  306 + @Test
  307 + public void testUpdateStatus_whenShadowMissing_insertWithTargetStatus() {
  308 + Long deviceId = 2L;
  309 + IotDeviceDO device = buildDevice();
  310 + when(deviceService.validateDeviceExists(deviceId)).thenReturn(device);
  311 + when(deviceShadowRedisDAO.get(deviceId)).thenReturn(null);
  312 + when(deviceShadowMapper.selectByDeviceId(deviceId)).thenReturn(null);
  313 +
  314 + service.updateStatus(deviceId, CommonStatusEnum.DISABLE.getStatus());
  315 +
  316 + ArgumentCaptor<IotDeviceShadowDO> captor = ArgumentCaptor.forClass(IotDeviceShadowDO.class);
  317 + verify(deviceShadowMapper).insert(captor.capture());
  318 + assertEquals(CommonStatusEnum.DISABLE.getStatus(), captor.getValue().getStatus());
  319 + verify(deviceShadowRedisDAO, never()).delete(deviceId);
  320 + }
  321 +
  322 + @Test
  323 + public void testUpdateStatus_whenShadowMissing_duplicateKey_fallbackUpdateAndEvictCache() {
  324 + Long deviceId = 2L;
  325 + IotDeviceDO device = buildDevice();
  326 + when(deviceService.validateDeviceExists(deviceId)).thenReturn(device);
  327 + when(deviceShadowRedisDAO.get(deviceId)).thenReturn(null);
  328 + when(deviceShadowMapper.selectByDeviceId(deviceId)).thenReturn(null);
  329 + when(deviceShadowMapper.insert(any(IotDeviceShadowDO.class))).thenThrow(new DuplicateKeyException("duplicate"));
  330 + when(deviceShadowMapper.updateStatusByDeviceId(deviceId, CommonStatusEnum.DISABLE.getStatus())).thenReturn(1);
  331 +
  332 + service.updateStatus(deviceId, CommonStatusEnum.DISABLE.getStatus());
  333 +
  334 + verify(deviceShadowMapper).updateStatusByDeviceId(deviceId, CommonStatusEnum.DISABLE.getStatus());
  335 + verify(deviceShadowRedisDAO).delete(deviceId);
  336 + }
  337 +
  338 + @Test
  339 + public void testResetShadow_casConflict_throwException() {
  340 + Long deviceId = 2L;
  341 + IotDeviceDO device = buildDevice();
  342 + IotDeviceShadowDO shadow = buildShadow(device, 2L);
  343 + when(deviceService.validateDeviceExists(deviceId)).thenReturn(device);
  344 + when(deviceShadowRedisDAO.get(deviceId)).thenReturn(null);
  345 + when(deviceShadowMapper.selectByDeviceId(deviceId)).thenReturn(shadow);
  346 + when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(deviceId), eq(2L))).thenReturn(0);
  347 +
  348 + assertThrows(ServiceException.class, () -> service.resetShadow(deviceId));
  349 + }
  350 +
  351 + @Test
  352 + public void testHandleDeviceShadowUpdate_catchBlock_dbFailureStillSendsReply() {
  353 + IotDeviceDO device = buildDevice();
  354 + when(deviceShadowRedisDAO.get(device.getId())).thenReturn(null);
  355 + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenThrow(new RuntimeException("DB down"));
  356 +
  357 + Map<String, Object> params = new HashMap<>();
  358 + params.put("method", "get");
  359 + IotDeviceMessage message = IotDeviceMessage.requestOf(device.getId(), 1L, "server-1",
  360 + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), params);
  361 +
  362 + assertDoesNotThrow(() -> service.handleDeviceShadowUpdate(device, message));
  363 +
  364 + ArgumentCaptor<IotDeviceMessage> captor = ArgumentCaptor.forClass(IotDeviceMessage.class);
  365 + verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1"));
  366 + IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams();
  367 + assertEquals(0L, reply.getVersion());
  368 + @SuppressWarnings("unchecked")
  369 + 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"));
  373 + }
  374 +
  375 + @Test
  376 + public void testHandleDeviceShadowUpdate_casShadowNotFound_send500() {
  377 + IotDeviceDO device = buildDevice();
  378 + IotDeviceShadowDO shadow = buildShadow(device, 2L);
  379 + when(deviceShadowRedisDAO.get(device.getId())).thenReturn(null);
  380 + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(shadow, (IotDeviceShadowDO) null);
  381 +
  382 + Map<String, Object> reported = new HashMap<>();
  383 + reported.put("Temperature", 25);
  384 + Map<String, Object> state = new HashMap<>();
  385 + state.put("reported", reported);
  386 + Map<String, Object> params = new HashMap<>();
  387 + params.put("method", "update");
  388 + params.put("state", state);
  389 + params.put("version", 2L);
  390 + IotDeviceMessage message = IotDeviceMessage.requestOf(device.getId(), 1L, "server-1",
  391 + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), params);
  392 +
  393 + service.handleDeviceShadowUpdate(device, message);
  394 +
  395 + ArgumentCaptor<IotDeviceMessage> captor = ArgumentCaptor.forClass(IotDeviceMessage.class);
  396 + verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1"));
  397 + IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams();
  398 + @SuppressWarnings("unchecked")
  399 + 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"));
  403 + verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any());
  404 + }
  405 +
  406 + private IotDeviceDO buildDevice() {
  407 + return IotDeviceDO.builder()
  408 + .id(2L)
  409 + .productId(10L)
  410 + .productKey("testProduct")
  411 + .deviceName("testDevice")
  412 + .build();
  413 + }
  414 +
  415 + private IotDeviceShadowDO buildShadow(IotDeviceDO device, Long version) {
  416 + return buildShadow(device, version, CommonStatusEnum.ENABLE.getStatus());
  417 + }
  418 +
  419 + private IotDeviceShadowDO buildShadow(IotDeviceDO device, Long version, Integer status) {
  420 + return IotDeviceShadowDO.builder()
  421 + .id(100L)
  422 + .deviceId(device.getId())
  423 + .productId(device.getProductId())
  424 + .productKey(device.getProductKey())
  425 + .deviceName(device.getDeviceName())
  426 + .desired(new HashMap<>())
  427 + .reported(new HashMap<>())
  428 + .metadata(new HashMap<>())
  429 + .version(version)
  430 + .shadowTime(1000L)
  431 + .status(status)
  432 + .build();
  433 + }
  434 +
  435 +}
yudao-module-iot/yudao-module-iot-biz/src/test/resources/sql/clean.sql
  View file @84c46ad
1 DELETE FROM "iot_scene_rule"; 1 DELETE FROM "iot_scene_rule";
2 DELETE FROM "iot_product"; 2 DELETE FROM "iot_product";
3 DELETE FROM "iot_device"; 3 DELETE FROM "iot_device";
  4 +DELETE FROM "iot_device_shadow";
4 DELETE FROM "iot_thing_model"; 5 DELETE FROM "iot_thing_model";
5 DELETE FROM "iot_device_data"; 6 DELETE FROM "iot_device_data";
6 DELETE FROM "iot_alert_config"; 7 DELETE FROM "iot_alert_config";
yudao-module-iot/yudao-module-iot-biz/src/test/resources/sql/create_tables.sql
  View file @84c46ad
@@ -181,6 +181,29 @@ CREATE TABLE IF NOT EXISTS &quot;iot_ota_record&quot; ( @@ -181,6 +181,29 @@ CREATE TABLE IF NOT EXISTS &quot;iot_ota_record&quot; (
181 PRIMARY KEY ("id") 181 PRIMARY KEY ("id")
182 ) COMMENT 'IoT OTA 升级记录表'; 182 ) COMMENT 'IoT OTA 升级记录表';
183 183
  184 +CREATE TABLE IF NOT EXISTS "iot_device_shadow"
  185 +(
  186 + "id" bigint NOT NULL GENERATED BY DEFAULT AS IDENTITY,
  187 + "device_id" bigint NOT NULL,
  188 + "product_id" bigint NOT NULL,
  189 + "product_key" varchar(64) NOT NULL DEFAULT '',
  190 + "device_name" varchar(255) NOT NULL DEFAULT '',
  191 + "reported" varchar(10000) DEFAULT NULL,
  192 + "desired" varchar(10000) DEFAULT NULL,
  193 + "metadata" varchar(10000) DEFAULT NULL,
  194 + "version" bigint NOT NULL DEFAULT 0,
  195 + "shadow_time" bigint NOT NULL DEFAULT 0,
  196 + "status" tinyint NOT NULL DEFAULT 0,
  197 + "creator" varchar(64) DEFAULT '',
  198 + "create_time" timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
  199 + "updater" varchar(64) DEFAULT '',
  200 + "update_time" timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
  201 + "deleted" bit NOT NULL DEFAULT FALSE,
  202 + "tenant_id" bigint NOT NULL DEFAULT '0',
  203 + PRIMARY KEY ("id"),
  204 + UNIQUE ("device_id")
  205 +) COMMENT 'IoT 设备影子表';
  206 +
184 CREATE TABLE IF NOT EXISTS "iot_data_rule" ( 207 CREATE TABLE IF NOT EXISTS "iot_data_rule" (
185 "id" bigint NOT NULL GENERATED BY DEFAULT AS IDENTITY, 208 "id" bigint NOT NULL GENERATED BY DEFAULT AS IDENTITY,
186 "name" varchar(128) NOT NULL, 209 "name" varchar(128) NOT NULL,
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotDeviceMessageMethodEnum.java
  View file @84c46ad
@@ -67,6 +67,11 @@ public enum IotDeviceMessageMethodEnum implements ArrayValuable&lt;String&gt; { @@ -67,6 +67,11 @@ public enum IotDeviceMessageMethodEnum implements ArrayValuable&lt;String&gt; {
67 OTA_UPGRADE("thing.ota.upgrade", "OTA 固件信息推送", false), 67 OTA_UPGRADE("thing.ota.upgrade", "OTA 固件信息推送", false),
68 OTA_PROGRESS("thing.ota.progress", "OTA 升级进度上报", true), 68 OTA_PROGRESS("thing.ota.progress", "OTA 升级进度上报", true),
69 69
  70 + // ========== 设备影子 ==========
  71 +
  72 + SHADOW_UPDATE("thing.shadow.update", "设备影子更新", true),
  73 + SHADOW_GET("thing.shadow.get", "设备影子获取", false),
  74 +
70 ; 75 ;
71 76
72 public static final String[] ARRAYS = Arrays.stream(values()).map(IotDeviceMessageMethodEnum::getMethod) 77 public static final String[] ARRAYS = Arrays.stream(values()).map(IotDeviceMessageMethodEnum::getMethod)
@@ -77,7 +82,16 @@ public enum IotDeviceMessageMethodEnum implements ArrayValuable&lt;String&gt; { @@ -77,7 +82,16 @@ public enum IotDeviceMessageMethodEnum implements ArrayValuable&lt;String&gt; {
77 */ 82 */
78 public static final Set<String> REPLY_DISABLED = SetUtils.asSet( 83 public static final Set<String> REPLY_DISABLED = SetUtils.asSet(
79 STATE_UPDATE.getMethod(), 84 STATE_UPDATE.getMethod(),
80 - OTA_PROGRESS.getMethod() // 参考阿里云，OTA 升级进度上报，不进行回复 85 + OTA_PROGRESS.getMethod(), // 参考阿里云，OTA 升级进度上报，不进行回复
  86 + SHADOW_UPDATE.getMethod() // 设备影子由影子 Service 主动下发 thing.shadow.get
  87 + );
  88 +
  89 + /**
  90 + * 透传方法集合：设备侧使用原始线格式，不套 IotDeviceMessage 封套
  91 + */
  92 + public static final Set<String> PASSTHROUGH_METHODS = SetUtils.asSet(
  93 + SHADOW_UPDATE.getMethod(),
  94 + SHADOW_GET.getMethod()
81 ); 95 );
82 96
83 private final String method; 97 private final String method;
@@ -100,4 +114,8 @@ public enum IotDeviceMessageMethodEnum implements ArrayValuable&lt;String&gt; { @@ -100,4 +114,8 @@ public enum IotDeviceMessageMethodEnum implements ArrayValuable&lt;String&gt; {
100 return REPLY_DISABLED.contains(method); 114 return REPLY_DISABLED.contains(method);
101 } 115 }
102 116
  117 + public static boolean isPassthrough(String method) {
  118 + return PASSTHROUGH_METHODS.contains(method);
  119 + }
  120 +
103 } 121 }
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/shadow/IotDeviceShadowErrorCodeEnum.java 0 → 100644
  View file @84c46ad
  1 +package cn.iocoder.yudao.module.iot.core.enums.shadow;
  2 +
  3 +import lombok.AllArgsConstructor;
  4 +import lombok.Getter;
  5 +
  6 +/**
  7 + * IoT 设备影子设备侧错误码枚举
  8 + * <p>
  9 + * 用于回填 update_reply.payload.content，不走 CommonResult
  10 + *
  11 + * @author 芋道源码
  12 + */
  13 +@Getter
  14 +@AllArgsConstructor
  15 +public enum IotDeviceShadowErrorCodeEnum {
  16 +
  17 + JSON_FORMAT_ERROR(400, "JSON 格式错误"),
  18 + METHOD_MISSING(401, "缺少 method"),
  19 + STATE_MISSING(402, "缺少 state"),
  20 + REPORTED_MISSING(404, "缺少 reported"),
  21 + REPORTED_EMPTY(405, "reported 为空"),
  22 + INVALID_METHOD(406, "无效内层 method"),
  23 + SHADOW_EMPTY(407, "影子内容为空"),
  24 + PROPERTY_COUNT_EXCEEDED(408, "影子属性数超过限制"),
  25 + VERSION_CONFLICT(409, "影子版本冲突"),
  26 + SERVER_ERROR(500, "服务端异常");
  27 +
  28 + private final Integer code;
  29 +
  30 + private final String message;
  31 +
  32 +}
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/shadow/IotDeviceShadowMetadatas.java 0 → 100644
  View file @84c46ad
  1 +package cn.iocoder.yudao.module.iot.core.enums.shadow;
  2 +
  3 +import cn.hutool.core.map.MapUtil;
  4 +
  5 +import java.util.HashMap;
  6 +import java.util.Map;
  7 +import java.util.Set;
  8 +
  9 +import static cn.iocoder.yudao.module.iot.core.enums.shadow.IotDeviceShadowProtocolConstants.*;
  10 +
  11 +/**
  12 + * IoT 设备影子 metadata 读写工具
  13 + * <p>
  14 + * 结构：{ desired: { identifier: { timestamp } }, reported: { ... } }
  15 + *
  16 + * @author 芋道源码
  17 + */
  18 +public final class IotDeviceShadowMetadatas {
  19 +
  20 + private IotDeviceShadowMetadatas() {
  21 + }
  22 +
  23 + /**
  24 + * 获取 desired 下某属性的更新时间戳
  25 + *
  26 + * @param metadata 元数据
  27 + * @param identifier 属性标识符
  28 + * @return 时间戳（epoch 秒），不存在返回 null
  29 + */
  30 + public static Long getDesiredTimestamp(Map<String, Object> metadata, String identifier) {
  31 + return getTimestamp(metadata, STATE_DESIRED, identifier);
  32 + }
  33 +
  34 + /**
  35 + * 获取 reported 下某属性的更新时间戳
  36 + *
  37 + * @param metadata 元数据
  38 + * @param identifier 属性标识符
  39 + * @return 时间戳（epoch 秒），不存在返回 null
  40 + */
  41 + public static Long getReportedTimestamp(Map<String, Object> metadata, String identifier) {
  42 + return getTimestamp(metadata, STATE_REPORTED, identifier);
  43 + }
  44 +
  45 + /**
  46 + * 写入 desired 下多个属性的更新时间戳（原地修改 metadata）
  47 + *
  48 + * @param metadata 元数据
  49 + * @param keys 属性标识符集合
  50 + * @param timestamp 时间戳（epoch 秒）
  51 + */
  52 + public static void putDesiredTimestamps(Map<String, Object> metadata, Set<String> keys, long timestamp) {
  53 + putTimestamps(metadata, STATE_DESIRED, keys, timestamp);
  54 + }
  55 +
  56 + /**
  57 + * 写入 reported 下多个属性的更新时间戳（原地修改 metadata）
  58 + *
  59 + * @param metadata 元数据
  60 + * @param keys 属性标识符集合
  61 + * @param timestamp 时间戳（epoch 秒）
  62 + */
  63 + public static void putReportedTimestamps(Map<String, Object> metadata, Set<String> keys, long timestamp) {
  64 + putTimestamps(metadata, STATE_REPORTED, keys, timestamp);
  65 + }
  66 +
  67 + @SuppressWarnings("unchecked")
  68 + private static Long getTimestamp(Map<String, Object> metadata, String type, String identifier) {
  69 + if (metadata == null) {
  70 + return null;
  71 + }
  72 + Object typeMeta = metadata.get(type);
  73 + if (!(typeMeta instanceof Map)) {
  74 + return null;
  75 + }
  76 + Object keyMeta = ((Map<String, Object>) typeMeta).get(identifier);
  77 + if (!(keyMeta instanceof Map)) {
  78 + return null;
  79 + }
  80 + Object timestamp = ((Map<String, Object>) keyMeta).get(METADATA_TIMESTAMP);
  81 + return timestamp instanceof Number ? ((Number) timestamp).longValue() : null;
  82 + }
  83 +
  84 + @SuppressWarnings("unchecked")
  85 + private static void putTimestamps(Map<String, Object> metadata, String type, Set<String> keys, long timestamp) {
  86 + Map<String, Object> typeMeta = (Map<String, Object>) metadata.computeIfAbsent(type, k -> new HashMap<>());
  87 + for (String key : keys) {
  88 + typeMeta.put(key, MapUtil.of(METADATA_TIMESTAMP, timestamp));
  89 + }
  90 + }
  91 +
  92 +}
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/shadow/IotDeviceShadowProtocolConstants.java 0 → 100644
  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 0 → 100644
  View file @84c46ad
  1 +package cn.iocoder.yudao.module.iot.core.topic.shadow;
  2 +
  3 +import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
  4 +import lombok.AllArgsConstructor;
  5 +import lombok.Data;
  6 +import lombok.NoArgsConstructor;
  7 +
  8 +import java.util.HashMap;
  9 +import java.util.Map;
  10 +
  11 +import static cn.iocoder.yudao.module.iot.core.enums.shadow.IotDeviceShadowProtocolConstants.*;
  12 +
  13 +/**
  14 + * IoT 设备影子下行 Reply DTO
  15 + * <p>
  16 + * 用于 {@link IotDeviceMessageMethodEnum#SHADOW_GET} 消息的 params 参数（透传原始线 body）
  17 + *
  18 + * @author 芋道源码
  19 + */
  20 +@Data
  21 +@NoArgsConstructor
  22 +@AllArgsConstructor
  23 +public class IotDeviceShadowReplyDTO {
  24 +
  25 + /**
  26 + * 内层方法：get_reply / update_reply
  27 + */
  28 + private String method;
  29 +
  30 + /**
  31 + * 回复载荷
  32 + */
  33 + private Object payload;
  34 +
  35 + /**
  36 + * 影子版本号
  37 + */
  38 + private Long version;
  39 +
  40 + /**
  41 + * 时间戳（epoch 秒）
  42 + */
  43 + private Long timestamp;
  44 +
  45 + public static IotDeviceShadowReplyDTO getReplyOf(Map<String, Object> desired, Map<String, Object> reported,
  46 + Long version, Long timestamp) {
  47 + Map<String, Object> state = new HashMap<>();
  48 + state.put(STATE_DESIRED, desired != null ? desired : new HashMap<>());
  49 + 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);
  53 + return new IotDeviceShadowReplyDTO()
  54 + .setMethod(INNER_METHOD_GET_REPLY)
  55 + .setPayload(payload)
  56 + .setVersion(version)
  57 + .setTimestamp(timestamp);
  58 + }
  59 +
  60 + 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);
  64 + return new IotDeviceShadowReplyDTO()
  65 + .setMethod(INNER_METHOD_UPDATE_REPLY)
  66 + .setPayload(payload)
  67 + .setVersion(version)
  68 + .setTimestamp(timestamp);
  69 + }
  70 +
  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);
  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);
  79 + return new IotDeviceShadowReplyDTO()
  80 + .setMethod(INNER_METHOD_UPDATE_REPLY)
  81 + .setPayload(payload)
  82 + .setVersion(version)
  83 + .setTimestamp(timestamp);
  84 + }
  85 +
  86 +}
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/shadow/IotDeviceShadowUpdateReqDTO.java 0 → 100644
  View file @84c46ad
  1 +package cn.iocoder.yudao.module.iot.core.topic.shadow;
  2 +
  3 +import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
  4 +import lombok.AllArgsConstructor;
  5 +import lombok.Data;
  6 +import lombok.NoArgsConstructor;
  7 +
  8 +import java.util.Map;
  9 +
  10 +/**
  11 + * IoT 设备影子上行 Request DTO
  12 + * <p>
  13 + * 用于 {@link IotDeviceMessageMethodEnum#SHADOW_UPDATE} 消息的 params 参数（透传原始线 body）
  14 + *
  15 + * @author 芋道源码
  16 + */
  17 +@Data
  18 +@NoArgsConstructor
  19 +@AllArgsConstructor
  20 +public class IotDeviceShadowUpdateReqDTO {
  21 +
  22 + /**
  23 + * 内层方法：get / update
  24 + */
  25 + private String method;
  26 +
  27 + /**
  28 + * 影子状态
  29 + */
  30 + private ShadowState state;
  31 +
  32 + /**
  33 + * 设备已知版本号，可缺省；-1 表示清空影子
  34 + */
  35 + private Long version;
  36 +
  37 + @Data
  38 + @NoArgsConstructor
  39 + @AllArgsConstructor
  40 + public static class ShadowState {
  41 +
  42 + /**
  43 + * 上报状态
  44 + */
  45 + private Map<String, Object> reported;
  46 +
  47 + /**
  48 + * 期望状态
  49 + */
  50 + private Map<String, Object> desired;
  51 +
  52 + }
  53 +
  54 +}
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/upstream/IotMqttUpstreamHandler.java
  View file @84c46ad
@@ -2,6 +2,8 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.handler.upstream; @@ -2,6 +2,8 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.handler.upstream;
2 2
3 import cn.hutool.core.lang.Assert; 3 import cn.hutool.core.lang.Assert;
4 import cn.hutool.core.util.ArrayUtil; 4 import cn.hutool.core.util.ArrayUtil;
  5 +import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
  6 +import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
5 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; 7 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
6 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.manager.IotMqttConnectionManager; 8 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.manager.IotMqttConnectionManager;
7 import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService; 9 import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
@@ -9,6 +11,10 @@ import cn.iocoder.yudao.module.iot.gateway.util.IotMqttTopicUtils; @@ -9,6 +11,10 @@ import cn.iocoder.yudao.module.iot.gateway.util.IotMqttTopicUtils;
9 import io.vertx.mqtt.MqttEndpoint; 11 import io.vertx.mqtt.MqttEndpoint;
10 import lombok.extern.slf4j.Slf4j; 12 import lombok.extern.slf4j.Slf4j;
11 13
  14 +import java.util.Arrays;
  15 +import java.util.Map;
  16 +import java.util.stream.Collectors;
  17 +
12 18
13 /** 19 /**
14 * IoT 网关 MQTT 上行消息处理器：处理业务消息（属性上报、事件上报等） 20 * IoT 网关 MQTT 上行消息处理器：处理业务消息（属性上报、事件上报等）
@@ -59,13 +65,27 @@ public class IotMqttUpstreamHandler extends IotMqttAbstractHandler { @@ -59,13 +65,27 @@ public class IotMqttUpstreamHandler extends IotMqttAbstractHandler {
59 } 65 }
60 66
61 // 2.1 反序列化消息 67 // 2.1 反序列化消息
62 - IotDeviceMessage message = deviceMessageService.deserializeDeviceMessage(payload, productKey, deviceName);  
63 - if (message == null) {  
64 - log.warn("[handleBusinessRequest][消息解码失败，客户端 ID: {}，主题: {}]", clientId, topic);  
65 - return; 68 + String topicMethod = Arrays.stream(topicParts, 4, topicParts.length)
  69 + .collect(Collectors.joining("."));
  70 + IotDeviceMessage message;
  71 + if (IotDeviceMessageMethodEnum.isPassthrough(topicMethod)) {
  72 + // 影子透传：从 topic 解析 yudao method，原始 body 作为 params
  73 + Map<String, Object> params = JsonUtils.parseObject(payload, Map.class);
  74 + if (params == null) {
  75 + log.warn("[handleBusinessRequest][影子透传 payload 解析失败，客户端 ID: {}，主题: {}]", clientId, topic);
  76 + return;
  77 + }
  78 + String requestId = (String) params.get("id");
  79 + message = IotDeviceMessage.requestOf(requestId, topicMethod, params);
  80 + } else {
  81 + message = deviceMessageService.deserializeDeviceMessage(payload, productKey, deviceName);
  82 + if (message == null) {
  83 + log.warn("[handleBusinessRequest][消息解码失败，客户端 ID: {}，主题: {}]", clientId, topic);
  84 + return;
  85 + }
  86 + // 2.2 标准化回复消息的 method（MQTT 协议中，设备回复消息的 method 会携带 _reply 后缀）
  87 + IotMqttTopicUtils.normalizeReplyMethod(message);
66 } 88 }
67 - // 2.2 标准化回复消息的 method（MQTT 协议中，设备回复消息的 method 会携带 _reply 后缀）  
68 - IotMqttTopicUtils.normalizeReplyMethod(message);  
69 89
70 // 3. 处理业务消息 90 // 3. 处理业务消息
71 deviceMessageService.sendDeviceMessage(message, productKey, deviceName, serverId); 91 deviceMessageService.sendDeviceMessage(message, productKey, deviceName, serverId);
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceMessageServiceImpl.java
  View file @84c46ad
@@ -2,7 +2,9 @@ package cn.iocoder.yudao.module.iot.gateway.service.device.message; @@ -2,7 +2,9 @@ package cn.iocoder.yudao.module.iot.gateway.service.device.message;
2 2
3 import cn.hutool.core.lang.Assert; 3 import cn.hutool.core.lang.Assert;
4 import cn.hutool.core.util.StrUtil; 4 import cn.hutool.core.util.StrUtil;
  5 +import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
5 import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO; 6 import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO;
  7 +import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
6 import cn.iocoder.yudao.module.iot.core.enums.IotSerializeTypeEnum; 8 import cn.iocoder.yudao.module.iot.core.enums.IotSerializeTypeEnum;
7 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; 9 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
8 import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer; 10 import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer;
@@ -40,6 +42,10 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { @@ -40,6 +42,10 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
40 @Override 42 @Override
41 public byte[] serializeDeviceMessage(IotDeviceMessage message, 43 public byte[] serializeDeviceMessage(IotDeviceMessage message,
42 String productKey, String deviceName) { 44 String productKey, String deviceName) {
  45 + // 影子透传：直接序列化 params 为原始线 JSON，忽略产品 serializeType
  46 + if (IotDeviceMessageMethodEnum.isPassthrough(message.getMethod())) {
  47 + return JsonUtils.toJsonByte(message.getParams());
  48 + }
43 // 1.1 获取设备信息 49 // 1.1 获取设备信息
44 IotDeviceRespDTO device = deviceService.getDeviceFromCache(productKey, deviceName); 50 IotDeviceRespDTO device = deviceService.getDeviceFromCache(productKey, deviceName);
45 if (device == null) { 51 if (device == null) {
@@ -56,6 +62,10 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { @@ -56,6 +62,10 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
56 @Override 62 @Override
57 public byte[] serializeDeviceMessage(IotDeviceMessage message, 63 public byte[] serializeDeviceMessage(IotDeviceMessage message,
58 IotSerializeTypeEnum serializeType) { 64 IotSerializeTypeEnum serializeType) {
  65 + // 影子透传
  66 + if (IotDeviceMessageMethodEnum.isPassthrough(message.getMethod())) {
  67 + return JsonUtils.toJsonByte(message.getParams());
  68 + }
59 // 1. 获取序列化器 69 // 1. 获取序列化器
60 IotMessageSerializer serializer = messageSerializerManager.get(serializeType); 70 IotMessageSerializer serializer = messageSerializerManager.get(serializeType);
61 if (serializer == null) { 71 if (serializer == null) {
yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceShadowPassthroughTest.java 0 → 100644
  View file @84c46ad
  1 +package cn.iocoder.yudao.module.iot.gateway.service.device.message;
  2 +
  3 +import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
  4 +import cn.iocoder.yudao.framework.test.core.ut.BaseMockitoUnitTest;
  5 +import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
  6 +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  7 +import cn.iocoder.yudao.module.iot.core.topic.shadow.IotDeviceShadowReplyDTO;
  8 +import org.junit.jupiter.api.Test;
  9 +import org.mockito.InjectMocks;
  10 +
  11 +import java.util.HashMap;
  12 +import java.util.Map;
  13 +
  14 +import static org.junit.jupiter.api.Assertions.assertEquals;
  15 +import static org.junit.jupiter.api.Assertions.assertNull;
  16 +
  17 +/**
  18 + * 设备影子网关透传序列化单元测试
  19 + */
  20 +public class IotDeviceShadowPassthroughTest extends BaseMockitoUnitTest {
  21 +
  22 + @InjectMocks
  23 + private IotDeviceMessageServiceImpl deviceMessageService;
  24 +
  25 + @Test
  26 + public void testSerializeDeviceMessage_passthrough_noEnvelope() {
  27 + Map<String, Object> payload = new HashMap<>();
  28 + payload.put("method", "get_reply");
  29 + payload.put("version", 2L);
  30 + payload.put("timestamp", 1469564576L);
  31 + IotDeviceMessage message = IotDeviceMessage.requestOf(
  32 + IotDeviceMessageMethodEnum.SHADOW_GET.getMethod(), payload);
  33 +
  34 + byte[] bytes = deviceMessageService.serializeDeviceMessage(message, null);
  35 +
  36 + @SuppressWarnings("unchecked")
  37 + Map<String, Object> result = JsonUtils.parseObject(bytes, Map.class);
  38 + assertEquals("get_reply", result.get("method"));
  39 + assertEquals(2, ((Number) result.get("version")).intValue());
  40 + assertNull(result.get("id"));
  41 + assertNull(result.get("code"));
  42 + assertNull(result.get("requestId"));
  43 + }
  44 +
  45 + @Test
  46 + public void testDeserializeUpstreamShadow_fromTopicMethod() {
  47 + // 模拟上行透传：原始 body 不含 thing.shadow.update 封套 method
  48 + Map<String, Object> body = new HashMap<>();
  49 + body.put("method", "get");
  50 + IotDeviceMessage message = IotDeviceMessage.requestOf("req-1",
  51 + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), body);
  52 +
  53 + assertEquals(IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), message.getMethod());
  54 + assertEquals("get", ((Map<?, ?>) message.getParams()).get("method"));
  55 + }
  56 +
  57 + @Test
  58 + public void testSerializeUpdateReply_passthrough() {
  59 + IotDeviceShadowReplyDTO reply = IotDeviceShadowReplyDTO.updateReplyOf(3L, 1469564576L);
  60 + IotDeviceMessage message = IotDeviceMessage.requestOf(
  61 + IotDeviceMessageMethodEnum.SHADOW_GET.getMethod(), reply);
  62 +
  63 + byte[] bytes = deviceMessageService.serializeDeviceMessage(message, null);
  64 + @SuppressWarnings("unchecked")
  65 + Map<String, Object> result = JsonUtils.parseObject(bytes, Map.class);
  66 + assertEquals("update_reply", result.get("method"));
  67 + assertNull(result.get("id"));
  68 + }
  69 +
  70 +}