qn-ai-pc/src/utils/mqtt.ts

import mqtt from 'mqtt'

export interface Options {
  host: string
  port: number // ws -> 8083; wss -> 8084
  protocol: mqtt.MqttProtocol // ws or wss
  clean: boolean
  clientId: string
  connectTimeout: number
  reconnectPeriod: number
  username: string
  password: string
}
export class MqttService {
  client: mqtt.MqttClient | null
  options: Options
  constructor(options: Options) {
    this.client = null
    this.options = options
  }

  connect() {
    return new Promise((resolve, reject) => {
      const { protocol, host, port } = this.options
      const connectUrl = `${protocol}://${host}:${port}/mqtt`
      this.client = mqtt.connect(connectUrl, this.options)
      console.log(this.client)
      this.client.on('connect', () => {
        console.log('连接成功')
        resolve(true)
      })
      this.client.on('error', (error: any) => {
        console.log('连接失败！', error)
        reject(error)
      })
    })
  }

  subscribe(topic: string) {
    if (!this.client || !this.client.connected) {
      throw new Error('MQTT client is not connected.')
    }

    return new Promise((resolve, reject) => {
      this.client.subscribe(topic, (err: any) => {
        if (err) {
          console.log('订阅失败！', err)
          reject(err)
        }
        else {
          console.log('订阅成功！', topic)
          resolve(true)
        }
      })
    })
  }

  onMessage(topic: string, callback: any) {
    if (!this.client || !this.client.connected) {
      throw new Error('MQTT client is not connected.')
    }
    this.client.on(
      'message',
      (receivedTopic: string, message: { toString: () => any }) => {
        if (receivedTopic === topic) {
          callback(JSON.parse(message.toString()))
        }
      },
    )
  }

  unsubscribe(topic: string) {
    if (!this.client || !this.client.connected) {
      throw new Error('MQTT client is not connected.')
    }

    this.client.unsubscribe(topic, (err: any) => {
      if (err) {
        console.log('取消订阅失败！', err)
      }
      else {
        console.log('取消订阅成功！', topic)
      }
    })
  }

  end() {
    if (this.client) {
      this.client.end()
      console.log('mqtt已断开')
    }
  }
}