smsb-web-develop/src/utils/mqtt.js

import mqtt from 'mqtt'
import SM4 from '@/utils/sm4'

const willTopic = 'web/offline'

const sm4 = new SM4({
  // encrypt/decypt main key; cannot be omitted
  // key: 'fELIMxLsdoSgRZnX',
  key: [102, 69, 76, 73, 77, 120, 76, 115, 100, 111, 83, 103, 82, 90, 110, 88],
  // optional; can be 'cbc' or 'ecb'
  // default is ecb
  mode: 'ecb',
  // optional; can be 'uint8array' or 'text'
  // default is uint8array
  outType: 'uint8array'
})

const host = process.env.VUE_APP_MQTT_URL || `${location.protocol === 'https:' ? 'wss' : 'ws'}://${location.host}/mqtt`
const username = process.env.VUE_APP_MQTT_USER_NAME
const password = process.env.VUE_APP_MQTT_PASSWORD

const options = {
  username,
  password,
  clientId: 'mqttjs_' + Math.random().toString(16).slice(2),
  keepalive: 60,
  protocolId: 'MQTT',
  protocolVersion: 4,
  clean: true,
  reconnectPeriod: 1000,
  connectTimeout: 30 * 1000,
  will: {
    topic: willTopic,
    payload: 'Connection Closed abnormally..!',
    qos: 0,
    retain: false
  }
}

const whiteList = [
  {
    type: 'topic',
    val: /^\$SYS/
  }, {
    type: 'topic',
    eq: true,
    val: willTopic
  }, {
    type: 'topic',
    val: /^\d+\/\d+\/(status|screenshot)/
  }, {
    val: /messageId/
  }
]

function isWhite (item, topic, message) {
  switch (item.type) {
    case 'topic':
      return item.eq ? topic === item.val : item.val.test(topic)
    default:
      return item.val.test(message)
  }
}

let mqttClient

function changeState (state) {
  mqttClient && (mqttClient.state = state)
}

function start () {
  console.log('Connecting mqtt client...')
  const client = mqtt.connect(host, options)

  client.on('error', err => {
    console.log('MQTT connection error: ', err)
    client.end(true)
    mqttClient = null
  })

  client.on('connect', () => {
    console.log('MQTT connected')
    changeState('connect')
  })

  client.on('reconnect', () => {
    console.log('MQTT reconnecting...')
    changeState('reconnect')
  })

  client.on('close', () => {
    console.log('MQTT closed')
    changeState('close')
  })

  client.on('disconnect', () => {
    console.log('MQTT disconnect')
    changeState('disconnect')
  })

  client.on('offline', () => {
    console.log('MQTT offline')
    changeState('offline')
  })

  client.on('end', () => {
    console.log('MQTT end')
    mqttClient = null
  })

  client.on('message', (topic, message) => {
    console.log('MQTT topic', topic)

    try {
      const s = utf8ArrayBufferToString(message)
      if (s) {
        if (whiteList.some(item => isWhite(item, topic, s))) {
          console.log('white list')
          message = s
        } else {
          const decodeStr = sm4.decrypt(message)
          if (decodeStr) {
            console.log('sm4')
            message = decodeStr
          } else {
            console.log('unencrypted')
            message = s
          }
        }
      } else {
        console.log('null')
        message = s
      }
    } catch (e) {
      console.warn(e)
      message = utf8ArrayBufferToString(message)
    }

    console.log(message)

    cbs.forEach(cb => cb(topic, message))
  })

  return client
}

const decoder = new TextDecoder('utf-8')
function utf8ArrayBufferToString (buffer) {
  return decoder.decode(buffer)
}

function inst () {
  if (!mqttClient) {
    mqttClient = {
      state: 'init',
      client: start()
    }
  }
  return mqttClient
}

export function close () {
  if (mqttClient) {
    mqttClient.client.end(true)
    mqttClient = null
  }
}

export function subscribe (topic, fn) {
  if (topic) {
    console.info(`MQTT subscribe ${topic}`)
    inst().client.subscribe(topic, e => {
      e && console.warn('MQTT subscribe', e)
    })
  }
  listen(fn)
}

export function unsubscribe (topic, fn) {
  if (topic && mqttClient) {
    console.info(`MQTT unsubscribe ${topic}`)
    mqttClient.client.unsubscribe(topic, e => {
      e && console.warn('MQTT unsubscribe', e)
    })
  }
  unlisten(fn)
}

const cbs = []
export function listen (fn) {
  fn && cbs.push(fn)
}

export function unlisten (fn) {
  if (fn) {
    const index = cbs.findIndex(item => item === fn)
    if (~index) {
      cbs.splice(index, 1)
    }
  }
}

export function publish (topic, message, encode) {
  return new Promise((resolve, reject) => {
    const mqttInst = inst()
    if (mqttInst && mqttInst.state === 'connect') {
      mqttInst.client.publish(topic, message && encode ? sm4.encrypt(message) : message, err => {
        if (err) {
          console.warn(err)
          reject(err)
        } else {
          resolve()
        }
      })
    } else {
      reject('not connected')
    }
  })
}