fix: memory leak in connected clients counter (#979)

Co-authored-by: Uladzimir Danko <[email protected]>

aedes.js

@@ -172,7 +172,7 @@ function Aedes (opts) {
     if (that.clients[clientId] && serverId !== that.id) {
       if (that.clients[clientId].closed) {
         // remove the client from the list if it is already closed
-        delete that.clients[clientId]
+        that.deleteClient(clientId)
         done()
       } else {
         that.clients[clientId].close(done)
@@ -316,15 +316,19 @@ Aedes.prototype._finishRegisterClient = function (client) {
 }
 
 Aedes.prototype.unregisterClient = function (client) {
-  this.connectedClients--
-  delete this.clients[client.id]
+  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)
 }

lib/handlers/subscribe.js

@@ -155,6 +155,13 @@ function addSubs (sub, done) {
     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)

test/events.js

@@ -223,18 +223,20 @@ test('Test backpressure aedes published function', function (t) {
 })
 
 test('clear closed clients when the same clientId is managed by another broker', function (t) {
-  t.plan(1)
+  t.plan(2)
 
   const clientId = 'closed-client'
-  const broker = aedes()
+  const aedesBroker = aedes()
 
   // simulate a closed client on the broker
-  broker.clients[clientId] = { closed: true }
+  aedesBroker.clients[clientId] = { closed: true, broker: aedesBroker }
+  aedesBroker.connectedClients = 1
 
   // simulate the creation of the same client on another broker of the cluster
-  broker.publish({ topic: '$SYS/anotherbroker/new/clients', payload: clientId }, () => {
-    t.equal(broker.clients[clientId], undefined) // check that the closed client was removed
+  aedesBroker.publish({ topic: '$SYS/anotherbroker/new/clients', payload: clientId }, () => {
+    t.equal(aedesBroker.clients[clientId], undefined) // check that the closed client was removed
+    t.equal(aedesBroker.connectedClients, 0)
   })
 
-  t.teardown(broker.close.bind(broker))
+  t.teardown(aedesBroker.close.bind(aedesBroker))
 })