emq问题修改

YeeAmosFireAutoSysService/src/main/java/com/yeejoin/amos/fas/business/service/impl/ContingencyPlanServiceImpl.java

@@ -29,6 +29,7 @@ import org.apache.commons.lang3.StringUtils;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.eclipse.paho.client.mqttv3.MqttException;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
 import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
@@ -41,6 +42,7 @@ import org.springframework.transaction.support.TransactionSynchronizationManager
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.ObjectUtils;
 import org.typroject.tyboot.component.emq.EmqKeeper;
+import org.typroject.tyboot.component.emq.EmqxListener;
 import org.typroject.tyboot.core.foundation.utils.Bean;
 import org.typroject.tyboot.core.foundation.utils.ValidationUtil;
 
@@ -799,50 +801,60 @@ public class ContingencyPlanServiceImpl implements IContingencyPlanService {
     @Override
     public void subscribeTopic() {
         try {
-            emqKeeper.getMqttClient().subscribe(DELETE_SYNC_PLAN_DOC, (s, mqttMessage) -> {
-                byte[] payload = mqttMessage.getPayload();
-                try {
-                    List<Long> ids = (List<Long>) ClazzUtils.deserializableObject(payload);
-                    if (!ValidationUtil.isEmpty(ids)) {
-                        planDocMapper.logicDeleteByDocIdList(ids);
-                    }
-                } catch (Exception e) {
-                    logger.error("预案文档删除同步出错", e);
+            emqKeeper.subscript(DELETE_SYNC_PLAN_DOC, 1, new EmqxListener() {
+                @Override
+                public void processMessage(String topic, MqttMessage message)  {
+
+                        byte[] payload = message.getPayload();
+                        try {
+                            List<Long> ids = (List<Long>) ClazzUtils.deserializableObject(payload);
+                            if (!ValidationUtil.isEmpty(ids)) {
+                                planDocMapper.logicDeleteByDocIdList(ids);
+                            }
+                        } catch (Exception e) {
+                            logger.error("预案文档删除同步出错", e);
+                        }
                 }
             });
-        } catch (MqttException e) {
+        } catch (Exception e) {
             logger.fatal("订阅文档删除同步消息失败，资源删除或取消无法同步", e);
         }
         try {
-            emqKeeper.getMqttClient().subscribe(DELETE_SYNC_PLAN_RULE, (s, mqttMessage) -> {
-                byte[] payload = mqttMessage.getPayload();
-                try {
-                    String[] ids = new String(payload).split(",");
-                    if (!ValidationUtil.isEmpty(ids)) {
-                        planRuleMapper.logicDeleteByRuleIdList(Arrays.asList(ids));
-                    }
-                } catch (Exception e) {
-                    logger.error("预案规则删除同步出错", e);
+            emqKeeper.subscript(DELETE_SYNC_PLAN_RULE, 1, new EmqxListener() {
+                @Override
+                public void processMessage(String topic, MqttMessage message)  {
+                        byte[] payload = message.getPayload();
+                        try {
+                            String[] ids = new String(payload).split(",");
+                            if (!ValidationUtil.isEmpty(ids)) {
+                                planRuleMapper.logicDeleteByRuleIdList(Arrays.asList(ids));
+                            }
+                        } catch (Exception e) {
+                            logger.error("预案规则删除同步出错", e);
+                        }
                 }
             });
-        } catch (MqttException e) {
+        } catch (Exception e) {
             logger.fatal("订阅规则删除同步消息失败，资源删除或取消无法同步", e);
         }
         try {
-            emqKeeper.getMqttClient().subscribe(VIEW_3D_OPEN_STATUS, (s, mqttMessage) -> {
-                Map msg = JSON.parseObject(mqttMessage.toString());
-                if (msg.containsKey("status") && msg.containsKey("seq")) {
-                    String seq = msg.get("seq").toString();
-                    String status = msg.get("status").toString();
-                    String key = KEY_VIEW_3D_PAGE + seq;
-                    if ("offLine".equals(status)) {
-                        redisTemplate.delete(key);
-                    } else {
-                        redisTemplate.opsForValue().set(key, status);
+            emqKeeper.subscript(VIEW_3D_OPEN_STATUS, 1, new EmqxListener() {
+                @Override
+                public void processMessage(String topic, MqttMessage message)  {
+                    Map msg = JSON.parseObject(message.toString());
+                    if (msg.containsKey("status") && msg.containsKey("seq")) {
+                        String seq = msg.get("seq").toString();
+                        String status = msg.get("status").toString();
+                        String key = KEY_VIEW_3D_PAGE + seq;
+                        if ("offLine".equals(status)) {
+                            redisTemplate.delete(key);
+                        } else {
+                            redisTemplate.opsForValue().set(key, status);
+                        }
                     }
                 }
             });
-        } catch (MqttException e) {
+        } catch (Exception e) {
             logger.fatal("订阅规则删除同步消息失败，资源删除或取消无法同步", e);
         }
     }