diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index de8e85fc..7f97fd2a 100644 --- a/docker/docker-compose.yml +++ b/docker/docker-compose.yml @@ -46,7 +46,7 @@ services: ] emqx: - image: emqx:5.1 + image: emqx:5.6.1 container_name: emqx ports: - 1883:1883 diff --git a/fastbee-common/src/main/java/com/fastbee/common/core/mq/message/ReportDataBo.java b/fastbee-common/src/main/java/com/fastbee/common/core/mq/message/ReportDataBo.java index c4eec915..78bf8a0c 100644 --- a/fastbee-common/src/main/java/com/fastbee/common/core/mq/message/ReportDataBo.java +++ b/fastbee-common/src/main/java/com/fastbee/common/core/mq/message/ReportDataBo.java @@ -56,8 +56,24 @@ public class ReportDataBo { private GwDeviceBo gwDeviceBo; + /** + * 用户ID + */ private Long userId; + + /** + * 用户名称 + */ private String userName; + + /** + * 设备名称 + */ private String deviceName; + /** + * 消息类型(详见枚举：{@link com.fastbee.common.enums.ServerType }) + */ + private String serverType; + } diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/emq/subscribeCallback.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/emq/subscribeCallback.java index a7079a9a..8618056b 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/emq/subscribeCallback.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/emq/subscribeCallback.java @@ -32,9 +32,9 @@ public class subscribeCallback implements IMqttMessageListener { @Override public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception { String message = new String(mqttMessage.getPayload()); - log.info("接收消息主题 : " + topic); - log.info("接收消息Qos : " + mqttMessage.getQos()); - log.info("接收消息内容 : " + message); + log.info("接收消息主题 : {}", topic); + log.info("接收消息Qos : {}", mqttMessage.getQos()); + log.info("接收消息内容 : {}", message); //这里默认设备编号长度超过9位 String[] split = topic.split("/"); diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DataHandlerImpl.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DataHandlerImpl.java index 7eae4549..31aa7e89 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DataHandlerImpl.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DataHandlerImpl.java @@ -8,6 +8,7 @@ import com.fastbee.common.core.thingsModel.ThingsModelSimpleItem; import com.fastbee.common.core.thingsModel.ThingsModelValuesInput; import com.fastbee.common.enums.DeviceLogTypeEnum; import com.fastbee.common.enums.FunctionReplyStatus; +import com.fastbee.common.enums.ServerType; import com.fastbee.common.enums.TopicType; import com.fastbee.common.enums.scenemodel.SceneModelTagOpreationEnum; import com.fastbee.common.exception.ServiceException; @@ -25,6 +26,7 @@ import com.fastbee.iot.domain.*; import com.fastbee.iot.mapper.SceneTagPointsMapper; import com.fastbee.iot.model.scenemodel.SceneModelTagCacheVO; import com.fastbee.iot.model.scenemodel.SceneModelTagCycleVO; +import com.fastbee.iot.ruleEngine.RuleProcess; import com.fastbee.iot.service.IDeviceService; import com.fastbee.iot.service.IEventLogService; import com.fastbee.iot.service.IFunctionLogService; @@ -50,7 +52,6 @@ import java.util.*; @Slf4j public class DataHandlerImpl implements IDataHandler { - @Resource private IDeviceService deviceService; @Resource @@ -60,6 +61,8 @@ public class DataHandlerImpl implements IDataHandler { @Resource private IRuleEngine ruleEngine; @Resource + private RuleProcess ruleProcess; + @Resource private ConditionTriggerTaskExecutor conditionTriggerExecutor; @Resource private MqttRemoteManager remoteManager; @@ -97,15 +100,28 @@ public class DataHandlerImpl implements IDataHandler { ThingsModelValuesInput input = new ThingsModelValuesInput(); input.setProductId(bo.getProductId()); // 这里上报设备编号是转的大写，后面存缓存也是使用大写的，所以在查询物模型的值时添加把设备编号转大写后取值 - input.setDeviceNumber(bo.getSerialNumber().toUpperCase()); + String serialNumber = bo.getSerialNumber().toUpperCase(); + input.setDeviceNumber(serialNumber); input.setThingsModelValueRemarkItem(thingsModelSimpleItems); List result = deviceService.reportDeviceThingsModelValue(input, bo.getType(), bo.isShadow()); // todo 只有设备上报进入规则引擎流程 if (bo.isRuleEngine() && !bo.getSerialNumber().startsWith("server-")) { + // 原有规则引擎（场景联动） ruleEngine.ruleMatch(bo); // 条件触发任务检查：根据本次上报值决定是否下发 Modbus 指令 conditionTriggerExecutor.checkAndTrigger(bo.getSerialNumber(), result); + // 数据桥接规则引擎（HTTP推送 / MQTT桥接 / 数据库存储） + if (ServerType.TCP.getCode().equals(bo.getServerType()) && !CollectionUtils.isEmpty(result)) { + try { + // payload 与 MQTT 上报格式一致：ThingsModelSimpleItem JSON 数组（含计算物模型） + String bridgePayload = JSON.toJSONString(result); + ruleProcess.processRuleScript(serialNumber, 1, "tcp/property/post", bridgePayload); + } catch (Exception e) { + log.error("数据桥接规则引擎执行异常 sn={}, msg={}", serialNumber, e.getMessage(), e); + } + } } + // 发送至前端 PushMessageBo messageBo = new PushMessageBo(); messageBo.setTopic(topicsUtils.buildTopic(bo.getProductId(), bo.getSerialNumber(), TopicType.WS_SERVICE_INVOKE)); @@ -135,7 +151,6 @@ public class DataHandlerImpl implements IDataHandler { } } - /** * 上报事件 * @@ -204,7 +219,7 @@ public class DataHandlerImpl implements IDataHandler { if (null == sceneModelTag) { return "场景运算型变量计算错误：变量为空"; } - if ((StringUtils.isEmpty(sceneModelTag.getAliasFormule()) && org.apache.commons.collections4.CollectionUtils.isEmpty(sceneModelTag.getTagPointsList()))) { + if ((StringUtils.isEmpty(sceneModelTag.getAliasFormule()) && CollectionUtils.isEmpty(sceneModelTag.getTagPointsList()))) { return "场景运算型变量计算错误：没有计算公式"; } String checkMsg = sceneModelTagService.checkAliasFormule(sceneModelTag); @@ -268,7 +283,7 @@ public class DataHandlerImpl implements IDataHandler { * @param sceneModelTag 变量类 * @return void * @param: value 值 - * @param: now 执行时间 + * @param: now 执行时间 */ private void saveSceneModelTagValue(SceneModelTag sceneModelTag, String value, LocalDateTime now) { // 保存运算型变量值,存缓存 @@ -289,7 +304,7 @@ public class DataHandlerImpl implements IDataHandler { deviceLog.setCreateTime(new Date()); logService.saveDeviceLog(deviceLog); } - //发送至前端 + // 发送至前端 List sendMsg = new ArrayList<>(); sendMsg.add(sceneModelTagCacheVO); PushMessageBo messageBo = new PushMessageBo(); diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceReportMessageServiceImpl.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceReportMessageServiceImpl.java index 6ef55528..f0bd59f2 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceReportMessageServiceImpl.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DeviceReportMessageServiceImpl.java @@ -298,6 +298,7 @@ public class DeviceReportMessageServiceImpl implements IDeviceReportMessageServi report.setGwDeviceBo(message.getGwDeviceBo()); // 属性上报执行规则引擎 report.setRuleEngine(true); + report.setServerType(message.getServerType() == null ? "" : message.getServerType().getCode()); dataHandler.reportData(report); } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/ruleEngine/MessageArrivedHandler.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/ruleEngine/MessageArrivedHandler.java index 22f1ec0a..b52eb4d0 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/ruleEngine/MessageArrivedHandler.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/ruleEngine/MessageArrivedHandler.java @@ -22,7 +22,7 @@ class MessageArrivedHandler implements IMqttMessageListener { @Override public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception { - log.info("Message arrived on topic '" + topic + "': " + new String(mqttMessage.getPayload())); + log.info("Message arrived on topic '{}': {}", topic, new String(mqttMessage.getPayload())); if (this.topic.equals(topic)) { // 处理特定主题的消息 ScriptCondition scriptCondition = ScriptCondition.builder() diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/ruleEngine/MqttClientFactory.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/ruleEngine/MqttClientFactory.java index ff4db523..d6bd83b9 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/ruleEngine/MqttClientFactory.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/ruleEngine/MqttClientFactory.java @@ -57,7 +57,7 @@ public class MqttClientFactory { token.waitForCompletion(); CLIENT_OPTIONS_CACHE.put(config.getKey(), options); } catch (MqttException e) { - log.error("=>mqtt客户端创建错误，原因：" + e.getMessage()); + log.error("=>mqtt客户端创建错误，原因：{}", e.getMessage()); } return client; } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/ruleEngine/RuleProcess.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/ruleEngine/RuleProcess.java index 767dd374..c82b951c 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/ruleEngine/RuleProcess.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/ruleEngine/RuleProcess.java @@ -57,7 +57,7 @@ public class RuleProcess { .payload(payload) .topic(topic) .build(); - //返回处理完的消息上下文 + // 返回处理完的消息上下文 return scriptService.execRuleScript(scriptCondition, context); } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/ScriptServiceImpl.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/ScriptServiceImpl.java index 4caef218..9e99943f 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/ScriptServiceImpl.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/ScriptServiceImpl.java @@ -242,7 +242,7 @@ public class ScriptServiceImpl implements IScriptService { @Override public AjaxResult validateScript(Script ruleScript) { // 检查安全性检查 - String pattern = ".*while|for\\s*\\(|InputStream|OutputStream|Reader|Writer|File|Socket.*"; + String pattern = ".*while\\s*\\(|InputStream|OutputStream|Reader|Writer|File|Socket.*"; Pattern r = Pattern.compile(pattern); Matcher m = r.matcher(ruleScript.getScriptData()); if (m.find()) { @@ -278,7 +278,7 @@ public class ScriptServiceImpl implements IScriptService { @Override public MsgContext execRuleScript(ScriptCondition scriptCondition, Object... contextBeanArray) { List