【修复】IoT 数据流转规则：属性上报场景下多规则命中同一数据目的会重复推送

YunaiV
1 parent 87c8d7e2
Showing 6 changed files with 386 additions and 12 deletions
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataRuleServiceImpl.java
yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataRuleServiceImplTest.java
yudao-module-iot/yudao-module-iot-biz/src/test/resources/sql/clean.sql
yudao-module-iot/yudao-module-iot-biz/src/test/resources/sql/create_tables.sql
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceMessageUtils.java
yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceMessageUtilsTest.java
@@ -10,6 +10,7 @@ import cn.iocoder.yudao.framework.common.util.object.ObjectUtils;
 import cn.iocoder.yudao.framework.common.util.spring.SpringUtils;
 import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.data.rule.IotDataRulePageReqVO;
 import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.data.rule.IotDataRuleSaveReqVO;
+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.util.IotDeviceMessageUtils;
 import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
@@ -212,34 +213,76 @@ public class IotDataRuleServiceImpl implements IotDataRuleService {
     @Override
     public void executeDataRule(IotDeviceMessage message) {
         try {
-            // 1. 获取匹配的数据流转规则
             Long deviceId = message.getDeviceId();
             String method = message.getMethod();
-            String identifier = IotDeviceMessageUtils.getIdentifier(message);
-            List<IotDataRuleDO> rules = getSelf().getDataRuleListByConditionFromCache(deviceId, method, identifier);
-            if (CollUtil.isEmpty(rules)) {
+            // 1. 匹配命中的规则
+            List<IotDataRuleDO> matchedRules;
+            Object identifierForLog;
+            if (IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod().equals(method)) {
+                // 属性上报：params 含多个属性 key，每个 key 都可能命中规则
+                Set<String> identifiers = IotDeviceMessageUtils.getPropertyIdentifiers(message);
+                matchedRules = matchPropertyPostDataRules(deviceId, method, identifiers);
+                identifierForLog = identifiers;
+            } else {
+                // 其他消息（事件 / 服务调用 / 状态）：单一 identifier
+                String identifier = IotDeviceMessageUtils.getIdentifier(message);
+                matchedRules = getSelf().getDataRuleListByConditionFromCache(deviceId, method, identifier);
+                identifierForLog = identifier;
+            }
+            if (CollUtil.isEmpty(matchedRules)) {
                 log.debug("[executeDataRule][设备({}) 方法({}) 标识符({}) 没有匹配的数据流转规则]",
-                        deviceId, method, identifier);
+                        deviceId, method, identifierForLog);
                 return;
             }
             log.info("[executeDataRule][设备({}) 方法({}) 标识符({}) 匹配到 {} 条数据流转规则]",
-                    deviceId, method, identifier, rules.size());
+                    deviceId, method, identifierForLog, matchedRules.size());
-            // 2. 遍历规则，执行数据流转
-            rules.forEach(rule -> executeDataRule(message, rule));
+            // 2. 跨规则去重 sink，避免多条规则命中同一数据目的时重复推送
+            Set<Long> processedSinkIds = new HashSet<>();
+            matchedRules.forEach(rule -> executeDataRule(message, rule, processedSinkIds));
         } catch (Exception e) {
             log.error("[executeDataRule][消息({}) 执行数据流转规则异常]", message, e);
         }
     }
     /**
+     * 匹配属性上报场景下命中的数据流转规则
+     *
+     * 同一规则可能同时匹配「任意属性」与具体 identifier，按 ruleId 去重后返回
+     *
+     * @param deviceId    设备编号
+     * @param method      消息方法
+     * @param identifiers 上报消息中包含的属性标识符集合
+     * @return 命中的数据流转规则列表
+     */
+    private List<IotDataRuleDO> matchPropertyPostDataRules(Long deviceId, String method, Set<String> identifiers) {
+        LinkedHashMap<Long, IotDataRuleDO> matchedRuleMap = new LinkedHashMap<>();
+        // 情况一：先匹配未填 identifier 的「任意属性」规则，默认就匹配
+        collectMatchedRules(matchedRuleMap, deviceId, method, null);
+        // 情况二：再针对每个上报的属性标识符匹配限定具体 identifier 的规则
+        identifiers.forEach(identifier -> collectMatchedRules(matchedRuleMap, deviceId, method, identifier));
+        return new ArrayList<>(matchedRuleMap.values());
+    }
+
+    private void collectMatchedRules(Map<Long, IotDataRuleDO> matchedRuleMap,
+                                     Long deviceId, String method, String identifier) {
+        getSelf().getDataRuleListByConditionFromCache(deviceId, method, identifier)
+                .forEach(rule -> matchedRuleMap.putIfAbsent(rule.getId(), rule));
+    }
+
+    /**
      * 为指定规则的所有数据目的执行数据流转
      *
-     * @param message 设备消息
-     * @param rule    数据流转规则
+     * @param message          设备消息
+     * @param rule             数据流转规则
+     * @param processedSinkIds 已处理的数据目的编号集合，跨规则去重，避免同一数据目的被重复推送
      */
-    private void executeDataRule(IotDeviceMessage message, IotDataRuleDO rule) {
+    private void executeDataRule(IotDeviceMessage message, IotDataRuleDO rule, Set<Long> processedSinkIds) {
         rule.getSinkIds().forEach(sinkId -> {
+            // 同一消息下，多条规则命中同一数据目的时只推送一次
+            if (!processedSinkIds.add(sinkId)) {
+                return;
+            }
             try {
                 // 获取数据目的配置
                 IotDataSinkDO dataSink = dataSinkService.getDataSinkFromCache(sinkId);
+package cn.iocoder.yudao.module.iot.service.rule.data;
+
+import cn.hutool.core.map.MapUtil;
+import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
+import cn.iocoder.yudao.framework.test.core.ut.BaseDbUnitTest;
+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.IotDataRuleDO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataSinkDO;
+import cn.iocoder.yudao.module.iot.dal.mysql.rule.IotDataRuleMapper;
+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.data.action.IotDataRuleAction;
+import cn.iocoder.yudao.module.iot.service.thingmodel.IotThingModelService;
+import jakarta.annotation.Resource;
+import org.junit.jupiter.api.Test;
+import org.springframework.context.annotation.Import;
+import org.springframework.test.context.bean.override.mockito.MockitoBean;
+
+import java.util.Map;
+
+import static cn.iocoder.yudao.framework.test.core.util.RandomUtils.randomLongId;
+import static cn.iocoder.yudao.framework.test.core.util.RandomUtils.randomPojo;
+import static java.util.Collections.singletonList;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * {@link IotDataRuleServiceImpl} 的单元测试类
+ *
+ * @author 芋道源码
+ */
+@Import(IotDataRuleServiceImpl.class)
+class IotDataRuleServiceImplTest extends BaseDbUnitTest {
+
+    @Resource
+    private IotDataRuleServiceImpl dataRuleService;
+
+    @Resource
+    private IotDataRuleMapper dataRuleMapper;
+
+    @MockitoBean
+    private IotDataSinkService dataSinkService;
+    @MockitoBean
+    private IotDataRuleAction dataRuleAction;
+    @MockitoBean
+    private IotProductService productService;
+    @MockitoBean
+    private IotDeviceService deviceService;
+    @MockitoBean
+    private IotThingModelService thingModelService;
+
+    @Test
+    public void testExecuteDataRule_propertyPost_singleIdentifierMatched() {
+        // 准备参数
+        Long deviceId = randomLongId();
+        String identifier = "temperature";
+        IotDeviceMessage message = createPropertyPostMessage(deviceId,
+                MapUtil.<String, Object>builder().put(identifier, 25.5).build());
+        // mock 数据：插入一条限定 identifier=temperature 的规则
+        Long sinkId = randomLongId();
+        insertEnabledRule(deviceId, IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod(), identifier, sinkId);
+        // mock 方法
+        IotDataSinkDO sink = mockEnabledSink(sinkId);
+
+        // 调用
+        dataRuleService.executeDataRule(message);
+
+        // 断言：sink action 被调用一次
+        verify(dataRuleAction).execute(eq(message), eq(sink));
+    }
+
+    @Test
+    public void testExecuteDataRule_propertyPost_multiIdentifierOneMatched() {
+        // 准备参数：上报 {temperature, humidity}，规则只限定 humidity
+        Long deviceId = randomLongId();
+        IotDeviceMessage message = createPropertyPostMessage(deviceId,
+                MapUtil.<String, Object>builder().put("temperature", 25.5).put("humidity", 60).build());
+        // mock 数据
+        Long sinkId = randomLongId();
+        insertEnabledRule(deviceId, IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod(), "humidity", sinkId);
+        // mock 方法
+        IotDataSinkDO sink = mockEnabledSink(sinkId);
+
+        // 调用
+        dataRuleService.executeDataRule(message);
+
+        // 断言
+        verify(dataRuleAction).execute(eq(message), eq(sink));
+    }
+
+    @Test
+    public void testExecuteDataRule_propertyPost_multiIdentifierDeduped() {
+        // 准备参数：上报 {temperature, humidity}，规则 identifier=null 不限定属性
+        Long deviceId = randomLongId();
+        IotDeviceMessage message = createPropertyPostMessage(deviceId,
+                MapUtil.<String, Object>builder().put("temperature", 25.5).put("humidity", 60).build());
+        // mock 数据：identifier=null 时两个属性 key 都会命中同一条规则，需在 sink 调用前去重
+        Long sinkId = randomLongId();
+        insertEnabledRule(deviceId, IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod(), null, sinkId);
+        // mock 方法
+        IotDataSinkDO sink = mockEnabledSink(sinkId);
+
+        // 调用
+        dataRuleService.executeDataRule(message);
+
+        // 断言：去重后只触发一次，而不是 2 次
+        verify(dataRuleAction).execute(eq(message), eq(sink));
+    }
+
+    @Test
+    public void testExecuteDataRule_propertyPost_multiRuleSameSinkDeduped() {
+        // 准备参数：上报 {temperature, humidity}，两条规则分别命中不同 identifier，但都指向同一 sink
+        Long deviceId = randomLongId();
+        IotDeviceMessage message = createPropertyPostMessage(deviceId,
+                MapUtil.<String, Object>builder().put("temperature", 25.5).put("humidity", 60).build());
+        // mock 数据：插入两条规则，identifier 分别为 temperature 与 humidity，sinkId 相同
+        Long sinkId = randomLongId();
+        insertEnabledRule(deviceId, IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod(), "temperature", sinkId);
+        insertEnabledRule(deviceId, IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod(), "humidity", sinkId);
+        // mock 方法
+        IotDataSinkDO sink = mockEnabledSink(sinkId);
+
+        // 调用
+        dataRuleService.executeDataRule(message);
+
+        // 断言：跨规则去重后，sink action 只触发一次，而不是 2 次
+        verify(dataRuleAction).execute(eq(message), eq(sink));
+    }
+
+    @Test
+    public void testExecuteDataRule_propertyPost_emptyParamsMatchesWildcardRule() {
+        // 准备参数：上报空属性，规则 identifier=null 不限定属性，按"任意 property report 都同步"语义应命中
+        Long deviceId = randomLongId();
+        IotDeviceMessage message = createPropertyPostMessage(deviceId,
+                MapUtil.<String, Object>builder().build());
+        // mock 数据
+        Long sinkId = randomLongId();
+        insertEnabledRule(deviceId, IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod(), null, sinkId);
+        // mock 方法
+        IotDataSinkDO sink = mockEnabledSink(sinkId);
+
+        // 调用
+        dataRuleService.executeDataRule(message);
+
+        // 断言
+        verify(dataRuleAction).execute(eq(message), eq(sink));
+    }
+
+    @Test
+    public void testExecuteDataRule_propertyPost_noIdentifierMatched() {
+        // 准备参数：上报 {temperature}，规则限定 humidity
+        Long deviceId = randomLongId();
+        IotDeviceMessage message = createPropertyPostMessage(deviceId,
+                MapUtil.<String, Object>builder().put("temperature", 25.5).build());
+        // mock 数据
+        Long sinkId = randomLongId();
+        insertEnabledRule(deviceId, IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod(), "humidity", sinkId);
+
+        // 调用
+        dataRuleService.executeDataRule(message);
+
+        // 断言：sink action 不应被调用
+        verify(dataRuleAction, never()).execute(any(), any());
+    }
+
+    @Test
+    public void testExecuteDataRule_eventPost_singleIdentifierMatched() {
+        // 准备参数：事件触发器走单 identifier 路径（与改动前行为保持一致）
+        Long deviceId = randomLongId();
+        String identifier = "alarm";
+        IotDeviceMessage message = randomPojo(IotDeviceMessage.class, o -> {
+            o.setDeviceId(deviceId);
+            o.setMethod(IotDeviceMessageMethodEnum.EVENT_POST.getMethod());
+            o.setParams(MapUtil.<String, Object>builder()
+                    .put("identifier", identifier).put("value", "fired").build());
+        });
+        // mock 数据
+        Long sinkId = randomLongId();
+        insertEnabledRule(deviceId, IotDeviceMessageMethodEnum.EVENT_POST.getMethod(), identifier, sinkId);
+        // mock 方法
+        IotDataSinkDO sink = mockEnabledSink(sinkId);
+
+        // 调用
+        dataRuleService.executeDataRule(message);
+
+        // 断言
+        verify(dataRuleAction).execute(eq(message), eq(sink));
+    }
+
+    // ========== 辅助方法 ==========
+
+    private IotDeviceMessage createPropertyPostMessage(Long deviceId, Map<String, Object> params) {
+        return randomPojo(IotDeviceMessage.class, o -> {
+            o.setDeviceId(deviceId);
+            o.setMethod(IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod());
+            o.setParams(params);
+        });
+    }
+
+    /**
+     * 向 H2 中插入一条启用状态的数据流转规则，命中后会路由到 {@code sinkId}
+     */
+    private void insertEnabledRule(Long deviceId, String method, String identifier, Long sinkId) {
+        IotDataRuleDO.SourceConfig config = randomPojo(IotDataRuleDO.SourceConfig.class, o -> {
+            o.setDeviceId(deviceId);
+            o.setMethod(method);
+            o.setIdentifier(identifier);
+        });
+        IotDataRuleDO rule = randomPojo(IotDataRuleDO.class, o -> {
+            o.setId(null);
+            o.setStatus(CommonStatusEnum.ENABLE.getStatus());
+            o.setSourceConfigs(singletonList(config));
+            o.setSinkIds(singletonList(sinkId));
+        });
+        dataRuleMapper.insert(rule);
+    }
+
+    /**
+     * 构造一个启用状态的数据流转目的并塞入对应 mock；返回 sink 用于断言
+     */
+    private IotDataSinkDO mockEnabledSink(Long sinkId) {
+        IotDataSinkDO sink = randomPojo(IotDataSinkDO.class, o -> {
+            o.setId(sinkId);
+            o.setStatus(CommonStatusEnum.ENABLE.getStatus());
+        });
+        when(dataSinkService.getDataSinkFromCache(sinkId)).thenReturn(sink);
+        when(dataRuleAction.getType()).thenReturn(sink.getType());
+        return sink;
+    }
+
+}
@@ -8,3 +8,4 @@ DELETE FROM &quot;iot_alert_record&quot;;
 DELETE FROM "iot_ota_firmware";
 DELETE FROM "iot_ota_task";
 DELETE FROM "iot_ota_record";
+DELETE FROM "iot_data_rule";
@@ -180,3 +180,19 @@ CREATE TABLE IF NOT EXISTS &quot;iot_ota_record&quot; (
     "tenant_id" bigint NOT NULL DEFAULT '0',
     PRIMARY KEY ("id")
 ) COMMENT 'IoT OTA 升级记录表';
+
+CREATE TABLE IF NOT EXISTS "iot_data_rule" (
+    "id" bigint NOT NULL GENERATED BY DEFAULT AS IDENTITY,
+    "name" varchar(128) NOT NULL,
+    "description" varchar(256) DEFAULT '',
+    "status" int NOT NULL,
+    "source_configs" varchar(10000) NOT NULL,
+    "sink_ids" varchar(512) NOT NULL,
+    "creator" varchar(64) DEFAULT '',
+    "create_time" timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updater" varchar(64) DEFAULT '',
+    "update_time" timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "deleted" bit NOT NULL DEFAULT FALSE,
+    "tenant_id" bigint NOT NULL DEFAULT '0',
+    PRIMARY KEY ("id")
+) COMMENT 'IoT 数据流转规则';
 package cn.iocoder.yudao.module.iot.core.util;
 import cn.hutool.core.lang.Assert;
-import cn.hutool.core.map.MapUtil;
 import cn.hutool.core.util.IdUtil;
 import cn.hutool.core.util.ReflectUtil;
 import cn.hutool.core.util.StrUtil;
@@ -10,7 +9,9 @@ import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
 import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
+import java.util.LinkedHashSet;
 import java.util.Map;
+import java.util.Set;
 /**
  * IoT 设备【消息】的工具类
@@ -90,6 +91,25 @@ public class IotDeviceMessageUtils {
     }
     /**
+     * 获取属性上报消息中包含的所有属性标识符
+     *
+     * 仅支持扁平结构：{ temperature: 25.5, humidity: 60 }，顶层 key 即属性标识符
+     *
+     * @param message 设备消息
+     * @return 属性标识符集合，不为 null
+     */
+    public static Set<String> getPropertyIdentifiers(IotDeviceMessage message) {
+        if (message == null) {
+            return new LinkedHashSet<>();
+        }
+        Map<String, Object> params = parseParamsToMap(message.getParams());
+        if (params == null) {
+            return new LinkedHashSet<>();
+        }
+        return new LinkedHashSet<>(params.keySet());
+    }
+
+    /**
      * 判断消息中是否包含指定的标识符
      * <p>
      * 对于不同消息类型的处理：
@@ -7,6 +7,7 @@ import org.junit.jupiter.api.Test;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Set;
 import static org.junit.jupiter.api.Assertions.*;
@@ -289,4 +290,62 @@ public class IotDeviceMessageUtilsTest {
         assertTrue(notContainsResult);
         assertEquals(!containsResult, notContainsResult);
     }
+
+    // ========== getPropertyIdentifiers 测试 ==========
+
+    @Test
+    public void testGetPropertyIdentifiers_flatStructure() {
+        // 扁平结构：顶层 key 即标识符
+        IotDeviceMessage message = new IotDeviceMessage();
+        Map<String, Object> params = new HashMap<>();
+        params.put("temperature", 25.5);
+        params.put("humidity", 60);
+        message.setParams(params);
+
+        Set<String> identifiers = IotDeviceMessageUtils.getPropertyIdentifiers(message);
+        assertEquals(2, identifiers.size());
+        assertTrue(identifiers.contains("temperature"));
+        assertTrue(identifiers.contains("humidity"));
+    }
+
+    @Test
+    public void testGetPropertyIdentifiers_nullMessage() {
+        // 入参为 null：返回空集合
+        Set<String> identifiers = IotDeviceMessageUtils.getPropertyIdentifiers(null);
+        assertNotNull(identifiers);
+        assertTrue(identifiers.isEmpty());
+    }
+
+    @Test
+    public void testGetPropertyIdentifiers_nullParams() {
+        // params 为 null：返回空集合
+        IotDeviceMessage message = new IotDeviceMessage();
+        message.setParams(null);
+
+        Set<String> identifiers = IotDeviceMessageUtils.getPropertyIdentifiers(message);
+        assertTrue(identifiers.isEmpty());
+    }
+
+    @Test
+    public void testGetPropertyIdentifiers_emptyParams() {
+        // params 为空 Map：返回空集合
+        IotDeviceMessage message = new IotDeviceMessage();
+        message.setParams(new HashMap<>());
+
+        Set<String> identifiers = IotDeviceMessageUtils.getPropertyIdentifiers(message);
+        assertTrue(identifiers.isEmpty());
+    }
+
+    @Test
+    public void testGetPropertyIdentifiers_jsonStringParams() {
+        // params 为 JSON 字符串：parseParamsToMap 解析后正常提取顶层标识符
+        IotDeviceMessage message = new IotDeviceMessage();
+        message.setParams("{\"temperature\":25.5,\"humidity\":60}");
+
+        Set<String> identifiers = IotDeviceMessageUtils.getPropertyIdentifiers(message);
+        assertEquals(2, identifiers.size());
+        assertTrue(identifiers.contains("temperature"));
+        assertTrue(identifiers.contains("humidity"));
+    }
+
 }