diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataRuleServiceImpl.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataRuleServiceImpl.java index ed52067..ccde54e 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataRuleServiceImpl.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataRuleServiceImpl.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 rules = getSelf().getDataRuleListByConditionFromCache(deviceId, method, identifier); - if (CollUtil.isEmpty(rules)) { + // 1. 匹配命中的规则 + List matchedRules; + Object identifierForLog; + if (IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod().equals(method)) { + // 属性上报：params 含多个属性 key，每个 key 都可能命中规则 + Set 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 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 matchPropertyPostDataRules(Long deviceId, String method, Set identifiers) { + LinkedHashMap 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 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 processedSinkIds) { rule.getSinkIds().forEach(sinkId -> { + // 同一消息下，多条规则命中同一数据目的时只推送一次 + if (!processedSinkIds.add(sinkId)) { + return; + } try { // 获取数据目的配置 IotDataSinkDO dataSink = dataSinkService.getDataSinkFromCache(sinkId); diff --git a/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataRuleServiceImplTest.java b/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataRuleServiceImplTest.java new file mode 100644 index 0000000..3bdc8c3 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataRuleServiceImplTest.java @@ -0,0 +1,235 @@ +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.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.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.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.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.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.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.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 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; + } + +} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/test/resources/sql/clean.sql b/yudao-module-iot/yudao-module-iot-biz/src/test/resources/sql/clean.sql index ae1c5e5..454a57f 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/test/resources/sql/clean.sql +++ b/yudao-module-iot/yudao-module-iot-biz/src/test/resources/sql/clean.sql @@ -8,3 +8,4 @@ DELETE FROM "iot_alert_record"; DELETE FROM "iot_ota_firmware"; DELETE FROM "iot_ota_task"; DELETE FROM "iot_ota_record"; +DELETE FROM "iot_data_rule"; diff --git a/yudao-module-iot/yudao-module-iot-biz/src/test/resources/sql/create_tables.sql b/yudao-module-iot/yudao-module-iot-biz/src/test/resources/sql/create_tables.sql index 306c66b..0bd2ad2 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/test/resources/sql/create_tables.sql +++ b/yudao-module-iot/yudao-module-iot-biz/src/test/resources/sql/create_tables.sql @@ -180,3 +180,19 @@ CREATE TABLE IF NOT EXISTS "iot_ota_record" ( "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 数据流转规则'; diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceMessageUtils.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceMessageUtils.java index 1c9f8d7..c49c4a0 100644 --- a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceMessageUtils.java +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceMessageUtils.java @@ -1,7 +1,6 @@ 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 getPropertyIdentifiers(IotDeviceMessage message) { + if (message == null) { + return new LinkedHashSet<>(); + } + Map params = parseParamsToMap(message.getParams()); + if (params == null) { + return new LinkedHashSet<>(); + } + return new LinkedHashSet<>(params.keySet()); + } + + /** * 判断消息中是否包含指定的标识符 *

* 对于不同消息类型的处理： diff --git a/yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceMessageUtilsTest.java b/yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceMessageUtilsTest.java index 4f7d614..a0c4650 100644 --- a/yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceMessageUtilsTest.java +++ b/yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceMessageUtilsTest.java @@ -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 params = new HashMap<>(); + params.put("temperature", 25.5); + params.put("humidity", 60); + message.setParams(params); + + Set identifiers = IotDeviceMessageUtils.getPropertyIdentifiers(message); + assertEquals(2, identifiers.size()); + assertTrue(identifiers.contains("temperature")); + assertTrue(identifiers.contains("humidity")); + } + + @Test + public void testGetPropertyIdentifiers_nullMessage() { + // 入参为 null：返回空集合 + Set identifiers = IotDeviceMessageUtils.getPropertyIdentifiers(null); + assertNotNull(identifiers); + assertTrue(identifiers.isEmpty()); + } + + @Test + public void testGetPropertyIdentifiers_nullParams() { + // params 为 null：返回空集合 + IotDeviceMessage message = new IotDeviceMessage(); + message.setParams(null); + + Set identifiers = IotDeviceMessageUtils.getPropertyIdentifiers(message); + assertTrue(identifiers.isEmpty()); + } + + @Test + public void testGetPropertyIdentifiers_emptyParams() { + // params 为空 Map：返回空集合 + IotDeviceMessage message = new IotDeviceMessage(); + message.setParams(new HashMap<>()); + + Set 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 identifiers = IotDeviceMessageUtils.getPropertyIdentifiers(message); + assertEquals(2, identifiers.size()); + assertTrue(identifiers.contains("temperature")); + assertTrue(identifiers.contains("humidity")); + } + }