Repository: websockets/ws (js)

Issue:
Last frame's payload not delivered and code 1006 returned
The title may not be 100% accurate, I apologize for that but it's not too clear to me what's happening...
### Versions

I can say for sure that this happens with version 1.1.1 but not with 0.4.32.

**Edit:** It doesn't work with version 0.8.1 either. The only difference with respect to 1.1.1 is that the close event code is still 1000.
### Client code

The client code is not really important, you can use the basic:

``` js
var WebSocket = require('ws');
var ws = new WebSocket(process.argv[2]);

ws.on('message', function (data, flags) {
    console.log('>', data);
});

ws.on('close', function (code, message) {
    console.error('#', code, message);
});
```

or even [`wscat`](https://github.com/websockets/wscat), what really matter is the `ws` version.
### How to reproduce

This behavior has been observed with the Google Chrome's Remote Debugging Protocol WebSocket; (cyrus-and/chrome-remote-interface#39).

Here are the step to reproduce this issue:
1. start a new Chrome session with `--remote-debugging-port=9222`;
2. fetch a list of the currently open tabs:
   
   ```
   $ curl http://localhost:9222/json
   [ {
      "description": "",
      "devtoolsFrontendUrl": "/devtools/inspector.html?ws=localhost:9222/devtools/page/952808E7-26BC-470B-B1DF-B61519786481",
      "faviconUrl": "https://www.google.it/favicon.ico",
      "id": "952808E7-26BC-470B-B1DF-B61519786481",
      "title": "New Tab",
      "type": "page",
      "url": "chrome://newtab/",
      "webSocketDebuggerUrl": "ws://localhost:9222/devtools/page/952808E7-26BC-470B-B1DF-B61519786481"
   },
   [...]
   ```
   
   Locate the new empty tab (`chrome://newtab/`) and not the WebSocket URL (in this case `ws://localhost:9222/devtools/page/952808E7-26BC-470B-B1DF-B61519786481`).

3. connect to that URL:
   
   ```
   $ wscat -c ws://localhost:9222/devtools/page/952808E7-26BC-470B-B1DF-B61519786481
   connected (press CTRL+C to quit)
   ```
4. from Chrome open the DevTools for that tab (this will close the WebSocket).
#### What happens with 1.1.1

```
connected (press CTRL+C to quit)
disconnected
```

Using a custom client we obtain the close event code 1006.

Using Wireshark we notice:
- `Sec-WebSocket-Extensions` is set to `permessage-deflate; client_max_window_bits=15` in the handshake reply;
- `Fin: True` and `Reserved: 0x04` and the payload is not human-readable in the last (and only) WebSocket frame.
#### What happens with 0.4.32

```
connected (press CTRL+C to quit)
< { "method": "Inspector.detached", "params": { "reason": "replaced_with_devtools"} }
disconnected
```

Using a custom client we obtain the close event code 1000.

Using Wireshark we notice:
- `Sec-WebSocket-Extensions` is not set in the handshake reply;
- `Fin: True` and `Reserved: 0x00` and the payload is the above JSON text in the last (and only) WebSocket frame.


Pull request:
Consume all data before emitting the `close` event
This makes `Receiver.prototype.cleanup()` accept an optional callback which is called when all buffered data is consumed or an error occurs.

Using this callback we can handle cases where the parser is decompressing data and the other peer closes the underlying TCP connection. Previously when this happened the buffered data was simply discarded. Now all data should be handled correctly.

Fixes #799 and #923.

This is based and supersedes #840.

@websockets/admin this really needs to be reviewed carefully so when you have time please take a look.

Patch:
diff --git a/bench/parser.benchmark.js b/bench/parser.benchmark.js
index 63883f8e..2924b544 100644
--- a/bench/parser.benchmark.js
+++ b/bench/parser.benchmark.js
@@ -43,8 +43,10 @@ const binaryDataPacket2 = createBinaryPacket(65535);
 const binaryDataPacket3 = createBinaryPacket(200 * 1024);
 const binaryDataPacket4 = createBinaryPacket(1024 * 1024);
 
-const receiver = new Receiver();
 const suite = new benchmark.Suite();
+const receiver = new Receiver();
+
+receiver.onmessage = receiver.onclose = receiver.onping = () => {};
 
 suite.add('ping message', () => receiver.add(pingPacket1));
 suite.add('ping with no data', () => receiver.add(pingPacket2));
diff --git a/lib/Receiver.js b/lib/Receiver.js
index 07cbb9b6..b79b1937 100644
--- a/lib/Receiver.js
+++ b/lib/Receiver.js
@@ -20,8 +20,6 @@ const GET_MASK = 3;
 const GET_DATA = 4;
 const HANDLE_DATA = 5;
 
-const noop = () => {};
-
 /**
  * HyBi Receiver implementation.
  */
@@ -51,13 +49,15 @@ class Receiver {
     this.messageLength = 0;
     this.fragments = [];
 
+    this.cleanupCallback = null;
+    this.hadError = false;
     this.dead = false;
 
-    this.onmessage = noop;
-    this.onclose = noop;
-    this.onerror = noop;
-    this.onping = noop;
-    this.onpong = noop;
+    this.onmessage = null;
+    this.onclose = null;
+    this.onerror = null;
+    this.onping = null;
+    this.onpong = null;
 
     this.state = START;
   }
@@ -69,19 +69,17 @@ class Receiver {
    * @private
    */
   readBuffer (bytes) {
-    var bufoff = 0;
+    var offset = 0;
     var dst;
     var l;
 
-    if (bytes === this.buffers[0].length) {
-      this.bufferedBytes -= bytes;
-      return this.buffers.shift();
-    }
+    this.bufferedBytes -= bytes;
+
+    if (bytes === this.buffers[0].length) return this.buffers.shift();
 
     if (bytes < this.buffers[0].length) {
       dst = this.buffers[0].slice(0, bytes);
       this.buffers[0] = this.buffers[0].slice(bytes);
-      this.bufferedBytes -= bytes;
       return dst;
     }
 
@@ -90,15 +88,13 @@ class Receiver {
     while (bytes > 0) {
       l = this.buffers[0].length;
 
-      if (bytes > l) {
-        this.buffers[0].copy(dst, bufoff);
-        bufoff += l;
+      if (bytes >= l) {
+        this.buffers[0].copy(dst, offset);
+        offset += l;
         this.buffers.shift();
-        this.bufferedBytes -= l;
       } else {
-        this.buffers[0].copy(dst, bufoff, 0, bytes);
+        this.buffers[0].copy(dst, offset, 0, bytes);
         this.buffers[0] = this.buffers[0].slice(bytes);
-        this.bufferedBytes -= bytes;
       }
 
       bytes -= l;
@@ -107,6 +103,21 @@ class Receiver {
     return dst;
   }
 
+  /**
+   * Checks if the number of buffered bytes is bigger or equal than `n` and
+   * calls `cleanup` if necessary.
+   *
+   * @param {Number} n The number of bytes to check against
+   * @return {Boolean} `true` if `bufferedBytes >= n`, else `false`
+   * @private
+   */
+  hasBufferedBytes (n) {
+    if (this.bufferedBytes >= n) return true;
+
+    if (this.dead) this.cleanup(this.cleanupCallback);
+    return false;
+  }
+
   /**
    * Adds new data to the parser.
    *
@@ -142,7 +153,7 @@ class Receiver {
    * @private
    */
   start () {
-    if (this.bufferedBytes < 2) return;
+    if (!this.hasBufferedBytes(2)) return;
 
     const buf = this.readBuffer(2);
 
@@ -222,7 +233,7 @@ class Receiver {
    * @private
    */
   getPayloadLength16 () {
-    if (this.bufferedBytes < 2) return;
+    if (!this.hasBufferedBytes(2)) return;
 
     this.payloadLength = this.readBuffer(2).readUInt16BE(0, true);
     this.haveLength();
@@ -234,7 +245,7 @@ class Receiver {
    * @private
    */
   getPayloadLength64 () {
-    if (this.bufferedBytes < 8) return;
+    if (!this.hasBufferedBytes(8)) return;
 
     const buf = this.readBuffer(8);
     const num = buf.readUInt32BE(0, true);
@@ -277,7 +288,7 @@ class Receiver {
    * @private
    */
   getMask () {
-    if (this.bufferedBytes < 4) return;
+    if (!this.hasBufferedBytes(4)) return;
 
     this.mask = this.readBuffer(4);
     this.state = GET_DATA;
@@ -293,7 +304,7 @@ class Receiver {
     var data = EMPTY_BUFFER;
 
     if (this.payloadLength) {
-      if (this.bufferedBytes < this.payloadLength) return;
+      if (!this.hasBufferedBytes(this.payloadLength)) return;
 
       data = this.readBuffer(this.payloadLength);
       if (this.masked) bufferUtil.unmask(data, this.mask);
@@ -319,8 +330,6 @@ class Receiver {
     const extension = this.extensions[PerMessageDeflate.extensionName];
 
     extension.decompress(data, this.fin, (err, buf) => {
-      if (this.dead) return;
-
       if (err) {
         this.error(err, err.closeCode === 1009 ? 1009 : 1007);
         return;
@@ -374,7 +383,7 @@ class Receiver {
     if (this.opcode === 0x08) {
       if (data.length === 0) {
         this.onclose(1000, '', { masked: this.masked });
-        this.cleanup();
+        this.cleanup(this.cleanupCallback);
       } else if (data.length === 1) {
         this.error(new Error('invalid payload length'), 1002);
       } else {
@@ -393,7 +402,7 @@ class Receiver {
         }
 
         this.onclose(code, buf.toString(), { masked: this.masked });
-        this.cleanup();
+        this.cleanup(this.cleanupCallback);
       }
 
       return;
@@ -417,7 +426,8 @@ class Receiver {
    */
   error (err, code) {
     this.onerror(err, code);
-    this.cleanup();
+    this.hadError = true;
+    this.cleanup(this.cleanupCallback);
   }
 
   /**
@@ -466,21 +476,29 @@ class Receiver {
   /**
    * Releases resources used by the receiver.
    *
+   * @param {Function} cb Callback
    * @public
    */
-  cleanup () {
+  cleanup (cb) {
     this.dead = true;
 
-    this.extensions = null;
-    this.fragments = null;
-    this.buffers = null;
-    this.mask = null;
-
-    this.onmessage = null;
-    this.onclose = null;
-    this.onerror = null;
-    this.onping = null;
-    this.onpong = null;
+    if (!this.hadError && this.state === HANDLE_DATA) {
+      this.cleanupCallback = cb;
+    } else {
+      this.extensions = null;
+      this.fragments = null;
+      this.buffers = null;
+      this.mask = null;
+
+      this.cleanupCallback = null;
+      this.onmessage = null;
+      this.onclose = null;
+      this.onerror = null;
+      this.onping = null;
+      this.onpong = null;
+
+      if (cb) cb();
+    }
   }
 }
 
diff --git a/lib/WebSocket.js b/lib/WebSocket.js
index 98b5eeec..b2cc9196 100644
--- a/lib/WebSocket.js
+++ b/lib/WebSocket.js
@@ -22,6 +22,7 @@ const Sender = require('./Sender');
 const GUID = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11';
 const closeTimeout = 30 * 1000; // Allow 30 seconds to terminate the connection cleanly.
 const protocolVersion = 13;
+const noop = () => {};
 
 /**
  * Class representing a WebSocket.
@@ -54,6 +55,7 @@ class WebSocket extends EventEmitter {
 
     this._finalize = this.finalize.bind(this);
     this._binaryType = 'nodebuffer';
+    this._finalizeCalled = false;
     this._closeMessage = null;
     this._closeTimer = null;
     this._closeCode = null;
@@ -164,15 +166,16 @@ class WebSocket extends EventEmitter {
   }
 
   /**
-   * Clean up and release internal resources and emit the `close` event.
+   * Clean up and release internal resources.
    *
    * @param {(Boolean|Error)} Indicates whether or not an error occurred
    * @private
    */
   finalize (error) {
-    if (this.readyState === WebSocket.CLOSED) return;
+    if (this._finalizeCalled) return;
 
-    this.readyState = WebSocket.CLOSED;
+    this.readyState = WebSocket.CLOSING;
+    this._finalizeCalled = true;
 
     clearTimeout(this._closeTimer);
     this._closeTimer = null;
@@ -183,7 +186,6 @@ class WebSocket extends EventEmitter {
     // 1006.
     //
     if (error) this._closeCode = 1006;
-    this.emit('close', this._closeCode || 1006, this._closeMessage || '');
 
     if (this._socket) {
       this._ultron.destroy();
@@ -203,9 +205,21 @@ class WebSocket extends EventEmitter {
     }
 
     if (this._receiver) {
-      this._receiver.cleanup();
+      this._receiver.cleanup(() => this.emitClose());
       this._receiver = null;
+    } else {
+      this.emitClose();
     }
+  }
+
+  /**
+   * Emit the `close` event.
+   *
+   * @private
+   */
+  emitClose () {
+    this.readyState = WebSocket.CLOSED;
+    this.emit('close', this._closeCode || 1006, this._closeMessage || '');
 
     if (this.extensions[PerMessageDeflate.extensionName]) {
       this.extensions[PerMessageDeflate.extensionName].cleanup();
@@ -214,7 +228,7 @@ class WebSocket extends EventEmitter {
     this.extensions = null;
 
     this.removeAllListeners();
-    this.on('error', function onerror () {}); // catch all errors after this
+    this.on('error', noop); // Catch all errors after this.
   }
 
   /**
@@ -255,9 +269,7 @@ class WebSocket extends EventEmitter {
     }
 
     if (this.readyState === WebSocket.CLOSING) {
-      if (this._closeCode && this._isServer) {
-        this.terminate();
-      }
+      if (this._closeCode) this.terminate();
       return;
     }
 
@@ -265,7 +277,7 @@ class WebSocket extends EventEmitter {
     this._sender.close(code, data, !this._isServer, (err) => {
       if (err) this.emit('error', err);
 
-      if (this._closeCode && this._isServer) {
+      if (this._closeCode) {
         this.terminate();
       } else {
         //


Test patch:
diff --git a/test/Receiver.test.js b/test/Receiver.test.js
index 18dc711a..7129228b 100644
--- a/test/Receiver.test.js
+++ b/test/Receiver.test.js
@@ -309,39 +309,50 @@ describe('Receiver', function () {
 
   it('resets `totalPayloadLength` only on final frame (unfragmented)', function () {
     const p = new Receiver({}, 10);
+    let message;
+
+    p.onmessage = function (msg) {
+      message = msg;
+    };
 
     assert.strictEqual(p.totalPayloadLength, 0);
     p.add(Buffer.from('810548656c6c6f', 'hex'));
     assert.strictEqual(p.totalPayloadLength, 0);
+    assert.strictEqual(message, 'Hello');
   });
 
   it('resets `totalPayloadLength` only on final frame (fragmented)', function () {
     const p = new Receiver({}, 10);
+    let message;
 
-    const frame1 = '01024865';
-    const frame2 = '80036c6c6f';
+    p.onmessage = function (msg) {
+      message = msg;
+    };
 
     assert.strictEqual(p.totalPayloadLength, 0);
-    p.add(Buffer.from(frame1, 'hex'));
+    p.add(Buffer.from('01024865', 'hex'));
     assert.strictEqual(p.totalPayloadLength, 2);
-    p.add(Buffer.from(frame2, 'hex'));
+    p.add(Buffer.from('80036c6c6f', 'hex'));
     assert.strictEqual(p.totalPayloadLength, 0);
+    assert.strictEqual(message, 'Hello');
   });
 
   it('resets `totalPayloadLength` only on final frame (fragmented + ping)', function () {
     const p = new Receiver({}, 10);
+    const data = [];
 
-    const frame1 = '01024865';
-    const frame2 = '8900';
-    const frame3 = '80036c6c6f';
+    p.onmessage = p.onping = function (buf) {
+      data.push(buf.toString());
+    };
 
     assert.strictEqual(p.totalPayloadLength, 0);
-    p.add(Buffer.from(frame1, 'hex'));
+    p.add(Buffer.from('02024865', 'hex'));
     assert.strictEqual(p.totalPayloadLength, 2);
-    p.add(Buffer.from(frame2, 'hex'));
+    p.add(Buffer.from('8900', 'hex'));
     assert.strictEqual(p.totalPayloadLength, 2);
-    p.add(Buffer.from(frame3, 'hex'));
+    p.add(Buffer.from('80036c6c6f', 'hex'));
     assert.strictEqual(p.totalPayloadLength, 0);
+    assert.deepStrictEqual(data, ['', 'Hello']);
   });
 
   it('raises an error when RSV1 is on and permessage-deflate is disabled', function (done) {
@@ -538,7 +549,7 @@ describe('Receiver', function () {
     p.add(Buffer.from([0x88, 0x01, 0x00]));
   });
 
-  it('raises an error if a close frame contains a invalid close code', function (done) {
+  it('raises an error if a close frame contains an invalid close code', function (done) {
     const p = new Receiver();
 
     p.error = function (err, code) {
@@ -582,7 +593,7 @@ describe('Receiver', function () {
     p.add(Buffer.from(frame, 'hex'));
   });
 
-  it('raises an error on a 200 KiB long unmasked binary message when maxpayload is 20 KiB', function (done) {
+  it('raises an error on a 200 KiB long unmasked binary message when `maxPayload` is 20 KiB', function (done) {
     const p = new Receiver({}, 20 * 1024);
     const msg = crypto.randomBytes(200 * 1024);
 
@@ -650,13 +661,9 @@ describe('Receiver', function () {
     });
   });
 
-  it('will not crash if another message is received after receiving a message that exceeds maxpayload', function (done) {
-    const perMessageDeflate = new PerMessageDeflate({}, false, 2);
-    perMessageDeflate.accept([{}]);
-
-    const p = new Receiver({ 'permessage-deflate': perMessageDeflate }, 2);
-    const buf1 = Buffer.from('foooooooooooooooooooooooooooooooooooooooooooooo');
-    const buf2 = Buffer.from('baaaarrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrr');
+  it('doesn\'t crash if data is received after `maxPayload` is exceeded', function (done) {
+    const p = new Receiver({}, 5);
+    const buf = crypto.randomBytes(10);
 
     let gotError = false;
 
@@ -665,46 +672,140 @@ describe('Receiver', function () {
       assert.strictEqual(code, 1009);
     };
 
-    perMessageDeflate.compress(buf1, false, function (err, compressed1) {
+    p.add(Buffer.from([0x82, buf.length]));
+
+    assert.ok(gotError);
+    assert.strictEqual(p.onerror, null);
+
+    p.add(buf);
+    done();
+  });
+
+  it('consumes all data before calling `cleanup` callback (1/4)', function (done) {
+    const perMessageDeflate = new PerMessageDeflate();
+    perMessageDeflate.accept([{}]);
+
+    const p = new Receiver({ 'permessage-deflate': perMessageDeflate });
+    const buf = Buffer.from('Hello');
+    const results = [];
+
+    p.onmessage = (message) => results.push(message);
+
+    perMessageDeflate.compress(buf, true, (err, data) => {
       if (err) return done(err);
 
-      p.add(Buffer.from([0x41, compressed1.length]));
-      p.add(compressed1);
+      const frame = Buffer.concat([Buffer.from([0xc1, data.length]), data]);
 
-      assert.ok(gotError);
-      assert.strictEqual(p.onerror, null);
+      p.add(frame);
+      p.add(frame);
 
-      perMessageDeflate.compress(buf2, true, function (err, compressed2) {
-        if (err) return done(err);
+      assert.strictEqual(p.state, 5);
+      assert.strictEqual(p.bufferedBytes, frame.length);
 
-        p.add(Buffer.from([0x80, compressed2.length]));
-        p.add(compressed2);
+      p.cleanup(() => {
+        assert.deepStrictEqual(results, ['Hello', 'Hello']);
+        assert.strictEqual(p.onmessage, null);
         done();
       });
     });
   });
 
-  it('can cleanup when consuming data', function (done) {
+  it('consumes all data before calling `cleanup` callback (2/4)', function (done) {
     const perMessageDeflate = new PerMessageDeflate();
     perMessageDeflate.accept([{}]);
 
     const p = new Receiver({ 'permessage-deflate': perMessageDeflate });
     const buf = Buffer.from('Hello');
+    const results = [];
 
-    perMessageDeflate.compress(buf, true, function (err, compressed) {
+    p.onclose = (code, reason) => results.push(code, reason);
+    p.onmessage = (message) => results.push(message);
+
+    perMessageDeflate.compress(buf, true, (err, data) => {
       if (err) return done(err);
 
-      const data = Buffer.concat([Buffer.from([0xc1, compressed.length]), compressed]);
+      const textFrame = Buffer.concat([Buffer.from([0xc1, data.length]), data]);
+      const closeFrame = Buffer.from([0x88, 0x00]);
 
-      p.add(data);
-      p.add(data);
+      p.add(textFrame);
+      p.add(textFrame);
+      p.add(closeFrame);
 
       assert.strictEqual(p.state, 5);
-      assert.strictEqual(p.bufferedBytes, data.length);
+      assert.strictEqual(p.bufferedBytes, textFrame.length + closeFrame.length);
 
-      p.cleanup();
+      p.cleanup(() => {
+        assert.deepStrictEqual(results, ['Hello', 'Hello', 1000, '']);
+        assert.strictEqual(p.onmessage, null);
+        done();
+      });
+    });
+  });
+
+  it('consumes all data before calling `cleanup` callback (3/4)', function (done) {
+    const perMessageDeflate = new PerMessageDeflate();
+    perMessageDeflate.accept([{}]);
+
+    const p = new Receiver({ 'permessage-deflate': perMessageDeflate });
+    const buf = Buffer.from('Hello');
+    const results = [];
 
-      perMessageDeflate._inflate.flush(done);
+    p.onerror = (err, code) => results.push(err.message, code);
+    p.onmessage = (message) => results.push(message);
+
+    perMessageDeflate.compress(buf, true, (err, data) => {
+      if (err) return done(err);
+
+      const textFrame = Buffer.concat([Buffer.from([0xc1, data.length]), data]);
+      const invalidFrame = Buffer.from([0xa0, 0x00]);
+
+      p.add(textFrame);
+      p.add(textFrame);
+      p.add(invalidFrame);
+
+      assert.strictEqual(p.state, 5);
+      assert.strictEqual(p.bufferedBytes, textFrame.length + invalidFrame.length);
+
+      p.cleanup(() => {
+        assert.deepStrictEqual(results, [
+          'Hello',
+          'Hello',
+          'RSV2 and RSV3 must be clear',
+          1002
+        ]);
+        assert.strictEqual(p.onmessage, null);
+        done();
+      });
+    });
+  });
+
+  it('consumes all data before calling `cleanup` callback (4/4)', function (done) {
+    const perMessageDeflate = new PerMessageDeflate();
+    perMessageDeflate.accept([{}]);
+
+    const p = new Receiver({ 'permessage-deflate': perMessageDeflate });
+    const buf = Buffer.from('Hello');
+    const results = [];
+
+    p.onmessage = (message) => results.push(message);
+
+    perMessageDeflate.compress(buf, true, (err, data) => {
+      if (err) return done(err);
+
+      const textFrame = Buffer.concat([Buffer.from([0xc1, data.length]), data]);
+      const incompleteFrame = Buffer.from([0x82, 0x0a, 0x00, 0x00]);
+
+      p.add(textFrame);
+      p.add(incompleteFrame);
+
+      assert.strictEqual(p.state, 5);
+      assert.strictEqual(p.bufferedBytes, incompleteFrame.length);
+
+      p.cleanup(() => {
+        assert.deepStrictEqual(results, ['Hello']);
+        assert.strictEqual(p.onmessage, null);
+        done();
+      });
     });
   });
 });
diff --git a/test/WebSocket.test.js b/test/WebSocket.test.js
index 24732207..7d938880 100644
--- a/test/WebSocket.test.js
+++ b/test/WebSocket.test.js
@@ -945,7 +945,7 @@ describe('WebSocket', function () {
       server.createServer(++port, (srv) => {
         const ws = new WebSocket(`ws://localhost:${port}`);
 
-        ws.on('open', () => ws.close(1000, 'some reason', { mask: true }));
+        ws.on('open', () => ws.close(1000, 'some reason'));
 
         srv.on('close', (code, message, flags) => {
           assert.ok(flags.masked);
@@ -963,7 +963,7 @@ describe('WebSocket', function () {
 
         ws.on('open', () => {
           connectedOnce = true;
-          ws.close(1000, 'some reason', {mask: true});
+          ws.close(1000, 'some reason');
         });
 
         ws.on('close', () => {
@@ -974,28 +974,28 @@ describe('WebSocket', function () {
       });
     });
 
-    it('consumes all data when the server socket closed', function (done) {
-      const wss = new WebSocketServer({ port: ++port }, () => {
-        wss.on('connection', (conn) => {
-          conn.send('foo');
-          conn.send('bar');
-          conn.send('baz');
-          conn.close();
-        });
-
+    it('permits all buffered data to be delivered', function (done) {
+      const wss = new WebSocketServer({
+        perMessageDeflate: { threshold: 0 },
+        port: ++port
+      }, () => {
         const ws = new WebSocket(`ws://localhost:${port}`);
         const messages = [];
 
-        ws.on('message', (message) => {
-          messages.push(message);
-          if (messages.length === 3) {
-            assert.deepStrictEqual(messages, ['foo', 'bar', 'baz']);
-
-            wss.close(done);
-            ws.terminate();
-          }
+        ws.on('message', (message) => messages.push(message));
+        ws.on('close', (code) => {
+          assert.strictEqual(code, 1000);
+          assert.deepStrictEqual(messages, ['foo', 'bar', 'baz']);
+          wss.close(done);
         });
       });
+
+      wss.on('connection', (ws) => {
+        ws.send('foo');
+        ws.send('bar');
+        ws.send('baz');
+        ws.close();
+      });
     });
 
     it('allows close code 1013', function (done) {
@@ -1010,6 +1010,39 @@ describe('WebSocket', function () {
 
       wss.on('connection', (ws) => ws.close(1013));
     });
+
+    it('closes the connection when an error occurs', function (done) {
+      const server = http.createServer();
+      const wss = new WebSocketServer({ server });
+      let closed = false;
+
+      wss.on('connection', (ws) => {
+        ws.on('error', (err) => {
+          assert.ok(err instanceof Error);
+          assert.strictEqual(err.message, 'RSV2 and RSV3 must be clear');
+
+          ws.on('close', (code, reason) => {
+            assert.strictEqual(code, 1006);
+            assert.strictEqual(reason, '');
+
+            closed = true;
+          });
+        });
+      });
+
+      server.listen(++port, () => {
+        const ws = new WebSocket(`ws://localhost:${port}`);
+
+        ws.on('open', () => ws._socket.write(Buffer.from([0xa1, 0x00])));
+        ws.on('close', (code, reason) => {
+          assert.strictEqual(code, 1002);
+          assert.strictEqual(reason, '');
+          assert.ok(closed);
+
+          server.close(done);
+        });
+      });
+    });
   });
 
   describe('WHATWG API emulation', function () {
@@ -1613,6 +1646,30 @@ describe('WebSocket', function () {
       });
     });
 
+    it('consumes all received data when connection is closed abnormally', function (done) {
+      const wss = new WebSocketServer({
+        perMessageDeflate: { threshold: 0 },
+        port: ++port
+      }, () => {
+        const ws = new WebSocket(`ws://localhost:${port}`);
+        const messages = [];
+
+        ws.on('message', (message) => messages.push(message));
+        ws.on('close', (code) => {
+          assert.strictEqual(code, 1006);
+          assert.deepStrictEqual(messages, ['foo', 'bar', 'baz', 'qux']);
+          wss.close(done);
+        });
+      });
+
+      wss.on('connection', (ws) => {
+        ws.send('foo');
+        ws.send('bar');
+        ws.send('baz');
+        ws.send('qux', () => ws._socket.end());
+      });
+    });
+
     describe('#send', function () {
       it('can set the compress option true when perMessageDeflate is disabled', function (done) {
         const wss = new WebSocketServer({ port: ++port }, () => {