Repository: websockets/ws
Pull request #997 (2017-02-11 09:52:25)

Write to the socket as soon as possible
When permessage-deflate is enabled the outbound throughput performance is throttled by `process.nextTick()`. If the are `n` messages in the queue, it will take `n` ticks before all messages are dispatched even when compression is skipped.

This patch prevents `process.nextTick()` from being used by slightly changing the handling logic of the queued messages.


Patch:
diff --git a/examples/express-session-parse/index.js b/examples/express-session-parse/index.js
new file mode 100644
index 00000000..bac2d8e9
--- /dev/null
+++ b/examples/express-session-parse/index.js
@@ -0,0 +1,80 @@
+'use strict';
+
+const session = require('express-session');
+const express = require('express');
+const http = require('http');
+const uuid = require('uuid');
+
+const WebSocket = require('../..');
+
+const app = express();
+
+//
+// We need the same instance of the session parser in express and
+// WebSocket server.
+//
+const sessionParser = session({
+  saveUninitialized: false,
+  secret: '$eCuRiTy',
+  resave: false
+});
+
+//
+// Serve static files from the 'public' folder.
+//
+app.use(express.static('public'));
+app.use(sessionParser);
+
+app.post('/login', (req, res) => {
+  //
+  // "Log in" user and set userId to session.
+  //
+  const id = uuid.v4();
+
+  console.log(`Updating session for user ${id}`);
+  req.session.userId = id;
+  res.send({ result: 'OK', message: 'Session updated' });
+});
+
+app.delete('/logout', (request, response) => {
+  console.log('Destroying session');
+  request.session.destroy();
+  response.send({ result: 'OK', message: 'Session destroyed' });
+});
+
+//
+// Create HTTP server by ourselves.
+//
+const server = http.createServer(app);
+
+const wss = new WebSocket.Server({
+  verifyClient: (info, done) => {
+    console.log('Parsing session from request...');
+    sessionParser(info.req, {}, () => {
+      console.log('Session is parsed!');
+
+      //
+      // We can reject the connection by returning false to done(). For example,
+      // reject here if user is unknown.
+      //
+      done(info.req.session.userId);
+    });
+  },
+  server
+});
+
+wss.on('connection', (ws) => {
+  ws.on('message', (message) => {
+    const session = ws.upgradeReq.session;
+
+    //
+    // Here we can now use session parameters.
+    //
+    console.log(`WS message ${message} from user ${session.userId}`);
+  });
+});
+
+//
+// Start the server.
+//
+server.listen(8080, () => console.log('Listening on http://localhost:8080'));
diff --git a/examples/express-session-parse/package.json b/examples/express-session-parse/package.json
new file mode 100644
index 00000000..cf96cbc4
--- /dev/null
+++ b/examples/express-session-parse/package.json
@@ -0,0 +1,11 @@
+{
+  "author": "",
+  "name": "express-session-parse",
+  "version": "0.0.0",
+  "repository": "websockets/ws",
+  "dependencies": {
+    "express": "~4.14.1",
+    "express-session": "~1.15.1",
+    "uuid": "~3.0.1"
+  }
+}
diff --git a/examples/express-session-parse/public/app.js b/examples/express-session-parse/public/app.js
new file mode 100644
index 00000000..916fafd4
--- /dev/null
+++ b/examples/express-session-parse/public/app.js
@@ -0,0 +1,46 @@
+/* global fetch, WebSocket, location */
+(() => {
+  const messages = document.querySelector('#messages');
+  const wsButton = document.querySelector('#wsButton');
+  const logout = document.querySelector('#logout');
+  const login = document.querySelector('#login');
+
+  const showMessage = (message) => {
+    messages.textContent += `\n${message}`;
+    messages.scrollTop = messages.scrollHeight;
+  };
+
+  const handleResponse = (response) => {
+    return response.ok
+      ? response.json().then((data) => JSON.stringify(data, null, 2))
+      : Promise.reject(new Error('Unexpected response'));
+  };
+
+  login.onclick = () => {
+    fetch('/login', { method: 'POST', credentials: 'same-origin' })
+      .then(handleResponse)
+      .then(showMessage)
+      .catch((err) => showMessage(err.message));
+  };
+
+  logout.onclick = () => {
+    fetch('/logout', { method: 'DELETE', credentials: 'same-origin' })
+      .then(handleResponse)
+      .then(showMessage)
+      .catch((err) => showMessage(err.message));
+  };
+
+  let ws;
+
+  wsButton.onclick = () => {
+    if (ws) {
+      ws.onerror = ws.onopen = ws.onclose = null;
+      ws.close();
+    }
+
+    ws = new WebSocket(`ws://${location.host}`);
+    ws.onerror = () => showMessage('WebSocket error');
+    ws.onopen = () => showMessage('WebSocket connection established');
+    ws.onclose = () => showMessage('WebSocket connection closed');
+  };
+})();
diff --git a/examples/express-session-parse/public/index.html b/examples/express-session-parse/public/index.html
new file mode 100644
index 00000000..c99949c7
--- /dev/null
+++ b/examples/express-session-parse/public/index.html
@@ -0,0 +1,21 @@
+<!DOCTYPE html>
+<html lang="en">
+  <head>
+      <meta charset="utf-8">
+      <title>Express session demo</title>
+  </head>
+  <body>
+    <h1>Choose an action.</h1>
+    <button id="login" type="button" title="Simulate login">
+      Simulate login
+    </button>
+    <button id="logout" type="button" title="Simulate logout">
+      Simulate logout
+    </button>
+    <button id="wsButton" type="button" title="Open WebSocket connection">
+      Open WebSocket connection
+    </button>
+    <pre id="messages" style="height: 400px; overflow: scroll"></pre>
+    <script src="app.js"></script>
+  </body>
+</html>
diff --git a/lib/PerMessageDeflate.js b/lib/PerMessageDeflate.js
index d1569546..92a26525 100644
--- a/lib/PerMessageDeflate.js
+++ b/lib/PerMessageDeflate.js
@@ -288,7 +288,8 @@ class PerMessageDeflate {
 
   compress (data, fin, callback) {
     if (!data || data.length === 0) {
-      return callback(null, EMPTY_BLOCK);
+      process.nextTick(callback, null, EMPTY_BLOCK);
+      return;
     }
 
     var endpoint = this._isServer ? 'server' : 'client';
diff --git a/lib/Receiver.js b/lib/Receiver.js
index 09e9973b..71da72c7 100644
--- a/lib/Receiver.js
+++ b/lib/Receiver.js
@@ -16,11 +16,9 @@ const EMPTY_BUFFER = Buffer.alloc(0);
 const GET_INFO = 0;
 const GET_PAYLOAD_LENGTH_16 = 1;
 const GET_PAYLOAD_LENGTH_64 = 2;
-const HAVE_LENGTH = 3;
-const GET_MASK = 4;
-const GET_DATA = 5;
-const HANDLE_DATA = 6;
-const INFLATING = 7;
+const GET_MASK = 3;
+const GET_DATA = 4;
+const INFLATING = 5;
 
 /**
  * HyBi Receiver implementation.
@@ -54,6 +52,7 @@ class Receiver {
     this.cleanupCallback = null;
     this.hadError = false;
     this.dead = false;
+    this.loop = false;
 
     this.onmessage = null;
     this.onclose = null;
@@ -117,6 +116,7 @@ class Receiver {
   hasBufferedBytes (n) {
     if (this.bufferedBytes >= n) return true;
 
+    this.loop = false;
     if (this.dead) this.cleanup(this.cleanupCallback);
     return false;
   }
@@ -140,21 +140,27 @@ class Receiver {
    * @private
    */
   startLoop () {
-    while (true) {
-      if (this.state === GET_INFO) {
-        if (!this.getInfo()) break;
-      } else if (this.state === GET_PAYLOAD_LENGTH_16) {
-        if (!this.getPayloadLength16()) break;
-      } else if (this.state === GET_PAYLOAD_LENGTH_64) {
-        if (!this.getPayloadLength64()) break;
-      } else if (this.state === HAVE_LENGTH) {
-        if (!this.haveLength()) break;
-      } else if (this.state === GET_MASK) {
-        if (!this.getMask()) break;
-      } else if (this.state === GET_DATA) {
-        if (!this.getData()) break;
-      } else { // `HANDLE_DATA` or `INFLATING`
-        break;
+    this.loop = true;
+
+    while (this.loop) {
+      switch (this.state) {
+        case GET_INFO:
+          this.getInfo();
+          break;
+        case GET_PAYLOAD_LENGTH_16:
+          this.getPayloadLength16();
+          break;
+        case GET_PAYLOAD_LENGTH_64:
+          this.getPayloadLength64();
+          break;
+        case GET_MASK:
+          this.getMask();
+          break;
+        case GET_DATA:
+          this.getData();
+          break;
+        default: // `INFLATING`
+          this.loop = false;
       }
     }
   }
@@ -162,24 +168,23 @@ class Receiver {
   /**
    * Reads the first two bytes of a frame.
    *
-   * @return {Boolean} `true` if the operation is successful, else `false`
    * @private
    */
   getInfo () {
-    if (!this.hasBufferedBytes(2)) return false;
+    if (!this.hasBufferedBytes(2)) return;
 
     const buf = this.readBuffer(2);
 
     if ((buf[0] & 0x30) !== 0x00) {
       this.error(new Error('RSV2 and RSV3 must be clear'), 1002);
-      return false;
+      return;
     }
 
     const compressed = (buf[0] & 0x40) === 0x40;
 
     if (compressed && !this.extensions[PerMessageDeflate.extensionName]) {
       this.error(new Error('RSV1 must be clear'), 1002);
-      return false;
+      return;
     }
 
     this.fin = (buf[0] & 0x80) === 0x80;
@@ -189,40 +194,40 @@ class Receiver {
     if (this.opcode === 0x00) {
       if (compressed) {
         this.error(new Error('RSV1 must be clear'), 1002);
-        return false;
+        return;
       }
 
       if (!this.fragmented) {
         this.error(new Error(`invalid opcode: ${this.opcode}`), 1002);
-        return false;
+        return;
       } else {
         this.opcode = this.fragmented;
       }
     } else if (this.opcode === 0x01 || this.opcode === 0x02) {
       if (this.fragmented) {
         this.error(new Error(`invalid opcode: ${this.opcode}`), 1002);
-        return false;
+        return;
       }
 
       this.compressed = compressed;
     } else if (this.opcode > 0x07 && this.opcode < 0x0b) {
       if (!this.fin) {
         this.error(new Error('FIN must be set'), 1002);
-        return false;
+        return;
       }
 
       if (compressed) {
         this.error(new Error('RSV1 must be clear'), 1002);
-        return false;
+        return;
       }
 
       if (this.payloadLength > 0x7d) {
         this.error(new Error('invalid payload length'), 1002);
-        return false;
+        return;
       }
     } else {
       this.error(new Error(`invalid opcode: ${this.opcode}`), 1002);
-      return false;
+      return;
     }
 
     if (!this.fin && !this.fragmented) this.fragmented = this.opcode;
@@ -231,33 +236,28 @@ class Receiver {
 
     if (this.payloadLength === 126) this.state = GET_PAYLOAD_LENGTH_16;
     else if (this.payloadLength === 127) this.state = GET_PAYLOAD_LENGTH_64;
-    else this.state = HAVE_LENGTH;
-
-    return true;
+    else this.haveLength();
   }
 
   /**
    * Gets extended payload length (7+16).
    *
-   * @return {Boolean} `true` if payload length has been read, else `false`
    * @private
    */
   getPayloadLength16 () {
-    if (!this.hasBufferedBytes(2)) return false;
+    if (!this.hasBufferedBytes(2)) return;
 
     this.payloadLength = this.readBuffer(2).readUInt16BE(0, true);
-    this.state = HAVE_LENGTH;
-    return true;
+    this.haveLength();
   }
 
   /**
    * Gets extended payload length (7+64).
    *
-   * @return {Boolean} `true` if payload length has been read, else `false`
    * @private
    */
   getPayloadLength64 () {
-    if (!this.hasBufferedBytes(8)) return false;
+    if (!this.hasBufferedBytes(8)) return;
 
     const buf = this.readBuffer(8);
     const num = buf.readUInt32BE(0, true);
@@ -268,62 +268,54 @@ class Receiver {
     //
     if (num > Math.pow(2, 53 - 32) - 1) {
       this.error(new Error('max payload size exceeded'), 1009);
-      return false;
+      return;
     }
 
     this.payloadLength = num * Math.pow(2, 32) + buf.readUInt32BE(4, true);
-    this.state = HAVE_LENGTH;
-    return true;
+    this.haveLength();
   }
 
   /**
    * Payload length has been read.
    *
-   * @return {Boolean} `false` if payload length exceeds `maxPayload`, else `true`
    * @private
    */
   haveLength () {
     if (this.opcode < 0x08 && this.maxPayloadExceeded(this.payloadLength)) {
-      return false;
+      return;
     }
 
     if (this.masked) this.state = GET_MASK;
     else this.state = GET_DATA;
-    return true;
   }
 
   /**
    * Reads mask bytes.
    *
-   * @return {Boolean} `true` if the mask has been read, else `false`
    * @private
    */
   getMask () {
-    if (!this.hasBufferedBytes(4)) return false;
+    if (!this.hasBufferedBytes(4)) return;
 
     this.mask = this.readBuffer(4);
     this.state = GET_DATA;
-    return true;
   }
 
   /**
    * Reads data bytes.
    *
-   * @return {Boolean} `true` if the data bytes have been read, else `false`
    * @private
    */
   getData () {
     var data = EMPTY_BUFFER;
 
     if (this.payloadLength) {
-      if (!this.hasBufferedBytes(this.payloadLength)) return false;
+      if (!this.hasBufferedBytes(this.payloadLength)) return;
 
       data = this.readBuffer(this.payloadLength);
       if (this.masked) bufferUtil.unmask(data, this.mask);
     }
 
-    this.state = HANDLE_DATA;
-
     if (this.opcode > 0x07) {
       this.controlMessage(data);
     } else if (this.compressed) {
@@ -332,8 +324,6 @@ class Receiver {
     } else if (this.pushFragment(data)) {
       this.dataMessage();
     }
-
-    return true;
   }
 
   /**
@@ -399,6 +389,7 @@ class Receiver {
     if (this.opcode === 0x08) {
       if (data.length === 0) {
         this.onclose(1000, '', { masked: this.masked });
+        this.loop = false;
         this.cleanup(this.cleanupCallback);
       } else if (data.length === 1) {
         this.error(new Error('invalid payload length'), 1002);
@@ -418,6 +409,7 @@ class Receiver {
         }
 
         this.onclose(code, buf.toString(), { masked: this.masked });
+        this.loop = false;
         this.cleanup(this.cleanupCallback);
       }
 
@@ -442,6 +434,7 @@ class Receiver {
   error (err, code) {
     this.onerror(err, code);
     this.hadError = true;
+    this.loop = false;
     this.cleanup(this.cleanupCallback);
   }
 
@@ -497,7 +490,7 @@ class Receiver {
   cleanup (cb) {
     this.dead = true;
 
-    if (!this.hadError && this.state === INFLATING) {
+    if (!this.hadError && (this.loop || this.state === INFLATING)) {
       this.cleanupCallback = cb;
     } else {
       this.extensions = null;
diff --git a/lib/Sender.js b/lib/Sender.js
index b36d6013..ee520cdd 100644
--- a/lib/Sender.js
+++ b/lib/Sender.js
@@ -29,8 +29,8 @@ class Sender {
     this.firstFragment = true;
     this.compress = false;
 
-    this.processing = false;
     this.bufferedBytes = 0;
+    this.deflating = false;
     this.queue = [];
 
     this.onerror = null;
@@ -55,7 +55,7 @@ class Sender {
     buf.writeUInt16BE(code || 1000, 0, true);
     if (buf.length > 2) buf.write(data, 2);
 
-    if (this.perMessageDeflate) {
+    if (this.deflating) {
       this.enqueue([this.doClose, buf, mask, cb]);
     } else {
       this.doClose(buf, mask, cb);
@@ -78,8 +78,6 @@ class Sender {
       fin: true,
       mask
     }, cb);
-
-    if (this.perMessageDeflate) this.continue();
   }
 
   /**
@@ -103,7 +101,7 @@ class Sender {
       }
     }
 
-    if (this.perMessageDeflate) {
+    if (this.deflating) {
       this.enqueue([this.doPing, data, mask, readOnly]);
     } else {
       this.doPing(data, mask, readOnly);
@@ -126,8 +124,6 @@ class Sender {
       readOnly,
       mask
     });
-
-    if (this.perMessageDeflate) this.continue();
   }
 
   /**
@@ -151,7 +147,7 @@ class Sender {
       }
     }
 
-    if (this.perMessageDeflate) {
+    if (this.deflating) {
       this.enqueue([this.doPong, data, mask, readOnly]);
     } else {
       this.doPong(data, mask, readOnly);
@@ -174,8 +170,6 @@ class Sender {
       readOnly,
       mask
     });
-
-    if (this.perMessageDeflate) this.continue();
   }
 
   /**
@@ -220,14 +214,20 @@ class Sender {
     if (options.fin) this.firstFragment = true;
 
     if (this.perMessageDeflate) {
-      this.enqueue([this.dispatch, data, {
+      const opts = {
         compress: this.compress,
         mask: options.mask,
         fin: options.fin,
         readOnly,
         opcode,
         rsv1
-      }, cb]);
+      };
+
+      if (this.deflating) {
+        this.enqueue([this.dispatch, data, opts, cb]);
+      } else {
+        this.dispatch(data, opts, cb);
+      }
     } else {
       this.frameAndSend(data, {
         mask: options.mask,
@@ -256,10 +256,10 @@ class Sender {
   dispatch (data, options, cb) {
     if (!options.compress) {
       this.frameAndSend(data, options, cb);
-      this.continue();
       return;
     }
 
+    this.deflating = true;
     this.perMessageDeflate.compress(data, options.fin, (err, buf) => {
       if (err) {
         if (cb) cb(err);
@@ -269,7 +269,8 @@ class Sender {
 
       options.readOnly = false;
       this.frameAndSend(buf, options, cb);
-      this.continue();
+      this.deflating = false;
+      this.dequeue();
     });
   }
 
@@ -349,32 +350,17 @@ class Sender {
   }
 
   /**
-   * Executes a queued send operation.
+   * Executes queued send operations.
    *
    * @private
    */
   dequeue () {
-    if (this.processing) return;
-
-    const params = this.queue.shift();
-    if (!params) return;
+    while (!this.deflating && this.queue.length) {
+      const params = this.queue.shift();
 
-    if (params[1]) this.bufferedBytes -= params[1].length;
-    this.processing = true;
-
-    params[0].apply(this, params.slice(1));
-  }
-
-  /**
-   * Signals the completion of a send operation.
-   *
-   * @private
-   */
-  continue () {
-    process.nextTick(() => {
-      this.processing = false;
-      this.dequeue();
-    });
+      if (params[1]) this.bufferedBytes -= params[1].length;
+      params[0].apply(this, params.slice(1));
+    }
   }
 
   /**
@@ -386,7 +372,6 @@ class Sender {
   enqueue (params) {
     if (params[1]) this.bufferedBytes += params[1].length;
     this.queue.push(params);
-    this.dequeue();
   }
 }
 


Test patch:
diff --git a/test/Receiver.test.js b/test/Receiver.test.js
index db7c721b..1290388a 100644
--- a/test/Receiver.test.js
+++ b/test/Receiver.test.js
@@ -377,7 +377,7 @@ describe('Receiver', function () {
   it('raises an error when RSV1 is on and permessage-deflate is disabled', function (done) {
     const p = new Receiver();
 
-    p.error = function (err, code) {
+    p.onerror = function (err, code) {
       assert.ok(err instanceof Error);
       assert.strictEqual(err.message, 'RSV1 must be clear');
       assert.strictEqual(code, 1002);
@@ -393,7 +393,7 @@ describe('Receiver', function () {
 
     const p = new Receiver({ 'permessage-deflate': perMessageDeflate });
 
-    p.error = function (err, code) {
+    p.onerror = function (err, code) {
       assert.ok(err instanceof Error);
       assert.strictEqual(err.message, 'RSV1 must be clear');
       assert.strictEqual(code, 1002);
@@ -406,7 +406,7 @@ describe('Receiver', function () {
   it('raises an error when RSV2 is on', function (done) {
     const p = new Receiver();
 
-    p.error = function (err, code) {
+    p.onerror = function (err, code) {
       assert.ok(err instanceof Error);
       assert.strictEqual(err.message, 'RSV2 and RSV3 must be clear');
       assert.strictEqual(code, 1002);
@@ -419,7 +419,7 @@ describe('Receiver', function () {
   it('raises an error when RSV3 is on', function (done) {
     const p = new Receiver();
 
-    p.error = function (err, code) {
+    p.onerror = function (err, code) {
       assert.ok(err instanceof Error);
       assert.strictEqual(err.message, 'RSV2 and RSV3 must be clear');
       assert.strictEqual(code, 1002);
@@ -432,7 +432,7 @@ describe('Receiver', function () {
   it('raises an error if the first frame in a fragmented message has opcode 0', function (done) {
     const p = new Receiver();
 
-    p.error = function (err, code) {
+    p.onerror = function (err, code) {
       assert.ok(err instanceof Error);
       assert.strictEqual(err.message, 'invalid opcode: 0');
       assert.strictEqual(code, 1002);
@@ -445,7 +445,7 @@ describe('Receiver', function () {
   it('raises an error if a frame has opcode 1 in the middle of a fragmented message', function (done) {
     const p = new Receiver();
 
-    p.error = function (err, code) {
+    p.onerror = function (err, code) {
       assert.ok(err instanceof Error);
       assert.strictEqual(err.message, 'invalid opcode: 1');
       assert.strictEqual(code, 1002);
@@ -459,7 +459,7 @@ describe('Receiver', function () {
   it('raises an error if a frame has opcode 2 in the middle of a fragmented message', function (done) {
     const p = new Receiver();
 
-    p.error = function (err, code) {
+    p.onerror = function (err, code) {
       assert.ok(err instanceof Error);
       assert.strictEqual(err.message, 'invalid opcode: 2');
       assert.strictEqual(code, 1002);
@@ -473,7 +473,7 @@ describe('Receiver', function () {
   it('raises an error when a control frame has the FIN bit off', function (done) {
     const p = new Receiver();
 
-    p.error = function (err, code) {
+    p.onerror = function (err, code) {
       assert.ok(err instanceof Error);
       assert.strictEqual(err.message, 'FIN must be set');
       assert.strictEqual(code, 1002);
@@ -489,7 +489,7 @@ describe('Receiver', function () {
 
     const p = new Receiver({ 'permessage-deflate': perMessageDeflate });
 
-    p.error = function (err, code) {
+    p.onerror = function (err, code) {
       assert.ok(err instanceof Error);
       assert.strictEqual(err.message, 'RSV1 must be clear');
       assert.strictEqual(code, 1002);
@@ -502,7 +502,7 @@ describe('Receiver', function () {
   it('raises an error when a control frame has the FIN bit off', function (done) {
     const p = new Receiver();
 
-    p.error = function (err, code) {
+    p.onerror = function (err, code) {
       assert.ok(err instanceof Error);
       assert.strictEqual(err.message, 'FIN must be set');
       assert.strictEqual(code, 1002);
@@ -515,7 +515,7 @@ describe('Receiver', function () {
   it('raises an error when a control frame has a payload bigger than 125 B', function (done) {
     const p = new Receiver();
 
-    p.error = function (err, code) {
+    p.onerror = function (err, code) {
       assert.ok(err instanceof Error);
       assert.strictEqual(err.message, 'invalid payload length');
       assert.strictEqual(code, 1002);
@@ -528,7 +528,7 @@ describe('Receiver', function () {
   it('raises an error when a data frame has a payload bigger than 2^53 - 1 B', function (done) {
     const p = new Receiver();
 
-    p.error = function (err, code) {
+    p.onerror = function (err, code) {
       assert.ok(err instanceof Error);
       assert.strictEqual(err.message, 'max payload size exceeded');
       assert.strictEqual(code, 1009);
@@ -545,7 +545,7 @@ describe('Receiver', function () {
   it('raises an error if a text frame contains invalid UTF-8 data', function (done) {
     const p = new Receiver();
 
-    p.error = function (err, code) {
+    p.onerror = function (err, code) {
       assert.ok(err instanceof Error);
       assert.strictEqual(err.message, 'invalid utf8 sequence');
       assert.strictEqual(code, 1007);
@@ -558,7 +558,7 @@ describe('Receiver', function () {
   it('raises an error if a close frame has a payload of 1 B', function (done) {
     const p = new Receiver();
 
-    p.error = function (err, code) {
+    p.onerror = function (err, code) {
       assert.ok(err instanceof Error);
       assert.strictEqual(err.message, 'invalid payload length');
       assert.strictEqual(code, 1002);
@@ -571,7 +571,7 @@ describe('Receiver', function () {
   it('raises an error if a close frame contains an invalid close code', function (done) {
     const p = new Receiver();
 
-    p.error = function (err, code) {
+    p.onerror = function (err, code) {
       assert.ok(err instanceof Error);
       assert.strictEqual(err.message, 'invalid status code: 0');
       assert.strictEqual(code, 1002);
@@ -584,7 +584,7 @@ describe('Receiver', function () {
   it('raises an error if a close frame contains invalid UTF-8 data', function (done) {
     const p = new Receiver();
 
-    p.error = function (err, code) {
+    p.onerror = function (err, code) {
       assert.ok(err instanceof Error);
       assert.strictEqual(err.message, 'invalid utf8 sequence');
       assert.strictEqual(code, 1007);
@@ -602,7 +602,7 @@ describe('Receiver', function () {
     const frame = '82' + util.getHybiLengthAsHexString(msg.length, true) + mask +
       util.mask(msg, mask).toString('hex');
 
-    p.error = function (err, code) {
+    p.onerror = function (err, code) {
       assert.ok(err instanceof Error);
       assert.strictEqual(err.message, 'max payload size exceeded');
       assert.strictEqual(code, 1009);
@@ -619,7 +619,7 @@ describe('Receiver', function () {
     const frame = '82' + util.getHybiLengthAsHexString(msg.length, false) +
       msg.toString('hex');
 
-    p.error = function (err, code) {
+    p.onerror = function (err, code) {
       assert.ok(err instanceof Error);
       assert.strictEqual(err.message, 'max payload size exceeded');
       assert.strictEqual(code, 1009);
@@ -718,7 +718,7 @@ describe('Receiver', function () {
       p.add(frame);
       p.add(frame);
 
-      assert.strictEqual(p.state, 7);
+      assert.strictEqual(p.state, 5);
       assert.strictEqual(p.bufferedBytes, frame.length);
 
       p.cleanup(() => {
@@ -750,7 +750,7 @@ describe('Receiver', function () {
       p.add(textFrame);
       p.add(closeFrame);
 
-      assert.strictEqual(p.state, 7);
+      assert.strictEqual(p.state, 5);
       assert.strictEqual(p.bufferedBytes, textFrame.length + closeFrame.length);
 
       p.cleanup(() => {
@@ -782,7 +782,7 @@ describe('Receiver', function () {
       p.add(textFrame);
       p.add(invalidFrame);
 
-      assert.strictEqual(p.state, 7);
+      assert.strictEqual(p.state, 5);
       assert.strictEqual(p.bufferedBytes, textFrame.length + invalidFrame.length);
 
       p.cleanup(() => {
@@ -817,7 +817,7 @@ describe('Receiver', function () {
       p.add(textFrame);
       p.add(incompleteFrame);
 
-      assert.strictEqual(p.state, 7);
+      assert.strictEqual(p.state, 5);
       assert.strictEqual(p.bufferedBytes, incompleteFrame.length);
 
       p.cleanup(() => {
diff --git a/test/WebSocket.test.js b/test/WebSocket.test.js
index 13b9d4bb..f7ea03d9 100644
--- a/test/WebSocket.test.js
+++ b/test/WebSocket.test.js
@@ -143,7 +143,9 @@ describe('WebSocket', function () {
 
       it('takes into account the data in the sender queue', function (done) {
         const wss = new WebSocketServer({ port: ++port }, () => {
-          const ws = new WebSocket(`ws://localhost:${port}`);
+          const ws = new WebSocket(`ws://localhost:${port}`, {
+            perMessageDeflate: { threshold: 0 }
+          });
 
           ws.on('open', () => {
             ws.send('foo');