Repository: websockets/ws
Pull request #1018 (2017-02-26 13:20:40)

Receiver support binaryType ‘fragments’ …
- emit message fragments array to avoid copyfull concat

Note:
binaryType used to affect before only in WHATWG API emulation
and the optional conversion to ArrayBuffer was done inside MessageEvent.
However in this change the receiver emits messages according to binaryType
so now also EventEmitter listeners added with on(‘message’, data => {})
will get the data according to binaryType.

guymguym: @lpinca Hi, would love to get your review.
See the Note I wrote above - let me know if you think this is a breaking change or not, and if you think it's a good direction.
Thanks
lpinca: @guymguym I've requested some changes but LGTM.
guymguym: @lpinca Fixed all comments. Thanks for the review!
lpinca: Can you please run the parser benchmark and post the results before and after this change? Thanks.
guymguym: Before:
```
$ git checkout master
Switched to branch 'master'
Your branch is up-to-date with 'origin/master'.

$ node bench/parser.benchmark.js 
ping message x 1,516,912 ops/sec ±0.66% (90 runs sampled)
ping with no data x 14,433,226 ops/sec ±1.00% (88 runs sampled)
close message x 12,816,502 ops/sec ±1.34% (87 runs sampled)
masked text message (20 bytes) x 1,010,852 ops/sec ±1.13% (88 runs sampled)
binary data (125 bytes) x 1,284,809 ops/sec ±1.29% (90 runs sampled)
binary data (65535 bytes) x 186,970 ops/sec ±1.17% (87 runs sampled)
binary data (200 KiB) x 69,283 ops/sec ±1.10% (86 runs sampled)
binary data (1 MiB) x 12,406 ops/sec ±0.90% (89 runs sampled)

$ node bench/parser.benchmark.js 
ping message x 1,500,888 ops/sec ±0.68% (89 runs sampled)
ping with no data x 13,664,804 ops/sec ±1.78% (83 runs sampled)
close message x 12,325,815 ops/sec ±1.19% (85 runs sampled)
masked text message (20 bytes) x 945,751 ops/sec ±0.98% (91 runs sampled)
binary data (125 bytes) x 1,219,381 ops/sec ±1.06% (87 runs sampled)
binary data (65535 bytes) x 184,835 ops/sec ±1.39% (87 runs sampled)
binary data (200 KiB) x 69,053 ops/sec ±1.00% (86 runs sampled)
binary data (1 MiB) x 12,375 ops/sec ±0.96% (87 runs sampled)
```

After:
```
$ git checkout guymguym-receiver-fragments 
Switched to branch 'guymguym-receiver-fragments'
Your branch is up-to-date with 'origin/guymguym-receiver-fragments'.

$ node bench/parser.benchmark.js 
ping message x 1,537,659 ops/sec ±0.98% (89 runs sampled)
ping with no data x 14,070,824 ops/sec ±0.85% (87 runs sampled)
close message x 12,873,083 ops/sec ±1.19% (85 runs sampled)
masked text message (20 bytes) x 944,098 ops/sec ±0.89% (87 runs sampled)
binary data (125 bytes) x 1,404,478 ops/sec ±1.30% (85 runs sampled)
binary data (65535 bytes) x 188,281 ops/sec ±1.31% (84 runs sampled)
binary data (200 KiB) x 70,403 ops/sec ±1.05% (86 runs sampled)
binary data (1 MiB) x 12,387 ops/sec ±0.91% (87 runs sampled)

$ node bench/parser.benchmark.js 
ping message x 1,528,134 ops/sec ±0.96% (90 runs sampled)
ping with no data x 14,189,599 ops/sec ±0.87% (89 runs sampled)
close message x 12,886,198 ops/sec ±1.64% (88 runs sampled)
masked text message (20 bytes) x 1,048,379 ops/sec ±1.38% (89 runs sampled)
binary data (125 bytes) x 1,412,051 ops/sec ±0.91% (89 runs sampled)
binary data (65535 bytes) x 189,763 ops/sec ±1.09% (89 runs sampled)
binary data (200 KiB) x 70,405 ops/sec ±1.15% (86 runs sampled)
binary data (1 MiB) x 12,353 ops/sec ±0.86% (86 runs sampled)
```
guymguym: Travis failed for node v4 but I think it's transient.
I don't have the option to re-trigger without commit, do you?
lpinca: Yes, test seems to be flaky. The timeout is probably too short. I'll restart.

I'll keep this open for a while so others can weigh in. If there are no objections I'll merge this tomorrow.

Patch:
diff --git a/.gitignore b/.gitignore
index 4b173c69..1e80e987 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,3 +1,4 @@
 node_modules/
 coverage/
 npm-debug.log
+.vscode/
diff --git a/doc/ws.md b/doc/ws.md
index cdc6aee0..038ba765 100644
--- a/doc/ws.md
+++ b/doc/ws.md
@@ -277,7 +277,10 @@ Register an event listener emulating the `EventTarget` interface.
 - {String}
 
 A string indicating the type of binary data being transmitted by the connection.
-This should be either "nodebuffer" or "arraybuffer". Defaults to "nodebuffer".
+This should be either "nodebuffer" or "arraybuffer" or "fragments". Defaults to "nodebuffer".
+Type "fragments" will emit the array of fragments as received from the sender,
+without copyfull concatenation, which is useful for the performance of binary protocols 
+transfering large messages with multiple fragments.
 
 ### websocket.bufferedAmount
 
diff --git a/lib/Constants.js b/lib/Constants.js
index 732df742..b1ab31fb 100644
--- a/lib/Constants.js
+++ b/lib/Constants.js
@@ -3,3 +3,5 @@
 exports.GUID = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11';
 exports.EMPTY_BUFFER = Buffer.alloc(0);
 exports.NOOP = () => {};
+
+exports.BINARY_TYPES = ['nodebuffer', 'arraybuffer', 'fragments'];
diff --git a/lib/EventTarget.js b/lib/EventTarget.js
index 9720e4a2..e30b1b3e 100644
--- a/lib/EventTarget.js
+++ b/lib/EventTarget.js
@@ -28,7 +28,7 @@ class MessageEvent extends Event {
   /**
    * Create a new `MessageEvent`.
    *
-   * @param {(String|Buffer|ArrayBuffer)} data The received data
+   * @param {(String|Buffer|ArrayBuffer|Buffer[])} data The received data
    * @param {Boolean} isBinary Specifies if `data` is binary
    * @param {WebSocket} target A reference to the target to which the event was dispatched
    */
@@ -100,9 +100,6 @@ const EventTarget = {
     if (typeof listener !== 'function') return;
 
     function onMessage (data, flags) {
-      if (flags.binary && this.binaryType === 'arraybuffer') {
-        data = new Uint8Array(data).buffer;
-      }
       listener.call(this, new MessageEvent(data, !!flags.binary, this));
     }
 
diff --git a/lib/Receiver.js b/lib/Receiver.js
index b43c3602..11d9efbf 100644
--- a/lib/Receiver.js
+++ b/lib/Receiver.js
@@ -28,10 +28,12 @@ class Receiver {
    *
    * @param {Object} extensions An object containing the negotiated extensions
    * @param {Number} maxPayload The maximum allowed message length
+   * @param {String} binaryType The type for binary data, see constants.BINARY_TYPES
    */
-  constructor (extensions, maxPayload) {
+  constructor (extensions, maxPayload, binaryType) {
     this.extensions = extensions || {};
     this.maxPayload = maxPayload | 0;
+    this.binaryType = binaryType || constants.BINARY_TYPES[0];
 
     this.bufferedBytes = 0;
     this.buffers = [];
@@ -352,20 +354,28 @@ class Receiver {
    */
   dataMessage () {
     if (this.fin) {
-      const buf = this.fragments.length > 1
-        ? Buffer.concat(this.fragments, this.messageLength)
-        : this.fragments.length === 1
-          ? this.fragments[0]
-          : constants.EMPTY_BUFFER;
-
+      const fragments = this.fragments;
+      const messageLength = this.messageLength;
       this.totalPayloadLength = 0;
-      this.fragments.length = 0;
+      this.fragments = [];
       this.messageLength = 0;
       this.fragmented = 0;
 
       if (this.opcode === 2) {
-        this.onmessage(buf, { masked: this.masked, binary: true });
+        var data;
+
+        if (this.binaryType === 'nodebuffer') {
+          data = bufferFromFragments(fragments, messageLength);
+        } else if (this.binaryType === 'arraybuffer') {
+          data = new Uint8Array(bufferFromFragments(fragments, messageLength)).buffer;
+        } else {
+          data = fragments;
+        }
+
+        this.onmessage(data, { masked: this.masked, binary: true });
       } else {
+        const buf = bufferFromFragments(fragments, messageLength);
+
         if (!isValidUTF8(buf)) {
           this.error(new Error('invalid utf8 sequence'), 1007);
           return;
@@ -509,4 +519,20 @@ class Receiver {
   }
 }
 
+/**
+ * Make a buffer from a list of fragments.
+ * Optimized for the common case of a single fragment in order to
+ * avoid copyfull concat and simply return the fragment buffer.
+ *
+ * @param {Buffer[]} fragments
+ * @param {Number} messageLength
+ * @return {Buffer}
+ * @private
+ */
+function bufferFromFragments (fragments, messageLength) {
+  if (fragments.length === 1) return fragments[0];
+  if (fragments.length > 1) return Buffer.concat(fragments, messageLength);
+  return constants.EMPTY_BUFFER;
+}
+
 module.exports = Receiver;
diff --git a/lib/WebSocket.js b/lib/WebSocket.js
index c0800adf..642330ba 100644
--- a/lib/WebSocket.js
+++ b/lib/WebSocket.js
@@ -53,7 +53,7 @@ class WebSocket extends EventEmitter {
     this.protocol = '';
 
     this._finalize = this.finalize.bind(this);
-    this._binaryType = 'nodebuffer';
+    this._binaryType = constants.BINARY_TYPES[0];
     this._finalizeCalled = false;
     this._closeMessage = null;
     this._closeTimer = null;
@@ -98,10 +98,17 @@ class WebSocket extends EventEmitter {
   }
 
   set binaryType (type) {
-    if (type === 'arraybuffer' || type === 'nodebuffer') {
-      this._binaryType = type;
-    } else {
-      throw new SyntaxError('unsupported binaryType: must be either "nodebuffer" or "arraybuffer"');
+    // silently ignore unsupported types
+    if (constants.BINARY_TYPES.indexOf(type) < 0) {
+      return;
+    }
+
+    this._binaryType = type;
+
+    // update the receiver if already created,
+    // if not then it will take the value once created
+    if (this._receiver) {
+      this._receiver.binaryType = type;
     }
   }
 
@@ -116,7 +123,7 @@ class WebSocket extends EventEmitter {
     socket.setTimeout(0);
     socket.setNoDelay();
 
-    this._receiver = new Receiver(this.extensions, this.maxPayload);
+    this._receiver = new Receiver(this.extensions, this.maxPayload, this.binaryType);
     this._sender = new Sender(socket, this.extensions);
     this._ultron = new Ultron(socket);
     this._socket = socket;


Test patch:
diff --git a/test/Receiver.test.js b/test/Receiver.test.js
index 1290388a..6d4e0771 100644
--- a/test/Receiver.test.js
+++ b/test/Receiver.test.js
@@ -5,6 +5,7 @@ const crypto = require('crypto');
 
 const PerMessageDeflate = require('../lib/PerMessageDeflate');
 const Receiver = require('../lib/Receiver');
+const Sender = require('../lib/Sender');
 const util = require('./hybi-util');
 
 describe('Receiver', function () {
@@ -827,4 +828,85 @@ describe('Receiver', function () {
       });
     });
   });
+
+  it('can emit nodebuffer of fragmented binary message', function (done) {
+    const p = new Receiver();
+    const frags = [
+      crypto.randomBytes(7321),
+      crypto.randomBytes(137),
+      crypto.randomBytes(285787),
+      crypto.randomBytes(3)
+    ];
+
+    p.binaryType = 'nodebuffer';
+    p.onmessage = function (data) {
+      assert.ok(Buffer.isBuffer(data));
+      assert.ok(data.equals(Buffer.concat(frags)));
+      done();
+    };
+
+    addBinaryFragments(p, frags);
+  });
+
+  it('can emit arraybuffer of fragmented binary message', function (done) {
+    const p = new Receiver();
+    const frags = [
+      crypto.randomBytes(19221),
+      crypto.randomBytes(954),
+      crypto.randomBytes(623987)
+    ];
+
+    p.binaryType = 'arraybuffer';
+    p.onmessage = function (data) {
+      assert.ok(data instanceof ArrayBuffer);
+      assert.ok(Buffer.from(data).equals(Buffer.concat(frags)));
+      done();
+    };
+
+    addBinaryFragments(p, frags);
+  });
+
+  it('can emit fragments of fragmented binary message', function (done) {
+    const p = new Receiver();
+    const frags = [
+      crypto.randomBytes(17),
+      crypto.randomBytes(419872),
+      crypto.randomBytes(83),
+      crypto.randomBytes(9928),
+      crypto.randomBytes(1)
+    ];
+
+    p.binaryType = 'fragments';
+    p.onmessage = function (data) {
+      assert.ok(Array.isArray(data));
+      assert.ok(data.length === frags.length);
+      for (let i = 0; i < frags.length; ++i) {
+        assert.ok(Buffer.isBuffer(data[i]));
+        assert.ok(frags[i].equals(data[i]));
+      }
+      done();
+    };
+
+    addBinaryFragments(p, frags);
+  });
 });
+
+/**
+ * Adds a list of binary fragments to the receiver prefixing each one
+ * with info byte and length.
+ *
+ * @param {Receiver} receiver
+ * @param {Buffer[]} frags
+ * @private
+ */
+function addBinaryFragments (receiver, frags) {
+  for (let i = 0; i < frags.length; ++i) {
+    Sender.frame(frags[i], {
+      opcode: i === 0 ? 2 : 0,
+      fin: i + 1 === frags.length,
+      mask: false,
+      rsv1: false,
+      readOnly: true
+    }).forEach(buf => receiver.add(buf));
+  }
+}
diff --git a/test/WebSocket.test.js b/test/WebSocket.test.js
index 8c336c9a..b1bcb0fe 100644
--- a/test/WebSocket.test.js
+++ b/test/WebSocket.test.js
@@ -1168,12 +1168,23 @@ describe('WebSocket', function () {
       assert.strictEqual(ws.onopen, listener);
     });
 
-    it('should throw an error when setting an invalid binary type', function () {
+    it('should ignore when setting an invalid binary type', function () {
       const ws = new WebSocket('ws://localhost', { agent: new CustomAgent() });
 
-      assert.throws(() => {
-        ws.binaryType = 'foo';
-      }, /^SyntaxError: unsupported binaryType: must be either "nodebuffer" or "arraybuffer"$/);
+      ws.binaryType = 'nodebuffer';
+      assert.ok(ws.binaryType === 'nodebuffer');
+      ws.binaryType = 'foo';
+      assert.ok(ws.binaryType === 'nodebuffer');
+      ws.binaryType = 'arraybuffer';
+      assert.ok(ws.binaryType === 'arraybuffer');
+      ws.binaryType = '';
+      assert.ok(ws.binaryType === 'arraybuffer');
+      ws.binaryType = 'fragments';
+      assert.ok(ws.binaryType === 'fragments');
+      ws.binaryType = 'buffer';
+      assert.ok(ws.binaryType === 'fragments');
+      ws.binaryType = 'nodebuffer';
+      assert.ok(ws.binaryType === 'nodebuffer');
     });
 
     it('should work the same as the EventEmitter api', function (done) {
@@ -1420,6 +1431,44 @@ describe('WebSocket', function () {
         };
       });
     });
+
+    it('should allow to update binaryType after receiver created', function (done) {
+      server.createServer(++port, (srv) => {
+        const ws = new WebSocket(`ws://localhost:${port}`);
+
+        function testType (binaryType, callback) {
+          const buf = Buffer.from(binaryType);
+          ws.binaryType = binaryType;
+          ws.onmessage = (messageEvent) => {
+            if (binaryType === 'nodebuffer') {
+              assert.ok(Buffer.isBuffer(messageEvent.data));
+              assert.ok(messageEvent.data.equals(buf));
+            } else if (binaryType === 'arraybuffer') {
+              assert.ok(messageEvent.data instanceof ArrayBuffer);
+              assert.ok(Buffer.from(messageEvent.data).equals(buf));
+            } else if (binaryType === 'fragments') {
+              assert.ok(Array.isArray(messageEvent.data));
+              assert.ok(messageEvent.data.length === 1);
+              assert.ok(Buffer.from(messageEvent.data[0]).equals(buf));
+            }
+            callback();
+          };
+          ws.send(buf);
+        }
+
+        ws.onopen =
+          () => testType('nodebuffer',
+            () => testType('arraybuffer',
+              () => testType('fragments',
+                () => {
+                  srv.close(done);
+                  ws.terminate();
+                }
+              )
+            )
+          );
+      });
+    });
   });
 
   describe('ssl', function () {