Repository: moscajs/aedes
Pull request #401 (2020-02-05 09:09:59)

Destroy socket when DISCONNECT sent
end() will send final data and allow some i/o activity to finish before destroying socket.
MQTT-3.14.4-2 states "MUST NOT send any more Control Packets on that Network Connection". In some case aedes still deliverQoS data to subscribers via  mqemitter when subscriber is conn.end(), broker then raises "connection closed" error.

Once we destroy the socket, mqemitter will not trigger deliver data to closed subscriber.

Fixed issue https://github.com/moscajs/aedes/issues/396


Patch:
diff --git a/lib/handlers/index.js b/lib/handlers/index.js
index 50bda5f..57c4907 100644
--- a/lib/handlers/index.js
+++ b/lib/handlers/index.js
@@ -51,8 +51,8 @@ function handle (client, packet, done) {
     case 'disconnect':
       // [MQTT-3.14.4-3]
       client._disconnected = true
-      // [MQTT-3.14.4-1]
-      client.conn.end()
+      // [MQTT-3.14.4-1] [MQTT-3.14.4-2]
+      client.conn.destroy()
       return
     default:
       client.conn.destroy()


Test patch:
diff --git a/test/basic.js b/test/basic.js
index c4e9f84..a8242d8 100644
--- a/test/basic.js
+++ b/test/basic.js
@@ -112,6 +112,25 @@ test('does not die badly on connection error', function (t) {
   })
 })
 
+// Guarded in mqtt-packet
+test('subscribe should have messageId', function (t) {
+  t.plan(1)
+
+  const s = connect(setup())
+  t.tearDown(s.broker.close.bind(s.broker))
+
+  s.inStream.write({
+    cmd: 'subscribe',
+    subscriptions: [{
+      topic: 'hello',
+      qos: 0
+    }]
+  })
+  s.broker.on('connectionError', function (client, err) {
+    t.ok(err.message, 'Invalid messageId')
+  })
+})
+
 test('unsubscribe', function (t) {
   t.plan(5)
 
@@ -234,7 +253,7 @@ test('disconnect', function (t) {
   const s = noError(connect(setup()), t)
   t.tearDown(s.broker.close.bind(s.broker))
 
-  s.outStream.on('finish', function () {
+  s.broker.on('clientDisconnect', function () {
     t.pass('closed stream')
   })
 
diff --git a/test/meta.js b/test/meta.js
index 27fd51e..cf42da1 100644
--- a/test/meta.js
+++ b/test/meta.js
@@ -1,7 +1,7 @@
 'use strict'
 
 const { test } = require('tap')
-const { setup, connect, subscribe } = require('./helper')
+const { setup, connect, subscribe, noError } = require('./helper')
 const aedes = require('../')
 
 test('count connected clients', function (t) {
@@ -187,7 +187,7 @@ test('dont emit unsubscribe event on client close', function (t) {
   const broker = aedes()
   t.tearDown(broker.close.bind(broker))
 
-  const s = connect(setup(broker), { clientId: 'abcde' })
+  const s = noError(connect(setup(broker), { clientId: 'abcde' }), t)
 
   broker.on('unsubscribe', function (unsubscriptions, client) {
     t.error('unsubscribe should not be emitted')
@@ -213,7 +213,7 @@ test('emit clientDisconnect event', function (t) {
     t.equal(client.id, 'abcde', 'client matches')
   })
 
-  const s = connect(setup(broker), { clientId: 'abcde' })
+  const s = noError(connect(setup(broker), { clientId: 'abcde' }), t)
 
   s.inStream.end({
     cmd: 'disconnect'
diff --git a/test/retain.js b/test/retain.js
index 4cf7e36..6dbbdaa 100644
--- a/test/retain.js
+++ b/test/retain.js
@@ -492,12 +492,12 @@ test('deliver QoS 0 retained message with QoS 1 subscription', function (t) {
 })
 
 test('disconnect and retain messages with QoS 1 [clean=false]', function (t) {
-  t.plan(8)
+  t.plan(7)
 
   const broker = aedes()
   t.tearDown(broker.close.bind(broker))
 
-  var subscriber = connect(setup(broker), { clean: false, clientId: 'abcde' })
+  var subscriber = noError(connect(setup(broker), { clean: false, clientId: 'abcde' }), t)
   const expected = {
     cmd: 'publish',
     topic: 'hello',
@@ -517,7 +517,7 @@ test('disconnect and retain messages with QoS 1 [clean=false]', function (t) {
       console.log('original', packet)
     })
 
-    const publisher = connect(setup(broker))
+    const publisher = noError(connect(setup(broker)), t)
 
     publisher.inStream.write({
       cmd: 'publish',
@@ -531,10 +531,6 @@ test('disconnect and retain messages with QoS 1 [clean=false]', function (t) {
     publisher.outStream.once('data', function (packet) {
       t.equal(packet.cmd, 'puback')
 
-      broker.on('clientError', function (client, err) {
-        t.equal(err.message, 'connection closed')
-      })
-
       subscriber = connect(setup(broker), { clean: false, clientId: 'abcde' }, function (connect) {
         t.equal(connect.sessionPresent, true, 'session present is set to true')
       })
@@ -556,12 +552,12 @@ test('disconnect and retain messages with QoS 1 [clean=false]', function (t) {
 })
 
 test('disconnect and two retain messages with QoS 1 [clean=false]', function (t) {
-  t.plan(17)
+  t.plan(15)
 
   const broker = aedes()
   t.tearDown(broker.close.bind(broker))
 
-  var subscriber = connect(setup(broker), { clean: false, clientId: 'abcde' })
+  var subscriber = noError(connect(setup(broker), { clean: false, clientId: 'abcde' }), t)
   const expected = {
     cmd: 'publish',
     topic: 'hello',
@@ -580,7 +576,7 @@ test('disconnect and two retain messages with QoS 1 [clean=false]', function (t)
       console.log('original', packet)
     })
 
-    const publisher = connect(setup(broker))
+    const publisher = noError(connect(setup(broker)), t)
 
     publisher.inStream.write({
       cmd: 'publish',
@@ -606,10 +602,6 @@ test('disconnect and two retain messages with QoS 1 [clean=false]', function (t)
       publisher.outStream.once('data', function (packet) {
         t.equal(packet.cmd, 'puback')
 
-        broker.on('clientError', function (client, err) {
-          t.equal(err.message, 'connection closed')
-        })
-
         subscriber = connect(setup(broker), { clean: false, clientId: 'abcde' }, function (connect) {
           t.equal(connect.sessionPresent, true, 'session present is set to true')
         })
diff --git a/test/will.js b/test/will.js
index 5375cef..cc7385c 100644
--- a/test/will.js
+++ b/test/will.js
@@ -2,7 +2,7 @@
 
 const { test } = require('tap')
 const memory = require('aedes-persistence')
-const { setup, connect } = require('./helper')
+const { setup, connect, noError } = require('./helper')
 const aedes = require('../')
 
 function willConnect (s, opts, connected) {
@@ -332,11 +332,11 @@ test('does not deliver will when client sends a DISCONNECT', function (t) {
   const broker = aedes()
   t.tearDown(broker.close.bind(broker))
 
-  const s = willConnect(setup(broker), {}, function () {
+  const s = noError(willConnect(setup(broker), {}, function () {
     s.inStream.end({
       cmd: 'disconnect'
     })
-  })
+  }), t)
 
   s.broker.mq.on('mywill', function (packet, cb) {
     t.fail(packet)
@@ -351,12 +351,12 @@ test('does not store multiple will with same clientid', function (t) {
 
   const broker = aedes()
 
-  var s = willConnect(setup(broker), opts, function () {
+  var s = noError(willConnect(setup(broker), opts, function () {
     // gracefully close client so no will is sent
     s.inStream.end({
       cmd: 'disconnect'
     })
-  })
+  }), t)
 
   broker.on('clientDisconnect', function (client) {
     // reconnect same client with will