From ea1f0cb462520cc0a3bd1633e02672a6c0ebac97 Mon Sep 17 00:00:00 2001 From: YunaiV Date: Thu, 26 Jun 2025 09:58:34 +0800 Subject: [PATCH] feat：【IoT 物联网】实现“数据流转”功能的执行（80%） --- yudao-module-iot/yudao-module-iot-api/src/main/java/cn/iocoder/yudao/module/iot/enums/device/IotDeviceMessageIdentifierEnum.java | 1 + yudao-module-iot/yudao-module-iot-api/src/main/java/cn/iocoder/yudao/module/iot/enums/rule/IotRuleSceneActionTypeEnum.java | 4 +--- yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotDataRuleMapper.java | 4 ++++ yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/RedisKeyConstants.java | 16 ++++++++++++++++ yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mq/consumer/rule/IotDataRuleMessageHandler.java | 50 ++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataRuleService.java | 14 +++++++++++--- yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataRuleServiceImpl.java | 120 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++---- yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataSinkService.java | 8 ++++++++ yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataSinkServiceImpl.java | 12 ++++++++++-- yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotRuleSceneDataBridgeAction.java | 61 ------------------------------------------------------------- yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/AbstractCacheableDataBridgeExecute.java | 114 ------------------------------------------------------------------------------------------------------------------ yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDataBridgeExecute.java | 45 --------------------------------------------- yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDataRuleAction.java | 28 ++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDataRuleCacheableAction.java | 117 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotHttpDataBridgeExecute.java | 88 ---------------------------------------------------------------------------------------- yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotHttpDataSinkAction.java | 91 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotKafkaDataRuleAction.java | 78 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotKafkaMQDataBridgeExecute.java | 78 ------------------------------------------------------------------------------ yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRabbitMQDataBridgeExecute.java | 77 ----------------------------------------------------------------------------- yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRabbitMQDataRuleAction.java | 73 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRedisStreamDataBridgeExecute.java | 80 -------------------------------------------------------------------------------- yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRedisStreamRuleAction.java | 80 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRocketMQDataBridgeExecute.java | 65 ----------------------------------------------------------------- yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRocketMQDataRuleAction.java | 65 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotRuleSceneServiceImpl.java | 5 ----- yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotDataBridgeExecuteTest.java | 14 +++++++------- 26 files changed, 756 insertions(+), 632 deletions(-) create mode 100644 yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mq/consumer/rule/IotDataRuleMessageHandler.java delete mode 100644 yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotRuleSceneDataBridgeAction.java delete mode 100644 yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/AbstractCacheableDataBridgeExecute.java delete mode 100644 yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDataBridgeExecute.java create mode 100644 yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDataRuleAction.java create mode 100644 yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDataRuleCacheableAction.java delete mode 100644 yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotHttpDataBridgeExecute.java create mode 100644 yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotHttpDataSinkAction.java create mode 100644 yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotKafkaDataRuleAction.java delete mode 100644 yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotKafkaMQDataBridgeExecute.java delete mode 100644 yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRabbitMQDataBridgeExecute.java create mode 100644 yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRabbitMQDataRuleAction.java delete mode 100644 yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRedisStreamDataBridgeExecute.java create mode 100644 yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRedisStreamRuleAction.java delete mode 100644 yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRocketMQDataBridgeExecute.java create mode 100644 yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRocketMQDataRuleAction.java diff --git a/yudao-module-iot/yudao-module-iot-api/src/main/java/cn/iocoder/yudao/module/iot/enums/device/IotDeviceMessageIdentifierEnum.java b/yudao-module-iot/yudao-module-iot-api/src/main/java/cn/iocoder/yudao/module/iot/enums/device/IotDeviceMessageIdentifierEnum.java index a06b43c..e9dbe2f 100644 --- a/yudao-module-iot/yudao-module-iot-api/src/main/java/cn/iocoder/yudao/module/iot/enums/device/IotDeviceMessageIdentifierEnum.java +++ b/yudao-module-iot/yudao-module-iot-api/src/main/java/cn/iocoder/yudao/module/iot/enums/device/IotDeviceMessageIdentifierEnum.java @@ -7,6 +7,7 @@ import lombok.RequiredArgsConstructor; /** * IoT 设备消息标识符枚举 */ +@Deprecated @Getter @RequiredArgsConstructor public enum IotDeviceMessageIdentifierEnum { diff --git a/yudao-module-iot/yudao-module-iot-api/src/main/java/cn/iocoder/yudao/module/iot/enums/rule/IotRuleSceneActionTypeEnum.java b/yudao-module-iot/yudao-module-iot-api/src/main/java/cn/iocoder/yudao/module/iot/enums/rule/IotRuleSceneActionTypeEnum.java index 6e6843b..5251852 100644 --- a/yudao-module-iot/yudao-module-iot-api/src/main/java/cn/iocoder/yudao/module/iot/enums/rule/IotRuleSceneActionTypeEnum.java +++ b/yudao-module-iot/yudao-module-iot-api/src/main/java/cn/iocoder/yudao/module/iot/enums/rule/IotRuleSceneActionTypeEnum.java @@ -40,9 +40,7 @@ public enum IotRuleSceneActionTypeEnum implements ArrayValuable { @Deprecated ALERT(2), // 告警执行 - - @Deprecated - DATA_BRIDGE(3); // 桥接执行 + ; private final Integer type; diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotDataRuleMapper.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotDataRuleMapper.java index d590232..7c0c17d 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotDataRuleMapper.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotDataRuleMapper.java @@ -31,4 +31,8 @@ public interface IotDataRuleMapper extends BaseMapperX { .apply(MyBatisUtils.findInSet("sink_ids", sinkId))); } + default List selectListByStatus(Integer status) { + return selectList(IotDataRuleDO::getStatus, status); + } + } \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/RedisKeyConstants.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/RedisKeyConstants.java index 5c4b742..1187677 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/RedisKeyConstants.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/RedisKeyConstants.java @@ -60,4 +60,20 @@ public interface RedisKeyConstants { */ String THING_MODEL_LIST = "iot:thing_model_list"; + /** + * 数据流转规则的数据缓存，使用 Spring Cache 操作 + * + * KEY 格式：data_rule_list_${deviceId}_${method}_${identifier} + * VALUE 数据类型：String 数组(JSON)，即 {@link cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataRuleDO} 列表 + */ + String DATA_RULE_LIST = "iot:data_rule_list"; + + /** + * 数据目的的数据缓存，使用 Spring Cache 操作 + * + * KEY 格式：data_sink_${id} + * VALUE 数据类型：String(JSON)，即 {@link cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataSinkDO} + */ + String DATA_SINK = "iot:data_sink"; + } diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mq/consumer/rule/IotDataRuleMessageHandler.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mq/consumer/rule/IotDataRuleMessageHandler.java new file mode 100644 index 0000000..c2b8226 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mq/consumer/rule/IotDataRuleMessageHandler.java @@ -0,0 +1,50 @@ +package cn.iocoder.yudao.module.iot.mq.consumer.rule; + +import cn.iocoder.yudao.framework.tenant.core.util.TenantUtils; +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.rule.data.IotDataRuleService; +import jakarta.annotation.PostConstruct; +import jakarta.annotation.Resource; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; + +// TODO @puhui999：后面重构哈 + +/** + * 针对 {@link IotDeviceMessage} 的消费者，处理数据流转 + * + * @author 芋道源码 + */ +@Component +@Slf4j +public class IotDataRuleMessageHandler implements IotMessageSubscriber { + + @Resource + private IotDataRuleService dataRuleService; + + @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_data_rule_consumer"; + } + + @Override + public void onMessage(IotDeviceMessage message) { + TenantUtils.execute(message.getTenantId(), () -> dataRuleService.executeDataRule(message)); + } + +} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataRuleService.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataRuleService.java index 42fdf30..1e0a813 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataRuleService.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataRuleService.java @@ -1,13 +1,14 @@ package cn.iocoder.yudao.module.iot.service.rule.data; -import java.util.List; - import cn.iocoder.yudao.framework.common.pojo.PageResult; 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.mq.message.IotDeviceMessage; import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataRuleDO; import jakarta.validation.Valid; +import java.util.List; + /** * IoT 数据流转规则 Service 接口 * @@ -59,6 +60,13 @@ public interface IotDataRuleService { * @param sinkId 数据目的编号 * @return 是否被使用 */ - List getDataRuleBySinkId(Long sinkId); + List getDataRuleListBySinkId(Long sinkId); + + /** + * 执行数据流转规则 + * + * @param message 消息 + */ + void executeDataRule(IotDeviceMessage message); } \ No newline at end of file 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 65c7a39..d7370c0 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 @@ -3,17 +3,28 @@ package cn.iocoder.yudao.module.iot.service.rule.data; import cn.hutool.core.collection.CollUtil; import cn.hutool.core.util.ObjUtil; import cn.hutool.core.util.StrUtil; +import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum; import cn.iocoder.yudao.framework.common.pojo.PageResult; import cn.iocoder.yudao.framework.common.util.object.BeanUtils; +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.mq.message.IotDeviceMessage; +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.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.dal.redis.RedisKeyConstants; 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 lombok.extern.slf4j.Slf4j; +import org.springframework.cache.annotation.CacheEvict; +import org.springframework.cache.annotation.Cacheable; import org.springframework.stereotype.Service; import org.springframework.validation.annotation.Validated; @@ -30,6 +41,7 @@ import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.DATA_RULE_NOT */ @Service @Validated +@Slf4j public class IotDataRuleServiceImpl implements IotDataRuleService { @Resource @@ -44,7 +56,11 @@ public class IotDataRuleServiceImpl implements IotDataRuleService { @Resource private IotDataSinkService dataSinkService; + @Resource + private List dataRuleActions; + @Override + @CacheEvict(value = RedisKeyConstants.DATA_RULE_LIST, allEntries = true) public Long createDataRule(IotDataRuleSaveReqVO createReqVO) { // 校验数据源配置和数据目的 validateDataRuleConfig(createReqVO); @@ -55,6 +71,7 @@ public class IotDataRuleServiceImpl implements IotDataRuleService { } @Override + @CacheEvict(value = RedisKeyConstants.DATA_RULE_LIST, allEntries = true) public void updateDataRule(IotDataRuleSaveReqVO updateReqVO) { // 校验存在 validateDataRuleExists(updateReqVO.getId()); @@ -67,6 +84,7 @@ public class IotDataRuleServiceImpl implements IotDataRuleService { } @Override + @CacheEvict(value = RedisKeyConstants.DATA_RULE_LIST, allEntries = true) public void deleteDataRule(Long id) { // 校验存在 validateDataRuleExists(id); @@ -116,15 +134,15 @@ public class IotDataRuleServiceImpl implements IotDataRuleService { * @param sourceConfigs 数据源配置列表 */ private void validateThingModelsExist(List sourceConfigs) { - Map> productIdToIdentifiers = new HashMap<>(); + Map> productIdIdentifiers = new HashMap<>(); for (IotDataRuleDO.SourceConfig config : sourceConfigs) { if (StrUtil.isEmpty(config.getIdentifier())) { continue; } - productIdToIdentifiers.computeIfAbsent(config.getProductId(), + productIdIdentifiers.computeIfAbsent(config.getProductId(), productId -> new HashSet<>()).add(config.getIdentifier()); } - for (Map.Entry> entry : productIdToIdentifiers.entrySet()) { + for (Map.Entry> entry : productIdIdentifiers.entrySet()) { thingModelService.validateThingModelsExist(entry.getKey(), entry.getValue()); } } @@ -140,8 +158,102 @@ public class IotDataRuleServiceImpl implements IotDataRuleService { } @Override - public List getDataRuleBySinkId(Long sinkId) { + public List getDataRuleListBySinkId(Long sinkId) { return dataRuleMapper.selectListBySinkId(sinkId); } + @Cacheable(value = RedisKeyConstants.DATA_RULE_LIST, + key = "#deviceId + '_' + #method + '_' + (#identifier ?: '')") + public List getDataRuleListByConditionFromCache(Long deviceId, String method, String identifier) { + // 1. 查询所有开启的数据流转规则 + List rules = dataRuleMapper.selectListByStatus(CommonStatusEnum.ENABLE.getStatus()); + // 2. 内存里过滤匹配的规则 + List matchedRules = new ArrayList<>(); + for (IotDataRuleDO rule : rules) { + IotDataRuleDO.SourceConfig found = CollUtil.findOne(rule.getSourceConfigs(), + config -> ObjectUtils.equalsAny(config.getDeviceId(), deviceId, IotDeviceDO.DEVICE_ID_ALL) + && (StrUtil.isNotEmpty(config.getMethod()) || ObjUtil.equal(config.getMethod(), method)) + && (StrUtil.isEmpty(config.getIdentifier()) || ObjUtil.equal(config.getIdentifier(), identifier))); + if (found != null) { + matchedRules.add(new IotDataRuleDO().setId(rule.getId()).setSinkIds(rule.getSinkIds())); + } + } + return matchedRules; + } + + @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)) { + log.debug("[executeDataRule][设备({}) 方法({}) 标识符({}) 没有匹配的数据流转规则]", + deviceId, method, identifier); + return; + } + log.info("[executeDataRule][设备({}) 方法({}) 标识符({}) 匹配到 {} 条数据流转规则]", + deviceId, method, identifier, rules.size()); + + // 2. 遍历规则，执行数据流转 + rules.forEach(rule -> executeDataRule(message, rule)); + } catch (Exception e) { + log.error("[executeDataRule][消息({}) 执行数据流转规则异常]", message, e); + } + } + + /** + * 为指定规则的所有数据目的执行数据流转 + * + * @param message 设备消息 + * @param rule 数据流转规则 + */ + private void executeDataRule(IotDeviceMessage message, IotDataRuleDO rule) { + rule.getSinkIds().forEach(sinkId -> { + try { + // 获取数据目的配置 + IotDataSinkDO dataSink = dataSinkService.getDataSinkFromCache(sinkId); + if (dataSink == null) { + log.error("[executeDataRule][规则({}) 对应的数据目的({}) 不存在]", rule.getId(), sinkId); + return; + } + if (CommonStatusEnum.isDisable(dataSink.getStatus())) { + log.info("[executeDataRule][规则({}) 对应的数据目的({}) 状态为禁用]", rule.getId(), sinkId); + return; + } + + // 执行数据桥接操作 + executeDataRuleAction(message, dataSink); + } catch (Exception e) { + log.error("[executeDataRule][规则({}) 数据目的({}) 执行异常]", rule.getId(), sinkId, e); + } + }); + } + + /** + * 执行数据流转操作 + * + * @param message 设备消息 + * @param dataSink 数据目的 + */ + private void executeDataRuleAction(IotDeviceMessage message, IotDataSinkDO dataSink) { + dataRuleActions.forEach(action -> { + if (ObjUtil.notEqual(action.getType(), dataSink.getType())) { + return; + } + try { + action.execute(message, dataSink); + log.info("[executeDataRuleAction][消息({}) 数据目的({}) 执行成功]", message.getId(), dataSink.getId()); + } catch (Exception e) { + log.error("[executeDataRuleAction][消息({}) 数据目的({}) 执行异常]", message.getId(), dataSink.getId(), e); + } + }); + } + + private IotDataRuleServiceImpl getSelf() { + return SpringUtils.getBean(IotDataRuleServiceImpl.class); + } + } \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataSinkService.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataSinkService.java index 307163a..d0e2a52 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataSinkService.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataSinkService.java @@ -47,6 +47,14 @@ public interface IotDataSinkService { IotDataSinkDO getDataSink(Long id); /** + * 从缓存中获得数据流转目的 + * + * @param id 编号 + * @return 数据流转目的 + */ + IotDataSinkDO getDataSinkFromCache(Long id); + + /** * 获得数据流转目的分页 * * @param pageReqVO 分页查询 diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataSinkServiceImpl.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataSinkServiceImpl.java index 2b964c9..9977afb 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataSinkServiceImpl.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataSinkServiceImpl.java @@ -7,7 +7,9 @@ import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.data.sink.IotDataSin import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.data.sink.IotDataSinkSaveReqVO; import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataSinkDO; import cn.iocoder.yudao.module.iot.dal.mysql.rule.IotDataSinkMapper; +import cn.iocoder.yudao.module.iot.dal.redis.RedisKeyConstants; import jakarta.annotation.Resource; +import org.springframework.cache.annotation.Cacheable; import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Service; import org.springframework.validation.annotation.Validated; @@ -16,8 +18,8 @@ import java.util.Collection; import java.util.List; import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception; -import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.DATA_SINK_NOT_EXISTS; import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.DATA_SINK_DELETE_FAIL_USED_BY_RULE; +import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.DATA_SINK_NOT_EXISTS; /** * IoT 数据流转目的 Service 实现类 @@ -56,7 +58,7 @@ public class IotDataSinkServiceImpl implements IotDataSinkService { // 校验存在 validateDataBridgeExists(id); // 校验是否被数据流转规则使用 - if (CollUtil.isNotEmpty(dataRuleService.getDataRuleBySinkId(id))) { + if (CollUtil.isNotEmpty(dataRuleService.getDataRuleListBySinkId(id))) { throw exception(DATA_SINK_DELETE_FAIL_USED_BY_RULE); } // 删除 @@ -75,6 +77,12 @@ public class IotDataSinkServiceImpl implements IotDataSinkService { } @Override + @Cacheable(value = RedisKeyConstants.DATA_SINK, key = "#id") + public IotDataSinkDO getDataSinkFromCache(Long id) { + return dataSinkMapper.selectById(id); + } + + @Override public PageResult getDataSinkPage(IotDataSinkPageReqVO pageReqVO) { return dataSinkMapper.selectPage(pageReqVO); } diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotRuleSceneDataBridgeAction.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotRuleSceneDataBridgeAction.java deleted file mode 100644 index 08b76be..0000000 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotRuleSceneDataBridgeAction.java +++ /dev/null @@ -1,61 +0,0 @@ -package cn.iocoder.yudao.module.iot.service.rule.data; - -import cn.hutool.core.lang.Assert; -import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum; -import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; -import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataSinkDO; -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.rule.data.action.IotDataBridgeExecute; -import cn.iocoder.yudao.module.iot.service.rule.scene.action.IotRuleSceneAction; -import jakarta.annotation.Resource; -import lombok.extern.slf4j.Slf4j; -import org.springframework.stereotype.Component; - -import java.util.List; - -/** - * IoT 数据流转目的的 {@link IotRuleSceneAction} 实现类 - * - * @author 芋道源码 - */ -@Deprecated -@Component -@Slf4j -public class IotRuleSceneDataBridgeAction implements IotRuleSceneAction { - - @Resource - private IotDataSinkService dataBridgeService; - @Resource - private List> dataBridgeExecutes; - - @Override - public void execute(IotDeviceMessage message, IotRuleSceneDO.ActionConfig config) throws Exception { - // 1.1 如果消息为空，直接返回 - if (message == null) { - return; - } - // 1.2 获得数据流转目的 - Assert.notNull(config.getDataBridgeId(), "数据流转目的编号不能为空"); - IotDataSinkDO dataBridge = dataBridgeService.getDataSink(config.getDataBridgeId()); - if (dataBridge == null || dataBridge.getConfig() == null) { - log.error("[execute][message({}) config({}) 对应的数据流转目的不存在]", message, config); - return; - } - if (CommonStatusEnum.isDisable(dataBridge.getStatus())) { - log.info("[execute][message({}) config({}) 对应的数据流转目的({}) 状态为禁用]", message, config, dataBridge); - return; - } - - // 2. 执行数据桥接操作 - for (IotDataBridgeExecute execute : dataBridgeExecutes) { - execute.execute(message, dataBridge); - } - } - - @Override - public IotRuleSceneActionTypeEnum getType() { - return IotRuleSceneActionTypeEnum.DATA_BRIDGE; - } - -} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/AbstractCacheableDataBridgeExecute.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/AbstractCacheableDataBridgeExecute.java deleted file mode 100644 index 584285e..0000000 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/AbstractCacheableDataBridgeExecute.java +++ /dev/null @@ -1,114 +0,0 @@ -package cn.iocoder.yudao.module.iot.service.rule.data.action; - -import cn.hutool.core.util.ObjUtil; -import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; -import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataSinkDO; -import com.google.common.cache.CacheBuilder; -import com.google.common.cache.CacheLoader; -import com.google.common.cache.LoadingCache; -import com.google.common.cache.RemovalListener; -import lombok.extern.slf4j.Slf4j; - -import java.time.Duration; - -// TODO @芋艿：数据库 -// TODO @芋艿：mqtt -// TODO @芋艿：tcp -// TODO @芋艿：websocket - -/** - * 带缓存功能的数据流转目的执行器抽象类 - * - * 该类提供了一个通用的缓存机制，用于管理各类数据桥接的生产者(Producer)实例。 - * - * 主要特点: - * - 基于Guava Cache实现高效的生产者实例缓存管理 - * - 自动处理生产者的生命周期（创建、获取、关闭） - * - 支持30分钟未访问自动过期清理机制 - * - 异常处理与日志记录，便于问题排查 - * - * 子类需要实现: - * - initProducer(Config) - 初始化特定类型的生产者实例 - * - closeProducer(Producer) - 关闭生产者实例并释放资源 - * - * @param 配置信息类型，用于初始化生产者 - * @param 生产者类型，负责将数据发送到目标系统 - * @author HUIHUI - */ -@Slf4j -public abstract class AbstractCacheableDataBridgeExecute implements IotDataBridgeExecute { - - /** - * Producer 缓存 - */ - private final LoadingCache PRODUCER_CACHE = CacheBuilder.newBuilder() - .expireAfterAccess(Duration.ofMinutes(30)) // 30 分钟未访问就提前过期 - .removalListener((RemovalListener) notification -> { - Producer producer = notification.getValue(); - if (producer == null) { - return; - } - - try { - closeProducer(producer); - log.info("[PRODUCER_CACHE][配置({}) 对应的 producer 已关闭]", notification.getKey()); - } catch (Exception e) { - log.error("[PRODUCER_CACHE][配置({}) 对应的 producer 关闭失败]", notification.getKey(), e); - } - }) - .build(new CacheLoader() { - - @Override - public Producer load(Config config) throws Exception { - try { - Producer producer = initProducer(config); - log.info("[PRODUCER_CACHE][配置({}) 对应的 producer 已创建并启动]", config); - return producer; - } catch (Exception e) { - log.error("[PRODUCER_CACHE][配置({}) 对应的 producer 创建启动失败]", config, e); - throw e; // 抛出异常，触发缓存加载失败机制 - } - } - - }); - - /** - * 获取生产者 - * - * @param config 配置信息 - * @return 生产者对象 - */ - protected Producer getProducer(Config config) throws Exception { - return PRODUCER_CACHE.get(config); - } - - /** - * 初始化生产者 - * - * @param config 配置信息 - * @return 生产者对象 - * @throws Exception 如果初始化失败 - */ - protected abstract Producer initProducer(Config config) throws Exception; - - /** - * 关闭生产者 - * - * @param producer 生产者对象 - */ - protected abstract void closeProducer(Producer producer) throws Exception; - - @Override - @SuppressWarnings({"unchecked"}) - public void execute(IotDeviceMessage message, IotDataSinkDO dataBridge) { - if (ObjUtil.notEqual(dataBridge.getType(), getType())) { - return; - } - try { - execute0(message, (Config) dataBridge.getConfig()); - } catch (Exception e) { - log.error("[execute][桥梁配置 config({}) 对应的 message({}) 发送异常]", dataBridge.getConfig(), message, e); - } - } - -} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDataBridgeExecute.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDataBridgeExecute.java deleted file mode 100644 index 48e7f47..0000000 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDataBridgeExecute.java +++ /dev/null @@ -1,45 +0,0 @@ -package cn.iocoder.yudao.module.iot.service.rule.data.action; - -import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; -import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataSinkDO; - -/** - * IoT 数据流转目的的执行器 execute 接口 - * - * @author HUIHUI - */ -public interface IotDataBridgeExecute { - - /** - * 获取数据流转目的类型 - * - * @return 数据流转目的类型 - */ - Integer getType(); - - /** - * 执行数据流转目的操作 - * - * @param message 设备消息 - * @param dataBridge 数据流转目的 - */ - @SuppressWarnings({"unchecked"}) - default void execute(IotDeviceMessage message, IotDataSinkDO dataBridge) throws Exception { - // 1.1 校验数据流转目的类型 - if (!getType().equals(dataBridge.getType())) { - return; - } - - // 1.2 执行对应的数据流转目的发送消息 - execute0(message, (Config) dataBridge.getConfig()); - } - - /** - * 【真正】执行数据流转目的操作 - * - * @param message 设备消息 - * @param config 桥梁配置 - */ - void execute0(IotDeviceMessage message, Config config) throws Exception; - -} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDataRuleAction.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDataRuleAction.java new file mode 100644 index 0000000..8e6458b --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDataRuleAction.java @@ -0,0 +1,28 @@ +package cn.iocoder.yudao.module.iot.service.rule.data.action; + +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; +import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataSinkDO; + +/** + * IoT 数据流转目的的执行器 action 接口 + * + * @author HUIHUI + */ +public interface IotDataRuleAction { + + /** + * 获取数据流转目的类型 + * + * @return 数据流转目的类型 + */ + Integer getType(); + + /** + * 执行数据流转目的操作 + * + * @param message 设备消息 + * @param dataSink 数据流转目的 + */ + void execute(IotDeviceMessage message, IotDataSinkDO dataSink); + +} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDataRuleCacheableAction.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDataRuleCacheableAction.java new file mode 100644 index 0000000..4319469 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDataRuleCacheableAction.java @@ -0,0 +1,117 @@ +package cn.iocoder.yudao.module.iot.service.rule.data.action; + +import cn.hutool.core.lang.Assert; +import cn.hutool.core.util.ObjUtil; +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; +import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataSinkDO; +import com.google.common.cache.CacheBuilder; +import com.google.common.cache.CacheLoader; +import com.google.common.cache.LoadingCache; +import com.google.common.cache.RemovalListener; +import lombok.extern.slf4j.Slf4j; + +import java.time.Duration; + +// TODO @芋艿：数据库 +// TODO @芋艿：mqtt +// TODO @芋艿：tcp +// TODO @芋艿：websocket + +/** + * 可缓存的 {@link IotDataRuleAction} 抽象实现 + * + * 该类提供了一个通用的缓存机制，用于管理各类数据桥接的生产者(Producer)实例。 + * + * 主要特点: + * - 基于Guava Cache实现高效的生产者实例缓存管理 + * - 自动处理生产者的生命周期（创建、获取、关闭） + * - 支持 30 分钟未访问自动过期清理机制 + * - 异常处理与日志记录，便于问题排查 + * + * 子类需要实现: + * - initProducer(Config) - 初始化特定类型的生产者实例 + * - closeProducer(Producer) - 关闭生产者实例并释放资源 + * + * @param 配置信息类型，用于初始化生产者 + * @param 生产者类型，负责将数据发送到目标系统 + * @author HUIHUI + */ +@Slf4j +public abstract class IotDataRuleCacheableAction implements IotDataRuleAction { + + /** + * Producer 缓存 + */ + private final LoadingCache PRODUCER_CACHE = CacheBuilder.newBuilder() + .expireAfterAccess(Duration.ofMinutes(30)) // 30 分钟未访问就提前过期 + .removalListener((RemovalListener) notification -> { + Producer producer = notification.getValue(); + try { + closeProducer(producer); + log.info("[PRODUCER_CACHE][配置({}) 对应的 producer 已关闭]", notification.getKey()); + } catch (Exception e) { + log.error("[PRODUCER_CACHE][配置({}) 对应的 producer 关闭失败]", notification.getKey(), e); + } + }) + .build(new CacheLoader() { + + @Override + public Producer load(Config config) throws Exception { + try { + Producer producer = initProducer(config); + log.info("[PRODUCER_CACHE][配置({}) 对应的 producer 已创建并启动]", config); + return producer; + } catch (Exception e) { + log.error("[PRODUCER_CACHE][配置({}) 对应的 producer 创建启动失败]", config, e); + throw e; // 抛出异常，触发缓存加载失败机制 + } + } + + }); + + /** + * 获取生产者 + * + * @param config 配置信息 + * @return 生产者对象 + */ + protected Producer getProducer(Config config) throws Exception { + return PRODUCER_CACHE.get(config); + } + + /** + * 初始化生产者 + * + * @param config 配置信息 + * @return 生产者对象 + * @throws Exception 如果初始化失败 + */ + protected abstract Producer initProducer(Config config) throws Exception; + + /** + * 关闭生产者 + * + * @param producer 生产者对象 + */ + protected abstract void closeProducer(Producer producer) throws Exception; + + @Override + @SuppressWarnings({"unchecked"}) + public void execute(IotDeviceMessage message, IotDataSinkDO dataSink) { + Assert.isTrue(ObjUtil.equal(dataSink.getType(), getType()), "类型({})不匹配", dataSink.getType()); + try { + execute(message, (Config) dataSink.getConfig()); + } catch (Exception e) { + log.error("[execute][桥梁配置 config({}) 对应的 message({}) 发送异常]", dataSink.getConfig(), message, e); + } + } + + /** + * 执行数据流转 + * + * @param message 设备消息 + * @param config 配置信息 + */ + protected abstract void execute(IotDeviceMessage message, Config config) throws Exception; + +} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotHttpDataBridgeExecute.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotHttpDataBridgeExecute.java deleted file mode 100644 index da02677..0000000 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotHttpDataBridgeExecute.java +++ /dev/null @@ -1,88 +0,0 @@ -package cn.iocoder.yudao.module.iot.service.rule.data.action; - -import cn.hutool.core.collection.CollUtil; -import cn.iocoder.yudao.framework.common.util.http.HttpUtils; -import cn.iocoder.yudao.framework.common.util.json.JsonUtils; -import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataSinkHttpConfig; -import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; -import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum; -import jakarta.annotation.Resource; -import lombok.extern.slf4j.Slf4j; -import org.springframework.http.*; -import org.springframework.stereotype.Component; -import org.springframework.web.client.RestTemplate; -import org.springframework.web.util.UriComponentsBuilder; - -import java.util.HashMap; -import java.util.Map; - -/** - * Http 的 {@link IotDataBridgeExecute} 实现类 - * - * @author HUIHUI - */ -@Component -@Slf4j -public class IotHttpDataBridgeExecute implements IotDataBridgeExecute { - - @Resource - private RestTemplate restTemplate; - - @Override - public Integer getType() { - return IotDataSinkTypeEnum.HTTP.getType(); - } - - @Override - @SuppressWarnings({"unchecked", "deprecation"}) - public void execute0(IotDeviceMessage message, IotDataSinkHttpConfig config) { - String url = null; - HttpMethod method = HttpMethod.valueOf(config.getMethod().toUpperCase()); - HttpEntity requestEntity = null; - ResponseEntity responseEntity = null; - try { - // 1.1 构建 Header - HttpHeaders headers = new HttpHeaders(); - if (CollUtil.isNotEmpty(config.getHeaders())) { - config.getHeaders().putAll(config.getHeaders()); - } - // TODO @puhui999：@yunai：可能需要通过设备查询到租户，然后 set -// headers.add(HEADER_TENANT_ID, message.getTenantId().toString()); - // 1.2 构建 URL - UriComponentsBuilder uriBuilder = UriComponentsBuilder.fromUriString(config.getUrl()); - if (CollUtil.isNotEmpty(config.getQuery())) { - config.getQuery().forEach(uriBuilder::queryParam); - } - // 1.3 构建请求体 - if (method == HttpMethod.GET) { - uriBuilder.queryParam("message", HttpUtils.encodeUtf8(JsonUtils.toJsonString(message))); - url = uriBuilder.build().toUriString(); - requestEntity = new HttpEntity<>(headers); - } else { - url = uriBuilder.build().toUriString(); - Map requestBody = JsonUtils.parseObject(config.getBody(), Map.class); - if (requestBody == null) { - requestBody = new HashMap<>(); - } - requestBody.put("message", message); - headers.add(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_UTF8_VALUE); - requestEntity = new HttpEntity<>(JsonUtils.toJsonString(requestBody), headers); - } - - // 2.1 发送请求 - responseEntity = restTemplate.exchange(url, method, requestEntity, String.class); - // 2.2 记录日志 - if (responseEntity.getStatusCode().is2xxSuccessful()) { - log.info("[executeHttp][message({}) config({}) url({}) method({}) requestEntity({}) 请求成功({})]", - message, config, url, method, requestEntity, responseEntity); - } else { - log.error("[executeHttp][message({}) config({}) url({}) method({}) requestEntity({}) 请求失败({})]", - message, config, url, method, requestEntity, responseEntity); - } - } catch (Exception e) { - log.error("[executeHttp][message({}) config({}) url({}) method({}) requestEntity({}) 请求异常({})]", - message, config, url, method, requestEntity, responseEntity, e); - } - } - -} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotHttpDataSinkAction.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotHttpDataSinkAction.java new file mode 100644 index 0000000..c23e346 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotHttpDataSinkAction.java @@ -0,0 +1,91 @@ +package cn.iocoder.yudao.module.iot.service.rule.data.action; + +import cn.hutool.core.collection.CollUtil; +import cn.hutool.core.lang.Assert; +import cn.iocoder.yudao.framework.common.util.http.HttpUtils; +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.IotDataSinkDO; +import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataSinkHttpConfig; +import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum; +import jakarta.annotation.Resource; +import lombok.extern.slf4j.Slf4j; +import org.springframework.http.*; +import org.springframework.stereotype.Component; +import org.springframework.web.client.RestTemplate; +import org.springframework.web.util.UriComponentsBuilder; + +import java.util.HashMap; +import java.util.Map; + +/** + * HTTP 的 {@link IotDataRuleAction} 实现类 + * + * @author HUIHUI + */ +@Component +@Slf4j +public class IotHttpDataSinkAction implements IotDataRuleAction { + + @Resource + private RestTemplate restTemplate; + + @Override + public Integer getType() { + return IotDataSinkTypeEnum.HTTP.getType(); + } + + @Override + public void execute(IotDeviceMessage message, IotDataSinkDO dataSink) { + IotDataSinkHttpConfig config = (IotDataSinkHttpConfig) dataSink.getConfig(); + Assert.notNull(config, "配置({})不能为空", dataSink.getId()); + String url = null; + HttpMethod method = HttpMethod.valueOf(config.getMethod().toUpperCase()); + HttpEntity requestEntity = null; + ResponseEntity responseEntity = null; + try { + // 1.1 构建 Header + HttpHeaders headers = new HttpHeaders(); + if (CollUtil.isNotEmpty(config.getHeaders())) { + config.getHeaders().putAll(config.getHeaders()); + } + // TODO @puhui999：@yunai：可能需要通过设备查询到租户，然后 set +// headers.add(HEADER_TENANT_ID, message.getTenantId().toString()); + // 1.2 构建 URL + UriComponentsBuilder uriBuilder = UriComponentsBuilder.fromUriString(config.getUrl()); + if (CollUtil.isNotEmpty(config.getQuery())) { + config.getQuery().forEach(uriBuilder::queryParam); + } + // 1.3 构建请求体 + if (method == HttpMethod.GET) { + uriBuilder.queryParam("message", HttpUtils.encodeUtf8(JsonUtils.toJsonString(message))); + url = uriBuilder.build().toUriString(); + requestEntity = new HttpEntity<>(headers); + } else { + url = uriBuilder.build().toUriString(); + Map requestBody = JsonUtils.parseObject(config.getBody(), Map.class); + if (requestBody == null) { + requestBody = new HashMap<>(); + } + requestBody.put("message", message); + headers.add(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_UTF8_VALUE); + requestEntity = new HttpEntity<>(JsonUtils.toJsonString(requestBody), headers); + } + + // 2.1 发送请求 + responseEntity = restTemplate.exchange(url, method, requestEntity, String.class); + // 2.2 记录日志 + if (responseEntity.getStatusCode().is2xxSuccessful()) { + log.info("[executeHttp][message({}) config({}) url({}) method({}) requestEntity({}) 请求成功({})]", + message, config, url, method, requestEntity, responseEntity); + } else { + log.error("[executeHttp][message({}) config({}) url({}) method({}) requestEntity({}) 请求失败({})]", + message, config, url, method, requestEntity, responseEntity); + } + } catch (Exception e) { + log.error("[executeHttp][message({}) config({}) url({}) method({}) requestEntity({}) 请求异常({})]", + message, config, url, method, requestEntity, responseEntity, e); + } + } + +} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotKafkaDataRuleAction.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotKafkaDataRuleAction.java new file mode 100644 index 0000000..5bbbe07 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotKafkaDataRuleAction.java @@ -0,0 +1,78 @@ +package cn.iocoder.yudao.module.iot.service.rule.data.action; + +import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataSinkKafkaConfig; +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; +import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum; +import lombok.extern.slf4j.Slf4j; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.common.serialization.StringSerializer; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.stereotype.Component; + +import java.time.Duration; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +/** + * Kafka 的 {@link IotDataRuleAction} 实现类 + * + * @author HUIHUI + */ +@ConditionalOnClass(name = "org.springframework.kafka.core.KafkaTemplate") +@Component +@Slf4j +public class IotKafkaDataRuleAction extends + IotDataRuleCacheableAction> { + + private static final Duration SEND_TIMEOUT = Duration.ofSeconds(10); + + @Override + public Integer getType() { + return IotDataSinkTypeEnum.KAFKA.getType(); + } + + @Override + public void execute(IotDeviceMessage message, IotDataSinkKafkaConfig config) throws Exception { + // 1. 获取或创建 KafkaTemplate + KafkaTemplate kafkaTemplate = getProducer(config); + + // 2. 发送消息并等待结果 + kafkaTemplate.send(config.getTopic(), message.toString()) + .get(SEND_TIMEOUT.getSeconds(), TimeUnit.SECONDS); // 添加超时等待 + log.info("[execute0][message({}) 发送成功]", message); + } + + @Override + protected KafkaTemplate initProducer(IotDataSinkKafkaConfig config) { + // 1.1 构建生产者配置 + Map props = new HashMap<>(); + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBootstrapServers()); + props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); + props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); + // 1.2 如果配置了认证信息 + if (config.getUsername() != null && config.getPassword() != null) { + props.put("security.protocol", "SASL_PLAINTEXT"); + props.put("sasl.mechanism", "PLAIN"); + props.put("sasl.jaas.config", + "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"" + + config.getUsername() + "\" password=\"" + config.getPassword() + "\";"); + } + // 1.3 如果启用 SSL + if (Boolean.TRUE.equals(config.getSsl())) { + props.put("security.protocol", "SSL"); + } + + // 2. 创建 KafkaTemplate + DefaultKafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory<>(props); + return new KafkaTemplate<>(producerFactory); + } + + @Override + protected void closeProducer(KafkaTemplate producer) { + producer.destroy(); + } + +} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotKafkaMQDataBridgeExecute.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotKafkaMQDataBridgeExecute.java deleted file mode 100644 index ec7eade..0000000 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotKafkaMQDataBridgeExecute.java +++ /dev/null @@ -1,78 +0,0 @@ -package cn.iocoder.yudao.module.iot.service.rule.data.action; - -import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataSinkKafkaConfig; -import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; -import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum; -import lombok.extern.slf4j.Slf4j; -import org.apache.kafka.clients.producer.ProducerConfig; -import org.apache.kafka.common.serialization.StringSerializer; -import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; -import org.springframework.kafka.core.DefaultKafkaProducerFactory; -import org.springframework.kafka.core.KafkaTemplate; -import org.springframework.stereotype.Component; - -import java.time.Duration; -import java.util.HashMap; -import java.util.Map; -import java.util.concurrent.TimeUnit; - -/** - * Kafka 的 {@link IotDataBridgeExecute} 实现类 - * - * @author HUIHUI - */ -@ConditionalOnClass(name = "org.springframework.kafka.core.KafkaTemplate") -@Component -@Slf4j -public class IotKafkaMQDataBridgeExecute extends - AbstractCacheableDataBridgeExecute> { - - private static final Duration SEND_TIMEOUT = Duration.ofMillis(10000); // 10 秒超时时间 - - @Override - public Integer getType() { - return IotDataSinkTypeEnum.KAFKA.getType(); - } - - @Override - public void execute0(IotDeviceMessage message, IotDataSinkKafkaConfig config) throws Exception { - // 1. 获取或创建 KafkaTemplate - KafkaTemplate kafkaTemplate = getProducer(config); - - // 2. 发送消息并等待结果 - kafkaTemplate.send(config.getTopic(), message.toString()) - .get(SEND_TIMEOUT.getSeconds(), TimeUnit.SECONDS); // 添加超时等待 - log.info("[execute0][message({}) 发送成功]", message); - } - - @Override - protected KafkaTemplate initProducer(IotDataSinkKafkaConfig config) { - // 1.1 构建生产者配置 - Map props = new HashMap<>(); - props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBootstrapServers()); - props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); - props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); - // 1.2 如果配置了认证信息 - if (config.getUsername() != null && config.getPassword() != null) { - props.put("security.protocol", "SASL_PLAINTEXT"); - props.put("sasl.mechanism", "PLAIN"); - props.put("sasl.jaas.config", - "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"" - + config.getUsername() + "\" password=\"" + config.getPassword() + "\";"); - } - // 1.3 如果启用 SSL - if (Boolean.TRUE.equals(config.getSsl())) { - props.put("security.protocol", "SSL"); - } - - // 2. 创建 KafkaTemplate - DefaultKafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory<>(props); - return new KafkaTemplate<>(producerFactory); - } - - @Override - protected void closeProducer(KafkaTemplate producer) { - producer.destroy(); - } - -} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRabbitMQDataBridgeExecute.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRabbitMQDataBridgeExecute.java deleted file mode 100644 index d8abaa7..0000000 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRabbitMQDataBridgeExecute.java +++ /dev/null @@ -1,77 +0,0 @@ -package cn.iocoder.yudao.module.iot.service.rule.data.action; - -import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataSinkRabbitMQConfig; -import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; -import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum; -import com.rabbitmq.client.Channel; -import com.rabbitmq.client.Connection; -import com.rabbitmq.client.ConnectionFactory; -import lombok.extern.slf4j.Slf4j; -import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; -import org.springframework.stereotype.Component; - -import java.nio.charset.StandardCharsets; - -/** - * RabbitMQ 的 {@link IotDataBridgeExecute} 实现类 - * - * @author HUIHUI - */ -@ConditionalOnClass(name = "com.rabbitmq.client.Channel") -@Component -@Slf4j -public class IotRabbitMQDataBridgeExecute extends - AbstractCacheableDataBridgeExecute { - - - @Override - public Integer getType() { - return IotDataSinkTypeEnum.RABBITMQ.getType(); - } - - @Override - public void execute0(IotDeviceMessage message, IotDataSinkRabbitMQConfig config) throws Exception { - // 1. 获取或创建 Channel - Channel channel = getProducer(config); - - // 2.1 声明交换机、队列和绑定关系 - channel.exchangeDeclare(config.getExchange(), "direct", true); - channel.queueDeclare(config.getQueue(), true, false, false, null); - channel.queueBind(config.getQueue(), config.getExchange(), config.getRoutingKey()); - - // 2.2 发送消息 - channel.basicPublish(config.getExchange(), config.getRoutingKey(), null, - message.toString().getBytes(StandardCharsets.UTF_8)); - log.info("[executeRabbitMQ][message({}) config({}) 发送成功]", message, config); - } - - @Override - @SuppressWarnings("resource") - protected Channel initProducer(IotDataSinkRabbitMQConfig config) throws Exception { - // 1. 创建连接工厂 - ConnectionFactory factory = new ConnectionFactory(); - factory.setHost(config.getHost()); - factory.setPort(config.getPort()); - factory.setVirtualHost(config.getVirtualHost()); - factory.setUsername(config.getUsername()); - factory.setPassword(config.getPassword()); - - // 2. 创建连接 - Connection connection = factory.newConnection(); - - // 3. 创建信道 - return connection.createChannel(); - } - - @Override - protected void closeProducer(Channel channel) throws Exception { - if (channel.isOpen()) { - channel.close(); - } - Connection connection = channel.getConnection(); - if (connection.isOpen()) { - connection.close(); - } - } - -} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRabbitMQDataRuleAction.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRabbitMQDataRuleAction.java new file mode 100644 index 0000000..89d3500 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRabbitMQDataRuleAction.java @@ -0,0 +1,73 @@ +package cn.iocoder.yudao.module.iot.service.rule.data.action; + +import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataSinkRabbitMQConfig; +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; +import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum; +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.Connection; +import com.rabbitmq.client.ConnectionFactory; +import lombok.extern.slf4j.Slf4j; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.stereotype.Component; + +import java.nio.charset.StandardCharsets; + +/** + * RabbitMQ 的 {@link IotDataRuleAction} 实现类 + * + * @author HUIHUI + */ +@ConditionalOnClass(name = "com.rabbitmq.client.Channel") +@Component +@Slf4j +public class IotRabbitMQDataRuleAction extends + IotDataRuleCacheableAction { + + @Override + public Integer getType() { + return IotDataSinkTypeEnum.RABBITMQ.getType(); + } + + @Override + public void execute(IotDeviceMessage message, IotDataSinkRabbitMQConfig config) throws Exception { + // 1.1 获取或创建 Channel + Channel channel = getProducer(config); + // 1.2 声明交换机、队列和绑定关系 + channel.exchangeDeclare(config.getExchange(), "direct", true); + channel.queueDeclare(config.getQueue(), true, false, false, null); + channel.queueBind(config.getQueue(), config.getExchange(), config.getRoutingKey()); + + // 2. 发送消息 + channel.basicPublish(config.getExchange(), config.getRoutingKey(), null, + message.toString().getBytes(StandardCharsets.UTF_8)); + log.info("[executeRabbitMQ][message({}) config({}) 发送成功]", message, config); + } + + @Override + @SuppressWarnings("resource") + protected Channel initProducer(IotDataSinkRabbitMQConfig config) throws Exception { + // 1. 创建连接工厂 + ConnectionFactory factory = new ConnectionFactory(); + factory.setHost(config.getHost()); + factory.setPort(config.getPort()); + factory.setVirtualHost(config.getVirtualHost()); + factory.setUsername(config.getUsername()); + factory.setPassword(config.getPassword()); + // 2. 创建连接 + Connection connection = factory.newConnection(); + // 3. 创建信道 + return connection.createChannel(); + } + + @Override + protected void closeProducer(Channel channel) throws Exception { + if (channel.isOpen()) { + channel.close(); + } + Connection connection = channel.getConnection(); + if (connection.isOpen()) { + connection.close(); + } + } + +} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRedisStreamDataBridgeExecute.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRedisStreamDataBridgeExecute.java deleted file mode 100644 index be33704..0000000 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRedisStreamDataBridgeExecute.java +++ /dev/null @@ -1,80 +0,0 @@ -package cn.iocoder.yudao.module.iot.service.rule.data.action; - -import cn.hutool.core.util.StrUtil; -import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataSinkRedisStreamConfig; -import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; -import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum; -import lombok.extern.slf4j.Slf4j; -import org.redisson.Redisson; -import org.redisson.api.RedissonClient; -import org.redisson.config.Config; -import org.redisson.config.SingleServerConfig; -import org.redisson.spring.data.connection.RedissonConnectionFactory; -import org.springframework.data.redis.connection.RedisConnectionFactory; -import org.springframework.data.redis.connection.stream.ObjectRecord; -import org.springframework.data.redis.connection.stream.StreamRecords; -import org.springframework.data.redis.core.RedisTemplate; -import org.springframework.data.redis.serializer.RedisSerializer; -import org.springframework.stereotype.Component; - -/** - * Redis Stream 的 {@link IotDataBridgeExecute} 实现类 - * - * @author HUIHUI - */ -@Component -@Slf4j -public class IotRedisStreamDataBridgeExecute extends - AbstractCacheableDataBridgeExecute> { - - @Override - public Integer getType() { - return IotDataSinkTypeEnum.REDIS_STREAM.getType(); - } - - @Override - public void execute0(IotDeviceMessage message, IotDataSinkRedisStreamConfig config) throws Exception { - // 1. 获取 RedisTemplate - RedisTemplate redisTemplate = getProducer(config); - - // 2. 创建并发送 Stream 记录 - ObjectRecord record = StreamRecords.newRecord() - .ofObject(message).withStreamKey(config.getTopic()); - String recordId = String.valueOf(redisTemplate.opsForStream().add(record)); - log.info("[executeRedisStream][消息发送成功] messageId: {}, config: {}", recordId, config); - } - - @Override - protected RedisTemplate initProducer(IotDataSinkRedisStreamConfig config) { - // 1.1 创建 Redisson 配置 - Config redissonConfig = new Config(); - SingleServerConfig serverConfig = redissonConfig.useSingleServer() - .setAddress("redis://" + config.getHost() + ":" + config.getPort()) - .setDatabase(config.getDatabase()); - // 1.2 设置密码（如果有） - if (StrUtil.isNotBlank(config.getPassword())) { - serverConfig.setPassword(config.getPassword()); - } - - // 创建 RedisTemplate 并配置 - RedissonClient redisson = Redisson.create(redissonConfig); - RedisTemplate template = new RedisTemplate<>(); - template.setConnectionFactory(new RedissonConnectionFactory(redisson)); - // 设置序列化器 - template.setKeySerializer(RedisSerializer.string()); - template.setHashKeySerializer(RedisSerializer.string()); - template.setValueSerializer(RedisSerializer.json()); - template.setHashValueSerializer(RedisSerializer.json()); - template.afterPropertiesSet(); - return template; - } - - @Override - protected void closeProducer(RedisTemplate producer) throws Exception { - RedisConnectionFactory factory = producer.getConnectionFactory(); - if (factory != null) { - ((RedissonConnectionFactory) factory).destroy(); - } - } - -} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRedisStreamRuleAction.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRedisStreamRuleAction.java new file mode 100644 index 0000000..9870c7d --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRedisStreamRuleAction.java @@ -0,0 +1,80 @@ +package cn.iocoder.yudao.module.iot.service.rule.data.action; + +import cn.hutool.core.util.StrUtil; +import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataSinkRedisStreamConfig; +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; +import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum; +import lombok.extern.slf4j.Slf4j; +import org.redisson.Redisson; +import org.redisson.api.RedissonClient; +import org.redisson.config.Config; +import org.redisson.config.SingleServerConfig; +import org.redisson.spring.data.connection.RedissonConnectionFactory; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.connection.stream.ObjectRecord; +import org.springframework.data.redis.connection.stream.StreamRecords; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.serializer.RedisSerializer; +import org.springframework.stereotype.Component; + +/** + * Redis Stream 的 {@link IotDataRuleAction} 实现类 + * + * @author HUIHUI + */ +@Component +@Slf4j +public class IotRedisStreamRuleAction extends + IotDataRuleCacheableAction> { + + @Override + public Integer getType() { + return IotDataSinkTypeEnum.REDIS_STREAM.getType(); + } + + @Override + public void execute(IotDeviceMessage message, IotDataSinkRedisStreamConfig config) throws Exception { + // 1. 获取 RedisTemplate + RedisTemplate redisTemplate = getProducer(config); + + // 2. 创建并发送 Stream 记录 + ObjectRecord record = StreamRecords.newRecord() + .ofObject(message).withStreamKey(config.getTopic()); + String recordId = String.valueOf(redisTemplate.opsForStream().add(record)); + log.info("[executeRedisStream][消息发送成功] messageId: {}, config: {}", recordId, config); + } + + @Override + protected RedisTemplate initProducer(IotDataSinkRedisStreamConfig config) { + // 1.1 创建 Redisson 配置 + Config redissonConfig = new Config(); + SingleServerConfig serverConfig = redissonConfig.useSingleServer() + .setAddress("redis://" + config.getHost() + ":" + config.getPort()) + .setDatabase(config.getDatabase()); + // 1.2 设置密码（如果有） + if (StrUtil.isNotBlank(config.getPassword())) { + serverConfig.setPassword(config.getPassword()); + } + + // 创建 RedisTemplate 并配置 + RedissonClient redisson = Redisson.create(redissonConfig); + RedisTemplate template = new RedisTemplate<>(); + template.setConnectionFactory(new RedissonConnectionFactory(redisson)); + // 设置序列化器 + template.setKeySerializer(RedisSerializer.string()); + template.setHashKeySerializer(RedisSerializer.string()); + template.setValueSerializer(RedisSerializer.json()); + template.setHashValueSerializer(RedisSerializer.json()); + template.afterPropertiesSet(); + return template; + } + + @Override + protected void closeProducer(RedisTemplate producer) throws Exception { + RedisConnectionFactory factory = producer.getConnectionFactory(); + if (factory != null) { + ((RedissonConnectionFactory) factory).destroy(); + } + } + +} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRocketMQDataBridgeExecute.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRocketMQDataBridgeExecute.java deleted file mode 100644 index 6a8d668..0000000 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRocketMQDataBridgeExecute.java +++ /dev/null @@ -1,65 +0,0 @@ -package cn.iocoder.yudao.module.iot.service.rule.data.action; - -import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataSinkRocketMQConfig; -import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; -import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum; -import lombok.extern.slf4j.Slf4j; -import org.apache.rocketmq.client.producer.DefaultMQProducer; -import org.apache.rocketmq.client.producer.SendResult; -import org.apache.rocketmq.client.producer.SendStatus; -import org.apache.rocketmq.common.message.Message; -import org.apache.rocketmq.remoting.common.RemotingHelper; -import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; -import org.springframework.stereotype.Component; - -/** - * RocketMQ 的 {@link IotDataBridgeExecute} 实现类 - * - * @author HUIHUI - */ -@ConditionalOnClass(name = "org.apache.rocketmq.client.producer.DefaultMQProducer") -@Component -@Slf4j -public class IotRocketMQDataBridgeExecute extends - AbstractCacheableDataBridgeExecute { - - @Override - public Integer getType() { - return IotDataSinkTypeEnum.ROCKETMQ.getType(); - } - - @Override - public void execute0(IotDeviceMessage message, IotDataSinkRocketMQConfig config) throws Exception { - // 1. 获取或创建 Producer - DefaultMQProducer producer = getProducer(config); - - // 2.1 创建消息对象，指定Topic、Tag和消息体 - Message msg = new Message( - config.getTopic(), - config.getTags(), - message.toString().getBytes(RemotingHelper.DEFAULT_CHARSET) - ); - // 2.2 发送同步消息并处理结果 - SendResult sendResult = producer.send(msg); - // 2.3 处理发送结果 - if (SendStatus.SEND_OK.equals(sendResult.getSendStatus())) { - log.info("[executeRocketMQ][message({}) config({}) 发送成功，结果({})]", message, config, sendResult); - } else { - log.error("[executeRocketMQ][message({}) config({}) 发送失败，结果({})]", message, config, sendResult); - } - } - - @Override - protected DefaultMQProducer initProducer(IotDataSinkRocketMQConfig config) throws Exception { - DefaultMQProducer producer = new DefaultMQProducer(config.getGroup()); - producer.setNamesrvAddr(config.getNameServer()); - producer.start(); - return producer; - } - - @Override - protected void closeProducer(DefaultMQProducer producer) { - producer.shutdown(); - } - -} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRocketMQDataRuleAction.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRocketMQDataRuleAction.java new file mode 100644 index 0000000..1a212ec --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRocketMQDataRuleAction.java @@ -0,0 +1,65 @@ +package cn.iocoder.yudao.module.iot.service.rule.data.action; + +import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataSinkRocketMQConfig; +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; +import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum; +import lombok.extern.slf4j.Slf4j; +import org.apache.rocketmq.client.producer.DefaultMQProducer; +import org.apache.rocketmq.client.producer.SendResult; +import org.apache.rocketmq.client.producer.SendStatus; +import org.apache.rocketmq.common.message.Message; +import org.apache.rocketmq.remoting.common.RemotingHelper; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.stereotype.Component; + +/** + * RocketMQ 的 {@link IotDataRuleAction} 实现类 + * + * @author HUIHUI + */ +@ConditionalOnClass(name = "org.apache.rocketmq.client.producer.DefaultMQProducer") +@Component +@Slf4j +public class IotRocketMQDataRuleAction extends + IotDataRuleCacheableAction { + + @Override + public Integer getType() { + return IotDataSinkTypeEnum.ROCKETMQ.getType(); + } + + @Override + public void execute(IotDeviceMessage message, IotDataSinkRocketMQConfig config) throws Exception { + // 1. 获取或创建 Producer + DefaultMQProducer producer = getProducer(config); + + // 2.1 创建消息对象，指定Topic、Tag和消息体 + Message msg = new Message( + config.getTopic(), + config.getTags(), + message.toString().getBytes(RemotingHelper.DEFAULT_CHARSET) + ); + // 2.2 发送同步消息并处理结果 + SendResult sendResult = producer.send(msg); + // 2.3 处理发送结果 + if (SendStatus.SEND_OK.equals(sendResult.getSendStatus())) { + log.info("[executeRocketMQ][message({}) config({}) 发送成功，结果({})]", message, config, sendResult); + } else { + log.error("[executeRocketMQ][message({}) config({}) 发送失败，结果({})]", message, config, sendResult); + } + } + + @Override + protected DefaultMQProducer initProducer(IotDataSinkRocketMQConfig config) throws Exception { + DefaultMQProducer producer = new DefaultMQProducer(config.getGroup()); + producer.setNamesrvAddr(config.getNameServer()); + producer.start(); + return producer; + } + + @Override + protected void closeProducer(DefaultMQProducer producer) { + producer.shutdown(); + } + +} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotRuleSceneServiceImpl.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotRuleSceneServiceImpl.java index b422e0d..fc77180 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotRuleSceneServiceImpl.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotRuleSceneServiceImpl.java @@ -214,11 +214,6 @@ public class IotRuleSceneServiceImpl implements IotRuleSceneService { .build()); action01.setDeviceControl(actionDeviceControl01); // ruleScene01.getActions().add(action01); // TODO 芋艿：先不测试了 - // 数据桥接（http） - IotRuleSceneDO.ActionConfig action02 = new IotRuleSceneDO.ActionConfig(); - action02.setType(IotRuleSceneActionTypeEnum.DATA_BRIDGE.getType()); - action02.setDataBridgeId(1L); - ruleScene01.getActions().add(action02); return ListUtil.toList(ruleScene01); } diff --git a/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotDataBridgeExecuteTest.java b/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotDataBridgeExecuteTest.java index 52599c4..d8cf9ee 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotDataBridgeExecuteTest.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotDataBridgeExecuteTest.java @@ -23,7 +23,7 @@ import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.when; /** - * {@link IotDataBridgeExecute} 实现类的单元测试 + * {@link IotDataRuleAction} 实现类的单元测试 * * @author HUIHUI */ @@ -37,7 +37,7 @@ public class IotDataBridgeExecuteTest extends BaseMockitoUnitTest { private RestTemplate restTemplate; @InjectMocks - private IotHttpDataBridgeExecute httpDataBridgeExecute; + private IotHttpDataSinkAction httpDataBridgeExecute; @BeforeEach public void setUp() { @@ -51,7 +51,7 @@ public class IotDataBridgeExecuteTest extends BaseMockitoUnitTest { @Test public void testKafkaMQDataBridge() throws Exception { // 1. 创建执行器实例 - IotKafkaMQDataBridgeExecute action = new IotKafkaMQDataBridgeExecute(); + IotKafkaDataRuleAction action = new IotKafkaDataRuleAction(); // 2. 创建配置 IotDataSinkKafkaConfig config = new IotDataSinkKafkaConfig() @@ -68,7 +68,7 @@ public class IotDataBridgeExecuteTest extends BaseMockitoUnitTest { @Test public void testRabbitMQDataBridge() throws Exception { // 1. 创建执行器实例 - IotRabbitMQDataBridgeExecute action = new IotRabbitMQDataBridgeExecute(); + IotRabbitMQDataRuleAction action = new IotRabbitMQDataRuleAction(); // 2. 创建配置 IotDataSinkRabbitMQConfig config = new IotDataSinkRabbitMQConfig() @@ -88,7 +88,7 @@ public class IotDataBridgeExecuteTest extends BaseMockitoUnitTest { @Test public void testRedisStreamDataBridge() throws Exception { // 1. 创建执行器实例 - IotRedisStreamDataBridgeExecute action = new IotRedisStreamDataBridgeExecute(); + IotRedisStreamRuleAction action = new IotRedisStreamRuleAction(); // 2. 创建配置 IotDataSinkRedisStreamConfig config = new IotDataSinkRedisStreamConfig() @@ -105,7 +105,7 @@ public class IotDataBridgeExecuteTest extends BaseMockitoUnitTest { @Test public void testRocketMQDataBridge() throws Exception { // 1. 创建执行器实例 - IotRocketMQDataBridgeExecute action = new IotRocketMQDataBridgeExecute(); + IotRocketMQDataRuleAction action = new IotRocketMQDataRuleAction(); // 2. 创建配置 IotDataSinkRocketMQConfig config = new IotDataSinkRocketMQConfig() @@ -142,7 +142,7 @@ public class IotDataBridgeExecuteTest extends BaseMockitoUnitTest { * @param type MQ 类型 * @throws Exception 如果执行过程中发生异常 */ - private void executeAndVerifyCache(IotDataBridgeExecute action, IotAbstractDataSinkConfig config, String type) + private void executeAndVerifyCache(IotDataRuleAction action, IotAbstractDataSinkConfig config, String type) throws Exception { log.info("[test{}DataBridge][第一次执行，应该会创建新的 producer]", type); action.execute(message, new IotDataSinkDO().setType(action.getType()).setConfig(config)); -- libgit2 0.21.4