diff --git a/sql/mysql/iot_device_shadow.sql b/sql/mysql/iot_device_shadow.sql new file mode 100644 index 0000000..0c899b1 --- /dev/null +++ b/sql/mysql/iot_device_shadow.sql @@ -0,0 +1,36 @@ +-- IoT 设备影子表 +CREATE TABLE IF NOT EXISTS `iot_device_shadow` +( + `id` bigint NOT NULL AUTO_INCREMENT COMMENT '编号', + `device_id` bigint NOT NULL COMMENT '设备编号', + `product_id` bigint NOT NULL COMMENT '产品编号', + `product_key` varchar(64) NOT NULL COMMENT '产品标识', + `device_name` varchar(255) NOT NULL COMMENT '设备名称', + `reported` json DEFAULT NULL COMMENT '上报状态', + `desired` json DEFAULT NULL COMMENT '期望状态', + `metadata` json DEFAULT NULL COMMENT '元数据', + `version` bigint NOT NULL DEFAULT 0 COMMENT '版本号', + `shadow_time` bigint NOT NULL DEFAULT 0 COMMENT '文档最后更新时间', + `status` tinyint NOT NULL DEFAULT 0 COMMENT '状态', + `creator` varchar(64) DEFAULT '' COMMENT '创建者', + `create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', + `updater` varchar(64) DEFAULT '' COMMENT '更新者', + `update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', + `deleted` bit(1) NOT NULL DEFAULT b'0' COMMENT '是否删除', + `tenant_id` bigint NOT NULL DEFAULT 0 COMMENT '租户编号', + PRIMARY KEY (`id`), + UNIQUE KEY `uk_device_id` (`device_id`) +) ENGINE = InnoDB + DEFAULT CHARSET = utf8mb4 COMMENT ='IoT 设备影子'; + +-- 菜单权限 +INSERT INTO `system_menu` (`id`, `name`, `permission`, `type`, `sort`, `parent_id`, `path`, `icon`, `component`, + `component_name`, `status`, `visible`, `keep_alive`, `always_show`, `creator`, `create_time`, + `updater`, `update_time`, `deleted`) +VALUES (4056, '设备影子查询', 'iot:device:shadow-query', 3, 13, 4008, '', '', '', '', 0, b'1', b'1', b'1', '1', NOW(), + '1', NOW(), b'0'); +INSERT INTO `system_menu` (`id`, `name`, `permission`, `type`, `sort`, `parent_id`, `path`, `icon`, `component`, + `component_name`, `status`, `visible`, `keep_alive`, `always_show`, `creator`, `create_time`, + `updater`, `update_time`, `deleted`) +VALUES (4057, '设备影子更新', 'iot:device:shadow-update', 3, 14, 4008, '', '', '', '', 0, b'1', b'1', b'1', '1', NOW(), + '1', NOW(), b'0'); diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/IotDeviceShadowController.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/IotDeviceShadowController.java new file mode 100644 index 0000000..cca5228 --- /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,110 @@ +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.module.iot.controller.admin.device.vo.shadow.IotDeviceShadowPropertyRespVO; +import cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow.IotDeviceShadowRespVO; +import cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow.IotDeviceShadowUpdateDesiredReqVO; +import cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow.IotDeviceShadowUpdateStatusReqVO; +import cn.iocoder.yudao.module.iot.core.enums.shadow.IotDeviceShadowMetadatas; +import cn.iocoder.yudao.module.iot.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; + +@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(); + 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(IotDeviceShadowMetadatas.getDesiredTimestamp(metadata, thingModel.getIdentifier())) + .setReportedTime(IotDeviceShadowMetadatas.getReportedTimestamp(metadata, thingModel.getIdentifier())); + }); + + return success(new IotDeviceShadowRespVO() + .setVersion(shadow.getVersion()) + .setShadowTime(shadow.getShadowTime()) + .setStatus(shadow.getStatus()) + .setProperties(properties) + .setDesired(desired) + .setReported(reported) + .setMetadata(metadata)); + } + + @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..da3f838 --- /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,40 @@ +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; + +@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..26acf9c --- /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,34 @@ +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; + +@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..1af9922 --- /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,21 @@ +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; + +@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..310002c --- /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,22 @@ +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; + +@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..86d02e9 --- /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,36 @@ +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 updateByDeviceIdAndVersion(IotDeviceShadowDO update, Long deviceId, Long expectVersion) { + return update(update, new LambdaUpdateWrapper() + .eq(IotDeviceShadowDO::getDeviceId, deviceId) + .eq(IotDeviceShadowDO::getVersion, expectVersion)); + } + + default int updateStatusByDeviceId(Long deviceId, Integer status) { + return update(new LambdaUpdateWrapper() + .set(IotDeviceShadowDO::getStatus, status) + .eq(IotDeviceShadowDO::getDeviceId, deviceId)); + } + + 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..405597c 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,14 @@ public interface RedisKeyConstants { String DEVICE_PROPERTY = "iot:device_property:%d"; /** + * 设备影子数据缓存，采用 STRING 结构 + *

+ * KEY 格式：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..11b8fcc --- /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,39 @@ +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 static cn.iocoder.yudao.module.iot.dal.redis.RedisKeyConstants.DEVICE_SHADOW; + +/** + * {@link IotDeviceShadowDO} 的 Redis DAO + */ +@Repository +public class DeviceShadowRedisDAO { + + @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)); + } + + 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 2829bdf..6f71a35 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 @@ -62,6 +62,10 @@ 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, "设备影子版本冲突"); + // ========== 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..5c609b5 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()); @@ -209,7 +213,10 @@ public class IotDeviceServiceImpl implements IotDeviceService { // 2. 删除设备 deviceMapper.deleteById(id); - // 3. 清空对应缓存 + // 3. 级联删除影子 + deviceShadowService.deleteByDeviceId(id); + + // 4. 清空对应缓存 deleteDeviceCache(device); } @@ -235,7 +242,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 e28f489..83b4192 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 @@ -37,6 +37,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 7bfef2a..86da0e8 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.tdengine.IotDeviceMessageMapper; 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.baomidou.mybatisplus.core.metadata.IPage; import com.baomidou.mybatisplus.extension.plugins.pagination.Page; @@ -65,6 +66,9 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { @Resource @Lazy // 延迟加载，避免循环依赖 private IotOtaTaskRecordService otaTaskRecordService; + @Resource + @Lazy // 延迟加载，避免循环依赖 + private IotDeviceShadowService deviceShadowService; @Resource private IotDeviceMessageMapper deviceMessageMapper; @@ -120,7 +124,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); @@ -229,6 +234,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..b406d0e --- /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,62 @@ +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 { + + /** + * 获取或创建设备影子 + * + * @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..21bb240 --- /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,445 @@ +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.util.StrUtil; +import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum; +import cn.iocoder.yudao.framework.common.util.json.JsonUtils; +import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum; +import cn.iocoder.yudao.module.iot.core.enums.shadow.IotDeviceShadowErrorCodeEnum; +import cn.iocoder.yudao.module.iot.core.enums.shadow.IotDeviceShadowMetadatas; +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; +import cn.iocoder.yudao.module.iot.core.topic.shadow.IotDeviceShadowReplyDTO; +import cn.iocoder.yudao.module.iot.core.topic.shadow.IotDeviceShadowUpdateReqDTO; +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.Getter; +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.annotation.Transactional; +import org.springframework.validation.annotation.Validated; + +import java.util.HashMap; +import java.util.Map; +import java.util.function.Function; + +import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception; +import static cn.iocoder.yudao.module.iot.core.enums.shadow.IotDeviceShadowProtocolConstants.INNER_METHOD_GET; +import static cn.iocoder.yudao.module.iot.core.enums.shadow.IotDeviceShadowProtocolConstants.INNER_METHOD_UPDATE; +import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.DEVICE_SHADOW_NOT_ENABLED; +import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.DEVICE_SHADOW_VERSION_CONFLICT; + +/** + * IoT 设备影子 Service 实现类 + * + * @author 芋道源码 + */ +@Service +@Validated +@Slf4j +public class IotDeviceShadowServiceImpl implements IotDeviceShadowService { + + 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) { + IotDeviceShadowDO shadow = getShadowFromCacheOrDb(device.getId()); + if (shadow != null) { + return shadow; + } + // 懒插入空文档 + IotDeviceShadowDO newShadow = buildEmptyShadow(device) + .status(CommonStatusEnum.ENABLE.getStatus()) + .build(); + try { + deviceShadowMapper.insert(newShadow); + return newShadow; + } catch (DuplicateKeyException ex) { + return deviceShadowMapper.selectByDeviceId(device.getId()); + } + } + + @Override + public void handleDeviceShadowUpdate(IotDeviceDO device, IotDeviceMessage message) { + try { + handleDeviceShadowUpdate0(device, message); + } catch (Exception ex) { + log.error("[handleDeviceShadowUpdate][device({}) message({}) 处理异常]", device.getId(), message, ex); + Long version = resolveShadowVersionSafely(device.getId()); + sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf(version, + IotDeviceShadowErrorCodeEnum.SERVER_ERROR.getCode(), + IotDeviceShadowErrorCodeEnum.SERVER_ERROR.getMessage(), DateUtil.currentSeconds())); + } + } + + private void handleDeviceShadowUpdate0(IotDeviceDO device, IotDeviceMessage message) { + // 1. 解析请求 + IotDeviceShadowUpdateReqDTO req = JsonUtils.convertObject(message.getParams(), IotDeviceShadowUpdateReqDTO.class); + if (req == null) { + sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf(0L, + IotDeviceShadowErrorCodeEnum.JSON_FORMAT_ERROR.getCode(), + IotDeviceShadowErrorCodeEnum.JSON_FORMAT_ERROR.getMessage(), DateUtil.currentSeconds())); + return; + } + if (StrUtil.isEmpty(req.getMethod())) { + sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf(resolveShadowVersionSafely(device.getId()), + IotDeviceShadowErrorCodeEnum.METHOD_MISSING.getCode(), + IotDeviceShadowErrorCodeEnum.METHOD_MISSING.getMessage(), DateUtil.currentSeconds())); + return; + } + + // 2. 获取影子并校验开关 + IotDeviceShadowDO shadow = getOrCreateShadow(device); + if (CommonStatusEnum.isDisable(shadow.getStatus())) { + log.info("[handleDeviceShadowUpdate0][device({}) 影子未启用，忽略上行]", device.getId()); + return; + } + + // 3. 内层 method 分发 + if (INNER_METHOD_GET.equals(req.getMethod())) { + sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.getReplyOf( + shadow.getDesired(), shadow.getReported(), shadow.getVersion(), shadow.getShadowTime())); + return; + } + if (INNER_METHOD_UPDATE.equals(req.getMethod())) { + handleReportedUpdate(device, message.getServerId(), req); + return; + } + sendShadowReplySafely(device, message.getServerId(), IotDeviceShadowReplyDTO.errorOf(shadow.getVersion(), + IotDeviceShadowErrorCodeEnum.INVALID_METHOD.getCode(), + IotDeviceShadowErrorCodeEnum.INVALID_METHOD.getMessage(), DateUtil.currentSeconds())); + } + + private void handleReportedUpdate(IotDeviceDO device, String serverId, IotDeviceShadowUpdateReqDTO req) { + Long deviceId = device.getId(); + + if (req.getState() == null) { + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(resolveShadowVersionSafely(deviceId), + IotDeviceShadowErrorCodeEnum.STATE_MISSING.getCode(), + IotDeviceShadowErrorCodeEnum.STATE_MISSING.getMessage(), DateUtil.currentSeconds())); + return; + } + if (req.getState().getReported() == null) { + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(resolveShadowVersionSafely(deviceId), + IotDeviceShadowErrorCodeEnum.REPORTED_MISSING.getCode(), + IotDeviceShadowErrorCodeEnum.REPORTED_MISSING.getMessage(), DateUtil.currentSeconds())); + return; + } + if (CollUtil.isEmpty(req.getState().getReported())) { + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(resolveShadowVersionSafely(deviceId), + IotDeviceShadowErrorCodeEnum.REPORTED_EMPTY.getCode(), + IotDeviceShadowErrorCodeEnum.REPORTED_EMPTY.getMessage(), DateUtil.currentSeconds())); + return; + } + + // version == -1：清空影子 + if (req.getVersion() != null && req.getVersion() == -1L) { + try { + resetShadowWithRetry(deviceId); + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.updateReplyOf(0L, DateUtil.currentSeconds())); + } catch (ShadowCasExhaustedException ex) { + handleCasExhaustedForDevice(device, serverId, ex); + } + return; + } + + try { + IotDeviceShadowDO committed = casUpdateShadow(deviceId, latest -> buildReportedUpdate(latest, req)); + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.updateReplyOf(committed.getVersion(), + committed.getShadowTime())); + } catch (ShadowUpdateRejectedException ex) { + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(ex.getVersion(), + ex.getErrorCode().getCode(), ex.getErrorCode().getMessage(), DateUtil.currentSeconds())); + } catch (ShadowCasExhaustedException ex) { + handleCasExhaustedForDevice(device, serverId, ex); + } + } + + @Override + public void updateDesiredState(Long deviceId, Map desired) { + IotDeviceDO device = deviceService.validateDeviceExists(deviceId); + IotDeviceShadowDO shadow = getOrCreateShadow(device); + if (CommonStatusEnum.isDisable(shadow.getStatus())) { + throw exception(DEVICE_SHADOW_NOT_ENABLED); + } + if (CollUtil.isEmpty(desired)) { + return; + } + + try { + IotDeviceShadowDO committed = casUpdateShadow(deviceId, latest -> buildDesiredUpdate(latest, desired)); + // 设备在线则主动推送全量影子（复用 CAS 返回值，避免二次查库） + String serverId = devicePropertyService.getDeviceServerId(deviceId); + if (StrUtil.isNotEmpty(serverId)) { + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.getReplyOf( + committed.getDesired(), committed.getReported(), + committed.getVersion(), committed.getShadowTime())); + } + } catch (ShadowCasExhaustedException ex) { + throw exception(DEVICE_SHADOW_VERSION_CONFLICT); + } + } + + @Override + public void resetShadow(Long deviceId) { + deviceService.validateDeviceExists(deviceId); + IotDeviceShadowDO shadow = getShadowFromCacheOrDb(deviceId); + if (shadow == null) { + return; + } + try { + resetShadowWithRetry(deviceId); + } catch (ShadowCasExhaustedException ex) { + throw exception(DEVICE_SHADOW_VERSION_CONFLICT); + } + } + + @Override + @Transactional(rollbackFor = Exception.class) + public void updateStatus(Long deviceId, Integer status) { + IotDeviceDO device = deviceService.validateDeviceExists(deviceId); + IotDeviceShadowDO shadow = getShadowFromCacheOrDb(deviceId); + if (shadow == null) { + // 影子行不存在时直接按目标 status 懒建，避免后续 getOrCreateShadow 始终以 ENABLE 插入 + IotDeviceShadowDO newShadow = buildEmptyShadow(device).status(status).build(); + try { + deviceShadowMapper.insert(newShadow); + } catch (DuplicateKeyException ex) { + int rows = deviceShadowMapper.updateStatusByDeviceId(deviceId, status); + if (rows == 0) { + throw exception(DEVICE_SHADOW_VERSION_CONFLICT); + } + deviceShadowRedisDAO.delete(deviceId); + } + } else { + int rows = deviceShadowMapper.updateStatusByDeviceId(deviceId, status); + if (rows == 0) { + throw exception(DEVICE_SHADOW_VERSION_CONFLICT); + } + deviceShadowRedisDAO.delete(deviceId); + } + } + + @Override + @Transactional(rollbackFor = Exception.class) + public void deleteByDeviceId(Long deviceId) { + deviceShadowMapper.deleteByDeviceId(deviceId); + deviceShadowRedisDAO.delete(deviceId); + } + + // ========== 内部方法 ========== + + private IotDeviceShadowDO buildReportedUpdate(IotDeviceShadowDO latest, IotDeviceShadowUpdateReqDTO req) { + if (req.getVersion() != null && req.getVersion() < latest.getVersion()) { + throw new ShadowUpdateRejectedException(latest.getVersion(), IotDeviceShadowErrorCodeEnum.VERSION_CONFLICT); + } + Map mergedReported = mergeMap(latest.getReported(), req.getState().getReported()); + if (mergedReported.size() > MAX_PROPERTY_COUNT) { + throw new ShadowUpdateRejectedException(latest.getVersion(), IotDeviceShadowErrorCodeEnum.PROPERTY_COUNT_EXCEEDED); + } + long now = DateUtil.currentSeconds(); + Map metadata = latest.getMetadata() != null ? new HashMap<>(latest.getMetadata()) : new HashMap<>(); + IotDeviceShadowMetadatas.putReportedTimestamps(metadata, req.getState().getReported().keySet(), now); + return IotDeviceShadowDO.builder() + .reported(mergedReported) + .metadata(metadata) + .version(latest.getVersion() + 1) + .shadowTime(now) + .build(); + } + + private IotDeviceShadowDO buildDesiredUpdate(IotDeviceShadowDO latest, Map desired) { + long now = DateUtil.currentSeconds(); + Map mergedDesired = mergeMap(latest.getDesired(), desired); + Map metadata = latest.getMetadata() != null ? new HashMap<>(latest.getMetadata()) : new HashMap<>(); + IotDeviceShadowMetadatas.putDesiredTimestamps(metadata, desired.keySet(), now); + return IotDeviceShadowDO.builder() + .desired(mergedDesired) + .metadata(metadata) + .version(latest.getVersion() + 1) + .shadowTime(now) + .build(); + } + + private void resetShadowWithRetry(Long deviceId) { + casUpdateShadow(deviceId, latest -> IotDeviceShadowDO.builder() + .desired(new HashMap<>()) + .reported(new HashMap<>()) + .metadata(new HashMap<>()) + .version(0L) + .shadowTime(DateUtil.currentSeconds()) + .build()); + } + + /** + * 乐观锁 CAS 更新：每次重试基于最新影子重新构建 update，避免 lost-update + * + * @return 写入成功后的 update DO（回填未更新字段，便于构建全量回复） + */ + private IotDeviceShadowDO casUpdateShadow(Long deviceId, Function updateBuilder) { + Long lastKnownVersion = 0L; + for (int i = 0; i < OPTIMISTIC_LOCK_RETRY; i++) { + // 首次尝试走缓存，重试时直读 DB 获取最新 version + IotDeviceShadowDO latest = i == 0 + ? getShadowFromCacheOrDb(deviceId) + : deviceShadowMapper.selectByDeviceId(deviceId); + if (latest == null) { + throw ShadowCasExhaustedException.shadowNotFound(); + } + lastKnownVersion = latest.getVersion(); + IotDeviceShadowDO update = updateBuilder.apply(latest); + int rows = deviceShadowMapper.updateByDeviceIdAndVersion(update, deviceId, latest.getVersion()); + if (rows > 0) { + deviceShadowRedisDAO.delete(deviceId); + enrichCommittedShadow(update, latest); + return update; + } + } + throw ShadowCasExhaustedException.versionConflict(lastKnownVersion); + } + + /** + * 将 CAS 成功写入的局部字段与 latest 快照合并，供全量 get_reply 使用 + */ + private static void enrichCommittedShadow(IotDeviceShadowDO committed, IotDeviceShadowDO latest) { + if (committed.getDesired() == null) { + committed.setDesired(latest.getDesired()); + } + if (committed.getReported() == null) { + committed.setReported(latest.getReported()); + } + if (committed.getMetadata() == null) { + committed.setMetadata(latest.getMetadata()); + } + } + + private void handleCasExhaustedForDevice(IotDeviceDO device, String serverId, ShadowCasExhaustedException ex) { + if (ex.isShadowNotFound()) { + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(0L, + IotDeviceShadowErrorCodeEnum.SERVER_ERROR.getCode(), + IotDeviceShadowErrorCodeEnum.SERVER_ERROR.getMessage(), DateUtil.currentSeconds())); + return; + } + sendVersionConflictReply(device, serverId, ex.getLastKnownVersion()); + } + + private void sendVersionConflictReply(IotDeviceDO device, String serverId, Long version) { + sendShadowReplySafely(device, serverId, IotDeviceShadowReplyDTO.errorOf(version, + IotDeviceShadowErrorCodeEnum.VERSION_CONFLICT.getCode(), + IotDeviceShadowErrorCodeEnum.VERSION_CONFLICT.getMessage(), DateUtil.currentSeconds())); + } + + private Long resolveShadowVersionSafely(Long deviceId) { + try { + IotDeviceShadowDO shadow = getShadowFromCacheOrDb(deviceId); + return shadow != null ? shadow.getVersion() : 0L; + } catch (Exception ex) { + log.warn("[resolveShadowVersionSafely][device({}) 读取 version 失败，使用保底 0]", deviceId, ex); + return 0L; + } + } + + private IotDeviceShadowDO.IotDeviceShadowDOBuilder buildEmptyShadow(IotDeviceDO device) { + return IotDeviceShadowDO.builder() + .deviceId(device.getId()) + .productId(device.getProductId()) + .productKey(device.getProductKey()) + .deviceName(device.getDeviceName()) + .reported(new HashMap<>()) + .desired(new HashMap<>()) + .metadata(new HashMap<>()) + .version(0L) + .shadowTime(DateUtil.currentSeconds()); + } + + private IotDeviceShadowDO getShadowFromCacheOrDb(Long deviceId) { + IotDeviceShadowDO shadow = deviceShadowRedisDAO.get(deviceId); + if (shadow != null) { + return shadow; + } + shadow = deviceShadowMapper.selectByDeviceId(deviceId); + if (shadow != null) { + deviceShadowRedisDAO.set(shadow); + } + return shadow; + } + + /** + * 下发影子回复，内部吞掉异常，保证上行处理链路不向上抛 + */ + private void sendShadowReplySafely(IotDeviceDO device, String serverId, IotDeviceShadowReplyDTO replyDTO) { + IotDeviceMessage shadowGetMsg = IotDeviceMessage.requestOf( + device.getId(), device.getTenantId(), serverId, + IotDeviceMessageMethodEnum.SHADOW_GET.getMethod(), replyDTO); + try { + deviceMessageService.sendDeviceMessage(shadowGetMsg, 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; + } + + @Getter + private static class ShadowUpdateRejectedException extends RuntimeException { + + private final Long version; + private final IotDeviceShadowErrorCodeEnum errorCode; + + private ShadowUpdateRejectedException(Long version, IotDeviceShadowErrorCodeEnum errorCode) { + super(errorCode.getMessage()); + this.version = version; + this.errorCode = errorCode; + } + + } + + @Getter + private static class ShadowCasExhaustedException extends RuntimeException { + + private final Long lastKnownVersion; + private final boolean shadowNotFound; + + private ShadowCasExhaustedException(Long lastKnownVersion, boolean shadowNotFound) { + this.lastKnownVersion = lastKnownVersion != null ? lastKnownVersion : 0L; + this.shadowNotFound = shadowNotFound; + } + + static ShadowCasExhaustedException shadowNotFound() { + return new ShadowCasExhaustedException(0L, true); + } + + static ShadowCasExhaustedException versionConflict(Long lastKnownVersion) { + return new ShadowCasExhaustedException(lastKnownVersion, false); + } + + } + +} 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..fef46b9 --- /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,435 @@ +package cn.iocoder.yudao.module.iot.service.device.shadow; + +import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum; +import cn.iocoder.yudao.framework.common.exception.ServiceException; +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.core.topic.shadow.IotDeviceShadowReplyDTO; +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.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.springframework.dao.DuplicateKeyException; + +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.*; +import static org.mockito.Mockito.*; + +/** + * {@link IotDeviceShadowServiceImpl} 的单元测试 + */ +public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest { + + @InjectMocks + private IotDeviceShadowServiceImpl service; + + @Mock + private IotDeviceShadowMapper deviceShadowMapper; + @Mock + private DeviceShadowRedisDAO deviceShadowRedisDAO; + @Mock + private IotDeviceService deviceService; + @Mock + private IotDevicePropertyService devicePropertyService; + @Mock + private IotDeviceMessageService deviceMessageService; + + @BeforeEach + public void setUp() { + reset(deviceShadowMapper, deviceShadowRedisDAO, deviceService, + devicePropertyService, deviceMessageService); + } + + @Test + public void testHandleDeviceShadowUpdate_get_replyFullShadow() { + IotDeviceDO device = buildDevice(); + IotDeviceShadowDO shadow = buildShadow(device, 2L); + when(deviceShadowRedisDAO.get(device.getId())).thenReturn(null); + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(shadow); + + Map params = new HashMap<>(); + params.put("method", "get"); + IotDeviceMessage message = IotDeviceMessage.requestOf(device.getId(), 1L, "server-1", + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), params); + + service.handleDeviceShadowUpdate(device, message); + + ArgumentCaptor captor = ArgumentCaptor.forClass(IotDeviceMessage.class); + verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1")); + IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams(); + assertEquals("get_reply", reply.getMethod()); + assertEquals(2L, reply.getVersion()); + } + + @Test + public void testHandleDeviceShadowUpdate_update_mergeReportedAndVersionIncrement() { + IotDeviceDO device = buildDevice(); + IotDeviceShadowDO shadow = buildShadow(device, 1L); + shadow.getReported().put("Temperature", 20); + when(deviceShadowRedisDAO.get(device.getId())).thenReturn(null); + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(shadow); + when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(1L))).thenReturn(1); + + Map reported = new HashMap<>(); + reported.put("Temperature", 25); + reported.put("Humidity", 60); + Map state = new HashMap<>(); + state.put("reported", reported); + Map params = new HashMap<>(); + params.put("method", "update"); + params.put("state", state); + params.put("version", 1L); + IotDeviceMessage message = IotDeviceMessage.requestOf(device.getId(), 1L, "server-1", + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), params); + + service.handleDeviceShadowUpdate(device, message); + + ArgumentCaptor updateCaptor = ArgumentCaptor.forClass(IotDeviceShadowDO.class); + verify(deviceShadowMapper).updateByDeviceIdAndVersion(updateCaptor.capture(), eq(device.getId()), eq(1L)); + assertEquals(25, updateCaptor.getValue().getReported().get("Temperature")); + assertEquals(60, updateCaptor.getValue().getReported().get("Humidity")); + assertEquals(2L, updateCaptor.getValue().getVersion()); + verify(deviceShadowRedisDAO).delete(device.getId()); + } + + @Test + public void testHandleDeviceShadowUpdate_versionConflict_send409() { + IotDeviceDO device = buildDevice(); + IotDeviceShadowDO shadow = buildShadow(device, 5L); + when(deviceShadowRedisDAO.get(device.getId())).thenReturn(null); + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(shadow); + + Map reported = new HashMap<>(); + reported.put("Temperature", 25); + Map state = new HashMap<>(); + state.put("reported", reported); + Map params = new HashMap<>(); + params.put("method", "update"); + params.put("state", state); + params.put("version", 3L); + IotDeviceMessage message = IotDeviceMessage.requestOf(device.getId(), 1L, "server-1", + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), params); + + service.handleDeviceShadowUpdate(device, message); + + ArgumentCaptor captor = ArgumentCaptor.forClass(IotDeviceMessage.class); + verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1")); + IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams(); + assertEquals("update_reply", reply.getMethod()); + @SuppressWarnings("unchecked") + Map payload = (Map) reply.getPayload(); + assertEquals("error", payload.get("status")); + verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any()); + } + + @Test + public void testHandleDeviceShadowUpdate_versionMinusOne_resetShadow() { + IotDeviceDO device = buildDevice(); + IotDeviceShadowDO shadow = buildShadow(device, 3L); + shadow.getDesired().put("PowerSwitch", 1); + shadow.getReported().put("Temperature", 20); + when(deviceShadowRedisDAO.get(device.getId())).thenReturn(null); + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(shadow); + when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(3L))).thenReturn(1); + + Map reported = new HashMap<>(); + reported.put("Temperature", 20); + Map state = new HashMap<>(); + state.put("reported", reported); + Map params = new HashMap<>(); + params.put("method", "update"); + params.put("state", state); + params.put("version", -1L); + IotDeviceMessage message = IotDeviceMessage.requestOf(device.getId(), 1L, "server-1", + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), params); + + service.handleDeviceShadowUpdate(device, message); + + ArgumentCaptor updateCaptor = ArgumentCaptor.forClass(IotDeviceShadowDO.class); + verify(deviceShadowMapper).updateByDeviceIdAndVersion(updateCaptor.capture(), eq(device.getId()), eq(3L)); + assertTrue(updateCaptor.getValue().getDesired().isEmpty()); + assertTrue(updateCaptor.getValue().getReported().isEmpty()); + assertEquals(0L, updateCaptor.getValue().getVersion()); + } + + @Test + public void testHandleDeviceShadowUpdate_disabled_ignore() { + IotDeviceDO device = buildDevice(); + IotDeviceShadowDO shadow = buildShadow(device, 0L, CommonStatusEnum.DISABLE.getStatus()); + when(deviceShadowRedisDAO.get(device.getId())).thenReturn(null); + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(shadow); + + Map params = new HashMap<>(); + params.put("method", "get"); + IotDeviceMessage message = IotDeviceMessage.requestOf(device.getId(), 1L, "server-1", + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), params); + + service.handleDeviceShadowUpdate(device, message); + + verify(deviceMessageService, never()).sendDeviceMessage(any(), any(), any()); + } + + @Test + public void testUpdateDesiredState_online_pushShadow() { + Long deviceId = 2L; + IotDeviceDO device = buildDevice(); + IotDeviceShadowDO shadow = buildShadow(device, 1L); + when(deviceService.validateDeviceExists(deviceId)).thenReturn(device); + when(deviceShadowRedisDAO.get(deviceId)).thenReturn(null); + when(deviceShadowMapper.selectByDeviceId(deviceId)).thenReturn(shadow); + when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(deviceId), eq(1L))).thenReturn(1); + when(devicePropertyService.getDeviceServerId(deviceId)).thenReturn("server-1"); + + Map desired = new HashMap<>(); + desired.put("PowerSwitch", 1); + service.updateDesiredState(deviceId, desired); + + ArgumentCaptor captor = ArgumentCaptor.forClass(IotDeviceMessage.class); + verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1")); + IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams(); + assertEquals("get_reply", reply.getMethod()); + assertEquals(2L, reply.getVersion()); + // getOrCreateShadow + CAS 各一次 SELECT，推送不再额外查库 + verify(deviceShadowMapper, times(2)).selectByDeviceId(deviceId); + verify(deviceShadowRedisDAO).delete(deviceId); + } + + @Test + public void testUpdateDesiredState_offline_noPush() { + Long deviceId = 2L; + IotDeviceDO device = buildDevice(); + IotDeviceShadowDO shadow = buildShadow(device, 0L); + when(deviceService.validateDeviceExists(deviceId)).thenReturn(device); + when(deviceShadowRedisDAO.get(deviceId)).thenReturn(null); + when(deviceShadowMapper.selectByDeviceId(deviceId)).thenReturn(shadow); + when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(deviceId), eq(0L))).thenReturn(1); + when(devicePropertyService.getDeviceServerId(deviceId)).thenReturn(null); + + Map desired = new HashMap<>(); + desired.put("PowerSwitch", 1); + service.updateDesiredState(deviceId, desired); + + verify(deviceMessageService, never()).sendDeviceMessage(any(), any(), any()); + } + + @Test + public void testHandleDeviceShadowUpdate_methodMissing_send401() { + IotDeviceDO device = buildDevice(); + IotDeviceShadowDO shadow = buildShadow(device, 3L); + when(deviceShadowRedisDAO.get(device.getId())).thenReturn(null); + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(shadow); + + Map params = new HashMap<>(); + params.put("state", Map.of("reported", Map.of("Temperature", 25))); + IotDeviceMessage message = IotDeviceMessage.requestOf(device.getId(), 1L, "server-1", + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), params); + + service.handleDeviceShadowUpdate(device, message); + + ArgumentCaptor captor = ArgumentCaptor.forClass(IotDeviceMessage.class); + verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1")); + IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams(); + @SuppressWarnings("unchecked") + Map payload = (Map) reply.getPayload(); + @SuppressWarnings("unchecked") + Map content = (Map) payload.get("content"); + assertEquals(401, content.get("errorcode")); + } + + @Test + public void testHandleDeviceShadowUpdate_update_retryReMergeFromLatest() { + 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); + + when(deviceShadowRedisDAO.get(device.getId())).thenReturn(null); + AtomicInteger selectCount = new AtomicInteger(0); + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenAnswer(invocation -> { + // 前两次 select 返回 v=1 的快照，之后返回 v=2 的最新快照 + return selectCount.incrementAndGet() <= 2 ? staleShadow : latestShadow; + }); + when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), anyLong())).thenAnswer(invocation -> { + Long expectVersion = invocation.getArgument(2); + if (expectVersion == 1L) { + return 0; + } + if (expectVersion == 2L) { + return 1; + } + return 0; + }); + + Map reported = new HashMap<>(); + reported.put("Temperature", 25); + Map state = new HashMap<>(); + state.put("reported", reported); + Map params = new HashMap<>(); + params.put("method", "update"); + params.put("state", state); + params.put("version", 2L); + IotDeviceMessage message = IotDeviceMessage.requestOf(device.getId(), 1L, "server-1", + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), params); + + service.handleDeviceShadowUpdate(device, message); + + ArgumentCaptor updateCaptor = ArgumentCaptor.forClass(IotDeviceShadowDO.class); + verify(deviceShadowMapper, atLeastOnce()).updateByDeviceIdAndVersion(updateCaptor.capture(), eq(device.getId()), any()); + IotDeviceShadowDO finalUpdate = updateCaptor.getAllValues().get(updateCaptor.getAllValues().size() - 1); + // 重试时应基于 v=2 的最新 reported 合并，保留 Humidity=50 + assertEquals(25, finalUpdate.getReported().get("Temperature")); + assertEquals(50, finalUpdate.getReported().get("Humidity")); + assertEquals(3L, finalUpdate.getVersion()); + + ArgumentCaptor replyCaptor = ArgumentCaptor.forClass(IotDeviceMessage.class); + verify(deviceMessageService).sendDeviceMessage(replyCaptor.capture(), eq(device), eq("server-1")); + IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) replyCaptor.getValue().getParams(); + assertEquals(3L, reply.getVersion()); + } + + @Test + public void testUpdateStatus_whenShadowMissing_insertWithTargetStatus() { + Long deviceId = 2L; + IotDeviceDO device = buildDevice(); + when(deviceService.validateDeviceExists(deviceId)).thenReturn(device); + when(deviceShadowRedisDAO.get(deviceId)).thenReturn(null); + when(deviceShadowMapper.selectByDeviceId(deviceId)).thenReturn(null); + + service.updateStatus(deviceId, CommonStatusEnum.DISABLE.getStatus()); + + ArgumentCaptor captor = ArgumentCaptor.forClass(IotDeviceShadowDO.class); + verify(deviceShadowMapper).insert(captor.capture()); + assertEquals(CommonStatusEnum.DISABLE.getStatus(), captor.getValue().getStatus()); + verify(deviceShadowRedisDAO, never()).delete(deviceId); + } + + @Test + public void testUpdateStatus_whenShadowMissing_duplicateKey_fallbackUpdateAndEvictCache() { + Long deviceId = 2L; + IotDeviceDO device = buildDevice(); + when(deviceService.validateDeviceExists(deviceId)).thenReturn(device); + when(deviceShadowRedisDAO.get(deviceId)).thenReturn(null); + when(deviceShadowMapper.selectByDeviceId(deviceId)).thenReturn(null); + when(deviceShadowMapper.insert(any(IotDeviceShadowDO.class))).thenThrow(new DuplicateKeyException("duplicate")); + when(deviceShadowMapper.updateStatusByDeviceId(deviceId, CommonStatusEnum.DISABLE.getStatus())).thenReturn(1); + + service.updateStatus(deviceId, CommonStatusEnum.DISABLE.getStatus()); + + verify(deviceShadowMapper).updateStatusByDeviceId(deviceId, CommonStatusEnum.DISABLE.getStatus()); + verify(deviceShadowRedisDAO).delete(deviceId); + } + + @Test + public void testResetShadow_casConflict_throwException() { + Long deviceId = 2L; + IotDeviceDO device = buildDevice(); + IotDeviceShadowDO shadow = buildShadow(device, 2L); + when(deviceService.validateDeviceExists(deviceId)).thenReturn(device); + when(deviceShadowRedisDAO.get(deviceId)).thenReturn(null); + when(deviceShadowMapper.selectByDeviceId(deviceId)).thenReturn(shadow); + when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(deviceId), eq(2L))).thenReturn(0); + + assertThrows(ServiceException.class, () -> service.resetShadow(deviceId)); + } + + @Test + public void testHandleDeviceShadowUpdate_catchBlock_dbFailureStillSendsReply() { + IotDeviceDO device = buildDevice(); + when(deviceShadowRedisDAO.get(device.getId())).thenReturn(null); + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenThrow(new RuntimeException("DB down")); + + Map params = new HashMap<>(); + params.put("method", "get"); + IotDeviceMessage message = IotDeviceMessage.requestOf(device.getId(), 1L, "server-1", + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), params); + + assertDoesNotThrow(() -> service.handleDeviceShadowUpdate(device, message)); + + ArgumentCaptor captor = ArgumentCaptor.forClass(IotDeviceMessage.class); + verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1")); + IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams(); + assertEquals(0L, reply.getVersion()); + @SuppressWarnings("unchecked") + Map payload = (Map) reply.getPayload(); + @SuppressWarnings("unchecked") + Map content = (Map) payload.get("content"); + assertEquals(500, content.get("errorcode")); + } + + @Test + public void testHandleDeviceShadowUpdate_casShadowNotFound_send500() { + IotDeviceDO device = buildDevice(); + IotDeviceShadowDO shadow = buildShadow(device, 2L); + when(deviceShadowRedisDAO.get(device.getId())).thenReturn(null); + when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(shadow, (IotDeviceShadowDO) null); + + Map reported = new HashMap<>(); + reported.put("Temperature", 25); + Map state = new HashMap<>(); + state.put("reported", reported); + Map params = new HashMap<>(); + params.put("method", "update"); + params.put("state", state); + params.put("version", 2L); + IotDeviceMessage message = IotDeviceMessage.requestOf(device.getId(), 1L, "server-1", + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), params); + + service.handleDeviceShadowUpdate(device, message); + + ArgumentCaptor captor = ArgumentCaptor.forClass(IotDeviceMessage.class); + verify(deviceMessageService).sendDeviceMessage(captor.capture(), eq(device), eq("server-1")); + IotDeviceShadowReplyDTO reply = (IotDeviceShadowReplyDTO) captor.getValue().getParams(); + @SuppressWarnings("unchecked") + Map payload = (Map) reply.getPayload(); + @SuppressWarnings("unchecked") + Map content = (Map) payload.get("content"); + assertEquals(500, content.get("errorcode")); + verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any()); + } + + 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 IotDeviceShadowDO.builder() + .id(100L) + .deviceId(device.getId()) + .productId(device.getProductId()) + .productKey(device.getProductKey()) + .deviceName(device.getDeviceName()) + .desired(new HashMap<>()) + .reported(new HashMap<>()) + .metadata(new HashMap<>()) + .version(version) + .shadowTime(1000L) + .status(status) + .build(); + } + +} 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 0bd2ad2..5cb1552 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 @@ -181,6 +181,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..c57f87b 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,11 @@ 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), + SHADOW_GET("thing.shadow.get", "设备影子获取", false), + ; public static final String[] ARRAYS = Arrays.stream(values()).map(IotDeviceMessageMethodEnum::getMethod) @@ -77,7 +82,16 @@ 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.get + ); + + /** + * 透传方法集合：设备侧使用原始线格式，不套 IotDeviceMessage 封套 + */ + public static final Set PASSTHROUGH_METHODS = SetUtils.asSet( + SHADOW_UPDATE.getMethod(), + SHADOW_GET.getMethod() ); private final String method; @@ -100,4 +114,8 @@ public enum IotDeviceMessageMethodEnum implements ArrayValuable { return REPLY_DISABLED.contains(method); } + public static boolean isPassthrough(String method) { + return PASSTHROUGH_METHODS.contains(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..6087003 --- /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 设备影子设备侧错误码枚举 + *

+ * 用于回填 update_reply.payload.content，不走 CommonResult + * + * @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"), + SHADOW_EMPTY(407, "影子内容为空"), + 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/enums/shadow/IotDeviceShadowMetadatas.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/shadow/IotDeviceShadowMetadatas.java new file mode 100644 index 0000000..4790b35 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/shadow/IotDeviceShadowMetadatas.java @@ -0,0 +1,92 @@ +package cn.iocoder.yudao.module.iot.core.enums.shadow; + +import cn.hutool.core.map.MapUtil; + +import java.util.HashMap; +import java.util.Map; +import java.util.Set; + +import static cn.iocoder.yudao.module.iot.core.enums.shadow.IotDeviceShadowProtocolConstants.*; + +/** + * IoT 设备影子 metadata 读写工具 + *

+ * 结构：{ desired: { identifier: { timestamp } }, reported: { ... } } + * + * @author 芋道源码 + */ +public final class IotDeviceShadowMetadatas { + + private IotDeviceShadowMetadatas() { + } + + /** + * 获取 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-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/shadow/IotDeviceShadowProtocolConstants.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/shadow/IotDeviceShadowProtocolConstants.java new file mode 100644 index 0000000..a6414fb --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/shadow/IotDeviceShadowProtocolConstants.java @@ -0,0 +1,42 @@ +package cn.iocoder.yudao.module.iot.core.enums.shadow; + +/** + * IoT 设备影子协议常量 + *

+ * 内层 method、state 键名等与设备侧透传 JSON 对齐 + * + * @author 芋道源码 + */ +public final class IotDeviceShadowProtocolConstants { + + private IotDeviceShadowProtocolConstants() { + } + + // ========== 内层 method ========== + + public static final String INNER_METHOD_GET = "get"; + public static final String INNER_METHOD_UPDATE = "update"; + public static final String INNER_METHOD_GET_REPLY = "get_reply"; + public static final String INNER_METHOD_UPDATE_REPLY = "update_reply"; + + // ========== state / metadata 键 ========== + + public static final String STATE_DESIRED = "desired"; + public static final String STATE_REPORTED = "reported"; + public static final String METADATA_TIMESTAMP = "timestamp"; + + // ========== payload 键 ========== + + public static final String PAYLOAD_STATUS = "status"; + public static final String PAYLOAD_STATE = "state"; + public static final String PAYLOAD_VERSION = "version"; + public static final String PAYLOAD_CONTENT = "content"; + public static final String PAYLOAD_ERROR_CODE = "errorcode"; + public static final String PAYLOAD_ERROR_MESSAGE = "errormessage"; + + // ========== payload status 值 ========== + + public static final String STATUS_SUCCESS = "success"; + public static final String STATUS_ERROR = "error"; + +} diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/shadow/IotDeviceShadowReplyDTO.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/shadow/IotDeviceShadowReplyDTO.java new file mode 100644 index 0000000..9f4816d --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/shadow/IotDeviceShadowReplyDTO.java @@ -0,0 +1,86 @@ +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.HashMap; +import java.util.Map; + +import static cn.iocoder.yudao.module.iot.core.enums.shadow.IotDeviceShadowProtocolConstants.*; + +/** + * IoT 设备影子下行 Reply DTO + *

+ * 用于 {@link IotDeviceMessageMethodEnum#SHADOW_GET} 消息的 params 参数（透传原始线 body） + * + * @author 芋道源码 + */ +@Data +@NoArgsConstructor +@AllArgsConstructor +public class IotDeviceShadowReplyDTO { + + /** + * 内层方法：get_reply / update_reply + */ + private String method; + + /** + * 回复载荷 + */ + private Object payload; + + /** + * 影子版本号 + */ + private Long version; + + /** + * 时间戳（epoch 秒） + */ + private Long timestamp; + + public static IotDeviceShadowReplyDTO getReplyOf(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 payload = new HashMap<>(); + payload.put(PAYLOAD_STATUS, STATUS_SUCCESS); + payload.put(PAYLOAD_STATE, state); + return new IotDeviceShadowReplyDTO() + .setMethod(INNER_METHOD_GET_REPLY) + .setPayload(payload) + .setVersion(version) + .setTimestamp(timestamp); + } + + public static IotDeviceShadowReplyDTO updateReplyOf(Long version, Long timestamp) { + Map payload = new HashMap<>(); + payload.put(PAYLOAD_STATUS, STATUS_SUCCESS); + payload.put(PAYLOAD_VERSION, version); + return new IotDeviceShadowReplyDTO() + .setMethod(INNER_METHOD_UPDATE_REPLY) + .setPayload(payload) + .setVersion(version) + .setTimestamp(timestamp); + } + + public static IotDeviceShadowReplyDTO errorOf(Long version, Integer errorcode, String errormessage, Long timestamp) { + Map content = new HashMap<>(); + content.put(PAYLOAD_ERROR_CODE, errorcode); + content.put(PAYLOAD_ERROR_MESSAGE, errormessage); + Map payload = new HashMap<>(); + payload.put(PAYLOAD_STATUS, STATUS_ERROR); + payload.put(PAYLOAD_VERSION, version); + payload.put(PAYLOAD_CONTENT, content); + return new IotDeviceShadowReplyDTO() + .setMethod(INNER_METHOD_UPDATE_REPLY) + .setPayload(payload) + .setVersion(version) + .setTimestamp(timestamp); + } + +} 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..ba49773 --- /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,54 @@ +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 参数（透传原始线 body） + * + * @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; + + /** + * 期望状态 + */ + private Map desired; + + } + +} diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/upstream/IotMqttUpstreamHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/upstream/IotMqttUpstreamHandler.java index 81a6ece..2b5ec37 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/upstream/IotMqttUpstreamHandler.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/upstream/IotMqttUpstreamHandler.java @@ -2,6 +2,8 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.handler.upstream; import cn.hutool.core.lang.Assert; import cn.hutool.core.util.ArrayUtil; +import cn.iocoder.yudao.framework.common.util.json.JsonUtils; +import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum; import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.manager.IotMqttConnectionManager; import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService; @@ -9,6 +11,10 @@ import cn.iocoder.yudao.module.iot.gateway.util.IotMqttTopicUtils; import io.vertx.mqtt.MqttEndpoint; import lombok.extern.slf4j.Slf4j; +import java.util.Arrays; +import java.util.Map; +import java.util.stream.Collectors; + /** * IoT 网关 MQTT 上行消息处理器：处理业务消息（属性上报、事件上报等） @@ -59,13 +65,27 @@ public class IotMqttUpstreamHandler extends IotMqttAbstractHandler { } // 2.1 反序列化消息 - IotDeviceMessage message = deviceMessageService.deserializeDeviceMessage(payload, productKey, deviceName); - if (message == null) { - log.warn("[handleBusinessRequest][消息解码失败，客户端 ID: {}，主题: {}]", clientId, topic); - return; + String topicMethod = Arrays.stream(topicParts, 4, topicParts.length) + .collect(Collectors.joining(".")); + IotDeviceMessage message; + if (IotDeviceMessageMethodEnum.isPassthrough(topicMethod)) { + // 影子透传：从 topic 解析 yudao method，原始 body 作为 params + Map params = JsonUtils.parseObject(payload, Map.class); + if (params == null) { + log.warn("[handleBusinessRequest][影子透传 payload 解析失败，客户端 ID: {}，主题: {}]", clientId, topic); + return; + } + String requestId = (String) params.get("id"); + message = IotDeviceMessage.requestOf(requestId, topicMethod, params); + } else { + message = deviceMessageService.deserializeDeviceMessage(payload, productKey, deviceName); + if (message == null) { + log.warn("[handleBusinessRequest][消息解码失败，客户端 ID: {}，主题: {}]", clientId, topic); + return; + } + // 2.2 标准化回复消息的 method（MQTT 协议中，设备回复消息的 method 会携带 _reply 后缀） + IotMqttTopicUtils.normalizeReplyMethod(message); } - // 2.2 标准化回复消息的 method（MQTT 协议中，设备回复消息的 method 会携带 _reply 后缀） - IotMqttTopicUtils.normalizeReplyMethod(message); // 3. 处理业务消息 deviceMessageService.sendDeviceMessage(message, productKey, deviceName, serverId); diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceMessageServiceImpl.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceMessageServiceImpl.java index ee0c4ae..8ffe1a7 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceMessageServiceImpl.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceMessageServiceImpl.java @@ -2,7 +2,9 @@ package cn.iocoder.yudao.module.iot.gateway.service.device.message; import cn.hutool.core.lang.Assert; import cn.hutool.core.util.StrUtil; +import cn.iocoder.yudao.framework.common.util.json.JsonUtils; import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO; +import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum; import cn.iocoder.yudao.module.iot.core.enums.IotSerializeTypeEnum; import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer; @@ -40,6 +42,10 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { @Override public byte[] serializeDeviceMessage(IotDeviceMessage message, String productKey, String deviceName) { + // 影子透传：直接序列化 params 为原始线 JSON，忽略产品 serializeType + if (IotDeviceMessageMethodEnum.isPassthrough(message.getMethod())) { + return JsonUtils.toJsonByte(message.getParams()); + } // 1.1 获取设备信息 IotDeviceRespDTO device = deviceService.getDeviceFromCache(productKey, deviceName); if (device == null) { @@ -56,6 +62,10 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { @Override public byte[] serializeDeviceMessage(IotDeviceMessage message, IotSerializeTypeEnum serializeType) { + // 影子透传 + if (IotDeviceMessageMethodEnum.isPassthrough(message.getMethod())) { + return JsonUtils.toJsonByte(message.getParams()); + } // 1. 获取序列化器 IotMessageSerializer serializer = messageSerializerManager.get(serializeType); if (serializer == null) { diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceShadowPassthroughTest.java b/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceShadowPassthroughTest.java new file mode 100644 index 0000000..bdf7a75 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceShadowPassthroughTest.java @@ -0,0 +1,70 @@ +package cn.iocoder.yudao.module.iot.gateway.service.device.message; + +import cn.iocoder.yudao.framework.common.util.json.JsonUtils; +import cn.iocoder.yudao.framework.test.core.ut.BaseMockitoUnitTest; +import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum; +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; +import cn.iocoder.yudao.module.iot.core.topic.shadow.IotDeviceShadowReplyDTO; +import org.junit.jupiter.api.Test; +import org.mockito.InjectMocks; + +import java.util.HashMap; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; + +/** + * 设备影子网关透传序列化单元测试 + */ +public class IotDeviceShadowPassthroughTest extends BaseMockitoUnitTest { + + @InjectMocks + private IotDeviceMessageServiceImpl deviceMessageService; + + @Test + public void testSerializeDeviceMessage_passthrough_noEnvelope() { + Map payload = new HashMap<>(); + payload.put("method", "get_reply"); + payload.put("version", 2L); + payload.put("timestamp", 1469564576L); + IotDeviceMessage message = IotDeviceMessage.requestOf( + IotDeviceMessageMethodEnum.SHADOW_GET.getMethod(), payload); + + byte[] bytes = deviceMessageService.serializeDeviceMessage(message, null); + + @SuppressWarnings("unchecked") + Map result = JsonUtils.parseObject(bytes, Map.class); + assertEquals("get_reply", result.get("method")); + assertEquals(2, ((Number) result.get("version")).intValue()); + assertNull(result.get("id")); + assertNull(result.get("code")); + assertNull(result.get("requestId")); + } + + @Test + public void testDeserializeUpstreamShadow_fromTopicMethod() { + // 模拟上行透传：原始 body 不含 thing.shadow.update 封套 method + Map body = new HashMap<>(); + body.put("method", "get"); + IotDeviceMessage message = IotDeviceMessage.requestOf("req-1", + IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), body); + + assertEquals(IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), message.getMethod()); + assertEquals("get", ((Map) message.getParams()).get("method")); + } + + @Test + public void testSerializeUpdateReply_passthrough() { + IotDeviceShadowReplyDTO reply = IotDeviceShadowReplyDTO.updateReplyOf(3L, 1469564576L); + IotDeviceMessage message = IotDeviceMessage.requestOf( + IotDeviceMessageMethodEnum.SHADOW_GET.getMethod(), reply); + + byte[] bytes = deviceMessageService.serializeDeviceMessage(message, null); + @SuppressWarnings("unchecked") + Map result = JsonUtils.parseObject(bytes, Map.class); + assertEquals("update_reply", result.get("method")); + assertNull(result.get("id")); + } + +}