kafka接消息逻辑中处理稳压泵

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

@@ -622,6 +622,16 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
             TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
                 @Override
                 public void afterCommit() {
+
+                    iotDatalist.forEach(iotDataVO -> {
+                        String indexKey = iotDataVO.getKey();
+                        String indexValue = iotDataVO.getValue().toString();
+                        // 稳压泵启停信号处理
+                        if (indexKey.equals(pressurePumpStart)) {
+                            pressurePump(indexKey, indexValue, iotDatalist, topicEntity);
+                        }
+                    });
+
                     equipmentSpecificAlarms.forEach(action -> {
                         if (AlarmStatusEnum.BJ.getCode() == action.getStatus()) {
                             alarmLogs.add(addEquipAlarmLogRecord(action));
@@ -900,6 +910,16 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                 TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
                     @Override
                     public void afterCommit() {
+
+                        iotDatalist.forEach(iotDataVO -> {
+                            String indexKey = iotDataVO.getKey();
+                            String indexValue = iotDataVO.getValue().toString();
+                            // 稳压泵启停信号处理
+                            if (indexKey.equals(pressurePumpStart)) {
+                                pressurePump(indexKey, indexValue, iotDatalist, topicEntity);
+                            }
+                        });
+
                         equipmentSpecificAlarms.forEach(action -> {
                             if (AlarmStatusEnum.BJ.getCode() == action.getStatus()) {
                                 alarmLogs.add(addEquipAlarmLogRecord(action));
@@ -2460,6 +2480,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 //    }
 
     private void pressurePump(String indexKey, String indexValue, List<IotDataVO> iotDatalist, TopicEntityVo topicEntity) {
+        log.info("开始处理稳压泵逻辑：{}值：{}", indexKey, indexValue);
         List<String> listIndex = new ArrayList<>();
         listIndex.add(pressurePumpStart);
         // 获取全部启停泵信号
@@ -2487,6 +2508,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                 PressurePumpValueEnum valueEnum = PressurePumpValueEnum.getByCode(pressurePumpEnum.getCompareValue());
                 assert valueEnum != null;
                 EquipmentSpecificIndex data = getPressurePumpDateByType(indexKey, valueEnum, topicEntity, equipmentSpeIndexList, pressurePumpEnum);
+                log.info("稳压泵获取{}， 值为{}", valueEnum.getDescribe(), data);
                 Date newDate = new Date();
                 // 2. 校验
                 if (!ObjectUtils.isEmpty(data.getUpdateDate())) {
@@ -2513,7 +2535,6 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
         } else {
             throw new BadRequest("装备物联编码错误，请确认！");
         }
-
         switch (valueEnum) {
             case LAST_STOP:
                 List<EquipmentSpecificIndex> lastStop = equipmentSpeIndexList.stream().filter(e ->
@@ -2613,6 +2634,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
         assert pumpCheckEnum != null;
         String leftValue = pressurePumpEnum.getLeftValue();
         String rightValue = pressurePumpEnum.getRightValue();
+        log.info("检验方式：{}，大于：{}，小于：{}， 间隔：{}", pumpCheckEnum.getDescribe(), leftValue, rightValue, diff);
         switch (pumpCheckEnum) {
             case LE:
                 if (StringUtil.isNotEmpty(rightValue)) {