20260426:数据桥接（数据库桥接）

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

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;
+
 }

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("接收消息主题 : {}", topic);
-        log.info("接收消息Qos : " + mqttMessage.getQos());
+        log.info("接收消息Qos : {}", mqttMessage.getQos());
-        log.info("接收消息内容 : " + message);
+        log.info("接收消息内容 : {}", message);
 
         //这里默认设备编号长度超过9位
         String[] split = topic.split("/");

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<ThingsModelSimpleItem> 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<SceneModelTagCacheVO> sendMsg = new ArrayList<>();
         sendMsg.add(sceneModelTagCacheVO);
         PushMessageBo messageBo = new PushMessageBo();

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);
     }
 

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()

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;
     }

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);
     }
 

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<Script> scripts = selectExecRuleScriptList(scriptCondition);
-        //如果查询不到脚本，则认为是不用处理
+        // 如果查询不到脚本，则认为是不用处理
         if (Objects.isNull(scripts) || scripts.isEmpty()) {
             return null;
         }
@@ -292,7 +292,7 @@ public class ScriptServiceImpl implements IScriptService {
             } else if (script.getScriptAction() == 4) {
                 el = "THEN(" + script.getScriptId() + ",mqttBridge" + ")";
             } else if (script.getScriptAction() == 5) {
-                el = "THEN(" + script.getScriptId() + ",mqttBridge" + ")";
+                el = "THEN(" + script.getScriptId() + ",databaseBridge" + ")";
             } else {
                 el = "THEN(" + script.getScriptId() + ")";
             }