package com.yeejoin.amos.boot.module.jxiop.biz.emqx;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
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.HealthStatusIndicatorServiceImpl;
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.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.stream.Collectors;

@Component
@Slf4j
public class WarningRecordStatusMessage extends EmqxListener {

    @Autowired
    protected EmqKeeper emqKeeper;
    /**
     * 预警状态修改消息 - 标准化
     */
    public static final String WARNING_CHANGE_MESSAGE = "+/warning/change";


    private static final BlockingQueue<JSONArray> blockingQueueFan = new LinkedBlockingQueue<JSONArray>();

    private static final BlockingQueue<JSONArray> blockingQueuePv = new LinkedBlockingQueue<JSONArray>();

    @Autowired
    private IdxBizFanWarningRecordServiceImpl idxBizFanWarningRecordService;

    @Autowired
    private IdxBizPvWarningRecordServiceImpl idxBizPvWarningRecordService;

    @PostConstruct
    void init() throws Exception {
        new Thread(taskRunnable).start();
        emqKeeper.subscript(WARNING_CHANGE_MESSAGE, 2, this);
    }

    @Override
    public void processMessage(String topic, MqttMessage message) throws Exception {
        if (topic.contains(HealthStatusIndicatorServiceImpl.SMART_ANALYSE_PV)) {
            log.info("预警状态改变消息-光伏{}", new String(message.getPayload()));
            JSONArray ja = JSON.parseArray(new String(message.getPayload()));
            blockingQueuePv.add(ja);
        } else if (topic.contains(HealthStatusIndicatorServiceImpl.SMART_ANALYSE_FAN)) {
            log.info("预警状态改变消息-风电{}", new String(message.getPayload()));
            JSONArray ja = JSON.parseArray(new String(message.getPayload()));
            blockingQueueFan.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 {
                    JSONArray analysisResultFan = blockingQueueFan.take();
                    jxIopUpdateFan(analysisResultFan);
                    JSONArray analysisResultPv = blockingQueuePv.take();
                    jxIopUpdatePv(analysisResultPv);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    };

    public void jxIopUpdateFan(JSONArray analysisResult) {
        log.info("修改预警状态信息：{}", analysisResult);
        List<JSONObject> taskList = JSONObject.parseArray(analysisResult.toJSONString(), JSONObject.class);
        List<String> traceIds = taskList.stream().map(t -> t.get("extAttr1").toString()).collect(Collectors.toList());
        LambdaUpdateWrapper<IdxBizFanWarningRecord> lambda = new LambdaUpdateWrapper<>();
        lambda.set(IdxBizFanWarningRecord::getDisposotionState, "已处置");
        lambda.set(IdxBizFanWarningRecord::getStatus, "1");
        lambda.set(IdxBizFanWarningRecord::getDisposotionDate, new Date());
        lambda.in(IdxBizFanWarningRecord::getSequenceNbr, traceIds);
        idxBizFanWarningRecordService.update(lambda);

    }

    public void jxIopUpdatePv(JSONArray analysisResult) {
        log.info("修改预警状态信息：{}", analysisResult);
        List<JSONObject> taskList = JSONObject.parseArray(analysisResult.toJSONString(), JSONObject.class);
        List<String> traceIds = taskList.stream().map(t -> t.get("extAttr1").toString()).collect(Collectors.toList());
        LambdaUpdateWrapper<IdxBizPvWarningRecord> lambda = new LambdaUpdateWrapper<>();
        lambda.set(IdxBizPvWarningRecord::getDisposotionState, "已处置");
        lambda.set(IdxBizPvWarningRecord::getStatus, "1");
        lambda.set(IdxBizPvWarningRecord::getDisposotionDate, new Date());
        lambda.in(IdxBizPvWarningRecord::getSequenceNbr, traceIds);
        idxBizPvWarningRecordService.update(lambda);
    }
}
