diff --git a/sql/mysql/ruoyi-vue-pro.sql b/sql/mysql/ruoyi-vue-pro.sql index 6657d42..11ac989 100644 --- a/sql/mysql/ruoyi-vue-pro.sql +++ b/sql/mysql/ruoyi-vue-pro.sql @@ -2799,7 +2799,7 @@ CREATE TABLE `system_menu` ( `update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', `deleted` bit(1) NOT NULL DEFAULT b'0' COMMENT '是否删除', PRIMARY KEY (`id`) USING BTREE -) ENGINE = InnoDB AUTO_INCREMENT = 13478 CHARACTER SET = utf8mb4 COLLATE = utf8mb4_unicode_ci COMMENT = '菜单权限表'; +) ENGINE = InnoDB AUTO_INCREMENT = 13480 CHARACTER SET = utf8mb4 COLLATE = utf8mb4_unicode_ci COMMENT = '菜单权限表'; -- ---------------------------- -- Records of system_menu @@ -5349,6 +5349,8 @@ INSERT INTO `system_menu` (`id`, `name`, `permission`, `type`, `sort`, `parent_i INSERT INTO `system_menu` (`id`, `name`, `permission`, `type`, `sort`, `parent_id`, `path`, `icon`, `component`, `component_name`, `status`, `visible`, `keep_alive`, `always_show`, `creator`, `create_time`, `updater`, `update_time`, `deleted`) VALUES (13475, '维修备件领用', 'eam:repair-order-act:maintenance', 3, 5, 13249, '', '', '', NULL, 0, b'1', b'1', b'1', '', '2026-09-26 20:50:45', '', '2026-09-26 20:50:45', b'0'); INSERT INTO `system_menu` (`id`, `name`, `permission`, `type`, `sort`, `parent_id`, `path`, `icon`, `component`, `component_name`, `status`, `visible`, `keep_alive`, `always_show`, `creator`, `create_time`, `updater`, `update_time`, `deleted`) VALUES (13476, '维修执行单取消', 'eam:repair-order-act:cancel', 3, 6, 13249, '', '', '', NULL, 0, b'1', b'1', b'1', '', '2026-09-26 20:50:45', '', '2026-09-26 20:50:45', b'0'); INSERT INTO `system_menu` (`id`, `name`, `permission`, `type`, `sort`, `parent_id`, `path`, `icon`, `component`, `component_name`, `status`, `visible`, `keep_alive`, `always_show`, `creator`, `create_time`, `updater`, `update_time`, `deleted`) VALUES (13477, '维修执行单修改', 'eam:repair-order-act:update', 3, 7, 13249, '', '', '', NULL, 0, b'1', b'1', b'1', '', '2026-09-26 20:50:45', '', '2026-09-26 20:50:45', b'0'); +INSERT INTO `system_menu` (`id`, `name`, `permission`, `type`, `sort`, `parent_id`, `path`, `icon`, `component`, `component_name`, `status`, `visible`, `keep_alive`, `always_show`, `creator`, `create_time`, `updater`, `update_time`, `deleted`) VALUES (13478, '设备影子查询', 'iot:device:shadow-query', 3, 13, 868, '', '', '', '', 0, b'1', b'1', b'1', '1', '2026-10-26 00:00:00', '1', '2026-10-26 00:00:00', b'0'); +INSERT INTO `system_menu` (`id`, `name`, `permission`, `type`, `sort`, `parent_id`, `path`, `icon`, `component`, `component_name`, `status`, `visible`, `keep_alive`, `always_show`, `creator`, `create_time`, `updater`, `update_time`, `deleted`) VALUES (13479, '设备影子更新', 'iot:device:shadow-update', 3, 14, 868, '', '', '', '', 0, b'1', b'1', b'1', '1', '2026-10-26 00:00:00', '1', '2026-10-26 00:00:00', b'0'); COMMIT; -- ---------------------------- diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/IotDeviceShadowController.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/IotDeviceShadowController.java new file mode 100644 index 0000000..37f5703 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/IotDeviceShadowController.java @@ -0,0 +1,113 @@ +package cn.iocoder.yudao.module.iot.controller.admin.device; + +import cn.hutool.core.lang.Assert; +import cn.hutool.core.map.MapUtil; +import cn.iocoder.yudao.framework.common.pojo.CommonResult; +import cn.iocoder.yudao.framework.common.util.object.BeanUtils; +import cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow.IotDeviceShadowPropertyRespVO; +import cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow.IotDeviceShadowRespVO; +import cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow.IotDeviceShadowUpdateDesiredReqVO; +import cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow.IotDeviceShadowUpdateStatusReqVO; +import cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils; +import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO; +import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceShadowDO; +import cn.iocoder.yudao.module.iot.dal.dataobject.thingmodel.IotThingModelDO; +import cn.iocoder.yudao.module.iot.dal.dataobject.thingmodel.model.ThingModelProperty; +import cn.iocoder.yudao.module.iot.enums.thingmodel.IotThingModelTypeEnum; +import cn.iocoder.yudao.module.iot.service.device.IotDeviceService; +import cn.iocoder.yudao.module.iot.service.device.shadow.IotDeviceShadowService; +import cn.iocoder.yudao.module.iot.service.thingmodel.IotThingModelService; +import io.swagger.v3.oas.annotations.Operation; +import io.swagger.v3.oas.annotations.Parameter; +import io.swagger.v3.oas.annotations.tags.Tag; +import jakarta.annotation.Resource; +import jakarta.validation.Valid; +import org.springframework.security.access.prepost.PreAuthorize; +import org.springframework.validation.annotation.Validated; +import org.springframework.web.bind.annotation.*; + +import java.util.List; +import java.util.Map; + +import static cn.iocoder.yudao.framework.common.pojo.CommonResult.success; +import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertList; + +/** + * 管理后台 - IoT 设备影子 Controller + * + * @author 芋道源码 + */ +@Tag(name = "管理后台 - IoT 设备影子") +@RestController +@RequestMapping("/iot/device/shadow") +@Validated +public class IotDeviceShadowController { + + @Resource + private IotDeviceShadowService deviceShadowService; + @Resource + private IotDeviceService deviceService; + @Resource + private IotThingModelService thingModelService; + + @GetMapping("/get") + @Operation(summary = "获取设备影子") + @Parameter(name = "deviceId", description = "设备编号", required = true) + @PreAuthorize("@ss.hasPermission('iot:device:shadow-query')") + public CommonResult getDeviceShadow(@RequestParam("deviceId") Long deviceId) { + IotDeviceDO device = deviceService.validateDeviceExists(deviceId); + IotDeviceShadowDO shadow = deviceShadowService.getOrCreateShadow(device); + + List thingModels = thingModelService.getThingModelListByProductIdAndType( + device.getProductId(), IotThingModelTypeEnum.PROPERTY.getType()); + Map desired = shadow.getDesired() != null ? shadow.getDesired() : MapUtil.empty(); + Map reported = shadow.getReported() != null ? shadow.getReported() : MapUtil.empty(); + Map metadata = shadow.getMetadata() != null ? shadow.getMetadata() : MapUtil.empty(); + // 拼接 VO + List properties = convertList(thingModels, thingModel -> { + ThingModelProperty prop = thingModel.getProperty(); + Assert.notNull(prop, "属性不能为空"); + return new IotDeviceShadowPropertyRespVO() + .setIdentifier(thingModel.getIdentifier()).setName(thingModel.getName()) + .setDataType(prop.getDataType()).setDataSpecs(prop.getDataSpecs()) + .setDataSpecsList(prop.getDataSpecsList()) + .setDesiredValue(desired.get(thingModel.getIdentifier())) + .setReportedValue(reported.get(thingModel.getIdentifier())) + .setDesiredTime(IotDeviceShadowProtocolUtils.getDesiredTimestamp(metadata, thingModel.getIdentifier())) + .setReportedTime(IotDeviceShadowProtocolUtils.getReportedTimestamp(metadata, thingModel.getIdentifier())); + }); + + IotDeviceShadowRespVO respVO = BeanUtils.toBean(shadow, IotDeviceShadowRespVO.class) + .setProperties(properties) + .setDesired(desired) + .setReported(reported) + .setMetadata(metadata); + return success(respVO); + } + + @PutMapping("/update-desired") + @Operation(summary = "更新设备影子期望状态") + @PreAuthorize("@ss.hasPermission('iot:device:shadow-update')") + public CommonResult updateDesiredState(@Valid @RequestBody IotDeviceShadowUpdateDesiredReqVO reqVO) { + deviceShadowService.updateDesiredState(reqVO.getDeviceId(), reqVO.getDesired()); + return success(true); + } + + @PutMapping("/update-status") + @Operation(summary = "更新设备影子开关状态") + @PreAuthorize("@ss.hasPermission('iot:device:shadow-update')") + public CommonResult updateShadowStatus(@Valid @RequestBody IotDeviceShadowUpdateStatusReqVO reqVO) { + deviceShadowService.updateStatus(reqVO.getDeviceId(), reqVO.getStatus()); + return success(true); + } + + @DeleteMapping("/reset") + @Operation(summary = "重置设备影子") + @Parameter(name = "deviceId", description = "设备编号", required = true) + @PreAuthorize("@ss.hasPermission('iot:device:shadow-update')") + public CommonResult resetShadow(@RequestParam("deviceId") Long deviceId) { + deviceShadowService.resetShadow(deviceId); + return success(true); + } + +} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/vo/shadow/IotDeviceShadowPropertyRespVO.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/vo/shadow/IotDeviceShadowPropertyRespVO.java new file mode 100644 index 0000000..323a6d4 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/vo/shadow/IotDeviceShadowPropertyRespVO.java @@ -0,0 +1,45 @@ +package cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow; + +import cn.iocoder.yudao.module.iot.dal.dataobject.thingmodel.model.dataType.ThingModelDataSpecs; +import io.swagger.v3.oas.annotations.media.Schema; +import lombok.Data; + +import java.util.List; + +/** + * 管理后台 - IoT 设备影子属性对比 Response VO + * + * @author 芋道源码 + */ +@Schema(description = "管理后台 - IoT 设备影子属性对比 Response VO") +@Data +public class IotDeviceShadowPropertyRespVO { + + @Schema(description = "标识符", example = "Temperature") + private String identifier; + + @Schema(description = "属性名称", example = "温度") + private String name; + + @Schema(description = "数据类型", example = "int") + private String dataType; + + @Schema(description = "数据定义") + private ThingModelDataSpecs dataSpecs; + + @Schema(description = "数据定义列表") + private List dataSpecsList; + + @Schema(description = "期望值") + private Object desiredValue; + + @Schema(description = "上报值") + private Object reportedValue; + + @Schema(description = "期望更新时间（epoch 秒）") + private Long desiredTime; + + @Schema(description = "上报更新时间（epoch 秒）") + private Long reportedTime; + +} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/vo/shadow/IotDeviceShadowRespVO.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/vo/shadow/IotDeviceShadowRespVO.java new file mode 100644 index 0000000..e712a75 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/vo/shadow/IotDeviceShadowRespVO.java @@ -0,0 +1,39 @@ +package cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow; + +import io.swagger.v3.oas.annotations.media.Schema; +import lombok.Data; + +import java.util.List; +import java.util.Map; + +/** + * 管理后台 - IoT 设备影子 Response VO + * + * @author 芋道源码 + */ +@Schema(description = "管理后台 - IoT 设备影子 Response VO") +@Data +public class IotDeviceShadowRespVO { + + @Schema(description = "版本号", example = "3") + private Long version; + + @Schema(description = "最后更新时间（epoch 秒）", example = "1469564576") + private Long shadowTime; + + @Schema(description = "是否启用", example = "0") + private Integer status; + + @Schema(description = "属性对比列表") + private List properties; + + @Schema(description = "期望状态") + private Map desired; + + @Schema(description = "上报状态") + private Map reported; + + @Schema(description = "元数据") + private Map metadata; + +} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/vo/shadow/IotDeviceShadowUpdateDesiredReqVO.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/vo/shadow/IotDeviceShadowUpdateDesiredReqVO.java new file mode 100644 index 0000000..814b0f3 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/vo/shadow/IotDeviceShadowUpdateDesiredReqVO.java @@ -0,0 +1,26 @@ +package cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow; + +import io.swagger.v3.oas.annotations.media.Schema; +import jakarta.validation.constraints.NotNull; +import lombok.Data; + +import java.util.Map; + +/** + * 管理后台 - IoT 设备影子更新期望状态 Request VO + * + * @author 芋道源码 + */ +@Schema(description = "管理后台 - IoT 设备影子更新期望状态 Request VO") +@Data +public class IotDeviceShadowUpdateDesiredReqVO { + + @Schema(description = "设备编号", requiredMode = Schema.RequiredMode.REQUIRED, example = "1") + @NotNull(message = "设备编号不能为空") + private Long deviceId; + + @Schema(description = "期望状态", requiredMode = Schema.RequiredMode.REQUIRED) + @NotNull(message = "期望状态不能为空") + private Map desired; + +} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/vo/shadow/IotDeviceShadowUpdateStatusReqVO.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/vo/shadow/IotDeviceShadowUpdateStatusReqVO.java new file mode 100644 index 0000000..a1a822c --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/vo/shadow/IotDeviceShadowUpdateStatusReqVO.java @@ -0,0 +1,27 @@ +package cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow; + +import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum; +import cn.iocoder.yudao.framework.common.validation.InEnum; +import io.swagger.v3.oas.annotations.media.Schema; +import jakarta.validation.constraints.NotNull; +import lombok.Data; + +/** + * 管理后台 - IoT 设备影子更新状态 Request VO + * + * @author 芋道源码 + */ +@Schema(description = "管理后台 - IoT 设备影子更新状态 Request VO") +@Data +public class IotDeviceShadowUpdateStatusReqVO { + + @Schema(description = "设备编号", requiredMode = Schema.RequiredMode.REQUIRED, example = "1") + @NotNull(message = "设备编号不能为空") + private Long deviceId; + + @Schema(description = "状态", requiredMode = Schema.RequiredMode.REQUIRED, example = "0") + @NotNull(message = "状态不能为空") + @InEnum(CommonStatusEnum.class) + private Integer status; + +} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/device/IotDeviceShadowDO.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/device/IotDeviceShadowDO.java new file mode 100644 index 0000000..b89524c --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/device/IotDeviceShadowDO.java @@ -0,0 +1,83 @@ +package cn.iocoder.yudao.module.iot.dal.dataobject.device; + +import cn.iocoder.yudao.framework.tenant.core.db.TenantBaseDO; +import cn.iocoder.yudao.module.iot.dal.dataobject.product.IotProductDO; +import com.baomidou.mybatisplus.annotation.KeySequence; +import com.baomidou.mybatisplus.annotation.TableField; +import com.baomidou.mybatisplus.annotation.TableId; +import com.baomidou.mybatisplus.annotation.TableName; +import com.baomidou.mybatisplus.extension.handlers.JacksonTypeHandler; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +import java.util.Map; + +/** + * IoT 设备影子 DO + * + * @author 芋道源码 + */ +@TableName(value = "iot_device_shadow", autoResultMap = true) +@KeySequence("iot_device_shadow_seq") +@Data +@Builder +@NoArgsConstructor +@AllArgsConstructor +public class IotDeviceShadowDO extends TenantBaseDO { + + /** + * 主键 + */ + @TableId + private Long id; + /** + * 设备编号 + */ + private Long deviceId; + /** + * 产品编号 + *

+ * 关联 {@link IotProductDO#getId()} + */ + private Long productId; + /** + * 产品标识 + */ + private String productKey; + /** + * 设备名称 + */ + private String deviceName; + /** + * 上报状态 + */ + @TableField(typeHandler = JacksonTypeHandler.class) + private Map reported; + /** + * 期望状态 + */ + @TableField(typeHandler = JacksonTypeHandler.class) + private Map desired; + /** + * 元数据：{ desired: { key: { timestamp } }, reported: { ... } } + */ + @TableField(typeHandler = JacksonTypeHandler.class) + private Map metadata; + /** + * 版本号（乐观锁） + */ + private Long version; + /** + * 文档最后更新时间（epoch 秒） + */ + private Long shadowTime; + /** + * 状态 + *

+ * 枚举 {@link cn.iocoder.yudao.framework.common.enums.CommonStatusEnum} + */ + private Integer status; + +} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/device/IotDeviceShadowMapper.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/device/IotDeviceShadowMapper.java new file mode 100644 index 0000000..8280dac --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/device/IotDeviceShadowMapper.java @@ -0,0 +1,35 @@ +package cn.iocoder.yudao.module.iot.dal.mysql.device; + +import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX; +import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceShadowDO; +import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; +import org.apache.ibatis.annotations.Mapper; + +/** + * IoT 设备影子 Mapper + * + * @author 芋道源码 + */ +@Mapper +public interface IotDeviceShadowMapper extends BaseMapperX { + + default IotDeviceShadowDO selectByDeviceId(Long deviceId) { + return selectOne(IotDeviceShadowDO::getDeviceId, deviceId); + } + + default int updateByDeviceId(IotDeviceShadowDO update, Long deviceId) { + return update(update, new LambdaUpdateWrapper() + .eq(IotDeviceShadowDO::getDeviceId, deviceId)); + } + + default int updateByDeviceIdAndVersion(IotDeviceShadowDO update, Long deviceId, Long expectVersion) { + return update(update, new LambdaUpdateWrapper() + .eq(IotDeviceShadowDO::getDeviceId, deviceId) + .eq(IotDeviceShadowDO::getVersion, expectVersion)); + } + + default int deleteByDeviceId(Long deviceId) { + return delete(IotDeviceShadowDO::getDeviceId, deviceId); + } + +} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/RedisKeyConstants.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/RedisKeyConstants.java index 95d2102..8f7fe58 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/RedisKeyConstants.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/RedisKeyConstants.java @@ -19,6 +19,15 @@ public interface RedisKeyConstants { String DEVICE_PROPERTY = "iot:device_property:%d"; /** + * 设备影子数据缓存，采用 STRING 结构 + *

+ * 影子缓存：DB 为准，Redis 仅作读缓存（cache-aside），写后失效，过期时间 60 秒。 + * KEY 格式：iot:device_shadow:{deviceId} + * VALUE 数据类型：String(JSON) {@link cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceShadowDO} + */ + String DEVICE_SHADOW = "iot:device_shadow:%d"; + + /** * 设备的最后上报时间，采用 ZSET 结构 * * KEY 格式：{deviceId} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/device/DeviceShadowRedisDAO.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/device/DeviceShadowRedisDAO.java new file mode 100644 index 0000000..dd02ba8 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/device/DeviceShadowRedisDAO.java @@ -0,0 +1,50 @@ +package cn.iocoder.yudao.module.iot.dal.redis.device; + +import cn.iocoder.yudao.framework.common.util.json.JsonUtils; +import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceShadowDO; +import jakarta.annotation.Resource; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.stereotype.Repository; + +import java.time.Duration; + +import static cn.iocoder.yudao.module.iot.dal.redis.RedisKeyConstants.DEVICE_SHADOW; + +/** + * {@link IotDeviceShadowDO} 的 Redis DAO + *

+ * 影子缓存：DB 为准，Redis 仅作读缓存（cache-aside），写后失效，并设置较短的过期时间兜底。 + * + * @author 芋道源码 + */ +@Repository +public class DeviceShadowRedisDAO { + + /** + * 缓存过期时间：并发回填旧数据时，最多在该时间内读到旧影子 + */ + private static final Duration TIMEOUT = Duration.ofSeconds(60); + + @Resource + private StringRedisTemplate stringRedisTemplate; + + public IotDeviceShadowDO get(Long deviceId) { + String redisKey = formatKey(deviceId); + String json = stringRedisTemplate.opsForValue().get(redisKey); + return JsonUtils.parseObject(json, IotDeviceShadowDO.class); + } + + public void set(IotDeviceShadowDO shadow) { + String redisKey = formatKey(shadow.getDeviceId()); + stringRedisTemplate.opsForValue().set(redisKey, JsonUtils.toJsonString(shadow), TIMEOUT); + } + + public void delete(Long deviceId) { + stringRedisTemplate.delete(formatKey(deviceId)); + } + + private static String formatKey(Long deviceId) { + return String.format(DEVICE_SHADOW, deviceId); + } + +} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/ErrorCodeConstants.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/ErrorCodeConstants.java index 1afd0fb..cb8b9b6 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/ErrorCodeConstants.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/ErrorCodeConstants.java @@ -67,6 +67,11 @@ public interface ErrorCodeConstants { ErrorCode DEVICE_MODBUS_POINT_NOT_EXISTS = new ErrorCode(1_050_007_000, "设备 Modbus 点位配置不存在"); ErrorCode DEVICE_MODBUS_POINT_EXISTS = new ErrorCode(1_050_007_001, "设备 Modbus 点位配置已存在"); + // ========== 设备影子 1-050-007-100 ========== + ErrorCode DEVICE_SHADOW_NOT_ENABLED = new ErrorCode(1_050_007_100, "设备影子未启用"); + ErrorCode DEVICE_SHADOW_VERSION_CONFLICT = new ErrorCode(1_050_007_101, "设备影子版本冲突"); + ErrorCode DEVICE_SHADOW_PROPERTY_COUNT_EXCEEDED = new ErrorCode(1_050_007_102, "设备影子属性数超过限制"); + // ========== OTA 固件相关 1-050-008-000 ========== ErrorCode OTA_FIRMWARE_NOT_EXISTS = new ErrorCode(1_050_008_000, "固件信息不存在"); diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/IotDeviceServiceImpl.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/IotDeviceServiceImpl.java index f05776f..46dc9bc 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/IotDeviceServiceImpl.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/IotDeviceServiceImpl.java @@ -37,6 +37,7 @@ import cn.iocoder.yudao.module.iot.dal.mysql.device.IotDeviceMapper; import cn.iocoder.yudao.module.iot.dal.redis.RedisKeyConstants; import cn.iocoder.yudao.module.iot.enums.product.IotProductDeviceTypeEnum; import cn.iocoder.yudao.module.iot.service.device.message.IotDeviceMessageService; +import cn.iocoder.yudao.module.iot.service.device.shadow.IotDeviceShadowService; import cn.iocoder.yudao.module.iot.service.product.IotProductService; import jakarta.annotation.Resource; import jakarta.validation.ConstraintViolationException; @@ -81,6 +82,9 @@ public class IotDeviceServiceImpl implements IotDeviceService { @Resource @Lazy // 延迟加载，解决循环依赖 private IotDeviceMessageService deviceMessageService; + @Resource + @Lazy // 延迟加载，解决循环依赖 + private IotDeviceShadowService deviceShadowService; private IotDeviceServiceImpl getSelf() { return SpringUtil.getBean(getClass()); @@ -197,6 +201,7 @@ public class IotDeviceServiceImpl implements IotDeviceService { } @Override + @Transactional(rollbackFor = Exception.class) public void deleteDevice(Long id) { // 1.1 校验存在 IotDeviceDO device = validateDeviceExists(id); @@ -209,7 +214,10 @@ public class IotDeviceServiceImpl implements IotDeviceService { // 2. 删除设备 deviceMapper.deleteById(id); - // 3. 清空对应缓存 + // 3. 级联删除影子 + deviceShadowService.deleteByDeviceId(id); + + // 4. 清空对应缓存（事务提交后生效） deleteDeviceCache(device); } @@ -235,7 +243,10 @@ public class IotDeviceServiceImpl implements IotDeviceService { // 2. 删除设备 deviceMapper.deleteByIds(ids); - // 3. 清空对应缓存 + // 3. 级联删除影子 + ids.forEach(deviceShadowService::deleteByDeviceId); + + // 4. 清空对应缓存（事务提交后生效） deleteDeviceCache(devices); } diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/message/IotDeviceMessageService.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/message/IotDeviceMessageService.java index 5323cca..da093e7 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/message/IotDeviceMessageService.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/message/IotDeviceMessageService.java @@ -30,6 +30,16 @@ public interface IotDeviceMessageService { IotDeviceMessage sendDeviceMessage(IotDeviceMessage message, IotDeviceDO device); /** + * 发送设备消息（指定 serverId，用于上行回复等场景） + * + * @param message 消息 + * @param device 设备 + * @param serverId 网关 serverId + * @return 设备消息 + */ + IotDeviceMessage sendDeviceMessage(IotDeviceMessage message, IotDeviceDO device, String serverId); + + /** * 发送设备消息 * * @param message 消息（“codec（编解码）字段” 部分字段） diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/message/IotDeviceMessageServiceImpl.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/message/IotDeviceMessageServiceImpl.java index 77f113b..09c8488 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/message/IotDeviceMessageServiceImpl.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/message/IotDeviceMessageServiceImpl.java @@ -28,6 +28,7 @@ import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceMessageDO; import cn.iocoder.yudao.module.iot.dal.storage.IotDeviceMessageDAO; import cn.iocoder.yudao.module.iot.service.device.IotDeviceService; import cn.iocoder.yudao.module.iot.service.device.property.IotDevicePropertyService; +import cn.iocoder.yudao.module.iot.service.device.shadow.IotDeviceShadowService; import cn.iocoder.yudao.module.iot.service.ota.IotOtaTaskRecordService; import com.google.common.base.Objects; import jakarta.annotation.Resource; @@ -63,6 +64,9 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { @Resource @Lazy // 延迟加载，避免循环依赖 private IotOtaTaskRecordService otaTaskRecordService; + @Resource + @Lazy // 延迟加载，避免循环依赖 + private IotDeviceShadowService deviceShadowService; @Resource private IotDeviceMessageDAO deviceMessageDAO; @@ -110,7 +114,8 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { return sendDeviceMessage(message, device, null); } - private IotDeviceMessage sendDeviceMessage(IotDeviceMessage message, IotDeviceDO device, String serverId) { + @Override + public IotDeviceMessage sendDeviceMessage(IotDeviceMessage message, IotDeviceDO device, String serverId) { // 1. 补充信息 appendDeviceMessage(message, device); @@ -219,6 +224,12 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { return null; } + // 设备影子更新 + if (Objects.equal(message.getMethod(), IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod())) { + deviceShadowService.handleDeviceShadowUpdate(device, message); + return null; + } + // 添加拓扑关系 if (Objects.equal(message.getMethod(), IotDeviceMessageMethodEnum.TOPO_ADD.getMethod())) { return deviceService.handleTopoAddMessage(message, device); diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowService.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowService.java new file mode 100644 index 0000000..cfdd793 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowService.java @@ -0,0 +1,64 @@ +package cn.iocoder.yudao.module.iot.service.device.shadow; + +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; +import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO; +import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceShadowDO; + +import java.util.Map; + +/** + * IoT 设备影子 Service 接口 + * + * @author 芋道源码 + */ +public interface IotDeviceShadowService { + + /** + * 获取或创建设备影子 + *

+ * 优先读缓存，仅用于展示；写入、版本校验等，需以 DB 为准 + * + * @param device 设备 + * @return 设备影子 + */ + IotDeviceShadowDO getOrCreateShadow(IotDeviceDO device); + + /** + * 处理设备影子上行更新 + * + * @param device 设备 + * @param message 上行消息 + */ + void handleDeviceShadowUpdate(IotDeviceDO device, IotDeviceMessage message); + + /** + * 更新期望状态 + * + * @param deviceId 设备编号 + * @param desired 期望状态 + */ + void updateDesiredState(Long deviceId, Map desired); + + /** + * 重置影子 + * + * @param deviceId 设备编号 + */ + void resetShadow(Long deviceId); + + /** + * 更新影子开关状态 + * + * @param deviceId 设备编号 + * @param status 状态 + */ + void updateStatus(Long deviceId, Integer status); + + /** + * 删除设备影子（设备删除时级联） + * + * @param deviceId 设备编号 + */ + void deleteByDeviceId(Long deviceId); + +} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImpl.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImpl.java new file mode 100644 index 0000000..ed5cc37 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImpl.java @@ -0,0 +1,392 @@ +package cn.iocoder.yudao.module.iot.service.device.shadow; + +import cn.hutool.core.collection.CollUtil; +import cn.hutool.core.date.DateUtil; +import cn.hutool.core.lang.Assert; +import cn.hutool.core.util.ObjUtil; +import cn.hutool.core.util.StrUtil; +import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum; +import cn.iocoder.yudao.framework.common.exception.ServiceException; +import cn.iocoder.yudao.framework.common.util.json.JsonUtils; +import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum; +import cn.iocoder.yudao.module.iot.core.enums.shadow.IotDeviceShadowErrorCodeEnum; +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; +import cn.iocoder.yudao.module.iot.core.topic.shadow.IotDeviceShadowUpdateReqDTO; +import cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils; +import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO; +import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceShadowDO; +import cn.iocoder.yudao.module.iot.dal.mysql.device.IotDeviceShadowMapper; +import cn.iocoder.yudao.module.iot.dal.redis.device.DeviceShadowRedisDAO; +import cn.iocoder.yudao.module.iot.service.device.IotDeviceService; +import cn.iocoder.yudao.module.iot.service.device.message.IotDeviceMessageService; +import cn.iocoder.yudao.module.iot.service.device.property.IotDevicePropertyService; +import jakarta.annotation.Resource; +import lombok.extern.slf4j.Slf4j; +import org.springframework.context.annotation.Lazy; +import org.springframework.dao.DuplicateKeyException; +import org.springframework.stereotype.Service; +import org.springframework.transaction.support.TransactionSynchronization; +import org.springframework.transaction.support.TransactionSynchronizationManager; +import org.springframework.validation.annotation.Validated; + +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; + +import static cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants.SUCCESS; +import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception; +import static cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils.INNER_METHOD_GET; +import static cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils.INNER_METHOD_UPDATE; +import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.*; + +/** + * IoT 设备影子 Service 实现类 + *

+ * DB 为准：写入、版本校验、开关判断都读 DB；Redis 仅用于管理后台查询的读加速 + * + * @author 芋道源码 + */ +@Service +@Validated +@Slf4j +public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { + + /** + * desired、reported 各自的最大属性数 + */ + private static final int MAX_PROPERTY_COUNT = 128; + /** + * 乐观锁冲突时的最大尝试次数 + */ + private static final int OPTIMISTIC_LOCK_RETRY = 3; + + @Resource + private IotDeviceShadowMapper deviceShadowMapper; + @Resource + private DeviceShadowRedisDAO deviceShadowRedisDAO; + + @Resource + private IotDeviceService deviceService; + @Resource + private IotDevicePropertyService devicePropertyService; + @Resource + @Lazy // 延迟加载，解决循环依赖 + private IotDeviceMessageService deviceMessageService; + + @Override + public IotDeviceShadowDO getOrCreateShadow(IotDeviceDO device) { + // 1. 优先读缓存：仅用于展示，允许在缓存过期前读到旧影子 + IotDeviceShadowDO shadow = deviceShadowRedisDAO.get(device.getId()); + if (shadow != null) { + return shadow; + } + // 2. 读 DB（不存在则创建），并回填缓存 + shadow = getOrCreateDeviceShadowFromDB(device, CommonStatusEnum.ENABLE.getStatus()); + deviceShadowRedisDAO.set(shadow); + return shadow; + } + + @Override + public void handleDeviceShadowUpdate(IotDeviceDO device, IotDeviceMessage message) { + try { + handleDeviceShadowUpdate0(device, message); + } catch (Exception ex) { + log.error("[handleDeviceShadowUpdate][device({}) message({}) 处理异常]", device.getId(), message, ex); + sendShadowErrorReply(device, message, IotDeviceShadowErrorCodeEnum.SERVER_ERROR, 0L); + } + } + + private void handleDeviceShadowUpdate0(IotDeviceDO device, IotDeviceMessage message) { + // 1.1 解析请求：params 结构为 { method, state: { reported }, version } + IotDeviceShadowUpdateReqDTO reqDTO = JsonUtils.convertObject(message.getParams(), + IotDeviceShadowUpdateReqDTO.class); + if (reqDTO == null) { + sendShadowErrorReply(device, message, IotDeviceShadowErrorCodeEnum.JSON_FORMAT_ERROR, 0L); + return; + } + // 1.2 校验影子已启用：读 DB，避免缓存中的旧开关 + IotDeviceShadowDO shadow = getOrCreateDeviceShadowFromDB(device, CommonStatusEnum.ENABLE.getStatus()); + if (CommonStatusEnum.isDisable(shadow.getStatus())) { + log.info("[handleDeviceShadowUpdate0][device({}) 影子未启用，忽略上行]", device.getId()); + return; + } + + // 2. 内层 method 分发 + if (INNER_METHOD_GET.equals(reqDTO.getMethod())) { + sendShadowGetReply(device, message.getServerId(), message.getRequestId(), shadow); + return; + } + if (INNER_METHOD_UPDATE.equals(reqDTO.getMethod())) { + handleReportedUpdate(device, message, reqDTO, shadow); + return; + } + sendShadowErrorReply(device, message, StrUtil.isEmpty(reqDTO.getMethod()) + ? IotDeviceShadowErrorCodeEnum.METHOD_MISSING : IotDeviceShadowErrorCodeEnum.INVALID_METHOD, + shadow.getVersion()); + } + + private void handleReportedUpdate(IotDeviceDO device, IotDeviceMessage message, + IotDeviceShadowUpdateReqDTO reqDTO, IotDeviceShadowDO shadow) { + // 1. version 为 -1 时，清空影子，无需 state + if (Objects.equals(reqDTO.getVersion(), -1L)) { + if (!resetDeviceShadow0(device.getId())) { + sendShadowVersionConflictReply(device, message); + return; + } + sendShadowReply(device, message.getServerId(), message.getRequestId(), SUCCESS.getCode(), SUCCESS.getMsg(), + IotDeviceShadowProtocolUtils.buildUpdateData(0L, DateUtil.currentSeconds())); + return; + } + + // 2. 校验 reported + IotDeviceShadowUpdateReqDTO.ShadowState state = reqDTO.getState(); + if (state == null) { + sendShadowErrorReply(device, message, IotDeviceShadowErrorCodeEnum.STATE_MISSING, shadow.getVersion()); + return; + } + if (state.getReported() == null) { + sendShadowErrorReply(device, message, IotDeviceShadowErrorCodeEnum.REPORTED_MISSING, shadow.getVersion()); + return; + } + if (state.getReported().isEmpty()) { + sendShadowErrorReply(device, message, IotDeviceShadowErrorCodeEnum.REPORTED_EMPTY, shadow.getVersion()); + return; + } + + // 3. 合并更新 reported + IotDeviceShadowDO newShadow; + try { + newShadow = updateDeviceShadowState(device.getId(), false, state.getReported(), reqDTO.getVersion()); + } catch (ServiceException ex) { // 合并后的属性数超过限制 + sendShadowErrorReply(device, message, IotDeviceShadowErrorCodeEnum.PROPERTY_COUNT_EXCEEDED, + shadow.getVersion()); + return; + } + if (newShadow == null) { + sendShadowVersionConflictReply(device, message); + return; + } + sendShadowReply(device, message.getServerId(), message.getRequestId(), SUCCESS.getCode(), SUCCESS.getMsg(), + IotDeviceShadowProtocolUtils.buildUpdateData(newShadow.getVersion(), newShadow.getShadowTime())); + } + + @Override + public void updateDesiredState(Long deviceId, Map desired) { + // 1. 校验影子已启用：读 DB，避免缓存中的旧开关 + IotDeviceDO device = deviceService.validateDeviceExists(deviceId); + IotDeviceShadowDO shadow = getOrCreateDeviceShadowFromDB(device, CommonStatusEnum.ENABLE.getStatus()); + if (CommonStatusEnum.isDisable(shadow.getStatus())) { + throw exception(DEVICE_SHADOW_NOT_ENABLED); + } + if (CollUtil.isEmpty(desired)) { + return; + } + + // 2. 合并更新 desired + IotDeviceShadowDO newShadow = updateDeviceShadowState(deviceId, true, desired, null); + if (newShadow == null) { + throw exception(DEVICE_SHADOW_VERSION_CONFLICT); + } + + // 3. 设备在线时，主动推送全量影子 + String serverId = devicePropertyService.getDeviceServerId(deviceId); + if (StrUtil.isNotEmpty(serverId)) { + sendShadowGetReply(device, serverId, null, newShadow); + } + } + + @Override + public void resetShadow(Long deviceId) { + deviceService.validateDeviceExists(deviceId); + if (!resetDeviceShadow0(deviceId)) { + throw exception(DEVICE_SHADOW_VERSION_CONFLICT); + } + } + + @Override + public void updateStatus(Long deviceId, Integer status) { + // 1. 获得影子：不存在时，直接按目标状态创建 + IotDeviceDO device = deviceService.validateDeviceExists(deviceId); + IotDeviceShadowDO shadow = getOrCreateDeviceShadowFromDB(device, status); + + // 2. 更新状态 + if (ObjUtil.notEqual(shadow.getStatus(), status)) { + deviceShadowMapper.updateByDeviceId(new IotDeviceShadowDO().setStatus(status), deviceId); + } + deleteDeviceShadowCache(deviceId); + } + + @Override + public void deleteByDeviceId(Long deviceId) { + deviceShadowMapper.deleteByDeviceId(deviceId); + deleteDeviceShadowCache(deviceId); + } + + /** + * 从 DB 获得影子，不存在时创建 + * + * @param device 设备 + * @param status 创建时的状态 + * @return 影子 + */ + private IotDeviceShadowDO getOrCreateDeviceShadowFromDB(IotDeviceDO device, Integer status) { + IotDeviceShadowDO shadow = deviceShadowMapper.selectByDeviceId(device.getId()); + if (shadow != null) { + return shadow; + } + // lazy 创建：首次访问时，才插入空影子；并发插入时，以先插入的为准 + shadow = new IotDeviceShadowDO().setDeviceId(device.getId()).setProductId(device.getProductId()) + .setProductKey(device.getProductKey()).setDeviceName(device.getDeviceName()) + .setDesired(new HashMap<>()).setReported(new HashMap<>()).setMetadata(new HashMap<>()) + .setVersion(0L).setShadowTime(DateUtil.currentSeconds()).setStatus(status); + try { + deviceShadowMapper.insert(shadow); + return shadow; + } catch (DuplicateKeyException ex) { + return deviceShadowMapper.selectByDeviceId(device.getId()); + } + } + + /** + * 乐观锁合并更新 desired 或 reported + *

+ * 每次尝试都读 DB 最新影子重新合并，避免丢失并发更新 + * + * @param deviceId 设备编号 + * @param desired true 更新 desired；false 更新 reported + * @param values 待合并的属性 + * @param expectVersion 期望的当前版本；null 表示不校验 + * @return 更新后的影子；版本冲突时，返回 null + */ + private IotDeviceShadowDO updateDeviceShadowState(Long deviceId, boolean desired, Map values, + Long expectVersion) { + for (int i = 0; i < OPTIMISTIC_LOCK_RETRY; i++) { + // 1. 校验版本 + IotDeviceShadowDO latest = deviceShadowMapper.selectByDeviceId(deviceId); + Assert.notNull(latest, "设备影子({})不存在", deviceId); + if (expectVersion != null && ObjUtil.notEqual(expectVersion, latest.getVersion())) { + return null; + } + + // 2.1 合并属性，并校验属性数 + Map state = mergeMap(desired ? latest.getDesired() : latest.getReported(), values); + if (state.size() > MAX_PROPERTY_COUNT) { + throw exception(DEVICE_SHADOW_PROPERTY_COUNT_EXCEEDED); + } + // 2.2 记录属性的更新时间 + long now = DateUtil.currentSeconds(); + Map metadata = mergeMap(latest.getMetadata(), null); + if (desired) { + IotDeviceShadowProtocolUtils.putDesiredTimestamps(metadata, values.keySet(), now); + } else { + IotDeviceShadowProtocolUtils.putReportedTimestamps(metadata, values.keySet(), now); + } + + // 3. 按 version 更新：成功则返回；失败则重试 + IotDeviceShadowDO update = new IotDeviceShadowDO() + .setDesired(desired ? state : latest.getDesired()) + .setReported(desired ? latest.getReported() : state) + .setMetadata(metadata).setVersion(latest.getVersion() + 1).setShadowTime(now); + if (deviceShadowMapper.updateByDeviceIdAndVersion(update, deviceId, latest.getVersion()) > 0) { + deleteDeviceShadowCache(deviceId); + return update; + } + } + return null; + } + + /** + * 乐观锁清空影子：desired、reported、metadata 置空，version 归零 + * + * @param deviceId 设备编号 + * @return 是否成功；影子不存在时，无需清空，也返回 true；版本冲突重试耗尽时，返回 false + */ + private boolean resetDeviceShadow0(Long deviceId) { + for (int i = 0; i < OPTIMISTIC_LOCK_RETRY; i++) { + IotDeviceShadowDO latest = deviceShadowMapper.selectByDeviceId(deviceId); + if (latest == null) { + return true; + } + IotDeviceShadowDO update = new IotDeviceShadowDO() + .setDesired(new HashMap<>()).setReported(new HashMap<>()).setMetadata(new HashMap<>()) + .setVersion(0L).setShadowTime(DateUtil.currentSeconds()); + if (deviceShadowMapper.updateByDeviceIdAndVersion(update, deviceId, latest.getVersion()) > 0) { + deleteDeviceShadowCache(deviceId); + return true; + } + } + return false; + } + + /** + * 删除影子缓存：存在事务时，在提交后删除，避免并发查询回填提交前的旧影子 + * + * @param deviceId 设备编号 + */ + private void deleteDeviceShadowCache(Long deviceId) { + if (!TransactionSynchronizationManager.isSynchronizationActive()) { + deviceShadowRedisDAO.delete(deviceId); + return; + } + TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() { + + @Override + public void afterCommit() { + deviceShadowRedisDAO.delete(deviceId); + } + + }); + } + + private void sendShadowGetReply(IotDeviceDO device, String serverId, String requestId, IotDeviceShadowDO shadow) { + sendShadowReply(device, serverId, requestId, SUCCESS.getCode(), SUCCESS.getMsg(), + IotDeviceShadowProtocolUtils.buildGetData(shadow.getDesired(), shadow.getReported(), + shadow.getVersion(), shadow.getShadowTime())); + } + + /** + * 回复版本冲突：带上 DB 当前版本，便于设备重新同步 + */ + private void sendShadowVersionConflictReply(IotDeviceDO device, IotDeviceMessage message) { + IotDeviceShadowDO latest = deviceShadowMapper.selectByDeviceId(device.getId()); + sendShadowErrorReply(device, message, IotDeviceShadowErrorCodeEnum.VERSION_CONFLICT, + latest != null ? latest.getVersion() : 0L); + } + + /** + * 回复错误：data 仍带 version，便于设备重新同步 + */ + private void sendShadowErrorReply(IotDeviceDO device, IotDeviceMessage message, + IotDeviceShadowErrorCodeEnum errorCode, Long version) { + sendShadowReply(device, message.getServerId(), message.getRequestId(), + errorCode.getCode(), errorCode.getMessage(), + IotDeviceShadowProtocolUtils.buildUpdateData(version, DateUtil.currentSeconds())); + } + + /** + * 下发影子回复（标准 IotDeviceMessage 封套） + *

+ * method 使用 {@link IotDeviceMessageMethodEnum#SHADOW_UPDATE}，带 code 即为回复， + * 网关下发到 thing/shadow/update_reply Topic；发送失败只打印日志，不影响上行处理 + */ + private void sendShadowReply(IotDeviceDO device, String serverId, String requestId, + Integer code, String msg, Map data) { + IotDeviceMessage reply = IotDeviceMessage.replyOf(requestId, + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), data, code, msg); + reply.setDeviceId(device.getId()).setTenantId(device.getTenantId()).setServerId(serverId); + try { + deviceMessageService.sendDeviceMessage(reply, device, serverId); + } catch (Exception ex) { + log.error("[sendShadowReply][device({}) 下发影子回复失败]", device.getId(), ex); + } + } + + private static Map mergeMap(Map oldMap, Map newMap) { + Map result = oldMap != null ? new HashMap<>(oldMap) : new HashMap<>(); + if (CollUtil.isNotEmpty(newMap)) { + result.putAll(newMap); + } + return result; + } + +} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/device/IotDeviceServiceImplTest.java b/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/device/IotDeviceServiceImplTest.java new file mode 100644 index 0000000..ab217a8 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/device/IotDeviceServiceImplTest.java @@ -0,0 +1,79 @@ +package cn.iocoder.yudao.module.iot.service.device; + +import cn.iocoder.yudao.framework.test.core.ut.BaseDbUnitTest; +import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO; +import cn.iocoder.yudao.module.iot.dal.mysql.device.IotDeviceMapper; +import cn.iocoder.yudao.module.iot.enums.product.IotProductDeviceTypeEnum; +import cn.iocoder.yudao.module.iot.service.device.message.IotDeviceMessageService; +import cn.iocoder.yudao.module.iot.service.device.shadow.IotDeviceShadowService; +import cn.iocoder.yudao.module.iot.service.product.IotProductService; +import jakarta.annotation.Resource; +import org.junit.jupiter.api.Test; +import org.springframework.context.annotation.Import; +import org.springframework.test.context.bean.override.mockito.MockitoBean; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.verify; + +/** + * {@link IotDeviceServiceImpl} 的单元测试 + * + * @author 芋道源码 + */ +@Import(IotDeviceServiceImpl.class) +public class IotDeviceServiceImplTest extends BaseDbUnitTest { + + @Resource + private IotDeviceServiceImpl deviceService; + + @Resource + private IotDeviceMapper deviceMapper; + + @MockitoBean + private IotProductService productService; + @MockitoBean + private IotDeviceGroupService deviceGroupService; + @MockitoBean + private IotDeviceMessageService deviceMessageService; + @MockitoBean + private IotDeviceShadowService deviceShadowService; + + @Test + public void testDeleteDevice_success() { + // mock 数据 + IotDeviceDO dbDevice = buildDevice(); + deviceMapper.insert(dbDevice); + // 准备参数 + Long id = dbDevice.getId(); + + // 调用 + deviceService.deleteDevice(id); + // 断言 + assertNull(deviceMapper.selectById(id)); + verify(deviceShadowService).deleteByDeviceId(id); + } + + @Test + public void testDeleteDevice_shadowDeleteFailed() { + // mock 数据 + IotDeviceDO dbDevice = buildDevice(); + deviceMapper.insert(dbDevice); + // 准备参数 + Long id = dbDevice.getId(); + // mock 影子删除失败 + RuntimeException failure = new RuntimeException("shadow delete failed"); + doThrow(failure).when(deviceShadowService).deleteByDeviceId(id); + + // 调用，并断言异常 + assertSame(failure, assertThrows(RuntimeException.class, () -> deviceService.deleteDevice(id))); + // 断言：设备删除已回滚 + assertNotNull(deviceMapper.selectById(id)); + } + + private static IotDeviceDO buildDevice() { + return IotDeviceDO.builder().deviceName("testDevice").productId(10L).productKey("testProduct") + .deviceType(IotProductDeviceTypeEnum.DIRECT.getType()).build(); + } + +} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImplTest.java b/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImplTest.java new file mode 100644 index 0000000..003fc53 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImplTest.java @@ -0,0 +1,702 @@ +package cn.iocoder.yudao.module.iot.service.device.shadow; + +import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum; +import cn.iocoder.yudao.framework.test.core.ut.BaseMockitoUnitTest; +import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum; +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; +import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO; +import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceShadowDO; +import cn.iocoder.yudao.module.iot.dal.mysql.device.IotDeviceShadowMapper; +import cn.iocoder.yudao.module.iot.dal.redis.device.DeviceShadowRedisDAO; +import cn.iocoder.yudao.module.iot.service.device.IotDeviceService; +import cn.iocoder.yudao.module.iot.service.device.message.IotDeviceMessageService; +import cn.iocoder.yudao.module.iot.service.device.property.IotDevicePropertyService; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.springframework.dao.DuplicateKeyException; +import org.springframework.transaction.support.TransactionSynchronization; +import org.springframework.transaction.support.TransactionSynchronizationManager; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static cn.iocoder.yudao.framework.test.core.util.AssertUtils.assertServiceException; +import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.*; +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.*; +import static org.mockito.Mockito.*; + +/** + * {@link IotDeviceShadowServiceImpl} 的单元测试 + * + * @author 芋道源码 + */ +public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest { + + @InjectMocks + private IotDeviceShadowServiceImpl deviceShadowService; + + @Mock + private IotDeviceShadowMapper deviceShadowMapper; + @Mock + private DeviceShadowRedisDAO deviceShadowRedisDAO; + @Mock + private IotDeviceService deviceService; + @Mock + private IotDevicePropertyService devicePropertyService; + @Mock + private IotDeviceMessageService deviceMessageService; + + // ========== getOrCreateShadow ========== + + @Test + public void testGetOrCreateShadow_cacheHit() { + // mock 数据 + IotDeviceDO device = buildDevice(); + IotDeviceShadowDO cacheShadow = buildShadow(device, 2L); + // mock 方法 + when(deviceShadowRedisDAO.get(device.getId())).thenReturn(cacheShadow); + + // 调用 + IotDeviceShadowDO shadow = deviceShadowService.getOrCreateShadow(device); + // 断言 + assertSame(cacheShadow, shadow); + verifyNoInteractions(deviceShadowMapper); + } + + @Test + public void testGetOrCreateShadow_notExists() { + // 准备参数 + IotDeviceDO device = buildDevice(); + // mock 方法 + when(deviceShadowRedisDAO.get(device.getId())).thenReturn(null); + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(null); + + // 调用 + IotDeviceShadowDO shadow = deviceShadowService.getOrCreateShadow(device); + // 断言：插入启用的空影子，并回填缓存 + verify(deviceShadowMapper).insert(shadow); + assertEquals(device.getId(), shadow.getDeviceId()); + assertEquals(CommonStatusEnum.ENABLE.getStatus(), shadow.getStatus()); + assertEquals(0L, shadow.getVersion()); + assertTrue(shadow.getReported().isEmpty()); + verify(deviceShadowRedisDAO).set(shadow); + } + + // ========== handleDeviceShadowUpdate：get ========== + + @Test + public void testHandleDeviceShadowUpdate_get() { + // mock 数据 + IotDeviceDO device = buildDevice(); + IotDeviceShadowDO dbShadow = buildShadow(device, 2L); + dbShadow.getDesired().put("PowerSwitch", 1); + // mock 方法 + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(dbShadow); + // 准备参数 + IotDeviceMessage message = buildMessage(device, Map.of("method", "get")); + + // 调用 + deviceShadowService.handleDeviceShadowUpdate(device, message); + // 断言：读 DB 回复全量影子 + IotDeviceMessage reply = captureReply(device); + assertEquals(IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), reply.getMethod()); + assertEquals("req-1", reply.getRequestId()); + assertEquals(0, reply.getCode()); + assertEquals(2L, getReplyVersion(reply)); + assertEquals(Map.of("PowerSwitch", 1), getReplyState(reply).get("desired")); + verify(deviceShadowRedisDAO, never()).get(any()); + } + + @Test + public void testHandleDeviceShadowUpdate_disabledInDb() { + // mock 数据：缓存为启用的旧影子，DB 已关闭 + IotDeviceDO device = buildDevice(); + lenient().when(deviceShadowRedisDAO.get(device.getId())).thenReturn(buildShadow(device, 0L)); + when(deviceShadowMapper.selectByDeviceId(device.getId())) + .thenReturn(buildShadow(device, 0L, CommonStatusEnum.DISABLE.getStatus())); + // 准备参数 + IotDeviceMessage message = buildMessage(device, buildUpdateParams(Map.of("Temperature", 25), null)); + + // 调用 + deviceShadowService.handleDeviceShadowUpdate(device, message); + // 断言：以 DB 为准，忽略上行 + verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any()); + verify(deviceMessageService, never()).sendDeviceMessage(any(), any(), any()); + } + + // ========== handleDeviceShadowUpdate：update ========== + + @Test + public void testHandleDeviceShadowUpdate_update_success() { + // mock 数据 + IotDeviceDO device = buildDevice(); + IotDeviceShadowDO dbShadow = buildShadow(device, 1L); + dbShadow.getReported().put("Temperature", 20); + dbShadow.getDesired().put("PowerSwitch", 1); + // mock 方法 + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(dbShadow); + when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(1L))).thenReturn(1); + // 准备参数 + IotDeviceMessage message = buildMessage(device, + buildUpdateParams(Map.of("Temperature", 25, "Humidity", 60), 1L)); + + // 调用 + deviceShadowService.handleDeviceShadowUpdate(device, message); + // 断言：合并 reported，保留 desired，version + 1 + IotDeviceShadowDO update = captureUpdate(device, 1L); + assertEquals(Map.of("Temperature", 25, "Humidity", 60), update.getReported()); + assertEquals(Map.of("PowerSwitch", 1), update.getDesired()); + assertEquals(2L, update.getVersion()); + verify(deviceShadowRedisDAO).delete(device.getId()); + // 断言：回复新版本 + IotDeviceMessage reply = captureReply(device); + assertEquals(0, reply.getCode()); + assertEquals(2L, getReplyVersion(reply)); + } + + @Test + public void testHandleDeviceShadowUpdate_update_staleCacheUseDbVersion() { + // mock 数据：缓存为 v1 的旧影子，DB 已是 v2 + IotDeviceDO device = buildDevice(); + lenient().when(deviceShadowRedisDAO.get(device.getId())).thenReturn(buildShadow(device, 1L)); + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(buildShadow(device, 2L)); + when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(2L))).thenReturn(1); + // 准备参数：设备携带 DB 当前版本 v2 + IotDeviceMessage message = buildMessage(device, buildUpdateParams(Map.of("Temperature", 25), 2L)); + + // 调用 + deviceShadowService.handleDeviceShadowUpdate(device, message); + // 断言：按 DB 版本写入成功 + assertEquals(3L, captureUpdate(device, 2L).getVersion()); + assertEquals(0, captureReply(device).getCode()); + verify(deviceShadowRedisDAO, never()).get(any()); + } + + @Test + public void testHandleDeviceShadowUpdate_update_versionConflict() { + // mock 数据 + IotDeviceDO device = buildDevice(); + // mock 方法 + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(buildShadow(device, 5L)); + // 准备参数：设备携带旧版本 v3 + IotDeviceMessage message = buildMessage(device, buildUpdateParams(Map.of("Temperature", 25), 3L)); + + // 调用 + deviceShadowService.handleDeviceShadowUpdate(device, message); + // 断言：不写入，回复 409 及 DB 当前版本 + verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any()); + IotDeviceMessage reply = captureReply(device); + assertEquals(409, reply.getCode()); + assertEquals(5L, getReplyVersion(reply)); + } + + @Test + public void testHandleDeviceShadowUpdate_update_versionNull() { + // mock 数据 + IotDeviceDO device = buildDevice(); + // mock 方法 + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(buildShadow(device, 5L)); + when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(5L))).thenReturn(1); + // 准备参数：不传 version，强制更新 + IotDeviceMessage message = buildMessage(device, buildUpdateParams(Map.of("Temperature", 25), null)); + + // 调用 + deviceShadowService.handleDeviceShadowUpdate(device, message); + // 断言 + assertEquals(6L, captureUpdate(device, 5L).getVersion()); + } + + @Test + public void testHandleDeviceShadowUpdate_update_retryFromLatest() { + // mock 数据：前两次读到 v1，之后读到 v2 + IotDeviceDO device = buildDevice(); + IotDeviceShadowDO staleShadow = buildShadow(device, 1L); + staleShadow.getReported().put("Temperature", 20); + IotDeviceShadowDO latestShadow = buildShadow(device, 2L); + latestShadow.getReported().put("Temperature", 30); + latestShadow.getReported().put("Humidity", 50); + // mock 方法：v1 写入冲突，v2 写入成功 + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(staleShadow, staleShadow, latestShadow); + when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(1L))).thenReturn(0); + when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(2L))).thenReturn(1); + // 准备参数 + IotDeviceMessage message = buildMessage(device, buildUpdateParams(Map.of("Temperature", 25), null)); + + // 调用 + deviceShadowService.handleDeviceShadowUpdate(device, message); + // 断言：重试时基于 v2 合并，保留 Humidity + IotDeviceShadowDO update = captureUpdate(device, 2L); + assertEquals(Map.of("Temperature", 25, "Humidity", 50), update.getReported()); + assertEquals(3L, update.getVersion()); + assertEquals(3L, getReplyVersion(captureReply(device))); + } + + @Test + public void testHandleDeviceShadowUpdate_update_retryExhausted() { + // mock 数据 + IotDeviceDO device = buildDevice(); + // mock 方法：每次写入都冲突 + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(buildShadow(device, 4L)); + when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(4L))).thenReturn(0); + // 准备参数 + IotDeviceMessage message = buildMessage(device, buildUpdateParams(Map.of("Temperature", 25), null)); + + // 调用 + deviceShadowService.handleDeviceShadowUpdate(device, message); + // 断言：重试 3 次后，回复 409 + verify(deviceShadowMapper, times(3)).updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(4L)); + verify(deviceShadowRedisDAO, never()).delete(any()); + IotDeviceMessage reply = captureReply(device); + assertEquals(409, reply.getCode()); + assertEquals(4L, getReplyVersion(reply)); + } + + @Test + public void testHandleDeviceShadowUpdate_update_propertyCountExceeded() { + // mock 数据：已有 128 个属性 + IotDeviceDO device = buildDevice(); + IotDeviceShadowDO dbShadow = buildShadow(device, 1L); + for (int i = 0; i < 128; i++) { + dbShadow.getReported().put("p" + i, i); + } + // mock 方法 + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(dbShadow); + // 准备参数 + IotDeviceMessage message = buildMessage(device, buildUpdateParams(Map.of("Temperature", 25), 1L)); + + // 调用 + deviceShadowService.handleDeviceShadowUpdate(device, message); + // 断言：不写入，回复 408 + verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any()); + IotDeviceMessage reply = captureReply(device); + assertEquals(408, reply.getCode()); + assertEquals(1L, getReplyVersion(reply)); + } + + @Test + public void testHandleDeviceShadowUpdate_resetWithoutState() { + // mock 数据 + IotDeviceDO device = buildDevice(); + // mock 方法 + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(buildShadow(device, 3L)); + when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(3L))).thenReturn(1); + // 准备参数：version 为 -1，且不传 state + IotDeviceMessage message = buildMessage(device, buildResetParams()); + + // 调用 + deviceShadowService.handleDeviceShadowUpdate(device, message); + // 断言：按 DB 版本清空影子，version 归零 + IotDeviceShadowDO update = captureUpdate(device, 3L); + assertTrue(update.getDesired().isEmpty()); + assertTrue(update.getReported().isEmpty()); + assertTrue(update.getMetadata().isEmpty()); + assertEquals(0L, update.getVersion()); + verify(deviceShadowRedisDAO).delete(device.getId()); + // 断言：回复成功 + IotDeviceMessage reply = captureReply(device); + assertEquals(0, reply.getCode()); + assertEquals(0L, getReplyVersion(reply)); + } + + @Test + public void testHandleDeviceShadowUpdate_resetRetryExhausted() { + // mock 数据 + IotDeviceDO device = buildDevice(); + // mock 方法：每次清空都冲突 + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(buildShadow(device, 3L)); + when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(3L))).thenReturn(0); + // 准备参数 + IotDeviceMessage message = buildMessage(device, buildResetParams()); + + // 调用 + deviceShadowService.handleDeviceShadowUpdate(device, message); + // 断言：重试 3 次后，回复 409 及 DB 当前版本 + verify(deviceShadowMapper, times(3)).updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(3L)); + verify(deviceShadowRedisDAO, never()).delete(any()); + IotDeviceMessage reply = captureReply(device); + assertEquals(409, reply.getCode()); + assertEquals(3L, getReplyVersion(reply)); + } + + @Test + public void testHandleDeviceShadowUpdate_stateMissing() { + // mock 数据 + IotDeviceDO device = buildDevice(); + // mock 方法 + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(buildShadow(device, 3L)); + // 准备参数 + IotDeviceMessage message = buildMessage(device, Map.of("method", "update", "version", 3L)); + + // 调用 + deviceShadowService.handleDeviceShadowUpdate(device, message); + // 断言 + IotDeviceMessage reply = captureReply(device); + assertEquals(402, reply.getCode()); + assertEquals(3L, getReplyVersion(reply)); + } + + @Test + public void testHandleDeviceShadowUpdate_invalidParams() { + // 准备参数 + IotDeviceDO device = buildDevice(); + IotDeviceMessage message = buildMessage(device, null); + + // 调用 + deviceShadowService.handleDeviceShadowUpdate(device, message); + // 断言 + assertEquals(400, captureReply(device).getCode()); + } + + @Test + public void testHandleDeviceShadowUpdate_methodMissing() { + // mock 数据 + IotDeviceDO device = buildDevice(); + // mock 方法 + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(buildShadow(device, 3L)); + // 准备参数 + IotDeviceMessage message = buildMessage(device, Map.of("state", Map.of("reported", Map.of("Temperature", 25)))); + + // 调用 + deviceShadowService.handleDeviceShadowUpdate(device, message); + // 断言 + assertEquals(401, captureReply(device).getCode()); + } + + @Test + public void testHandleDeviceShadowUpdate_dbFailure() { + // mock 数据 + IotDeviceDO device = buildDevice(); + // mock 方法 + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenThrow(new RuntimeException("DB down")); + // 准备参数 + IotDeviceMessage message = buildMessage(device, Map.of("method", "get")); + + // 调用 + assertDoesNotThrow(() -> deviceShadowService.handleDeviceShadowUpdate(device, message)); + // 断言：仍回复 500 + IotDeviceMessage reply = captureReply(device); + assertEquals(500, reply.getCode()); + assertEquals(0L, getReplyVersion(reply)); + } + + @Test + public void testHandleDeviceShadowUpdate_shadowDeletedDuringUpdate() { + // mock 数据 + IotDeviceDO device = buildDevice(); + // mock 方法：校验时存在，更新时已被删除 + when(deviceShadowMapper.selectByDeviceId(device.getId())) + .thenReturn(buildShadow(device, 2L), (IotDeviceShadowDO) null); + // 准备参数 + IotDeviceMessage message = buildMessage(device, buildUpdateParams(Map.of("Temperature", 25), 2L)); + + // 调用 + deviceShadowService.handleDeviceShadowUpdate(device, message); + // 断言 + verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any()); + assertEquals(500, captureReply(device).getCode()); + } + + // ========== updateDesiredState ========== + + @Test + public void testUpdateDesiredState_online() { + // mock 数据：缓存为关闭的旧影子，DB 已启用 + IotDeviceDO device = buildDevice(); + IotDeviceShadowDO dbShadow = buildShadow(device, 1L); + dbShadow.getReported().put("Temperature", 20); + lenient().when(deviceShadowRedisDAO.get(device.getId())) + .thenReturn(buildShadow(device, 1L, CommonStatusEnum.DISABLE.getStatus())); + when(deviceService.validateDeviceExists(device.getId())).thenReturn(device); + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(dbShadow); + when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(1L))).thenReturn(1); + when(devicePropertyService.getDeviceServerId(device.getId())).thenReturn("server-1"); + // 准备参数：数值、结构体、数组保持原类型 + Map desired = Map.of("Temperature", 25, "Location", Map.of("lat", 31.2), + "Modes", List.of(1, 2)); + + // 调用 + deviceShadowService.updateDesiredState(device.getId(), desired); + // 断言：合并 desired，保留 reported + IotDeviceShadowDO update = captureUpdate(device, 1L); + assertEquals(desired, update.getDesired()); + assertEquals(Map.of("Temperature", 20), update.getReported()); + assertEquals(2L, update.getVersion()); + verify(deviceShadowRedisDAO).delete(device.getId()); + // 断言：推送全量影子，desired 类型不变 + IotDeviceMessage push = captureReply(device); + assertEquals(IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), push.getMethod()); + assertEquals(0, push.getCode()); + assertEquals(2L, getReplyVersion(push)); + assertEquals(desired, getReplyState(push).get("desired")); + assertEquals(Map.of("Temperature", 20), getReplyState(push).get("reported")); + } + + @Test + public void testUpdateDesiredState_offline() { + // mock 数据 + IotDeviceDO device = buildDevice(); + // mock 方法 + when(deviceService.validateDeviceExists(device.getId())).thenReturn(device); + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(buildShadow(device, 0L)); + when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(0L))).thenReturn(1); + when(devicePropertyService.getDeviceServerId(device.getId())).thenReturn(null); + + // 调用 + deviceShadowService.updateDesiredState(device.getId(), Map.of("PowerSwitch", 1)); + // 断言 + verify(deviceMessageService, never()).sendDeviceMessage(any(), any(), any()); + } + + @Test + public void testUpdateDesiredState_disabledInDb() { + // mock 数据：缓存为启用的旧影子，DB 已关闭 + IotDeviceDO device = buildDevice(); + lenient().when(deviceShadowRedisDAO.get(device.getId())).thenReturn(buildShadow(device, 0L)); + when(deviceService.validateDeviceExists(device.getId())).thenReturn(device); + when(deviceShadowMapper.selectByDeviceId(device.getId())) + .thenReturn(buildShadow(device, 0L, CommonStatusEnum.DISABLE.getStatus())); + + // 调用，并断言异常 + assertServiceException(() -> deviceShadowService.updateDesiredState(device.getId(), Map.of("PowerSwitch", 1)), + DEVICE_SHADOW_NOT_ENABLED); + verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any()); + } + + @Test + public void testUpdateDesiredState_propertyCountExceeded() { + // mock 数据：已有 128 个属性 + IotDeviceDO device = buildDevice(); + IotDeviceShadowDO dbShadow = buildShadow(device, 1L); + for (int i = 0; i < 128; i++) { + dbShadow.getDesired().put("p" + i, i); + } + // mock 方法 + when(deviceService.validateDeviceExists(device.getId())).thenReturn(device); + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(dbShadow); + + // 调用，并断言异常 + assertServiceException(() -> deviceShadowService.updateDesiredState(device.getId(), Map.of("PowerSwitch", 1)), + DEVICE_SHADOW_PROPERTY_COUNT_EXCEEDED); + verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any()); + } + + @Test + public void testUpdateDesiredState_retryExhausted() { + // mock 数据 + IotDeviceDO device = buildDevice(); + // mock 方法：每次写入都冲突 + when(deviceService.validateDeviceExists(device.getId())).thenReturn(device); + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(buildShadow(device, 1L)); + when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(1L))).thenReturn(0); + + // 调用，并断言异常 + assertServiceException(() -> deviceShadowService.updateDesiredState(device.getId(), Map.of("PowerSwitch", 1)), + DEVICE_SHADOW_VERSION_CONFLICT); + verify(deviceShadowMapper, times(3)).updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(1L)); + verify(deviceMessageService, never()).sendDeviceMessage(any(), any(), any()); + } + + // ========== resetShadow ========== + + @Test + public void testResetShadow() { + // mock 数据 + IotDeviceDO device = buildDevice(); + // mock 方法 + when(deviceService.validateDeviceExists(device.getId())).thenReturn(device); + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(buildShadow(device, 2L)); + when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(2L))).thenReturn(1); + + // 调用 + deviceShadowService.resetShadow(device.getId()); + // 断言 + IotDeviceShadowDO update = captureUpdate(device, 2L); + assertTrue(update.getDesired().isEmpty()); + assertTrue(update.getReported().isEmpty()); + assertEquals(0L, update.getVersion()); + verify(deviceShadowRedisDAO).delete(device.getId()); + } + + @Test + public void testResetShadow_notExists() { + // mock 数据 + IotDeviceDO device = buildDevice(); + // mock 方法 + when(deviceService.validateDeviceExists(device.getId())).thenReturn(device); + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(null); + + // 调用 + deviceShadowService.resetShadow(device.getId()); + // 断言：影子不存在，无需清空 + verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any()); + verify(deviceShadowRedisDAO, never()).delete(any()); + } + + @Test + public void testResetShadow_retryExhausted() { + // mock 数据 + IotDeviceDO device = buildDevice(); + // mock 方法：每次清空都冲突 + when(deviceService.validateDeviceExists(device.getId())).thenReturn(device); + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(buildShadow(device, 2L)); + when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(2L))).thenReturn(0); + + // 调用，并断言异常 + assertServiceException(() -> deviceShadowService.resetShadow(device.getId()), DEVICE_SHADOW_VERSION_CONFLICT); + verify(deviceShadowMapper, times(3)).updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(2L)); + verify(deviceShadowRedisDAO, never()).delete(any()); + } + + // ========== updateStatus ========== + + @Test + public void testUpdateStatus_shadowMissing() { + // mock 数据 + IotDeviceDO device = buildDevice(); + // mock 方法 + when(deviceService.validateDeviceExists(device.getId())).thenReturn(device); + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(null); + + // 调用 + deviceShadowService.updateStatus(device.getId(), CommonStatusEnum.DISABLE.getStatus()); + // 断言：直接按目标状态创建 + ArgumentCaptor insertCaptor = ArgumentCaptor.forClass(IotDeviceShadowDO.class); + verify(deviceShadowMapper).insert(insertCaptor.capture()); + assertEquals(CommonStatusEnum.DISABLE.getStatus(), insertCaptor.getValue().getStatus()); + verify(deviceShadowMapper, never()).updateByDeviceId(any(), any()); + } + + @Test + public void testUpdateStatus_duplicateKey() { + // mock 数据 + IotDeviceDO device = buildDevice(); + // mock 方法：并发创建，插入冲突后读到已启用的影子 + when(deviceService.validateDeviceExists(device.getId())).thenReturn(device); + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(null, buildShadow(device, 0L)); + when(deviceShadowMapper.insert(any(IotDeviceShadowDO.class))).thenThrow(new DuplicateKeyException("duplicate")); + + // 调用 + deviceShadowService.updateStatus(device.getId(), CommonStatusEnum.DISABLE.getStatus()); + // 断言：改为更新状态，并删除缓存 + ArgumentCaptor updateCaptor = ArgumentCaptor.forClass(IotDeviceShadowDO.class); + verify(deviceShadowMapper).updateByDeviceId(updateCaptor.capture(), eq(device.getId())); + assertEquals(CommonStatusEnum.DISABLE.getStatus(), updateCaptor.getValue().getStatus()); + assertNull(updateCaptor.getValue().getVersion()); + verify(deviceShadowRedisDAO).delete(device.getId()); + } + + @Test + public void testUpdateStatus_inTransaction() { + // mock 数据 + IotDeviceDO device = buildDevice(); + // mock 方法 + when(deviceService.validateDeviceExists(device.getId())).thenReturn(device); + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(buildShadow(device, 0L)); + // mock 事务 + TransactionSynchronizationManager.initSynchronization(); + try { + // 调用 + deviceShadowService.updateStatus(device.getId(), CommonStatusEnum.DISABLE.getStatus()); + // 断言：提交前不删除缓存 + verify(deviceShadowMapper).updateByDeviceId(any(), eq(device.getId())); + verify(deviceShadowRedisDAO, never()).delete(any()); + // 断言：提交后删除缓存 + TransactionSynchronizationManager.getSynchronizations().forEach(TransactionSynchronization::afterCommit); + verify(deviceShadowRedisDAO).delete(device.getId()); + } finally { + TransactionSynchronizationManager.clearSynchronization(); + } + } + + // ========== deleteByDeviceId ========== + + @Test + public void testDeleteByDeviceId_inTransaction() { + // 准备参数 + Long deviceId = 2L; + // mock 事务 + TransactionSynchronizationManager.initSynchronization(); + try { + // 调用 + deviceShadowService.deleteByDeviceId(deviceId); + // 断言：提交前不删除缓存 + verify(deviceShadowMapper).deleteByDeviceId(deviceId); + verify(deviceShadowRedisDAO, never()).delete(any()); + // 断言：回滚时不删除缓存 + TransactionSynchronizationManager.getSynchronizations() + .forEach(sync -> sync.afterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK)); + verify(deviceShadowRedisDAO, never()).delete(any()); + // 断言：提交后删除缓存 + TransactionSynchronizationManager.getSynchronizations().forEach(TransactionSynchronization::afterCommit); + verify(deviceShadowRedisDAO).delete(deviceId); + } finally { + TransactionSynchronizationManager.clearSynchronization(); + } + } + + // ========== 私有方法 ========== + + private IotDeviceDO buildDevice() { + return IotDeviceDO.builder().id(2L).productId(10L).productKey("testProduct").deviceName("testDevice").build(); + } + + private IotDeviceShadowDO buildShadow(IotDeviceDO device, Long version) { + return buildShadow(device, version, CommonStatusEnum.ENABLE.getStatus()); + } + + private IotDeviceShadowDO buildShadow(IotDeviceDO device, Long version, Integer status) { + return new IotDeviceShadowDO().setId(100L).setDeviceId(device.getId()).setProductId(device.getProductId()) + .setProductKey(device.getProductKey()).setDeviceName(device.getDeviceName()) + .setDesired(new HashMap<>()).setReported(new HashMap<>()).setMetadata(new HashMap<>()) + .setVersion(version).setShadowTime(1000L).setStatus(status); + } + + private static Map buildUpdateParams(Map reported, Long version) { + Map params = new HashMap<>(); + params.put("method", "update"); + params.put("state", Map.of("reported", reported)); + params.put("version", version); + return params; + } + + private static Map buildResetParams() { + Map params = new HashMap<>(); + params.put("method", "update"); + params.put("version", -1L); + return params; + } + + private static IotDeviceMessage buildMessage(IotDeviceDO device, Object params) { + return IotDeviceMessage.requestOf(device.getId(), 1L, "server-1", + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), params).setRequestId("req-1"); + } + + private IotDeviceShadowDO captureUpdate(IotDeviceDO device, Long expectVersion) { + ArgumentCaptor updateCaptor = ArgumentCaptor.forClass(IotDeviceShadowDO.class); + verify(deviceShadowMapper).updateByDeviceIdAndVersion(updateCaptor.capture(), eq(device.getId()), + eq(expectVersion)); + return updateCaptor.getValue(); + } + + private IotDeviceMessage captureReply(IotDeviceDO device) { + ArgumentCaptor replyCaptor = ArgumentCaptor.forClass(IotDeviceMessage.class); + verify(deviceMessageService).sendDeviceMessage(replyCaptor.capture(), eq(device), eq("server-1")); + return replyCaptor.getValue(); + } + + @SuppressWarnings("unchecked") + private static Long getReplyVersion(IotDeviceMessage reply) { + Object version = ((Map) reply.getData()).get("version"); + return version instanceof Number ? ((Number) version).longValue() : null; + } + + @SuppressWarnings("unchecked") + private static Map getReplyState(IotDeviceMessage reply) { + return (Map) ((Map) reply.getData()).get("state"); + } + +} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/test/resources/sql/clean.sql b/yudao-module-iot/yudao-module-iot-biz/src/test/resources/sql/clean.sql index 454a57f..eb632d8 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/test/resources/sql/clean.sql +++ b/yudao-module-iot/yudao-module-iot-biz/src/test/resources/sql/clean.sql @@ -1,6 +1,7 @@ DELETE FROM "iot_scene_rule"; DELETE FROM "iot_product"; DELETE FROM "iot_device"; +DELETE FROM "iot_device_shadow"; DELETE FROM "iot_thing_model"; DELETE FROM "iot_device_data"; DELETE FROM "iot_alert_config"; diff --git a/yudao-module-iot/yudao-module-iot-biz/src/test/resources/sql/create_tables.sql b/yudao-module-iot/yudao-module-iot-biz/src/test/resources/sql/create_tables.sql index b1cc3bd..7fe0e94 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/test/resources/sql/create_tables.sql +++ b/yudao-module-iot/yudao-module-iot-biz/src/test/resources/sql/create_tables.sql @@ -189,6 +189,29 @@ CREATE TABLE IF NOT EXISTS "iot_ota_record" ( PRIMARY KEY ("id") ) COMMENT 'IoT OTA 升级记录表'; +CREATE TABLE IF NOT EXISTS "iot_device_shadow" +( + "id" bigint NOT NULL GENERATED BY DEFAULT AS IDENTITY, + "device_id" bigint NOT NULL, + "product_id" bigint NOT NULL, + "product_key" varchar(64) NOT NULL DEFAULT '', + "device_name" varchar(255) NOT NULL DEFAULT '', + "reported" varchar(10000) DEFAULT NULL, + "desired" varchar(10000) DEFAULT NULL, + "metadata" varchar(10000) DEFAULT NULL, + "version" bigint NOT NULL DEFAULT 0, + "shadow_time" bigint NOT NULL DEFAULT 0, + "status" tinyint NOT NULL DEFAULT 0, + "creator" varchar(64) DEFAULT '', + "create_time" timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updater" varchar(64) DEFAULT '', + "update_time" timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP, + "deleted" bit NOT NULL DEFAULT FALSE, + "tenant_id" bigint NOT NULL DEFAULT '0', + PRIMARY KEY ("id"), + UNIQUE ("device_id") +) COMMENT 'IoT 设备影子表'; + CREATE TABLE IF NOT EXISTS "iot_data_rule" ( "id" bigint NOT NULL GENERATED BY DEFAULT AS IDENTITY, "name" varchar(128) NOT NULL, diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotDeviceMessageMethodEnum.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotDeviceMessageMethodEnum.java index 3b4495e..1e536ee 100644 --- a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotDeviceMessageMethodEnum.java +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotDeviceMessageMethodEnum.java @@ -67,6 +67,10 @@ public enum IotDeviceMessageMethodEnum implements ArrayValuable { OTA_UPGRADE("thing.ota.upgrade", "OTA 固件信息推送", false), OTA_PROGRESS("thing.ota.progress", "OTA 升级进度上报", true), + // ========== 设备影子 ========== + + SHADOW_UPDATE("thing.shadow.update", "设备影子更新", true), + ; public static final String[] ARRAYS = Arrays.stream(values()).map(IotDeviceMessageMethodEnum::getMethod) @@ -77,7 +81,8 @@ public enum IotDeviceMessageMethodEnum implements ArrayValuable { */ public static final Set REPLY_DISABLED = SetUtils.asSet( STATE_UPDATE.getMethod(), - OTA_PROGRESS.getMethod() // 参考阿里云，OTA 升级进度上报，不进行回复 + OTA_PROGRESS.getMethod(), // 参考阿里云，OTA 升级进度上报，不进行回复 + SHADOW_UPDATE.getMethod() // 设备影子由影子 Service 主动下发回复（thing/shadow/update_reply） ); private final String method; diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/shadow/IotDeviceShadowErrorCodeEnum.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/shadow/IotDeviceShadowErrorCodeEnum.java new file mode 100644 index 0000000..9bbce77 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/shadow/IotDeviceShadowErrorCodeEnum.java @@ -0,0 +1,32 @@ +package cn.iocoder.yudao.module.iot.core.enums.shadow; + +import lombok.AllArgsConstructor; +import lombok.Getter; + +/** + * IoT 设备影子回复的错误码枚举 + *

+ * 用于设备影子回复（Topic：thing/shadow/update_reply）的 code、msg； + * 管理后台接口的业务异常，使用 biz 模块的 ErrorCodeConstants + * + * @author 芋道源码 + */ +@Getter +@AllArgsConstructor +public enum IotDeviceShadowErrorCodeEnum { + + JSON_FORMAT_ERROR(400, "JSON 格式错误"), + METHOD_MISSING(401, "缺少 method"), + STATE_MISSING(402, "缺少 state"), + REPORTED_MISSING(404, "缺少 reported"), + REPORTED_EMPTY(405, "reported 为空"), + INVALID_METHOD(406, "无效内层 method"), + PROPERTY_COUNT_EXCEEDED(408, "影子属性数超过限制"), + VERSION_CONFLICT(409, "影子版本冲突"), + SERVER_ERROR(500, "服务端异常"); + + private final Integer code; + + private final String message; + +} diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/shadow/IotDeviceShadowUpdateReqDTO.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/shadow/IotDeviceShadowUpdateReqDTO.java new file mode 100644 index 0000000..b75edca --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/shadow/IotDeviceShadowUpdateReqDTO.java @@ -0,0 +1,49 @@ +package cn.iocoder.yudao.module.iot.core.topic.shadow; + +import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum; +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.NoArgsConstructor; + +import java.util.Map; + +/** + * IoT 设备影子上行 Request DTO + *

+ * 用于 {@link IotDeviceMessageMethodEnum#SHADOW_UPDATE} 标准封套的 params 参数 + * + * @author 芋道源码 + */ +@Data +@NoArgsConstructor +@AllArgsConstructor +public class IotDeviceShadowUpdateReqDTO { + + /** + * 内层方法：get / update + */ + private String method; + + /** + * 影子状态 + */ + private ShadowState state; + + /** + * 设备已知版本号，可缺省；-1 表示清空影子 + */ + private Long version; + + @Data + @NoArgsConstructor + @AllArgsConstructor + public static class ShadowState { + + /** + * 上报状态 + */ + private Map reported; + + } + +} diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceShadowProtocolUtils.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceShadowProtocolUtils.java new file mode 100644 index 0000000..9e9e9ad --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceShadowProtocolUtils.java @@ -0,0 +1,131 @@ +package cn.iocoder.yudao.module.iot.core.util; + +import cn.hutool.core.map.MapUtil; + +import java.util.HashMap; +import java.util.Map; +import java.util.Set; + +/** + * IoT 设备影子协议工具 + *

+ * 内层 method、state 键名、reply data 字段等与设备侧标准封套 params/data 对齐；并提供 metadata 读写。 + * + * @author 芋道源码 + */ +public final class IotDeviceShadowProtocolUtils { + + private IotDeviceShadowProtocolUtils() { + } + + // ========== 内层 method（上行 params.method） ========== + + public static final String INNER_METHOD_GET = "get"; + public static final String INNER_METHOD_UPDATE = "update"; + + // ========== state / metadata 键 ========== + + public static final String STATE_DESIRED = "desired"; + public static final String STATE_REPORTED = "reported"; + public static final String METADATA_TIMESTAMP = "timestamp"; + + // ========== reply data 键 ========== + + public static final String PAYLOAD_STATE = "state"; + public static final String PAYLOAD_VERSION = "version"; + + /** + * 构建 get 响应 data：{ state:{desired,reported}, version, timestamp } + */ + public static Map buildGetData(Map desired, Map reported, + Long version, Long timestamp) { + Map state = new HashMap<>(); + state.put(STATE_DESIRED, desired != null ? desired : new HashMap<>()); + state.put(STATE_REPORTED, reported != null ? reported : new HashMap<>()); + Map data = new HashMap<>(); + data.put(PAYLOAD_STATE, state); + data.put(PAYLOAD_VERSION, version); + data.put(METADATA_TIMESTAMP, timestamp); + return data; + } + + /** + * 构建 update 响应 / 错误 data：{ version, timestamp } + */ + public static Map buildUpdateData(Long version, Long timestamp) { + Map data = new HashMap<>(); + data.put(PAYLOAD_VERSION, version); + data.put(METADATA_TIMESTAMP, timestamp); + return data; + } + + /** + * 获取 desired 下某属性的更新时间戳 + * + * @param metadata 元数据 + * @param identifier 属性标识符 + * @return 时间戳（epoch 秒），不存在返回 null + */ + public static Long getDesiredTimestamp(Map metadata, String identifier) { + return getTimestamp(metadata, STATE_DESIRED, identifier); + } + + /** + * 获取 reported 下某属性的更新时间戳 + * + * @param metadata 元数据 + * @param identifier 属性标识符 + * @return 时间戳（epoch 秒），不存在返回 null + */ + public static Long getReportedTimestamp(Map metadata, String identifier) { + return getTimestamp(metadata, STATE_REPORTED, identifier); + } + + /** + * 写入 desired 下多个属性的更新时间戳（原地修改 metadata） + * + * @param metadata 元数据 + * @param keys 属性标识符集合 + * @param timestamp 时间戳（epoch 秒） + */ + public static void putDesiredTimestamps(Map metadata, Set keys, long timestamp) { + putTimestamps(metadata, STATE_DESIRED, keys, timestamp); + } + + /** + * 写入 reported 下多个属性的更新时间戳（原地修改 metadata） + * + * @param metadata 元数据 + * @param keys 属性标识符集合 + * @param timestamp 时间戳（epoch 秒） + */ + public static void putReportedTimestamps(Map metadata, Set keys, long timestamp) { + putTimestamps(metadata, STATE_REPORTED, keys, timestamp); + } + + @SuppressWarnings("unchecked") + private static Long getTimestamp(Map metadata, String type, String identifier) { + if (metadata == null) { + return null; + } + Object typeMeta = metadata.get(type); + if (!(typeMeta instanceof Map)) { + return null; + } + Object keyMeta = ((Map) typeMeta).get(identifier); + if (!(keyMeta instanceof Map)) { + return null; + } + Object timestamp = ((Map) keyMeta).get(METADATA_TIMESTAMP); + return timestamp instanceof Number ? ((Number) timestamp).longValue() : null; + } + + @SuppressWarnings("unchecked") + private static void putTimestamps(Map metadata, String type, Set keys, long timestamp) { + Map typeMeta = (Map) metadata.computeIfAbsent(type, k -> new HashMap<>()); + for (String key : keys) { + typeMeta.put(key, MapUtil.of(METADATA_TIMESTAMP, timestamp)); + } + } + +} diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/util/IotMqttTopicUtilsTest.java b/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/util/IotMqttTopicUtilsTest.java new file mode 100644 index 0000000..55dcf2a --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/util/IotMqttTopicUtilsTest.java @@ -0,0 +1,37 @@ +package cn.iocoder.yudao.module.iot.gateway.util; + +import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * {@link IotMqttTopicUtils} 的单元测试 + * + * @author 芋道源码 + */ +public class IotMqttTopicUtilsTest { + + @Test + public void testBuildTopicByMethod_shadowUpdateReply() { + // 准备参数 + String method = IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(); + + // 调用 + String topic = IotMqttTopicUtils.buildTopicByMethod(method, "pk", "dn", true); + // 断言 + assertEquals("/sys/pk/dn/thing/shadow/update_reply", topic); + } + + @Test + public void testBuildTopicByMethod_serviceInvokeReply() { + // 准备参数 + String method = IotDeviceMessageMethodEnum.SERVICE_INVOKE.getMethod(); + + // 调用 + String topic = IotMqttTopicUtils.buildTopicByMethod(method, "pk", "dn", true); + // 断言 + assertEquals("/sys/pk/dn/thing/service/invoke_reply", topic); + } + +}