feat：【IoT 物联网】实现 OTA updateOtaRecordProgress

YunaiV
1 parent af1f993a
Showing 11 changed files with 134 additions and 9 deletions
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/ota/IotOtaTaskRecordMapper.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/ErrorCodeConstants.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/ota/IotOtaTaskRecordStatusEnum.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/IotDeviceService.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/IotDeviceServiceImpl.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/message/IotDeviceMessageServiceImpl.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/ota/IotOtaFirmwareService.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/ota/IotOtaFirmwareServiceImpl.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/ota/IotOtaTaskRecordService.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/ota/IotOtaTaskRecordServiceImpl.java
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotDeviceMessageMethodEnum.java
@@ -67,6 +67,12 @@ public interface IotOtaTaskRecordMapper extends BaseMapperX&lt;IotOtaTaskRecordDO&gt; 
                 .inIfPresent(IotOtaTaskRecordDO::getStatus, statuses));
     }
+    default List<IotOtaTaskRecordDO> selectListByDeviceIdAndStatus(Long deviceId, Set<Integer> statuses) {
+        return selectList(new LambdaQueryWrapperX<IotOtaTaskRecordDO>()
+                .eqIfPresent(IotOtaTaskRecordDO::getDeviceId, deviceId)
+                .inIfPresent(IotOtaTaskRecordDO::getStatus, statuses));
+    }
+
     default List<IotOtaTaskRecordDO> selectListByStatus(Integer status) {
         return selectList(IotOtaTaskRecordDO::getStatus, status);
     }
@@ -61,6 +61,7 @@ public interface ErrorCodeConstants {
     ErrorCode OTA_TASK_RECORD_NOT_EXISTS = new ErrorCode(1_050_008_200, "升级记录不存在");
     ErrorCode OTA_TASK_RECORD_CANCEL_FAIL_STATUS_ERROR = new ErrorCode(1_050_008_201, "取消 OTA 升级记录失败，原因：记录状态不是进行中");
+    ErrorCode OTA_TASK_RECORD_UPDATE_PROGRESS_FAIL_NO_EXISTS = new ErrorCode(1_050_008_202, "更新 OTA 升级记录进度失败，原因：该设备没有进行中的升级记录");
     // ========== IoT 数据流转规则 1-050-010-000 ==========
     ErrorCode DATA_RULE_NOT_EXISTS = new ErrorCode(1_050_010_000, "数据流转规则不存在");
 package cn.iocoder.yudao.module.iot.enums.ota;
+import cn.hutool.core.util.ArrayUtil;
 import cn.iocoder.yudao.framework.common.core.ArrayValuable;
 import cn.iocoder.yudao.framework.common.util.collection.SetUtils;
 import lombok.Getter;
@@ -49,4 +50,8 @@ public enum IotOtaTaskRecordStatusEnum implements ArrayValuable&lt;Integer&gt; {
         return ARRAYS;
     }
+    public static IotOtaTaskRecordStatusEnum of(Integer status) {
+        return ArrayUtil.firstMatch(o -> o.getStatus().equals(status), values());
+    }
+
 }
 \ No newline at end of file
@@ -263,4 +263,12 @@ public interface IotDeviceService {
         return convertMap(getDeviceList(ids), IotDeviceDO::getId);
     }
+    /**
+     * 更新设备固件版本
+     *
+     * @param deviceId 设备编号
+     * @param firmwareId 固件编号
+     */
+    void updateDeviceFirmware(Long deviceId, Long firmwareId);
+
 }
@@ -485,6 +485,19 @@ public class IotDeviceServiceImpl implements IotDeviceService {
         return deviceMapper.selectByIds(ids);
     }
+    @Override
+    public void updateDeviceFirmware(Long deviceId, Long firmwareId) {
+        // 1. 校验设备是否存在
+        IotDeviceDO device = validateDeviceExists(deviceId);
+        
+        // 2. 更新设备固件版本
+        IotDeviceDO updateObj = new IotDeviceDO().setId(deviceId).setFirmwareId(firmwareId);
+        deviceMapper.updateById(updateObj);
+        
+        // 3. 清空对应缓存
+        deleteDeviceCache(device);
+    }
+
     private IotDeviceServiceImpl getSelf() {
         return SpringUtil.getBean(getClass());
     }
@@ -22,11 +22,13 @@ 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.ota.IotOtaTaskRecordService;
 import com.baomidou.mybatisplus.core.metadata.IPage;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import com.google.common.base.Objects;
 import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.annotation.Lazy;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 import org.springframework.validation.annotation.Validated;
@@ -54,6 +56,9 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
     private IotDeviceService deviceService;
     @Resource
     private IotDevicePropertyService devicePropertyService;
+    @Resource
+    @Lazy // 延迟加载，避免循环依赖
+    private IotOtaTaskRecordService otaTaskRecordService;
     @Resource
     private IotDeviceMessageMapper deviceMessageMapper;
@@ -192,6 +197,12 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
             return null;
         }
+        // OTA 上报升级进度
+        if (Objects.equal(message.getMethod(), IotDeviceMessageMethodEnum.OTA_PROGRESS.getMethod())) {
+            otaTaskRecordService.updateOtaRecordProgress(device, message);
+            return null;
+        }
+
         // TODO @芋艿：这里可以按需，添加别的逻辑；
         return null;
     }
@@ -7,12 +7,12 @@ import cn.iocoder.yudao.module.iot.controller.admin.ota.vo.firmware.IotOtaFirmwa
 import cn.iocoder.yudao.module.iot.dal.dataobject.ota.IotOtaFirmwareDO;
 import jakarta.validation.Valid;
-import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertMap;
-
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
+import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertMap;
+
 /**
  * OTA 固件管理 Service 接口
  *
@@ -44,6 +44,15 @@ public interface IotOtaFirmwareService {
     IotOtaFirmwareDO getOtaFirmware(Long id);
     /**
+     * 根据产品、版本号，获取 OTA 固件信息
+     *
+     * @param productId 产品编号
+     * @param version 版本号
+     * @return OTA 固件信息
+     */
+    IotOtaFirmwareDO getOtaFirmwareByProductIdAndVersion(Long productId, String version);
+
+    /**
      * 获取 OTA 固件信息列表
      *
      * @param ids 固件编号集合
@@ -21,10 +21,8 @@ import java.io.ByteArrayInputStream;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
-import java.util.Map;
 import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
-import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertMap;
 import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.OTA_FIRMWARE_NOT_EXISTS;
 import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.OTA_FIRMWARE_PRODUCT_VERSION_DUPLICATE;
@@ -82,6 +80,11 @@ public class IotOtaFirmwareServiceImpl implements IotOtaFirmwareService {
     }
     @Override
+    public IotOtaFirmwareDO getOtaFirmwareByProductIdAndVersion(Long productId, String version) {
+        return otaFirmwareMapper.selectByProductIdAndVersion(productId, version);
+    }
+
+    @Override
     public List<IotOtaFirmwareDO> getOtaFirmwareList(Collection<Long> ids) {
         if (ids == null || ids.isEmpty()) {
             return Collections.emptyList();
@@ -2,6 +2,7 @@ package cn.iocoder.yudao.module.iot.service.ota;
 import cn.iocoder.yudao.framework.common.pojo.PageResult;
 import cn.iocoder.yudao.module.iot.controller.admin.ota.vo.task.record.IotOtaTaskRecordPageReqVO;
+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.ota.IotOtaFirmwareDO;
 import cn.iocoder.yudao.module.iot.dal.dataobject.ota.IotOtaTaskRecordDO;
@@ -91,4 +92,12 @@ public interface IotOtaTaskRecordService {
      */
     boolean pushOtaTaskRecord(IotOtaTaskRecordDO record, IotOtaFirmwareDO fireware, IotDeviceDO device);
+    /**
+     * 更新 OTA 升级记录进度
+     *
+     * @param device   设备信息
+     * @param message  设备消息
+     */
+    void updateOtaRecordProgress(IotDeviceDO device, IotDeviceMessage message);
+
 }
@@ -3,6 +3,7 @@ package cn.iocoder.yudao.module.iot.service.ota;
 import cn.hutool.core.collection.CollUtil;
 import cn.hutool.core.convert.Convert;
 import cn.hutool.core.lang.Assert;
+import cn.hutool.core.map.MapUtil;
 import cn.hutool.core.util.StrUtil;
 import cn.iocoder.yudao.framework.common.pojo.PageResult;
 import cn.iocoder.yudao.module.iot.controller.admin.ota.vo.task.record.IotOtaTaskRecordPageReqVO;
@@ -17,14 +18,14 @@ import cn.iocoder.yudao.module.iot.service.device.message.IotDeviceMessageServic
 import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
 import org.springframework.validation.annotation.Validated;
 import java.util.*;
 import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
 import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.*;
-import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.OTA_TASK_RECORD_NOT_EXISTS;
-import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.OTA_TASK_RECORD_CANCEL_FAIL_STATUS_ERROR;
+import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.*;
 /**
  * OTA 升级任务记录 Service 实现类
@@ -38,6 +39,8 @@ public class IotOtaTaskRecordServiceImpl implements IotOtaTaskRecordService {
     private IotOtaTaskRecordMapper otaTaskRecordMapper;
     @Resource
+    private IotOtaFirmwareService otaFirmwareService;
+    @Resource
     private IotOtaTaskService otaTaskService;
     @Resource
     private IotDeviceMessageService deviceMessageService;
@@ -158,6 +161,57 @@ public class IotOtaTaskRecordServiceImpl implements IotOtaTaskRecordService {
         return upgradeRecord;
     }
+    @Override
+    @Transactional(rollbackFor = Exception.class)
+    @SuppressWarnings("unchecked")
+    public void updateOtaRecordProgress(IotDeviceDO device, IotDeviceMessage message) {
+        // 1.1 参数解析
+        Map<String, Object> params = (Map<String, Object>) message.getParams();
+        String version = MapUtil.getStr(params, "version");
+        Assert.notBlank(version, "version 不能为空");
+        Integer status = MapUtil.getInt(params, "status");
+        Assert.notNull(status, "status 不能为空");
+        Assert.notNull(IotOtaTaskRecordStatusEnum.of(status), "status 状态不正确");
+        String description = MapUtil.getStr(params, "description");
+        Integer progress = MapUtil.getInt(params, "progress");
+        Assert.notNull(progress, "progress 不能为空");
+        Assert.isTrue(progress >= 0 && progress <= 100, "progress 必须在 0-100 之间");
+        // 1.2 查询 OTA 升级记录
+        List<IotOtaTaskRecordDO> records = otaTaskRecordMapper.selectListByDeviceIdAndStatus(
+                device.getId(), IotOtaTaskRecordStatusEnum.IN_PROCESS_STATUSES);
+        if (CollUtil.isEmpty(records)) {
+            throw exception(OTA_TASK_RECORD_UPDATE_PROGRESS_FAIL_NO_EXISTS);
+        }
+        if (records.size() > 1) {
+            log.warn("[updateOtaRecordProgress][message({}) 对应升级记录过多({})]", message, records);
+        }
+        IotOtaTaskRecordDO record = CollUtil.getFirst(records);
+        // 1.3 查询 OTA 固件
+        IotOtaFirmwareDO firmware = otaFirmwareService.getOtaFirmwareByProductIdAndVersion(
+                device.getProductId(), version);
+        if (firmware == null) {
+            throw exception(OTA_FIRMWARE_NOT_EXISTS);
+        }
+
+        // 2. 更新 OTA 升级记录状态
+        int updateCount = otaTaskRecordMapper.updateByIdAndStatus(
+                record.getId(), IotOtaTaskRecordStatusEnum.IN_PROCESS_STATUSES,
+                IotOtaTaskRecordDO.builder().status(status).description(description).progress(progress).build());
+        if (updateCount == 0) {
+            throw exception(OTA_TASK_RECORD_UPDATE_PROGRESS_FAIL_NO_EXISTS);
+        }
+
+        // 3. 如果升级成功，则更新设备固件版本
+        if (IotOtaTaskRecordStatusEnum.SUCCESS.getStatus().equals(status)) {
+            deviceService.updateDeviceFirmware(device.getId(), firmware.getId());
+        }
+
+        // 4. 如果状态是“已结束”（非进行中），则更新任务状态
+        if (!IotOtaTaskRecordStatusEnum.IN_PROCESS_STATUSES.contains(status)) {
+            checkAndUpdateOtaTaskStatus(record.getTaskId());
+        }
+    }
+
     /**
      * 检查并更新任务状态
      * 如果任务下没有进行中的记录，则将任务状态更新为已结束
@@ -30,10 +30,12 @@ public enum IotDeviceMessageMethodEnum implements ArrayValuable&lt;String&gt; {
     PROPERTY_SET("thing.property.set", "属性设置", false),
     // ========== 设备事件 ==========
+    // 可参考：https://help.aliyun.com/zh/iot/user-guide/device-properties-events-and-services
     EVENT_POST("thing.event.post", "事件上报", true),
     // ========== 设备服务调用 ==========
+    // 可参考：https://help.aliyun.com/zh/iot/user-guide/device-properties-events-and-services
     SERVICE_INVOKE("thing.service.invoke", "服务调用", false),
@@ -43,9 +45,10 @@ public enum IotDeviceMessageMethodEnum implements ArrayValuable&lt;String&gt; {
     CONFIG_PUSH("thing.config.push", "配置推送", true),
     // ========== OTA 固件 ==========
+    // 可参考：https://help.aliyun.com/zh/iot/user-guide/perform-ota-updates
-    OTA_UPGRADE("thing.ota.upgrade", "OTA 推送固定信息", false),
-    OTA_PROGRESS("thing.ota.progress", "OTA 上报升级进度", true),
+    OTA_UPGRADE("thing.ota.upgrade", "OTA 固定信息推送", false),
+    OTA_PROGRESS("thing.ota.progress", "OTA 升级进度上报", true),
     ;
     public static final String[] ARRAYS = Arrays.stream(values()).map(IotDeviceMessageMethodEnum::getMethod)
@@ -54,7 +57,10 @@ public enum IotDeviceMessageMethodEnum implements ArrayValuable&lt;String&gt; {
     /**
      * 不进行 reply 回复的方法集合
      */
-    public static final Set<String> REPLY_DISABLED = Set.of(STATE_UPDATE.getMethod());
+    public static final Set<String> REPLY_DISABLED = Set.of(
+            STATE_UPDATE.getMethod(),
+            OTA_PROGRESS.getMethod() // 参考阿里云，OTA 升级进度上报，不进行回复
+    );
     private final String method;