lilishop-uniapp/pages/promotion/live/mqtt/mqtt.js

// #ifndef MP-WEIXIN
import api from "@/config/api.js";
import mqtt from "mqtt";

const mqttEndpoint = api.mqtt;

class MqttClient {
  constructor(options = {}) {
    const defaultOptions = {
      endpoint: mqttEndpoint,
      username: "admin",
      password: "hivemq",
      ssl: true,
      keepalive: 30,
      clean: true,
      reconnectPeriod: 15000,
      connectTimeout: 20000,
      topicHandlers: [],
    };

    this.options = { ...defaultOptions, ...options };
    this.client = null;
    this.connected = false;
    this.connecting = false;
    this.reconnecting = false;
    this.error = null;
    this.topics = [];
    this.connectCount = 0;
    this.MAX_CONNECT_RETRIES = 5;
    this.manualDisconnect = false;

    this.onConnectCallback = null;
    this.onReconnectCallback = null;
    this.onErrorCallback = null;
    this.onCloseCallback = null;
    this.onOfflineCallback = null;
    this.onMessageCallback = null;

    this.onConnect = this.onConnect.bind(this);
    this.onReconnect = this.onReconnect.bind(this);
    this.onError = this.onError.bind(this);
    this.onMessage = this.onMessage.bind(this);
    this.onClose = this.onClose.bind(this);
    this.onOffline = this.onOffline.bind(this);
  }

  connect() {
    try {
      if (!this.options.endpoint) {
        throw new Error("MQTT 地址未配置，请检查 config/api.js 的 mqtt 字段");
      }

      console.log("[MQTT] 开始连接:", this.options.endpoint);
      this.manualDisconnect = false;

      if (this.client) {
        this.disconnect();
      }

      this.connecting = true;
      this.error = null;

      const connectionOptions = {
        clientId: this.options.clientId,
        keepalive: this.options.keepalive,
        clean: this.options.clean,
        reconnectPeriod: this.options.reconnectPeriod,
        connectTimeout: this.options.connectTimeout,
        username: this.options.username || undefined,
        password: this.options.password || undefined,
      };

      this.client = mqtt.connect(this.options.endpoint, connectionOptions);

      this.client.on("connect", this.onConnect);
      this.client.on("reconnect", this.onReconnect);
      this.client.on("error", this.onError);
      this.client.on("message", this.onMessage);
      this.client.on("close", this.onClose);
      this.client.on("offline", this.onOffline);

      const timeoutId = setTimeout(() => {
        if (!this.connected) {
          this.onError(new Error("连接超时"));
          if (this.client && typeof this.client.end === "function" && !this.manualDisconnect) {
            try {
              this.client.end(true);
            } catch (e) {
              console.warn("[MQTT] 超时关闭客户端失败:", e);
            }
          }
        }
      }, this.options.connectTimeout);

      this.client.once("connect", () => {
        clearTimeout(timeoutId);
      });
    } catch (err) {
      console.error("[MQTT] 连接异常:", err);
      this.onError(err);
    }

    return this;
  }

  disconnect() {
    this.manualDisconnect = true;

    if (this.client) {
      try {
        if (this.client.removeAllListeners) {
          this.client.removeAllListeners();
        }
        if (typeof this.client.end === "function") {
          this.client.end(true);
        }
        console.log("[MQTT] 连接已断开");
      } catch (err) {
        console.warn("[MQTT] 断开连接异常:", err);
      }
    }

    this.client = null;
    this.connected = false;
    this.connecting = false;
    this.reconnecting = false;
    this.topics = [];
    this.error = null;
    this.connectCount = 0;

    return this;
  }

  onConnect() {
    console.log("[MQTT] 连接成功");
    this.connected = true;
    this.connecting = false;
    this.reconnecting = false;
    this.error = null;
    this.connectCount = 0;
    this.subscribeToTopics();
    this.onConnectCallback?.();
  }

  onReconnect() {
    console.log("[MQTT] 重连中...");
    this.connected = false;
    this.connecting = true;
    this.reconnecting = true;
    this.connectCount++;

    if (this.connectCount > this.MAX_CONNECT_RETRIES) {
      console.warn("[MQTT] 重连次数超限，停止重连");
      if (this.client && typeof this.client.end === "function") {
        try {
          this.client.end(true);
        } catch (err) {
          console.warn("[MQTT] 停止客户端异常:", err);
        }
      }
      this.client = null;
      this.onError(new Error("重连次数超限"));
    }

    this.onReconnectCallback?.(this.connectCount);
  }

  onError(err) {
    console.error("[MQTT] 错误:", err?.message || err);
    this.error = err;
    this.connecting = false;
    this.onErrorCallback?.(err);
  }

  onClose() {
    console.log("[MQTT] 连接关闭");
    this.connected = false;
    this.onCloseCallback?.();
  }

  onOffline() {
    console.log("[MQTT] 客户端离线");
    this.connected = false;
    this.onOfflineCallback?.();
  }

  onMessage(topic, message) {
    const msgStr = message.toString();

    this.options.topicHandlers
      .filter((handler) => handler.topic === topic || this.matchTopicPattern(handler.topic, topic))
      .forEach((handler) => {
        try {
          handler.handler(msgStr, topic);
        } catch (err) {
          console.error(`[MQTT] 处理主题 ${topic} 消息失败:`, err);
        }
      });

    this.onMessageCallback?.(topic, msgStr);
  }

  isClientValid() {
    return this.client && this.connected && typeof this.client.subscribe === "function";
  }

  matchTopicPattern(pattern, topic) {
    const regex = new RegExp(
      `^${pattern
        .replace(/\+/g, "[^/]+")
        .replace(/#$/, ".*")
        .replace(/\//g, "\\/")}$`
    );
    return regex.test(topic);
  }

  subscribe(topic, options = { qos: 0 }, handler) {
    if (!this.isClientValid()) {
      console.warn("[MQTT] 客户端无效，无法订阅");
      return false;
    }

    try {
      this.client.subscribe(topic, options, (err, granted) => {
        if (err) {
          console.error(`[MQTT] 订阅 ${topic} 失败:`, err);
          return;
        }
        console.log(`[MQTT] 订阅成功: ${topic}`, granted);

        const idx = this.topics.findIndex((t) => t.topic === topic);
        if (idx >= 0) {
          this.topics[idx] = { topic, qos: options.qos, handler };
        } else {
          this.topics.push({ topic, qos: options.qos, handler });
        }

        if (handler && typeof handler === "function") {
          const hIdx = this.options.topicHandlers.findIndex((h) => h.topic === topic);
          if (hIdx >= 0) {
            this.options.topicHandlers[hIdx] = { topic, qos: options.qos, handler };
          } else {
            this.options.topicHandlers.push({ topic, qos: options.qos, handler });
          }
        }
      });
      return true;
    } catch (err) {
      console.error(`[MQTT] 订阅 ${topic} 异常:`, err);
      return false;
    }
  }

  subscribeToTopics() {
    if (!this.isClientValid()) return;

    this.options.topicHandlers.forEach(({ topic, qos = 0, handler }) => {
      this.subscribe(topic, { qos }, handler);
    });
  }

  onConnected(callback) {
    this.onConnectCallback = callback;
    return this;
  }
  onErrorOccurred(callback) {
    this.onErrorCallback = callback;
    return this;
  }
  onClosed(callback) {
    this.onCloseCallback = callback;
    return this;
  }
  onClientOffline(callback) {
    this.onOfflineCallback = callback;
    return this;
  }

  isConnected() {
    return this.connected;
  }
}

let mqttInstance = null;

export function createMqttClient(options = {}) {
  if (mqttInstance && mqttInstance.isConnected()) {
    mqttInstance.disconnect();
  }
  mqttInstance = new MqttClient(options);
  return mqttInstance;
}

export { MqttClient };
// #endif

// #ifdef MP-WEIXIN
class MqttClient {
  connect() {
    return Promise.resolve(this);
  }
  disconnect() {}
  isConnected() {
    return false;
  }
  subscribe() {
    return this;
  }
  on() {
    return this;
  }
}

export function createMqttClient() {
  return new MqttClient();
}

export { MqttClient };
// #endif