fix(sendPacket): drain leak (#1401)

Author: BertKleewein

lib/client.js

@@ -175,7 +175,7 @@ function sendPacket (client, packet, cb) {
   debug('sendPacket :: writing to stream')
   const result = mqttPacket.writeToStream(packet, client.stream, client.options)
   debug('sendPacket :: writeToStream result %s', result)
-  if (!result && cb) {
+  if (!result && cb && cb !== nop) {
     debug('sendPacket :: handle events on `drain` once through callback.')
     client.stream.once('drain', cb)
   } else if (cb) {
@@ -190,6 +190,8 @@ function flush (queue) {
     Object.keys(queue).forEach(function (messageId) {
       if (typeof queue[messageId].cb === 'function') {
         queue[messageId].cb(new Error('Connection closed'))
+        // This is suspicious.  Why do we only delete this if we have a callbck?
+        // If this is by-design, then adding no as callback would cause this to get deleted unintentionally.
         delete queue[messageId]
       }
     })
@@ -569,7 +571,7 @@ MqttClient.prototype._handlePacket = function (packet, done) {
 
 MqttClient.prototype._checkDisconnecting = function (callback) {
   if (this.disconnecting) {
-    if (callback) {
+    if (callback && callback !== nop) {
       callback(new Error('client disconnecting'))
     } else {
       this.emit('error', new Error('client disconnecting'))
@@ -1581,7 +1583,7 @@ MqttClient.prototype._handleAck = function (packet) {
   const that = this
   let err
 
-  if (!cb) {
+  if (!cb || cb === nop) {
     debug('_handleAck :: Server sent an ack in error. Ignoring.')
     // Server sent an ack in error, ignore it.
     return