fix: Memory leak on client.js 'connected' event #348

lib/client.js

@@ -48,6 +48,15 @@ function Client (broker, conn, req) {
 
   this.connDetails = null
 
+  // queue packets received before client fires 'connect' event. Prevents memory leaks on 'connect' event
+  this.parser._queue = []
+
+  this.on('connected', () => {
+    while (this.parser._queue.length > 0) {
+      handle(this, this.parser._queue.shift(), this._nextBatch)
+    }
+  })
+
   this.parser.on('packet', enqueue)
 
   function nextBatch (err) {
@@ -330,9 +339,7 @@ function enqueue (packet) {
   if (client.connackSent || client._parsingBatch === 1) {
     handle(client, packet, client._nextBatch)
   } else {
-    client.on('connected', () => {
-      handle(client, packet, client._nextBatch)
-    })
+    this._queue.push(packet)
   }
 }
 

test/connect.js

@@ -332,6 +332,46 @@ test('reject clients with wrong protocol name', function (t) {
   broker.on('closed', t.end.bind(t))
 })
 
+test('After first CONNECT Packet, others are queued until \'connect\' event', function (t) {
+  t.plan(2)
+
+  var broker = aedes()
+
+  var publishP = {
+    cmd: 'publish',
+    topic: 'hello',
+    payload: Buffer.from('world'),
+    qos: 0,
+    retain: false
+  }
+
+  var connectP = {
+    cmd: 'connect',
+    protocolId: 'MQTT',
+    protocolVersion: 4,
+    clean: true,
+    clientId: 'abcde',
+    keepalive: 0
+  }
+
+  var s = setup(broker, false)
+  s.inStream.write(connectP)
+
+  for (let i = 0; i < 10; i++) {
+    s.inStream.write(publishP)
+  }
+
+  broker.on('client', function (client) {
+    t.equal(client.parser._queue.length, 10, 'Packets have been queued')
+
+    client.on('connected', () => {
+      t.equal(client.parser._queue.length, 0, 'Queue is empty')
+      s.conn.destroy()
+      broker.close(t.end)
+    })
+  })
+})
+
 ;[[0, null, false], [1, null, true], [1, new Error('connection banned'), false], [1, new Error('connection banned'), true]].forEach(function (ele) {
   var plan = ele[0]
   var err = ele[1]