[major] Do not decode Buffers to strings

Avoid decoding text messages and close reasons to strings. Pass them as
`Buffer`s to the listeners of their respective events. Also, make
listeners of the `'message'` event take a boolean argument to speficy
whether or not the message is binary.

Refs: #1878
Refs: #1804

Author: lpinca

bench/speed.js

@@ -19,7 +19,9 @@ if (cluster.isMaster) {
   });
 
   wss.on('connection', (ws) => {
-    ws.on('message', (data) => ws.send(data));
+    ws.on('message', (data, isBinary) => {
+      ws.send(data, { binary: isBinary });
+    });
   });
 
   server.listen(path ? { path } : { port }, () => cluster.fork());

doc/ws.md

@@ -300,11 +300,12 @@ it defaults to `/`.
 ### Event: 'close'
 
 - `code` {Number}
-- `reason` {String}
+- `reason` {Buffer}
 
 Emitted when the connection is closed. `code` is a numeric value indicating the
 status code explaining why the connection has been closed. `reason` is a
-human-readable string explaining why the connection has been closed.
+`Buffer` containing a human-readable string explaining why the connection has
+been closed.
 
 ### Event: 'error'
 
@@ -315,9 +316,11 @@ of the string values defined below under [WS Error Codes](#ws-error-codes).
 
 ### Event: 'message'
 
-- `data` {String|Buffer|ArrayBuffer|Buffer[]}
+- `data` {Buffer|ArrayBuffer|Buffer[]}
+- `isBinary` {Boolean}
 
-Emitted when a message is received from the server.
+Emitted when a message is received. `data` is the message content. `isBinary`
+specifies whether the message is binary or not.
 
 ### Event: 'open'
 
@@ -389,8 +392,7 @@ following ways:
 
 - `code` {Number} A numeric value indicating the status code explaining why the
   connection is being closed.
-- `reason` {String} A human-readable string explaining why the connection is
-  closing.
+- `reason` {String|Buffer} The reason why the connection is closing.
 
 Initiate a closing handshake.
 

lib/event-target.js

@@ -128,12 +128,15 @@ const EventTarget = {
   addEventListener(type, listener, options) {
     if (typeof listener !== 'function') return;
 
-    function onMessage(data) {
-      listener.call(this, new MessageEvent(data, this));
+    function onMessage(data, isBinary) {
+      listener.call(
+        this,
+        new MessageEvent(isBinary ? data : data.toString(), this)
+      );
     }
 
     function onClose(code, message) {
-      listener.call(this, new CloseEvent(code, message, this));
+      listener.call(this, new CloseEvent(code, message.toString(), this));
     }
 
     function onError(error) {

lib/receiver.js

@@ -499,7 +499,7 @@ class Receiver extends Writable {
           data = fragments;
         }
 
-        this.emit('message', data);
+        this.emit('message', data, true);
       } else {
         const buf = concat(fragments, messageLength);
 
@@ -514,7 +514,7 @@ class Receiver extends Writable {
           );
         }
 
-        this.emit('message', buf.toString());
+        this.emit('message', buf, false);
       }
     }
 
@@ -533,7 +533,7 @@ class Receiver extends Writable {
       this._loop = false;
 
       if (data.length === 0) {
-        this.emit('conclude', 1005, '');
+        this.emit('conclude', 1005, EMPTY_BUFFER);
         this.end();
       } else if (data.length === 1) {
         return error(
@@ -568,7 +568,7 @@ class Receiver extends Writable {
           );
         }
 
-        this.emit('conclude', code, buf.toString());
+        this.emit('conclude', code, buf);
         this.end();
       }
     } else if (this._opcode === 0x09) {

lib/sender.js

@@ -98,7 +98,7 @@ class Sender {
    * Sends a close message to the other peer.
    *
    * @param {Number} [code] The status code component of the body
-   * @param {String} [data] The message component of the body
+   * @param {(String|Buffer)} [data] The message component of the body
    * @param {Boolean} [mask=false] Specifies whether or not to mask the message
    * @param {Function} [cb] Callback
    * @public
@@ -110,7 +110,7 @@ class Sender {
       buf = EMPTY_BUFFER;
     } else if (typeof code !== 'number' || !isValidStatusCode(code)) {
       throw new TypeError('First argument must be a valid error code number');
-    } else if (data === undefined || data === '') {
+    } else if (data === undefined || !data.length) {
       buf = Buffer.allocUnsafe(2);
       buf.writeUInt16BE(code, 0);
     } else {
@@ -122,7 +122,12 @@ class Sender {
 
       buf = Buffer.allocUnsafe(2 + length);
       buf.writeUInt16BE(code, 0);
-      buf.write(data, 2);
+
+      if (typeof data === 'string') {
+        buf.write(data, 2);
+      } else {
+        buf.set(data, 2);
+      }
     }
 
     if (this._deflating) {

lib/stream.js

@@ -72,8 +72,11 @@ function createWebSocketStream(ws, options) {
     writableObjectMode: false
   });
 
-  ws.on('message', function message(msg) {
-    if (!duplex.push(msg)) {
+  ws.on('message', function message(msg, isBinary) {
+    const data =
+      !isBinary && duplex._readableState.objectMode ? msg.toString() : msg;
+
+    if (!duplex.push(data)) {
       resumeOnReceiverDrain = false;
       ws._socket.pause();
     }

lib/websocket.js

@@ -48,7 +48,7 @@ class WebSocket extends EventEmitter {
     this._closeCode = 1006;
     this._closeFrameReceived = false;
     this._closeFrameSent = false;
-    this._closeMessage = '';
+    this._closeMessage = EMPTY_BUFFER;
     this._closeTimer = null;
     this._extensions = {};
     this._protocol = '';
@@ -219,7 +219,8 @@ class WebSocket extends EventEmitter {
    *              +---+
    *
    * @param {Number} [code] Status code explaining why the connection is closing
-   * @param {String} [data] A string explaining why the connection is closing
+   * @param {(String|Buffer)} [data] The reason why the connection is
+   *     closing
    * @public
    */
   close(code, data) {
@@ -821,7 +822,7 @@ function sendAfterClose(websocket, data, cb) {
  * The listener of the `Receiver` `'conclude'` event.
  *
  * @param {Number} code The status code
- * @param {String} reason The reason for closing
+ * @param {Buffer} reason The reason for closing
  * @private
  */
 function receiverOnConclude(code, reason) {
@@ -875,11 +876,12 @@ function receiverOnFinish() {
 /**
  * The listener of the `Receiver` `'message'` event.
  *
- * @param {(String|Buffer|ArrayBuffer|Buffer[])} data The message
+ * @param {Buffer|ArrayBuffer|Buffer[])} data The message
+ * @param {Boolean} isBinary Specifies whether the message is binary or not
  * @private
  */
-function receiverOnMessage(data) {
-  this[kWebSocket].emit('message', data);
+function receiverOnMessage(data, isBinary) {
+  this[kWebSocket].emit('message', data, isBinary);
 }
 
 /**

test/create-websocket-stream.test.js

@@ -9,6 +9,7 @@ const { randomBytes } = require('crypto');
 const createWebSocketStream = require('../lib/stream');
 const Sender = require('../lib/sender');
 const WebSocket = require('..');
+const { EMPTY_BUFFER } = require('../lib/constants');
 
 describe('createWebSocketStream', () => {
   it('is exposed as a property of the `WebSocket` class', () => {
@@ -58,11 +59,12 @@ describe('createWebSocketStream', () => {
       });
 
       wss.on('connection', (ws) => {
-        ws.on('message', (message) => {
+        ws.on('message', (message, isBinary) => {
           ws.on('close', (code, reason) => {
-            assert.ok(message.equals(chunk));
+            assert.deepStrictEqual(message, chunk);
+            assert.ok(isBinary);
             assert.strictEqual(code, 1005);
-            assert.strictEqual(reason, '');
+            assert.strictEqual(reason, EMPTY_BUFFER);
             wss.close(done);
           });
         });
@@ -229,7 +231,7 @@ describe('createWebSocketStream', () => {
         ws._socket.write(Buffer.from([0x85, 0x00]));
         ws.on('close', (code, reason) => {
           assert.strictEqual(code, 1002);
-          assert.strictEqual(reason, '');
+          assert.deepStrictEqual(reason, EMPTY_BUFFER);
 
           serverClientCloseEventEmitted = true;
           if (duplexCloseEventEmitted) wss.close(done);
@@ -538,5 +540,33 @@ describe('createWebSocketStream', () => {
         });
       });
     });
+
+    it('converts text messages to strings in readable object mode', (done) => {
+      const wss = new WebSocket.Server({ port: 0 }, () => {
+        const events = [];
+        const ws = new WebSocket(`ws://localhost:${wss.address().port}`);
+        const duplex = createWebSocketStream(ws, { readableObjectMode: true });
+
+        duplex.on('data', (data) => {
+          events.push('data');
+          assert.strictEqual(data, 'foo');
+        });
+
+        duplex.on('end', () => {
+          events.push('end');
+          duplex.end();
+        });
+
+        duplex.on('close', () => {
+          assert.deepStrictEqual(events, ['data', 'end']);
+          wss.close(done);
+        });
+      });
+
+      wss.on('connection', (ws) => {
+        ws.send('foo');
+        ws.close();
+      });
+    });
   });
 });