client.js

'use strict'

const mqtt = require('mqtt-packet')
const EventEmitter = require('events')
const util = require('util')
const eos = require('end-of-stream')
const Packet = require('aedes-packet')
const write = require('./write')
const QoSPacket = require('./qos-packet')
const handleSubscribe = require('./handlers/subscribe')
const handleUnsubscribe = require('./handlers/unsubscribe')
const handle = require('./handlers')
const { pipeline } = require('stream')
const { through } = require('./utils')

module.exports = Client

function Client (broker, conn, req) {
  const that = this

  // metadata
  this.closed = false
  this.connecting = false
  this.connected = false
  this.connackSent = false
  this.errored = false

  // mqtt params
  this.id = null
  this.clean = true
  this.version = null

  this.subscriptions = {}
  this.duplicates = {}

  this.broker = broker
  this.conn = conn
  conn.client = this

  this._disconnected = false
  this._authorized = false
  this._parsingBatch = 1
  this._nextId = Math.ceil(Math.random() * 65535)

  this.req = req
  this.connDetails = req ? req.connDetails : null

  // we use two variables for the will
  // because we store in _will while
  // we are authenticating
  this.will = null
  this._will = null

  this._parser = mqtt.parser()
  this._parser.client = this
  this._parser._queue = [] // queue packets received before client fires 'connect' event. Prevents memory leaks on 'connect' event
  this._parser.on('packet', enqueue)
  this.once('connected', dequeue)

  function nextBatch (err) {
    if (err) {
      that.emit('error', err)
      return
    }

    const client = that

    if (client._paused) {
      return
    }

    that._parsingBatch--
    if (that._parsingBatch <= 0) {
      that._parsingBatch = 0
      const buf = client.conn.read(null)
      if (buf) {
        client._parser.parse(buf)
      }
    }
  }
  this._nextBatch = nextBatch

  conn.on('readable', nextBatch)

  this.on('error', onError)
  conn.on('error', this.emit.bind(this, 'error'))
  this._parser.on('error', this.emit.bind(this, 'error'))

  conn.on('end', this.close.bind(this))
  this._eos = eos(this.conn, this.close.bind(this))

  const getToForwardPacket = (_packet) => {
    // Mqttv5 3.8.3.1: https://docs.oasis-open.org/mqtt/mqtt/v5.0/mqtt-v5.0.html#_Toc3901169
    // prevent to forward messages sent by the same client when no-local flag is set
    if (_packet.clientId === that.id && _packet.nl) return

    const toForward = dedupe(that, _packet) &&
      that.broker.authorizeForward(that, _packet)

    return toForward
  }

  this.deliver0 = function deliverQoS0 (_packet, cb) {
    const toForward = getToForwardPacket(_packet)

    if (toForward) {
      // Give nodejs some time to clear stacks, or we will see
      // "Maximum call stack size exceeded" in a very high load
      setImmediate(() => {
        const packet = new Packet(toForward, broker)
        packet.qos = 0
        write(that, packet, function (err) {
          that._onError(err)
          cb() // don't pass the error here or it will be thrown by mqemitter
        })
      })
    } else {
      setImmediate(cb)
    }
  }

  this.deliverQoS = function deliverQoS (_packet, cb) {
    // downgrade to qos0 if requested by publish
    if (_packet.qos === 0) {
      that.deliver0(_packet, cb)
      return
    }
    const toForward = getToForwardPacket(_packet)

    if (toForward) {
      setImmediate(() => {
        const packet = new QoSPacket(toForward, that)
        // Downgrading to client subscription qos if needed
        const clientSub = that.subscriptions[packet.topic]
        if (clientSub && (clientSub.qos || 0) < packet.qos) {
          packet.qos = clientSub.qos
        }
        packet.writeCallback = cb
        if (that.clean || packet.retain) {
          writeQoS(null, that, packet)
        } else {
          broker.persistence.outgoingUpdate(that, packet, writeQoS)
        }
      })
    } else if (that.clean === false) {
      that.broker.persistence.outgoingClearMessageId(that, _packet, noop)
      // we consider this to be an error, since the packet is undefined
      // so there's nothing to send
      setImmediate(cb)
    } else {
      setImmediate(cb)
    }
  }

  this._keepaliveTimer = null
  this._keepaliveInterval = -1

  this._connectTimer = setTimeout(function () {
    that.emit('error', new Error('connect did not arrive in time'))
  }, broker.connectTimeout)
}

function dedupe (client, packet) {
  const id = packet.brokerId
  if (!id) {
    return true
  }
  const duplicates = client.duplicates
  const counter = packet.brokerCounter
  const result = (duplicates[id] || 0) < counter
  if (result) {
    duplicates[id] = counter
  }
  return result
}

function writeQoS (err, client, packet) {
  if (err) {
    // is this right, or we should ignore thins?
    client.emit('error', err)
    // don't pass the error here or it will be thrown by mqemitter
    packet.writeCallback()
  } else {
    write(client, packet, function (err) {
      if (err) {
        client.emit('error', err)
      }
      // don't pass the error here or it will be thrown by mqemitter
      packet.writeCallback()
    })
  }
}

function drainRequest (req) {
  req.callback()
}

function onError (err) {
  if (!err) return

  this.errored = true
  this.conn.removeAllListeners('error')
  this.conn.on('error', noop)
  // hack to clean up the write callbacks in case of error
  const state = this.conn._writableState
  const list = typeof state.getBuffer === 'function' ? state.getBuffer() : state.buffer
  list.forEach(drainRequest)
  this.broker.emit(this.id ? 'clientError' : 'connectionError', this, err)
  this.close()
}

util.inherits(Client, EventEmitter)

Client.prototype._onError = onError

Client.prototype.publish = function (message, done) {
  const packet = new Packet(message, this.broker)
  const that = this
  if (packet.qos === 0) {
    // skip offline and send it as it is
    this.deliver0(packet, done)
    return
  }
  if (!this.clean && this.id) {
    this.broker.persistence.outgoingEnqueue({
      clientId: this.id
    }, packet, function deliver (err) {
      if (err) {
        return done(err)
      }
      that.deliverQoS(packet, done)
    })
  } else {
    that.deliverQoS(packet, done)
  }
}

Client.prototype.subscribe = function (packet, done) {
  if (!packet.subscriptions) {
    if (!Array.isArray(packet)) {
      packet = [packet]
    }
    packet = {
      subscriptions: packet
    }
  }
  handleSubscribe(this, packet, false, done)
}

Client.prototype.unsubscribe = function (packet, done) {
  if (!packet.unsubscriptions) {
    if (!Array.isArray(packet)) {
      packet = [packet]
    }
    packet = {
      unsubscriptions: packet
    }
  }
  handleUnsubscribe(this, packet, done)
}

Client.prototype.close = function (done) {
  if (this.closed) {
    if (typeof done === 'function') {
      done()
    }
    return
  }

  const that = this
  const conn = this.conn

  this.closed = true

  this._parser.removeAllListeners('packet')
  conn.removeAllListeners('readable')

  this._parser._queue = null

  if (this._keepaliveTimer) {
    this._keepaliveTimer.clear()
    this._keepaliveInterval = -1
    this._keepaliveTimer = null
  }

  if (this._connectTimer) {
    clearTimeout(this._connectTimer)
    this._connectTimer = null
  }

  this._eos()
  this._eos = noop

  handleUnsubscribe(
    this,
    {
      unsubscriptions: Object.keys(this.subscriptions)
    },
    finish)

  function finish () {
    const will = that.will
    // _disconnected is set only if client is disconnected with a valid disconnect packet
    if (!that._disconnected && will) {
      that.broker.authorizePublish(that, will, function (err) {
        if (err) { return done() }
        that.broker.publish(will, that, done)

        function done () {
          that.broker.persistence.delWill({
            id: that.id,
            brokerId: that.broker.id
          }, noop)
        }
      })
    } else if (will) {
      // delete the persisted will even on clean disconnect https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc385349232
      that.broker.persistence.delWill({
        id: that.id,
        brokerId: that.broker.id
      }, noop)
    }

    that.will = null // this function might be called twice
    that._will = null

    that.connected = false
    that.connecting = false

    conn.removeAllListeners('error')
    conn.on('error', noop)

    if (that.broker.clients[that.id] && that._authorized) {
      that.broker.unregisterClient(that)
    }

    // clear up the drain event listeners
    that.conn.emit('drain')
    that.conn.removeAllListeners('drain')

    conn.destroy()

    if (typeof done === 'function') {
      done()
    }
  }
}

Client.prototype.pause = function () {
  this._paused = true
}

Client.prototype.resume = function () {
  this._paused = false
  this._nextBatch()
}

function enqueue (packet) {
  const client = this.client
  client._parsingBatch++
  // already connected or it's the first packet
  if (client.connackSent || client._parsingBatch === 1) {
    handle(client, packet, client._nextBatch)
  } else {
    if (this._queue.length < client.broker.queueLimit) {
      this._queue.push(packet)
    } else {
      this.emit('error', new Error('Client queue limit reached'))
    }
  }
}

function dequeue () {
  const q = this._parser._queue
  if (q) {
    for (let i = 0, len = q.length; i < len; i++) {
      handle(this, q[i], this._nextBatch)
    }

    this._parser._queue = null
  }
}

Client.prototype.emptyOutgoingQueue = function (done) {
  const client = this
  const persistence = client.broker.persistence

  function filter (packet, enc, next) {
    persistence.outgoingClearMessageId(client, packet, next)
  }

  pipeline(
    persistence.outgoingStream(client),
    through(filter),
    done
  )
}

function noop () {}