package com.ruoyi.dj.mqtt.client;

import com.ruoyi.dj.mqtt.common.CommonFieldsVo;
import com.ruoyi.dj.mqtt.config.DJIMqttConfig;
import com.ruoyi.dj.mqtt.domain.DrcOperVo;
import com.ruoyi.dj.mqtt.domain.DrcVo;
import com.ruoyi.dj.mqtt.service.MqttMessageService;
import com.ruoyi.dj.mqtt.utils.PayloadUtil;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;

/**
 * MQTT客户端实现类，负责与DJI云进行MQTT通信
 */
@Slf4j
@Component
public class DJICloudMQTTClient implements MqttCallback {

    @Autowired
    private DJIMqttConfig djiMqttConfig;

    @Autowired
    private MqttMessageService mqttMessageService;

    private MqttClient mqttClient;

    /**
     * 用于存储响应的Future，键为请求的公共字段，值为对应的Future
     */
    private final Map<CommonFieldsVo, CompletableFuture<String>> responseFutures = new ConcurrentHashMap<>();
    private final Map<DrcVo, CompletableFuture<String>> drcResponseFutures = new ConcurrentHashMap<>();
    private final Map<DrcOperVo, CompletableFuture<String>> drcOperVoCompletableFutureConcurrentHashMap = new ConcurrentHashMap<>();

    /**
     * 初始化客户端并连接到MQTT服务器
     *
     * @throws MqttException 如果连接失败
     */
    @PostConstruct
    public void connect() throws MqttException {
        // 创建MQTT客户端并设置回调
//        String uniqueClientId = UUID.randomUUID().toString();
        mqttClient = new MqttClient(djiMqttConfig.getBroker(), "4363452132146456346");
//        mqttClient = new MqttClient(djiMqttConfig.getBroker(), uniqueClientId);
        mqttClient.setCallback(this);

        // 设置连接选项，包括清空会话、用户名和密码
        MqttConnectOptions connOpts = new MqttConnectOptions();
        connOpts.setCleanSession(false);
        connOpts.setUserName(djiMqttConfig.getUsername());
        connOpts.setPassword(djiMqttConfig.getPassword().toCharArray());
        connOpts.setAutomaticReconnect(true);  // 启用自动重连

        // 连接到MQTT服务器并订阅指定的主题
        log.info("连接到服务器: " + djiMqttConfig.getBroker());
        mqttClient.connect(connOpts);
        for (String topic : djiMqttConfig.getTopics()) {
            mqttClient.subscribe(topic);
            log.info("订阅主题: " + topic);
        }
    }

    /**
     * 向指定主题发布消息，并等待响应
     *
     * @param topic          消息发布的主题
     * @param payload        消息的内容
     * @param commonFieldsVo 请求的公共字段，用于匹配响应
     * @return 响应的消息内容
     * @throws Exception 如果发布消息或等待响应失败
     */
    public String publishMessage(String topic, String payload, CommonFieldsVo commonFieldsVo) {
        try {
            CompletableFuture<String> future = new CompletableFuture<>();
            // 把网关置空、因为有的响应是不带网关这个字段的所以都不需要这个字段来做判断,否则代码会一直阻塞等待
            commonFieldsVo.setGateway(null);
            responseFutures.put(commonFieldsVo, future);

            // 设置消息质量等级为2，并发布消息
            MqttMessage message = new MqttMessage(payload.getBytes());
            message.setQos(2);
            mqttClient.publish(topic, message);
            log.info("消息已发送到主题: " + topic);

            // 等待响应
            return future.get();
        } catch (MqttException e) {
            throw new RuntimeException(e);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (ExecutionException e) {
            throw new RuntimeException(e);
        }
    }

    public String publishMessageWithDrc(String topic, String payload, CommonFieldsVo commonFieldsVo) {
        try {
            DrcVo drcVo = new DrcVo();
            drcVo.setMethod(commonFieldsVo.getMethod());
            CompletableFuture<String> future = new CompletableFuture<>();
            drcResponseFutures.put(drcVo, future);

            // 设置消息质量等级为2，并发布消息
            MqttMessage message = new MqttMessage(payload.getBytes());
            message.setQos(2);
            mqttClient.publish(topic, message);
            log.info("消息已发送到主题: " + topic);

            // 等待响应
            return future.get();
        } catch (MqttException e) {
            throw new RuntimeException(e);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (ExecutionException e) {
            throw new RuntimeException(e);
        }
    }

    public String publishMessageWithDrcOper(String topic, String payload, CommonFieldsVo commonFieldsVo) {
        try {
            DrcOperVo drcOperVo = new DrcOperVo();
            drcOperVo.setMethod(commonFieldsVo.getMethod());
            // 从 payload 的 json 字符串内截取 data 下的 seq 字段、用于匹配响应
            Long seqFromJson = PayloadUtil.getSeqFromJson(payload);
            drcOperVo.setSeq(seqFromJson);
            CompletableFuture<String> future = new CompletableFuture<>();
            drcOperVoCompletableFutureConcurrentHashMap.put(drcOperVo, future);

            // 设置消息质量等级为2，并发布消息
            MqttMessage message = new MqttMessage(payload.getBytes());
            message.setQos(2);
            mqttClient.publish(topic, message);
            log.info("消息已发送到主题: " + topic);
            // 等待响应
            return future.get();
        } catch (MqttException e) {
            throw new RuntimeException(e);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (ExecutionException e) {
            throw new RuntimeException(e);
        }
    }


    public void publishMessage(String topic, String payload) {
        try {
            log.info("发送消息到主题: " + topic);
            log.info("发生消息的内容: " + payload);

            // 设置消息质量等级为2，并发布消息
            MqttMessage message = new MqttMessage(payload.getBytes());
            message.setQos(2);
            mqttClient.publish(topic, message);
            log.info("消息已发送到主题: " + topic);
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }


    /**
     * 当MQTT连接丢失时调用
     *
     * @param cause 连接丢失的原因
     */
    @Override
    public void connectionLost(Throwable cause) {
        log.info("连接丢失! 原因: " + cause.getMessage());
    }

    /**
     * 当收到消息时调用
     *
     * @param topic   收到消息的主题
     * @param message 收到的消息
     * @throws Exception 如果处理消息失败
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        String payload = new String(message.getPayload());
        mqttMessageService.handleMessage(topic, payload);
    }

    /**
     * 当消息成功交付给服务器时调用
     *
     * @param token 消息交付的令牌
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        log.info("消息传递完成.");
    }

    /**
     * 根据公共字段获取对应的响应Future
     *
     * @param commonFieldsVo 公共字段
     * @return 响应的Future
     */
    public CompletableFuture<String> getResponseFuture(CommonFieldsVo commonFieldsVo) {
        return responseFutures.get(commonFieldsVo);
    }


    public CompletableFuture<String> getDrcResponseFutures(DrcVo drcVo) {
        return drcResponseFutures.get(drcVo);
    }

    /**
     * 完成指定公共字段的响应Future，并从map中移除
     *
     * @param commonFieldsVo 公共字段
     * @param response       响应内容
     */
    public void completeResponseFuture(CommonFieldsVo commonFieldsVo, String response) {
        CompletableFuture<String> future = responseFutures.get(commonFieldsVo);
        if (future != null) {
            future.complete(response);
            responseFutures.remove(commonFieldsVo);
        }
    }

    public void completeDrcResponseFuture(DrcVo drcVo, String response) {
        CompletableFuture<String> future = drcResponseFutures.get(drcVo);
        if (future != null) {
            future.complete(response);
            drcResponseFutures.remove(drcVo);
        }
    }

    /**
     * 根据 DrcOperVo 获取对应的响应 Future
     *
     * @param drcOperVo DrcOperVo 对象
     * @return 响应的 Future
     */
    public CompletableFuture<String> getDrcOperVoResponseFuture(DrcOperVo drcOperVo) {
        return drcOperVoCompletableFutureConcurrentHashMap.get(drcOperVo);
    }

    /**
     * 完成指定 DrcOperVo 的响应 Future，并从 map 中移除
     *
     * @param drcOperVo DrcOperVo 对象
     * @param response  响应内容
     */
    public void completeDrcOperVoResponseFuture(DrcOperVo drcOperVo, String response) {
        CompletableFuture<String> future = drcOperVoCompletableFutureConcurrentHashMap.get(drcOperVo);
        if (future != null) {
            future.complete(response);
            drcOperVoCompletableFutureConcurrentHashMap.remove(drcOperVo);
        }
    }

}
