Merge remote-tracking branch 'origin/developer' into developer

amos-boot-system-jxiop/amos-boot-module-jxiop-analyse-biz/src/main/java/com/yeejoin/amos/boot/module/jxiop/biz/emqx/WarningRecordStatusMessage.java

+package com.yeejoin.amos.boot.module.jxiop.biz.emqx;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
+import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
+import com.baomidou.mybatisplus.core.toolkit.ObjectUtils;
+import com.yeejoin.amos.boot.module.jxiop.biz.entity.IdxBizFanWarningRecord;
+import com.yeejoin.amos.boot.module.jxiop.biz.entity.IdxBizPvWarningRecord;
+import com.yeejoin.amos.boot.module.jxiop.biz.service.impl.IdxBizFanWarningRecordServiceImpl;
+import com.yeejoin.amos.boot.module.jxiop.biz.service.impl.IdxBizPvWarningRecordServiceImpl;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+import org.typroject.tyboot.component.emq.EmqKeeper;
+import org.typroject.tyboot.component.emq.EmqxListener;
+
+import javax.annotation.PostConstruct;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+@Component
+@Slf4j
+public class WarningRecordStatusMessage extends EmqxListener {
+
+    @Autowired
+    protected EmqKeeper emqKeeper;
+
+    // 江西电建接收红黄绿码主题
+    private static final String QUESTION_STATUS_CHANGE = "question/status/change";
+
+    private static final BlockingQueue<JSONObject> blockingQueue = new LinkedBlockingQueue<JSONObject>();
+
+    @Autowired
+    private IdxBizFanWarningRecordServiceImpl idxBizFanWarningRecordService;
+
+    @Autowired
+    private IdxBizPvWarningRecordServiceImpl idxBizPvWarningRecordService;
+
+    @PostConstruct
+    void init() throws Exception {
+        new Thread(taskRunnable).start();
+        emqKeeper.subscript(QUESTION_STATUS_CHANGE, 2, this);
+    }
+
+    @Override
+    public void processMessage(String topic, MqttMessage message) throws Exception {
+        log.info("人员赋码消息{}", new String(message.getPayload()));
+        JSONObject ja = JSON.parseObject(new String(message.getPayload()));
+        blockingQueue.add(ja);
+    }
+
+    Runnable taskRunnable = new Runnable() {
+        @Override
+        public void run() {
+            boolean isRun = true;
+            int k = 0;
+            while (isRun) {
+                k++;
+                isRun = k < Integer.MAX_VALUE;
+                try {
+                    JSONObject analysisResult = blockingQueue.take();
+                    jxIopUpdate(analysisResult);
+                } catch (Exception e) {
+                    e.printStackTrace();
+                }
+            }
+        }
+    };
+
+    public void jxIopUpdate(JSONObject analysisResult) {
+        log.info("修改预警状态信息：{}", analysisResult);
+        if (ObjectUtils.isNotEmpty(analysisResult) && analysisResult.get("warningObjectType").toString().equals("pv")) {
+            LambdaUpdateWrapper<IdxBizFanWarningRecord> lambda = new LambdaUpdateWrapper<>();
+            lambda.set(IdxBizFanWarningRecord::getDisposotionState, "已处置");
+            lambda.eq(IdxBizFanWarningRecord::getSequenceNbr, analysisResult.get("objectId"));
+            idxBizFanWarningRecordService.update(lambda);
+        } else if (ObjectUtils.isNotEmpty(analysisResult) && analysisResult.get("warningObjectType").toString().equals("fan")) {
+            LambdaUpdateWrapper<IdxBizPvWarningRecord> lambda = new LambdaUpdateWrapper<>();
+            lambda.set(IdxBizPvWarningRecord::getDisposotionState, "已处置");
+            lambda.eq(IdxBizPvWarningRecord::getSequenceNbr, analysisResult.get("objectId"));
+            idxBizPvWarningRecordService.update(lambda);
+        }
+    }
+}

amos-boot-system-jxiop/amos-boot-module-jxiop-analyse-biz/src/main/java/com/yeejoin/amos/boot/module/jxiop/biz/emqx/WarningRecordStatusMessage2.java

+package com.yeejoin.amos.boot.module.jxiop.biz.emqx;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
+import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
+import com.baomidou.mybatisplus.core.toolkit.ObjectUtils;
+import com.yeejoin.amos.boot.module.jxiop.biz.entity.IdxBizFanWarningRecord;
+import com.yeejoin.amos.boot.module.jxiop.biz.entity.IdxBizPvWarningRecord;
+import com.yeejoin.amos.boot.module.jxiop.biz.service.impl.IdxBizFanWarningRecordServiceImpl;
+import com.yeejoin.amos.boot.module.jxiop.biz.service.impl.IdxBizPvWarningRecordServiceImpl;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+import org.typroject.tyboot.component.emq.EmqKeeper;
+import org.typroject.tyboot.component.emq.EmqxListener;
+
+import javax.annotation.PostConstruct;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+/**
+ * 处理无需处置情况
+ */
+@Component
+@Slf4j
+public class WarningRecordStatusMessage2 extends EmqxListener {
+
+    @Autowired
+    protected EmqKeeper emqKeeper;
+
+    /**
+     * 无需处置
+     */
+    public static final String  NOT_DISPOSE_AMOS = "not/dispose/amos";
+
+    private static final BlockingQueue<JSONObject> blockingQueue = new LinkedBlockingQueue<JSONObject>();
+
+    @Autowired
+    private IdxBizFanWarningRecordServiceImpl idxBizFanWarningRecordService;
+
+    @Autowired
+    private IdxBizPvWarningRecordServiceImpl idxBizPvWarningRecordService;
+
+    @PostConstruct
+    void init() throws Exception {
+        new Thread(taskRunnable).start();
+        emqKeeper.subscript(NOT_DISPOSE_AMOS, 2, this);
+    }
+
+    @Override
+    public void processMessage(String topic, MqttMessage message) throws Exception {
+        log.info("修改预警状态消息{}", new String(message.getPayload()));
+        JSONObject ja = JSON.parseObject(new String(message.getPayload()));
+        blockingQueue.add(ja);
+    }
+
+    Runnable taskRunnable = new Runnable() {
+        @Override
+        public void run() {
+            boolean isRun = true;
+            int k = 0;
+            while (isRun) {
+                k++;
+                isRun = k < Integer.MAX_VALUE;
+                try {
+                    JSONObject analysisResult = blockingQueue.take();
+                    jxIopUpdate(analysisResult);
+                } catch (Exception e) {
+                    e.printStackTrace();
+                }
+            }
+        }
+    };
+
+    public void jxIopUpdate(JSONObject analysisResult) {
+        log.info("修改预警状态信息：{}", analysisResult);
+        if (ObjectUtils.isNotEmpty(analysisResult) && analysisResult.get("warningObjectType").toString().equals("pv")) {
+            LambdaUpdateWrapper<IdxBizFanWarningRecord> lambda = new LambdaUpdateWrapper<>();
+            lambda.set(IdxBizFanWarningRecord::getDisposotionState, "已处置");
+            List<String> traceIds = (List<String>) analysisResult.get("traceIds");
+            lambda.in(IdxBizFanWarningRecord::getSequenceNbr, traceIds);
+            idxBizFanWarningRecordService.update(lambda);
+        } else if (ObjectUtils.isNotEmpty(analysisResult) && analysisResult.get("warningObjectType").toString().equals("fan")) {
+            LambdaUpdateWrapper<IdxBizPvWarningRecord> lambda = new LambdaUpdateWrapper<>();
+            lambda.set(IdxBizPvWarningRecord::getDisposotionState, "已处置");
+            List<String> traceIds = (List<String>) analysisResult.get("traceIds");
+            lambda.eq(IdxBizPvWarningRecord::getSequenceNbr, traceIds);
+            idxBizPvWarningRecordService.update(lambda);
+        }
+    }
+}