首次提交

2026-07-13 12:06:16 +08:00
commit e3c810b1a6
1690 changed files with 228324 additions and 0 deletions
@@ -0,0 +1,399 @@
+'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 () {}
@@ -0,0 +1,272 @@
+'use strict'
+
+const retimer = require('retimer')
+const { pipeline } = require('stream')
+const write = require('../write')
+const QoSPacket = require('../qos-packet')
+const { through } = require('../utils')
+const handleSubscribe = require('./subscribe')
+const uniqueId = require('hyperid')()
+
+function Connack (arg) {
+  this.cmd = 'connack'
+  this.returnCode = arg.returnCode
+  this.sessionPresent = arg.sessionPresent
+}
+
+function ClientPacketStatus (client, packet) {
+  this.client = client
+  this.packet = packet
+}
+
+const connectActions = [
+  authenticate,
+  setKeepAlive,
+  fetchSubs,
+  restoreSubs,
+  storeWill,
+  registerClient,
+  doConnack,
+  emptyQueue
+]
+
+const errorMessages = [
+  '',
+  'unacceptable protocol version',
+  'identifier rejected',
+  'Server unavailable',
+  'bad user name or password',
+  'not authorized',
+  'keep alive limit exceeded'
+]
+
+function handleConnect (client, packet, done) {
+  clearTimeout(client._connectTimer)
+  client._connectTimer = null
+  client.connecting = true
+  client.broker.preConnect(client, packet, negate)
+
+  function negate (err, successful) {
+    if (!err && successful === true) {
+      setImmediate(init, client, packet, done)
+    } else {
+      client.connecting = false
+      done(err)
+    }
+  }
+}
+
+function init (client, packet, done) {
+  const clientId = packet.clientId
+  let returnCode = 0
+  // [MQTT-3.1.2-2]
+  if (packet.protocolVersion < 3 || packet.protocolVersion > 4) {
+    returnCode = 1
+  }
+  // MQTT 3.1.0 allows <= 23 client id length
+  if (packet.protocolVersion === 3 && clientId.length > client.broker.maxClientsIdLength) {
+    returnCode = 2
+  }
+  // check if the client keepalive is compatible with broker settings
+  if (client.broker.keepaliveLimit && (!packet.keepalive || packet.keepalive > client.broker.keepaliveLimit)) {
+    returnCode = 6
+  }
+  if (returnCode > 0) {
+    const error = new Error(errorMessages[returnCode])
+    error.errorCode = returnCode
+    doConnack(
+      { client, returnCode, sessionPresent: false },
+      done.bind(this, error))
+    return
+  }
+
+  client.id = clientId || 'aedes_' + uniqueId()
+  client.clean = packet.clean
+  client.version = packet.protocolVersion
+  client._will = packet.will
+
+  client.broker._series(
+    new ClientPacketStatus(client, packet),
+    connectActions,
+    { returnCode: 0, sessionPresent: false }, // [MQTT-3.1.4-4], [MQTT-3.2.2-4]
+    function (err) {
+      this.client.connecting = false
+      if (!err) {
+        this.client.connected = true
+        this.client.broker.emit('clientReady', client)
+        this.client.emit('connected')
+      }
+      done(err)
+    })
+}
+
+function authenticate (arg, done) {
+  const client = this.client
+  client.pause()
+  client.broker.authenticate(
+    client,
+    this.packet.username,
+    this.packet.password,
+    negate)
+
+  function negate (err, successful) {
+    if (client.closed || client.broker.closed) {
+      // a hack, sometimes client.close() or broker.close() happened
+      // before authenticate() comes back
+      // we stop here for not to register it and deregister it in write()
+      return
+    }
+    if (!err && successful) {
+      client._authorized = true
+      return done()
+    }
+
+    if (err) {
+      const errCode = err.returnCode
+      if (errCode && (errCode >= 2 && errCode <= 5)) {
+        arg.returnCode = errCode
+      } else {
+        arg.returnCode = 5
+      }
+      if (!err.message) {
+        err.message = errorMessages[arg.returnCode]
+      }
+    } else {
+      arg.returnCode = 5
+      err = new Error(errorMessages[arg.returnCode])
+    }
+    err.errorCode = arg.returnCode
+    arg.client = client
+    doConnack(arg,
+      // [MQTT-3.2.2-5]
+      client.close.bind(client, done.bind(this, err)))
+  }
+}
+
+function setKeepAlive (arg, done) {
+  if (this.packet.keepalive > 0) {
+    const client = this.client
+    // [MQTT-3.1.2-24]
+    client._keepaliveInterval = (this.packet.keepalive * 1500) + 1
+    client._keepaliveTimer = retimer(function keepaliveTimeout () {
+      client.broker.emit('keepaliveTimeout', client)
+      client.emit('error', new Error('keep alive timeout'))
+    }, client._keepaliveInterval)
+  }
+  done()
+}
+
+function fetchSubs (arg, done) {
+  const client = this.client
+  if (!this.packet.clean) {
+    client.broker.persistence.subscriptionsByClient({
+      id: client.id,
+      done,
+      arg
+    }, gotSubs)
+    return
+  }
+  arg.sessionPresent = false // [MQTT-3.2.2-1]
+  client.broker.persistence.cleanSubscriptions(
+    client,
+    done)
+}
+
+function gotSubs (err, subs, client) {
+  if (err) {
+    return client.done(err)
+  }
+  client.arg.subs = subs
+  client.done()
+}
+
+function restoreSubs (arg, done) {
+  if (arg.subs) {
+    handleSubscribe(this.client, { subscriptions: arg.subs }, true, done)
+    arg.sessionPresent = !!arg.subs // cast to boolean, [MQTT-3.2.2-2]
+    return
+  }
+  arg.sessionPresent = false // [MQTT-3.2.2-1], [MQTT-3.2.2-3]
+  done()
+}
+
+function storeWill (arg, done) {
+  const client = this.client
+  client.will = client._will
+  // delete any existing will messages from persistence
+  client.broker.persistence.delWill(client, function () {
+    if (client.will) {
+      client.broker.persistence.putWill(
+        client,
+        client.will,
+        done)
+    } else {
+      done()
+    }
+  })
+}
+
+function registerClient (arg, done) {
+  const client = this.client
+  client.broker.registerClient(client)
+  done()
+}
+
+function doConnack (arg, done) {
+  const client = arg.client || this.client
+  const connack = new Connack(arg)
+  write(client, connack, function (err) {
+    if (!err) {
+      client.broker.emit('connackSent', connack, client)
+      client.connackSent = true
+    }
+    done(err)
+  })
+}
+
+// push any queued messages (included retained messages) at the disconnected time
+// when QoS > 0 and session is true
+function emptyQueue (arg, done) {
+  const client = this.client
+  const persistence = client.broker.persistence
+  const outgoing = persistence.outgoingStream(client)
+
+  client.resume()
+
+  pipeline(
+    outgoing,
+    through(function clearQueue (data, enc, next) {
+      const packet = new QoSPacket(data, client)
+      // Here we are deliberatly passing only the error
+      // This is because there is no destination stream so the "client"
+      // Object filled the buffer up to the highWaterMark preventing stored messages
+      // being sent
+      packet.writeCallback = (error, _client) => next(error)
+      persistence.outgoingUpdate(client, packet, emptyQueueFilter)
+    }),
+    done
+  )
+}
+
+function emptyQueueFilter (err, client, packet) {
+  const next = packet.writeCallback
+
+  if (err) {
+    client.emit('error', err)
+    return next()
+  }
+
+  const authorized = (packet.cmd === 'publish')
+    ? client.broker.authorizeForward(client, packet)
+    : true
+
+  const persistence = client.broker.persistence
+
+  if (client.clean || !authorized) {
+    persistence.outgoingClearMessageId(client, packet, next)
+  } else {
+    write(client, packet, next)
+  }
+}
+
+module.exports = handleConnect
@@ -0,0 +1,77 @@
+'use strict'
+
+const handleConnect = require('./connect')
+const handleSubscribe = require('./subscribe')
+const handleUnsubscribe = require('./unsubscribe')
+const handlePublish = require('./publish')
+const handlePuback = require('./puback')
+const handlePubrel = require('./pubrel')
+const handlePubrec = require('./pubrec')
+const handlePing = require('./ping')
+
+function handle (client, packet, done) {
+  if (packet.cmd === 'connect') {
+    if (client.connecting || client.connected) {
+      // [MQTT-3.1.0-2]
+      finish(client.conn, packet, done)
+      return
+    }
+    handleConnect(client, packet, done)
+    return
+  }
+
+  if (!client.connecting && !client.connected) {
+    // [MQTT-3.1.0-1]
+    finish(client.conn, packet, done)
+    return
+  }
+
+  switch (packet.cmd) {
+    case 'publish':
+      handlePublish(client, packet, done)
+      break
+    case 'subscribe':
+      handleSubscribe(client, packet, false, done)
+      break
+    case 'unsubscribe':
+      handleUnsubscribe(client, packet, done)
+      break
+    case 'pubcomp':
+    case 'puback':
+      handlePuback(client, packet, done)
+      break
+    case 'pubrel':
+      handlePubrel(client, packet, done)
+      break
+    case 'pubrec':
+      handlePubrec(client, packet, done)
+      break
+    case 'pingreq':
+      handlePing(client, packet, done)
+      break
+    case 'disconnect':
+      // [MQTT-3.14.4-3]
+      client._disconnected = true
+      // [MQTT-3.14.4-1] [MQTT-3.14.4-2]
+      client.conn.destroy()
+      done()
+      return
+    default:
+      client.conn.destroy()
+      done()
+      return
+  }
+
+  if (client._keepaliveInterval > 0) {
+    client._keepaliveTimer.reschedule(client._keepaliveInterval)
+  }
+}
+
+function finish (conn, packet, done) {
+  conn.destroy()
+  const error = new Error('Invalid protocol')
+  error.info = packet
+  done(error)
+}
+
+module.exports = handle
@@ -0,0 +1,13 @@
+'use strict'
+
+const write = require('../write')
+const pingResp = {
+  cmd: 'pingresp'
+}
+
+function handlePing (client, packet, done) {
+  client.broker.emit('ping', packet, client)
+  write(client, pingResp, done)
+}
+
+module.exports = handlePing
@@ -0,0 +1,11 @@
+'use strict'
+
+function handlePuback (client, packet, done) {
+  const persistence = client.broker.persistence
+  persistence.outgoingClearMessageId(client, packet, function (err, origPacket) {
+    client.broker.emit('ack', origPacket, client)
+    done(err)
+  })
+}
+
+module.exports = handlePuback
@@ -0,0 +1,65 @@
+'use strict'
+
+const write = require('../write')
+
+function PubAck (packet) {
+  this.cmd = 'puback'
+  this.messageId = packet.messageId
+}
+
+function PubRec (packet) {
+  this.cmd = 'pubrec'
+  this.messageId = packet.messageId
+}
+
+const publishActions = [
+  authorizePublish,
+  enqueuePublish
+]
+function handlePublish (client, packet, done) {
+  const topic = packet.topic
+  let err
+  if (topic.length === 0) {
+    err = new Error('empty topic not allowed in PUBLISH')
+    return done(err)
+  }
+  if (topic.indexOf('#') > -1) {
+    err = new Error('# is not allowed in PUBLISH')
+    return done(err)
+  }
+  if (topic.indexOf('+') > -1) {
+    err = new Error('+ is not allowed in PUBLISH')
+    return done(err)
+  }
+  client.broker._series(client, publishActions, packet, done)
+}
+
+function enqueuePublish (packet, done) {
+  const client = this
+
+  switch (packet.qos) {
+    case 2:
+      client.broker.persistence.incomingStorePacket(client, packet, function (err) {
+        if (err) { return done(err) }
+        write(client, new PubRec(packet), done)
+      })
+      break
+    case 1:
+      write(client, new PubAck(packet), function (err) {
+        if (err) { return done(err) }
+        client.broker.publish(packet, client, done)
+      })
+      break
+    case 0:
+      client.broker.publish(packet, client, done)
+      break
+    default:
+      // nothing to do
+  }
+}
+
+function authorizePublish (packet, done) {
+  this.broker.authorizePublish(this, packet, done)
+}
+
+module.exports = handlePublish
@@ -0,0 +1,30 @@
+'use strict'
+
+const write = require('../write')
+
+function PubRel (packet) {
+  this.cmd = 'pubrel'
+  this.messageId = packet.messageId
+}
+
+function handlePubrec (client, packet, done) {
+  const pubrel = new PubRel(packet)
+
+  if (client.clean) {
+    write(client, pubrel, done)
+    return
+  }
+
+  client.broker.persistence.outgoingUpdate(
+    client, pubrel, reply)
+
+  function reply (err) {
+    if (err) {
+      done(err)
+    } else {
+      write(client, pubrel, done)
+    }
+  }
+}
+
+module.exports = handlePubrec
@@ -0,0 +1,50 @@
+'use strict'
+
+const write = require('../write')
+
+function ClientPacketStatus (client, packet) {
+  this.client = client
+  this.packet = packet
+}
+
+function PubComp (packet) {
+  this.cmd = 'pubcomp'
+  this.messageId = packet.messageId
+}
+
+const pubrelActions = [
+  pubrelGet,
+  pubrelPublish,
+  pubrelWrite,
+  pubrelDel
+]
+function handlePubrel (client, packet, done) {
+  client.broker._series(
+    new ClientPacketStatus(client, packet),
+    pubrelActions, {}, done)
+}
+
+function pubrelGet (arg, done) {
+  const persistence = this.client.broker.persistence
+  persistence.incomingGetPacket(this.client, this.packet, reply)
+
+  function reply (err, packet) {
+    arg.packet = packet
+    done(err)
+  }
+}
+
+function pubrelPublish (arg, done) {
+  this.client.broker.publish(arg.packet, this.client, done)
+}
+
+function pubrelWrite (arg, done) {
+  write(this.client, new PubComp(arg.packet), done)
+}
+
+function pubrelDel (arg, done) {
+  const persistence = this.client.broker.persistence
+  persistence.incomingDelPacket(this.client, arg.packet, done)
+}
+
+module.exports = handlePubrel
@@ -0,0 +1,255 @@
+'use strict'
+
+const fastfall = require('fastfall')
+const Packet = require('aedes-packet')
+const { through } = require('../utils')
+const { validateTopic, $SYS_PREFIX } = require('../utils')
+const write = require('../write')
+
+const subscribeTopicActions = fastfall([
+  authorize,
+  storeSubscriptions,
+  addSubs
+])
+const restoreTopicActions = fastfall([
+  authorize,
+  addSubs
+])
+
+function SubAck (packet, granted) {
+  this.cmd = 'suback'
+  this.messageId = packet.messageId
+  // the qos granted
+  this.granted = granted
+}
+
+function Subscription (qos, func, rh, rap, nl) {
+  this.qos = qos
+  this.func = func
+
+  // retain-handling indicates how retained messages should be
+  // handled when a new subscription is created
+  // (see [MQTT-3.3.1-9] through [MQTT-3.3.1-11])
+  this.rh = rh
+
+  // retain-as-published indicates whether to leave the retain flag as-is (true)
+  // or to clear it before sending to subscriptions (false) default false
+  // (see [MQTT-3.3.1-12] through [MQTT-3.3.1-13])
+  this.rap = rap
+
+  // no-local indicates that a client should not receive its own
+  // messages (see [MQTT-3.8.3-3])
+  this.nl = nl
+}
+
+function SubscribeState (client, packet, restore, finish) {
+  this.client = client
+  this.packet = packet
+  this.actions = restore ? restoreTopicActions : subscribeTopicActions
+  this.finish = finish
+  this.subState = []
+}
+
+function SubState (client, packet, granted, rh, rap, nl) {
+  this.client = client
+  this.packet = packet
+  this.granted = granted
+  this.rh = rh
+  this.rap = rap
+  this.nl = nl
+}
+
+// if same subscribed topic in subs array, we pick up the last one
+function _dedupe (subs) {
+  const dedupeSubs = {}
+  for (let i = 0; i < subs.length; i++) {
+    const sub = subs[i]
+    dedupeSubs[sub.topic] = sub
+  }
+  const ret = []
+  for (const key in dedupeSubs) {
+    ret.push(dedupeSubs[key])
+  }
+  return ret
+}
+
+function handleSubscribe (client, packet, restore, done) {
+  packet.subscriptions = packet.subscriptions.length === 1 ? packet.subscriptions : _dedupe(packet.subscriptions)
+  client.broker._parallel(
+    new SubscribeState(client, packet, restore, done), // what will be this in the functions
+    doSubscribe, // function to call
+    packet.subscriptions, // first argument of the function
+    restore ? done : completeSubscribe // the function to be called when the parallel ends
+  )
+}
+
+function doSubscribe (sub, done) {
+  const s = new SubState(this.client, this.packet, sub.qos, sub.rh, sub.rap, sub.nl)
+  this.subState.push(s)
+  this.actions.call(s, sub, done)
+}
+
+function authorize (sub, done) {
+  const err = validateTopic(sub.topic, 'SUBSCRIBE')
+  if (err) {
+    return done(err)
+  }
+  const client = this.client
+  client.broker.authorizeSubscribe(client, sub, done)
+}
+
+function blockDollarSignTopics (func) {
+  return function deliverSharp (packet, cb) {
+    // '$' is 36
+    if (packet.topic.charCodeAt(0) === 36) {
+      cb()
+    } else {
+      func(packet, cb)
+    }
+  }
+}
+
+function storeSubscriptions (sub, done) {
+  if (!sub || typeof sub !== 'object') {
+    // means failure: MQTT 3.1.1 specs > 3.9.3 Payload
+    this.granted = 128
+    return done(null, null)
+  }
+
+  const packet = this.packet
+  const client = this.client
+
+  if (client.clean) {
+    return done(null, sub)
+  }
+
+  client.broker.persistence.addSubscriptions(client, packet.subscriptions, function addSub (err) {
+    done(err, sub)
+  })
+}
+
+function addSubs (sub, done) {
+  if (!sub || typeof sub !== 'object') {
+    return done()
+  }
+
+  const client = this.client
+  const broker = client.broker
+  const topic = sub.topic
+  const qos = sub.qos
+  const rh = this.rh
+  const rap = this.rap
+  const nl = this.nl
+  let func = qos > 0 ? client.deliverQoS : client.deliver0
+
+  const deliverFunc = func
+  func = function handlePacketSubscription (_packet, cb) {
+    _packet = new Packet(_packet, broker)
+    _packet.nl = nl
+    if (!rap) _packet.retain = false
+    deliverFunc(_packet, cb)
+  }
+
+  // [MQTT-4.7.2-1]
+  if (isStartsWithWildcard(topic)) {
+    func = blockDollarSignTopics(func)
+  }
+
+  if (client.closed || client.broker.closed) {
+    // a hack, sometimes client.close() or broker.close() happened
+    // before authenticate() comes back
+    // we don't continue subscription here
+    return
+  }
+
+  if (!client.subscriptions[topic]) {
+    client.subscriptions[topic] = new Subscription(qos, func, rh, rap, nl)
+    broker.subscribe(topic, func, done)
+  } else if (client.subscriptions[topic].qos !== qos || client.subscriptions[topic].rh !== rh || client.subscriptions[topic].rap !== rap || client.subscriptions[topic].nl !== nl) {
+    broker.unsubscribe(topic, client.subscriptions[topic].func)
+    client.subscriptions[topic] = new Subscription(qos, func, rh, rap, nl)
+    broker.subscribe(topic, func, done)
+  } else {
+    done()
+  }
+}
+
+// + is 43
+// # is 35
+function isStartsWithWildcard (topic) {
+  const code = topic.charCodeAt(0)
+  return code === 43 || code === 35
+}
+
+function completeSubscribe (err) {
+  const done = this.finish
+
+  if (err) {
+    return done(err)
+  }
+
+  const packet = this.packet
+  const client = this.client
+
+  if (packet.messageId !== undefined) {
+    // [MQTT-3.9.3-1]
+    write(client,
+      new SubAck(packet, this.subState.map(obj => obj.granted)),
+      done)
+  } else {
+    done()
+  }
+
+  const broker = client.broker
+
+  // subscriptions array to return as result in 'subscribe' event and $SYS
+  const subs = packet.subscriptions
+
+  // topics we need to retrieve retain values from
+  const topics = []
+
+  for (let i = 0; i < subs.length; i++) {
+    // skip topics that are not allowed
+    if (this.subState[i].granted !== 128) {
+      topics.push(subs[i].topic)
+    }
+    // set granted qos to subscriptions
+    subs[i].qos = this.subState[i].granted
+  }
+
+  this.subState = []
+
+  broker.emit('subscribe', subs, client)
+  broker.publish({
+    topic: $SYS_PREFIX + broker.id + '/new/subscribes',
+    payload: Buffer.from(JSON.stringify({
+      clientId: client.id,
+      subs
+    }), 'utf8')
+  }, noop)
+
+  // Conform to MQTT 3.1.1 section 3.1.2.4
+  // Restored sessions should not contain any retained message.
+  // Retained message should be only fetched from SUBSCRIBE.
+
+  if (topics.length > 0) {
+    const persistence = broker.persistence
+    const stream = persistence.createRetainedStreamCombi(topics)
+    stream.pipe(through(function sendRetained (packet, enc, cb) {
+      packet = new Packet({
+        cmd: packet.cmd,
+        qos: packet.qos,
+        topic: packet.topic,
+        payload: packet.payload,
+        retain: true
+      }, broker)
+      // this should not be deduped
+      packet.brokerId = null
+      client.deliverQoS(packet, cb)
+    }))
+  }
+}
+
+function noop () { }
+
+module.exports = handleSubscribe
@@ -0,0 +1,104 @@
+'use strict'
+
+const write = require('../write')
+const { validateTopic, $SYS_PREFIX } = require('../utils')
+
+function UnSubAck (packet) {
+  this.cmd = 'unsuback'
+  this.messageId = packet.messageId // [MQTT-3.10.4-4]
+}
+
+function UnsubscribeState (client, packet, finish) {
+  this.client = client
+  this.packet = packet
+  this.finish = finish
+}
+
+function handleUnsubscribe (client, packet, done) {
+  const broker = client.broker
+  const unsubscriptions = packet.unsubscriptions
+  let err
+
+  for (let i = 0; i < unsubscriptions.length; i++) {
+    err = validateTopic(unsubscriptions[i], 'UNSUBSCRIBE')
+    if (err) {
+      return done(err)
+    }
+  }
+
+  if (packet.messageId !== undefined) {
+    if (client.clean) {
+      return actualUnsubscribe(client, packet, done)
+    }
+
+    broker.persistence.removeSubscriptions(client, unsubscriptions, function (err) {
+      if (err) {
+        return done(err)
+      }
+
+      actualUnsubscribe(client, packet, done)
+    })
+  } else {
+    actualUnsubscribe(client, packet, done)
+  }
+}
+
+function actualUnsubscribe (client, packet, done) {
+  const broker = client.broker
+  broker._parallel(
+    new UnsubscribeState(client, packet, done),
+    doUnsubscribe,
+    packet.unsubscriptions,
+    completeUnsubscribe)
+}
+
+function doUnsubscribe (sub, done) {
+  const client = this.client
+  const broker = client.broker
+  const s = client.subscriptions[sub]
+
+  if (s) {
+    const func = s.func
+    delete client.subscriptions[sub]
+    broker.unsubscribe(
+      sub,
+      func,
+      done)
+  } else {
+    done()
+  }
+}
+
+function completeUnsubscribe (err) {
+  const client = this.client
+
+  if (err) {
+    client.emit('error', err)
+    return
+  }
+
+  const packet = this.packet
+  const done = this.finish
+
+  if (packet.messageId !== undefined) {
+    write(client, new UnSubAck(packet),
+      done)
+  } else {
+    done()
+  }
+
+  if ((!client.closed || client.clean === true) && packet.unsubscriptions.length > 0) {
+    client.broker.emit('unsubscribe', packet.unsubscriptions, client)
+    client.broker.publish({
+      topic: $SYS_PREFIX + client.broker.id + '/new/unsubscribes',
+      payload: Buffer.from(JSON.stringify({
+        clientId: client.id,
+        subs: packet.unsubscriptions
+      }), 'utf8')
+    }, noop)
+  }
+}
+
+function noop () { }
+
+module.exports = handleUnsubscribe
@@ -0,0 +1,25 @@
+'use strict'
+
+const Packet = require('aedes-packet')
+const util = require('util')
+
+function QoSPacket (original, client) {
+  Packet.call(this, original, client.broker)
+
+  this.writeCallback = client._onError.bind(client)
+
+  if (!original.messageId) {
+    this.messageId = client._nextId
+    if (client._nextId >= 65535) {
+      client._nextId = 1
+    } else {
+      client._nextId++
+    }
+  } else {
+    this.messageId = original.messageId
+  }
+}
+
+util.inherits(QoSPacket, Packet)
+
+module.exports = QoSPacket
@@ -0,0 +1,56 @@
+'use strict'
+
+const { Transform, Writable } = require('stream')
+
+function validateTopic (topic, message) {
+  if (!topic || topic.length === 0) { // [MQTT-3.8.3-3]
+    return new Error('impossible to ' + message + ' to an empty topic')
+  }
+
+  const end = topic.length - 1
+  const endMinus = end - 1
+  const slashInPreEnd = endMinus > 0 && topic.charCodeAt(endMinus) !== 47
+
+  for (let i = 0; i < topic.length; i++) {
+    switch (topic.charCodeAt(i)) {
+      case 35: { // #
+        const notAtTheEnd = i !== end
+        if (notAtTheEnd || slashInPreEnd) {
+          return new Error('# is only allowed in ' + message + ' in the last position')
+        }
+        break
+      }
+      case 43: { // +
+        const pastChar = i < end - 1 && topic.charCodeAt(i + 1) !== 47
+        const preChar = i > 1 && topic.charCodeAt(i - 1) !== 47
+        if (pastChar || preChar) {
+          return new Error('+ is only allowed in ' + message + ' between /')
+        }
+        break
+      }
+    }
+  }
+}
+
+function through (transform) {
+  return new Transform({
+    objectMode: true,
+    transform
+  })
+}
+
+function bulk (fn) {
+  return new Writable({
+    objectMode: true,
+    writev: function (chunks, cb) {
+      fn(chunks.map(chunk => chunk.chunk), cb)
+    }
+  })
+}
+
+module.exports = {
+  validateTopic,
+  through,
+  bulk,
+  $SYS_PREFIX: '$SYS/'
+}
@@ -0,0 +1,24 @@
+'use strict'
+
+const mqtt = require('mqtt-packet')
+
+function write (client, packet, done) {
+  let error = null
+  if (client.connecting || client.connected) {
+    try {
+      const result = mqtt.writeToStream(packet, client.conn)
+      if (!result && !client.errored) {
+        client.conn.once('drain', done)
+        return
+      }
+    } catch (e) {
+      error = new Error('packet received not valid')
+    }
+  } else {
+    error = new Error('connection closed')
+  }
+
+  setImmediate(done, error, client)
+}
+
+module.exports = write