package com.dcs.equipment.utils;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;

/**
 * MQTT 工具类
 * 提供 MQTT 连接、发布、订阅等功能
 * 
 * @author: wangpx
 * @date: 2025-07-24 14:06
 */
public class MqttUtil {
    
    private static final Logger log = LoggerFactory.getLogger(MqttUtil.class);
    
    // 存储活跃的 MQTT 客户端连接，key 为 host:port 格式
    private static final ConcurrentHashMap<String, MqttClient> clientMap = new ConcurrentHashMap<>();
    
    // 默认配置
    private static final int DEFAULT_QOS = 1;
    private static final int DEFAULT_TIMEOUT = 30;
    private static final int DEFAULT_KEEP_ALIVE = 60;
    
    /**
     * 创建 MQTT 客户端连接
     * 
     * @param host MQTT 服务器地址
     * @param port MQTT 服务器端口
     * @param username 用户名
     * @param password 密码
     * @param clientId 客户端ID
     * @param qos 服务质量
     * @param timeout 连接超时时间
     * @param keepAliveInterval 保活间隔
     * @param cleanSession 是否清理会话
     * @return MqttClient 实例
     * @throws MqttException MQTT 异常
     */
    public static MqttClient createClient(String host, int port, String username, String password, 
                                        String clientId, int qos, int timeout, int keepAliveInterval, 
                                        boolean cleanSession) throws MqttException {
        String brokerKey = host + ":" + port;
        String broker = "tcp://" + brokerKey;
        
        // 检查是否已存在连接
        if (clientMap.containsKey(brokerKey)) {
            MqttClient existingClient = clientMap.get(brokerKey);
            if (existingClient.isConnected()) {
                log.info("Broker {} 已存在连接", brokerKey);
                return existingClient;
            } else {
                log.info("Broker {} 的连接已断开，移除旧连接", brokerKey);
                clientMap.remove(brokerKey);
            }
        }
        
        log.info("为 Broker {} 创建新连接", brokerKey);
        MqttClient client = new MqttClient(broker, clientId, new MemoryPersistence());
        
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(cleanSession);
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setConnectionTimeout(timeout);
        options.setKeepAliveInterval(keepAliveInterval);
        options.setAutomaticReconnect(true);
        
        client.connect(options);
        clientMap.put(brokerKey, client);
        
        log.info("MQTT 客户端连接成功 - Broker: {}, ClientId: {}", brokerKey, clientId);
        return client;
    }
    
    /**
     * 创建 MQTT 客户端连接（使用默认参数）
     * 
     * @param host MQTT 服务器地址
     * @param port MQTT 服务器端口
     * @param username 用户名
     * @param password 密码
     * @return MqttClient 实例
     * @throws MqttException MQTT 异常
     */
    public static MqttClient createClient(String host, int port, String username, String password) throws MqttException {
        String clientId = "java_client_" + System.currentTimeMillis();
        return createClient(host, port, username, password, clientId, DEFAULT_QOS, DEFAULT_TIMEOUT, DEFAULT_KEEP_ALIVE, true);
    }
    
    /**
     * 创建 MQTT 客户端连接（指定客户端ID）
     * 
     * @param host MQTT 服务器地址
     * @param port MQTT 服务器端口
     * @param username 用户名
     * @param password 密码
     * @param clientId 客户端ID
     * @return MqttClient 实例
     * @throws MqttException MQTT 异常
     */
    public static MqttClient createClient(String host, int port, String username, String password, String clientId) throws MqttException {
        return createClient(host, port, username, password, clientId, DEFAULT_QOS, DEFAULT_TIMEOUT, DEFAULT_KEEP_ALIVE, true);
    }
    
    /**
     * 获取或创建 MQTT 客户端连接
     * 
     * @param host MQTT 服务器地址
     * @param port MQTT 服务器端口
     * @param username 用户名
     * @param password 密码
     * @param clientId 客户端ID
     * @param qos 服务质量
     * @param timeout 连接超时时间
     * @param keepAliveInterval 保活间隔
     * @param cleanSession 是否清理会话
     * @return MqttClient 实例
     * @throws MqttException MQTT 异常
     */
    public static MqttClient getOrCreateClient(String host, int port, String username, String password, 
                                             String clientId, int qos, int timeout, int keepAliveInterval, 
                                             boolean cleanSession) throws MqttException {
        String brokerKey = host + ":" + port;
        
        // 如果连接不存在，创建新连接
        if (!clientMap.containsKey(brokerKey)) {
            return createClient(host, port, username, password, clientId, qos, timeout, keepAliveInterval, cleanSession);
        }
        
        MqttClient client = clientMap.get(brokerKey);
        if (!client.isConnected()) {
            log.info("Broker {} 的连接已断开，重新连接", brokerKey);
            clientMap.remove(brokerKey);
            return createClient(host, port, username, password, clientId, qos, timeout, keepAliveInterval, cleanSession);
        }
        
        return client;
    }
    
    /**
     * 获取或创建 MQTT 客户端连接（使用默认参数）
     * 
     * @param host MQTT 服务器地址
     * @param port MQTT 服务器端口
     * @param username 用户名
     * @param password 密码
     * @return MqttClient 实例
     * @throws MqttException MQTT 异常
     */
    public static MqttClient getOrCreateClient(String host, int port, String username, String password) throws MqttException {
        String clientId = "java_client_" + System.currentTimeMillis();
        return getOrCreateClient(host, port, username, password, clientId, DEFAULT_QOS, DEFAULT_TIMEOUT, DEFAULT_KEEP_ALIVE, true);
    }
    
    /**
     * 发布消息
     * 
     * @param host MQTT 服务器地址
     * @param port MQTT 服务器端口
     * @param username 用户名
     * @param password 密码
     * @param topic 主题
     * @param payload 消息内容
     * @param qos 服务质量
     * @throws MqttException MQTT 异常
     */
    public static void publish(String host, int port, String username, String password, 
                             String topic, String payload, int qos) throws MqttException {
        MqttClient client = getOrCreateClient(host, port, username, password);
        
        MqttMessage message = new MqttMessage(payload.getBytes());
        message.setQos(qos);
        client.publish(topic, message);
        
        String brokerKey = host + ":" + port;
        log.info("消息已发布 - Broker: {}, Topic: {}, Payload: {}", brokerKey, topic, payload);
    }
    
    /**
     * 发布消息（使用默认 QoS）
     * 
     * @param host MQTT 服务器地址
     * @param port MQTT 服务器端口
     * @param username 用户名
     * @param password 密码
     * @param topic 主题
     * @param payload 消息内容
     * @throws MqttException MQTT 异常
     */
    public static void publish(String host, int port, String username, String password, 
                             String topic, String payload) throws MqttException {
        publish(host, port, username, password, topic, payload, DEFAULT_QOS);
    }
    
    /**
     * 订阅主题
     * 
     * @param host MQTT 服务器地址
     * @param port MQTT 服务器端口
     * @param username 用户名
     * @param password 密码
     * @param topic 主题
     * @param qos 服务质量
     * @param messageHandler 消息处理器
     * @throws MqttException MQTT 异常
     */
    public static void subscribe(String host, int port, String username, String password, 
                               String topic, int qos, Consumer<String> messageHandler) throws MqttException {
        MqttClient client = getOrCreateClient(host, port, username, password);
        String brokerKey = host + ":" + port;
        
        // 设置回调
        client.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable cause) {
                log.error("MQTT 连接丢失 - Broker: {}, 原因: {}", 
                        brokerKey, cause != null ? cause.getMessage() : "未知原因");
            }
            
            @Override
            public void messageArrived(String topic, MqttMessage message) {
                String payload = new String(message.getPayload());
                log.info("接收到消息 - Broker: {}, Topic: {}, Content: {}", 
                        brokerKey, topic, payload);
                if (messageHandler != null) {
                    messageHandler.accept(payload);
                }
            }
            
            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                log.debug("消息发送完成 - Broker: {}", brokerKey);
            }
        });
        
        client.subscribe(topic, qos);
        log.info("已订阅主题 - Broker: {}, Topic: {}", brokerKey, topic);
    }
    
    /**
     * 订阅主题（使用默认 QoS）
     * 
     * @param host MQTT 服务器地址
     * @param port MQTT 服务器端口
     * @param username 用户名
     * @param password 密码
     * @param topic 主题
     * @param messageHandler 消息处理器
     * @throws MqttException MQTT 异常
     */
    public static void subscribe(String host, int port, String username, String password, 
                               String topic, Consumer<String> messageHandler) throws MqttException {
        subscribe(host, port, username, password, topic, DEFAULT_QOS, messageHandler);
    }
    
    /**
     * 发送消息并等待响应
     * 
     * @param host MQTT 服务器地址
     * @param port MQTT 服务器端口
     * @param username 用户名
     * @param password 密码
     * @param topic 主题
     * @param payload 发送的消息
     * @param timeout 超时时间（秒）
     * @return CompletableFuture<String> 响应结果
     */
    public static CompletableFuture<String> sendAndWaitResponse(String host, int port, String username, String password, 
                                                              String topic, String payload, int timeout) {
        CompletableFuture<String> future = new CompletableFuture<>();
        String brokerKey = host + ":" + port;
        
        try {
            // 获取或创建客户端连接
            MqttClient client = getOrCreateClient(host, port, username, password);
            
            // 订阅响应主题
            subscribe(host, port, username, password, topic, message -> {
                if (!future.isDone()) {
                    future.complete(message);
                }
            });
            
            // 发送消息
            publish(host, port, username, password, topic, payload);
            
            // 设置超时 - 兼容 Java 8
            CompletableFuture.supplyAsync(() -> {
                try {
                    return future.get(timeout, TimeUnit.SECONDS);
                } catch (Exception e) {
                    if (e instanceof TimeoutException) {
                        log.error("等待响应超时 - Broker: {}, Topic: {}, Timeout: {} 秒", 
                                brokerKey, topic, timeout);
                    } else {
                        log.error("等待响应时发生错误 - Broker: {}, Topic: {}, Error: {}", 
                                brokerKey, topic, e.getMessage());
                    }
                    throw new RuntimeException(e);
                }
            }).whenComplete((result, throwable) -> {
                if (throwable != null) {
                    log.error("操作失败 - Broker: {}, Topic: {}, Error: {}", 
                            brokerKey, topic, throwable.getMessage());
                }
                // 注意：这里不自动断开连接，保持连接复用
            });
            
        } catch (MqttException e) {
            log.error("MQTT 操作失败 - Broker: {}, Topic: {}, Error: {}", 
                    brokerKey, topic, e.getMessage());
            future.completeExceptionally(e);
        }
        
        return future;
    }
    
    /**
     * 发送消息并等待响应（使用默认超时时间）
     * 
     * @param host MQTT 服务器地址
     * @param port MQTT 服务器端口
     * @param username 用户名
     * @param password 密码
     * @param topic 主题
     * @param payload 发送的消息
     * @return CompletableFuture<String> 响应结果
     */
    public static CompletableFuture<String> sendAndWaitResponse(String host, int port, String username, String password, 
                                                              String topic, String payload) {
        return sendAndWaitResponse(host, port, username, password, topic, payload, DEFAULT_TIMEOUT);
    }
    
    /**
     * 断开指定 Broker 的连接
     * 
     * @param host MQTT 服务器地址
     * @param port MQTT 服务器端口
     */
    public static void disconnect(String host, int port) {
        String brokerKey = host + ":" + port;
        MqttClient client = clientMap.remove(brokerKey);
        if (client != null && client.isConnected()) {
            try {
                client.disconnect();
                client.close();
                log.info("MQTT 客户端已断开连接 - Broker: {}", brokerKey);
            } catch (MqttException e) {
                log.error("断开连接失败 - Broker: {}, Error: {}", brokerKey, e.getMessage());
            }
        }
    }
    
    /**
     * 断开所有连接
     */
    public static void disconnectAll() {
        clientMap.keySet().forEach(brokerKey -> {
            String[] parts = brokerKey.split(":");
            if (parts.length == 2) {
                disconnect(parts[0], Integer.parseInt(parts[1]));
            }
        });
    }
    
    /**
     * 检查指定 Broker 是否连接
     * 
     * @param host MQTT 服务器地址
     * @param port MQTT 服务器端口
     * @return 是否连接
     */
    public static boolean isConnected(String host, int port) {
        String brokerKey = host + ":" + port;
        MqttClient client = clientMap.get(brokerKey);
        return client != null && client.isConnected();
    }
    
    /**
     * 获取活跃连接数
     * 
     * @return 连接数
     */
    public static int getActiveConnectionCount() {
        return (int) clientMap.values().stream()
                .filter(MqttClient::isConnected)
                .count();
    }
    
    /**
     * 获取所有活跃的 Broker 连接信息
     * 
     * @return Broker key 列表
     */
    public static java.util.Set<String> getActiveBrokers() {
        return clientMap.entrySet().stream()
                .filter(entry -> entry.getValue().isConnected())
                .map(java.util.Map.Entry::getKey)
                .collect(java.util.stream.Collectors.toSet());
    }
    
    /**
     * 测试方法
     */
    public static void main(String[] args) {
        try {
            // 测试参数
            String host = "222.243.138.146";
            int port = 1883;
            String username = "ygtx";
            String password = "T!8dFz$kW3#pRgM2";
            String clientId = "test_client_" + System.currentTimeMillis();
            
            // 测试发送消息并等待响应
            String topic = "yzkj/zkshi/esp32/12led";
            String payload = "100 01 cccccc";
            
            log.info("开始测试 MQTT 功能...");
            log.info("Broker: {}:{}", host, port);
            
            CompletableFuture<String> future = sendAndWaitResponse(host, port, username, password, topic, payload, 10);
            
            future.thenAccept(response -> {
                log.info("收到响应: {}", response);
            }).exceptionally(throwable -> {
                log.error("操作失败: {}", throwable.getMessage());
                return null;
            });
            
            // 等待完成
            future.join();
            
            // 显示连接信息
            log.info("活跃连接数: {}", getActiveConnectionCount());
            log.info("活跃 Broker: {}", getActiveBrokers());
            
        } catch (Exception e) {
            log.error("测试失败: {}", e.getMessage(), e);
        }
    }
}
