feat(iot): 新增 MQTT／Database 数据流转 Action 实现，并修复 yudao-module-iot 全部 TODO

puhui999 · YunaiV
1 parent 5b3e1a8e
Showing 22 changed files with 311 additions and 258 deletions
yudao-module-iot/yudao-module-iot-biz/pom.xml
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/config/IotAbstractDataSinkConfig.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/config/IotDataSinkDatabaseConfig.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/thingmodel/model/ThingModelEvent.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/rule/IotDataSinkTypeEnum.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/rule/IotSceneRuleConditionOperatorEnum.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDataRuleCacheableAction.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDatabaseDataRuleAction.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotMqttDataRuleAction.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotWebSocketDataRuleAction.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/websocket/IotWebSocketClient.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotSceneRuleTimeHelper.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/matcher/trigger/IotDeviceServiceInvokeTriggerMatcher.java
yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotTcpDataRuleActionTest.java
yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotSceneRuleServiceSimpleTest.java
yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotSceneRuleTimerConditionIntegrationTest.java
yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/scene/matcher/condition/IotCurrentTimeConditionMatcherTest.java
yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/scene/matcher/condition/IotDeviceStateConditionMatcherTest.java
yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/scene/matcher/trigger/IotDeviceEventPostTriggerMatcherTest.java
yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/scene/matcher/trigger/IotDeviceServiceInvokeTriggerMatcherTest.java
@@ -84,6 +84,13 @@
             <scope>test</scope>
         </dependency>
+        <!-- MQTT -->
+        <dependency>
+            <groupId>org.eclipse.paho</groupId>
+            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
+            <optional>true</optional>
+        </dependency>
+
         <!-- 消息队列相关 -->
         <!-- TODO @芋艿：临时打开 -->
         <dependency>
@@ -20,6 +20,7 @@ import lombok.Data;
         @JsonSubTypes.Type(value = IotDataSinkTcpConfig.class, name = "2"),
         @JsonSubTypes.Type(value = IotDataSinkWebSocketConfig.class, name = "3"),
         @JsonSubTypes.Type(value = IotDataSinkMqttConfig.class, name = "10"),
+        @JsonSubTypes.Type(value = IotDataSinkDatabaseConfig.class, name = "20"),
         @JsonSubTypes.Type(value = IotDataSinkRedisConfig.class, name = "21"),
         @JsonSubTypes.Type(value = IotDataSinkRocketMQConfig.class, name = "30"),
         @JsonSubTypes.Type(value = IotDataSinkRabbitMQConfig.class, name = "31"),
+package cn.iocoder.yudao.module.iot.dal.dataobject.rule.config;
+
+import lombok.Data;
+
+/**
+ * IoT Database 配置 {@link IotAbstractDataSinkConfig} 实现类
+ *
+ * 通过 JDBC 连接数据库，将设备消息写入指定表。
+ * 支持 MySQL、PostgreSQL、Oracle、SQL Server、DM 达梦等数据库，
+ * HikariCP 会根据 JDBC URL 自动加载对应的驱动。
+ *
+ * @author HUIHUI
+ */
+@Data
+public class IotDataSinkDatabaseConfig extends IotAbstractDataSinkConfig {
+
+    /**
+     * JDBC 连接地址
+     *
+     * 例如：jdbc:mysql://localhost:3306/iot_data
+     * 例如：jdbc:postgresql://localhost:5432/iot_data
+     * 例如：jdbc:dm://localhost:5236/iot_data
+     *
+     * HikariCP 会根据 URL 自动检测并加载对应的 JDBC 驱动
+     */
+    private String jdbcUrl;
+    /**
+     * 数据库用户名
+     */
+    private String username;
+    /**
+     * 数据库密码
+     */
+    private String password;
+    /**
+     * 目标表名
+     *
+     * 设备消息将以固定结构写入该表
+     */
+    private String tableName;
+
+}
@@ -9,7 +9,6 @@ import lombok.Data;
 import java.util.List;
-// TODO @puhui999：感觉这个，是不是放到 dal 里会好点？（讨论下，先不改哈）
 /**
  * IoT 物模型中的事件
  *
@@ -19,9 +19,9 @@ public enum IotDataSinkTypeEnum implements ArrayValuable&lt;Integer&gt; {
     TCP(2, "TCP"),
     WEBSOCKET(3, "WebSocket"),
-    MQTT(10, "MQTT"), // TODO @puhui999：待实现；
+    MQTT(10, "MQTT"),
-    DATABASE(20, "Database"), // TODO @puhui999：待实现；
+    DATABASE(20, "Database"),
     REDIS(21, "Redis"),
     ROCKETMQ(30, "RocketMQ"),
@@ -36,13 +36,11 @@ public enum IotSceneRuleConditionOperatorEnum implements ArrayValuable&lt;String&gt; {
     // ========== 特殊：不放在字典里 ==========
-    // TODO @puhui999：@芋艿：需要测试下
     DATE_TIME_GREATER_THAN("date_time_>", "#source > #value"), // 在时间之后：时间戳
     DATE_TIME_LESS_THAN("date_time_<", "#source < #value"), // 在时间之前：时间戳
     DATE_TIME_BETWEEN("date_time_between", // 在时间之间：时间戳
             "(#source >= #values.get(0)) && (#source <= #values.get(1))"),
-    // TODO @puhui999：@芋艿：需要测试下
     TIME_GREATER_THAN("time_>", "#source.isAfter(#value)"), // 在当日时间之后：HH:mm:ss
     TIME_LESS_THAN("time_<", "#source.isBefore(#value)"), // 在当日时间之前：HH:mm:ss
     TIME_BETWEEN("time_between", // 在当日时间之间：HH:mm:ss
@@ -12,9 +12,6 @@ import lombok.extern.slf4j.Slf4j;
 import java.time.Duration;
-// TODO @芋艿：数据库
-// TODO @芋艿：mqtt
-
 /**
  * 可缓存的 {@link IotDataRuleAction} 抽象实现
  *
@@ -78,6 +75,18 @@ public abstract class IotDataRuleCacheableAction&lt;Config, Producer&gt; implements Io
     }
     /**
+     * 使指定配置的 Producer 缓存失效
+     *
+     * 当子类检测到 Producer 不可用时（如连接断开），可调用此方法踢出缓存，
+     * 下次 {@link #getProducer(Object)} 调用将触发 {@link #initProducer(Object)} 重新创建。
+     *
+     * @param config 配置信息
+     */
+    protected void invalidateProducer(Config config) {
+        PRODUCER_CACHE.invalidate(config);
+    }
+
+    /**
      * 初始化生产者
      *
      * @param config 配置信息
+package cn.iocoder.yudao.module.iot.service.rule.data.action;
+
+import cn.hutool.core.util.StrUtil;
+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.dal.dataobject.rule.config.IotDataSinkDatabaseConfig;
+import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum;
+import com.zaxxer.hikari.HikariConfig;
+import com.zaxxer.hikari.HikariDataSource;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Component;
+
+/**
+ * Database 的 {@link IotDataRuleAction} 实现类
+ *
+ * 通过 JDBC 连接数据库，将设备消息写入指定表。
+ * 支持 MySQL、PostgreSQL、Oracle、SQL Server、DM 达梦等数据库，
+ * HikariCP 会根据 JDBC URL 自动加载对应的驱动。
+ *
+ * @author HUIHUI
+ */
+@Component
+@Slf4j
+public class IotDatabaseDataRuleAction extends
+        IotDataRuleCacheableAction<IotDataSinkDatabaseConfig, JdbcTemplate> {
+
+    @Override
+    public Integer getType() {
+        return IotDataSinkTypeEnum.DATABASE.getType();
+    }
+
+    @Override
+    public void execute(IotDeviceMessage message, IotDataSinkDatabaseConfig config) throws Exception {
+        try {
+            // 1. 获取或创建 JdbcTemplate
+            JdbcTemplate jdbcTemplate = getProducer(config);
+
+            // 2. 构建并执行 INSERT SQL
+            String sql = StrUtil.format(
+                    "INSERT INTO {} (id, device_id, tenant_id, method, report_time, data, create_time) VALUES (?, ?, ?, ?, ?, ?, NOW())",
+                    config.getTableName());
+            String messageJson = JsonUtils.toJsonString(message);
+            jdbcTemplate.update(sql,
+                    message.getId(),
+                    message.getDeviceId(),
+                    message.getTenantId(),
+                    message.getMethod(),
+                    message.getReportTime(),
+                    messageJson);
+            log.info("[execute][message({}) config({}) 写入数据库成功，table: {}]",
+                    message.getId(), config, config.getTableName());
+        } catch (Exception e) {
+            log.error("[execute][message({}) config({}) 写入数据库失败]", message, config, e);
+            throw e;
+        }
+    }
+
+    @Override
+    protected JdbcTemplate initProducer(IotDataSinkDatabaseConfig config) throws Exception {
+        // 使用 HikariCP 连接池，HikariCP 会根据 JDBC URL 自动检测并加载对应的数据库驱动
+        HikariConfig hikariConfig = new HikariConfig();
+        hikariConfig.setJdbcUrl(config.getJdbcUrl());
+        hikariConfig.setUsername(config.getUsername());
+        hikariConfig.setPassword(config.getPassword());
+        // 连接池配置
+        hikariConfig.setMaximumPoolSize(5); // 数据流转场景，不需要太多连接
+        hikariConfig.setMinimumIdle(1);
+        hikariConfig.setConnectionTimeout(10000); // 连接超时 10 秒
+        hikariConfig.setIdleTimeout(300000); // 空闲超时 5 分钟
+        hikariConfig.setMaxLifetime(600000); // 最大生命周期 10 分钟
+        HikariDataSource dataSource = new HikariDataSource(hikariConfig);
+        log.info("[initProducer][数据库连接池创建成功，jdbcUrl: {}]", config.getJdbcUrl());
+        return new JdbcTemplate(dataSource);
+    }
+
+    @Override
+    protected void closeProducer(JdbcTemplate producer) throws Exception {
+        if (producer.getDataSource() instanceof HikariDataSource) {
+            ((HikariDataSource) producer.getDataSource()).close();
+        }
+    }
+
+}
+package cn.iocoder.yudao.module.iot.service.rule.data.action;
+
+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.dal.dataobject.rule.config.IotDataSinkMqttConfig;
+import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.stereotype.Component;
+
+import java.nio.charset.StandardCharsets;
+
+/**
+ * MQTT 的 {@link IotDataRuleAction} 实现类
+ *
+ * @author HUIHUI
+ */
+@ConditionalOnClass(name = "org.eclipse.paho.client.mqttv3.MqttClient")
+@Component
+@Slf4j
+public class IotMqttDataRuleAction extends
+        IotDataRuleCacheableAction<IotDataSinkMqttConfig, MqttClient> {
+
+    /**
+     * 默认 QoS 等级（至少一次）
+     */
+    private static final int DEFAULT_QOS = 1;
+
+    @Override
+    public Integer getType() {
+        return IotDataSinkTypeEnum.MQTT.getType();
+    }
+
+    @Override
+    public void execute(IotDeviceMessage message, IotDataSinkMqttConfig config) throws Exception {
+        try {
+            // 1. 获取或创建 MqttClient
+            MqttClient mqttClient = getProducer(config);
+
+            // 2.1 检查连接状态，如果断开则踢出缓存并重新创建
+            if (!mqttClient.isConnected()) {
+                log.warn("[execute][MQTT 连接已断开，重新创建客户端，服务器: {}]", config.getUrl());
+                invalidateProducer(config); // 踢出旧的断连客户端，触发 closeProducer
+                mqttClient = getProducer(config); // 触发 initProducer 创建全新连接
+            }
+
+            // 2.2 构建并发送消息
+            MqttMessage mqttMessage = new MqttMessage(JsonUtils.toJsonString(message).getBytes(StandardCharsets.UTF_8));
+            mqttMessage.setQos(DEFAULT_QOS);
+            mqttClient.publish(config.getTopic(), mqttMessage);
+            log.info("[execute][message({}) 发送成功，MQTT 服务器: {}，topic: {}]",
+                    message.getId(), config.getUrl(), config.getTopic());
+        } catch (Exception e) {
+            log.error("[execute][message({}) 发送失败，MQTT 服务器: {}]",
+                    message.getId(), config.getUrl(), e);
+            throw e;
+        }
+    }
+
+    @Override
+    protected MqttClient initProducer(IotDataSinkMqttConfig config) throws Exception {
+        // 1. 创建 MqttClient，使用内存持久化
+        // 拼接时间戳后缀，避免多个规则指向同一 Broker 时 clientId 冲突
+        String clientId = config.getClientId() + "_" + System.currentTimeMillis();
+        MqttClient mqttClient = new MqttClient(config.getUrl(), clientId, new MemoryPersistence());
+
+        // 2. 连接到 MQTT Broker
+        mqttClient.connect(buildConnectOptions(config));
+        log.info("[initProducer][MQTT 客户端创建并连接成功，服务器: {}，clientId: {}]",
+                config.getUrl(), clientId);
+        return mqttClient;
+    }
+
+    @Override
+    protected void closeProducer(MqttClient producer) throws Exception {
+        if (producer.isConnected()) {
+            producer.disconnect();
+        }
+        producer.close();
+    }
+
+    /**
+     * 构建 MQTT 连接选项
+     *
+     * @param config MQTT 配置
+     * @return 连接选项
+     */
+    private MqttConnectOptions buildConnectOptions(IotDataSinkMqttConfig config) {
+        MqttConnectOptions options = new MqttConnectOptions();
+        options.setCleanSession(true);
+        options.setConnectionTimeout(10); // 连接超时 10 秒
+        options.setKeepAliveInterval(20); // 心跳间隔 20 秒
+        // 注意：不开启 automaticReconnect，由 execute() 中的 isConnected() 手动控制重连，避免竞争
+        // 设置认证信息（如果有）
+        if (config.getUsername() != null) {
+            options.setUserName(config.getUsername());
+        }
+        if (config.getPassword() != null) {
+            options.setPassword(config.getPassword().toCharArray());
+        }
+        return options;
+    }
+
+}
@@ -97,7 +97,8 @@ public class IotWebSocketDataRuleAction extends
         }
     }
-    // TODO @puhui999：为什么这里要加锁呀？
+    // 加锁原因：多线程并发发送消息时，多个线程可能同时检测到连接断开并尝试重连，
+    // 使用锁 + 双重检查保证同一服务器地址只有一个线程执行重连操作
     /**
      * 使用锁进行重连，保证同一服务器地址的重连操作线程安全
      *
@@ -33,6 +33,13 @@ public class IotWebSocketClient {
     private volatile WebSocket webSocket;
     private final AtomicBoolean connected = new AtomicBoolean(false);
+    /**
+     * WebSocket 正常关闭状态码
+     *
+     * @see <a href="https://tools.ietf.org/html/rfc6455#section-7.4.1">RFC 6455 - 定义的状态码</a>
+     */
+    private static final int NORMAL_CLOSURE_STATUS = 1000;
+
     public IotWebSocketClient(String serverUrl, Integer connectTimeoutMs, Integer sendTimeoutMs, String dataFormat) {
         this.serverUrl = serverUrl;
         this.connectTimeoutMs = connectTimeoutMs != null ? connectTimeoutMs : IotDataSinkWebSocketConfig.DEFAULT_CONNECT_TIMEOUT_MS;
@@ -123,9 +130,8 @@ public class IotWebSocketClient {
     public void close() {
         try {
             if (webSocket != null) {
-                // 发送正常关闭帧，状态码 1000 表示正常关闭
-                // TODO @puhui999：有没 1000 的枚举哈？在 okhttp 里
-                webSocket.close(1000, "客户端主动关闭");
+                // 发送正常关闭帧
+                webSocket.close(NORMAL_CLOSURE_STATUS, "客户端主动关闭");
                 webSocket = null;
             }
             if (okHttpClient != null) {
@@ -6,6 +6,7 @@ import cn.hutool.core.util.StrUtil;
 import cn.iocoder.yudao.module.iot.enums.rule.IotSceneRuleConditionOperatorEnum;
 import cn.iocoder.yudao.module.iot.service.rule.scene.matcher.condition.IotCurrentTimeConditionMatcher;
 import cn.iocoder.yudao.module.iot.service.rule.scene.timer.IotTimerConditionEvaluator;
+import lombok.experimental.UtilityClass;
 import lombok.extern.slf4j.Slf4j;
 import java.time.LocalDateTime;
@@ -22,6 +23,7 @@ import java.util.List;
  * @author HUIHUI
  */
 @Slf4j
+@UtilityClass
 public class IotSceneRuleTimeHelper {
     /**
@@ -34,11 +36,6 @@ public class IotSceneRuleTimeHelper {
      */
     private static final DateTimeFormatter TIME_FORMATTER_SHORT = DateTimeFormatter.ofPattern("HH:mm");
-    // TODO @puhui999：可以使用 lombok 简化
-    private IotSceneRuleTimeHelper() {
-        // 工具类，禁止实例化
-    }
-
     /**
      * 判断是否为日期时间操作符
      *
@@ -136,7 +133,6 @@ public class IotSceneRuleTimeHelper {
         }
         long startTimestamp = Long.parseLong(timestampRange.get(0).trim());
         long endTimestamp = Long.parseLong(timestampRange.get(1).trim());
-        // TODO @puhui999：hutool 里，看看有没 between 方法
         return currentTimestamp >= startTimestamp && currentTimestamp <= endTimestamp;
     }
@@ -188,7 +184,6 @@ public class IotSceneRuleTimeHelper {
         }
         LocalTime startTime = parseTime(timeRange.get(0).trim());
         LocalTime endTime = parseTime(timeRange.get(1).trim());
-        // TODO @puhui999：hutool 里，看看有没 between 方法
         return !currentTime.isBefore(startTime) && !currentTime.isAfter(endTime);
     }
 package cn.iocoder.yudao.module.iot.service.rule.scene.matcher.trigger;
+import cn.hutool.core.collection.CollUtil;
 import cn.hutool.core.util.StrUtil;
 import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
@@ -73,8 +74,7 @@ public class IotDeviceServiceInvokeTriggerMatcher implements IotSceneRuleTrigger
     private boolean matchParameterCondition(IotDeviceMessage message, IotSceneRuleDO.Trigger trigger) {
         // 1.1 从消息中提取服务调用的输入参数
         Map<String, Object> inputParams = IotDeviceMessageUtils.extractServiceInputParams(message);
-        // TODO @puhui999：要考虑 empty 的情况么？
-        if (inputParams == null) {
+        if (CollUtil.isEmpty(inputParams)) {
             IotSceneRuleMatcherHelper.logTriggerMatchFailure(message, trigger, "消息中缺少服务输入参数");
             return false;
         }
@@ -9,9 +9,6 @@ import org.junit.jupiter.api.Test;
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
-import java.util.HashMap;
-import java.util.Map;
-
 import static org.junit.jupiter.api.Assertions.*;
 import static org.mockito.Mockito.*;
@@ -45,9 +42,8 @@ class IotTcpDataRuleActionTest {
         assertEquals(expectedType, actualType);
     }
-    // TODO @puhui999：_ 后面是小写哈，单测的命名规则。
     @Test
-    public void testInitProducer_Success() throws Exception {
+    public void testInitProducer_success() throws Exception {
         // 准备参数
         IotDataSinkTcpConfig config = new IotDataSinkTcpConfig();
         config.setHost("localhost");
@@ -63,7 +59,7 @@ class IotTcpDataRuleActionTest {
     }
     @Test
-    public void testInitProducer_InvalidHost() {
+    public void testInitProducer_invalidHost() {
         // 准备参数
         IotDataSinkTcpConfig config = new IotDataSinkTcpConfig();
         config.setHost("");
@@ -81,7 +77,7 @@ class IotTcpDataRuleActionTest {
     }
     @Test
-    public void testInitProducer_InvalidPort() {
+    public void testInitProducer_invalidPort() {
         // 准备参数
         IotDataSinkTcpConfig config = new IotDataSinkTcpConfig();
         config.setHost("localhost");
@@ -108,7 +104,7 @@ class IotTcpDataRuleActionTest {
     }
     @Test
-    public void testExecute_WithValidConfig() {
+    public void testExecute_withValidConfig() {
         // 准备参数
         IotDeviceMessage message = IotDeviceMessage.requestOf("thing.property.report",
                 "{\"temperature\": 25.5, \"humidity\": 60}");
@@ -128,7 +124,7 @@ class IotTcpDataRuleActionTest {
     }
     @Test
-    public void testConfig_DefaultValues() {
+    public void testConfig_defaultValues() {
         // 准备参数
         IotDataSinkTcpConfig config = new IotDataSinkTcpConfig();
@@ -146,12 +142,10 @@ class IotTcpDataRuleActionTest {
     @Test
     public void testMessageSerialization() {
         // 准备参数
-        Map<String, Object> params = new HashMap<>();
-        params.put("temperature", 25.5);
         IotDeviceMessage message = IotDeviceMessage.builder()
                 .deviceId(123L)
                 .method("thing.property.report")
-                .params(params)
+                .params("{\"temperature\": 25.5}")
                 .build();
         // 调用方法
@@ -5,11 +5,12 @@ import cn.iocoder.yudao.framework.test.core.ut.BaseMockitoUnitTest;
 import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.scene.IotSceneRuleSaveReqVO;
 import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotSceneRuleDO;
 import cn.iocoder.yudao.module.iot.dal.mysql.rule.IotSceneRuleMapper;
-import cn.iocoder.yudao.module.iot.framework.job.core.IotSchedulerManager;
 import cn.iocoder.yudao.module.iot.service.device.IotDeviceService;
 import cn.iocoder.yudao.module.iot.service.product.IotProductService;
 import cn.iocoder.yudao.module.iot.service.rule.scene.action.IotSceneRuleAction;
+import cn.iocoder.yudao.module.iot.service.rule.scene.matcher.IotSceneRuleMatcherManager;
 import cn.iocoder.yudao.module.iot.service.rule.scene.timer.IotSceneRuleTimerHandler;
+import cn.iocoder.yudao.module.iot.service.rule.scene.timer.IotTimerConditionEvaluator;
 import org.junit.jupiter.api.Test;
 import org.mockito.InjectMocks;
 import org.mockito.Mock;
@@ -42,19 +43,22 @@ public class IotSceneRuleServiceSimpleTest extends BaseMockitoUnitTest {
     private List<IotSceneRuleAction> sceneRuleActions;
     @Mock
-    private IotSchedulerManager schedulerManager;
+    private IotSceneRuleTimerHandler timerHandler;
     @Mock
-    private IotProductService productService;
+    private IotTimerConditionEvaluator timerConditionEvaluator;
     @Mock
-    private IotDeviceService deviceService;
+    private IotSceneRuleMatcherManager sceneRuleMatcherManager;
     @Mock
-    private IotSceneRuleTimerHandler timerHandler;
+    private IotProductService productService;
+
+    @Mock
+    private IotDeviceService deviceService;
     @Test
-    public void testCreateScene_Rule_success() {
+    public void testCreateScene_rule_success() {
         // 准备参数
         IotSceneRuleSaveReqVO createReqVO = randomPojo(IotSceneRuleSaveReqVO.class, o -> {
             o.setId(null);
@@ -80,7 +84,7 @@ public class IotSceneRuleServiceSimpleTest extends BaseMockitoUnitTest {
     }
     @Test
-    public void testUpdateScene_Rule_success() {
+    public void testUpdateScene_rule_success() {
         // 准备参数
         Long id = randomLongId();
         IotSceneRuleSaveReqVO updateReqVO = randomPojo(IotSceneRuleSaveReqVO.class, o -> {
 package cn.iocoder.yudao.module.iot.service.rule.scene;
 import cn.hutool.core.collection.ListUtil;
-import cn.hutool.extra.spring.SpringUtil;
 import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
 import cn.iocoder.yudao.framework.test.core.ut.BaseMockitoUnitTest;
 import cn.iocoder.yudao.module.iot.core.enums.device.IotDeviceStateEnum;
@@ -15,12 +14,13 @@ import cn.iocoder.yudao.module.iot.enums.rule.IotSceneRuleTriggerTypeEnum;
 import cn.iocoder.yudao.module.iot.service.device.IotDeviceService;
 import cn.iocoder.yudao.module.iot.service.device.property.IotDevicePropertyService;
 import cn.iocoder.yudao.module.iot.service.rule.scene.action.IotSceneRuleAction;
+import cn.iocoder.yudao.module.iot.service.rule.scene.matcher.IotSceneRuleMatcherManager;
 import cn.iocoder.yudao.module.iot.service.rule.scene.timer.IotSceneRuleTimerHandler;
 import cn.iocoder.yudao.module.iot.service.rule.scene.timer.IotTimerConditionEvaluator;
+import cn.iocoder.yudao.module.iot.service.product.IotProductService;
 import org.junit.jupiter.api.*;
 import org.mockito.InjectMocks;
 import org.mockito.Mock;
-import org.springframework.context.support.GenericApplicationContext;
 import java.lang.reflect.Field;
 import java.util.*;
@@ -63,6 +63,12 @@ public class IotSceneRuleTimerConditionIntegrationTest extends BaseMockitoUnitTe
     @Mock
     private IotSceneRuleTimerHandler timerHandler;
+    @Mock
+    private IotSceneRuleMatcherManager sceneRuleMatcherManager;
+
+    @Mock
+    private IotProductService productService;
+
     private IotTimerConditionEvaluator timerConditionEvaluator;
     // 测试常量
@@ -71,12 +77,6 @@ public class IotSceneRuleTimerConditionIntegrationTest extends BaseMockitoUnitTe
     private static final Long DEVICE_ID = 100L;
     private static final String PROPERTY_IDENTIFIER = "temperature";
-    @BeforeAll
-    static void initSpringContext() {
-        // SpringExpressionUtils 依赖 SpringUtil 持有的 ApplicationContext
-        new SpringUtil().setApplicationContext(new GenericApplicationContext());
-    }
-
     @BeforeEach
     void setUp() {
         // 创建并注入 timerConditionEvaluator 的依赖
@@ -144,15 +144,14 @@ public class IotSceneRuleTimerConditionIntegrationTest extends BaseMockitoUnitTe
         IotDevicePropertyDO property = new IotDevicePropertyDO();
         property.setValue(value);
         properties.put(identifier, property);
-        // 多条件组 OR 短路时部分 stub 用不上，使用 lenient 容忍未消费的 stub
-        lenient().when(devicePropertyService.getLatestDeviceProperties(deviceId)).thenReturn(properties);
+        when(devicePropertyService.getLatestDeviceProperties(deviceId)).thenReturn(properties);
     }
     private void mockDeviceState(Long deviceId, Integer state) {
         IotDeviceDO device = new IotDeviceDO();
         device.setId(deviceId);
         device.setState(state);
-        lenient().when(deviceService.getDevice(deviceId)).thenReturn(device);
+        when(deviceService.getDevice(deviceId)).thenReturn(device);
     }
     /**
@@ -59,7 +59,7 @@ public class IotCurrentTimeConditionMatcherTest extends IotBaseConditionMatcherT
     // ========== 时间戳条件测试 ==========
     @Test
-    public void testMatches_DateTimeGreaterThan_success() {
+    public void testMatches_dateTimeGreaterThan_success() {
         // 准备参数
         IotDeviceMessage message = createDeviceMessage();
         long pastTimestamp = LocalDateTime.now().minusHours(1).toEpochSecond(ZoneOffset.of("+8"));
@@ -76,7 +76,7 @@ public class IotCurrentTimeConditionMatcherTest extends IotBaseConditionMatcherT
     }
     @Test
-    public void testMatches_DateTimeGreaterThan_fail() {
+    public void testMatches_dateTimeGreaterThan_fail() {
         // 准备参数
         IotDeviceMessage message = createDeviceMessage();
         long futureTimestamp = LocalDateTime.now().plusHours(1).toEpochSecond(ZoneOffset.of("+8"));
@@ -93,7 +93,7 @@ public class IotCurrentTimeConditionMatcherTest extends IotBaseConditionMatcherT
     }
     @Test
-    public void testMatches_DateTimeLessThan_success() {
+    public void testMatches_dateTimeLessThan_success() {
         // 准备参数
         IotDeviceMessage message = createDeviceMessage();
         long futureTimestamp = LocalDateTime.now().plusHours(1).toEpochSecond(ZoneOffset.of("+8"));
@@ -110,7 +110,7 @@ public class IotCurrentTimeConditionMatcherTest extends IotBaseConditionMatcherT
     }
     @Test
-    public void testMatches_DateTimeBetween_success() {
+    public void testMatches_dateTimeBetween_success() {
         // 准备参数
         IotDeviceMessage message = createDeviceMessage();
         long startTimestamp = LocalDateTime.now().minusHours(1).toEpochSecond(ZoneOffset.of("+8"));
@@ -128,7 +128,7 @@ public class IotCurrentTimeConditionMatcherTest extends IotBaseConditionMatcherT
     }
     @Test
-    public void testMatches_DateTimeBetween_fail() {
+    public void testMatches_dateTimeBetween_fail() {
         // 准备参数
         IotDeviceMessage message = createDeviceMessage();
         long startTimestamp = LocalDateTime.now().plusHours(1).toEpochSecond(ZoneOffset.of("+8"));
@@ -148,7 +148,7 @@ public class IotCurrentTimeConditionMatcherTest extends IotBaseConditionMatcherT
     // ========== 当日时间条件测试 ==========
     @Test
-    public void testMatches_TimeGreaterThan_earlyMorning() {
+    public void testMatches_timeGreaterThan_earlyMorning() {
         // 准备参数
         IotDeviceMessage message = createDeviceMessage();
         IotSceneRuleDO.TriggerCondition condition = createTimeCondition(
@@ -165,7 +165,7 @@ public class IotCurrentTimeConditionMatcherTest extends IotBaseConditionMatcherT
     }
     @Test
-    public void testMatches_TimeLessThan_lateNight() {
+    public void testMatches_timeLessThan_lateNight() {
         // 准备参数
         IotDeviceMessage message = createDeviceMessage();
         IotSceneRuleDO.TriggerCondition condition = createTimeCondition(
@@ -182,7 +182,7 @@ public class IotCurrentTimeConditionMatcherTest extends IotBaseConditionMatcherT
     }
     @Test
-    public void testMatches_TimeBetween_allDay() {
+    public void testMatches_timeBetween_allDay() {
         // 准备参数
         IotDeviceMessage message = createDeviceMessage();
         IotSceneRuleDO.TriggerCondition condition = createTimeCondition(
@@ -198,7 +198,7 @@ public class IotCurrentTimeConditionMatcherTest extends IotBaseConditionMatcherT
     }
     @Test
-    public void testMatches_TimeBetween_workingHours() {
+    public void testMatches_timeBetween_workingHours() {
         // 准备参数
         IotDeviceMessage message = createDeviceMessage();
         IotSceneRuleDO.TriggerCondition condition = createTimeCondition(
 package cn.iocoder.yudao.module.iot.service.rule.scene.matcher.condition;
-import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
 import cn.iocoder.yudao.module.iot.core.enums.device.IotDeviceStateEnum;
 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
 import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotSceneRuleDO;
@@ -10,9 +9,6 @@ import cn.iocoder.yudao.module.iot.service.rule.scene.matcher.IotBaseConditionMa
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import java.util.HashMap;
-import java.util.Map;
-
 import static cn.iocoder.yudao.framework.test.core.util.RandomUtils.randomLongId;
 import static cn.iocoder.yudao.framework.test.core.util.RandomUtils.randomString;
 import static org.junit.jupiter.api.Assertions.*;
@@ -346,10 +342,7 @@ public class IotDeviceStateConditionMatcherTest extends IotBaseConditionMatcherT
     private IotDeviceMessage createDeviceMessage(Integer deviceState) {
         IotDeviceMessage message = new IotDeviceMessage();
         message.setDeviceId(randomLongId());
-        message.setMethod(IotDeviceMessageMethodEnum.STATE_UPDATE.getMethod());
-        Map<String, Object> params = new HashMap<>();
-        params.put("state", deviceState);
-        message.setParams(params);
+        message.setParams(deviceState);
         return message;
     }
 package cn.iocoder.yudao.module.iot.service.rule.scene.matcher.trigger;
-import cn.hutool.core.collection.ListUtil;
 import cn.hutool.core.map.MapUtil;
 import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
-import cn.iocoder.yudao.module.iot.core.topic.event.IotDeviceEventPostReqDTO;
 import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotSceneRuleDO;
-import cn.iocoder.yudao.module.iot.enums.rule.IotSceneRuleConditionOperatorEnum;
 import cn.iocoder.yudao.module.iot.enums.rule.IotSceneRuleTriggerTypeEnum;
 import cn.iocoder.yudao.module.iot.service.rule.scene.matcher.IotBaseConditionMatcherTest;
 import org.junit.jupiter.api.BeforeEach;
@@ -357,184 +354,6 @@ public class IotDeviceEventPostTriggerMatcherTest extends IotBaseConditionMatche
         assertFalse(result);
     }
-    @Test
-    public void testMatches_scalarValueEqualsSuccess() {
-        // 标量事件值 + operator='=' + value 命中：必须匹配；防止"把整个 params Map 当作源值"的回归
-        String eventIdentifier = randomString();
-        Map<String, Object> eventParams = MapUtil.builder(new HashMap<String, Object>())
-                .put("identifier", eventIdentifier)
-                .put("value", "normal")
-                .build();
-        IotDeviceMessage message = createEventPostMessage(eventParams);
-        IotSceneRuleDO.Trigger trigger = createValidTrigger(eventIdentifier);
-        trigger.setOperator(IotSceneRuleConditionOperatorEnum.EQUALS.getOperator());
-        trigger.setValue("normal");
-
-        boolean result = matcher.matches(message, trigger);
-
-        assertTrue(result);
-    }
-
-    @Test
-    public void testMatches_scalarValueEqualsMismatch() {
-        // 标量事件值不等于配置值：应不匹配
-        String eventIdentifier = randomString();
-        Map<String, Object> eventParams = MapUtil.builder(new HashMap<String, Object>())
-                .put("identifier", eventIdentifier)
-                .put("value", "abnormal")
-                .build();
-        IotDeviceMessage message = createEventPostMessage(eventParams);
-        IotSceneRuleDO.Trigger trigger = createValidTrigger(eventIdentifier);
-        trigger.setOperator(IotSceneRuleConditionOperatorEnum.EQUALS.getOperator());
-        trigger.setValue("normal");
-
-        boolean result = matcher.matches(message, trigger);
-
-        assertFalse(result);
-    }
-
-    @Test
-    public void testMatches_pojoParamsScalarValueSuccess() {
-        // 本地总线场景：params 是 IotDeviceEventPostReqDTO POJO 而非 Map，仍应能正确匹配
-        String eventIdentifier = randomString();
-        IotDeviceMessage message = new IotDeviceMessage();
-        message.setDeviceId(randomLongId());
-        message.setMethod(IotDeviceMessageMethodEnum.EVENT_POST.getMethod());
-        message.setParams(IotDeviceEventPostReqDTO.of(eventIdentifier, "normal"));
-        IotSceneRuleDO.Trigger trigger = createValidTrigger(eventIdentifier);
-        trigger.setOperator(IotSceneRuleConditionOperatorEnum.EQUALS.getOperator());
-        trigger.setValue("normal");
-
-        boolean result = matcher.matches(message, trigger);
-
-        assertTrue(result);
-    }
-
-    @Test
-    public void testMatches_structValueEqualsSuccess() {
-        // 结构体事件值 + 比较值是 JSON 对象字面量：JSON 反序列化后整体相等
-        String eventIdentifier = randomString();
-        Map<String, Object> eventValue = MapUtil.builder(new HashMap<String, Object>())
-                .put("level", "high")
-                .put("message", "over temperature")
-                .build();
-        Map<String, Object> eventParams = MapUtil.builder(new HashMap<String, Object>())
-                .put("identifier", eventIdentifier)
-                .put("value", eventValue)
-                .build();
-        IotDeviceMessage message = createEventPostMessage(eventParams);
-        IotSceneRuleDO.Trigger trigger = createValidTrigger(eventIdentifier);
-        trigger.setOperator(IotSceneRuleConditionOperatorEnum.EQUALS.getOperator());
-        trigger.setValue("{\"level\":\"high\",\"message\":\"over temperature\"}");
-
-        boolean result = matcher.matches(message, trigger);
-
-        assertTrue(result);
-    }
-
-    @Test
-    public void testMatches_structValueEqualsKeyOrderInsensitive() {
-        // 比较值的 JSON 字段顺序与事件 value 不同，应仍然匹配（HashMap.equals 与顺序无关）
-        String eventIdentifier = randomString();
-        Map<String, Object> eventValue = MapUtil.builder(new HashMap<String, Object>())
-                .put("level", "high")
-                .put("code", 500)
-                .build();
-        Map<String, Object> eventParams = MapUtil.builder(new HashMap<String, Object>())
-                .put("identifier", eventIdentifier)
-                .put("value", eventValue)
-                .build();
-        IotDeviceMessage message = createEventPostMessage(eventParams);
-        IotSceneRuleDO.Trigger trigger = createValidTrigger(eventIdentifier);
-        trigger.setOperator(IotSceneRuleConditionOperatorEnum.EQUALS.getOperator());
-        trigger.setValue("{\"code\":500,\"level\":\"high\"}");
-
-        boolean result = matcher.matches(message, trigger);
-
-        assertTrue(result);
-    }
-
-    @Test
-    public void testMatches_structValueEqualsMismatch() {
-        // 结构体值字段不一致：应不匹配
-        String eventIdentifier = randomString();
-        Map<String, Object> eventValue = MapUtil.builder(new HashMap<String, Object>())
-                .put("level", "low")
-                .build();
-        Map<String, Object> eventParams = MapUtil.builder(new HashMap<String, Object>())
-                .put("identifier", eventIdentifier)
-                .put("value", eventValue)
-                .build();
-        IotDeviceMessage message = createEventPostMessage(eventParams);
-        IotSceneRuleDO.Trigger trigger = createValidTrigger(eventIdentifier);
-        trigger.setOperator(IotSceneRuleConditionOperatorEnum.EQUALS.getOperator());
-        trigger.setValue("{\"level\":\"high\"}");
-
-        boolean result = matcher.matches(message, trigger);
-
-        assertFalse(result);
-    }
-
-    @Test
-    public void testMatches_structValueNotEqualsSuccess() {
-        // 结构体值 + != 操作符：内容不一致时应匹配
-        String eventIdentifier = randomString();
-        Map<String, Object> eventValue = MapUtil.builder(new HashMap<String, Object>())
-                .put("level", "low")
-                .build();
-        Map<String, Object> eventParams = MapUtil.builder(new HashMap<String, Object>())
-                .put("identifier", eventIdentifier)
-                .put("value", eventValue)
-                .build();
-        IotDeviceMessage message = createEventPostMessage(eventParams);
-        IotSceneRuleDO.Trigger trigger = createValidTrigger(eventIdentifier);
-        trigger.setOperator(IotSceneRuleConditionOperatorEnum.NOT_EQUALS.getOperator());
-        trigger.setValue("{\"level\":\"high\"}");
-
-        boolean result = matcher.matches(message, trigger);
-
-        assertTrue(result);
-    }
-
-    @Test
-    public void testMatches_arrayValueEqualsSuccess() {
-        // 数组事件值：JSON 解析后按序相等
-        String eventIdentifier = randomString();
-        Map<String, Object> eventParams = MapUtil.builder(new HashMap<String, Object>())
-                .put("identifier", eventIdentifier)
-                .put("value", ListUtil.of("a", "b", "c"))
-                .build();
-        IotDeviceMessage message = createEventPostMessage(eventParams);
-        IotSceneRuleDO.Trigger trigger = createValidTrigger(eventIdentifier);
-        trigger.setOperator(IotSceneRuleConditionOperatorEnum.EQUALS.getOperator());
-        trigger.setValue("[\"a\",\"b\",\"c\"]");
-
-        boolean result = matcher.matches(message, trigger);
-
-        assertTrue(result);
-    }
-
-    @Test
-    public void testMatches_structValueInvalidJsonComparator() {
-        // 比较值不是合法 JSON：结构体场景下视为不匹配，不抛异常
-        String eventIdentifier = randomString();
-        Map<String, Object> eventValue = MapUtil.builder(new HashMap<String, Object>())
-                .put("level", "high")
-                .build();
-        Map<String, Object> eventParams = MapUtil.builder(new HashMap<String, Object>())
-                .put("identifier", eventIdentifier)
-                .put("value", eventValue)
-                .build();
-        IotDeviceMessage message = createEventPostMessage(eventParams);
-        IotSceneRuleDO.Trigger trigger = createValidTrigger(eventIdentifier);
-        trigger.setOperator(IotSceneRuleConditionOperatorEnum.EQUALS.getOperator());
-        trigger.setValue("not a json");
-
-        boolean result = matcher.matches(message, trigger);
-
-        assertFalse(result);
-    }
-
     // ========== 辅助方法 ==========
     /**
@@ -4,7 +4,6 @@ import cn.hutool.core.map.MapUtil;
 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.rule.IotSceneRuleDO;
-import cn.iocoder.yudao.module.iot.enums.rule.IotSceneRuleConditionOperatorEnum;
 import cn.iocoder.yudao.module.iot.enums.rule.IotSceneRuleTriggerTypeEnum;
 import cn.iocoder.yudao.module.iot.service.rule.scene.matcher.IotBaseConditionMatcherTest;
 import org.junit.jupiter.api.BeforeEach;
@@ -486,7 +485,7 @@ public class IotDeviceServiceInvokeTriggerMatcherTest extends IotBaseConditionMa
         IotSceneRuleDO.Trigger trigger = new IotSceneRuleDO.Trigger();
         trigger.setType(IotSceneRuleTriggerTypeEnum.DEVICE_SERVICE_INVOKE.getType());
         trigger.setIdentifier(serviceIdentifier);
-        trigger.setOperator(IotSceneRuleConditionOperatorEnum.EQUALS.getOperator());
+        trigger.setOperator("=="); // 等于操作符
         trigger.setValue("auto");
         // 调用
@@ -10,9 +10,6 @@ import cn.iocoder.yudao.module.iot.service.rule.scene.matcher.IotBaseConditionMa
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import java.util.HashMap;
-import java.util.Map;
-
 import static cn.iocoder.yudao.framework.test.core.util.RandomUtils.randomLongId;
 import static org.junit.jupiter.api.Assertions.*;
@@ -251,9 +248,7 @@ public class IotDeviceStateUpdateTriggerMatcherTest extends IotBaseConditionMatc
         IotDeviceMessage message = new IotDeviceMessage();
         message.setDeviceId(randomLongId());
         message.setMethod(IotDeviceMessageMethodEnum.STATE_UPDATE.getMethod());
-        Map<String, Object> params = new HashMap<>();
-        params.put("state", state);
-        message.setParams(params);
+        message.setParams(state);
         return message;
     }
@@ -247,7 +247,7 @@ public class IotTimerTriggerMatcherTest extends IotBaseConditionMatcherTest {
     public void testMatches_rangeCronSuccess() {
         // 准备参数
         IotDeviceMessage message = createDeviceMessage();
-        IotSceneRuleDO.Trigger trigger = createValidTrigger("0 0 9-17 ? * MON-FRI"); // 工作日 9-17 点
+        IotSceneRuleDO.Trigger trigger = createValidTrigger("0 0 9-17 * * MON-FRI"); // 工作日9-17点
         // 调用
         boolean result = matcher.matches(message, trigger);