feat!: migrate to async persistence (#1034)

This PR migrates Aedes to use the async persistence interface.

It contains the following sub items:

- [X] replace all callback calls to persistence by `.then()` calls
- [X] move all side effects (setting up persistence, attaching event
handlers etc) to a `listen()` method of Aedes
- [X] update `createBroker()` to be async and make it await `listen()`
as well
- [X] update exports so that they produce a warning if people use old
style calling
- [X] update Typescript typing
- [X] update documentation including examples
- [X] add a `migration.md` to help users migrate.
- [X] do some benchmark tests to see performances are kept
- [X] update node engines version to >=20
- [X] updates all dependencies to current except for Tap and
@sinonjs/fake-timers

BREAKING CHANGES: See
[docs](https://github.com/moscajs/aedes/blob/main/docs/MIGRATION.md)

Author: seriousme

README.md

@@ -56,8 +56,8 @@ Check the [Docker docs](https://github.com/moscajs/aedes-cli#docker)
 
 ## API
 
-- [Aedes object](./docs/Aedes.md)
-- [Client object](./docs/Client.md)
+- [Aedes class](./docs/Aedes.md)
+- [Client class](./docs/Client.md)
 
 ## Features
 

aedes.d.ts

@@ -1,9 +1,3 @@
-import Aedes, { AedesOptions } from './types/instance'
-
-export declare function createBroker (options?: AedesOptions): Aedes
-
 export * from './types/instance'
 export * from './types/packet'
 export * from './types/client'
-
-export { default } from './types/instance'

aedes.js

@@ -38,6 +38,7 @@ class Aedes extends EventEmitter {
     const that = this
 
     opts = Object.assign({}, defaultOptions, opts)
+    this.opts = opts
     this.id = opts.id || uuidv4()
     // +1 when construct a new aedes-packet
     // internal track for last brokerCounter
@@ -56,8 +57,7 @@ class Aedes extends EventEmitter {
       // 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)
@@ -75,6 +75,22 @@ class Aedes extends EventEmitter {
 
     this.clients = {}
     this.brokers = {}
+    this.closed = true
+  }
+
+  async listen () {
+    const opts = this.opts
+    const that = this
+
+    // metadata
+    this.connectedClients = 0
+    this.closed = false
+
+    this.persistence = opts.persistence || memory()
+    if (this.persistence.setup.constructor.name !== 'AsyncFunction') {
+      throw new Error('persistence.setup() must be an async function')
+    }
+    await this.persistence.setup(this)
 
     const heartbeatTopic = $SYS_PREFIX + that.id + '/heartbeat'
     const birthTopic = $SYS_PREFIX + that.id + '/birth'
@@ -137,7 +153,7 @@ class Aedes extends EventEmitter {
           that.persistence.delWill({
             id: will.clientId,
             brokerId: will.brokerId
-          }, done)
+          }).then(will => done(undefined, will), done)
         }
       })
     }
@@ -176,18 +192,16 @@ class Aedes extends EventEmitter {
         done()
       }
     })
-
-    // metadata
-    this.connectedClients = 0
-    this.closed = false
   }
 
   get version () {
     return version
   }
 
-  static createBroker (opts) {
-    return new Aedes(opts)
+  static async createBroker (opts) {
+    const aedes = new Aedes(opts)
+    await aedes.listen()
+    return aedes
   }
 
   publish (packet, client, done) {
@@ -263,7 +277,8 @@ class Aedes extends EventEmitter {
 
 function storeRetained (packet, done) {
   if (packet.retain) {
-    this.broker.persistence.storeRetained(packet, done)
+    this.broker.persistence.storeRetained(packet)
+      .then(() => done(null), done)
   } else {
     done()
   }
@@ -281,10 +296,8 @@ function enqueueOffline (packet, done) {
   enqueuer.packet = packet
   enqueuer.topic = packet.topic
   enqueuer.broker = this.broker
-  this.broker.persistence.subscriptionsByTopic(
-    packet.topic,
-    enqueuer.done
-  )
+  this.broker.persistence.subscriptionsByTopic(packet.topic)
+    .then(subs => enqueuer.done(null, subs), enqueuer.done)
 }
 
 class DoEnqueues {
@@ -319,7 +332,8 @@ class DoEnqueues {
       that.complete = null
       that.topic = null
 
-      broker.persistence.outgoingEnqueueCombi(subs, packet, complete)
+      broker.persistence.outgoingEnqueueCombi(subs, packet)
+        .then(() => complete(null), complete)
       broker._enqueuers.release(that)
     }
   }
@@ -390,6 +404,14 @@ class PublishState {
 
 function noop () {}
 
-module.exports = Aedes.createBroker
+function warnMigrate () {
+  throw new Error(
+` Aedes default export has been removed.
+ Use 'const aedes = await Aedes.createBroker()' instead.
+ See: https://github.com/moscajs/aedes/docs/MIGRATION.MD
+ `)
+}
+
+module.exports = warnMigrate
 module.exports.createBroker = Aedes.createBroker
 module.exports.Aedes = Aedes

benchmarks/server.js

@@ -2,24 +2,27 @@
 
 // To be used with cpuprofilify http://npm.im/cpuprofilify
 
-const aedes = require('../')()
-const server = require('net').createServer(aedes.handle)
-const port = 1883
+const { Aedes } = require('../')
 
-server.listen(port, function () {
-  console.error('server listening on port', port, 'pid', process.pid)
-})
+Aedes.createBroker().then(aedes => {
+  const server = require('net').createServer(aedes.handle)
+  const port = 1883
 
-aedes.on('clientError', function (client, err) {
-  console.error('client error', client.id, err.message)
-})
+  server.listen(port, function () {
+    console.error('server listening on port', port, 'pid', process.pid)
+  })
 
-// Cleanly shut down process on SIGTERM to ensure that perf-<pid>.map gets flushed
-process.on('SIGINT', onSIGINT)
+  aedes.on('clientError', function (client, err) {
+    console.error('client error', client.id, err.message)
+  })
 
-function onSIGINT () {
+  // Cleanly shut down process on SIGTERM to ensure that perf-<pid>.map gets flushed
+  process.on('SIGINT', onSIGINT)
+
+  function onSIGINT () {
   // IMPORTANT to log on stderr, to not clutter stdout which is purely for data, i.e. dtrace stacks
-  console.error('Caught SIGTERM, shutting down.')
-  server.close()
-  process.exit(0)
-}
+    console.error('Caught SIGTERM, shutting down.')
+    server.close()
+    process.exit(0)
+  }
+})

docs/Aedes.md

@@ -2,7 +2,9 @@
 # Aedes
 
 - [Aedes](#aedes)
-  - [new Aedes([options]) / new Aedes.Server([options])](#new-aedesoptions--new-aedesserveroptions)
+  - [new Aedes([options])](#new-aedesoptions)
+  - [Aedes.createBroker([options])](#aedescreatebrokeroptions)
+  - [aedes.listen()](#aedeslisten)
   - [aedes.id](#aedesid)
   - [aedes.connectedClients](#aedesconnectedclients)
   - [aedes.closed](#aedesclosed)
@@ -31,12 +33,13 @@
   - [Handler: authorizeForward (client, packet)](#handler-authorizeforward-client-packet)
   - [Handler: published (packet, client, callback)](#handler-published-packet-client-callback)
 
-## new Aedes([options]) / new Aedes.Server([options])
+## new Aedes([options])
 
 - options `<object>`
   - `mq` [`<MQEmitter>`](../README.md#mqemitter) middleware used to deliver messages to subscribed clients. In a cluster environment it is used also to share messages between brokers instances. __Default__: `mqemitter`
   - `concurrency` `<number>` maximum number of concurrent messages delivered by `mq`. __Default__: `100`
   - `persistence` [`<Persistence>`](../README.md#persistence) middleware that stores _QoS > 0, retained, will_ packets and _subscriptions_. __Default__: `aedes-persistence` (_in memory_)
+  Versions 1.x and above require persistence to support async access,see [MIGRATION.md][MIGRATION] for details.
   - `queueLimit` `<number>` maximum number of queued messages before client session is established. If number of queued items exceeds, `connectionError` throws an error `Client queue limit reached`. __Default__: `42`
   - `maxClientsIdLength` option to override MQTT 3.1.0 clients Id length limit. __Default__: `23`
   - `heartbeatInterval` `<number>` an interval in millisconds at which server beats its health signal in `$SYS/<aedes.id>/heartbeat` topic. __Default__: `60000`
@@ -45,9 +48,36 @@
   - `keepaliveLimit` `<number>` maximum client keep alive time allowed, 0 means no limit. __Default__: `0`
 - Returns `<Aedes>`
 
-Create a new Aedes server.
+Create a new Aedes server instance.
 
-Aedes is the class and function exposed by this module. It can be created by `Aedes()` or using `new Aedes()`. An variant `aedes.Server` is for TypeScript or ES modules.
+Aedes is the class exported by this module.
+The instance will only start listening after [aedes.listen()](#aedeslisten) is called.
+The recommended way to start an Aedes server is to use [Aedes.createBroker([options])](#aedescreatebrokeroptions) instead.
+
+## Aedes.createBroker([options])
+
+An async static method in the Aedes class which creates the instance and automatically awaits `listen()`.
+
+Using `Aedes.createBroker([options])` is the recommended way to start Aedes, example:
+
+```js
+const aedes = await Aedes.createBroker([options])
+```
+
+It uses the same options as [new Aedes([options])](#new-aedesoptions)
+
+## aedes.listen()
+
+Async method to make the aedes instance start listening.
+Example:
+
+```js
+const aedes = new Aedes([options])
+await aedes.listen()
+```
+
+You should typically not need to use this as it is more compact to use
+[Aedes.createBroker([options])](#aedescreatebrokeroptions) instead.
 
 ## aedes.id
 
@@ -173,7 +203,9 @@ Emitted when server is closed.
 A connection listener that pipe stream to aedes.
 
 ```js
-const aedes = require('./aedes')()
+const { Aedes } = require('/aedes')
+const aedes = await Aedes.createBroker()
+
 const server = require('net').createServer(aedes.handle)
 ```
 
@@ -414,3 +446,4 @@ same as [`Event: publish`](#event-publish), but provides a backpressure function
 [PINGREQ]: https://github.com/mqttjs/mqtt-packet#pingreq
 [PUBLISH]: https://github.com/mqttjs/mqtt-packet#publish
 [PUBREL]: https://github.com/mqttjs/mqtt-packet#pubrel
+[MIGRATION]: MIGRATION.md

docs/Examples.md

@@ -4,9 +4,13 @@
 ## Simple plain MQTT server
 
 ```js
-const aedes = require('aedes')()
-const server = require('net').createServer(aedes.handle)
+
+const { Aedes } = require('aedes')
+const { createServer } = require('net')
+
+const aedes = await Aedes.createBroker()
 const port = 1883
+const server = createServer(aedes.handle)
 
 server.listen(port, function () {
   console.log('server started and listening on port ', port)
@@ -21,7 +25,7 @@ import { createServer } from 'net'
 
 const port = 1883
 
-const aedes = new Aedes()
+const aedes = await Aedes.createBroker()
 const server = createServer(aedes.handle)
 
 server.listen(port, function () {
@@ -32,10 +36,11 @@ server.listen(port, function () {
 ## Simple plain MQTT server using server-factory
 
 ```js
-const aedes = require('aedes')()
+const { Aedes } = require('aedes')
 const { createServer } = require('aedes-server-factory')
 const port = 1883
 
+const aedes = await Aedes.createBroker()
 const server = createServer(aedes)
 
 server.listen(port, function () {
@@ -47,14 +52,14 @@ server.listen(port, function () {
 
 ```js
 const fs = require('fs')
-const aedes = require('aedes')()
+const { Aedes } = require('aedes')
 const port = 8883
 
 const options = {
   key: fs.readFileSync('YOUR_PRIVATE_KEY_FILE.pem'),
   cert: fs.readFileSync('YOUR_PUBLIC_CERT_FILE.pem')
 }
-
+const aedes = await Aedes.createBroker()
 const server = require('tls').createServer(options, aedes.handle)
 
 server.listen(port, function () {
@@ -65,11 +70,13 @@ server.listen(port, function () {
 ## MQTT server over WebSocket
 
 ```js
-const aedes = require('aedes')()
-const httpServer = require('http').createServer()
+const { Aedes } = require('aedes')
+const { createServer } require('http')
 const ws = require('ws')
 const port = 8888
 
+const aedes = await Aedes.createBroker()
+const httpServer = createServer()
 const wss = new ws.WebSocketServer({
   server:httpServer
 })
@@ -87,10 +94,11 @@ httpServer.listen(port, function () {
 ## MQTT server over WebSocket using server-factory
 
 ```js
-const aedes = require('aedes')()
+const { Aedes } = require('aedes')
 const { createServer } = require('aedes-server-factory')
 const port = 8888
 
+const aedes = await Aedes.createBroker()
 const httpServer = createServer(aedes, { ws: true })
 
 httpServer.listen(port, function () {