Fixed test server helper sometimes write after end.

redboltz · redboltz · commit 7c1368fee5bd · 2021-06-16T09:52:47.000+09:00
It caused Uncaught Error: write after end as follows.
It had happened very subtle timing.

```
  1) Websocket Client
       auto reconnect
         should resubscribe when reconnecting:
     Uncaught Error: write after end
      at writeAfterEnd (node_modules/duplexify/node_modules/readable-stream/lib/_stream_writable.js:288:12)
      at Connection.Writable.write (node_modules/duplexify/node_modules/readable-stream/lib/_stream_writable.js:332:20)
      at Connection.&lt;computed&gt; [as pingresp] (node_modules/mqtt-connection/connection.js:95:10)
      at Connection.&lt;anonymous&gt; (test/server_helpers_for_client_tests.js:96:20)
      at Connection.emitPacket (node_modules/mqtt-connection/connection.js:10:8)
      at addChunk (node_modules/duplexify/node_modules/readable-stream/lib/_stream_readable.js:291:12)
      at readableAddChunk (node_modules/duplexify/node_modules/readable-stream/lib/_stream_readable.js:278:11)
      at Connection.Readable.push (node_modules/duplexify/node_modules/readable-stream/lib/_stream_readable.js:245:10)
      at Connection.Duplexify._forward (node_modules/duplexify/index.js:170:26)
      at DestroyableTransform.onreadable (node_modules/duplexify/index.js:134:10)
      at emitReadable_ (node_modules/through2/node_modules/readable-stream/lib/_stream_readable.js:504:10)
      at emitReadable (node_modules/through2/node_modules/readable-stream/lib/_stream_readable.js:498:62)
      at addChunk (node_modules/through2/node_modules/readable-stream/lib/_stream_readable.js:298:29)
      at readableAddChunk (node_modules/through2/node_modules/readable-stream/lib/_stream_readable.js:278:11)
      at DestroyableTransform.Readable.push (node_modules/through2/node_modules/readable-stream/lib/_stream_readable.js:245:10)
      at DestroyableTransform.Transform.push (node_modules/through2/node_modules/readable-stream/lib/_stream_transform.js:148:32)
      at Parser.push (node_modules/mqtt-connection/lib/parseStream.js:19:12)
      at Parser._newPacket (node_modules/mqtt-packet/parser.js:672:12)
      at Parser.parse (node_modules/mqtt-packet/parser.js:43:45)
      at DestroyableTransform.process [as _transform] (node_modules/mqtt-connection/lib/parseStream.js:14:17)
      at DestroyableTransform.Transform._read (node_modules/through2/node_modules/readable-stream/lib/_stream_transform.js:184:10)
      at DestroyableTransform.Transform._write (node_modules/through2/node_modules/readable-stream/lib/_stream_transform.js:172:83)
      at doWrite (node_modules/through2/node_modules/readable-stream/lib/_stream_writable.js:428:64)
      at writeOrBuffer (node_modules/through2/node_modules/readable-stream/lib/_stream_writable.js:417:5)
      at DestroyableTransform.Writable.write (node_modules/through2/node_modules/readable-stream/lib/_stream_writable.js:334:11)
      at Socket.ondata (internal/streams/readable.js:719:22)
      at addChunk (internal/streams/readable.js:309:12)
      at readableAddChunk (internal/streams/readable.js:284:9)
      at Socket.Readable.push (internal/streams/readable.js:223:10)
      at TCP.onStreamRead (internal/stream_base_commons.js:188:23)
      at TCP.callbackTrampoline (internal/async_hooks.js:131:14)
```
diff --git a/test/server_helpers_for_client_tests.js b/test/server_helpers_for_client_tests.js
@@ -22,12 +22,14 @@ var MQTTConnection = require('mqtt-connection')
 function serverBuilder (protocol, handler) {
   var defaultHandler = function (serverClient) {
     serverClient.on('auth', function (packet) {
+      if (serverClient.writable) return false
       var rc = 'reasonCode'
       var connack = {}
       connack[rc] = 0
       serverClient.connack(connack)
     })
     serverClient.on('connect', function (packet) {
+      if (!serverClient.writable) return false
       var rc = 'returnCode'
       var connack = {}
       if (serverClient.options && serverClient.options.protocolVersion === 5) {
@@ -52,6 +54,7 @@ function serverBuilder (protocol, handler) {
     })
 
     serverClient.on('publish', function (packet) {
+      if (!serverClient.writable) return false
       setImmediate(function () {
         switch (packet.qos) {
           case 0:
@@ -67,10 +70,12 @@ function serverBuilder (protocol, handler) {
     })
 
     serverClient.on('pubrel', function (packet) {
+      if (!serverClient.writable) return false
       serverClient.pubcomp(packet)
     })
 
     serverClient.on('pubrec', function (packet) {
+      if (!serverClient.writable) return false
       serverClient.pubrel(packet)
     })
 
@@ -79,6 +84,7 @@ function serverBuilder (protocol, handler) {
     })
 
     serverClient.on('subscribe', function (packet) {
+      if (!serverClient.writable) return false
       serverClient.suback({
         messageId: packet.messageId,
         granted: packet.subscriptions.map(function (e) {
@@ -88,11 +94,13 @@ function serverBuilder (protocol, handler) {
     })
 
     serverClient.on('unsubscribe', function (packet) {
+      if (!serverClient.writable) return false
       packet.granted = packet.unsubscriptions.map(function () { return 0 })
       serverClient.unsuback(packet)
     })
 
     serverClient.on('pingreq', function () {
+      if (!serverClient.writable) return false
       serverClient.pingresp()
     })
 
