package cn.estsh.i3plus.ext.mes.pcn.apiservice.mqtt;

import cn.estsh.i3plus.ext.mes.pcn.api.base.IMesEquipmentLogService;
import cn.estsh.i3plus.ext.mes.pcn.apiservice.config.MqttConfig;
import cn.estsh.i3plus.ext.mes.pcn.apiservice.util.SpringUtils;
import cn.estsh.i3plus.ext.mes.pcn.pojo.mqtt.EquipLogMqttMsg;
import cn.estsh.i3plus.pojo.mes.bean.MesEquipmentVariable;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Slf4j
public class PcnMqttCallback implements MqttCallbackExtended {

    private static ExecutorService executorService = new ThreadPoolExecutor(100, 100,
            0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<>(10000), r -> {
        Thread thread = new Thread(r);
        thread.setName("executorService--"+r.hashCode());
        return thread;
    },new ThreadPoolExecutor.DiscardPolicy());

    private static ExecutorService executorServiceTwo = new ThreadPoolExecutor(100,100,
            0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<>(10000), r -> {
        Thread thread = new Thread(r);
        thread.setName("executorServiceTwo--"+r.hashCode());
        return thread;
    },new ThreadPoolExecutor.DiscardPolicy());

    //手动注入
    private MqttConfig mqttConfig = SpringUtils.getBean(MqttConfig.class);


    private PcnMqttClient myMqttClient;

    private IMesEquipmentLogService equipmentLogService;

    public PcnMqttCallback(PcnMqttClient myMqttClient) {
        this.myMqttClient = myMqttClient;
    }


    public PcnMqttCallback(PcnMqttClient myMqttClient, IMesEquipmentLogService equipmentLogService) {
        this.myMqttClient = myMqttClient;
        this.equipmentLogService = equipmentLogService;
    }
    /**
     * MQTT Broker连接成功时被调用的方法。在该方法中可以执行 订阅系统约定的主题（推荐使用）。
     * 如果 MQTT Broker断开连接之后又重新连接成功时，主题也需要再次订阅，将重新订阅主题放在连接成功后的回调方法中比较合理。
     *
     * @param reconnect
     * @param serverURI MQTT Broker的url
     */
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        String connectMode = reconnect ? "重连" : "直连";
        log.info("== MyMqttCallback ==> MQTT 连接成功，连接方式：{}，serverURI：{}", connectMode, serverURI);
        //订阅主题
        //myMqttClient.subscribe(mqttConfig.topic, 1);
    }


    /**
     * 丢失连接，可在这里做重连
     * 只会调用一次
     *
     * @param throwable
     */
    @Override
    public void connectionLost(Throwable throwable) {
        log.error("== MyMqttCallback ==> connectionLost 连接断开，5S之后尝试重连: {}", throwable.getMessage());
        long reconnectTimes = 1;
        while (true) {
            try {
                if (PcnMqttClient.getClient().isConnected()) {
                    //判断已经重新连接成功  需要重新订阅主题 可以在这个if里面订阅主题  或者 connectComplete（方法里面）  看你们自己选择
                    log.warn("== MyMqttCallback ==> mqtt reconnect success end  重新连接  重新订阅成功");
                    return;
                }
                reconnectTimes += 1;
                log.warn("== MyMqttCallback ==> mqtt reconnect times = {} try again...  mqtt重新连接时间 {}", reconnectTimes, reconnectTimes);
                PcnMqttClient.getClient().reconnect();
            } catch (MqttException e) {
                log.error("== MyMqttCallback ==> mqtt断连异常", e);
            }
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e1) {
            }
        }
    }

    /**
     * 接收到消息（subscribe订阅的主题消息）时被调用的方法
     *
     * @param topic
     * @param mqttMessage
     * @throws Exception 后得到的消息会执行到这里面
     */
    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage)  {
        log.info("== pcnMqttCallback ==> messageArrived 接收消息主题: {}，接收消息内容: {}", topic, new String(mqttMessage.getPayload()));
        executorService.execute(() -> saveData(topic, mqttMessage));
    }

    private void saveData(String topic, MqttMessage mqttMessage) {
        log.info("== pcnMqttCallback ==> messageArrived 接收消息主题: {}，异步处理开始 消息内容: {}", topic, new String(mqttMessage.getPayload()));

        try{
            String resStr = new String(mqttMessage.getPayload(), "UTF-8").trim();
            /**
             * 根据订阅的主题分别处理业务。可以通过if-else或者策略模式来分别处理不同的主题消息。
             */
            //topic1主题
            List<EquipLogMqttMsg> equipLogMqttMsgList = JSONObject.parseArray(resStr, EquipLogMqttMsg.class);
            for (EquipLogMqttMsg equipLogMqttMsg : equipLogMqttMsgList) {
                executorServiceTwo.execute(()-> equipmentLogService.updateValue(equipLogMqttMsg));
            }
            //业务处理
            //doSomething1(maps);
            log.info("== pcnMqttCallback ==> messageArrived 接收消息主题: {}，业务处理消息内容完成", topic);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 消息发送（publish）完成时被调用的方法
     *
     * @param iMqttDeliveryToken
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        log.info("== MyMqttCallback ==> deliveryComplete 消息发送完成，Complete= {}", iMqttDeliveryToken.isComplete());
    }

}
