Repository: moscajs/aedes
Pull request #1032 (2025-06-19 08:37:26)

chore: migrate to ES6 classes
This PR migrates Aedes to ES6 class syntax.
This should help in further refactoring.
I tried to migrate everything that looked like a class but I might have missed one as `this` is also used in functions that are not constructors but whose this will be set by `parallel` etc.

Kind regards,
Hans

coveralls: ## Pull Request Test Coverage Report for [Build 15753454238](https://coveralls.io/builds/74233415)


### Details

* **337** of **337**   **(100.0%)**  changed or added relevant lines in **9** files are covered.
* No unchanged relevant lines lost coverage.
* Overall coverage decreased (**-0.007%**) to **99.568%**

---



|  Totals | [![Coverage Status](https://coveralls.io/builds/74233415/badge)](https://coveralls.io/builds/74233415) |
| :-- | --: |
| Change from base [Build 15752165255](https://coveralls.io/builds/74232472): |  -0.007% |
| Covered Lines: | 810 |
| Relevant Lines: | 811 |

---
##### 💛  - [Coveralls](https://coveralls.io)

seriousme: @robertsLando 
the constructor of Aedes immediatly starts running,e.g. it contains:

```js
  this._heartbeatInterval = setInterval(heartbeat, opts.heartbeatInterval)
```
 and:

```js
  that.publish({
      topic: birthTopic,
      payload: bufId
    }, noop)
```

My suggestion would be to migrate these actions to a separate async method as discussed before where we also can initialize async persistence as those action should imo come after persistence has been initialized.

We already have Aedes.createBroker() which we could repurpose for that.

As long as we are using the callback persistence we can automagically call this in the createBroker factory.
In the next major version the way to start Aedes could then be:

```js
const { Aedes } = require("aedes")
const aedes = await Aedes.createBroker(opts)
```
What do you think?

robertsLando: @seriousme can I merge?
seriousme: Yes you can.

Patch:
diff --git a/aedes.js b/aedes.js
index 283351b..52d36c2 100644
--- a/aedes.js
+++ b/aedes.js
@@ -1,7 +1,6 @@
 'use strict'
 
 const EventEmitter = require('events')
-const util = require('util')
 const parallel = require('fastparallel')
 const series = require('fastseries')
 const { v4: uuidv4 } = require('uuid')
@@ -13,8 +12,6 @@ const mqemitter = require('mqemitter')
 const Client = require('./lib/client')
 const { $SYS_PREFIX, bulk } = require('./lib/utils')
 
-module.exports = Aedes.createBroker = Aedes
-
 const defaultOptions = {
   concurrency: 100,
   heartbeatInterval: 60000, // 1 minute
@@ -33,161 +30,236 @@ const defaultOptions = {
   keepaliveLimit: 0
 }
 
-function Aedes (opts) {
-  const that = this
-
-  if (!(this instanceof Aedes)) {
-    return new Aedes(opts)
-  }
-
-  opts = Object.assign({}, defaultOptions, opts)
-
-  this.id = opts.id || uuidv4()
-  // +1 when construct a new aedes-packet
-  // internal track for last brokerCounter
-  this.counter = 0
-  this.queueLimit = opts.queueLimit
-  this.connectTimeout = opts.connectTimeout
-  this.keepaliveLimit = opts.keepaliveLimit
-  this.maxClientsIdLength = opts.maxClientsIdLength
-  this.mq = opts.mq || mqemitter({
-    concurrency: opts.concurrency,
-    matchEmptyLevels: true // [MQTT-4.7.1-3]
-  })
-  this.handle = function handle (conn, req) {
-    conn.setMaxListeners(opts.concurrency * 2)
-    // create a new Client instance for a new connection
-    // return, just to please standard
-    return new Client(that, conn, req)
-  }
-  this.persistence = opts.persistence || memory()
-  this.persistence.broker = this
-  this._parallel = parallel()
-  this._series = series()
-  this._enqueuers = reusify(DoEnqueues)
-
-  this.preConnect = opts.preConnect
-  this.authenticate = opts.authenticate
-  this.authorizePublish = opts.authorizePublish
-  this.authorizeSubscribe = opts.authorizeSubscribe
-  this.authorizeForward = opts.authorizeForward
-  this.published = opts.published
+const version = require('./package.json').version
+
+class Aedes extends EventEmitter {
+  constructor (opts) {
+    super()
+    const that = this
+
+    opts = Object.assign({}, defaultOptions, opts)
+    this.id = opts.id || uuidv4()
+    // +1 when construct a new aedes-packet
+    // internal track for last brokerCounter
+    this.counter = 0
+    this.queueLimit = opts.queueLimit
+    this.connectTimeout = opts.connectTimeout
+    this.keepaliveLimit = opts.keepaliveLimit
+    this.maxClientsIdLength = opts.maxClientsIdLength
+    this.mq = opts.mq || mqemitter({
+      concurrency: opts.concurrency,
+      matchEmptyLevels: true // [MQTT-4.7.1-3]
+    })
+    this.handle = function handle (conn, req) {
+      conn.setMaxListeners(opts.concurrency * 2)
+      // create a new Client instance for a new connection
+      // return, just to please standard
+      return new Client(that, conn, req)
+    }
+    this.persistence = opts.persistence || memory()
+    this.persistence.broker = this
+    this._parallel = parallel()
+    this._series = series()
+    this._enqueuers = reusify(DoEnqueues)
 
-  this.decodeProtocol = opts.decodeProtocol
-  this.trustProxy = opts.trustProxy
-  this.trustedProxies = opts.trustedProxies
+    this.preConnect = opts.preConnect
+    this.authenticate = opts.authenticate
+    this.authorizePublish = opts.authorizePublish
+    this.authorizeSubscribe = opts.authorizeSubscribe
+    this.authorizeForward = opts.authorizeForward
+    this.published = opts.published
 
-  this.clients = {}
-  this.brokers = {}
+    this.decodeProtocol = opts.decodeProtocol
+    this.trustProxy = opts.trustProxy
+    this.trustedProxies = opts.trustedProxies
 
-  const heartbeatTopic = $SYS_PREFIX + that.id + '/heartbeat'
-  const birthTopic = $SYS_PREFIX + that.id + '/birth'
+    this.clients = {}
+    this.brokers = {}
 
-  this._heartbeatInterval = setInterval(heartbeat, opts.heartbeatInterval)
+    const heartbeatTopic = $SYS_PREFIX + that.id + '/heartbeat'
+    const birthTopic = $SYS_PREFIX + that.id + '/birth'
 
-  const bufId = Buffer.from(that.id, 'utf8')
+    this._heartbeatInterval = setInterval(heartbeat, opts.heartbeatInterval)
 
-  // in a cluster env this is used to warn other broker instances
-  // that this broker is alive
-  that.publish({
-    topic: birthTopic,
-    payload: bufId
-  }, noop)
+    const bufId = Buffer.from(that.id, 'utf8')
 
-  function heartbeat () {
+    // in a cluster env this is used to warn other broker instances
+    // that this broker is alive
     that.publish({
-      topic: heartbeatTopic,
+      topic: birthTopic,
       payload: bufId
     }, noop)
-  }
 
-  function deleteOldBrokers (broker) {
-    if (that.brokers[broker] + (3 * opts.heartbeatInterval) < Date.now()) {
-      delete that.brokers[broker]
+    function heartbeat () {
+      that.publish({
+        topic: heartbeatTopic,
+        payload: bufId
+      }, noop)
     }
-  }
 
-  this._clearWillInterval = setInterval(function () {
-    Object.keys(that.brokers).forEach(deleteOldBrokers)
+    function deleteOldBrokers (broker) {
+      if (that.brokers[broker] + (3 * opts.heartbeatInterval) < Date.now()) {
+        delete that.brokers[broker]
+      }
+    }
+
+    this._clearWillInterval = setInterval(function () {
+      Object.keys(that.brokers).forEach(deleteOldBrokers)
 
-    pipeline(
-      that.persistence.streamWill(that.brokers),
-      bulk(receiveWills),
-      function done (err) {
-        if (err) {
-          that.emit('error', err)
+      pipeline(
+        that.persistence.streamWill(that.brokers),
+        bulk(receiveWills),
+        function done (err) {
+          if (err) {
+            that.emit('error', err)
+          }
         }
-      }
-    )
-  }, opts.heartbeatInterval * 4)
+      )
+    }, opts.heartbeatInterval * 4)
 
-  function receiveWills (chunks, done) {
-    that._parallel(that, checkAndPublish, chunks, done)
-  }
+    function receiveWills (chunks, done) {
+      that._parallel(that, checkAndPublish, chunks, done)
+    }
+
+    function checkAndPublish (will, done) {
+      const notPublish = that.brokers[will.brokerId] !== undefined && that.brokers[will.brokerId] + (3 * opts.heartbeatInterval) >= Date.now()
+
+      if (notPublish) return done()
+
+      // randomize this, so that multiple brokers
+      // do not publish the same wills at the same time
+      this.authorizePublish(that.clients[will.clientId] || null, will, function (err) {
+        if (err) { return doneWill() }
+        that.publish(will, doneWill)
+
+        function doneWill (err) {
+          if (err) { return done(err) }
+          that.persistence.delWill({
+            id: will.clientId,
+            brokerId: will.brokerId
+          }, done)
+        }
+      })
+    }
 
-  function checkAndPublish (will, done) {
-    const notPublish =
-      that.brokers[will.brokerId] !== undefined && that.brokers[will.brokerId] + (3 * opts.heartbeatInterval) >= Date.now()
+    this.mq.on($SYS_PREFIX + '+/heartbeat', function storeBroker (packet, done) {
+      that.brokers[packet.payload.toString()] = Date.now()
+      done()
+    })
+
+    this.mq.on($SYS_PREFIX + '+/birth', function brokerBorn (packet, done) {
+      const brokerId = packet.payload.toString()
 
-    if (notPublish) return done()
+      // reset duplicates counter
+      if (brokerId !== that.id) {
+        for (const clientId in that.clients) {
+          delete that.clients[clientId].duplicates[brokerId]
+        }
+      }
 
-    // randomize this, so that multiple brokers
-    // do not publish the same wills at the same time
-    this.authorizePublish(that.clients[will.clientId] || null, will, function (err) {
-      if (err) { return doneWill() }
-      that.publish(will, doneWill)
+      done()
+    })
 
-      function doneWill (err) {
-        if (err) { return done(err) }
-        that.persistence.delWill({
-          id: will.clientId,
-          brokerId: will.brokerId
-        }, done)
+    this.mq.on($SYS_PREFIX + '+/new/clients', function closeSameClients (packet, done) {
+      const serverId = packet.topic.split('/')[1]
+      const clientId = packet.payload.toString()
+
+      if (that.clients[clientId] && serverId !== that.id) {
+        if (that.clients[clientId].closed) {
+          // remove the client from the list if it is already closed
+          that.deleteClient(clientId)
+          done()
+        } else {
+          that.clients[clientId].close(done)
+        }
+      } else {
+        done()
       }
     })
+
+    // metadata
+    this.connectedClients = 0
+    this.closed = false
   }
 
-  this.mq.on($SYS_PREFIX + '+/heartbeat', function storeBroker (packet, done) {
-    that.brokers[packet.payload.toString()] = Date.now()
-    done()
-  })
+  get version () {
+    return version
+  }
 
-  this.mq.on($SYS_PREFIX + '+/birth', function brokerBorn (packet, done) {
-    const brokerId = packet.payload.toString()
+  static createBroker (opts) {
+    return new Aedes(opts)
+  }
 
-    // reset duplicates counter
-    if (brokerId !== that.id) {
-      for (const clientId in that.clients) {
-        delete that.clients[clientId].duplicates[brokerId]
-      }
+  publish (packet, client, done) {
+    if (typeof client === 'function') {
+      done = client
+      client = null
     }
+    const p = new Packet(packet, this)
+    const publishFuncs = p.qos > 0 ? publishFuncsQoS : publishFuncsSimple
 
-    done()
-  })
+    this._series(new PublishState(this, client, packet), publishFuncs, p, done)
+  }
 
-  this.mq.on($SYS_PREFIX + '+/new/clients', function closeSameClients (packet, done) {
-    const serverId = packet.topic.split('/')[1]
-    const clientId = packet.payload.toString()
+  subscribe (topic, func, done) {
+    this.mq.on(topic, func, done)
+  }
 
-    if (that.clients[clientId] && serverId !== that.id) {
-      if (that.clients[clientId].closed) {
-        // remove the client from the list if it is already closed
-        that.deleteClient(clientId)
-        done()
-      } else {
-        that.clients[clientId].close(done)
-      }
+  unsubscribe (topic, func, done) {
+    this.mq.removeListener(topic, func, done)
+  }
+
+  registerClient (client) {
+    const that = this
+    if (this.clients[client.id]) {
+      // [MQTT-3.1.4-2]
+      this.clients[client.id].close(function closeClient () {
+        that._finishRegisterClient(client)
+      })
     } else {
-      done()
+      this._finishRegisterClient(client)
     }
-  })
+  }
 
-  // metadata
-  this.connectedClients = 0
-  this.closed = false
-}
+  _finishRegisterClient (client) {
+    this.connectedClients++
+    this.clients[client.id] = client
+    this.emit('client', client)
+    this.publish({
+      topic: $SYS_PREFIX + this.id + '/new/clients',
+      payload: Buffer.from(client.id, 'utf8')
+    }, noop)
+  }
 
-util.inherits(Aedes, EventEmitter)
+  unregisterClient (client) {
+    this.deleteClient(client.id)
+    this.emit('clientDisconnect', client)
+    this.publish({
+      topic: $SYS_PREFIX + this.id + '/disconnect/clients',
+      payload: Buffer.from(client.id, 'utf8')
+    }, noop)
+  }
+
+  deleteClient (clientId) {
+    this.connectedClients--
+    delete this.clients[clientId]
+  }
+
+  close (cb = noop) {
+    const that = this
+    if (this.closed) {
+      return cb()
+    }
+    this.closed = true
+    clearInterval(this._heartbeatInterval)
+    clearInterval(this._clearWillInterval)
+    this._parallel(this, closeClient, Object.keys(this.clients), doneClose)
+    function doneClose () {
+      that.emit('closed')
+      that.mq.close(cb)
+    }
+  }
+}
 
 function storeRetained (packet, done) {
   if (packet.retain) {
@@ -215,39 +287,41 @@ function enqueueOffline (packet, done) {
   )
 }
 
-function DoEnqueues () {
-  this.next = null
-  this.complete = null
-  this.packet = null
-  this.topic = null
-  this.broker = null
+class DoEnqueues {
+  constructor () {
+    this.next = null
+    this.complete = null
+    this.packet = null
+    this.topic = null
+    this.broker = null
+
+    const that = this
+
+    this.done = function doneEnqueue (err, subs) {
+      const broker = that.broker
+
+      if (err) {
+        // is this really recoverable?
+        // let's just error the whole aedes
+        // https://nodejs.org/api/events.html#events_error_events
+        broker.emit('error', err)
+        return
+      }
 
-  const that = this
+      if (that.topic.indexOf($SYS_PREFIX) === 0) {
+        subs = subs.filter(removeSharp)
+      }
 
-  this.done = function doneEnqueue (err, subs) {
-    const broker = that.broker
+      const packet = that.packet
+      const complete = that.complete
 
-    if (err) {
-      // is this really recoverable?
-      // let's just error the whole aedes
-      // https://nodejs.org/api/events.html#events_error_events
-      broker.emit('error', err)
-      return
-    }
+      that.packet = null
+      that.complete = null
+      that.topic = null
 
-    if (that.topic.indexOf($SYS_PREFIX) === 0) {
-      subs = subs.filter(removeSharp)
+      broker.persistence.outgoingEnqueueCombi(subs, packet, complete)
+      broker._enqueuers.release(that)
     }
-
-    const packet = that.packet
-    const complete = that.complete
-
-    that.packet = null
-    that.complete = null
-    that.topic = null
-
-    broker.persistence.outgoingEnqueueCombi(subs, packet, complete)
-    broker._enqueuers.release(that)
   }
 }
 
@@ -274,82 +348,11 @@ const publishFuncsQoS = [
   emitPacket,
   callPublished
 ]
-Aedes.prototype.publish = function (packet, client, done) {
-  if (typeof client === 'function') {
-    done = client
-    client = null
-  }
-  const p = new Packet(packet, this)
-  const publishFuncs = p.qos > 0 ? publishFuncsQoS : publishFuncsSimple
-
-  this._series(new PublishState(this, client, packet), publishFuncs, p, done)
-}
-
-Aedes.prototype.subscribe = function (topic, func, done) {
-  this.mq.on(topic, func, done)
-}
-
-Aedes.prototype.unsubscribe = function (topic, func, done) {
-  this.mq.removeListener(topic, func, done)
-}
-
-Aedes.prototype.registerClient = function (client) {
-  const that = this
-  if (this.clients[client.id]) {
-    // [MQTT-3.1.4-2]
-    this.clients[client.id].close(function closeClient () {
-      that._finishRegisterClient(client)
-    })
-  } else {
-    this._finishRegisterClient(client)
-  }
-}
-
-Aedes.prototype._finishRegisterClient = function (client) {
-  this.connectedClients++
-  this.clients[client.id] = client
-  this.emit('client', client)
-  this.publish({
-    topic: $SYS_PREFIX + this.id + '/new/clients',
-    payload: Buffer.from(client.id, 'utf8')
-  }, noop)
-}
-
-Aedes.prototype.unregisterClient = function (client) {
-  this.deleteClient(client.id)
-  this.emit('clientDisconnect', client)
-  this.publish({
-    topic: $SYS_PREFIX + this.id + '/disconnect/clients',
-    payload: Buffer.from(client.id, 'utf8')
-  }, noop)
-}
-
-Aedes.prototype.deleteClient = function (clientId) {
-  this.connectedClients--
-  delete this.clients[clientId]
-}
 
 function closeClient (client, cb) {
   this.clients[client].close(cb)
 }
 
-Aedes.prototype.close = function (cb = noop) {
-  const that = this
-  if (this.closed) {
-    return cb()
-  }
-  this.closed = true
-  clearInterval(this._heartbeatInterval)
-  clearInterval(this._clearWillInterval)
-  this._parallel(this, closeClient, Object.keys(this.clients), doneClose)
-  function doneClose () {
-    that.emit('closed')
-    that.mq.close(cb)
-  }
-}
-
-Aedes.prototype.version = require('./package.json').version
-
 function defaultPreConnect (client, packet, callback) {
   callback(null, true)
 }
@@ -377,10 +380,16 @@ function defaultPublished (packet, client, callback) {
   callback(null)
 }
 
-function PublishState (broker, client, packet) {
-  this.broker = broker
-  this.client = client
-  this.packet = packet
+class PublishState {
+  constructor (broker, client, packet) {
+    this.broker = broker
+    this.client = client
+    this.packet = packet
+  }
 }
 
 function noop () {}
+
+module.exports = Aedes.createBroker
+module.exports.createBroker = Aedes.createBroker
+module.exports.Aedes = Aedes
diff --git a/lib/client.js b/lib/client.js
index e5ee42d..ec0f224 100644
--- a/lib/client.js
+++ b/lib/client.js
@@ -13,151 +13,322 @@ 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
-    }
+class Client {
+  constructor (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
+      const client = that
 
-    if (client._paused) {
-      return
-    }
+      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)
+      that._parsingBatch--
+      if (that._parsingBatch <= 0) {
+        that._parsingBatch = 0
+        const buf = client.conn.read(null)
+        if (buf) {
+          client._parser.parse(buf)
+        }
       }
     }
-  }
-  this._nextBatch = nextBatch
+    this._nextBatch = nextBatch
 
-  conn.on('readable', 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'))
+    this.on('error', this._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))
+    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 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)
+      const toForward = dedupe(that, _packet) &&
+        that.broker.authorizeForward(that, _packet)
 
-    return toForward
-  }
+      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
+    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)
+      } 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)
   }
 
-  this.deliverQoS = function deliverQoS (_packet, cb) {
-    // downgrade to qos0 if requested by publish
-    if (_packet.qos === 0) {
-      that.deliver0(_packet, cb)
+  _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()
+  }
+
+  publish (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
     }
-    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)
+    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 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)
+      that.deliverQoS(packet, done)
+    }
+  }
+
+  subscribe (packet, done) {
+    if (!packet.subscriptions) {
+      if (!Array.isArray(packet)) {
+        packet = [packet]
+      }
+      packet = {
+        subscriptions: packet
+      }
     }
+    handleSubscribe(this, packet, false, done)
   }
 
-  this._keepaliveTimer = null
-  this._keepaliveInterval = -1
+  unsubscribe (packet, done) {
+    if (!packet.unsubscriptions) {
+      if (!Array.isArray(packet)) {
+        packet = [packet]
+      }
+      packet = {
+        unsubscriptions: packet
+      }
+    }
+    handleUnsubscribe(this, packet, done)
+  }
+
+  close (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()
+      }
+    }
+  }
+
+  pause () {
+    this._paused = true
+  }
 
-  this._connectTimer = setTimeout(function () {
-    that.emit('error', new Error('connect did not arrive in time'))
-  }, broker.connectTimeout)
+  resume () {
+    this._paused = false
+    this._nextBatch()
+  }
+
+  emptyOutgoingQueue (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 dedupe (client, packet) {
@@ -195,166 +366,8 @@ 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++
@@ -381,19 +394,5 @@ function dequeue () {
   }
 }
 
-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 () {}
+module.exports = Client
diff --git a/lib/handlers/connect.js b/lib/handlers/connect.js
index 43011d2..5d2a5a2 100644
--- a/lib/handlers/connect.js
+++ b/lib/handlers/connect.js
@@ -8,15 +8,19 @@ 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
+class Connack {
+  constructor (arg) {
+    this.cmd = 'connack'
+    this.returnCode = arg.returnCode
+    this.sessionPresent = arg.sessionPresent
+  }
 }
 
-function ClientPacketStatus (client, packet) {
-  this.client = client
-  this.packet = packet
+class ClientPacketStatus {
+  constructor (client, packet) {
+    this.client = client
+    this.packet = packet
+  }
 }
 
 const connectActions = [
diff --git a/lib/handlers/publish.js b/lib/handlers/publish.js
index e30c9db..e297769 100644
--- a/lib/handlers/publish.js
+++ b/lib/handlers/publish.js
@@ -2,14 +2,18 @@
 
 const write = require('../write')
 
-function PubAck (packet) {
-  this.cmd = 'puback'
-  this.messageId = packet.messageId
+class PubAck {
+  constructor (packet) {
+    this.cmd = 'puback'
+    this.messageId = packet.messageId
+  }
 }
 
-function PubRec (packet) {
-  this.cmd = 'pubrec'
-  this.messageId = packet.messageId
+class PubRec {
+  constructor (packet) {
+    this.cmd = 'pubrec'
+    this.messageId = packet.messageId
+  }
 }
 
 const publishActions = [
diff --git a/lib/handlers/pubrec.js b/lib/handlers/pubrec.js
index 5c914dd..2197b8d 100644
--- a/lib/handlers/pubrec.js
+++ b/lib/handlers/pubrec.js
@@ -2,9 +2,11 @@
 
 const write = require('../write')
 
-function PubRel (packet) {
-  this.cmd = 'pubrel'
-  this.messageId = packet.messageId
+class PubRel {
+  constructor (packet) {
+    this.cmd = 'pubrel'
+    this.messageId = packet.messageId
+  }
 }
 
 function handlePubrec (client, packet, done) {
diff --git a/lib/handlers/pubrel.js b/lib/handlers/pubrel.js
index 09dcc86..8fc39be 100644
--- a/lib/handlers/pubrel.js
+++ b/lib/handlers/pubrel.js
@@ -2,14 +2,18 @@
 
 const write = require('../write')
 
-function ClientPacketStatus (client, packet) {
-  this.client = client
-  this.packet = packet
+class ClientPacketStatus {
+  constructor (client, packet) {
+    this.client = client
+    this.packet = packet
+  }
 }
 
-function PubComp (packet) {
-  this.cmd = 'pubcomp'
-  this.messageId = packet.messageId
+class PubComp {
+  constructor (packet) {
+    this.cmd = 'pubcomp'
+    this.messageId = packet.messageId
+  }
 }
 
 const pubrelActions = [
diff --git a/lib/handlers/subscribe.js b/lib/handlers/subscribe.js
index 2ae0a46..bd04988 100644
--- a/lib/handlers/subscribe.js
+++ b/lib/handlers/subscribe.js
@@ -16,47 +16,55 @@ const restoreTopicActions = fastfall([
   addSubs
 ])
 
-function SubAck (packet, granted) {
-  this.cmd = 'suback'
-  this.messageId = packet.messageId
-  // the qos granted
-  this.granted = granted
+class SubAck {
+  constructor (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
+class Subscription {
+  constructor (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-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
+    // 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
+    // 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 = []
+class SubscribeState {
+  constructor (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
+class SubState {
+  constructor (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
diff --git a/lib/handlers/unsubscribe.js b/lib/handlers/unsubscribe.js
index e08c317..874b31b 100644
--- a/lib/handlers/unsubscribe.js
+++ b/lib/handlers/unsubscribe.js
@@ -3,15 +3,19 @@
 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]
+class UnSubAck {
+  constructor (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
+class UnsubscribeState {
+  constructor (client, packet, finish) {
+    this.client = client
+    this.packet = packet
+    this.finish = finish
+  }
 }
 
 function handleUnsubscribe (client, packet, done) {
diff --git a/lib/qos-packet.js b/lib/qos-packet.js
index 5527fe1..ebace14 100644
--- a/lib/qos-packet.js
+++ b/lib/qos-packet.js
@@ -1,25 +1,24 @@
 'use strict'
 
 const Packet = require('aedes-packet')
-const util = require('util')
 
-function QoSPacket (original, client) {
-  Packet.call(this, original, client.broker)
+class QoSPacket extends Packet {
+  constructor (original, client) {
+    super(original, client.broker)
 
-  this.writeCallback = client._onError.bind(client)
+    this.writeCallback = client._onError.bind(client)
 
-  if (!original.messageId) {
-    this.messageId = client._nextId
-    if (client._nextId >= 65535) {
-      client._nextId = 1
+    if (!original.messageId) {
+      this.messageId = client._nextId
+      if (client._nextId >= 65535) {
+        client._nextId = 1
+      } else {
+        client._nextId++
+      }
     } else {
-      client._nextId++
+      this.messageId = original.messageId
     }
-  } else {
-    this.messageId = original.messageId
   }
 }
 
-util.inherits(QoSPacket, Packet)
-
 module.exports = QoSPacket


Test patch:
diff --git a/test/basic.js b/test/basic.js
index 6ca103f..cb53e7c 100644
--- a/test/basic.js
+++ b/test/basic.js
@@ -4,8 +4,16 @@ const { test } = require('tap')
 const eos = require('end-of-stream')
 const { setup, connect, subscribe, subscribeMultiple, noError } = require('./helper')
 const aedes = require('../')
+const { Aedes } = require('../')
 const proxyquire = require('proxyquire')
 
+test('test Aedes constructor', function (t) {
+  t.plan(1)
+  const aedes = new Aedes()
+  t.teardown(aedes.close.bind(aedes))
+  t.equal(aedes instanceof Aedes, true, 'Aedes constructor works')
+})
+
 test('test aedes.createBroker', function (t) {
   t.plan(1)