chore: reduce production dependencies

aedes.js

@@ -1,14 +1,11 @@
 import EventEmitter from 'node:events'
-import parallel from 'fastparallel'
-import series from 'fastseries'
-import { v4 as uuidv4 } from 'uuid'
-import reusify from 'reusify'
+import { randomUUID } from 'node:crypto'
 import { pipeline } from 'stream'
 import Packet from 'aedes-packet'
 import memory from 'aedes-persistence'
 import mqemitter from 'mqemitter'
 import Client from './lib/client.js'
-import { $SYS_PREFIX, bulk } from './lib/utils.js'
+import { $SYS_PREFIX, bulk, ObjectPool } from './lib/utils.js'
 import pkg from './package.json' with { type: 'json' }
 
 const defaultOptions = {
@@ -37,7 +34,7 @@ export class Aedes extends EventEmitter {
 
     opts = Object.assign({}, defaultOptions, opts)
     this.opts = opts
-    this.id = opts.id || uuidv4()
+    this.id = opts.id || randomUUID()
     // +1 when construct a new aedes-packet
     // internal track for last brokerCounter
     this.counter = 0
@@ -56,9 +53,8 @@ export class Aedes extends EventEmitter {
       return new Client(that, conn, req)
     }
 
-    this._parallel = parallel()
-    this._series = series()
-    this._enqueuers = reusify(DoEnqueues)
+    this._series = runSeries
+    this._enqueuers = new ObjectPool(DoEnqueues)
 
     this.preConnect = opts.preConnect
     this.authenticate = opts.authenticate
@@ -131,8 +127,18 @@ export class Aedes extends EventEmitter {
       )
     }, opts.heartbeatInterval * 4)
 
-    function receiveWills (chunks, done) {
-      that._parallel(that, checkAndPublish, chunks, done)
+    async function receiveWills (chunks, done) {
+      try {
+        await Promise.all(chunks.map(chunk => new Promise((resolve, reject) => {
+          checkAndPublish(chunk, (err) => {
+            if (err) reject(err)
+            else resolve()
+          })
+        })))
+        done()
+      } catch (err) {
+        done(err)
+      }
     }
 
     function checkAndPublish (will, done) {
@@ -142,7 +148,7 @@ export class Aedes extends EventEmitter {
 
       // 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) {
+      that.authorizePublish(that.clients[will.clientId] || null, will, function (err) {
         if (err) { return doneWill() }
         that.publish(will, doneWill)
 
@@ -265,11 +271,14 @@ export class Aedes extends EventEmitter {
     this.closed = true
     clearInterval(this._heartbeatInterval)
     clearInterval(this._clearWillInterval)
-    this._parallel(this, closeClient, Object.keys(this.clients), doneClose)
-    function doneClose () {
+    const promises = []
+    for (const clientId of Object.keys(this.clients)) {
+      promises.push(closeClient(this.clients[clientId]))
+    }
+    Promise.all(promises).then(() => {
       that.emit('closed')
       that.mq.close(cb)
-    }
+    })
   }
 }
 
@@ -361,8 +370,21 @@ const publishFuncsQoS = [
   callPublished
 ]
 
-function closeClient (client, cb) {
-  this.clients[client].close(cb)
+// runSeries runs functions in fastseries style
+function runSeries (state, actions, packet, done) {
+  done = (done || noop).bind(state)
+  let i = 0
+  function next (err) {
+    if (err || i === actions.length) return done(err)
+    actions[i++].call(state, packet, next)
+  }
+  next()
+}
+
+async function closeClient (clientInstance) {
+  return new Promise((resolve) => {
+    clientInstance.close(resolve)
+  })
 }
 
 function defaultPreConnect (client, packet, callback) {
@@ -400,11 +422,11 @@ class PublishState {
   }
 }
 
-function noop () {}
+function noop () { }
 
 function warnMigrate () {
   throw new Error(
-` Aedes default export has been removed.
+    ` Aedes default export has been removed.
  Use 'const aedes = await Aedes.createBroker()' instead.
  See: https://github.com/moscajs/aedes/docs/MIGRATION.MD
  `)

docs/Client.md

@@ -62,7 +62,7 @@ a read-only flag indicates if client is closed or not.
 
 ## client.id
 
-- `<string>` __Default__: `aedes_${hyperid()}`
+- `<string>` __Default__: `aedes_${randomUUID()}`
 
 Client unique identifier, specified by CONNECT packet.
 

lib/client.js

@@ -1,14 +1,13 @@
 import mqtt from 'mqtt-packet'
 import EventEmitter from 'node:events'
 import util from 'util'
-import eos from 'end-of-stream'
 import Packet from 'aedes-packet'
 import write from './write.js'
 import QoSPacket from './qos-packet.js'
 import handleSubscribe from './handlers/subscribe.js'
 import handleUnsubscribe from './handlers/unsubscribe.js'
 import handle from './handlers/index.js'
-import { pipeline } from 'stream'
+import { pipeline, finished } from 'stream'
 import { through } from './utils.js'
 
 class Client {
@@ -84,7 +83,7 @@ class Client {
     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))
+    this._eos = finished(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

lib/handlers/connect.js

@@ -1,12 +1,9 @@
-import retimer from 'retimer'
+import { randomUUID } from 'node:crypto'
 import { pipeline } from 'stream'
 import write from '../write.js'
 import QoSPacket from '../qos-packet.js'
-import { through } from '../utils.js'
+import { through, retimer } from '../utils.js'
 import handleSubscribe from './subscribe.js'
-import hyperid from 'hyperid'
-
-const uniqueId = hyperid()
 
 class Connack {
   constructor (arg) {
@@ -84,7 +81,7 @@ function init (client, packet, done) {
     return
   }
 
-  client.id = clientId || 'aedes_' + uniqueId()
+  client.id = clientId || 'aedes_' + randomUUID()
   client.clean = packet.clean
   client.version = packet.protocolVersion
   client._will = packet.will

lib/handlers/subscribe.js

@@ -1,14 +1,32 @@
-import fastfall from 'fastfall'
 import Packet from 'aedes-packet'
 import { through, validateTopic, $SYS_PREFIX } from '../utils.js'
 import write from '../write.js'
 
-const subscribeTopicActions = fastfall([
+function runFall (fns) {
+  // run functions in fastfall style, only need the single argument function
+  return function (arg, cb) {
+    let i = 0
+    const ctx = this
+    function next (err, nextarg) {
+      if (err || i === fns.length) {
+        if (typeof cb === 'function') {
+          cb.call(ctx, err, nextarg)
+        }
+        return
+      }
+      const fn = fns[i++]
+      fn.call(ctx, nextarg, next)
+    }
+    next(null, arg)
+  }
+}
+
+const subscribeTopicActions = runFall([
   authorize,
   storeSubscriptions,
   addSubs
 ])
-const restoreTopicActions = fastfall([
+const restoreTopicActions = runFall([
   authorize,
   addSubs
 ])
@@ -78,14 +96,24 @@ function _dedupe (subs) {
   return ret
 }
 
-function handleSubscribe (client, packet, restore, done) {
+async 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
-  )
+  const state = new SubscribeState(client, packet, restore, done)
+  try {
+    await Promise.all(packet.subscriptions.map(sub => new Promise((resolve, reject) => {
+      doSubscribe.call(state, sub, (err, result) => {
+        if (err) reject(err)
+        else resolve(result)
+      })
+    })))
+    if (restore) {
+      done()
+    } else {
+      completeSubscribe.call(state)
+    }
+  } catch (err) {
+    done(err)
+  }
 }
 
 function doSubscribe (sub, done) {

lib/handlers/unsubscribe.js

@@ -40,13 +40,19 @@ function handleUnsubscribe (client, packet, done) {
   }
 }
 
-function actualUnsubscribe (client, packet, done) {
-  const broker = client.broker
-  broker._parallel(
-    new UnsubscribeState(client, packet, done),
-    doUnsubscribe,
-    packet.unsubscriptions,
-    completeUnsubscribe)
+async function actualUnsubscribe (client, packet, done) {
+  const state = new UnsubscribeState(client, packet, done)
+  try {
+    await Promise.all(packet.unsubscriptions.map(sub => new Promise((resolve, reject) => {
+      doUnsubscribe.call(state, sub, (err, result) => {
+        if (err) reject(err)
+        else resolve(result)
+      })
+    })))
+    completeUnsubscribe.call(state)
+  } catch (err) {
+    completeUnsubscribe.call(state, err)
+  }
 }
 
 function doUnsubscribe (sub, done) {

lib/utils.js

@@ -37,6 +37,40 @@ export function through (transform) {
   })
 }
 
+// DIY object pooling as a replacement for reusify
+export class ObjectPool {
+  constructor (ClassCtor) {
+    this.ClassCtor = ClassCtor
+    this.pool = []
+  }
+
+  get () {
+    return this.pool.pop() || new this.ClassCtor()
+  }
+
+  release (obj) {
+    // Optionally reset the object here if needed
+    this.pool.push(obj)
+  }
+}
+
+// Nodejs can clear timers in constant time these days
+// https://asafdav2.github.io/2017/node-js-timers/
+// Running retimers bench.js on Node 24.5 shows retimer to be 1 second faster on 1M invocations.
+// hence we can simplify retimer so we don't have to rely on a dependency
+export function retimer (fn, ms) {
+  let timeout = setTimeout(fn, ms)
+  return {
+    reschedule (newMs) {
+      clearTimeout(timeout)
+      timeout = setTimeout(fn, newMs)
+    },
+    clear () {
+      clearTimeout(timeout)
+    }
+  }
+}
+
 export function bulk (fn) {
   return new Writable({
     objectMode: true,

package.json

@@ -120,16 +120,8 @@
 	"dependencies": {
 		"aedes-packet": "^3.0.0",
 		"aedes-persistence": "^10.2.2",
-		"end-of-stream": "^1.4.5",
-		"fastfall": "^1.5.1",
-		"fastparallel": "^2.4.1",
-		"fastseries": "^2.0.0",
-		"hyperid": "^3.3.0",
 		"mqemitter": "^7.1.0",
-		"mqtt-packet": "^9.0.2",
-		"retimer": "^4.0.0",
-		"reusify": "^1.1.0",
-		"uuid": "^11.1.0"
+		"mqtt-packet": "^9.0.2"
 	},
 	"peerDependencies": {
 		"aedes-persistence-level": "^9.1.2",