feat: 组态页面数据推送逻辑优化

amos-boot-module/amos-boot-module-api/amos-boot-module-equip-api/src/main/java/com/yeejoin/equipmanage/common/enums/ConfigPageTopicEnum.java

@@ -15,13 +15,14 @@ import java.util.List;
 public enum ConfigPageTopicEnum {
 
     INTEGRATE("INTEGRATE_TOPIC/#", "集成页面统配主题"),
-    SYSTEMDETAIL("EQUIP_INDEX_ON_SYSTEM_DETAIL", "系统详情页面大屏初始化主题");
+    EQUIP_INDICATOR("EQUIP_INDEX_ON_SYSTEM_DETAIL", "系统详情页面测点初始化主题(单测点)"),
+    EQUIP_MULTI_INDICATOR("EQUIP_MULTI_INDEX_ON_SYSTEM_DETAIL", "系统详情页面测点初始化主题(设备绑定多测点)");
 
 	private String topic;
 
     private String describe;
-    
-    
+
+
     public static List<String> getEnumTopicList() {
         List<String> topics = new ArrayList<>();
         for (ConfigPageTopicEnum e : ConfigPageTopicEnum.values()) {
@@ -29,7 +30,7 @@ public enum ConfigPageTopicEnum {
        }
         return topics;
     }
-    
+
     public static boolean  isEqualsTopic(String topic) {
         for (ConfigPageTopicEnum e : ConfigPageTopicEnum.values()) {
         	if(e.getTopic().equals(topic)){

amos-boot-module/amos-boot-module-biz/amos-boot-module-equip-biz/src/main/java/com/yeejoin/equipmanage/listener/IntegratePageDataListener.java

@@ -8,13 +8,12 @@ import com.yeejoin.equipmanage.common.enums.ConfigPageTopicEnum;
 import com.yeejoin.equipmanage.service.IEquipmentSpecificIndexSerivce;
 import com.yeejoin.equipmanage.service.IEquipmentSpecificSerivce;
 import com.yeejoin.equipmanage.service.IFireFightingSystemService;
+import com.yeejoin.equipmanage.service.MqttSendGateway;
 import lombok.extern.slf4j.Slf4j;
-import org.eclipse.paho.client.mqttv3.MqttException;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 import org.springframework.util.ObjectUtils;
-import org.typroject.tyboot.component.emq.EmqKeeper;
 import org.typroject.tyboot.component.emq.EmqxListener;
 
 import java.util.*;
@@ -36,8 +35,12 @@ public class IntegratePageDataListener extends EmqxListener {
     @Autowired
     IEquipmentSpecificIndexSerivce equipmentSpecificIndexService;
 
+    static MqttSendGateway mqttSendGateway;
+
     @Autowired
-    EmqKeeper emqKeeper;
+    public void setMqttSendGateway(MqttSendGateway mqttSendGateway) {
+        IntegratePageDataListener.mqttSendGateway = mqttSendGateway;
+    }
 
     @Override
     public void processMessage(String topic, MqttMessage message) throws Exception {
@@ -51,32 +54,23 @@ public class IntegratePageDataListener extends EmqxListener {
                 String code = topic.substring(topic.indexOf(split) + 1);
                 fireFightingSystemService.integrationPageSysData(code, false);
             }
-        } else if (ConfigPageTopicEnum.SYSTEMDETAIL.getTopic().equalsIgnoreCase(topic)) {
+        } else if (ConfigPageTopicEnum.EQUIP_INDICATOR.getTopic().equalsIgnoreCase(topic)
+                || ConfigPageTopicEnum.EQUIP_MULTI_INDICATOR.getTopic().equalsIgnoreCase(topic)) {
             this.initializeIntegrationPageData(topic, msg);
         }
     }
 
     private void initializeIntegrationPageData(String topic, Map<String, Object> message) {
-        Object multiIndicator = message.get("multiIndicator");
         Object codes = message.get("codes");
 
-        if (ObjectUtils.isEmpty(codes)) {
-            return;
-        }
-
-        // 判断是否是设备绑定多测点
-        boolean isDeviceBindMultiIndicator = Objects.nonNull(multiIndicator) && Boolean.parseBoolean(String.valueOf(multiIndicator));
-
-        List<String> ids = JSON.parseArray(String.valueOf(codes), String.class);
-
         // 设备绑定多测点：codes为"equipment_specific_id"
-        if (isDeviceBindMultiIndicator) {
+        if (ConfigPageTopicEnum.EQUIP_MULTI_INDICATOR.getTopic().equalsIgnoreCase(topic) && Objects.nonNull(codes) && !ObjectUtils.isEmpty(codes)) {
+            List<String> equipIds = JSON.parseArray(String.valueOf(codes), String.class);
             List<EquipmentSpecificIndex> indices = equipmentSpecificIndexService.list(
                     Wrappers.<EquipmentSpecificIndex>lambdaQuery()
                             .select(EquipmentSpecificIndex::getEquipmentSpecificId, EquipmentSpecificIndex::getEquipmentIndexKey, EquipmentSpecificIndex::getValue)
-                            .in(EquipmentSpecificIndex::getEquipmentSpecificId, ids)
+                            .in(EquipmentSpecificIndex::getEquipmentSpecificId, equipIds)
                             .isNotNull(EquipmentSpecificIndex::getValue)
-                            .ne(EquipmentSpecificIndex::getValue, "false")
                             .ne(EquipmentSpecificIndex::getValue, "")
                             .orderByDesc(EquipmentSpecificIndex::getEmergencyLevel)
             );
@@ -85,26 +79,22 @@ public class IntegratePageDataListener extends EmqxListener {
             JSONObject emqMessage;
             for (Map.Entry<Long, List<EquipmentSpecificIndex>> entry : groupedIndices.entrySet()) {
                 Long equipmentSpecificId = entry.getKey();
-                List<EquipmentSpecificIndex> valuedIndexes = entry.getValue();
+                List<EquipmentSpecificIndex> specificIndexList = entry.getValue();
 
-                List<HashMap<String, String>> indexKey = valuedIndexes.stream().map(index -> new HashMap<String, String>() {{
+                List<HashMap<String, String>> valuedIndexes = specificIndexList.stream().map(index -> new HashMap<String, String>() {{
                     put("key", index.getEquipmentIndexKey());
                     put("value", index.getValue());
                 }}).collect(Collectors.toList());
                 emqMessage = new JSONObject() {{
                     this.put("code", String.valueOf(equipmentSpecificId));
-                    this.put("indexKey", indexKey);
+                    this.put("valuedIndexes", valuedIndexes);
                 }};
-
-                try {
-                    emqKeeper.getMqttClient().publish(topic, JSON.toJSONString(emqMessage).getBytes(), 1, false);
-                } catch (MqttException e) {
-                    log.error("发布消息失败", e);
-                }
+                mqttSendGateway.sendToMqtt(ConfigPageTopicEnum.EQUIP_MULTI_INDICATOR.getTopic(), emqMessage.toJSONString());
             }
         }
         // 单测点：codes为"equipment_specific_index_id"
-        else {
+        if (ConfigPageTopicEnum.EQUIP_INDICATOR.getTopic().equalsIgnoreCase(topic) && Objects.nonNull(codes) && !ObjectUtils.isEmpty(codes)) {
+            List<String> ids = JSON.parseArray(String.valueOf(codes), String.class);
             List<EquipmentSpecificIndex> indices = equipmentSpecificIndexService.list(
                     Wrappers.<EquipmentSpecificIndex>lambdaQuery()
                             .select(EquipmentSpecificIndex::getId, EquipmentSpecificIndex::getEquipmentIndexKey, EquipmentSpecificIndex::getValue)
@@ -118,12 +108,7 @@ public class IntegratePageDataListener extends EmqxListener {
                     this.put("value", index.getValue());
                     this.put("status", index.getValue());
                 }};
-
-                try {
-                    emqKeeper.getMqttClient().publish(topic, JSON.toJSONString(emqMessage).getBytes(), 1, false);
-                } catch (MqttException e) {
-                    log.error("发布消息失败", e);
-                }
+                mqttSendGateway.sendToMqtt(ConfigPageTopicEnum.EQUIP_INDICATOR.getTopic(), emqMessage.toJSONString());
             }
         }
     }

amos-boot-module/amos-boot-module-biz/amos-boot-module-equip-biz/src/main/java/com/yeejoin/equipmanage/service/impl/ApplicationRunnerImpl.java

@@ -63,6 +63,7 @@ public class ApplicationRunnerImpl implements ApplicationRunner {
 
         maintenanceResourceDataService.subscribeTopic();
         emqKeeper.subscript(ConfigPageTopicEnum.INTEGRATE.getTopic(), 2, integratePageDataListener);
-        emqKeeper.subscript(ConfigPageTopicEnum.SYSTEMDETAIL.getTopic(), 2, integratePageDataListener);
+        emqKeeper.subscript(ConfigPageTopicEnum.EQUIP_INDICATOR.getTopic(), 2, integratePageDataListener);
+        emqKeeper.subscript(ConfigPageTopicEnum.EQUIP_MULTI_INDICATOR.getTopic(), 2, integratePageDataListener);
     }
 }

amos-boot-module/amos-boot-module-biz/amos-boot-module-equip-biz/src/main/java/com/yeejoin/equipmanage/service/impl/MqttReceiveServiceImpl.java

@@ -2697,18 +2697,23 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                             .eq(EquipmentSpecificIndex::getEquipmentSpecificId, equipmentSpecificId)
                             .isNotNull(EquipmentSpecificIndex::getValue)
                             .ne(EquipmentSpecificIndex::getValue, "")
+                            .orderByDesc(EquipmentSpecificIndex::getEmergencyLevel)
             );
-            List<HashMap<String, String>> indexKey = equipmentSpecificIndices.stream().map(index -> new HashMap<String, String>() {{
+            List<HashMap<String, String>> valuedIndexes = equipmentSpecificIndices.stream().map(index -> new HashMap<String, String>() {{
                 put("key", index.getEquipmentIndexKey());
                 put("value", index.getValue());
             }}).collect(Collectors.toList());
-            JSONObject message = new JSONObject() {{
+            JSONObject message1 = new JSONObject() {{
                 put("code", String.valueOf(specificIndex.getId()));
                 put("status", specificIndex.getValue());
                 put("value", specificIndex.getValue());
-                put("indexKey", indexKey);
             }};
-            mqttSendGateway.sendToMqtt(ConfigPageTopicEnum.SYSTEMDETAIL.getTopic(), message.toJSONString());
+            JSONObject message2 = new JSONObject() {{
+                put("code", String.valueOf(specificIndex.getEquipmentSpecificId()));
+                put("valuedIndexes", valuedIndexes);
+            }};
+            mqttSendGateway.sendToMqtt(ConfigPageTopicEnum.EQUIP_INDICATOR.getTopic(), message1.toJSONString());
+            mqttSendGateway.sendToMqtt(ConfigPageTopicEnum.EQUIP_MULTI_INDICATOR.getTopic(), message2.toJSONString());
         }
     }
 }