hfln_miniprogram/src/utils/globalMqtt.js

import mqtt from './mqtt.js';

class MqttService {
    constructor() {
        if (!MqttService.instance) {
            this.cmdClient = null;
            this.dataClient = null;
            this.cmdConnected = false;
            this.dataConnected = false;
            this.connections = new Map(); // 内部管理多连接
            MqttService.instance = this;
        }
        return MqttService.instance;
    }

    // ---------------- 连接 CMD ----------------
    connectCmd(userId) {
        if (this.cmdConnected && this.cmdClient) {
            console.log("CMD MQTT 已连接，复用现有实例");
            return Promise.resolve(this.cmdClient);
        }

        const brokerName = 'CMD';
        const url = 'wxs://cmd.radar-power.cn/mqtt/';
        const clientId = `xcx_mqtt_cmd1_${userId}_${Math.random().toString(16).substring(2, 8)}`;

        return this.connectToBroker(brokerName, { url, clientId, username: 'lnradar', password: 'lnradar' })
            .then(client => {
                this.cmdClient = client;
                this.cmdConnected = true;

                // 默认订阅 CMD 主题
                const topic = `/mps/wx_${userId}/notice`;
                this.subscribe('CMD', topic, (msg) => console.log('CMD 消息:', msg));

                uni.$emit('mqtt-ready', this.cmdClient);
                return client;
            });
    }

    // ---------------- 连接 DATA ----------------
    connectData(userId) {

        if (this.dataConnected && this.dataClient) {
            console.log("DATA MQTT 已连接，复用现有实例");
            return Promise.resolve(this.dataClient);
        }
        uni.showToast({ title: "平台连接中...", icon: "loading", duration: 1000 });
        const brokerName = 'DATA';
        let url = ""
        url = "wxs://data.radar-power.cn/mqtt/";
        // if (__wxConfig.envVersion == 'develop') {
        //     url = "wxs://data.radar-power.asia:8084/mqtt/";
        // }
        // if (__wxConfig.envVersion == 'trial') {
        //     url = "wxs://data.radar-power.asia:8084/mqtt/";
        // }

        // if (__wxConfig.envVersion == 'release') {
        //     url = "wxs://data.radar-power.cn/mqtt/";
        // }

        const clientId = `xcx_mqtt_data1_${userId}_${Date.now()}`;
        return this.connectToBroker(brokerName, { url, clientId, username: 'lnradar', password: 'lnradar' })
            .then(client => {
                this.dataClient = client;
                this.dataConnected = true;
                uni.$emit('mqttData-ready', this.dataClient);
                return client;
            });
    }

    // ---------------- 内部通用连接方法 ----------------
    connectToBroker(brokerName, config) {
        return new Promise((resolve, reject) => {
            if (this.connections.has(brokerName)) {
                const existingConn = this.connections.get(brokerName);
                if (existingConn.connected) {
                    resolve(existingConn.client);
                    return;
                }
            }

            const client = mqtt.connect(config.url, {
                clientId: config.clientId,
                username: config.username,
                password: config.password,
                reconnectPeriod: config.reconnectPeriod || 500,
                wsOptions: {
                    WebSocket: url => wx.connectSocket({
                        url,
                        header: { 'content-type': 'application/json' },
                        protocols: ['mqtt'],
                    })
                },
            });

            const connection = {
                client,
                connected: false,
                subscriptions: new Map(), // topic => [callbacks]
            };
            this.connections.set(brokerName, connection);

            // ---------------- 连接成功 ----------------
            client.on('connect', () => {
                console.log(`${brokerName} MQTT 连接成功`);
                connection.connected = true;

                // 重连时恢复所有订阅
                connection.subscriptions.forEach((callbacks, topic) => {
                    connection.client.subscribe(topic, (err) => {
                        if (err) console.error(`重连订阅失败: ${topic}`, err);
                        else console.log(`✅ 重连订阅成功: ${topic}`);
                    });
                });

                resolve(client);
            });

            // ---------------- 连接错误 ----------------
            client.on('error', (err) => {
                uni.showModal({
                    title: '提示',
                    content: '连接平台失败,请重新登录',
                    showCancel: false,
                    success: (res) => {
                        if (res.confirm) {
                            uni.clearStorageSync();
                            uni.reLaunch({ url: "/pagesA/loginNew/loginNew" });
                        }
                    }
                });
                console.error(`${brokerName} MQTT 连接错误:`, err);
                connection.connected = false;
                reject(err);
            });

            // ---------------- 连接断开 ----------------
            client.on('close', () => {
                uni.showToast({ title: `平台已断开`, icon: "none", duration: 1500 });
                console.log(`${brokerName} MQTT 断开`);
                connection.connected = false;
                if (brokerName === 'DATA') this.dataConnected = false;
                if (brokerName === 'CMD') this.cmdConnected = false;
            });

            // ---------------- 接收消息 ----------------
            client.on('message', (topic, message) => {
                const callbacks = connection.subscriptions.get(topic) || [];
                callbacks.forEach(cb => {
                    try { cb(message.toString(), topic, brokerName); }
                    catch (err) { console.error('Message callback error:', err); }
                });
            });

            // ---------------- 连接超时 ----------------
            setTimeout(() => {
                if (!connection.connected) reject(new Error(`${brokerName} 连接超时`));
            }, 10000);
        });
    }

    // ---------------- 发布消息 ----------------
    publish(clientType, topic, message) {
        const client = clientType === 'CMD' ? this.cmdClient : this.dataClient;
        if (client) client.publish(topic, message);
        else console.warn(`${clientType} MQTT 未连接，无法 publish`);
    }

    // ---------------- 订阅 ----------------
    subscribe(clientType, topic, callback, onSubscribeDone) {
        const connection = clientType === 'CMD' ? this.connections.get('CMD') : this.connections.get('DATA');
        if (!connection || !connection.connected) {
            console.warn(`${clientType} MQTT 未连接，无法 subscribe`);
            if (onSubscribeDone) onSubscribeDone(new Error('未连接'));
            return;
        }

        // 保存回调
        if (!connection.subscriptions.has(topic)) connection.subscriptions.set(topic, []);
        const callbacks = connection.subscriptions.get(topic);
        if (!callbacks.includes(callback)) callbacks.push(callback);

        // 执行订阅
        connection.client.subscribe(topic, (err) => {
            if (err) console.error(`Subscribe error on ${clientType}:`, err);
            if (onSubscribeDone) onSubscribeDone(err);
        });

        // 返回取消订阅函数
        return () => this.unsubscribe(clientType, topic, callback);
    }

    // ---------------- 取消订阅 ----------------
    unsubscribe(clientType, topic, callback) {
        const connection = clientType === 'CMD' ? this.connections.get('CMD') : this.connections.get('DATA');
        if (!connection) return;

        const callbacks = connection.subscriptions.get(topic);
        if (callbacks) {
            const index = callbacks.indexOf(callback);
            if (index > -1) callbacks.splice(index, 1);
            if (callbacks.length === 0) {
                connection.subscriptions.delete(topic);
                connection.client.unsubscribe(topic);
            }
        }
    }

    // ---------------- 手动恢复订阅 ----------------
    resubscribeAll(brokerName) {
        const connection = this.connections.get(brokerName);
        if (!connection || !connection.connected) return;

        connection.subscriptions.forEach((callbacks, topic) => {
            connection.client.subscribe(topic, (err) => {
                if (err) console.error(`手动恢复订阅失败: ${topic}`, err);
                else console.log(`✅ 手动恢复订阅成功: ${topic}`);
            });
        });
    }

    // ---------------- 断开连接 ----------------
    disconnectAll() {
        if (this.cmdClient) { this.cmdClient.end(true); this.cmdClient = null; this.cmdConnected = false; }
        if (this.dataClient) { this.dataClient.end(true); this.dataClient = null; this.dataConnected = false; }
        this.connections.forEach(conn => conn.client.end(true));
        this.connections.clear();
        console.log('所有 MQTT 连接已关闭');
    }

    disconnectData() {
        if (this.dataClient) {
            this.dataClient.end(true);
            this.dataClient = null;
            this.dataConnected = false;
            this.connections.delete("DATA");
            console.log("DATA MQTT 连接已关闭");
        }
    }
}

// 单例导出
const instance = new MqttService();
export default instance;