reactor：【IoT 物联网】重新梳理下行消息的逻辑（未测试，用于相互 review 作用）

YunaiV
1 parent 4ea6e08f
Showing 46 changed files with 622 additions and 1424 deletions
yudao-framework/yudao-common/src/main/java/cn/iocoder/yudao/framework/common/util/object/ObjectUtils.java
yudao-module-iot/yudao-module-iot-api/src/main/java/cn/iocoder/yudao/module/iot/api/device/dto/control/upstream/IotDeviceRegisterReqDTO.java
yudao-module-iot/yudao-module-iot-api/src/main/java/cn/iocoder/yudao/module/iot/api/device/dto/control/upstream/IotDeviceRegisterSubReqDTO.java
yudao-module-iot/yudao-module-iot-api/src/main/java/cn/iocoder/yudao/module/iot/api/device/dto/control/upstream/IotDeviceTopologyAddReqDTO.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/IotDeviceController.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/IotDeviceLogController.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/IotDevicePropertyController.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/vo/control/IotDeviceDownstreamReqVO.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/vo/control/IotDeviceUpstreamReqVO.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/vo/message/IotDeviceMessageSendReqVO.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/ota/vo/upgrade/record/IotOtaUpgradeRecordRespVO.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/statistics/IotStatisticsController.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/ota/IotOtaUpgradeRecordDO.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/device/DeviceServerIdRedisDAO.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/framework/tdengine/config/TDengineTableInitRunner.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/job/device/IotDeviceOfflineCheckJob.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mq/consumer/device/IotDeviceLogMessageSubscriber.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mq/consumer/device/IotDevicePropertyMessageSubscriber.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
@@ -60,4 +60,8 @@ public class ObjectUtils {
         return Arrays.asList(array).contains(obj);
     }
  
+    public static boolean isNotAllEmpty(Object... objs) {
+        return !ObjectUtil.isAllEmpty(objs);
+    }
+
 }
-package cn.iocoder.yudao.module.iot.api.device.dto.control.upstream;
-
-import lombok.Data;
-
-/**
- * IoT 设备【注册】自己 Request DTO
- *
- * @author 芋道源码
- */
-@Data
-public class IotDeviceRegisterReqDTO extends IotDeviceUpstreamAbstractReqDTO {
-}
-package cn.iocoder.yudao.module.iot.api.device.dto.control.upstream;
-
-import jakarta.validation.constraints.NotEmpty;
-import lombok.Data;
-
-import java.util.List;
-
-/**
- * IoT 设备【注册】子设备 Request DTO
- *
- * @author 芋道源码
- */
-@Data
-public class IotDeviceRegisterSubReqDTO extends IotDeviceUpstreamAbstractReqDTO {
-
-    // TODO @芋艿：看看要不要优化命名
-    /**
-     * 子设备数组
-     */
-    @NotEmpty(message = "子设备不能为空")
-    private List<Device> params;
-
-    /**
-     * 设备信息
-     */
-    @Data
-    public static class Device {
-
-        /**
-         * 产品标识
-         */
-        @NotEmpty(message = "产品标识不能为空")
-        private String productKey;
-
-        /**
-         * 设备名称
-         */
-        @NotEmpty(message = "设备名称不能为空")
-        private String deviceName;
-
-    }
-
-}
-package cn.iocoder.yudao.module.iot.api.device.dto.control.upstream;
-
-import jakarta.validation.constraints.NotEmpty;
-import lombok.Data;
-
-import java.util.List;
-
-// TODO @芋艿：要写清楚，是来自设备网关，还是设备。
-/**
- * IoT 设备【拓扑】添加 Request DTO
- */
-@Data
-public class IotDeviceTopologyAddReqDTO extends IotDeviceUpstreamAbstractReqDTO {
-
-    // TODO @芋艿：看看要不要优化命名
-    /**
-     * 子设备数组
-     */
-    @NotEmpty(message = "子设备不能为空")
-    private List<IotDeviceRegisterSubReqDTO.Device> params;
-
-    /**
-     * 设备信息
-     */
-    @Data
-    public static class Device {
-
-        /**
-         * 产品标识
-         */
-        @NotEmpty(message = "产品标识不能为空")
-        private String productKey;
-
-        /**
-         * 设备名称
-         */
-        @NotEmpty(message = "设备名称不能为空")
-        private String deviceName;
-
-        // TODO @芋艿：阿里云还有 sign 签名
-
-    }
-
-}
@@ -6,13 +6,12 @@ import cn.iocoder.yudao.framework.common.pojo.PageParam;
 import cn.iocoder.yudao.framework.common.pojo.PageResult;
 import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
 import cn.iocoder.yudao.framework.excel.core.util.ExcelUtils;
-import cn.iocoder.yudao.module.iot.controller.admin.device.vo.control.IotDeviceDownstreamReqVO;
-import cn.iocoder.yudao.module.iot.controller.admin.device.vo.control.IotDeviceUpstreamReqVO;
 import cn.iocoder.yudao.module.iot.controller.admin.device.vo.device.*;
+import cn.iocoder.yudao.module.iot.controller.admin.device.vo.message.IotDeviceMessageSendReqVO;
+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.service.device.IotDeviceService;
-import cn.iocoder.yudao.module.iot.service.device.control.IotDeviceDownstreamService;
-import cn.iocoder.yudao.module.iot.service.device.control.IotDeviceUpstreamService;
+import cn.iocoder.yudao.module.iot.service.device.message.IotDeviceMessageService;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.Parameter;
 import io.swagger.v3.oas.annotations.tags.Tag;
@@ -42,9 +41,7 @@ public class IotDeviceController {
     @Resource
     private IotDeviceService deviceService;
     @Resource
-    private IotDeviceUpstreamService deviceUpstreamService;
-    @Resource
-    private IotDeviceDownstreamService deviceDownstreamService;
+    private IotDeviceMessageService deviceMessageService;
  
     @PostMapping("/create")
     @Operation(summary = "创建设备")
@@ -161,19 +158,12 @@ public class IotDeviceController {
         ExcelUtils.write(response, "设备导入模板.xls", "数据", IotDeviceImportExcelVO.class, list);
     }
  
-    @PostMapping("/upstream")
-    @Operation(summary = "设备上行", description = "可用于设备模拟")
+    // TODO @芋艿：需要重构
+    @PostMapping("/send-message")
+    @Operation(summary = "发送消息", description = "可用于设备模拟")
     @PreAuthorize("@ss.hasPermission('iot:device:upstream')")
-    public CommonResult<Boolean> upstreamDevice(@Valid @RequestBody IotDeviceUpstreamReqVO upstreamReqVO) {
-        deviceUpstreamService.upstreamDevice(upstreamReqVO);
-        return success(true);
-    }
-
-    @PostMapping("/downstream")
-    @Operation(summary = "设备下行", description = "可用于设备模拟")
-    @PreAuthorize("@ss.hasPermission('iot:device:downstream')")
-    public CommonResult<Boolean> downstreamDevice(@Valid @RequestBody IotDeviceDownstreamReqVO downstreamReqVO) {
-        deviceDownstreamService.downstreamDevice(downstreamReqVO);
+    public CommonResult<Boolean> upstreamDevice(@Valid @RequestBody IotDeviceMessageSendReqVO sendReqVO) {
+        deviceMessageService.sendDeviceMessage(BeanUtils.toBean(sendReqVO, IotDeviceMessage.class));
         return success(true);
     }
  
@@ -5,8 +5,8 @@ import cn.iocoder.yudao.framework.common.pojo.PageResult;
 import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
 import cn.iocoder.yudao.module.iot.controller.admin.device.vo.data.IotDeviceLogPageReqVO;
 import cn.iocoder.yudao.module.iot.controller.admin.device.vo.data.IotDeviceLogRespVO;
-import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceLogDO;
-import cn.iocoder.yudao.module.iot.service.device.data.IotDeviceLogService;
+import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceMessageDO;
+import cn.iocoder.yudao.module.iot.service.device.property.IotDeviceLogService;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.tags.Tag;
 import jakarta.annotation.Resource;
@@ -32,7 +32,7 @@ public class IotDeviceLogController {
     @Operation(summary = "获得设备日志分页")
     @PreAuthorize("@ss.hasPermission('iot:device:log-query')")
     public CommonResult<PageResult<IotDeviceLogRespVO>> getDeviceLogPage(@Valid IotDeviceLogPageReqVO pageReqVO) {
-        PageResult<IotDeviceLogDO> pageResult = deviceLogService.getDeviceLogPage(pageReqVO);
+        PageResult<IotDeviceMessageDO> pageResult = deviceLogService.getDeviceLogPage(pageReqVO);
         return success(BeanUtils.toBean(pageResult, IotDeviceLogRespVO.class));
     }
  
@@ -12,7 +12,7 @@ import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
 import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDevicePropertyDO;
 import cn.iocoder.yudao.module.iot.dal.dataobject.thingmodel.IotThingModelDO;
 import cn.iocoder.yudao.module.iot.service.device.IotDeviceService;
-import cn.iocoder.yudao.module.iot.service.device.data.IotDevicePropertyService;
+import cn.iocoder.yudao.module.iot.service.device.property.IotDevicePropertyService;
 import cn.iocoder.yudao.module.iot.service.thingmodel.IotThingModelService;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.Parameter;
-package cn.iocoder.yudao.module.iot.controller.admin.device.vo.control;
-
-import cn.iocoder.yudao.framework.common.validation.InEnum;
-import cn.iocoder.yudao.module.iot.enums.device.IotDeviceMessageTypeEnum;
-import io.swagger.v3.oas.annotations.media.Schema;
-import jakarta.validation.constraints.NotEmpty;
-import jakarta.validation.constraints.NotNull;
-import lombok.Data;
-
-@Schema(description = "管理后台 - IoT 设备下行 Request VO") // 服务调用、属性设置、属性获取等
-@Data
-public class IotDeviceDownstreamReqVO {
-
-    @Schema(description = "设备编号", requiredMode = Schema.RequiredMode.REQUIRED, example = "177")
-    @NotNull(message = "设备编号不能为空")
-    private Long id;
-
-    @Schema(description = "消息类型", requiredMode = Schema.RequiredMode.REQUIRED, example = "property")
-    @NotEmpty(message = "消息类型不能为空")
-    @InEnum(IotDeviceMessageTypeEnum.class)
-    private String type;
-
-    @Schema(description = "标识符", requiredMode = Schema.RequiredMode.REQUIRED, example = "report")
-    @NotEmpty(message = "标识符不能为空")
-    private String identifier; // 参见 IotDeviceMessageIdentifierEnum 枚举类
-
-    @Schema(description = "请求参数", requiredMode = Schema.RequiredMode.REQUIRED)
-    private Object data; // 例如说：服务调用的 params、属性设置的 properties
-
-}
-package cn.iocoder.yudao.module.iot.controller.admin.device.vo.control;
-
-import cn.iocoder.yudao.framework.common.validation.InEnum;
-import cn.iocoder.yudao.module.iot.enums.device.IotDeviceMessageTypeEnum;
-import io.swagger.v3.oas.annotations.media.Schema;
-import jakarta.validation.constraints.NotEmpty;
-import jakarta.validation.constraints.NotNull;
-import lombok.Data;
-
-@Schema(description = "管理后台 - IoT 设备上行 Request VO") // 属性上报、事件上报、状态变更等
-@Data
-public class IotDeviceUpstreamReqVO {
-
-    @Schema(description = "设备编号", requiredMode = Schema.RequiredMode.REQUIRED, example = "177")
-    @NotNull(message = "设备编号不能为空")
-    private Long id;
-
-    @Schema(description = "消息类型", requiredMode = Schema.RequiredMode.REQUIRED, example = "property")
-    @NotEmpty(message = "消息类型不能为空")
-    @InEnum(IotDeviceMessageTypeEnum.class)
-    private String type;
-
-    @Schema(description = "标识符", requiredMode = Schema.RequiredMode.REQUIRED, example = "report")
-    @NotEmpty(message = "标识符不能为空")
-    private String identifier; // 参见 IotDeviceMessageIdentifierEnum 枚举类
-
-    @Schema(description = "请求参数", requiredMode = Schema.RequiredMode.REQUIRED)
-    private Object data; // 例如说：属性上报的 properties、事件上报的 params
-
-}
+package cn.iocoder.yudao.module.iot.controller.admin.device.vo.message;
+
+import cn.iocoder.yudao.framework.common.validation.InEnum;
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import io.swagger.v3.oas.annotations.media.Schema;
+import jakarta.validation.constraints.NotEmpty;
+import jakarta.validation.constraints.NotNull;
+import lombok.Data;
+
+@Schema(description = "管理后台 - IoT 设备消息发送 Request VO") // 属性上报、事件上报、状态变更等
+@Data
+public class IotDeviceMessageSendReqVO {
+
+    @Schema(description = "请求方法", requiredMode = Schema.RequiredMode.REQUIRED, example = "report")
+    @NotEmpty(message = "请求方法不能为空")
+    @InEnum(IotDeviceMessageMethodEnum.class)
+    private String method;
+
+    @Schema(description = "请求参数")
+    private Object params; // 例如说：属性上报的 properties、事件上报的 params
+
+    @Schema(description = "设备编号", requiredMode = Schema.RequiredMode.REQUIRED, example = "177")
+    @NotNull(message = "设备编号不能为空")
+    private Long deviceId;
+
+}
 package cn.iocoder.yudao.module.iot.controller.admin.ota.vo.upgrade.record;
  
 import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceMessageDO;
 import cn.iocoder.yudao.module.iot.dal.dataobject.ota.IotOtaFirmwareDO;
 import cn.iocoder.yudao.module.iot.dal.dataobject.ota.IotOtaUpgradeTaskDO;
 import com.fhs.core.trans.anno.Trans;
@@ -89,7 +90,7 @@ public class IotOtaUpgradeRecordRespVO {
      * 升级进度描述
      * <p>
      * 注意，只记录设备最后一次的升级进度描述
-     * 如果想看历史记录，可以查看 {@link cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceLogDO} 设备日志
+     * 如果想看历史记录，可以查看 {@link IotDeviceMessageDO} 设备日志
      */
     @Schema(description = "升级进度描述", requiredMode = Schema.RequiredMode.REQUIRED, example = "10")
     private String description;
@@ -6,7 +6,7 @@ import cn.iocoder.yudao.module.iot.controller.admin.statistics.vo.IotStatisticsR
 import cn.iocoder.yudao.module.iot.controller.admin.statistics.vo.IotStatisticsSummaryRespVO;
 import cn.iocoder.yudao.module.iot.core.enums.IotDeviceStateEnum;
 import cn.iocoder.yudao.module.iot.service.device.IotDeviceService;
-import cn.iocoder.yudao.module.iot.service.device.data.IotDeviceLogService;
+import cn.iocoder.yudao.module.iot.service.device.property.IotDeviceLogService;
 import cn.iocoder.yudao.module.iot.service.product.IotProductCategoryService;
 import cn.iocoder.yudao.module.iot.service.product.IotProductService;
 import io.swagger.v3.oas.annotations.Operation;
 package cn.iocoder.yudao.module.iot.dal.dataobject.device;
  
-import cn.hutool.core.util.IdUtil;
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
-import cn.iocoder.yudao.module.iot.dal.dataobject.product.IotProductDO;
-import cn.iocoder.yudao.module.iot.enums.device.IotDeviceMessageIdentifierEnum;
-import cn.iocoder.yudao.module.iot.enums.device.IotDeviceMessageTypeEnum;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
 import lombok.NoArgsConstructor;
  
 /**
- * IoT 设备日志数据 DO
+ * IoT 设备消息数据 DO
  *
  * 目前使用 TDengine 存储
  *
@@ -21,80 +18,84 @@ import lombok.NoArgsConstructor;
 @Builder
 @NoArgsConstructor
 @AllArgsConstructor
-public class IotDeviceLogDO {
-
-    /**
-     * 日志编号
-     *
-     * 通过 {@link IdUtil#fastSimpleUUID()} 生成
-     */
-    private String id;
+public class IotDeviceMessageDO {
  
     /**
      * 消息编号
-     *
-     * 对应 {@link IotDeviceMessage#getMessageId()} 字段
      */
-    private String messageId;
-
+    private String id;
     /**
-     * 产品标识
-     * <p>
-     * 关联 {@link IotProductDO#getProductKey()}
+     * 上报时间戳
      */
-    private String productKey;
+    private Long reportTime;
     /**
-     * 设备名称
-     *
-     * 关联 {@link IotDeviceDO#getDeviceName()}
+     * 存储时序时间
      */
-    private String deviceName;
+    private Long ts;
+
     /**
      * 设备编号
      *
      * 关联 {@link IotDeviceDO#getId()}
      */
     private Long deviceId;
+    /**
+     * 租户编号
+     */
+    private Long tenantId;
  
     /**
-     * 日志类型
+     * 服务编号，该消息由哪个 server 发送
+     */
+    private String serverId;
+
+    /**
+     * 是否上行消息
      *
-     * 枚举 {@link IotDeviceMessageTypeEnum}
+     * 由 {@link cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils#isUpstreamMessage(IotDeviceMessage)} 计算。
+     * 计算并存储的目的：方便计算多少条上行、多少条下行
      */
-    private String type;
+    private Boolean upstream;
     /**
-     * 标识符
+     * 是否回复消息
      *
-     * 枚举 {@link IotDeviceMessageIdentifierEnum}
+     * 由 {@link cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils#isReplyMessage(IotDeviceMessage)} 计算。
+     * 计算并存储的目的：方便计算多少条请求、多少条回复
      */
-    private String identifier;
+    private Boolean reply;
+
+    // ========== codec（编解码）字段 ==========
  
     /**
-     * 数据内容
+     * 请求编号
      *
-     * 存储具体的消息数据内容，通常是 JSON 格式
+     * 由设备生成，对应阿里云 IoT 的 Alink 协议中的 id、华为云 IoTDA 协议的 request_id
      */
-    private String content;
+    private String requestId;
     /**
-     * 响应码
+     * 请求方法
      *
-     * 目前只有 server 下行消息给 device 设备时，才会有响应码
+     * 枚举 {@link IotDeviceMessageMethodEnum}
+     * 例如说：thing.property.report 属性上报
      */
-    private Integer code;
-
+    private String method;
     /**
-     * 上报时间戳
+     * 请求参数
+     *
+     * 例如说：属性上报的 properties、事件上报的 params
      */
-    private Long reportTime;
-
+    private Object params;
     /**
-     * 租户编号
+     * 响应结果
      */
-    private Long tenantId;
-
+    private Object data;
     /**
-     * 时序时间
+     * 响应错误码
      */
-    private Long ts;
+    private Integer code;
+    /**
+     * 响应提示
+     */
+    private String msg;
  
 }
@@ -2,6 +2,7 @@ package cn.iocoder.yudao.module.iot.dal.dataobject.ota;
  
 import cn.iocoder.yudao.framework.mybatis.core.dataobject.BaseDO;
 import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceMessageDO;
 import com.baomidou.mybatisplus.annotation.KeySequence;
 import com.baomidou.mybatisplus.annotation.TableId;
 import com.baomidou.mybatisplus.annotation.TableName;
@@ -77,7 +78,7 @@ public class IotOtaUpgradeRecordDO extends BaseDO {
      * 升级进度描述
      *
      * 注意，只记录设备最后一次的升级进度描述
-     * 如果想看历史记录，可以查看 {@link cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceLogDO} 设备日志
+     * 如果想看历史记录，可以查看 {@link IotDeviceMessageDO} 设备日志
      */
     private String description;
     /**
@@ -43,17 +43,6 @@ public class DeviceServerIdRedisDAO {
     }
  
     /**
-     * 删除设备关联的网关 serverId
-     *
-     * @param productKey 产品标识
-     * @param deviceName 设备名称
-     */
-    public void delete(String productKey, String deviceName) {
-        String hashKey = buildHashKey(productKey, deviceName);
-        stringRedisTemplate.opsForHash().delete(RedisKeyConstants.DEVICE_SERVER_ID, hashKey);
-    }
-
-    /**
      * 构建 HASH KEY
      *
      * @param productKey 产品标识
@@ -64,4 +53,4 @@ public class DeviceServerIdRedisDAO {
         return productKey + StrUtil.COMMA + deviceName;
     }
  
-} 
 \ No newline at end of file
+}
 \ No newline at end of file
 package cn.iocoder.yudao.module.iot.dal.tdengine;
  
 import cn.iocoder.yudao.module.iot.controller.admin.device.vo.data.IotDeviceLogPageReqVO;
-import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceLogDO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceMessageDO;
 import cn.iocoder.yudao.module.iot.framework.tdengine.core.annotation.TDengineDS;
 import com.baomidou.mybatisplus.annotation.InterceptorIgnore;
 import com.baomidou.mybatisplus.core.metadata.IPage;
@@ -12,48 +12,48 @@ import java.util.List;
 import java.util.Map;
  
 /**
- * 设备日志 {@link IotDeviceLogDO} Mapper 接口
+ * 设备消息 {@link IotDeviceMessageDO} Mapper 接口
  */
 @Mapper
 @TDengineDS
 @InterceptorIgnore(tenantLine = "true") // 避免 SQL 解析，因为 JSqlParser 对 TDengine 的 SQL 解析会报错
-public interface IotDeviceLogMapper {
+public interface IotDeviceMessageMapper {
  
     /**
-     * 创建设备日志超级表
+     * 创建设备消息超级表
      */
-    void createDeviceLogSTable();
+    void createSTable();
  
     /**
-     * 查询设备日志表是否存在
+     * 查询设备消息表是否存在
      *
      * @return 存在则返回表名；不存在则返回 null
      */
-    String showDeviceLogSTable();
+    String showSTable();
  
     /**
-     * 插入设备日志数据
+     * 插入设备消息数据
      *
      * 如果子表不存在，会自动创建子表
      *
-     * @param log 设备日志数据
+     * @param message 设备消息数据
      */
-    void insert(IotDeviceLogDO log);
+    void insert(IotDeviceMessageDO message);
  
     /**
-     * 获得设备日志分页
+     * 获得设备消息分页
      *
      * @param reqVO 分页查询条件
-     * @return 设备日志列表
+     * @return 设备消息列表
      */
-    IPage<IotDeviceLogDO> selectPage(IPage<IotDeviceLogDO> page,
-                                     @Param("reqVO") IotDeviceLogPageReqVO reqVO);
+    IPage<IotDeviceMessageDO> selectPage(IPage<IotDeviceMessageDO> page,
+                                         @Param("reqVO") IotDeviceLogPageReqVO reqVO);
  
     /**
-     * 统计设备日志数量
+     * 统计设备消息数量
      *
-     * @param createTime 创建时间，如果为空，则统计所有日志数量
-     * @return 日志数量
+     * @param createTime 创建时间，如果为空，则统计所有消息数量
+     * @return 消息数量
      */
     Long selectCountByCreateTime(@Param("createTime") Long createTime);
  
@@ -62,14 +62,14 @@ public interface IotDeviceLogMapper {
     /**
      * 查询每个小时设备上行消息数量
      */
-    List<Map<String, Object>> selectDeviceLogUpCountByHour(@Param("deviceKey") String deviceKey,
+    List<Map<String, Object>> selectDeviceLogUpCountByHour(@Param("deviceId") Long deviceId,
                                                            @Param("startTime") Long startTime,
                                                            @Param("endTime") Long endTime);
  
     /**
      * 查询每个小时设备下行消息数量
      */
-    List<Map<String, Object>> selectDeviceLogDownCountByHour(@Param("deviceKey") String deviceKey,
+    List<Map<String, Object>> selectDeviceLogDownCountByHour(@Param("deviceId") Long deviceId,
                                                              @Param("startTime") Long startTime,
                                                              @Param("endTime") Long endTime);
  
 package cn.iocoder.yudao.module.iot.framework.tdengine.config;
  
-import cn.iocoder.yudao.module.iot.service.device.data.IotDeviceLogService;
+import cn.iocoder.yudao.module.iot.service.device.message.IotDeviceMessageService;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.ApplicationArguments;
@@ -17,16 +17,16 @@ import org.springframework.stereotype.Component;
 @Slf4j
 public class TDengineTableInitRunner implements ApplicationRunner {
  
-    private final IotDeviceLogService deviceLogService;
+    private final IotDeviceMessageService deviceMessageService;
  
     @Override
     public void run(ApplicationArguments args) {
         try {
-            // 初始化设备日志表
-            deviceLogService.defineDeviceLog();
+            // 初始化设备消息表
+            deviceMessageService.defineDeviceMessageStable();
         } catch (Exception ex) {
-            // 初始化失败时打印错误日志并退出系统
-            log.error("[run][TDengine初始化设备日志表结构失败，系统无法正常运行，即将退出]", ex);
+            // 初始化失败时打印错误消息并退出系统
+            log.error("[run][TDengine初始化设备消息表结构失败，系统无法正常运行，即将退出]", ex);
             System.exit(1);
         }
     }
@@ -7,10 +7,10 @@ import cn.iocoder.yudao.framework.quartz.core.handler.JobHandler;
 import cn.iocoder.yudao.framework.tenant.core.job.TenantJob;
 import cn.iocoder.yudao.module.iot.core.enums.IotDeviceStateEnum;
 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
-import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer;
 import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
 import cn.iocoder.yudao.module.iot.service.device.IotDeviceService;
-import cn.iocoder.yudao.module.iot.service.device.data.IotDevicePropertyService;
+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 org.springframework.stereotype.Component;
  
@@ -43,10 +43,10 @@ public class IotDeviceOfflineCheckJob implements JobHandler {
     private IotDeviceService deviceService;
     @Resource
     private IotDevicePropertyService devicePropertyService;
-
     @Resource
-    private IotDeviceMessageProducer deviceMessageProducer;
+    private IotDeviceMessageService deviceMessageService;
  
+    // TODO @芋艿：需要重构下；
     @Override
     @TenantJob
     public String execute(String param) {
@@ -69,8 +69,7 @@ public class IotDeviceOfflineCheckJob implements JobHandler {
             }
             offlineDeviceKeys.add(new String[]{device.getProductKey(), device.getDeviceName()});
             // 为什么不直接更新状态呢？因为通过 IotDeviceMessage 可以经过一系列的处理，例如说记录日志等等
-            deviceMessageProducer.sendDeviceMessage(IotDeviceMessage.of(device.getProductKey(), device.getDeviceName())
-                    .ofStateOffline());
+            deviceMessageService.sendDeviceMessage(IotDeviceMessage.buildStateOffline().setDeviceId(device.getId()));
         }
         return JsonUtils.toJsonString(offlineDeviceKeys);
     }
-package cn.iocoder.yudao.module.iot.mq.consumer.device;
-
-import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
-import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber;
-import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
-import cn.iocoder.yudao.module.iot.service.device.data.IotDeviceLogService;
-import jakarta.annotation.PostConstruct;
-import jakarta.annotation.Resource;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.stereotype.Component;
-
-/**
- * 针对 {@link IotDeviceMessage} 的消费者：记录设备日志
- *
- * @author 芋道源码
- */
-@Component
-@Slf4j
-public class IotDeviceLogMessageSubscriber implements IotMessageSubscriber<IotDeviceMessage> {
-
-    @Resource
-    private IotMessageBus messageBus;
-
-    @Resource
-    private IotDeviceLogService deviceLogService;
-
-    @PostConstruct
-    public void init() {
-        messageBus.register(this);
-    }
-
-    @Override
-    public String getTopic() {
-        return IotDeviceMessage.MESSAGE_BUS_DEVICE_MESSAGE_TOPIC;
-    }
-
-    @Override
-    public String getGroup() {
-        return "iot_device_log_consumer";
-    }
-
-    @Override
-    public void onMessage(IotDeviceMessage message) {
-        log.info("[onMessage][消息内容({})]", message);
-        deviceLogService.createDeviceLog(message);
-    }
-
-}
 package cn.iocoder.yudao.module.iot.mq.consumer.device;
  
 import cn.hutool.core.util.ObjectUtil;
-import cn.iocoder.yudao.framework.tenant.core.util.TenantUtils;
-import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageIdentifierEnum;
-import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageTypeEnum;
+import cn.iocoder.yudao.framework.common.util.object.ObjectUtils;
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
 import cn.iocoder.yudao.module.iot.core.enums.IotDeviceStateEnum;
 import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
 import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber;
 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
-import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer;
 import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
 import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
 import cn.iocoder.yudao.module.iot.service.device.IotDeviceService;
-import cn.iocoder.yudao.module.iot.service.device.data.IotDevicePropertyService;
+import cn.iocoder.yudao.module.iot.service.device.message.IotDeviceMessageService;
+import cn.iocoder.yudao.module.iot.service.device.property.IotDevicePropertyService;
 import jakarta.annotation.PostConstruct;
 import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
  
 import java.time.LocalDateTime;
-import java.util.Objects;
  
 /**
- * 针对 {@link IotDeviceMessage} 的消费者：记录设备状态
+ * 针对 {@link IotDeviceMessage} 的业务处理器：调用 method 对应的逻辑。例如说：
+ * 1. {@link IotDeviceMessageMethodEnum#PROPERTY_REPORT} 属性上报时，记录设备属性
  *
- * 特殊：如果是离线的设备，将自动上线
- *
- * @author 芋道源码
+ * @author alwayssuper
  */
 @Component
 @Slf4j
-public class IotDeviceStateMessageSubscriber implements IotMessageSubscriber<IotDeviceMessage> {
+public class IotDeviceMessageSubscriber implements IotMessageSubscriber<IotDeviceMessage> {
  
     @Resource
     private IotDeviceService deviceService;
     @Resource
     private IotDevicePropertyService devicePropertyService;
+    @Resource
+    private IotDeviceMessageService deviceMessageService;
  
     @Resource
     private IotMessageBus messageBus;
-    @Resource
-    private IotDeviceMessageProducer deviceMessageProducer;
  
     @PostConstruct
     public void init() {
@@ -54,53 +51,49 @@ public class IotDeviceStateMessageSubscriber implements IotMessageSubscriber&lt;Iot
  
     @Override
     public String getGroup() {
-        return "iot_device_state_consumer";
+        return "iot_device_message_consumer";
     }
  
     @Override
     public void onMessage(IotDeviceMessage message) {
-        // 1.1 只处理上行消息，或者是 STATE 相关的消息
-        if (!IotDeviceMessageUtils.isUpstreamMessage(message)
-            && ObjectUtil.notEqual(message.getType(), IotDeviceMessageTypeEnum.STATE.getType())) {
+        if (!IotDeviceMessageUtils.isUpstreamMessage(message)) {
+            log.error("[onMessage][message({}) 非上行消息，不进行处理]", message);
             return;
         }
-        // 1.2 校验设备是否存在
-        IotDeviceDO device = deviceService.getDeviceByProductKeyAndDeviceNameFromCache(
-                message.getProductKey(), message.getDeviceName());
-        if (device == null) {
-            log.error("[onMessage][消息({}) 对应的设备部存在]", message);
-            return;
-        }
-
-        // 2. 处理消息
-        TenantUtils.execute(device.getTenantId(), () -> onMessage(message, device));
-    }
  
-    private void onMessage(IotDeviceMessage message, IotDeviceDO device) {
-        // 更新设备的最后时间
+        // 1.1 更新设备的最后时间
+        // TODO 芋艿：后续加缓存；
+        IotDeviceDO device = deviceService.validateDeviceExists(message.getDeviceId());
         devicePropertyService.updateDeviceReportTime(device.getProductKey(), device.getDeviceName(), LocalDateTime.now());
+        // 1.2 更新设备的连接 server
+        devicePropertyService.updateDeviceServerId(device.getProductKey(), device.getDeviceName(), message.getServerId());
+
+        // 2. 未上线的设备，强制上线
+        forceDeviceOnline(message, device);
+
+        // 3. 核心：处理消息
+        deviceMessageService.handleUpstreamDeviceMessage(message, device);
+    }
  
-        // 情况一：STATE 相关的消息
-        if (Objects.equals(message.getType(), IotDeviceMessageTypeEnum.STATE.getType())) {
-            if (Objects.equals(message.getIdentifier(), IotDeviceMessageIdentifierEnum.STATE_ONLINE.getIdentifier())) {
-                deviceService.updateDeviceState(device.getId(), IotDeviceStateEnum.ONLINE.getState());
-                devicePropertyService.updateDeviceServerId(device.getProductKey(), device.getDeviceName(), message.getServerId());
-            } else {
-                deviceService.updateDeviceState(device.getId(), IotDeviceStateEnum.OFFLINE.getState());
-                devicePropertyService.deleteDeviceServerId(device.getProductKey(), device.getDeviceName());
-            }
-            // TODO 芋艿：子设备的关联
+    private void forceDeviceOnline(IotDeviceMessage message, IotDeviceDO device) {
+        // 已经在线，无需处理
+        if (ObjectUtil.equal(device.getState(), IotDeviceStateEnum.ONLINE.getState())) {
+            return;
+        }
+        // 如果是 STATE 相关的消息，无需处理，不然就重复处理状态了
+        if (ObjectUtils.equalsAny(message.getMethod(), IotDeviceMessageMethodEnum.STATE_ONLINE.getMethod(),
+                IotDeviceMessageMethodEnum.STATE_OFFLINE.getMethod())) {
             return;
         }
  
-        // 情况二：非 STATE 相关的消息
-        devicePropertyService.updateDeviceServerId(device.getProductKey(), device.getDeviceName(), message.getServerId());
         // 特殊：设备非在线时，主动标记设备为在线
-        // 为什么不直接更新状态呢？因为通过 IotDeviceMessage 可以经过一系列的处理，例如说记录日志等等
-        if (ObjectUtil.notEqual(device.getState(), IotDeviceStateEnum.ONLINE.getState())) {
-            deviceMessageProducer.sendDeviceMessage(IotDeviceMessage.of(message.getProductKey(), message.getDeviceName())
-                    .ofStateOnline());
+        // 为什么不直接更新状态呢？因为通过 IotDeviceMessage 可以经过一系列的处理，例如说记录日志、规则引擎等等
+        try {
+            deviceMessageService.sendDeviceMessage(IotDeviceMessage.buildStateOnline().setDeviceId(device.getId()));
+        } catch (Exception e) {
+            // 注意：即使执行失败，也不影响主流程
+            log.error("[forceDeviceOnline][message({}) device({}) 强制设备上线失败]", message, device, e);
         }
     }
  
-}
 \ No newline at end of file
+}
-package cn.iocoder.yudao.module.iot.mq.consumer.device;
-
-import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
-import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber;
-import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
-import cn.iocoder.yudao.module.iot.enums.device.IotDeviceMessageIdentifierEnum;
-import cn.iocoder.yudao.module.iot.enums.device.IotDeviceMessageTypeEnum;
-import cn.iocoder.yudao.module.iot.service.device.data.IotDevicePropertyService;
-import com.google.common.base.Objects;
-import jakarta.annotation.PostConstruct;
-import jakarta.annotation.Resource;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.stereotype.Component;
-
-/**
- * 针对 {@link IotDeviceMessage} 的消费者：记录设备属性
- *
- * @author alwayssuper
- */
-@Component
-@Slf4j
-public class IotDevicePropertyMessageSubscriber implements IotMessageSubscriber<IotDeviceMessage> {
-
-    @Resource
-    private IotDevicePropertyService deviceDataService;
-
-    @Resource
-    private IotMessageBus messageBus;
-
-    @PostConstruct
-    public void init() {
-        messageBus.register(this);
-    }
-
-    @Override
-    public String getTopic() {
-        return IotDeviceMessage.MESSAGE_BUS_DEVICE_MESSAGE_TOPIC;
-    }
-
-    @Override
-    public String getGroup() {
-        return "iot_device_property_consumer";
-    }
-
-    @Override
-    public void onMessage(IotDeviceMessage message) {
-        if (Objects.equal(message.getType(), IotDeviceMessageTypeEnum.PROPERTY.getType())
-                && Objects.equal(message.getIdentifier(), IotDeviceMessageIdentifierEnum.PROPERTY_REPORT.getIdentifier())) {
-            // 保存设备属性
-            deviceDataService.saveDeviceProperty(message);
-        }
-    }
-
-}
@@ -63,6 +63,14 @@ public interface IotDeviceService {
     /**
      * 更新设备状态
      *
+     * @param device 设备信息
+     * @param state 状态
+     */
+    void updateDeviceState(IotDeviceDO device, Integer state);
+
+    /**
+     * 更新设备状态
+     *
      * @param id    编号
      * @param state 状态
      */
@@ -272,12 +272,9 @@ public class IotDeviceServiceImpl implements IotDeviceService {
     }
  
     @Override
-    public void updateDeviceState(Long id, Integer state) {
-        // 1. 校验存在
-        IotDeviceDO device = validateDeviceExists(id);
-
-        // 2. 更新状态和时间
-        IotDeviceDO updateObj = new IotDeviceDO().setId(id).setState(state);
+    public void updateDeviceState(IotDeviceDO device, Integer state) {
+        // 1. 更新状态和时间
+        IotDeviceDO updateObj = new IotDeviceDO().setId(device.getId()).setState(state);
         if (device.getOnlineTime() == null
                 && Objects.equals(state, IotDeviceStateEnum.ONLINE.getState())) {
             updateObj.setActiveTime(LocalDateTime.now());
@@ -289,11 +286,19 @@ public class IotDeviceServiceImpl implements IotDeviceService {
         }
         deviceMapper.updateById(updateObj);
  
-        // 3. 清空对应缓存
+        // 2. 清空对应缓存
         deleteDeviceCache(device);
     }
  
     @Override
+    public void updateDeviceState(Long id, Integer state) {
+        // 校验存在
+        IotDeviceDO device = validateDeviceExists(id);
+        // 执行更新
+        updateDeviceState(device, state);
+    }
+
+    @Override
     public Long getDeviceCountByProductId(Long productId) {
         return deviceMapper.selectCountByProductId(productId);
     }
-package cn.iocoder.yudao.module.iot.service.device.control;
-
-import cn.iocoder.yudao.module.iot.controller.admin.device.vo.control.IotDeviceDownstreamReqVO;
-import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
-import jakarta.validation.Valid;
-
-/**
- * IoT 设备下行 Service 接口
- *
- * 目的：服务端 -> 网关 -> 设备
- *
- * @author 芋道源码
- */
-public interface IotDeviceDownstreamService {
-
-    /**
-     * 设备下行，可用于设备模拟
-     *
-     * @param downstreamReqVO 设备下行请求 VO
-     * @return 下行消息
-     */
-    IotDeviceMessage downstreamDevice(@Valid IotDeviceDownstreamReqVO downstreamReqVO);
-
-}
-package cn.iocoder.yudao.module.iot.service.device.control;
-
-import cn.hutool.core.util.StrUtil;
-import cn.iocoder.yudao.framework.common.exception.ServiceException;
-import cn.iocoder.yudao.module.iot.controller.admin.device.vo.control.IotDeviceDownstreamReqVO;
-import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
-import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer;
-import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
-import cn.iocoder.yudao.module.iot.enums.device.IotDeviceMessageIdentifierEnum;
-import cn.iocoder.yudao.module.iot.enums.device.IotDeviceMessageTypeEnum;
-import cn.iocoder.yudao.module.iot.service.device.IotDeviceService;
-import cn.iocoder.yudao.module.iot.service.device.data.IotDevicePropertyService;
-import jakarta.annotation.Resource;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.stereotype.Service;
-import org.springframework.validation.annotation.Validated;
-
-import java.util.Map;
-import java.util.Objects;
-
-import static cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants.BAD_REQUEST;
-import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
-import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.DEVICE_DOWNSTREAM_FAILED_SERVER_ID_NULL;
-
-/**
- * IoT 设备下行 Service 实现类
- *
- * @author 芋道源码
- */
-@Service
-@Validated
-@Slf4j
-public class IotDeviceDownstreamServiceImpl implements IotDeviceDownstreamService {
-
-    @Resource
-    private IotDeviceService deviceService;
-    @Resource
-    private IotDevicePropertyService devicePropertyService;
-
-    @Resource
-    private IotDeviceMessageProducer deviceMessageProducer;
-
-    @Override
-    public IotDeviceMessage downstreamDevice(IotDeviceDownstreamReqVO downstreamReqVO) {
-        // 1. 校验设备是否存在
-        IotDeviceDO device = deviceService.validateDeviceExists(downstreamReqVO.getId());
-        // TODO 芋艿：父设备的处理
-        IotDeviceDO parentDevice = null;
-
-        // 2. 构建消息
-        IotDeviceMessage message = buildDownstreamDeviceMessage(downstreamReqVO, device, parentDevice);
-
-        // 3.1 发送给网关
-        String serverId = devicePropertyService.getDeviceServerId(message.getProductKey(), message.getDeviceName());
-        if (StrUtil.isEmpty(serverId)) {
-            throw exception(DEVICE_DOWNSTREAM_FAILED_SERVER_ID_NULL);
-        }
-        deviceMessageProducer.sendDeviceMessageToGateway(serverId, message);
-
-        // 3.2 发送给服务器（用于设备日志等的记录）
-        deviceMessageProducer.sendDeviceMessage(message);
-        return message;
-    }
-
-    @SuppressWarnings("unchecked")
-    private IotDeviceMessage buildDownstreamDeviceMessage(IotDeviceDownstreamReqVO downstreamReqVO,
-                                                          IotDeviceDO device, IotDeviceDO parentDevice) {
-        IotDeviceMessage message = IotDeviceMessage.of(getProductKey(device, parentDevice),
-                getDeviceName(device, parentDevice));
-        // 服务调用
-        if (Objects.equals(downstreamReqVO.getType(), IotDeviceMessageTypeEnum.SERVICE.getType())) {
-            // TODO @芋艿：待实现
-//            return invokeDeviceService(downstreamReqVO, device, parentDevice);
-        }
-        // 属性相关
-        if (Objects.equals(downstreamReqVO.getType(), IotDeviceMessageTypeEnum.PROPERTY.getType())) {
-            // 属性设置
-            if (Objects.equals(downstreamReqVO.getIdentifier(),
-                    IotDeviceMessageIdentifierEnum.PROPERTY_SET.getIdentifier())) {
-                if (!(downstreamReqVO.getData() instanceof Map<?, ?>)) {
-                    throw new ServiceException(BAD_REQUEST.getCode(), "data 不是 Map 类型");
-                }
-                return message.ofPropertySet((Map<String, Object>) downstreamReqVO.getData());
-            }
-            // 属性获取
-            if (Objects.equals(downstreamReqVO.getIdentifier(),
-                    IotDeviceMessageIdentifierEnum.PROPERTY_GET.getIdentifier())) {
-                // TODO @芋艿：待实现
-//                return getDeviceProperty(downstreamReqVO, device, parentDevice);
-            }
-        }
-        // 配置下发
-        if (Objects.equals(downstreamReqVO.getType(), IotDeviceMessageTypeEnum.CONFIG.getType())
-                && Objects.equals(downstreamReqVO.getIdentifier(),
-                IotDeviceMessageIdentifierEnum.CONFIG_SET.getIdentifier())) {
-            // TODO @芋艿：待实现
-//            return setDeviceConfig(downstreamReqVO, device, parentDevice);
-        }
-        // OTA 升级
-        if (Objects.equals(downstreamReqVO.getType(), IotDeviceMessageTypeEnum.OTA.getType())) {
-            // TODO @芋艿：待实现
-//            return otaUpgrade(downstreamReqVO, device, parentDevice);
-        }
-        // TODO @芋艿：取消设备的网关的时，要不要下发 REGISTER_UNREGISTER_SUB ？
-        throw new IllegalArgumentException("不支持的下行消息类型：" + downstreamReqVO);
-    }
-
-//    /**
-//     * 调用设备服务
-//     *
-//     * @param downstreamReqVO 下行请求
-//     * @param device          设备
-//     * @param parentDevice    父设备
-//     * @return 下发消息
-//     */
-//    @SuppressWarnings("unchecked")
-//    private IotDeviceMessage invokeDeviceService(IotDeviceDownstreamReqVO downstreamReqVO,
-//                                                 IotDeviceDO device, IotDeviceDO parentDevice) {
-//        // 1. 参数校验
-//        if (!(downstreamReqVO.getData() instanceof Map<?, ?>)) {
-//            throw new ServiceException(BAD_REQUEST.getCode(), "data 不是 Map 类型");
-//        }
-//        // TODO @super：【可优化】过滤掉不合法的服务
-//
-//        // 2. 发送请求
-//        String url = String.format("sys/%s/%s/thing/service/%s",
-//                getProductKey(device, parentDevice), getDeviceName(device, parentDevice),
-//                downstreamReqVO.getIdentifier());
-//        IotDeviceServiceInvokeReqDTO reqDTO = new IotDeviceServiceInvokeReqDTO()
-//                .setParams((Map<String, Object>) downstreamReqVO.getData());
-////        CommonResult<Boolean> result = requestPlugin(url, reqDTO, device);
-//        CommonResult<Boolean> result = null;
-//
-//        // 3. 发送设备消息
-//        IotDeviceMessage message = new IotDeviceMessage().setRequestId(reqDTO.getRequestId())
-//                .setType(IotDeviceMessageTypeEnum.SERVICE.getType()).setIdentifier(reqDTO.getIdentifier())
-//                .setData(reqDTO.getParams());
-//        sendDeviceMessage(message, device, result.getCode());
-//
-//        // 4. 如果不成功，抛出异常，提示用户
-//        if (result.isError()) {
-//            log.error("[invokeDeviceService][设备({})服务调用失败，请求参数：({})，响应结果：({})]",
-//                    device.getDeviceKey(), reqDTO, result);
-//            throw exception(DEVICE_DOWNSTREAM_FAILED, result.getMsg());
-//        }
-//        return message;
-//    }
-
-//    /**
-//     * 获取设备属性
-//     *
-//     * @param downstreamReqVO 下行请求
-//     * @param device          设备
-//     * @param parentDevice    父设备
-//     * @return 下发消息
-//     */
-//    @SuppressWarnings("unchecked")
-//    private IotDeviceMessage getDeviceProperty(IotDeviceDownstreamReqVO downstreamReqVO,
-//                                               IotDeviceDO device, IotDeviceDO parentDevice) {
-//        // 1. 参数校验
-//        if (!(downstreamReqVO.getData() instanceof List<?>)) {
-//            throw new ServiceException(BAD_REQUEST.getCode(), "data 不是 List 类型");
-//        }
-//        // TODO @super：【可优化】过滤掉不合法的属性
-//
-//        // 2. 发送请求
-//        String url = String.format("sys/%s/%s/thing/service/property/get",
-//                getProductKey(device, parentDevice), getDeviceName(device, parentDevice));
-//        IotDevicePropertyGetReqDTO reqDTO = new IotDevicePropertyGetReqDTO()
-//                .setIdentifiers((List<String>) downstreamReqVO.getData());
-////        CommonResult<Boolean> result = requestPlugin(url, reqDTO, device);
-//        CommonResult<Boolean> result = null;
-//
-//        // 3. 发送设备消息
-//        IotDeviceMessage message = new IotDeviceMessage().setRequestId(reqDTO.getRequestId())
-//                .setType(IotDeviceMessageTypeEnum.PROPERTY.getType())
-//                .setIdentifier(IotDeviceMessageIdentifierEnum.PROPERTY_SET.getIdentifier())
-//                .setData(reqDTO.getIdentifiers());
-//        sendDeviceMessage(message, device, result.getCode());
-//
-//        // 4. 如果不成功，抛出异常，提示用户
-//        if (result.isError()) {
-//            log.error("[getDeviceProperty][设备({})属性获取失败，请求参数：({})，响应结果：({})]",
-//                    device.getDeviceKey(), reqDTO, result);
-//            throw exception(DEVICE_DOWNSTREAM_FAILED, result.getMsg());
-//        }
-//        return message;
-//    }
-
-//    /**
-//     * 设置设备配置
-//     *
-//     * @param downstreamReqVO 下行请求
-//     * @param device          设备
-//     * @param parentDevice    父设备
-//     * @return 下发消息
-//     */
-//    @SuppressWarnings({ "unchecked", "unused" })
-//    private IotDeviceMessage setDeviceConfig(IotDeviceDownstreamReqVO downstreamReqVO,
-//                                             IotDeviceDO device, IotDeviceDO parentDevice) {
-//        // 1. 参数转换，无需校验
-//        Map<String, Object> config = JsonUtils.parseObject(device.getConfig(), Map.class);
-//
-//        // 2. 发送请求
-//        String url = String.format("sys/%s/%s/thing/service/config/set",
-//                getProductKey(device, parentDevice), getDeviceName(device, parentDevice));
-//        IotDeviceConfigSetReqDTO reqDTO = new IotDeviceConfigSetReqDTO()
-//                .setConfig(config);
-////        CommonResult<Boolean> result = requestPlugin(url, reqDTO, device);
-//        CommonResult<Boolean> result = null;
-//
-//        // 3. 发送设备消息
-//        IotDeviceMessage message = new IotDeviceMessage().setRequestId(reqDTO.getRequestId())
-//                .setType(IotDeviceMessageTypeEnum.CONFIG.getType())
-//                .setIdentifier(IotDeviceMessageIdentifierEnum.CONFIG_SET.getIdentifier())
-//                .setData(reqDTO.getConfig());
-//        sendDeviceMessage(message, device, result.getCode());
-//
-//        // 4. 如果不成功，抛出异常，提示用户
-//        if (result.isError()) {
-//            log.error("[setDeviceConfig][设备({})配置下发失败，请求参数：({})，响应结果：({})]",
-//                    device.getDeviceKey(), reqDTO, result);
-//            throw exception(DEVICE_DOWNSTREAM_FAILED, result.getMsg());
-//        }
-//        return message;
-//    }
-
-//    /**
-//     * 设备 OTA 升级
-//     *
-//     * @param downstreamReqVO 下行请求
-//     * @param device          设备
-//     * @param parentDevice    父设备
-//     * @return 下发消息
-//     */
-//    private IotDeviceMessage otaUpgrade(IotDeviceDownstreamReqVO downstreamReqVO,
-//                                        IotDeviceDO device, IotDeviceDO parentDevice) {
-//        // 1. 参数校验
-//        if (!(downstreamReqVO.getData() instanceof Map<?, ?> data)) {
-//            throw new ServiceException(BAD_REQUEST.getCode(), "data 不是 Map 类型");
-//        }
-//
-//        // 2. 发送请求
-//        String url = String.format("ota/%s/%s/upgrade",
-//                getProductKey(device, parentDevice), getDeviceName(device, parentDevice));
-//        IotDeviceOtaUpgradeReqDTO reqDTO = IotDeviceOtaUpgradeReqDTO.build(data);
-////        CommonResult<Boolean> result = requestPlugin(url, reqDTO, device);
-//        CommonResult<Boolean> result = null;
-//
-//        // 3. 发送设备消息
-//        IotDeviceMessage message = new IotDeviceMessage().setRequestId(reqDTO.getRequestId())
-//                .setType(IotDeviceMessageTypeEnum.OTA.getType())
-//                .setIdentifier(IotDeviceMessageIdentifierEnum.OTA_UPGRADE.getIdentifier())
-//                .setData(downstreamReqVO.getData());
-//        sendDeviceMessage(message, device, result.getCode());
-//
-//        // 4. 如果不成功，抛出异常，提示用户
-//        if (result.isError()) {
-//            log.error("[otaUpgrade][设备({}) OTA 升级失败，请求参数：({})，响应结果：({})]",
-//                    device.getDeviceKey(), reqDTO, result);
-//            throw exception(DEVICE_DOWNSTREAM_FAILED, result.getMsg());
-//        }
-//        return message;
-//    }
-
-    private String getDeviceName(IotDeviceDO device, IotDeviceDO parentDevice) {
-        return parentDevice != null ? parentDevice.getDeviceName() : device.getDeviceName();
-    }
-
-    private String getProductKey(IotDeviceDO device, IotDeviceDO parentDevice) {
-        return parentDevice != null ? parentDevice.getProductKey() : device.getProductKey();
-    }
-
-}
-package cn.iocoder.yudao.module.iot.service.device.control;
-
-import cn.iocoder.yudao.module.iot.controller.admin.device.vo.control.IotDeviceUpstreamReqVO;
-import jakarta.validation.Valid;
-
-/**
- * IoT 设备上行 Service 接口
- *
- * 目的：设备 -> 网关 -> 服务端
- *
- * @author 芋道源码
- */
-public interface IotDeviceUpstreamService {
-
-    /**
-     * 设备上行，可用于设备模拟
-     *
-     * @param simulatorReqVO 设备上行请求 VO
-     */
-    void upstreamDevice(@Valid IotDeviceUpstreamReqVO simulatorReqVO);
-
-//    /**
-//     * 上报设备事件数据
-//     *
-//     * @param reportReqDTO 设备事件
-//     */
-//    void reportDeviceEvent(IotDeviceEventReportReqDTO reportReqDTO);
-//
-//    /**
-//     * 注册设备
-//     *
-//     * @param registerReqDTO 注册设备 DTO
-//     */
-//    void registerDevice(IotDeviceRegisterReqDTO registerReqDTO);
-//
-//    /**
-//     * 注册子设备
-//     *
-//     * @param registerReqDTO 注册子设备 DTO
-//     */
-//    void registerSubDevice(IotDeviceRegisterSubReqDTO registerReqDTO);
-//
-//    /**
-//     * 添加设备拓扑
-//     *
-//     * @param addReqDTO 添加设备拓扑 DTO
-//     */
-//    void addDeviceTopology(IotDeviceTopologyAddReqDTO addReqDTO);
-
-}
-package cn.iocoder.yudao.module.iot.service.device.control;
-
-import cn.hutool.core.collection.CollUtil;
-import cn.hutool.core.util.IdUtil;
-import cn.hutool.core.util.ObjUtil;
-import cn.iocoder.yudao.framework.tenant.core.util.TenantUtils;
-import cn.iocoder.yudao.module.iot.api.device.dto.control.upstream.*;
-import cn.iocoder.yudao.module.iot.controller.admin.device.vo.control.IotDeviceUpstreamReqVO;
-import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
-import cn.iocoder.yudao.module.iot.enums.device.IotDeviceMessageTypeEnum;
-import cn.iocoder.yudao.module.iot.enums.product.IotProductDeviceTypeEnum;
-import cn.iocoder.yudao.module.iot.service.device.IotDeviceService;
-import cn.iocoder.yudao.module.iot.service.device.data.IotDevicePropertyService;
-import jakarta.annotation.Resource;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.stereotype.Service;
-import org.springframework.validation.annotation.Validated;
-
-import java.time.LocalDateTime;
-import java.util.Map;
-import java.util.Objects;
-
-/**
- * IoT 设备上行 Service 实现类
- *
- * @author 芋道源码
- */
-@Service
-@Validated
-@Slf4j
-public class IotDeviceUpstreamServiceImpl implements IotDeviceUpstreamService {
-
-    @Resource
-    private IotDeviceService deviceService;
-    @Resource
-    private IotDevicePropertyService devicePropertyService;
-
-    // TODO @芋艿：需要重新实现下；
-    @Override
-    @SuppressWarnings("unchecked")
-    public void upstreamDevice(IotDeviceUpstreamReqVO simulatorReqVO) {
-        // 1. 校验存在
-        IotDeviceDO device = deviceService.validateDeviceExists(simulatorReqVO.getId());
-
-        // 2.1 情况一：属性上报
-        String requestId = IdUtil.fastSimpleUUID();
-        if (Objects.equals(simulatorReqVO.getType(), IotDeviceMessageTypeEnum.PROPERTY.getType())) {
-            reportDeviceProperty(((IotDevicePropertyReportReqDTO) new IotDevicePropertyReportReqDTO()
-                    .setRequestId(requestId).setReportTime(LocalDateTime.now())
-                    .setProductKey(device.getProductKey()).setDeviceName(device.getDeviceName()))
-                    .setProperties((Map<String, Object>) simulatorReqVO.getData()));
-            return;
-        }
-        // 2.2 情况二：事件上报
-        if (Objects.equals(simulatorReqVO.getType(), IotDeviceMessageTypeEnum.EVENT.getType())) {
-            reportDeviceEvent(((IotDeviceEventReportReqDTO) new IotDeviceEventReportReqDTO().setRequestId(requestId)
-                    .setReportTime(LocalDateTime.now())
-                    .setProductKey(device.getProductKey()).setDeviceName(device.getDeviceName()))
-                    .setIdentifier(simulatorReqVO.getIdentifier())
-                    .setParams((Map<String, Object>) simulatorReqVO.getData()));
-            return;
-        }
-        // 2.3 情况三：状态变更
-        if (Objects.equals(simulatorReqVO.getType(), IotDeviceMessageTypeEnum.STATE.getType())) {
-            // TODO @芋艿：这里未搞完
-            return;
-        }
-        throw new IllegalArgumentException("未知的类型：" + simulatorReqVO.getType());
-    }
-
-//    @Override TODO 芋艿：待重新实现
-    public void reportDeviceProperty(IotDevicePropertyReportReqDTO reportReqDTO) {
-        // 1.1 获得设备
-        log.info("[reportDeviceProperty][上报设备属性: {}]", reportReqDTO);
-        IotDeviceDO device = deviceService.getDeviceByProductKeyAndDeviceNameFromCache(
-                reportReqDTO.getProductKey(), reportReqDTO.getDeviceName());
-        if (device == null) {
-            log.error("[reportDeviceProperty][设备({}/{})不存在]",
-                    reportReqDTO.getProductKey(), reportReqDTO.getDeviceName());
-            return;
-        }
-
-        // 2. 发送设备消息
-//        IotDeviceMessage message = BeanUtils.toBean(reportReqDTO, IotDeviceMessage.class)
-//                .setType(IotDeviceMessageTypeEnum.PROPERTY.getType())
-//                .setIdentifier(IotDeviceMessageIdentifierEnum.PROPERTY_REPORT.getIdentifier())
-//                .setData(reportReqDTO.getProperties());
-//        sendDeviceMessage(message, device);
-    }
-
-    //    @Override TODO 芋艿：待重新实现
-    public void reportDeviceEvent(IotDeviceEventReportReqDTO reportReqDTO) {
-        // 1.1 获得设备
-        log.info("[reportDeviceEvent][上报设备事件: {}]", reportReqDTO);
-        IotDeviceDO device = deviceService.getDeviceByProductKeyAndDeviceNameFromCache(
-                reportReqDTO.getProductKey(), reportReqDTO.getDeviceName());
-        if (device == null) {
-            log.error("[reportDeviceEvent][设备({}/{})不存在]",
-                    reportReqDTO.getProductKey(), reportReqDTO.getDeviceName());
-            return;
-        }
-
-        // 2. 发送设备消息
-//        IotDeviceMessage message = BeanUtils.toBean(reportReqDTO, IotDeviceMessage.class)
-//                .setType(IotDeviceMessageTypeEnum.EVENT.getType())
-//                .setIdentifier(reportReqDTO.getIdentifier())
-//                .setData(reportReqDTO.getParams());
-//        sendDeviceMessage(message, device);
-    }
-
-    //    @Override TODO 芋艿：待重新实现
-    public void registerDevice(IotDeviceRegisterReqDTO registerReqDTO) {
-        log.info("[registerDevice][注册设备: {}]", registerReqDTO);
-        registerDevice0(registerReqDTO.getProductKey(), registerReqDTO.getDeviceName(), null, registerReqDTO);
-    }
-
-    private void registerDevice0(String productKey, String deviceName, Long gatewayId,
-            IotDeviceUpstreamAbstractReqDTO registerReqDTO) {
-        // 1.1 注册设备
-        IotDeviceDO device = deviceService.getDeviceByProductKeyAndDeviceNameFromCache(productKey, deviceName);
-        boolean registerNew = device == null;
-        if (device == null) {
-            device = deviceService.createDevice(productKey, deviceName, gatewayId);
-            log.info("[registerDevice0][消息({}) 设备({}/{}) 成功注册]", registerReqDTO, productKey, device);
-        } else if (gatewayId != null && ObjUtil.notEqual(device.getGatewayId(), gatewayId)) {
-            Long deviceId = device.getId();
-            TenantUtils.execute(device.getTenantId(),
-                    () -> deviceService.updateDeviceGateway(deviceId, gatewayId));
-            log.info("[registerDevice0][消息({}) 设备({}/{}) 更新网关设备编号({})]",
-                    registerReqDTO, productKey, device, gatewayId);
-        }
-        // 1.2 记录设备的最后时间
-//        updateDeviceLastTime(device, registerReqDTO);
-
-        // 2. 发送设备消息
-        if (registerNew) {
-//            IotDeviceMessage message = BeanUtils.toBean(registerReqDTO, IotDeviceMessage.class)
-//                    .setType(IotDeviceMessageTypeEnum.REGISTER.getType())
-//                    .setIdentifier(IotDeviceMessageIdentifierEnum.REGISTER_REGISTER.getIdentifier());
-//            sendDeviceMessage(message, device);
-        }
-    }
-
-    //    @Override TODO 芋艿：待重新实现
-    public void registerSubDevice(IotDeviceRegisterSubReqDTO registerReqDTO) {
-        // 1.1 注册子设备
-        log.info("[registerSubDevice][注册子设备: {}]", registerReqDTO);
-        IotDeviceDO device = deviceService.getDeviceByProductKeyAndDeviceNameFromCache(
-                registerReqDTO.getProductKey(), registerReqDTO.getDeviceName());
-        if (device == null) {
-            log.error("[registerSubDevice][设备({}/{}) 不存在]",
-                    registerReqDTO.getProductKey(), registerReqDTO.getDeviceName());
-            return;
-        }
-        if (!IotProductDeviceTypeEnum.isGateway(device.getDeviceType())) {
-            log.error("[registerSubDevice][设备({}/{}) 不是网关设备({})，无法进行注册]",
-                    registerReqDTO.getProductKey(), registerReqDTO.getDeviceName(), device);
-            return;
-        }
-        // 1.2 记录设备的最后时间
-//        updateDeviceLastTime(device, registerReqDTO);
-
-        // 2. 处理子设备
-        if (CollUtil.isNotEmpty(registerReqDTO.getParams())) {
-            registerReqDTO.getParams().forEach(subDevice -> registerDevice0(
-                    subDevice.getProductKey(), subDevice.getDeviceName(), device.getId(), registerReqDTO));
-            // TODO @芋艿：后续要处理，每个设备是否成功
-        }
-
-        // 3. 发送设备消息
-//        IotDeviceMessage message = BeanUtils.toBean(registerReqDTO, IotDeviceMessage.class)
-//                .setType(IotDeviceMessageTypeEnum.REGISTER.getType())
-//                .setIdentifier(IotDeviceMessageIdentifierEnum.REGISTER_REGISTER_SUB.getIdentifier())
-//                .setData(registerReqDTO.getParams());
-//        sendDeviceMessage(message, device);
-    }
-
-    //    @Override TODO 芋艿：待重新实现
-    public void addDeviceTopology(IotDeviceTopologyAddReqDTO addReqDTO) {
-        // 1.1 获得设备
-        log.info("[addDeviceTopology][添加设备拓扑: {}]", addReqDTO);
-        IotDeviceDO device = deviceService.getDeviceByProductKeyAndDeviceNameFromCache(
-                addReqDTO.getProductKey(), addReqDTO.getDeviceName());
-        if (device == null) {
-            log.error("[addDeviceTopology][设备({}/{}) 不存在]",
-                    addReqDTO.getProductKey(), addReqDTO.getDeviceName());
-            return;
-        }
-        if (!IotProductDeviceTypeEnum.isGateway(device.getDeviceType())) {
-            log.error("[addDeviceTopology][设备({}/{}) 不是网关设备({})，无法进行拓扑添加]",
-                    addReqDTO.getProductKey(), addReqDTO.getDeviceName(), device);
-            return;
-        }
-
-        // 2. 处理拓扑
-        if (CollUtil.isNotEmpty(addReqDTO.getParams())) {
-            TenantUtils.execute(device.getTenantId(), () -> {
-                addReqDTO.getParams().forEach(subDevice -> {
-                    IotDeviceDO subDeviceDO = deviceService.getDeviceByProductKeyAndDeviceNameFromCache(
-                            subDevice.getProductKey(), subDevice.getDeviceName());
-                    // TODO @芋艿：后续要处理，每个设备是否成功
-                    if (subDeviceDO == null) {
-                        log.error("[addDeviceTopology][子设备({}/{}) 不存在]",
-                                subDevice.getProductKey(), subDevice.getDeviceName());
-                        return;
-                    }
-                    deviceService.updateDeviceGateway(subDeviceDO.getId(), device.getId());
-                    log.info("[addDeviceTopology][子设备({}/{}) 添加到网关设备({}) 成功]",
-                            subDevice.getProductKey(), subDevice.getDeviceName(), device);
-                });
-            });
-        }
-
-        // 3. 发送设备消息
-//        IotDeviceMessage message = BeanUtils.toBean(addReqDTO, IotDeviceMessage.class)
-//                .setType(IotDeviceMessageTypeEnum.TOPOLOGY.getType())
-//                .setIdentifier(IotDeviceMessageIdentifierEnum.TOPOLOGY_ADD.getIdentifier())
-//                .setData(addReqDTO.getParams());
-//        sendDeviceMessage(message, device);
-    }
-
-}
+package cn.iocoder.yudao.module.iot.service.device.message;
+
+import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
+import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
+
+/**
+ * IoT 设备消息 Service 接口
+ *
+ * @author 芋道源码
+ */
+public interface IotDeviceMessageService {
+
+    /**
+     * 初始化设备消息的 TDengine 超级表
+     *
+     * 系统启动时，会自动初始化一次
+     */
+    void defineDeviceMessageStable();
+
+    /**
+     * 发送设备消息
+     *
+     * @param message 消息（“codec（编解码）字段” 部分字段）
+     * @param device 设备
+     * @return 设备消息
+     */
+    IotDeviceMessage sendDeviceMessage(IotDeviceMessage message, IotDeviceDO device);
+
+    /**
+     * 发送设备消息
+     *
+     * @param message 消息（“codec（编解码）字段” 部分字段）
+     * @return 设备消息
+     */
+    IotDeviceMessage sendDeviceMessage(IotDeviceMessage message);
+
+    /**
+     * 处理设备上行的消息，包括如下步骤：
+     *
+     * 1. 处理消息
+     * 2. 记录消息
+     * 3. 回复消息
+     *
+     * @param message 消息
+     * @param device 设备
+     */
+    void handleUpstreamDeviceMessage(IotDeviceMessage message, IotDeviceDO device);
+
+}
+package cn.iocoder.yudao.module.iot.service.device.message;
+
+import cn.hutool.core.util.StrUtil;
+import cn.iocoder.yudao.framework.common.exception.ServiceException;
+import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceStateEnum;
+import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
+import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer;
+import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
+import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
+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 com.google.common.base.Objects;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import org.springframework.validation.annotation.Validated;
+
+import java.time.LocalDateTime;
+
+import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
+import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.DEVICE_DOWNSTREAM_FAILED_SERVER_ID_NULL;
+
+/**
+ * IoT 设备消息 Service 实现类
+ *
+ * @author 芋道源码
+ */
+@Service
+@Validated
+@Slf4j
+public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
+
+    @Resource
+    private IotDeviceService deviceService;
+    @Resource
+    private IotDevicePropertyService devicePropertyService;
+
+    @Resource
+    private IotDeviceMessageMapper deviceLogMapper;
+
+    @Resource
+    private IotDeviceMessageProducer deviceMessageProducer;
+
+    @Override
+    public void defineDeviceMessageStable() {
+        if (StrUtil.isNotEmpty(deviceLogMapper.showSTable())) {
+            log.info("[defineDeviceMessageStable][设备消息超级表已存在，创建跳过]");
+            return;
+        }
+        log.info("[defineDeviceMessageStable][设备消息超级表不存在，创建开始...]");
+        deviceLogMapper.createSTable();
+        log.info("[defineDeviceMessageStable][设备消息超级表不存在，创建成功]");
+    }
+
+    // TODO @芋艿：要不要异步记录；
+    private void createDeviceLog(IotDeviceMessage message) {
+        IotDeviceMessageDO messageDO = BeanUtils.toBean(message, IotDeviceMessageDO.class)
+                .setUpstream(IotDeviceMessageUtils.isUpstreamMessage(message));
+        deviceLogMapper.insert(messageDO);
+    }
+
+    @Override
+    public IotDeviceMessage sendDeviceMessage(IotDeviceMessage message) {
+        IotDeviceDO device = deviceService.validateDeviceExists(message.getDeviceId());
+        return sendDeviceMessage(message, device);
+    }
+
+    // TODO @芋艿：针对连接网关的设备，是不是 productKey、deviceName 需要调整下；
+    @Override
+    public IotDeviceMessage sendDeviceMessage(IotDeviceMessage message, IotDeviceDO device) {
+        // 1. 补充信息
+        appendDeviceMessage(message, device);
+
+        // 2.1 情况一：发送上行消息
+        boolean upstream = IotDeviceMessageUtils.isUpstreamMessage(message);
+        if (upstream) {
+            deviceMessageProducer.sendDeviceMessage(message);
+            return message;
+        }
+
+        // 2.2 情况二：发送下行消息
+        // 如果是下行消息，需要校验 serverId 存在
+        String serverId = devicePropertyService.getDeviceServerId(device.getProductKey(), device.getDeviceName());
+        if (StrUtil.isEmpty(serverId)) {
+            throw exception(DEVICE_DOWNSTREAM_FAILED_SERVER_ID_NULL);
+        }
+        deviceMessageProducer.sendDeviceMessageToGateway(serverId, message);
+        // 特殊：记录消息日志。原因：上行消息，消费时，已经会记录；下行消息，因为消费在 Gateway 端，所以需要在这里记录
+        createDeviceLog(message);
+        return message;
+    }
+
+    /**
+     * 补充消息的后端字段
+     *
+     * @param message    消息
+     * @param device 设备信息
+     * @return 消息
+     */
+    private IotDeviceMessage appendDeviceMessage(IotDeviceMessage message, IotDeviceDO device) {
+        message.setId(IotDeviceMessageUtils.generateMessageId()).setReportTime(LocalDateTime.now())
+                .setDeviceId(device.getId()).setTenantId(device.getTenantId());
+        // 特殊：如果设备没有指定 requestId，则使用 messageId
+        if (StrUtil.isEmpty(message.getRequestId())) {
+            message.setRequestId(message.getId());
+        }
+        return message;
+    }
+
+    @Override
+    public void handleUpstreamDeviceMessage(IotDeviceMessage message, IotDeviceDO device) {
+        // 1. 理消息
+        Object replyData = null;
+        ServiceException serviceException = null;
+        try {
+            replyData = handleUpstreamDeviceMessage0(message, device);
+        } catch (ServiceException ex) {
+            serviceException = ex;
+            log.warn("[onMessage][message({}) 业务异常]", message, serviceException);
+        } catch (Exception ex) {
+            log.error("[onMessage][message({}) 发生异常]", message, ex);
+            throw ex;
+        }
+
+        // 2. 记录消息
+        createDeviceLog(message);
+
+        // 3. 回复消息。前提：非 _reply 消息，并且非禁用回复的消息
+        if (IotDeviceMessageUtils.isReplyMessage(message)
+            || IotDeviceMessageMethodEnum.isReplyDisabled(message.getMethod())) {
+            return;
+        }
+        sendDeviceMessage(IotDeviceMessage.replyOf(message.getRequestId(), message.getMethod(), replyData,
+                serviceException != null ? serviceException.getCode() : null,
+                serviceException != null ? serviceException.getMessage() : null));
+    }
+
+    // TODO @芋艿：可优化：未来逻辑复杂后，可以独立拆除 Processor 处理器
+    private Object handleUpstreamDeviceMessage0(IotDeviceMessage message, IotDeviceDO device) {
+        // 设备上线
+        if (Objects.equal(message.getMethod(), IotDeviceMessageMethodEnum.STATE_ONLINE.getMethod())) {
+            deviceService.updateDeviceState(device, IotDeviceStateEnum.ONLINE.getState());
+            // TODO 芋艿：子设备的关联
+            return null;
+        }
+        // 设备下线
+        if (Objects.equal(message.getMethod(), IotDeviceMessageMethodEnum.STATE_OFFLINE.getMethod())) {
+            deviceService.updateDeviceState(device, IotDeviceStateEnum.OFFLINE.getState());
+            // TODO 芋艿：子设备的关联
+            return null;
+        }
+
+        // 属性上报
+        if (Objects.equal(message.getMethod(), IotDeviceMessageMethodEnum.PROPERTY_REPORT.getMethod())) {
+            devicePropertyService.saveDeviceProperty(device, message);
+            return null;
+        }
+
+        // TODO @芋艿：这里可以按需，添加别的逻辑；
+        return null;
+    }
+
+}
-package cn.iocoder.yudao.module.iot.service.device.data;
+package cn.iocoder.yudao.module.iot.service.device.property;
  
 import cn.iocoder.yudao.framework.common.pojo.PageResult;
 import cn.iocoder.yudao.module.iot.controller.admin.device.vo.data.IotDeviceLogPageReqVO;
-import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
-import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceLogDO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceMessageDO;
  
 import javax.annotation.Nullable;
 import java.time.LocalDateTime;
@@ -18,26 +17,12 @@ import java.util.Map;
 public interface IotDeviceLogService {
  
     /**
-     * 初始化 TDengine 超级表
-     *
-     * 系统启动时，会自动初始化一次
-     */
-    void defineDeviceLog();
-
-    /**
-     * 插入设备日志
-     *
-     * @param message 设备数据
-     */
-    void createDeviceLog(IotDeviceMessage message);
-
-    /**
      * 获得设备日志分页
      *
      * @param pageReqVO 分页查询
      * @return 设备日志分页
      */
-    PageResult<IotDeviceLogDO> getDeviceLogPage(IotDeviceLogPageReqVO pageReqVO);
+    PageResult<IotDeviceMessageDO> getDeviceLogPage(IotDeviceLogPageReqVO pageReqVO);
  
     /**
      * 获得设备日志数量
-package cn.iocoder.yudao.module.iot.service.device.data;
+package cn.iocoder.yudao.module.iot.service.device.property;
  
 import cn.hutool.core.date.LocalDateTimeUtil;
 import cn.hutool.core.map.MapUtil;
-import cn.hutool.core.util.IdUtil;
-import cn.hutool.core.util.StrUtil;
 import cn.iocoder.yudao.framework.common.pojo.PageResult;
-import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
-import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
 import cn.iocoder.yudao.module.iot.controller.admin.device.vo.data.IotDeviceLogPageReqVO;
-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.IotDeviceLogDO;
-import cn.iocoder.yudao.module.iot.dal.tdengine.IotDeviceLogMapper;
+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 com.baomidou.mybatisplus.core.metadata.IPage;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
@@ -40,39 +34,12 @@ public class IotDeviceLogServiceImpl implements IotDeviceLogService {
     private IotDeviceService deviceService;
  
     @Resource
-    private IotDeviceLogMapper deviceLogMapper;
+    private IotDeviceMessageMapper deviceLogMapper;
  
     @Override
-    public void defineDeviceLog() {
-        if (StrUtil.isNotEmpty(deviceLogMapper.showDeviceLogSTable())) {
-            log.info("[defineDeviceLog][设备日志超级表已存在，创建跳过]");
-            return;
-        }
-
-        log.info("[defineDeviceLog][设备日志超级表不存在，创建开始...]");
-        deviceLogMapper.createDeviceLogSTable();
-        log.info("[defineDeviceLog][设备日志超级表不存在，创建成功]");
-    }
-
-    @Override
-    public void createDeviceLog(IotDeviceMessage message) {
-        IotDeviceDO device = deviceService.getDeviceByProductKeyAndDeviceNameFromCache(
-                message.getProductKey(), message.getDeviceName());
-        if (device == null) {
-            log.error("[createDeviceLog][设备({}/{}) 不存在]", message.getProductKey(), message.getDeviceName());
-            return;
-        }
-        IotDeviceLogDO log = BeanUtils.toBean(message, IotDeviceLogDO.class)
-                .setId(IdUtil.fastSimpleUUID())
-                .setContent(JsonUtils.toJsonString(message.getData()))
-                .setTenantId(device.getTenantId());
-        deviceLogMapper.insert(log);
-    }
-
-    @Override
-    public PageResult<IotDeviceLogDO> getDeviceLogPage(IotDeviceLogPageReqVO pageReqVO) {
+    public PageResult<IotDeviceMessageDO> getDeviceLogPage(IotDeviceLogPageReqVO pageReqVO) {
         try {
-            IPage<IotDeviceLogDO> page = deviceLogMapper.selectPage(
+            IPage<IotDeviceMessageDO> page = deviceLogMapper.selectPage(
                     new Page<>(pageReqVO.getPageNo(), pageReqVO.getPageSize()), pageReqVO);
             return new PageResult<>(page.getRecords(), page.getTotal());
         } catch (Exception exception) {
@@ -92,7 +59,8 @@ public class IotDeviceLogServiceImpl implements IotDeviceLogService {
     @Override
     public List<Map<Long, Integer>> getDeviceLogUpCountByHour(String deviceKey, Long startTime, Long endTime) {
         // TODO @super：不能只基于数据库统计。因为有一些小时，可能出现没数据的情况，导致前端展示的图是不全的。可以参考 CrmStatisticsCustomerService 来实现
-        List<Map<String, Object>> list = deviceLogMapper.selectDeviceLogUpCountByHour(deviceKey, startTime, endTime);
+        // TODO @芋艿：这里实现，需要调整
+        List<Map<String, Object>> list = deviceLogMapper.selectDeviceLogUpCountByHour(0L, startTime, endTime);
         return list.stream()
                 .map(map -> {
                     // 从Timestamp获取时间戳
@@ -108,7 +76,8 @@ public class IotDeviceLogServiceImpl implements IotDeviceLogService {
     // TODO @super：getDeviceLogDownCountByHour 融合到 getDeviceLogUpCountByHour
     @Override
     public List<Map<Long, Integer>> getDeviceLogDownCountByHour(String deviceKey, Long startTime, Long endTime) {
-        List<Map<String, Object>> list = deviceLogMapper.selectDeviceLogDownCountByHour(deviceKey, startTime, endTime);
+        // TODO @芋艿：这里实现，需要调整
+        List<Map<String, Object>> list = deviceLogMapper.selectDeviceLogDownCountByHour(0L, startTime, endTime);
         return list.stream()
                 .map(map -> {
                     // 从Timestamp获取时间戳
-package cn.iocoder.yudao.module.iot.service.device.data;
+package cn.iocoder.yudao.module.iot.service.device.property;
  
 import cn.iocoder.yudao.framework.common.pojo.PageResult;
 import cn.iocoder.yudao.module.iot.controller.admin.device.vo.data.IotDevicePropertyHistoryPageReqVO;
 import cn.iocoder.yudao.module.iot.controller.admin.device.vo.data.IotDevicePropertyRespVO;
 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.IotDevicePropertyDO;
 import jakarta.validation.Valid;
  
@@ -30,9 +31,10 @@ public interface IotDevicePropertyService {
     /**
      * 保存设备数据
      *
+     * @param device 设备
      * @param message 设备消息
      */
-    void saveDeviceProperty(IotDeviceMessage message);
+    void saveDeviceProperty(IotDeviceDO device, IotDeviceMessage message);
  
     /**
      * 获得设备属性最新数据
@@ -79,14 +81,6 @@ public interface IotDevicePropertyService {
     void updateDeviceServerId(String productKey, String deviceName, String serverId);
  
     /**
-     * 删除设备关联的网关 serverId
-     *
-     * @param productKey 产品标识
-     * @param deviceName 设备名称
-     */
-    void deleteDeviceServerId(String productKey, String deviceName);
-
-    /**
      * 获得设备关联的网关 serverId
      *
      * @param productKey 产品标识
-package cn.iocoder.yudao.module.iot.service.device.data;
+package cn.iocoder.yudao.module.iot.service.device.property;
  
 import cn.hutool.core.collection.CollUtil;
 import cn.hutool.core.date.LocalDateTimeUtil;
 import cn.hutool.core.map.MapUtil;
 import cn.hutool.core.util.StrUtil;
 import cn.iocoder.yudao.framework.common.pojo.PageResult;
-import cn.iocoder.yudao.framework.tenant.core.aop.TenantIgnore;
 import cn.iocoder.yudao.module.iot.controller.admin.device.vo.data.IotDevicePropertyHistoryPageReqVO;
 import cn.iocoder.yudao.module.iot.controller.admin.device.vo.data.IotDevicePropertyRespVO;
 import cn.iocoder.yudao.module.iot.controller.admin.thingmodel.model.dataType.ThingModelDateOrTextDataSpecs;
@@ -121,21 +120,13 @@ public class IotDevicePropertyServiceImpl implements IotDevicePropertyService {
     }
  
     @Override
-    @TenantIgnore
-    public void saveDeviceProperty(IotDeviceMessage message) {
+    public void saveDeviceProperty(IotDeviceDO device, IotDeviceMessage message) {
         if (!(message.getData() instanceof Map)) {
             log.error("[saveDeviceProperty][消息内容({}) 的 data 类型不正确]", message);
             return;
         }
-        // 1. 获得设备信息
-        IotDeviceDO device = deviceService.getDeviceByProductKeyAndDeviceNameFromCache(
-                message.getProductKey(), message.getDeviceName());
-        if (device == null) {
-            log.error("[saveDeviceProperty][消息({}) 对应的设备不存在]", message);
-            return;
-        }
  
-        // 2. 根据物模型，拼接合法的属性
+        // 1. 根据物模型，拼接合法的属性
         // TODO @芋艿：【待定 004】赋能后，属性到底以 thingModel 为准（ik），还是 db 的表结构为准（tl）？
         List<IotThingModelDO> thingModels = thingModelService.getThingModelListByProductKeyFromCache(device.getProductKey());
         Map<String, Object> properties = new HashMap<>();
@@ -151,11 +142,11 @@ public class IotDevicePropertyServiceImpl implements IotDevicePropertyService {
             return;
         }
  
-        // 3.1 保存设备属性【数据】
+        // 2.1 保存设备属性【数据】
         devicePropertyMapper.insert(device, properties,
                 LocalDateTimeUtil.toEpochMilli(message.getReportTime()));
  
-        // 3.2 保存设备属性【日志】
+        // 2.2 保存设备属性【日志】
         Map<String, IotDevicePropertyDO> properties2 = convertMap(properties.entrySet(), Map.Entry::getKey, entry ->
                 IotDevicePropertyDO.builder().value(entry.getValue()).updateTime(message.getReportTime()).build());
         deviceDataRedisDAO.putAll(device.getProductKey(), device.getDeviceName(), properties2);
@@ -210,11 +201,6 @@ public class IotDevicePropertyServiceImpl implements IotDevicePropertyService {
     }
  
     @Override
-    public void deleteDeviceServerId(String productKey, String deviceName) {
-        deviceServerIdRedisDAO.delete(productKey, deviceName);
-    }
-
-    @Override
     public String getDeviceServerId(String productKey, String deviceName) {
         return deviceServerIdRedisDAO.get(productKey, deviceName);
     }
@@ -8,7 +8,7 @@ import cn.iocoder.yudao.module.iot.controller.admin.product.vo.product.IotProduc
 import cn.iocoder.yudao.module.iot.dal.dataobject.product.IotProductDO;
 import cn.iocoder.yudao.module.iot.dal.mysql.product.IotProductMapper;
 import cn.iocoder.yudao.module.iot.enums.product.IotProductStatusEnum;
-import cn.iocoder.yudao.module.iot.service.device.data.IotDevicePropertyService;
+import cn.iocoder.yudao.module.iot.service.device.property.IotDevicePropertyService;
 import com.baomidou.dynamic.datasource.annotation.DSTransactional;
 import jakarta.annotation.Resource;
 import org.springframework.context.annotation.Lazy;
@@ -309,8 +309,10 @@ public class IotRuleSceneServiceImpl implements IotRuleSceneService {
     private List<IotRuleSceneDO> getMatchedRuleSceneListByMessage(IotDeviceMessage message) {
         // 1. 匹配设备
         // TODO @芋艿：可能需要 getSelf(); 缓存
-        List<IotRuleSceneDO> ruleScenes = getRuleSceneListByProductKeyAndDeviceNameFromCache(
-                message.getProductKey(), message.getDeviceName());
+        List<IotRuleSceneDO> ruleScenes = null;
+        // TODO @芋艿：这里需要适配
+//        List<IotRuleSceneDO> ruleScenes = getRuleSceneListByProductKeyAndDeviceNameFromCache(
+//                message.getProductKey(), message.getDeviceName());
         if (CollUtil.isEmpty(ruleScenes)) {
             return ruleScenes;
         }
@@ -329,10 +331,11 @@ public class IotRuleSceneServiceImpl implements IotRuleSceneService {
                 }
                 // 2.3 多个条件，只需要满足一个即可
                 IotRuleSceneDO.TriggerCondition matchedCondition = CollUtil.findOne(trigger.getConditions(), condition -> {
-                    if (ObjUtil.notEqual(message.getType(), condition.getType())
-                            || ObjUtil.notEqual(message.getIdentifier(), condition.getIdentifier())) {
-                        return false;
-                    }
+                    // TODO @芋艿：这里的逻辑，需要适配
+//                    if (ObjUtil.notEqual(message.getType(), condition.getType())
+//                            || ObjUtil.notEqual(message.getIdentifier(), condition.getIdentifier())) {
+//                        return false;
+//                    }
                     // 多个条件参数，必须全部满足。所以，下面的逻辑就是找到一个不满足的条件参数
                     IotRuleSceneDO.TriggerConditionParameter notMatchedParameter = CollUtil.findOne(condition.getParameters(),
                             parameter -> !isTriggerConditionParameterMatched(message, parameter, ruleScene, trigger));
 package cn.iocoder.yudao.module.iot.service.rule.action;
  
 import cn.hutool.core.lang.Assert;
-import cn.iocoder.yudao.module.iot.controller.admin.device.vo.control.IotDeviceDownstreamReqVO;
 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.rule.IotRuleSceneDO;
 import cn.iocoder.yudao.module.iot.enums.rule.IotRuleSceneActionTypeEnum;
 import cn.iocoder.yudao.module.iot.service.device.IotDeviceService;
-import cn.iocoder.yudao.module.iot.service.device.control.IotDeviceDownstreamService;
+import cn.iocoder.yudao.module.iot.service.device.message.IotDeviceMessageService;
 import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
@@ -22,9 +21,9 @@ import org.springframework.stereotype.Component;
 public class IotRuleSceneDeviceControlAction implements IotRuleSceneAction {
  
     @Resource
-    private IotDeviceDownstreamService deviceDownstreamService;
-    @Resource
     private IotDeviceService deviceService;
+    @Resource
+    private IotDeviceMessageService deviceMessageService;
  
     @Override
     public void execute(IotDeviceMessage message, IotRuleSceneDO.ActionConfig config) {
@@ -38,9 +37,9 @@ public class IotRuleSceneDeviceControlAction implements IotRuleSceneAction {
                 return;
             }
             try {
-                IotDeviceMessage downstreamMessage = deviceDownstreamService.downstreamDevice(new IotDeviceDownstreamReqVO()
-                        .setId(device.getId()).setType(control.getType()).setIdentifier(control.getIdentifier())
-                        .setData(control.getData()));
+                // TODO @芋艿：@puhui999：这块可能要改，从 type => method
+                IotDeviceMessage downstreamMessage = deviceMessageService.sendDeviceMessage(IotDeviceMessage.requestOf(
+                        control.getType() + control.getIdentifier(), control.getData()).setDeviceId(device.getId()));
                 log.info("[execute][message({}) config({}) 下发消息({})成功]", message, config, downstreamMessage);
             } catch (Exception e) {
                 log.error("[execute][message({}) config({}) 下发消息失败]", message, config, e);
@@ -2,51 +2,62 @@
 <!DOCTYPE mapper
         PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
         "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
-<mapper namespace="cn.iocoder.yudao.module.iot.dal.tdengine.IotDeviceLogMapper">
+<mapper namespace="cn.iocoder.yudao.module.iot.dal.tdengine.IotDeviceMessageMapper">
  
-    <update id="createDeviceLogSTable">
-        CREATE STABLE IF NOT EXISTS device_log (
+    <update id="createSTable">
+        CREATE STABLE IF NOT EXISTS device_message (
             ts TIMESTAMP,
             id NCHAR(50),
-            message_id NCHAR(50),
-            type NCHAR(50),
-            identifier NCHAR(255),
-            content NCHAR(1024),
-            code INT,
             report_time TIMESTAMP,
-            tenant_id BIGINT
+            device_id BIGINT,
+            tenant_id BIGINT,
+            server_id NCHAR(50),
+            upstream BOOL,
+            request_id NCHAR(50),
+            method NCHAR(100),
+            params NCHAR(2048),
+            data NCHAR(2048),
+            code INT
         ) TAGS (
-            product_key NCHAR(50),
-            device_name NCHAR(50)
+            device_id BIGINT
         )
     </update>
  
-    <select id="showDeviceLogSTable" resultType="String">
-        SHOW STABLES LIKE 'device_log'
+    <select id="showSTable" resultType="String">
+        SHOW STABLES LIKE 'device_message'
     </select>
  
     <insert id="insert">
-        INSERT INTO device_log_${productKey}_${deviceName} (
-            ts, id, message_id, type, identifier,
-            content, code, report_time, tenant_id
+        INSERT INTO device_message_${deviceId} (
+            ts, id, report_time, device_id, tenant_id,
+            server_id, upstream, request_id, method, params,
+            data, code
         )
-        USING device_log
-        TAGS ('${productKey}', '${deviceName}')
+        USING device_message
+        TAGS (#{deviceId})
         VALUES (
-            NOW, #{id}, #{messageId}, #{type}, #{identifier},
-            #{content}, #{code}, #{reportTime}, #{tenantId}
+            #{ts}, #{id}, #{reportTime}, #{deviceId}, #{tenantId},
+            #{serverId}, #{upstream}, #{requestId}, #{method}, #{params},
+            #{data}, #{code}
         )
     </insert>
  
-    <select id="selectPage" resultType="cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceLogDO">
-        SELECT ts, id, device_key, product_key, type, identifier, content, report_time
-        FROM device_log_${productKey}_${deviceName}
+    <select id="selectPage" resultType="cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceMessageDO">
+        SELECT ts, id, report_time, device_id, tenant_id, server_id, upstream, 
+               request_id, method, params, data, code
+        FROM device_message_${reqVO.deviceId}
         <where>
-            <if test="reqVO.type != null and reqVO.type != ''">
-                AND type = #{reqVO.type}
+            <if test="reqVO.method != null and reqVO.method != ''">
+                AND method = #{reqVO.method}
             </if>
-            <if test="reqVO.identifier != null and reqVO.identifier != ''">
-                AND identifier LIKE CONCAT('%',#{reqVO.identifier},'%')
+            <if test="reqVO.upstream != null">
+                AND upstream = #{reqVO.upstream}
+            </if>
+            <if test="reqVO.startTime != null">
+                AND ts >= #{reqVO.startTime}
+            </if>
+            <if test="reqVO.endTime != null">
+                AND ts &lt;= #{reqVO.endTime}
             </if>
         </where>
         ORDER BY ts DESC
@@ -54,7 +65,7 @@
  
     <select id="selectCountByCreateTime" resultType="Long">
         SELECT COUNT(*)
-        FROM device_log
+        FROM device_message
         <where>
             <if test="createTime != null">
                 AND ts >= #{createTime}
@@ -68,11 +79,11 @@
             COUNT(*) as data
         FROM
         <choose>
-            <when test="deviceKey != null and deviceKey != ''">
-                device_log_${deviceKey}
+            <when test="deviceId != null">
+                device_message_${deviceId}
             </when>
             <otherwise>
-                device_log
+                device_message
             </otherwise>
         </choose>
         <where>
@@ -82,9 +93,7 @@
             <if test="endTime != null">
                 AND ts &lt;= #{endTime}
             </if>
-            AND (
-                 identifier IN ('online', 'offline', 'pull', 'progress', 'report', 'register', 'register_sub')
-            )
+            AND upstream = true
         </where>
         GROUP BY TIMETRUNCATE(ts, 1h)
         ORDER BY time ASC
@@ -96,11 +105,11 @@
             COUNT(*) as data
         FROM
         <choose>
-            <when test="deviceKey != null and deviceKey != ''">
-                device_log_${deviceKey}
+            <when test="deviceId != null">
+                device_message_${deviceId}
             </when>
             <otherwise>
-                device_log
+                device_message
             </otherwise>
         </choose>
         <where>
@@ -110,10 +119,10 @@
             <if test="endTime != null">
                 AND ts &lt;= #{endTime}
             </if>
-            AND identifier IN ('set', 'get', 'upgrade', 'unregister_sub', 'topology_add')
+            AND upstream = false
         </where>
         GROUP BY TIMETRUNCATE(ts, 1h)
         ORDER BY time ASC
     </select>
  
-</mapper>
 \ No newline at end of file
+</mapper> 
 \ No newline at end of file
-package cn.iocoder.yudao.module.iot.core.enums;
-
-import lombok.Getter;
-import lombok.RequiredArgsConstructor;
-
-// TODO @芋艿：需要添加对应的 DTO，以及上下行的链路，网关、网关服务、设备等
-
-/**
- * IoT 设备消息标识符枚举
- */
-@Getter
-@RequiredArgsConstructor
-public enum IotDeviceMessageIdentifierEnum {
-
-    PROPERTY_GET("get"), // 下行 TODO 芋艿：【讨论】貌似这个“上行”更合理？device 主动拉取配置。和 IotDevicePropertyGetReqDTO 一样的配置
-    PROPERTY_SET("set"), // 下行
-    PROPERTY_REPORT("report"), // 上行
-
-    STATE_ONLINE("online"), // 上行
-    STATE_OFFLINE("offline"), // 上行
-
-    CONFIG_GET("get"), // 上行 TODO 芋艿：【讨论】暂时没有上行的场景
-    CONFIG_SET("set"), // 下行
-
-    SERVICE_INVOKE("${identifier}"), // 下行
-    SERVICE_REPLY_SUFFIX("_reply"), // 芋艿：TODO 芋艿：【讨论】上行 or 下行
-
-    OTA_UPGRADE("upgrade"), // 下行
-    OTA_PULL("pull"), // 上行
-    OTA_PROGRESS("progress"), // 上行
-    OTA_REPORT("report"), // 上行
-
-    REGISTER_REGISTER("register"), // 上行
-    REGISTER_REGISTER_SUB("register_sub"), // 上行
-    REGISTER_UNREGISTER_SUB("unregister_sub"), // 下行
-
-    TOPOLOGY_ADD("topology_add"), // 下行;
-    ;
-
-    /**
-     * 标志符
-     */
-    private final String identifier;
-
-}
 \ No newline at end of file
 package cn.iocoder.yudao.module.iot.core.enums;
  
+import cn.hutool.core.util.ArrayUtil;
+import cn.iocoder.yudao.framework.common.core.ArrayValuable;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
  
+import java.util.Arrays;
+import java.util.Set;
+
 /**
  * IoT 设备消息的方法枚举
  *
@@ -10,16 +15,47 @@ import lombok.Getter;
  */
 @Getter
 @AllArgsConstructor
-public enum IotDeviceMessageMethodEnum {
+public enum IotDeviceMessageMethodEnum implements ArrayValuable<String> {
  
     // ========== 设备状态 ==========
  
-    STATE_ONLINE("thing.state.online"),
-    STATE_OFFLINE("thing.state.offline"),
+    STATE_ONLINE("thing.state.online", true),
+    STATE_OFFLINE("thing.state.offline", true),
+
+    // ========== 设备属性 ==========
+    // 可参考 https://help.aliyun.com/zh/iot/user-guide/device-properties-events-and-services
+    PROPERTY_REPORT("thing.property.report", true),
+
+    PROPERTY_SET("thing.property.set", false),
+    PROPERTY_GET("thing.property.get", false),
+
+    //
  
     ;
  
+    public static final String[] ARRAYS = Arrays.stream(values()).map(IotDeviceMessageMethodEnum::getMethod).toArray(String[]::new);
+
+    /**
+     * 不进行 reply 回复的方法集合
+     */
+    public static final Set<String> REPLY_DISABLED = Set.of(STATE_ONLINE.getMethod(), STATE_OFFLINE.getMethod());
+
     private final String method;
  
+    private final Boolean upstream;
+
+    @Override
+    public String[] array() {
+        return ARRAYS;
+    }
+
+    public static IotDeviceMessageMethodEnum of(String method) {
+        return ArrayUtil.firstMatch(item -> item.getMethod().equals(method),
+                IotDeviceMessageMethodEnum.values());
+    }
+
+    public static boolean isReplyDisabled(String method) {
+        return REPLY_DISABLED.contains(method);
+    }
  
 }
@@ -14,7 +14,7 @@ import java.util.Arrays;
 public enum IotDeviceMessageTypeEnum implements ArrayValuable<String> {
  
     STATE("state"), // 设备状态
-    PROPERTY("property"), // 设备属性：可参考 https://help.aliyun.com/zh/iot/user-guide/device-properties-events-and-services 设备属性、事件、服务
+//    PROPERTY("property"), // 设备属性：可参考 https://help.aliyun.com/zh/iot/user-guide/device-properties-events-and-services 设备属性、事件、服务
     EVENT("event"), // 设备事件：可参考 https://help.aliyun.com/zh/iot/user-guide/device-properties-events-and-services 设备属性、事件、服务
     SERVICE("service"), // 设备服务：可参考 https://help.aliyun.com/zh/iot/user-guide/device-properties-events-and-services 设备属性、事件、服务
     CONFIG("config"), // 设备配置：可参考 https://help.aliyun.com/zh/iot/user-guide/remote-configuration-1 远程配置
 package cn.iocoder.yudao.module.iot.core.mq.message;
  
+import cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants;
 import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
 import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
 import lombok.AllArgsConstructor;
@@ -43,6 +44,20 @@ public class IotDeviceMessage {
      */
     private LocalDateTime reportTime;
  
+    /**
+     * 设备编号
+     */
+    private Long deviceId;
+    /**
+     * 租户编号
+     */
+    private Long tenantId;
+
+    /**
+     * 服务编号，该消息由哪个 server 发送
+     */
+    private String serverId;
+
     // ========== codec（编解码）字段 ==========
  
     /**
@@ -72,84 +87,53 @@ public class IotDeviceMessage {
      * 响应错误码
      */
     private Integer code;
-
-    // ========== 后端字段 ==========
-
     /**
-     * 设备编号
+     * 返回结果信息
      */
-    private Long deviceId;
-    /**
-     * 租户编号
-     */
-    private Long tenantId;
+    private String msg;
  
-    /**
-     * 服务编号，该消息由哪个 server 服务进行消费
-     */
-    private String serverId;
+    // ========== 基础方法：只传递“codec（编解码）字段” ==========
+
+    public static IotDeviceMessage requestOf(String method) {
+        return requestOf(null, method, null);
+    }
  
-//    public IotDeviceMessage ofPropertyReport(Map<String, Object> properties) {
-//        this.setType(IotDeviceMessageTypeEnum.PROPERTY.getType());
-//        this.setIdentifier(IotDeviceMessageIdentifierEnum.PROPERTY_REPORT.getIdentifier());
-//        this.setData(properties);
-//        return this;
-//    }
-//
-//    public IotDeviceMessage ofPropertySet(Map<String, Object> properties) {
-//        this.setType(IotDeviceMessageTypeEnum.PROPERTY.getType());
-//        this.setIdentifier(IotDeviceMessageIdentifierEnum.PROPERTY_SET.getIdentifier());
-//        this.setData(properties);
-//        return this;
-//    }
-//
-//    public IotDeviceMessage ofStateOnline() {
-//        this.setType(IotDeviceMessageTypeEnum.STATE.getType());
-//        this.setIdentifier(IotDeviceMessageIdentifierEnum.STATE_ONLINE.getIdentifier());
-//        return this;
-//    }
-//
-//    public IotDeviceMessage ofStateOffline() {
-//        this.setType(IotDeviceMessageTypeEnum.STATE.getType());
-//        this.setIdentifier(IotDeviceMessageIdentifierEnum.STATE_OFFLINE.getIdentifier());
-//        return this;
-//    }
-//
-//    public static IotDeviceMessage of(String productKey, String deviceName) {
-//        return of(productKey, deviceName,
-//                null, null);
-//    }
-//
-//    public static IotDeviceMessage of(String productKey, String deviceName,
-//                                      String serverId) {
-//        return of(productKey, deviceName,
-//                null, serverId);
-//    }
-//
-//    public static IotDeviceMessage of(String productKey, String deviceName,
-//                                      LocalDateTime reportTime, String serverId) {
-//        if (reportTime == null) {
-//            reportTime = LocalDateTime.now();
-//        }
-//        String messageId = IotDeviceMessageUtils.generateMessageId();
-//        return IotDeviceMessage.builder()
-//                .messageId(messageId).reportTime(reportTime)
-//                .productKey(productKey).deviceName(deviceName)
-//                .serverId(serverId).build();
-//    }
-
-    public static IotDeviceMessage of(String requestId, String method, Object params) {
-        return of(requestId, method, params, null, null);
+    public static IotDeviceMessage requestOf(String method, Object params) {
+        return requestOf(null, method, params);
+    }
+
+    public static IotDeviceMessage requestOf(String requestId, String method, Object params) {
+        return of(requestId, method, params, null, null, null);
+    }
+
+    public static IotDeviceMessage replyOf(String requestId, String method,
+                                           Object data, Integer code, String msg) {
+        if (code == null) {
+            code = GlobalErrorCodeConstants.SUCCESS.getCode();
+            msg = GlobalErrorCodeConstants.SUCCESS.getMsg();
+        }
+        return of(requestId, method, null, data, code, msg);
     }
  
     public static IotDeviceMessage of(String requestId, String method,
-                                      Object params, Object data, Integer code) {
+                                      Object params, Object data, Integer code, String msg) {
         // 通用参数
         IotDeviceMessage message = new IotDeviceMessage()
                 .setId(IotDeviceMessageUtils.generateMessageId()).setReportTime(LocalDateTime.now());
         // 当前参数
-        message.setRequestId(requestId).setMethod(method).setParams(params).setData(data).setCode(code);
+        message.setRequestId(requestId).setMethod(method).setParams(params)
+                .setData(data).setCode(code).setMsg(msg);
         return message;
     }
  
+    // ========== 核心方法：在 of 基础方法之上，添加对应 method ==========
+
+    public static IotDeviceMessage buildStateOnline() {
+        return requestOf(IotDeviceMessageMethodEnum.STATE_ONLINE.getMethod());
+    }
+
+    public static IotDeviceMessage buildStateOffline() {
+        return requestOf(IotDeviceMessageMethodEnum.STATE_OFFLINE.getMethod());
+    }
+
 }
 package cn.iocoder.yudao.module.iot.core.util;
  
+import cn.hutool.core.lang.Assert;
 import cn.hutool.core.util.IdUtil;
-import cn.hutool.core.util.StrUtil;
 import cn.hutool.system.SystemUtil;
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  
 /**
@@ -18,15 +19,28 @@ public class IotDeviceMessageUtils {
         return IdUtil.fastSimpleUUID();
     }
  
-    // TODO @芋艿：需要优化下；
     /**
      * 是否是上行消息：由设备发送
      *
      * @param message 消息
      * @return 是否
      */
+    @SuppressWarnings("SimplifiableConditionalExpression")
     public static boolean isUpstreamMessage(IotDeviceMessage message) {
-        return StrUtil.isNotEmpty(message.getServerId());
+        IotDeviceMessageMethodEnum methodEnum = IotDeviceMessageMethodEnum.of(message.getMethod());
+        Assert.notNull(methodEnum, "无法识别的消息方法：" + message.getMethod());
+        // 注意：回复消息时，需要取反
+        return !isReplyMessage(message) ? methodEnum.getUpstream() : !methodEnum.getUpstream();
+    }
+
+    /**
+     * 是否是回复消息，通过 {@link IotDeviceMessage#getCode()} 非空进行识别
+     *
+     * @param message 消息
+     * @return 是否
+     */
+    public static boolean isReplyMessage(IotDeviceMessage message) {
+        return message.getCode() != null;
     }
  
     // ========== Topic 相关 ==========
 package cn.iocoder.yudao.module.iot.gateway.codec.alink;
  
 import cn.hutool.core.lang.Assert;
+import cn.iocoder.yudao.framework.common.pojo.CommonResult;
 import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
 import cn.iocoder.yudao.module.iot.gateway.codec.IotDeviceMessageCodec;
@@ -48,18 +49,23 @@ public class IotAlinkDeviceMessageCodec implements IotDeviceMessageCodec {
          * 响应结果
          */
         private Object data;
-
         /**
          * 响应错误码
          */
         private Integer code;
+        /**
+         * 响应提示
+         *
+         * 特殊：这里阿里云是 message，为了保持和项目的 {@link CommonResult#getMsg()} 一致。
+         */
+        private String msg;
  
     }
  
     @Override
     public byte[] encode(IotDeviceMessage message) {
         AlinkMessage alinkMessage = new AlinkMessage(message.getRequestId(), AlinkMessage.VERSION_1,
-                message.getMethod(), message.getParams(), message.getData(), message.getCode());
+                message.getMethod(), message.getParams(), message.getData(), message.getCode(), message.getMsg());
         return JsonUtils.toJsonByte(alinkMessage);
     }
  
@@ -69,8 +75,8 @@ public class IotAlinkDeviceMessageCodec implements IotDeviceMessageCodec {
         AlinkMessage alinkMessage = JsonUtils.parseObject(bytes, AlinkMessage.class);
         Assert.notNull(alinkMessage, "消息不能为空");
         Assert.equals(alinkMessage.getVersion(), AlinkMessage.VERSION_1, "消息版本号必须是 1.0");
-        return IotDeviceMessage.of(alinkMessage.getId(),
-                alinkMessage.getMethod(), alinkMessage.getParams(), alinkMessage.getData(), alinkMessage.getCode());
+        return IotDeviceMessage.of(alinkMessage.getId(), alinkMessage.getMethod(), alinkMessage.getParams(),
+                alinkMessage.getData(), alinkMessage.getCode(), alinkMessage.getMsg());
     }
  
     @Override
@@ -3,8 +3,6 @@ package cn.iocoder.yudao.module.iot.gateway.config;
 import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
 import cn.iocoder.yudao.module.iot.gateway.protocol.http.IotHttpDownstreamSubscriber;
 import cn.iocoder.yudao.module.iot.gateway.protocol.http.IotHttpUpstreamProtocol;
-import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttDownstreamSubscriber;
-import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
@@ -44,16 +42,17 @@ public class IotGatewayConfiguration {
     @Slf4j
     public static class MqttProtocolConfiguration {
  
-        @Bean
-        public IotMqttUpstreamProtocol iotMqttUpstreamProtocol(IotGatewayProperties gatewayProperties) {
-            return new IotMqttUpstreamProtocol(gatewayProperties.getProtocol().getEmqx());
-        }
-
-        @Bean
-        public IotMqttDownstreamSubscriber iotMqttDownstreamSubscriber(IotMqttUpstreamProtocol mqttUpstreamProtocol,
-                                                                       IotMessageBus messageBus) {
-            return new IotMqttDownstreamSubscriber(mqttUpstreamProtocol, messageBus);
-        }
+        // TODO @haohao：临时注释，避免报错
+//        @Bean
+//        public IotMqttUpstreamProtocol iotMqttUpstreamProtocol(IotGatewayProperties gatewayProperties) {
+//            return new IotMqttUpstreamProtocol(gatewayProperties.getProtocol().getEmqx());
+//        }
+//
+//        @Bean
+//        public IotMqttDownstreamSubscriber iotMqttDownstreamSubscriber(IotMqttUpstreamProtocol mqttUpstreamProtocol,
+//                                                                       IotMessageBus messageBus) {
+//            return new IotMqttDownstreamSubscriber(mqttUpstreamProtocol, messageBus);
+//        }
     }
  
 }
 package cn.iocoder.yudao.module.iot.gateway.protocol.http.router;
  
-import cn.hutool.core.collection.CollUtil;
 import cn.hutool.core.lang.Assert;
 import cn.hutool.core.map.MapUtil;
 import cn.hutool.core.text.StrPool;
-import cn.hutool.core.util.StrUtil;
 import cn.hutool.extra.spring.SpringUtil;
 import cn.iocoder.yudao.framework.common.pojo.CommonResult;
 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
 import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer;
-import cn.iocoder.yudao.module.iot.gateway.enums.IotDeviceTopicEnum;
 import cn.iocoder.yudao.module.iot.gateway.protocol.http.IotHttpUpstreamProtocol;
 import cn.iocoder.yudao.module.iot.gateway.service.message.IotDeviceMessageService;
-import io.vertx.core.json.JsonObject;
 import io.vertx.ext.web.RoutingContext;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
  
-import java.util.Map;
-
 /**
  * IoT 网关 HTTP 协议的【上行】处理器
  *
@@ -61,135 +55,4 @@ public class IotHttpUpstreamHandler extends IotHttpAbstractHandler {
         return CommonResult.success(MapUtil.of("messageId", message.getId()));
     }
  
-    /**
-     * 判断是否是属性上报路径
-     *
-     * @param path 路径
-     * @return 是否是属性上报路径
-     */
-    private boolean isPropertyPostPath(String path) {
-        return StrUtil.endWith(path, IotDeviceTopicEnum.PROPERTY_POST_TOPIC.getTopic());
-    }
-
-    /**
-     * 判断是否是事件上报路径
-     *
-     * @param path 路径
-     * @return 是否是事件上报路径
-     */
-    private boolean isEventPostPath(String path) {
-        return StrUtil.contains(path, IotDeviceTopicEnum.EVENT_POST_TOPIC_PREFIX.getTopic())
-                && StrUtil.endWith(path, IotDeviceTopicEnum.EVENT_POST_TOPIC_SUFFIX.getTopic());
-    }
-
-    /**
-     * 处理属性上报请求
-     *
-     * @param routingContext 路由上下文
-     * @param productKey     产品 Key
-     * @param deviceName     设备名称
-     * @param body           请求体
-     */
-    private void handlePropertyPost(RoutingContext routingContext, String productKey, String deviceName,
-                                    JsonObject body) {
-        // 1.1 构建设备消息
-        IotDeviceMessage message = IotDeviceMessage.of(productKey, deviceName, protocol.getServerId())
-//                .ofPropertyReport(parsePropertiesFromBody(body))
-                ;
-        // 1.2 发送消息
-        deviceMessageProducer.sendDeviceMessage(message);
-
-//        // 2. 返回响应
-//        sendResponse(routingContext, null);
-    }
-
-    /**
-     * 处理事件上报请求
-     *
-     * @param routingContext 路由上下文
-     * @param productKey     产品 Key
-     * @param deviceName     设备名称
-     * @param identifier     事件标识符
-     * @param body           请求体
-     */
-    private void handleEventPost(RoutingContext routingContext, String productKey, String deviceName,
-                                 String identifier, JsonObject body) {
-//        // 处理事件上报
-//        IotDeviceEventReportReqDTO reportReqDTO = parseEventReportRequest(productKey, deviceName, identifier,
-//                requestId, body);
-//
-//        // 事件上报
-//        CommonResult<Boolean> result = deviceUpstreamApi.reportDeviceEvent(reportReqDTO);
-    }
-
-    // TODO @芋艿：这块在看看
-    /**
-     * 从请求体解析属性
-     *
-     * @param body 请求体
-     * @return 属性映射
-     */
-    private Map<String, Object> parsePropertiesFromBody(JsonObject body) {
-        Map<String, Object> properties = MapUtil.newHashMap();
-        JsonObject params = body.getJsonObject("params");
-        if (CollUtil.isEmpty(params)) {
-            return properties;
-        }
-
-        // 将标准格式的 params 转换为平台需要的 properties 格式
-        for (String key : params.fieldNames()) {
-            Object valueObj = params.getValue(key);
-            // 如果是复杂结构（包含 value 和 time）
-            if (valueObj instanceof JsonObject) {
-                JsonObject valueJson = (JsonObject) valueObj;
-                properties.put(key, valueJson.containsKey("value") ? valueJson.getValue("value") : valueObj);
-            } else {
-                properties.put(key, valueObj);
-            }
-        }
-        return properties;
-    }
-
-//    /**
-//     * 解析事件上报请求
-//     *
-//     * @param productKey 产品 Key
-//     * @param deviceName 设备名称
-//     * @param identifier 事件标识符
-//     * @param requestId  请求 ID
-//     * @param body       请求体
-//     * @return 事件上报请求 DTO
-//     */
-//    private IotDeviceEventReportReqDTO parseEventReportRequest(String productKey, String deviceName, String identifier,
-//                                                               String requestId, JsonObject body) {
-//        // 解析参数
-//        Map<String, Object> params = parseParamsFromBody(body);
-//
-//        // 构建事件上报请求 DTO
-//        return ((IotDeviceEventReportReqDTO) new IotDeviceEventReportReqDTO()
-//                .setRequestId(requestId)
-//                .setProcessId(IotNetComponentCommonUtils.getProcessId())
-//                .setReportTime(LocalDateTime.now())
-//                .setProductKey(productKey)
-//                .setDeviceName(deviceName)).setIdentifier(identifier).setParams(params);
-//    }
-
-    /**
-     * 从请求体解析参数
-     *
-     * @param body 请求体
-     * @return 参数映射
-     */
-    private Map<String, Object> parseParamsFromBody(JsonObject body) {
-        Map<String, Object> params = MapUtil.newHashMap();
-        JsonObject paramsJson = body.getJsonObject("params");
-        if (CollUtil.isEmpty(paramsJson)) {
-            return params;
-        }
-
-        for (String key : paramsJson.fieldNames()) {
-            params.put(key, paramsJson.getValue(key));
-        }
-        return params;
-    }
 }
 \ No newline at end of file
@@ -96,7 +96,7 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
             return null;
         }
  
-        IotDeviceMessage message = IotDeviceMessage.of(null,
+        IotDeviceMessage message = IotDeviceMessage.requestOf(null,
                 IotDeviceMessageMethodEnum.STATE_ONLINE.getMethod(), null);
  
         return appendDeviceMessage(message, deviceInfo, serverId);
@@ -112,9 +112,8 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
             return null;
         }
  
-        IotDeviceMessage message = IotDeviceMessage.of(null,
-                IotDeviceMessageMethodEnum.STATE_OFFLINE.getMethod(), null);
-
+        IotDeviceMessage message = IotDeviceMessage.requestOf(IotDeviceMessageMethodEnum.STATE_OFFLINE.getMethod(),
+                null);
         return appendDeviceMessage(message, deviceInfo, serverId);
     }
  
@@ -122,23 +121,18 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
      * 补充消息的后端字段
      *
      * @param message    消息
-     * @param deviceInfo 设备信息
+     * @param device 设备信息
      * @param serverId   设备连接的 serverId
      * @return 消息
      */
     private IotDeviceMessage appendDeviceMessage(IotDeviceMessage message,
-                                                 IotDeviceCacheService.DeviceInfo deviceInfo, String serverId) {
+                                                 IotDeviceCacheService.DeviceInfo device, String serverId) {
         message.setId(IotDeviceMessageUtils.generateMessageId()).setReportTime(LocalDateTime.now())
-                .setDeviceId(deviceInfo.getDeviceId()).setTenantId(deviceInfo.getTenantId()).setServerId(serverId);
-
+                .setDeviceId(device.getDeviceId()).setTenantId(device.getTenantId()).setServerId(serverId);
         // 特殊：如果设备没有指定 requestId，则使用 messageId
         if (StrUtil.isEmpty(message.getRequestId())) {
             message.setRequestId(message.getId());
         }
-
-        log.debug("[appendDeviceMessage][消息字段补充完成][deviceId: {}][tenantId: {}]",
-                deviceInfo.getDeviceId(), deviceInfo.getTenantId());
-
         return message;
     }