fix: possible leak in connect and disconnect (moscajs#511)

* fix: possible leak in handleConnect negate

* docs: add missing id option

* fix: possible leak in client disconnect

* test preConect error

* test disconnect

aedes.d.ts

@@ -58,6 +58,7 @@ declare namespace aedes {
 
   interface AedesOptions {
     mq?: any
+    id?: string
     persistence?: any
     concurrency?: number
     heartbeatInterval?: number

docs/Aedes.md

@@ -41,6 +41,7 @@
   - `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`
+  - `id` `<string>` aedes broker unique identifier. __Default__: `uuidv4()`
   - `connectTimeout` `<number>` maximum waiting time in milliseconds waiting for a [`CONNECT`][CONNECT] packet. __Default__: `30000`
 - Returns `<Aedes>`
 

lib/client.js

@@ -288,6 +288,7 @@ Client.prototype.close = function (done) {
     finish)
 
   function finish () {
+    // _disconnected is set only if client is disconnected with a valid disconnect packet
     if (!that._disconnected && that.will) {
       const will = Object.assign({}, that.will)
       that.broker.authorizePublish(that, will, function (err) {

lib/handlers/connect.js

@@ -48,13 +48,10 @@ function handleConnect (client, packet, done) {
   function negate (err, successful) {
     if (!err && successful === true) {
       setImmediate(init, client, packet, done)
-      return
-    }
-    client.connecting = false
-    if (err) {
-      client.broker.emit('connectionError', client, err)
+    } else {
+      client.connecting = false
+      done(err)
     }
-    client.conn.destroy()
   }
 }
 

lib/handlers/index.js

@@ -54,9 +54,11 @@ function handle (client, packet, done) {
       client._disconnected = true
       // [MQTT-3.14.4-1] [MQTT-3.14.4-2]
       client.conn.destroy()
+      done()
       return
     default:
       client.conn.destroy()
+      done()
       return
   }
 

test/connect.js

@@ -433,6 +433,46 @@ test('second CONNECT Packet sent from a Client as a protocol violation and disco
   })
 })
 
+test('connect handler calls done when preConnect throws error', function (t) {
+  t.plan(1)
+
+  const broker = aedes({
+    preConnect: function (client, done) {
+      done(Error('error in preconnect'))
+    }
+  })
+
+  t.tearDown(broker.close.bind(broker))
+
+  const s = setup(broker)
+
+  var handleConnect = require('../lib/handlers/connect')
+
+  handleConnect(s.client, {}, function done (err) {
+    t.equal(err.message, 'error in preconnect', 'calls done with error')
+  })
+})
+
+test('handler calls done when disconnect or unknown packet cmd is received', function (t) {
+  t.plan(2)
+
+  const broker = aedes()
+
+  t.tearDown(broker.close.bind(broker))
+
+  const s = setup(broker)
+
+  var handle = require('../lib/handlers/index')
+
+  handle(s.client, { cmd: 'disconnect' }, function done () {
+    t.pass('calls done when disconnect cmd is received')
+  })
+
+  handle(s.client, { cmd: 'fsfadgragae' }, function done () {
+    t.pass('calls done when unknown cmd is received')
+  })
+})
+
 test('reject second CONNECT Packet sent while first CONNECT still in preConnect stage', function (t) {
   t.plan(2)
 

test/types/index.ts

@@ -8,6 +8,7 @@ const broker = Server({
   concurrency: 100,
   heartbeatInterval: 60000,
   connectTimeout: 30000,
+  id: 'aedes',
   preConnect: (client: Client, callback) => {
     if (client.req) {
       callback(new Error('not websocket stream'), false)