Repository: websockets/ws
Pull request #1302 (2018-02-15 07:20:23)

Make `Receiver` inherit from `stream.Writable`
This makes `Receiver` inherit from `stream.Writable`.

The main reason for this change is to have back-pressure properly handled when permessage-deflate is enabled.

There are no significant performance differences but memory usage is a lot higher. 100k connections use ~400 MB on master and ~600 MB with this patch.

I'm opening this PR to discuss whether or not this makes sense.

lpinca: Managed to reduce memory consumption by not using `Readable.prototype.pipe()`. Now 100k connections use ~450 MB which is ~10% more memory than master.

I think this is a good trade off.
lpinca: If there is anyone watching this, can you please share your opinion about `net.Socket` errors? Should we re-emit them or not?

- Node.js core is not consistent about these errors on the HTTP stack. On the server they are not re-emitted, or better they are re-emitted using the [`'clientError'`](https://nodejs.org/dist/latest/docs/api/http.html#http_event_clienterror) event on the server but you can omit the `'error'` listener. On the client they are re-emitted on the [`ClientRequest`](https://nodejs.org/dist/latest/docs/api/http.html#http_class_http_clientrequest) instance and you can't omit the `'error'` listener.
- From https://github.com/websockets/ws/issues/1256 it seems that these errors end up being ignored as there is not much that can be done to prevent them. If the common action is to filter them out, I think there is no point in re-emitting them as the WebSocket is closed anyway with an abnormal closure status code.
- The WebSocket browser implementation does not call the `onerror` listener if a network error occurs after the handshake completes.

cc: @websockets/admin 
lpinca: This library has ~15M monthly downloads, it's a dependency of many popular project, it's used by some big corporations but almost no one actively participate in its development. This is weird.

@3rd-Eden can you please weigh in when you have time?
3rd-Eden: I simply do not know enough of Node.js net code internal to make a comment that will be remotely useful here. 

The performance loss is acceptable as we finally handle back pressure correctly. We can see if we can performance back in different ways. 
lpinca: Performance is the same, it's memory usage that's slightly higher but that's acceptable.

My doubt is whether or not we should re-emit network errors like `ECONNRESET`, `EPIPE`, `ECONNABORTED`. All of these were swallowed before version `3.3.3` and from #1256 it seems that the common action is to ignore them.
3rd-Eden: @lpinca I agree, I don't think those errors are usefull because it just means that the connection got closed in some way. 
lpinca: Ok, thank you.

Patch:
diff --git a/.travis.yml b/.travis.yml
index 1d63d7d3..e21953a5 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -5,6 +5,6 @@ node_js:
   - "8"
   - "6"
   - "4"
-  - "4.2.0"
+  - "4.2.2"
 after_success:
   - "npm install coveralls@3 && nyc report --reporter=text-lcov | coveralls"
diff --git a/appveyor.yml b/appveyor.yml
index 2502b8c8..394bc1e8 100644
--- a/appveyor.yml
+++ b/appveyor.yml
@@ -4,7 +4,7 @@ environment:
     - nodejs_version: "8"
     - nodejs_version: "6"
     - nodejs_version: "4"
-    - nodejs_version: "4.2.0"
+    - nodejs_version: "4.2.2"
 platform:
   - x86
   - x64
diff --git a/doc/ws.md b/doc/ws.md
index 464df38b..f505ed5e 100644
--- a/doc/ws.md
+++ b/doc/ws.md
@@ -244,8 +244,7 @@ human-readable string explaining why the connection has been closed.
 
 - `error` {Error}
 
-Emitted when an error occurs. Errors from the underlying `net.Socket` are
-forwarded here.
+Emitted when an error occurs.
 
 ### Event: 'message'
 
diff --git a/lib/constants.js b/lib/constants.js
index 39044146..613a6766 100644
--- a/lib/constants.js
+++ b/lib/constants.js
@@ -4,7 +4,11 @@ const safeBuffer = require('safe-buffer');
 
 const Buffer = safeBuffer.Buffer;
 
-exports.BINARY_TYPES = ['nodebuffer', 'arraybuffer', 'fragments'];
-exports.GUID = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11';
-exports.EMPTY_BUFFER = Buffer.alloc(0);
-exports.NOOP = () => {};
+module.exports = {
+  BINARY_TYPES: ['nodebuffer', 'arraybuffer', 'fragments'],
+  GUID: '258EAFA5-E914-47DA-95CA-C5AB0DC85B11',
+  kStatusCode: Symbol('status-code'),
+  kWebSocket: Symbol('websocket'),
+  EMPTY_BUFFER: Buffer.alloc(0),
+  NOOP: () => {}
+};
diff --git a/lib/permessage-deflate.js b/lib/permessage-deflate.js
index cabab971..05832f28 100644
--- a/lib/permessage-deflate.js
+++ b/lib/permessage-deflate.js
@@ -5,19 +5,20 @@ const Limiter = require('async-limiter');
 const zlib = require('zlib');
 
 const bufferUtil = require('./buffer-util');
+const constants = require('./constants');
 
 const Buffer = safeBuffer.Buffer;
 
 const TRAILER = Buffer.from([0x00, 0x00, 0xff, 0xff]);
 const EMPTY_BLOCK = Buffer.from([0x00]);
 
+const kPerMessageDeflate = Symbol('permessage-deflate');
 const kWriteInProgress = Symbol('write-in-progress');
 const kPendingClose = Symbol('pending-close');
 const kTotalLength = Symbol('total-length');
 const kCallback = Symbol('callback');
 const kBuffers = Symbol('buffers');
 const kError = Symbol('error');
-const kOwner = Symbol('owner');
 
 //
 // We limit zlib concurrency, which prevents severe memory fragmentation
@@ -346,15 +347,11 @@ class PerMessageDeflate {
         : this.params[key];
 
       this._inflate = zlib.createInflateRaw(
-        Object.assign(
-          {},
-          this._options.zlibInflateOptions,
-          { windowBits }
-        )
+        Object.assign({}, this._options.zlibInflateOptions, { windowBits })
       );
+      this._inflate[kPerMessageDeflate] = this;
       this._inflate[kTotalLength] = 0;
       this._inflate[kBuffers] = [];
-      this._inflate[kOwner] = this;
       this._inflate.on('error', inflateOnError);
       this._inflate.on('data', inflateOnData);
     }
@@ -492,15 +489,15 @@ function inflateOnData (chunk) {
   this[kTotalLength] += chunk.length;
 
   if (
-    this[kOwner]._maxPayload < 1 ||
-    this[kTotalLength] <= this[kOwner]._maxPayload
+    this[kPerMessageDeflate]._maxPayload < 1 ||
+    this[kTotalLength] <= this[kPerMessageDeflate]._maxPayload
   ) {
     this[kBuffers].push(chunk);
     return;
   }
 
   this[kError] = new RangeError('Max payload size exceeded');
-  this[kError].closeCode = 1009;
+  this[kError][constants.kStatusCode] = 1009;
   this.removeListener('data', inflateOnData);
   this.reset();
 }
@@ -516,6 +513,7 @@ function inflateOnError (err) {
   // There is no need to call `Zlib#close()` as the handle is automatically
   // closed when an error is emitted.
   //
-  this[kOwner]._inflate = null;
+  this[kPerMessageDeflate]._inflate = null;
+  err[constants.kStatusCode] = 1007;
   this[kCallback](err);
 }
diff --git a/lib/receiver.js b/lib/receiver.js
index ad1187b8..a66cc10e 100644
--- a/lib/receiver.js
+++ b/lib/receiver.js
@@ -1,6 +1,7 @@
 'use strict';
 
 const safeBuffer = require('safe-buffer');
+const stream = require('stream');
 
 const PerMessageDeflate = require('./permessage-deflate');
 const bufferUtil = require('./buffer-util');
@@ -18,8 +19,10 @@ const INFLATING = 5;
 
 /**
  * HyBi Receiver implementation.
+ *
+ * @extends stream.Writable
  */
-class Receiver {
+class Receiver extends stream.Writable {
   /**
    * Creates a Receiver instance.
    *
@@ -28,7 +31,10 @@ class Receiver {
    * @param {String} binaryType The type for binary data
    */
   constructor (extensions, maxPayload, binaryType) {
+    super();
+
     this._binaryType = binaryType || constants.BINARY_TYPES[0];
+    this[constants.kWebSocket] = undefined;
     this._extensions = extensions || {};
     this._maxPayload = maxPayload | 0;
 
@@ -37,46 +43,43 @@ class Receiver {
 
     this._compressed = false;
     this._payloadLength = 0;
+    this._mask = undefined;
     this._fragmented = 0;
     this._masked = false;
     this._fin = false;
-    this._mask = null;
     this._opcode = 0;
 
     this._totalPayloadLength = 0;
     this._messageLength = 0;
     this._fragments = [];
 
-    this._cleanupCallback = null;
-    this._isCleaningUp = false;
-    this._hadError = false;
+    this._state = GET_INFO;
     this._loop = false;
+  }
 
-    this.add = this.add.bind(this);
-    this.onmessage = null;
-    this.onclose = null;
-    this.onerror = null;
-    this.onping = null;
-    this.onpong = null;
+  /**
+   * Implements `Writable.prototype._write()`.
+   *
+   * @param {Buffer} chunk The chunk of data to write
+   * @param {String} encoding The character encoding of `chunk`
+   * @param {Function} cb Callback
+   */
+  _write (chunk, encoding, cb) {
+    if (this._opcode === 0x08) return cb();
 
-    this._state = GET_INFO;
+    this._bufferedBytes += chunk.length;
+    this._buffers.push(chunk);
+    this.startLoop(cb);
   }
 
   /**
-   * Consumes `n` bytes from the buffered data, calls `cleanup` if necessary.
+   * Consumes `n` bytes from the buffered data.
    *
    * @param {Number} n The number of bytes to consume
-   * @return {(Buffer|null)} The consumed bytes or `null` if `n` bytes are not
-   *     available
+   * @return {Buffer} The consumed bytes
    * @private
    */
   consume (n) {
-    if (this._bufferedBytes < n) {
-      this._loop = false;
-      if (this._isCleaningUp) this.cleanup(this._cleanupCallback);
-      return null;
-    }
-
     this._bufferedBytes -= n;
 
     if (n === this._buffers[0].length) return this._buffers.shift();
@@ -105,74 +108,66 @@ class Receiver {
     return dst;
   }
 
-  /**
-   * Adds new data to the parser.
-   *
-   * @param {Buffer} chunk A chunk of data
-   * @public
-   */
-  add (chunk) {
-    this._bufferedBytes += chunk.length;
-    this._buffers.push(chunk);
-    this.startLoop();
-  }
-
   /**
    * Starts the parsing loop.
    *
+   * @param {Function} cb Callback
    * @private
    */
-  startLoop () {
+  startLoop (cb) {
+    var err;
     this._loop = true;
 
     do {
       switch (this._state) {
         case GET_INFO:
-          this.getInfo();
+          err = this.getInfo();
           break;
         case GET_PAYLOAD_LENGTH_16:
-          this.getPayloadLength16();
+          err = this.getPayloadLength16();
           break;
         case GET_PAYLOAD_LENGTH_64:
-          this.getPayloadLength64();
+          err = this.getPayloadLength64();
           break;
         case GET_MASK:
           this.getMask();
           break;
         case GET_DATA:
-          this.getData();
+          err = this.getData(cb);
           break;
         default: // `INFLATING`
           this._loop = false;
+          return;
       }
     } while (this._loop);
+
+    cb(err);
   }
 
   /**
    * Reads the first two bytes of a frame.
    *
+   * @return {(RangeError|undefined)} A possible error
    * @private
    */
   getInfo () {
+    if (this._bufferedBytes < 2) {
+      this._loop = false;
+      return;
+    }
+
     const buf = this.consume(2);
-    if (buf === null) return;
 
     if ((buf[0] & 0x30) !== 0x00) {
-      this.error(
-        new RangeError('Invalid WebSocket frame: RSV2 and RSV3 must be clear'),
-        1002
-      );
-      return;
+      this._loop = false;
+      return error(RangeError, 'RSV2 and RSV3 must be clear', true, 1002);
     }
 
     const compressed = (buf[0] & 0x40) === 0x40;
 
     if (compressed && !this._extensions[PerMessageDeflate.extensionName]) {
-      this.error(
-        new RangeError('Invalid WebSocket frame: RSV1 must be clear'),
-        1002
-      );
-      return;
+      this._loop = false;
+      return error(RangeError, 'RSV1 must be clear', true, 1002);
     }
 
     this._fin = (buf[0] & 0x80) === 0x80;
@@ -181,102 +176,85 @@ class Receiver {
 
     if (this._opcode === 0x00) {
       if (compressed) {
-        this.error(
-          new RangeError('Invalid WebSocket frame: RSV1 must be clear'),
-          1002
-        );
-        return;
+        this._loop = false;
+        return error(RangeError, 'RSV1 must be clear', true, 1002);
       }
 
       if (!this._fragmented) {
-        this.error(
-          new RangeError('Invalid WebSocket frame: invalid opcode 0'),
-          1002
-        );
-        return;
-      } else {
-        this._opcode = this._fragmented;
+        this._loop = false;
+        return error(RangeError, 'invalid opcode 0', true, 1002);
       }
+
+      this._opcode = this._fragmented;
     } else if (this._opcode === 0x01 || this._opcode === 0x02) {
       if (this._fragmented) {
-        this.error(
-          new RangeError(
-            `Invalid WebSocket frame: invalid opcode ${this._opcode}`
-          ),
-          1002
-        );
-        return;
+        this._loop = false;
+        return error(RangeError, `invalid opcode ${this._opcode}`, true, 1002);
       }
 
       this._compressed = compressed;
     } else if (this._opcode > 0x07 && this._opcode < 0x0b) {
       if (!this._fin) {
-        this.error(
-          new RangeError('Invalid WebSocket frame: FIN must be set'),
-          1002
-        );
-        return;
+        this._loop = false;
+        return error(RangeError, 'FIN must be set', true, 1002);
       }
 
       if (compressed) {
-        this.error(
-          new RangeError('Invalid WebSocket frame: RSV1 must be clear'),
-          1002
-        );
-        return;
+        this._loop = false;
+        return error(RangeError, 'RSV1 must be clear', true, 1002);
       }
 
       if (this._payloadLength > 0x7d) {
-        this.error(
-          new RangeError(
-            `Invalid WebSocket frame: invalid payload length ` +
-              `${this._payloadLength}`
-          ),
+        this._loop = false;
+        return error(
+          RangeError,
+          `invalid payload length ${this._payloadLength}`,
+          true,
           1002
         );
-        return;
       }
     } else {
-      this.error(
-        new RangeError(
-          `Invalid WebSocket frame: invalid opcode ${this._opcode}`
-        ),
-        1002
-      );
-      return;
+      this._loop = false;
+      return error(RangeError, `invalid opcode ${this._opcode}`, true, 1002);
     }
 
     if (!this._fin && !this._fragmented) this._fragmented = this._opcode;
-
     this._masked = (buf[1] & 0x80) === 0x80;
 
     if (this._payloadLength === 126) this._state = GET_PAYLOAD_LENGTH_16;
     else if (this._payloadLength === 127) this._state = GET_PAYLOAD_LENGTH_64;
-    else this.haveLength();
+    else return this.haveLength();
   }
 
   /**
    * Gets extended payload length (7+16).
    *
+   * @return {(RangeError|undefined)} A possible error
    * @private
    */
   getPayloadLength16 () {
-    const buf = this.consume(2);
-    if (buf === null) return;
+    if (this._bufferedBytes < 2) {
+      this._loop = false;
+      return;
+    }
 
-    this._payloadLength = buf.readUInt16BE(0, true);
-    this.haveLength();
+    this._payloadLength = this.consume(2).readUInt16BE(0, true);
+    return this.haveLength();
   }
 
   /**
    * Gets extended payload length (7+64).
    *
+   * @return {(RangeError|undefined)} A possible error
    * @private
    */
   getPayloadLength64 () {
-    const buf = this.consume(8);
-    if (buf === null) return;
+    if (this._bufferedBytes < 8) {
+      this._loop = false;
+      return;
+    }
 
+    const buf = this.consume(8);
     const num = buf.readUInt32BE(0, true);
 
     //
@@ -284,27 +262,32 @@ class Receiver {
     // if payload length is greater than this number.
     //
     if (num > Math.pow(2, 53 - 32) - 1) {
-      this.error(
-        new RangeError(
-          'Unsupported WebSocket frame: payload length > 2^53 - 1'
-        ),
+      this._loop = false;
+      return error(
+        RangeError,
+        'Unsupported WebSocket frame: payload length > 2^53 - 1',
+        false,
         1009
       );
-      return;
     }
 
     this._payloadLength = num * Math.pow(2, 32) + buf.readUInt32BE(4, true);
-    this.haveLength();
+    return this.haveLength();
   }
 
   /**
    * Payload length has been read.
    *
+   * @return {(RangeError|undefined)} A possible error
    * @private
    */
   haveLength () {
-    if (this._opcode < 0x08 && this.maxPayloadExceeded(this._payloadLength)) {
-      return;
+    if (this._payloadLength && this._opcode < 0x08) {
+      this._totalPayloadLength += this._payloadLength;
+      if (this._totalPayloadLength > this._maxPayload && this._maxPayload > 0) {
+        this._loop = false;
+        return error(RangeError, 'Max payload size exceeded', false, 1009);
+      }
     }
 
     if (this._masked) this._state = GET_MASK;
@@ -317,60 +300,88 @@ class Receiver {
    * @private
    */
   getMask () {
-    this._mask = this.consume(4);
-    if (this._mask === null) return;
+    if (this._bufferedBytes < 4) {
+      this._loop = false;
+      return;
+    }
 
+    this._mask = this.consume(4);
     this._state = GET_DATA;
   }
 
   /**
    * Reads data bytes.
    *
+   * @param {Function} cb Callback
+   * @return {(Error|RangeError|undefined)} A possible error
    * @private
    */
-  getData () {
+  getData (cb) {
     var data = constants.EMPTY_BUFFER;
 
     if (this._payloadLength) {
-      data = this.consume(this._payloadLength);
-      if (data === null) return;
+      if (this._bufferedBytes < this._payloadLength) {
+        this._loop = false;
+        return;
+      }
 
+      data = this.consume(this._payloadLength);
       if (this._masked) bufferUtil.unmask(data, this._mask);
     }
 
-    if (this._opcode > 0x07) {
-      this.controlMessage(data);
-    } else if (this._compressed) {
+    if (this._opcode > 0x07) return this.controlMessage(data);
+
+    if (this._compressed) {
       this._state = INFLATING;
-      this.decompress(data);
-    } else if (this.pushFragment(data)) {
-      this.dataMessage();
+      this.decompress(data, cb);
+      return;
+    }
+
+    if (data.length) {
+      //
+      // This message is not compressed so its lenght is the sum of the payload
+      // length of all fragments.
+      //
+      this._messageLength = this._totalPayloadLength;
+      this._fragments.push(data);
     }
+
+    return this.dataMessage();
   }
 
   /**
    * Decompresses data.
    *
    * @param {Buffer} data Compressed data
+   * @param {Function} cb Callback
    * @private
    */
-  decompress (data) {
+  decompress (data, cb) {
     const perMessageDeflate = this._extensions[PerMessageDeflate.extensionName];
 
     perMessageDeflate.decompress(data, this._fin, (err, buf) => {
-      if (err) {
-        this.error(err, err.closeCode === 1009 ? 1009 : 1007);
-        return;
+      if (err) return cb(err);
+
+      if (buf.length) {
+        this._messageLength += buf.length;
+        if (this._messageLength > this._maxPayload && this._maxPayload > 0) {
+          return cb(error(RangeError, 'Max payload size exceeded', false, 1009));
+        }
+
+        this._fragments.push(buf);
       }
 
-      if (this.pushFragment(buf)) this.dataMessage();
-      this.startLoop();
+      const er = this.dataMessage();
+      if (er) return cb(er);
+
+      this.startLoop(cb);
     });
   }
 
   /**
    * Handles a data message.
    *
+   * @return {(Error|undefined)} A possible error
    * @private
    */
   dataMessage () {
@@ -394,19 +405,16 @@ class Receiver {
           data = fragments;
         }
 
-        this.onmessage(data);
+        this.emit('message', data);
       } else {
         const buf = toBuffer(fragments, messageLength);
 
         if (!validation.isValidUTF8(buf)) {
-          this.error(
-            new Error('Invalid WebSocket frame: invalid UTF-8 sequence'),
-            1007
-          );
-          return;
+          this._loop = false;
+          return error(Error, 'invalid UTF-8 sequence', true, 1007);
         }
 
-        this.onmessage(buf.toString());
+        this.emit('message', buf.toString());
       }
     }
 
@@ -417,149 +425,68 @@ class Receiver {
    * Handles a control message.
    *
    * @param {Buffer} data Data to handle
+   * @return {(Error|RangeError|undefined)} A possible error
    * @private
    */
   controlMessage (data) {
     if (this._opcode === 0x08) {
+      this._loop = false;
+
       if (data.length === 0) {
-        this._loop = false;
-        this.onclose(1005, '');
-        this.cleanup(this._cleanupCallback);
+        this.emit('close', 1005, '');
+        this.end();
       } else if (data.length === 1) {
-        this.error(
-          new RangeError('Invalid WebSocket frame: invalid payload length 1'),
-          1002
-        );
+        return error(RangeError, 'invalid payload length 1', true, 1002);
       } else {
         const code = data.readUInt16BE(0, true);
 
         if (!validation.isValidStatusCode(code)) {
-          this.error(
-            new RangeError(
-              `Invalid WebSocket frame: invalid status code ${code}`
-            ),
-            1002
-          );
-          return;
+          return error(RangeError, `invalid status code ${code}`, true, 1002);
         }
 
         const buf = data.slice(2);
 
         if (!validation.isValidUTF8(buf)) {
-          this.error(
-            new Error('Invalid WebSocket frame: invalid UTF-8 sequence'),
-            1007
-          );
-          return;
+          return error(Error, 'invalid UTF-8 sequence', true, 1007);
         }
 
-        this._loop = false;
-        this.onclose(code, buf.toString());
-        this.cleanup(this._cleanupCallback);
+        this.emit('close', code, buf.toString());
+        this.end();
       }
 
       return;
     }
 
-    if (this._opcode === 0x09) this.onping(data);
-    else this.onpong(data);
+    if (this._opcode === 0x09) this.emit('ping', data);
+    else this.emit('pong', data);
 
     this._state = GET_INFO;
   }
-
-  /**
-   * Handles an error.
-   *
-   * @param {Error} err The error
-   * @param {Number} code Close code
-   * @private
-   */
-  error (err, code) {
-    this._hadError = true;
-    this._loop = false;
-    this.onerror(err, code);
-    this.cleanup(this._cleanupCallback);
-  }
-
-  /**
-   * Checks payload size, disconnects socket when it exceeds `maxPayload`.
-   *
-   * @param {Number} length Payload length
-   * @private
-   */
-  maxPayloadExceeded (length) {
-    if (length === 0 || this._maxPayload < 1) return false;
-
-    const fullLength = this._totalPayloadLength + length;
-
-    if (fullLength <= this._maxPayload) {
-      this._totalPayloadLength = fullLength;
-      return false;
-    }
-
-    this.error(new RangeError('Max payload size exceeded'), 1009);
-    return true;
-  }
-
-  /**
-   * Appends a fragment in the fragments array after checking that the sum of
-   * fragment lengths does not exceed `maxPayload`.
-   *
-   * @param {Buffer} fragment The fragment to add
-   * @return {Boolean} `true` if `maxPayload` is not exceeded, else `false`
-   * @private
-   */
-  pushFragment (fragment) {
-    if (fragment.length === 0) return true;
-
-    const totalLength = this._messageLength + fragment.length;
-
-    if (this._maxPayload < 1 || totalLength <= this._maxPayload) {
-      this._messageLength = totalLength;
-      this._fragments.push(fragment);
-      return true;
-    }
-
-    this.error(new RangeError('Max payload size exceeded'), 1009);
-    return false;
-  }
-
-  /**
-   * Releases resources used by the receiver.
-   *
-   * @param {Function} cb Callback
-   * @public
-   */
-  cleanup (cb) {
-    if (this._extensions === null) {
-      if (cb) cb();
-      return;
-    }
-
-    if (!this._hadError && (this._loop || this._state === INFLATING)) {
-      this._cleanupCallback = cb;
-      this._isCleaningUp = true;
-      return;
-    }
-
-    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();
-  }
 }
 
 module.exports = Receiver;
 
+/**
+ * Builds an error object.
+ *
+ * @param {(Error|RangeError)} ErrorCtor The error constructor
+ * @param {String} message The error message
+ * @param {Boolean} prefix Specifies whether or not to add a default prefix to
+ *     `message`
+ * @param {Number} statusCode The status code
+ * @return {(Error|RangeError)} The error
+ * @private
+ */
+function error (ErrorCtor, message, prefix, statusCode) {
+  const err = new ErrorCtor(
+    prefix ? `Invalid WebSocket frame: ${message}` : message
+  );
+
+  Error.captureStackTrace(err, error);
+  err[constants.kStatusCode] = statusCode;
+  return err;
+}
+
 /**
  * Makes a buffer from a list of fragments.
  *
diff --git a/lib/websocket.js b/lib/websocket.js
index a5f606f9..f3b460e3 100644
--- a/lib/websocket.js
+++ b/lib/websocket.js
@@ -14,6 +14,7 @@ const Receiver = require('./receiver');
 const Sender = require('./sender');
 
 const readyStates = ['CONNECTING', 'OPEN', 'CLOSING', 'CLOSED'];
+const kWebSocket = constants.kWebSocket;
 const protocolVersions = [8, 13];
 const closeTimeout = 30 * 1000; // Allow 30 seconds to terminate the connection cleanly.
 
@@ -37,19 +38,16 @@ class WebSocket extends EventEmitter {
     this.protocol = '';
 
     this._binaryType = constants.BINARY_TYPES[0];
-    this._finalize = this.finalize.bind(this);
     this._closeFrameReceived = false;
     this._closeFrameSent = false;
     this._closeMessage = '';
     this._closeTimer = null;
-    this._finalized = false;
     this._closeCode = 1006;
     this._extensions = {};
     this._isServer = true;
     this._receiver = null;
     this._sender = null;
     this._socket = null;
-    this._error = null;
 
     if (address !== null) {
       if (!protocols) {
@@ -119,126 +117,75 @@ class WebSocket extends EventEmitter {
    * @private
    */
   setSocket (socket, head, maxPayload) {
-    socket.setTimeout(0);
-    socket.setNoDelay();
-
-    socket.on('close', this._finalize);
-    socket.on('error', this._finalize);
-    socket.on('end', this._finalize);
+    const receiver = new Receiver(
+      this._extensions,
+      maxPayload,
+      this._binaryType
+    );
 
-    this._receiver = new Receiver(this._extensions, maxPayload, this.binaryType);
     this._sender = new Sender(socket, this._extensions);
+    this._receiver = receiver;
     this._socket = socket;
 
-    if (head.length > 0) socket.unshift(head);
-
-    socket.on('data', this._receiver.add);
-
-    this._receiver.onmessage = (data) => this.emit('message', data);
-    this._receiver.onping = (data) => {
-      this.pong(data, !this._isServer, constants.NOOP);
-      this.emit('ping', data);
-    };
-    this._receiver.onpong = (data) => this.emit('pong', data);
-    this._receiver.onclose = (code, reason) => {
-      //
-      // Discard any additional data that is received on the socket.
-      //
-      this._socket.removeListener('data', this._receiver.add);
+    receiver[kWebSocket] = this;
+    socket[kWebSocket] = this;
 
-      this._closeFrameReceived = true;
-      this._closeMessage = reason;
-      this._closeCode = code;
+    receiver.on('close', receiverOnClose);
+    receiver.on('drain', receiverOnDrain);
+    receiver.on('error', receiverOnError);
+    receiver.on('message', receiverOnMessage);
+    receiver.on('ping', receiverOnPing);
+    receiver.on('pong', receiverOnPong);
 
-      if (code === 1005) this.close();
-      else this.close(code, reason);
-    };
-    this._receiver.onerror = (error, code) => {
-      if (this._error) return;
+    socket.setTimeout(0);
+    socket.setNoDelay();
 
-      this._closeCode = code;
+    if (head.length > 0) socket.unshift(head);
 
-      if (!this._finalized) this.finalize(error);
-      else this.emit('error', error);
-    };
+    socket.on('close', socketOnClose);
+    socket.on('data', socketOnData);
+    socket.on('end', socketOnEnd);
+    socket.on('error', socketOnError);
 
     this.readyState = WebSocket.OPEN;
     this.emit('open');
   }
 
   /**
-   * Clean up internal resources and emit the `'close'` event.
+   * Emit the `'close'` event.
    *
-   * @param {(Boolean|Error)} error Indicates whether or not an error occurred
    * @private
    */
-  finalize (error) {
-    if (this._finalized) return;
-
-    this.readyState = WebSocket.CLOSING;
-    this._finalized = true;
+  emitClose () {
+    this.readyState = WebSocket.CLOSED;
 
     if (!this._socket) {
-      //
-      // `error` is always an `Error` instance in this case.
-      //
-      this.emit('error', error);
-      this.readyState = WebSocket.CLOSED;
       this.emit('close', this._closeCode, this._closeMessage);
       return;
     }
 
-    clearTimeout(this._closeTimer);
-
-    this._socket.removeListener('data', this._receiver.add);
-    this._socket.removeListener('close', this._finalize);
-    this._socket.removeListener('error', this._finalize);
-    this._socket.removeListener('end', this._finalize);
-    this._socket.on('error', constants.NOOP);
-
-    if (error) {
-      if (error !== true) this._error = error;
-      this._socket.destroy();
-    } else {
-      this._socket.end();
+    if (this._extensions[PerMessageDeflate.extensionName]) {
+      this._extensions[PerMessageDeflate.extensionName].cleanup();
     }
 
-    this._receiver.cleanup(() => {
-      const err = this._error;
-
-      if (err) {
-        this._error = null;
-        this.emit('error', err);
-      }
-
-      this.readyState = WebSocket.CLOSED;
-
-      if (this._extensions[PerMessageDeflate.extensionName]) {
-        this._extensions[PerMessageDeflate.extensionName].cleanup();
-      }
-
-      this.emit('close', this._closeCode, this._closeMessage);
-    });
+    this._receiver.removeAllListeners();
+    this.emit('close', this._closeCode, this._closeMessage);
   }
 
   /**
    * Start a closing handshake.
    *
-   *            +----------+     +-----------+   +----------+
-   *     + - - -|ws.close()|---->|close frame|-->|ws.close()|- - - -
-   *            +----------+     +-----------+   +----------+       |
-   *     |      +----------+     +-----------+         |
-   *            |ws.close()|<----|close frame|<--------+            |
-   *            +----------+     +-----------+         |
-   *  CLOSING         |              +---+             |         CLOSING
-   *                  |          +---|fin|<------------+
-   *     |            |          |   +---+                          |
-   *                  |          |   +---+      +-------------+
-   *     |            +----------+-->|fin|----->|ws.finalize()| - - +
-   *                             |   +---+      +-------------+
-   *     |     +-------------+   |
-   *      - - -|ws.finalize()|<--+
-   *           +-------------+
+   *          +----------+   +-----------+   +----------+
+   *     - - -|ws.close()|-->|close frame|-->|ws.close()|- - -
+   *    |     +----------+   +-----------+   +----------+     |
+   *          +----------+   +-----------+         |
+   * CLOSING  |ws.close()|<--|close frame|<--+-----+       CLOSING
+   *          +----------+   +-----------+   |
+   *    |           |                        |   +---+        |
+   *                +------------------------+-->|fin| - - - -
+   *    |         +---+                      |   +---+
+   *     - - - - -|fin|<---------------------+
+   *              +---+
    *
    * @param {Number} code Status code explaining why the connection is closing
    * @param {String} data A string explaining why the connection is closing
@@ -247,11 +194,8 @@ class WebSocket extends EventEmitter {
   close (code, data) {
     if (this.readyState === WebSocket.CLOSED) return;
     if (this.readyState === WebSocket.CONNECTING) {
-      this._req.abort();
-      this.finalize(
-        new Error('WebSocket was closed before the connection was established')
-      );
-      return;
+      const msg = 'WebSocket was closed before the connection was established';
+      return abortHandshake(this, this._req, msg);
     }
 
     if (this.readyState === WebSocket.CLOSING) {
@@ -269,14 +213,17 @@ class WebSocket extends EventEmitter {
 
       this._closeFrameSent = true;
 
-      if (!this._finalized) {
+      if (this._socket.writable) {
         if (this._closeFrameReceived) this._socket.end();
 
         //
-        // Ensure that the connection is cleaned up even when the closing
-        // handshake fails.
+        // Ensure that the connection is closed even if the closing handshake
+        // fails.
         //
-        this._closeTimer = setTimeout(this._finalize, closeTimeout, true);
+        this._closeTimer = setTimeout(
+          this._socket.destroy.bind(this._socket),
+          closeTimeout
+        );
       }
     });
   }
@@ -397,14 +344,14 @@ class WebSocket extends EventEmitter {
   terminate () {
     if (this.readyState === WebSocket.CLOSED) return;
     if (this.readyState === WebSocket.CONNECTING) {
-      this._req.abort();
-      this.finalize(
-        new Error('WebSocket was closed before the connection was established')
-      );
-      return;
+      const msg = 'WebSocket was closed before the connection was established';
+      return abortHandshake(this, this._req, msg);
     }
 
-    this.finalize(true);
+    if (this._socket) {
+      this.readyState = WebSocket.CLOSING;
+      this._socket.destroy();
+    }
   }
 }
 
@@ -619,30 +566,31 @@ function initAsClient (address, protocols, options) {
 
   if (agent) requestOptions.agent = agent;
 
-  this._req = httpObj.get(requestOptions);
+  var req = this._req = httpObj.get(requestOptions);
 
   if (options.handshakeTimeout) {
-    this._req.setTimeout(options.handshakeTimeout, () => {
-      this._req.abort();
-      this.finalize(new Error('Opening handshake has timed out'));
-    });
+    req.setTimeout(
+      options.handshakeTimeout,
+      abortHandshake.bind(null, this, req, 'Opening handshake has timed out')
+    );
   }
 
-  this._req.on('error', (error) => {
+  req.on('error', (err) => {
     if (this._req.aborted) return;
 
-    this._req = null;
-    this.finalize(error);
+    req = this._req = null;
+    this.readyState = WebSocket.CLOSING;
+    this.emit('error', err);
+    this.emitClose();
   });
 
-  this._req.on('response', (res) => {
-    if (!this.emit('unexpected-response', this._req, res)) {
-      this._req.abort();
-      this.finalize(new Error(`Unexpected server response: ${res.statusCode}`));
-    }
+  req.on('response', (res) => {
+    if (this.emit('unexpected-response', req, res)) return;
+
+    abortHandshake(this, req, `Unexpected server response: ${res.statusCode}`);
   });
 
-  this._req.on('upgrade', (res, socket, head) => {
+  req.on('upgrade', (res, socket, head) => {
     this.emit('upgrade', res);
 
     //
@@ -651,15 +599,15 @@ function initAsClient (address, protocols, options) {
     //
     if (this.readyState !== WebSocket.CONNECTING) return;
 
-    this._req = null;
+    req = this._req = null;
 
     const digest = crypto.createHash('sha1')
       .update(key + constants.GUID, 'binary')
       .digest('base64');
 
     if (res.headers['sec-websocket-accept'] !== digest) {
-      socket.destroy();
-      return this.finalize(new Error('Invalid Sec-WebSocket-Accept header'));
+      abortHandshake(this, socket, 'Invalid Sec-WebSocket-Accept header');
+      return;
     }
 
     const serverProt = res.headers['sec-websocket-protocol'];
@@ -675,8 +623,8 @@ function initAsClient (address, protocols, options) {
     }
 
     if (protError) {
-      socket.destroy();
-      return this.finalize(new Error(protError));
+      abortHandshake(this, socket, protError);
+      return;
     }
 
     if (serverProt) this.protocol = serverProt;
@@ -694,8 +642,7 @@ function initAsClient (address, protocols, options) {
           this._extensions[PerMessageDeflate.extensionName] = perMessageDeflate;
         }
       } catch (err) {
-        socket.destroy();
-        this.finalize(new Error('Invalid Sec-WebSocket-Extensions header'));
+        abortHandshake(this, socket, 'Invalid Sec-WebSocket-Extensions header');
         return;
       }
     }
@@ -703,3 +650,195 @@ function initAsClient (address, protocols, options) {
     this.setSocket(socket, head, 0);
   });
 }
+
+/**
+ * Abort the handshake and emit an error.
+ *
+ * @param {WebSocket} websocket The WebSocket instance
+ * @param {(http.ClientRequest|net.Socket)} stream The request to abort or the
+ *     socket to destroy
+ * @param {String} message The error message
+ * @private
+ */
+function abortHandshake (websocket, stream, message) {
+  websocket.readyState = WebSocket.CLOSING;
+
+  const err = new Error(message);
+  Error.captureStackTrace(err, abortHandshake);
+
+  if (stream.setHeader) {
+    stream.abort();
+    stream.once('abort', websocket.emitClose.bind(websocket));
+    websocket.emit('error', err);
+  } else {
+    stream.destroy(err);
+    stream.once('error', websocket.emit.bind(websocket, 'error'));
+    stream.once('close', websocket.emitClose.bind(websocket));
+  }
+}
+
+/**
+ * The listener of the `Receiver` `'close'` event.
+ *
+ * @param {Number} code The status code
+ * @param {String} reason The reason for closing
+ * @private
+ */
+function receiverOnClose (code, reason) {
+  const websocket = this[kWebSocket];
+
+  websocket._socket.removeListener('data', socketOnData);
+  websocket._socket.resume();
+
+  websocket._closeFrameReceived = true;
+  websocket._closeMessage = reason;
+  websocket._closeCode = code;
+
+  if (code === 1005) websocket.close();
+  else websocket.close(code, reason);
+}
+
+/**
+ * The listener of the `Receiver` `'drain'` event.
+ *
+ * @private
+ */
+function receiverOnDrain () {
+  this[kWebSocket]._socket.resume();
+}
+
+/**
+ * The listener of the `Receiver` `'error'` event.
+ *
+ * @param {(RangeError|Error)} err The emitted error
+ * @private
+ */
+function receiverOnError (err) {
+  const websocket = this[kWebSocket];
+
+  websocket.readyState = WebSocket.CLOSING;
+  websocket._closeCode = err[constants.kStatusCode];
+  websocket.emit('error', err);
+  websocket._socket.destroy();
+}
+
+/**
+ * The listener of the `Receiver` `'finish'` event.
+ *
+ * @private
+ */
+function receiverOnFinish () {
+  this[kWebSocket].emitClose();
+}
+
+/**
+ * The listener of the `Receiver` `'message'` event.
+ *
+ * @param {(String|Buffer|ArrayBuffer|Buffer[])} data The message
+ * @private
+ */
+function receiverOnMessage (data) {
+  this[kWebSocket].emit('message', data);
+}
+
+/**
+ * The listener of the `Receiver` `'ping'` event.
+ *
+ * @param {Buffer} data The data included in the ping frame
+ * @private
+ */
+function receiverOnPing (data) {
+  const websocket = this[kWebSocket];
+
+  websocket.pong(data, !websocket._isServer, constants.NOOP);
+  websocket.emit('ping', data);
+}
+
+/**
+ * The listener of the `Receiver` `'pong'` event.
+ *
+ * @param {Buffer} data The data included in the pong frame
+ * @private
+ */
+function receiverOnPong (data) {
+  this[kWebSocket].emit('pong', data);
+}
+
+/**
+ * The listener of the `net.Socket` `'close'` event.
+ *
+ * @private
+ */
+function socketOnClose () {
+  const websocket = this[kWebSocket];
+
+  this.removeListener('close', socketOnClose);
+  this.removeListener('data', socketOnData);
+  this.removeListener('end', socketOnEnd);
+  this[kWebSocket] = undefined;
+
+  websocket.readyState = WebSocket.CLOSING;
+
+  //
+  // The close frame might not have been received or the `'end'` event emitted,
+  // for example, if the socket was destroyed due to an error. Ensure that the
+  // `receiver` stream is closed after writing any remaining buffered data to
+  // it.
+  //
+  websocket._socket.read();
+  websocket._receiver.end();
+
+  clearTimeout(websocket._closeTimer);
+
+  if (
+    websocket._receiver._writableState.finished ||
+    websocket._receiver._writableState.errorEmitted
+  ) {
+    websocket.emitClose();
+  } else {
+    websocket._receiver.on('error', receiverOnFinish);
+    websocket._receiver.on('finish', receiverOnFinish);
+  }
+}
+
+/**
+ * The listener of the `net.Socket` `'data'` event.
+ *
+ * @param {Buffer} chunk A chunk of data
+ * @private
+ */
+function socketOnData (chunk) {
+  if (!this[kWebSocket]._receiver.write(chunk)) {
+    this.pause();
+  }
+}
+
+/**
+ * The listener of the `net.Socket` `'end'` event.
+ *
+ * @private
+ */
+function socketOnEnd () {
+  const websocket = this[kWebSocket];
+
+  websocket.readyState = WebSocket.CLOSING;
+  websocket._receiver.end();
+  this.end();
+}
+
+/**
+ * The listener of the `net.Socket` `'error'` event.
+ *
+ * @private
+ */
+function socketOnError () {
+  const websocket = this[kWebSocket];
+
+  this.removeListener('error', socketOnError);
+  this.on('error', constants.NOOP);
+
+  if (websocket) {
+    websocket.readyState = WebSocket.CLOSING;
+    this.destroy();
+  }
+}


Test patch:
diff --git a/test/receiver.test.js b/test/receiver.test.js
index 4335a12c..f235719e 100644
--- a/test/receiver.test.js
+++ b/test/receiver.test.js
@@ -5,48 +5,52 @@ const assert = require('assert');
 const crypto = require('crypto');
 
 const PerMessageDeflate = require('../lib/permessage-deflate');
+const constants = require('../lib/constants');
 const Receiver = require('../lib/receiver');
 const Sender = require('../lib/sender');
 
+const kStatusCode = constants.kStatusCode;
 const Buffer = safeBuffer.Buffer;
 
 describe('Receiver', function () {
-  it('can parse unmasked text message', function (done) {
-    const p = new Receiver();
+  it('parses an unmasked text message', function (done) {
+    const receiver = new Receiver();
 
-    p.onmessage = function (data) {
+    receiver.on('message', (data) => {
       assert.strictEqual(data, 'Hello');
       done();
-    };
+    });
 
-    p.add(Buffer.from('810548656c6c6f', 'hex'));
+    receiver.write(Buffer.from('810548656c6c6f', 'hex'));
   });
 
-  it('can parse close message', function (done) {
-    const p = new Receiver();
+  it('parses a close message', function (done) {
+    const receiver = new Receiver();
 
-    p.onclose = function (code, data) {
+    receiver.on('close', (code, data) => {
       assert.strictEqual(code, 1005);
       assert.strictEqual(data, '');
       done();
-    };
+    });
 
-    p.add(Buffer.from('8800', 'hex'));
+    receiver.write(Buffer.from('8800', 'hex'));
   });
 
-  it('can parse masked text message', function (done) {
-    const p = new Receiver();
+  it('parses a masked text message', function (done) {
+    const receiver = new Receiver();
 
-    p.onmessage = function (data) {
+    receiver.on('message', (data) => {
       assert.strictEqual(data, '5:::{"name":"echo"}');
       done();
-    };
+    });
 
-    p.add(Buffer.from('81933483a86801b992524fa1c60959e68a5216e6cb005ba1d5', 'hex'));
+    receiver.write(
+      Buffer.from('81933483a86801b992524fa1c60959e68a5216e6cb005ba1d5', 'hex')
+    );
   });
 
-  it('can parse a masked text message longer than 125 B', function (done) {
-    const p = new Receiver();
+  it('parses a masked text message longer than 125 B', function (done) {
+    const receiver = new Receiver();
     const msg = 'A'.repeat(200);
 
     const list = Sender.frame(Buffer.from(msg), {
@@ -59,17 +63,17 @@ describe('Receiver', function () {
 
     const frame = Buffer.concat(list);
 
-    p.onmessage = function (data) {
+    receiver.on('message', (data) => {
       assert.strictEqual(data, msg);
       done();
-    };
+    });
 
-    p.add(frame.slice(0, 2));
-    setImmediate(() => p.add(frame.slice(2)));
+    receiver.write(frame.slice(0, 2));
+    setImmediate(() => receiver.write(frame.slice(2)));
   });
 
-  it('can parse a really long masked text message', function (done) {
-    const p = new Receiver();
+  it('parses a really long masked text message', function (done) {
+    const receiver = new Receiver();
     const msg = 'A'.repeat(64 * 1024);
 
     const list = Sender.frame(Buffer.from(msg), {
@@ -82,16 +86,16 @@ describe('Receiver', function () {
 
     const frame = Buffer.concat(list);
 
-    p.onmessage = function (data) {
+    receiver.on('message', (data) => {
       assert.strictEqual(data, msg);
       done();
-    };
+    });
 
-    p.add(frame);
+    receiver.write(frame);
   });
 
-  it('can parse a fragmented masked text message of 300 B', function (done) {
-    const p = new Receiver();
+  it('parses a 300 B fragmented masked text message', function (done) {
+    const receiver = new Receiver();
     const msg = 'A'.repeat(300);
 
     const fragment1 = msg.substr(0, 150);
@@ -108,17 +112,17 @@ describe('Receiver', function () {
       Object.assign({ fin: true, opcode: 0x00 }, options)
     ));
 
-    p.onmessage = function (data) {
+    receiver.on('message', (data) => {
       assert.strictEqual(data, msg);
       done();
-    };
+    });
 
-    p.add(frame1);
-    p.add(frame2);
+    receiver.write(frame1);
+    receiver.write(frame2);
   });
 
-  it('can parse a ping message', function (done) {
-    const p = new Receiver();
+  it('parses a ping message', function (done) {
+    const receiver = new Receiver();
     const msg = 'Hello';
 
     const list = Sender.frame(Buffer.from(msg), {
@@ -131,27 +135,27 @@ describe('Receiver', function () {
 
     const frame = Buffer.concat(list);
 
-    p.onping = function (data) {
+    receiver.on('ping', (data) => {
       assert.strictEqual(data.toString(), msg);
       done();
-    };
+    });
 
-    p.add(frame);
+    receiver.write(frame);
   });
 
-  it('can parse a ping with no data', function (done) {
-    const p = new Receiver();
+  it('parses a ping message with no data', function (done) {
+    const receiver = new Receiver();
 
-    p.onping = function (data) {
+    receiver.on('ping', (data) => {
       assert.ok(data.equals(Buffer.alloc(0)));
       done();
-    };
+    });
 
-    p.add(Buffer.from('8900', 'hex'));
+    receiver.write(Buffer.from('8900', 'hex'));
   });
 
-  it('can parse a fragmented masked text message of 300 B with a ping in the middle (1/2)', function (done) {
-    const p = new Receiver();
+  it('parses a 300 B fragmented masked text message with a ping in the middle (1/2)', function (done) {
+    const receiver = new Receiver();
     const msg = 'A'.repeat(300);
     const pingMessage = 'Hello';
 
@@ -175,23 +179,23 @@ describe('Receiver', function () {
 
     let gotPing = false;
 
-    p.onmessage = function (data) {
+    receiver.on('message', (data) => {
       assert.strictEqual(data, msg);
       assert.ok(gotPing);
       done();
-    };
-    p.onping = function (data) {
+    });
+    receiver.on('ping', (data) => {
       gotPing = true;
       assert.strictEqual(data.toString(), pingMessage);
-    };
+    });
 
-    p.add(frame1);
-    p.add(frame2);
-    p.add(frame3);
+    receiver.write(frame1);
+    receiver.write(frame2);
+    receiver.write(frame3);
   });
 
-  it('can parse a fragmented masked text message of 300 B with a ping in the middle (2/2)', function (done) {
-    const p = new Receiver();
+  it('parses a 300 B fragmented masked text message with a ping in the middle (2/2)', function (done) {
+    const receiver = new Receiver();
     const msg = 'A'.repeat(300);
     const pingMessage = 'Hello';
 
@@ -225,23 +229,23 @@ describe('Receiver', function () {
 
     let gotPing = false;
 
-    p.onmessage = function (data) {
+    receiver.on('message', (data) => {
       assert.strictEqual(data, msg);
       assert.ok(gotPing);
       done();
-    };
-    p.onping = function (data) {
+    });
+    receiver.on('ping', (data) => {
       gotPing = true;
       assert.strictEqual(data.toString(), pingMessage);
-    };
+    });
 
     for (let i = 0; i < chunks.length; ++i) {
-      p.add(chunks[i]);
+      receiver.write(chunks[i]);
     }
   });
 
-  it('can parse a 100 B long masked binary message', function (done) {
-    const p = new Receiver();
+  it('parses a 100 B masked binary message', function (done) {
+    const receiver = new Receiver();
     const msg = crypto.randomBytes(100);
 
     const list = Sender.frame(msg, {
@@ -254,16 +258,16 @@ describe('Receiver', function () {
 
     const frame = Buffer.concat(list);
 
-    p.onmessage = function (data) {
+    receiver.on('message', (data) => {
       assert.ok(data.equals(msg));
       done();
-    };
+    });
 
-    p.add(frame);
+    receiver.write(frame);
   });
 
-  it('can parse a 256 B long masked binary message', function (done) {
-    const p = new Receiver();
+  it('parses a 256 B masked binary message', function (done) {
+    const receiver = new Receiver();
     const msg = crypto.randomBytes(256);
 
     const list = Sender.frame(msg, {
@@ -276,16 +280,16 @@ describe('Receiver', function () {
 
     const frame = Buffer.concat(list);
 
-    p.onmessage = function (data) {
+    receiver.on('message', (data) => {
       assert.ok(data.equals(msg));
       done();
-    };
+    });
 
-    p.add(frame);
+    receiver.write(frame);
   });
 
-  it('can parse a 200 KiB long masked binary message', function (done) {
-    const p = new Receiver();
+  it('parses a 200 KiB masked binary message', function (done) {
+    const receiver = new Receiver();
     const msg = crypto.randomBytes(200 * 1024);
 
     const list = Sender.frame(msg, {
@@ -298,16 +302,16 @@ describe('Receiver', function () {
 
     const frame = Buffer.concat(list);
 
-    p.onmessage = function (data) {
+    receiver.on('message', (data) => {
       assert.ok(data.equals(msg));
       done();
-    };
+    });
 
-    p.add(frame);
+    receiver.write(frame);
   });
 
-  it('can parse a 200 KiB long unmasked binary message', function (done) {
-    const p = new Receiver();
+  it('parses a 200 KiB unmasked binary message', function (done) {
+    const receiver = new Receiver();
     const msg = crypto.randomBytes(200 * 1024);
 
     const list = Sender.frame(msg, {
@@ -320,63 +324,63 @@ describe('Receiver', function () {
 
     const frame = Buffer.concat(list);
 
-    p.onmessage = function (data) {
+    receiver.on('message', (data) => {
       assert.ok(data.equals(msg));
       done();
-    };
+    });
 
-    p.add(frame);
+    receiver.write(frame);
   });
 
-  it('can parse compressed message', function (done) {
+  it('parses a compressed message', function (done) {
     const perMessageDeflate = new PerMessageDeflate();
     perMessageDeflate.accept([{}]);
 
-    const p = new Receiver({ 'permessage-deflate': perMessageDeflate });
+    const receiver = new Receiver({ 'permessage-deflate': perMessageDeflate });
     const buf = Buffer.from('Hello');
 
-    p.onmessage = function (data) {
+    receiver.on('message', (data) => {
       assert.strictEqual(data, 'Hello');
       done();
-    };
+    });
 
-    perMessageDeflate.compress(buf, true, function (err, compressed) {
+    perMessageDeflate.compress(buf, true, (err, data) => {
       if (err) return done(err);
 
-      p.add(Buffer.from([0xc1, compressed.length]));
-      p.add(compressed);
+      receiver.write(Buffer.from([0xc1, data.length]));
+      receiver.write(data);
     });
   });
 
-  it('can parse compressed fragments', function (done) {
+  it('parses a compressed and fragmented message', function (done) {
     const perMessageDeflate = new PerMessageDeflate();
     perMessageDeflate.accept([{}]);
 
-    const p = new Receiver({ 'permessage-deflate': perMessageDeflate });
+    const receiver = new Receiver({ 'permessage-deflate': perMessageDeflate });
     const buf1 = Buffer.from('foo');
     const buf2 = Buffer.from('bar');
 
-    p.onmessage = function (data) {
+    receiver.on('message', (data) => {
       assert.strictEqual(data, 'foobar');
       done();
-    };
+    });
 
-    perMessageDeflate.compress(buf1, false, function (err, compressed1) {
+    perMessageDeflate.compress(buf1, false, function (err, fragment1) {
       if (err) return done(err);
 
-      p.add(Buffer.from([0x41, compressed1.length]));
-      p.add(compressed1);
+      receiver.write(Buffer.from([0x41, fragment1.length]));
+      receiver.write(fragment1);
 
-      perMessageDeflate.compress(buf2, true, function (err, compressed2) {
+      perMessageDeflate.compress(buf2, true, function (err, fragment2) {
         if (err) return done(err);
 
-        p.add(Buffer.from([0x80, compressed2.length]));
-        p.add(compressed2);
+        receiver.write(Buffer.from([0x80, fragment2.length]));
+        receiver.write(fragment2);
       });
     });
   });
 
-  it('can parse a buffer with thousands of frames', function (done) {
+  it('parses a buffer with thousands of frames', function (done) {
     const buf = Buffer.allocUnsafe(40000);
 
     for (let i = 0; i < buf.length; i += 2) {
@@ -384,335 +388,357 @@ describe('Receiver', function () {
       buf[i + 1] = 0x00;
     }
 
-    const p = new Receiver();
+    const receiver = new Receiver();
     let counter = 0;
 
-    p.onmessage = function (data) {
+    receiver.on('message', (data) => {
       assert.strictEqual(data, '');
       if (++counter === 20000) done();
-    };
+    });
 
-    p.add(buf);
+    receiver.write(buf);
   });
 
-  it('resets `totalPayloadLength` only on final frame (unfragmented)', function () {
-    const p = new Receiver({}, 10);
-    let message;
+  it('resets `totalPayloadLength` only on final frame (unfragmented)', function (done) {
+    const receiver = new Receiver({}, 10);
 
-    p.onmessage = function (msg) {
-      message = msg;
-    };
+    receiver.on('message', (data) => {
+      assert.strictEqual(receiver._totalPayloadLength, 0);
+      assert.strictEqual(data, 'Hello');
+      done();
+    });
 
-    assert.strictEqual(p._totalPayloadLength, 0);
-    p.add(Buffer.from('810548656c6c6f', 'hex'));
-    assert.strictEqual(p._totalPayloadLength, 0);
-    assert.strictEqual(message, 'Hello');
+    assert.strictEqual(receiver._totalPayloadLength, 0);
+    receiver.write(Buffer.from('810548656c6c6f', 'hex'));
   });
 
-  it('resets `totalPayloadLength` only on final frame (fragmented)', function () {
-    const p = new Receiver({}, 10);
-    let message;
+  it('resets `totalPayloadLength` only on final frame (fragmented)', function (done) {
+    const receiver = new Receiver({}, 10);
 
-    p.onmessage = function (msg) {
-      message = msg;
-    };
+    receiver.on('message', (data) => {
+      assert.strictEqual(receiver._totalPayloadLength, 0);
+      assert.strictEqual(data, 'Hello');
+      done();
+    });
 
-    assert.strictEqual(p._totalPayloadLength, 0);
-    p.add(Buffer.from('01024865', 'hex'));
-    assert.strictEqual(p._totalPayloadLength, 2);
-    p.add(Buffer.from('80036c6c6f', 'hex'));
-    assert.strictEqual(p._totalPayloadLength, 0);
-    assert.strictEqual(message, 'Hello');
+    assert.strictEqual(receiver._totalPayloadLength, 0);
+    receiver.write(Buffer.from('01024865', 'hex'));
+    assert.strictEqual(receiver._totalPayloadLength, 2);
+    receiver.write(Buffer.from('80036c6c6f', 'hex'));
   });
 
-  it('resets `totalPayloadLength` only on final frame (fragmented + ping)', function () {
-    const p = new Receiver({}, 10);
-    const data = [];
+  it('resets `totalPayloadLength` only on final frame (fragmented + ping)', function (done) {
+    const receiver = new Receiver({}, 10);
+    let data;
 
-    p.onmessage = p.onping = function (buf) {
-      data.push(buf.toString());
-    };
+    receiver.on('ping', (buf) => {
+      assert.strictEqual(receiver._totalPayloadLength, 2);
+      data = buf.toString();
+    });
+    receiver.on('message', (buf) => {
+      assert.strictEqual(receiver._totalPayloadLength, 0);
+      assert.strictEqual(data, '');
+      assert.strictEqual(buf.toString(), 'Hello');
+      done();
+    });
+
+    assert.strictEqual(receiver._totalPayloadLength, 0);
+    receiver.write(Buffer.from('02024865', 'hex'));
+    receiver.write(Buffer.from('8900', 'hex'));
+    receiver.write(Buffer.from('80036c6c6f', 'hex'));
+  });
+
+  it('ignores any data after a close frame', function (done) {
+    const perMessageDeflate = new PerMessageDeflate();
+    perMessageDeflate.accept([{}]);
+
+    const receiver = new Receiver({ 'permessage-deflate': perMessageDeflate });
+    const results = [];
+    const push = results.push.bind(results);
 
-    assert.strictEqual(p._totalPayloadLength, 0);
-    p.add(Buffer.from('02024865', 'hex'));
-    assert.strictEqual(p._totalPayloadLength, 2);
-    p.add(Buffer.from('8900', 'hex'));
-    assert.strictEqual(p._totalPayloadLength, 2);
-    p.add(Buffer.from('80036c6c6f', 'hex'));
-    assert.strictEqual(p._totalPayloadLength, 0);
-    assert.deepStrictEqual(data, ['', 'Hello']);
+    receiver.on('close', push).on('message', push);
+    receiver.on('finish', () => {
+      assert.deepStrictEqual(results, ['', 1005, '']);
+      done();
+    });
+
+    receiver.write(Buffer.from([0xc1, 0x01, 0x00]));
+    receiver.write(Buffer.from([0x88, 0x00]));
+    receiver.write(Buffer.from([0x81, 0x00]));
   });
 
-  it('raises an error when RSV1 is on and permessage-deflate is disabled', function (done) {
-    const p = new Receiver();
+  it('emits an error if RSV1 is on and permessage-deflate is disabled', function (done) {
+    const receiver = new Receiver();
 
-    p.onerror = function (err, code) {
+    receiver.on('error', (err) => {
       assert.ok(err instanceof RangeError);
       assert.strictEqual(
         err.message,
         'Invalid WebSocket frame: RSV1 must be clear'
       );
-      assert.strictEqual(code, 1002);
+      assert.strictEqual(err[kStatusCode], 1002);
       done();
-    };
+    });
 
-    p.add(Buffer.from([0xc2, 0x80, 0x00, 0x00, 0x00, 0x00]));
+    receiver.write(Buffer.from([0xc2, 0x80, 0x00, 0x00, 0x00, 0x00]));
   });
 
-  it('raises an error when RSV1 is on and opcode is 0', function (done) {
+  it('emits an error if RSV1 is on and opcode is 0', function (done) {
     const perMessageDeflate = new PerMessageDeflate();
     perMessageDeflate.accept([{}]);
 
-    const p = new Receiver({ 'permessage-deflate': perMessageDeflate });
+    const receiver = new Receiver({ 'permessage-deflate': perMessageDeflate });
 
-    p.onerror = function (err, code) {
+    receiver.on('error', (err) => {
       assert.ok(err instanceof RangeError);
       assert.strictEqual(
         err.message,
         'Invalid WebSocket frame: RSV1 must be clear'
       );
-      assert.strictEqual(code, 1002);
+      assert.strictEqual(err[kStatusCode], 1002);
       done();
-    };
+    });
 
-    p.add(Buffer.from([0x40, 0x00]));
+    receiver.write(Buffer.from([0x40, 0x00]));
   });
 
-  it('raises an error when RSV2 is on', function (done) {
-    const p = new Receiver();
+  it('emits an error if RSV2 is on', function (done) {
+    const receiver = new Receiver();
 
-    p.onerror = function (err, code) {
+    receiver.on('error', (err) => {
       assert.ok(err instanceof RangeError);
       assert.strictEqual(
         err.message,
         'Invalid WebSocket frame: RSV2 and RSV3 must be clear'
       );
-      assert.strictEqual(code, 1002);
+      assert.strictEqual(err[kStatusCode], 1002);
       done();
-    };
+    });
 
-    p.add(Buffer.from([0xa2, 0x00]));
+    receiver.write(Buffer.from([0xa2, 0x00]));
   });
 
-  it('raises an error when RSV3 is on', function (done) {
-    const p = new Receiver();
+  it('emits an error if RSV3 is on', function (done) {
+    const receiver = new Receiver();
 
-    p.onerror = function (err, code) {
+    receiver.on('error', (err) => {
       assert.ok(err instanceof RangeError);
       assert.strictEqual(
         err.message,
         'Invalid WebSocket frame: RSV2 and RSV3 must be clear'
       );
-      assert.strictEqual(code, 1002);
+      assert.strictEqual(err[kStatusCode], 1002);
       done();
-    };
+    });
 
-    p.add(Buffer.from([0x92, 0x00]));
+    receiver.write(Buffer.from([0x92, 0x00]));
   });
 
-  it('raises an error if the first frame in a fragmented message has opcode 0', function (done) {
-    const p = new Receiver();
+  it('emits an error if the first frame in a fragmented message has opcode 0', function (done) {
+    const receiver = new Receiver();
 
-    p.onerror = function (err, code) {
+    receiver.on('error', (err) => {
       assert.ok(err instanceof RangeError);
       assert.strictEqual(
         err.message,
         'Invalid WebSocket frame: invalid opcode 0'
       );
-      assert.strictEqual(code, 1002);
+      assert.strictEqual(err[kStatusCode], 1002);
       done();
-    };
+    });
 
-    p.add(Buffer.from([0x00, 0x00]));
+    receiver.write(Buffer.from([0x00, 0x00]));
   });
 
-  it('raises an error if a frame has opcode 1 in the middle of a fragmented message', function (done) {
-    const p = new Receiver();
+  it('emits an error if a frame has opcode 1 in the middle of a fragmented message', function (done) {
+    const receiver = new Receiver();
 
-    p.onerror = function (err, code) {
+    receiver.on('error', (err) => {
       assert.ok(err instanceof RangeError);
       assert.strictEqual(
         err.message,
         'Invalid WebSocket frame: invalid opcode 1'
       );
-      assert.strictEqual(code, 1002);
+      assert.strictEqual(err[kStatusCode], 1002);
       done();
-    };
+    });
 
-    p.add(Buffer.from([0x01, 0x00]));
-    p.add(Buffer.from([0x01, 0x00]));
+    receiver.write(Buffer.from([0x01, 0x00]));
+    receiver.write(Buffer.from([0x01, 0x00]));
   });
 
-  it('raises an error if a frame has opcode 2 in the middle of a fragmented message', function (done) {
-    const p = new Receiver();
+  it('emits an error if a frame has opcode 2 in the middle of a fragmented message', function (done) {
+    const receiver = new Receiver();
 
-    p.onerror = function (err, code) {
+    receiver.on('error', (err) => {
       assert.ok(err instanceof RangeError);
       assert.strictEqual(
         err.message,
         'Invalid WebSocket frame: invalid opcode 2'
       );
-      assert.strictEqual(code, 1002);
+      assert.strictEqual(err[kStatusCode], 1002);
       done();
-    };
+    });
 
-    p.add(Buffer.from([0x01, 0x00]));
-    p.add(Buffer.from([0x02, 0x00]));
+    receiver.write(Buffer.from([0x01, 0x00]));
+    receiver.write(Buffer.from([0x02, 0x00]));
   });
 
-  it('raises an error when a control frame has the FIN bit off', function (done) {
-    const p = new Receiver();
+  it('emits an error if a control frame has the FIN bit off', function (done) {
+    const receiver = new Receiver();
 
-    p.onerror = function (err, code) {
+    receiver.on('error', (err) => {
       assert.ok(err instanceof RangeError);
       assert.strictEqual(
         err.message,
         'Invalid WebSocket frame: FIN must be set'
       );
-      assert.strictEqual(code, 1002);
+      assert.strictEqual(err[kStatusCode], 1002);
       done();
-    };
+    });
 
-    p.add(Buffer.from([0x09, 0x00]));
+    receiver.write(Buffer.from([0x09, 0x00]));
   });
 
-  it('raises an error when a control frame has the RSV1 bit on', function (done) {
+  it('emits an error if a control frame has the RSV1 bit on', function (done) {
     const perMessageDeflate = new PerMessageDeflate();
     perMessageDeflate.accept([{}]);
 
-    const p = new Receiver({ 'permessage-deflate': perMessageDeflate });
+    const receiver = new Receiver({ 'permessage-deflate': perMessageDeflate });
 
-    p.onerror = function (err, code) {
+    receiver.on('error', (err) => {
       assert.ok(err instanceof RangeError);
       assert.strictEqual(
         err.message,
         'Invalid WebSocket frame: RSV1 must be clear'
       );
-      assert.strictEqual(code, 1002);
+      assert.strictEqual(err[kStatusCode], 1002);
       done();
-    };
+    });
 
-    p.add(Buffer.from([0xc9, 0x00]));
+    receiver.write(Buffer.from([0xc9, 0x00]));
   });
 
-  it('raises an error when a control frame has the FIN bit off', function (done) {
-    const p = new Receiver();
+  it('emits an error if a control frame has the FIN bit off', function (done) {
+    const receiver = new Receiver();
 
-    p.onerror = function (err, code) {
+    receiver.on('error', (err) => {
       assert.ok(err instanceof RangeError);
       assert.strictEqual(
         err.message,
         'Invalid WebSocket frame: FIN must be set'
       );
-      assert.strictEqual(code, 1002);
+      assert.strictEqual(err[kStatusCode], 1002);
       done();
-    };
+    });
 
-    p.add(Buffer.from([0x09, 0x00]));
+    receiver.write(Buffer.from([0x09, 0x00]));
   });
 
-  it('raises an error when a control frame has a payload bigger than 125 B', function (done) {
-    const p = new Receiver();
+  it('emits an error if a control frame has a payload bigger than 125 B', function (done) {
+    const receiver = new Receiver();
 
-    p.onerror = function (err, code) {
+    receiver.on('error', (err) => {
       assert.ok(err instanceof RangeError);
       assert.strictEqual(
         err.message,
         'Invalid WebSocket frame: invalid payload length 126'
       );
-      assert.strictEqual(code, 1002);
+      assert.strictEqual(err[kStatusCode], 1002);
       done();
-    };
+    });
 
-    p.add(Buffer.from([0x89, 0x7e]));
+    receiver.write(Buffer.from([0x89, 0x7e]));
   });
 
-  it('raises an error when a data frame has a payload bigger than 2^53 - 1 B', function (done) {
-    const p = new Receiver();
+  it('emits an error if a data frame has a payload bigger than 2^53 - 1 B', function (done) {
+    const receiver = new Receiver();
 
-    p.onerror = function (err, code) {
+    receiver.on('error', (err) => {
       assert.ok(err instanceof RangeError);
       assert.strictEqual(
         err.message,
         'Unsupported WebSocket frame: payload length > 2^53 - 1'
       );
-      assert.strictEqual(code, 1009);
+      assert.strictEqual(err[kStatusCode], 1009);
       done();
-    };
+    });
 
-    p.add(Buffer.from([0x82, 0x7f]));
-    setImmediate(() => p.add(Buffer.from([
+    receiver.write(Buffer.from([0x82, 0x7f]));
+    setImmediate(() => receiver.write(Buffer.from([
       0x00, 0x20, 0x00, 0x00,
       0x00, 0x00, 0x00, 0x00
     ])));
   });
 
-  it('raises an error if a text frame contains invalid UTF-8 data', function (done) {
-    const p = new Receiver();
+  it('emits an error if a text frame contains invalid UTF-8 data', function (done) {
+    const receiver = new Receiver();
 
-    p.onerror = function (err, code) {
+    receiver.on('error', (err) => {
       assert.ok(err instanceof Error);
       assert.strictEqual(
         err.message,
         'Invalid WebSocket frame: invalid UTF-8 sequence'
       );
-      assert.strictEqual(code, 1007);
+      assert.strictEqual(err[kStatusCode], 1007);
       done();
-    };
+    });
 
-    p.add(Buffer.from([0x81, 0x04, 0xce, 0xba, 0xe1, 0xbd]));
+    receiver.write(Buffer.from([0x81, 0x04, 0xce, 0xba, 0xe1, 0xbd]));
   });
 
-  it('raises an error if a close frame has a payload of 1 B', function (done) {
-    const p = new Receiver();
+  it('emits an error if a close frame has a payload of 1 B', function (done) {
+    const receiver = new Receiver();
 
-    p.onerror = function (err, code) {
+    receiver.on('error', (err) => {
       assert.ok(err instanceof RangeError);
       assert.strictEqual(
         err.message,
         'Invalid WebSocket frame: invalid payload length 1'
       );
-      assert.strictEqual(code, 1002);
+      assert.strictEqual(err[kStatusCode], 1002);
       done();
-    };
+    });
 
-    p.add(Buffer.from([0x88, 0x01, 0x00]));
+    receiver.write(Buffer.from([0x88, 0x01, 0x00]));
   });
 
-  it('raises an error if a close frame contains an invalid close code', function (done) {
-    const p = new Receiver();
+  it('emits an error if a close frame contains an invalid close code', function (done) {
+    const receiver = new Receiver();
 
-    p.onerror = function (err, code) {
+    receiver.on('error', (err) => {
       assert.ok(err instanceof RangeError);
       assert.strictEqual(
         err.message,
         'Invalid WebSocket frame: invalid status code 0'
       );
-      assert.strictEqual(code, 1002);
+      assert.strictEqual(err[kStatusCode], 1002);
       done();
-    };
+    });
 
-    p.add(Buffer.from([0x88, 0x02, 0x00, 0x00]));
+    receiver.write(Buffer.from([0x88, 0x02, 0x00, 0x00]));
   });
 
-  it('raises an error if a close frame contains invalid UTF-8 data', function (done) {
-    const p = new Receiver();
+  it('emits an error if a close frame contains invalid UTF-8 data', function (done) {
+    const receiver = new Receiver();
 
-    p.onerror = function (err, code) {
+    receiver.on('error', (err) => {
       assert.ok(err instanceof Error);
       assert.strictEqual(
         err.message,
         'Invalid WebSocket frame: invalid UTF-8 sequence'
       );
-      assert.strictEqual(code, 1007);
+      assert.strictEqual(err[kStatusCode], 1007);
       done();
-    };
+    });
 
-    p.add(Buffer.from([0x88, 0x06, 0x03, 0xef, 0xce, 0xba, 0xe1, 0xbd]));
+    receiver.write(
+      Buffer.from([0x88, 0x06, 0x03, 0xef, 0xce, 0xba, 0xe1, 0xbd])
+    );
   });
 
-  it('raises an error on a 200 KiB long masked binary message when `maxPayload` is 20 KiB', function (done) {
-    const p = new Receiver({}, 20 * 1024);
+  it('emits an error if a frame payload length is bigger than `maxPayload`', function (done) {
+    const receiver = new Receiver({}, 20 * 1024);
     const msg = crypto.randomBytes(200 * 1024);
 
     const list = Sender.frame(msg, {
@@ -725,221 +751,73 @@ describe('Receiver', function () {
 
     const frame = Buffer.concat(list);
 
-    p.onerror = function (err, code) {
+    receiver.on('error', (err) => {
       assert.ok(err instanceof RangeError);
       assert.strictEqual(err.message, 'Max payload size exceeded');
-      assert.strictEqual(code, 1009);
+      assert.strictEqual(err[kStatusCode], 1009);
       done();
-    };
-
-    p.add(frame);
-  });
-
-  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);
-
-    const list = Sender.frame(msg, {
-      fin: true,
-      rsv1: false,
-      opcode: 0x02,
-      mask: false,
-      readOnly: true
     });
 
-    const frame = Buffer.concat(list);
-
-    p.onerror = function (err, code) {
-      assert.ok(err instanceof RangeError);
-      assert.strictEqual(err.message, 'Max payload size exceeded');
-      assert.strictEqual(code, 1009);
-      done();
-    };
-
-    p.add(frame);
+    receiver.write(frame);
   });
 
-  it('raises an error on a compressed message that exceeds `maxPayload`', function (done) {
+  it('emits an error if the message length exceeds `maxPayload`', function (done) {
     const perMessageDeflate = new PerMessageDeflate({}, false, 25);
     perMessageDeflate.accept([{}]);
 
-    const p = new Receiver({ 'permessage-deflate': perMessageDeflate }, 25);
+    const receiver = new Receiver({
+      'permessage-deflate': perMessageDeflate
+    }, 25);
     const buf = Buffer.from('A'.repeat(50));
 
-    p.onerror = function (err, code) {
+    receiver.on('error', (err) => {
       assert.ok(err instanceof RangeError);
       assert.strictEqual(err.message, 'Max payload size exceeded');
-      assert.strictEqual(code, 1009);
+      assert.strictEqual(err[kStatusCode], 1009);
       done();
-    };
+    });
 
     perMessageDeflate.compress(buf, true, function (err, data) {
       if (err) return done(err);
 
-      p.add(Buffer.from([0xc1, data.length]));
-      p.add(data);
+      receiver.write(Buffer.from([0xc1, data.length]));
+      receiver.write(data);
     });
   });
 
-  it('raises an error if the sum of fragment lengths exceeds `maxPayload`', function (done) {
+  it('emits an error if the sum of fragment lengths exceeds `maxPayload`', function (done) {
     const perMessageDeflate = new PerMessageDeflate({}, false, 25);
     perMessageDeflate.accept([{}]);
 
-    const p = new Receiver({ 'permessage-deflate': perMessageDeflate }, 25);
+    const receiver = new Receiver({
+      'permessage-deflate': perMessageDeflate
+    }, 25);
     const buf = Buffer.from('A'.repeat(15));
 
-    p.onerror = function (err, code) {
+    receiver.on('error', (err) => {
       assert.ok(err instanceof RangeError);
       assert.strictEqual(err.message, 'Max payload size exceeded');
-      assert.strictEqual(code, 1009);
+      assert.strictEqual(err[kStatusCode], 1009);
       done();
-    };
+    });
 
     perMessageDeflate.compress(buf, false, function (err, fragment1) {
       if (err) return done(err);
 
-      p.add(Buffer.from([0x41, fragment1.length]));
-      p.add(fragment1);
+      receiver.write(Buffer.from([0x41, fragment1.length]));
+      receiver.write(fragment1);
 
       perMessageDeflate.compress(buf, true, function (err, fragment2) {
         if (err) return done(err);
 
-        p.add(Buffer.from([0x80, fragment2.length]));
-        p.add(fragment2);
-      });
-    });
-  });
-
-  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);
-
-      const frame = Buffer.concat([Buffer.from([0xc1, data.length]), data]);
-
-      p.add(frame);
-      p.add(frame);
-
-      assert.strictEqual(p._state, 5);
-      assert.strictEqual(p._bufferedBytes, frame.length);
-
-      p.cleanup(() => {
-        assert.deepStrictEqual(results, ['Hello', 'Hello']);
-        assert.strictEqual(p.onmessage, null);
-        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 = [];
-
-    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 textFrame = Buffer.concat([Buffer.from([0xc1, data.length]), data]);
-      const closeFrame = Buffer.from([0x88, 0x00]);
-
-      p.add(textFrame);
-      p.add(textFrame);
-      p.add(closeFrame);
-
-      assert.strictEqual(p._state, 5);
-      assert.strictEqual(p._bufferedBytes, textFrame.length + closeFrame.length);
-
-      p.cleanup(() => {
-        assert.deepStrictEqual(results, ['Hello', 'Hello', 1005, '']);
-        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 = [];
-
-    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',
-          'Invalid WebSocket frame: RSV2 and RSV3 must be clear',
-          1002
-        ]);
-        assert.strictEqual(p.onmessage, null);
-        done();
+        receiver.write(Buffer.from([0x80, fragment2.length]));
+        receiver.write(fragment2);
       });
     });
   });
 
-  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();
-      });
-    });
-  });
-
-  it('can emit nodebuffer of fragmented binary message', function (done) {
-    const p = new Receiver();
+  it("honors the 'nodebuffer' binary type", function (done) {
+    const receiver = new Receiver();
     const frags = [
       crypto.randomBytes(7321),
       crypto.randomBytes(137),
@@ -947,12 +825,11 @@ describe('Receiver', function () {
       crypto.randomBytes(3)
     ];
 
-    p.binaryType = 'nodebuffer';
-    p.onmessage = (data) => {
+    receiver.on('message', (data) => {
       assert.ok(Buffer.isBuffer(data));
       assert.ok(data.equals(Buffer.concat(frags)));
       done();
-    };
+    });
 
     frags.forEach((frag, i) => {
       Sender.frame(frag, {
@@ -961,24 +838,24 @@ describe('Receiver', function () {
         readOnly: true,
         mask: false,
         rsv1: false
-      }).forEach((buf) => p.add(buf));
+      }).forEach((buf) => receiver.write(buf));
     });
   });
 
-  it('can emit arraybuffer of fragmented binary message', function (done) {
-    const p = new Receiver();
+  it("honors the 'arraybuffer' binary type", function (done) {
+    const receiver = new Receiver();
     const frags = [
       crypto.randomBytes(19221),
       crypto.randomBytes(954),
       crypto.randomBytes(623987)
     ];
 
-    p._binaryType = 'arraybuffer';
-    p.onmessage = (data) => {
+    receiver._binaryType = 'arraybuffer';
+    receiver.on('message', (data) => {
       assert.ok(data instanceof ArrayBuffer);
       assert.ok(Buffer.from(data).equals(Buffer.concat(frags)));
       done();
-    };
+    });
 
     frags.forEach((frag, i) => {
       Sender.frame(frag, {
@@ -987,12 +864,12 @@ describe('Receiver', function () {
         readOnly: true,
         mask: false,
         rsv1: false
-      }).forEach((buf) => p.add(buf));
+      }).forEach((buf) => receiver.write(buf));
     });
   });
 
-  it('can emit fragments of fragmented binary message', function (done) {
-    const p = new Receiver();
+  it("honors the 'fragments' binary type", function (done) {
+    const receiver = new Receiver();
     const frags = [
       crypto.randomBytes(17),
       crypto.randomBytes(419872),
@@ -1001,11 +878,11 @@ describe('Receiver', function () {
       crypto.randomBytes(1)
     ];
 
-    p._binaryType = 'fragments';
-    p.onmessage = (data) => {
+    receiver._binaryType = 'fragments';
+    receiver.on('message', (data) => {
       assert.deepStrictEqual(data, frags);
       done();
-    };
+    });
 
     frags.forEach((frag, i) => {
       Sender.frame(frag, {
@@ -1014,7 +891,7 @@ describe('Receiver', function () {
         readOnly: true,
         mask: false,
         rsv1: false
-      }).forEach((buf) => p.add(buf));
+      }).forEach((buf) => receiver.write(buf));
     });
   });
 });
diff --git a/test/websocket.test.js b/test/websocket.test.js
index 3801fdd9..66e59acd 100644
--- a/test/websocket.test.js
+++ b/test/websocket.test.js
@@ -8,7 +8,6 @@ const crypto = require('crypto');
 const https = require('https');
 const http = require('http');
 const dns = require('dns');
-const net = require('net');
 const fs = require('fs');
 const os = require('os');
 
@@ -393,6 +392,28 @@ describe('WebSocket', function () {
       });
     });
 
+    it('does not re-emit `net.Socket` errors', function (done) {
+      const wss = new WebSocket.Server({ port: 0 }, () => {
+        const ws = new WebSocket(`ws://localhost:${wss.address().port}`);
+
+        ws.on('open', () => {
+          ws._socket.on('error', (err) => {
+            assert.ok(err instanceof Error);
+            assert.ok(err.message.startsWith('write E'));
+            ws.on('close', (code, message) => {
+              assert.strictEqual(message, '');
+              assert.strictEqual(code, 1006);
+              wss.close(done);
+            });
+          });
+
+          for (const client of wss.clients) client.terminate();
+          ws.send('foo');
+          ws.send('bar');
+        });
+      });
+    });
+
     it("emits an 'upgrade' event", function (done) {
       const wss = new WebSocket.Server({ port: 0 }, () => {
         const ws = new WebSocket(`ws://localhost:${wss.address().port}`);
@@ -1160,36 +1181,6 @@ describe('WebSocket', function () {
       });
     });
 
-    it('emits an error if the close frame can not be sent', function (done) {
-      const wss = new WebSocket.Server({ port: 0 }, () => {
-        const socket = net.createConnection(wss.address().port, () => {
-          socket.write(
-            'GET / HTTP/1.1\r\n' +
-            'Host: localhost\r\n' +
-            'Upgrade: websocket\r\n' +
-            'Connection: Upgrade\r\n' +
-            'Sec-WebSocket-Key: qqFVFwaCnSMXiqfezY/AZQ==\r\n' +
-            'Sec-WebSocket-Version: 13\r\n' +
-            '\r\n'
-          );
-          socket.destroy();
-        });
-
-        wss.on('connection', (ws) => {
-          ws.on('error', (err) => {
-            assert.ok(err instanceof Error);
-            assert.ok(err.message.startsWith('write E'));
-            ws.on('close', (code, message) => {
-              assert.strictEqual(message, '');
-              assert.strictEqual(code, 1006);
-              wss.close(done);
-            });
-          });
-          ws.close();
-        });
-      });
-    });
-
     it('sends the close status code only when necessary', function (done) {
       let sent;
       const wss = new WebSocket.Server({ port: 0 }, () => {