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

import mqtt from 'mqtt'
import { GATEWAY_WS } from '@/constant'
import store from '@/store'
import SM4 from '@/utils/sm4'

const willTopic = 'web/offline'

export 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 = `${GATEWAY_WS}${process.env.VUE_APP_MQTT_PROXY}`
const username = process.env.VUE_APP_MQTT_USER_NAME
const password = process.env.VUE_APP_MQTT_PASSWORD

const whiteList = [
  {
    type: 'topic',
    val: /^\$SYS/
  }, {
    type: 'topic',
    eq: true,
    val: willTopic
  }, {
    type: 'topic',
    val: /^\d+\/\d+\/(status|screenshot|restart)/
  }, {
    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)
  }
}

export function createClient (options) {
  return mqtt.connect(host, {
    username,
    password,
    clientId: `mqtt_${store.getters.account}_${Math.random().toString(16).slice(2)}`,
    protocolId: 'MQTT',
    protocolVersion: 4,
    clean: true,
    keepalive: 30,
    reconnectPeriod: 1000,
    connectTimeout: 30 * 1000,
    ...options
  })
}

const cbs = []
let mqttClient

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

function start () {
  console.log('Connecting mqtt client...')
  const client = createClient({
    will: {
      topic: willTopic,
      payload: 'Connection Closed abnormally..!',
      qos: 0,
      retain: false
    }
  })

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

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

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

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

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

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

  client.on('message', (topic, payload) => {
    console.log('MQTT topic', topic)
    if (cbs.length) {
      const message = decodePayload(topic, payload)
      cbs.forEach(cb => cb(topic, message))
    }
  })

  return client
}

export function decodePayload (topic, payload) {
  try {
    const s = utf8ArrayBufferToString(payload)
    if (s) {
      payload = decodeMessage(topic, payload, s)
    } else {
      console.log('null')
      payload = s
    }
    console.log(payload)
  } catch (e) {
    console.warn(e)
  }
  return payload
}

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

function decodeMessage (topic, arrayBuffer, message) {
  if (topic && whiteList.some(item => isWhite(item, topic, message))) {
    console.log('white list')
  } else {
    const decodeStr = sm4.decrypt(arrayBuffer)
    if (decodeStr) {
      console.log('sm4')
      return decodeStr
    }
    console.log('unencrypted')
  }
  return message
}

export 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)
}

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') {
      console.log('MQTT publish', topic)
      console.log('encode', !!encode)
      console.log(message)
      mqttInst.client.publish(topic, message && encode ? sm4.encrypt(message) : message, err => {
        if (err) {
          console.warn(err)
          reject(err)
        } else {
          resolve()
        }
      })
    } else {
      reject('not connected')
    }
  })
}