[feature] Add support for Blob (#2229)

Closes #2206

Author: lpinca

doc/ws.md

@@ -466,10 +466,11 @@ does nothing if `type` is not one of `'close'`, `'error'`, `'message'`, or
 - {String}
 
 A string indicating the type of binary data being transmitted by the connection.
-This should be one of "nodebuffer", "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 transferring large messages with multiple fragments.
+This should be one of "nodebuffer", "arraybuffer", "blob", 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 transferring large messages with multiple
+fragments.
 
 ### websocket.bufferedAmount
 
@@ -538,7 +539,8 @@ is a noop if the ready state is `CONNECTING` or `CLOSED`.
 
 ### websocket.ping([data[, mask]][, callback])
 
-- `data` {Array|Number|Object|String|ArrayBuffer|Buffer|DataView|TypedArray} The
+- `data`
+ {Array|Number|Object|String|ArrayBuffer|Buffer|DataView|TypedArray|Blob} The
  data to send in the ping frame.
 - `mask` {Boolean} Specifies whether `data` should be masked or not. Defaults to
  `true` when `websocket` is not a server client.
@@ -550,7 +552,8 @@ Send a ping. This method throws an error if the ready state is `CONNECTING`.
 
 ### websocket.pong([data[, mask]][, callback])
 
-- `data` {Array|Number|Object|String|ArrayBuffer|Buffer|DataView|TypedArray} The
+- `data`
+ {Array|Number|Object|String|ArrayBuffer|Buffer|DataView|TypedArray|Blob} The
  data to send in the pong frame.
 - `mask` {Boolean} Specifies whether `data` should be masked or not. Defaults to
  `true` when `websocket` is not a server client.
@@ -588,7 +591,8 @@ only removes listeners added with
 
 ### websocket.send(data[, options][, callback])
 
-- `data` {Array|Number|Object|String|ArrayBuffer|Buffer|DataView|TypedArray} The
+- `data`
+ {Array|Number|Object|String|ArrayBuffer|Buffer|DataView|TypedArray|Blob} The
  data to send. `Object` values are only supported if they conform to the
  requirements of [`Buffer.from()`][]. If those constraints are not met, a
  `TypeError` is thrown.

lib/constants.js

@@ -1,9 +1,15 @@
 'use strict';
 
+const BINARY_TYPES = ['nodebuffer', 'arraybuffer', 'fragments'];
+const hasBlob = typeof Blob !== 'undefined';
+
+if (hasBlob) BINARY_TYPES.push('blob');
+
 module.exports = {
- BINARY_TYPES: ['nodebuffer', 'arraybuffer', 'fragments'],
+ BINARY_TYPES,
  EMPTY_BUFFER: Buffer.alloc(0),
  GUID: '258EAFA5-E914-47DA-95CA-C5AB0DC85B11',
+ hasBlob,
  kForOnEventAttribute: Symbol('kIsForOnEventAttribute'),
  kListener: Symbol('kListener'),
  kStatusCode: Symbol('status-code'),

lib/receiver.js

@@ -559,6 +559,8 @@ class Receiver extends Writable {
  data = concat(fragments, messageLength);
  } else if (this._binaryType === 'arraybuffer') {
  data = toArrayBuffer(concat(fragments, messageLength));
+ } else if (this._binaryType === 'blob') {
+ data = new Blob(fragments);
  } else {
  data = fragments;
  }

lib/sender.js

@@ -6,8 +6,8 @@ const { Duplex } = require('stream');
 const { randomFillSync } = require('crypto');
 
 const PerMessageDeflate = require('./permessage-deflate');
-const { EMPTY_BUFFER } = require('./constants');
-const { isValidStatusCode } = require('./validation');
+const { EMPTY_BUFFER, kWebSocket, NOOP } = require('./constants');
+const { isBlob, isValidStatusCode } = require('./validation');
 const { mask: applyMask, toBuffer } = require('./buffer-util');
 
 const kByteLength = Symbol('kByteLength');
@@ -16,6 +16,10 @@ const RANDOM_POOL_SIZE = 8 * 1024;
 let randomPool;
 let randomPoolPointer = RANDOM_POOL_SIZE;
 
+const DEFAULT = 0;
+const DEFLATING = 1;
+const GET_BLOB_DATA = 2;
+
 /**
  * HyBi Sender implementation.
  */
@@ -42,8 +46,10 @@ class Sender {
  this._compress = false;
 
  this._bufferedBytes = 0;
- this._deflating = false;
  this._queue = [];
+ this._state = DEFAULT;
+ this.onerror = NOOP;
+ this[kWebSocket] = undefined;
  }
 
  /**
@@ -210,7 +216,7 @@ class Sender {
  rsv1: false
  };
 
- if (this._deflating) {
+ if (this._state !== DEFAULT) {
  this.enqueue([this.dispatch, buf, false, options, cb]);
  } else {
  this.sendFrame(Sender.frame(buf, options), cb);
@@ -232,6 +238,9 @@ class Sender {
  if (typeof data === 'string') {
  byteLength = Buffer.byteLength(data);
  readOnly = false;
+ } else if (isBlob(data)) {
+ byteLength = data.size;
+ readOnly = false;
  } else {
  data = toBuffer(data);
  byteLength = data.length;
@@ -253,7 +262,13 @@ class Sender {
  rsv1: false
  };
 
- if (this._deflating) {
+ if (isBlob(data)) {
+ if (this._state !== DEFAULT) {
+ this.enqueue([this.getBlobData, data, false, options, cb]);
+ } else {
+ this.getBlobData(data, false, options, cb);
+ }
+ } else if (this._state !== DEFAULT) {
  this.enqueue([this.dispatch, data, false, options, cb]);
  } else {
  this.sendFrame(Sender.frame(data, options), cb);
@@ -275,6 +290,9 @@ class Sender {
  if (typeof data === 'string') {
  byteLength = Buffer.byteLength(data);
  readOnly = false;
+ } else if (isBlob(data)) {
+ byteLength = data.size;
+ readOnly = false;
  } else {
  data = toBuffer(data);
  byteLength = data.length;
@@ -296,7 +314,13 @@ class Sender {
  rsv1: false
  };
 
- if (this._deflating) {
+ if (isBlob(data)) {
+ if (this._state !== DEFAULT) {
+ this.enqueue([this.getBlobData, data, false, options, cb]);
+ } else {
+ this.getBlobData(data, false, options, cb);
+ }
+ } else if (this._state !== DEFAULT) {
  this.enqueue([this.dispatch, data, false, options, cb]);
  } else {
  this.sendFrame(Sender.frame(data, options), cb);
@@ -330,6 +354,9 @@ class Sender {
  if (typeof data === 'string') {
  byteLength = Buffer.byteLength(data);
  readOnly = false;
+ } else if (isBlob(data)) {
+ byteLength = data.size;
+ readOnly = false;
  } else {
  data = toBuffer(data);
  byteLength = data.length;
@@ -357,40 +384,94 @@ class Sender {
 
  if (options.fin) this._firstFragment = true;
 
- if (perMessageDeflate) {
- const opts = {
-  [kByteLength]: byteLength,
-  fin: options.fin,
-  generateMask: this._generateMask,
-  mask: options.mask,
-  maskBuffer: this._maskBuffer,
-  opcode,
-  readOnly,
-  rsv1
- };
-
- if (this._deflating) {
- this.enqueue([this.dispatch, data, this._compress, opts, cb]);
+ const opts = {
+ [kByteLength]: byteLength,
+ fin: options.fin,
+ generateMask: this._generateMask,
+ mask: options.mask,
+ maskBuffer: this._maskBuffer,
+ opcode,
+ readOnly,
+ rsv1
+ };
+
+ if (isBlob(data)) {
+ if (this._state !== DEFAULT) {
+ this.enqueue([this.getBlobData, data, this._compress, opts, cb]);
  } else {
- this.dispatch(data, this._compress, opts, cb);
+ this.getBlobData(data, this._compress, opts, cb);
  }
+ } else if (this._state !== DEFAULT) {
+ this.enqueue([this.dispatch, data, this._compress, opts, cb]);
  } else {
- this.sendFrame(
- Sender.frame(data, {
- [kByteLength]: byteLength,
- fin: options.fin,
- generateMask: this._generateMask,
- mask: options.mask,
- maskBuffer: this._maskBuffer,
- opcode,
- readOnly,
- rsv1: false
- }),
- cb
- );
+ this.dispatch(data, this._compress, opts, cb);
  }
  }
 
+ /**
+ * Gets the contents of a blob as binary data.
+ *
+ * @param {Blob} blob The blob
+ * @param {Boolean} [compress=false] Specifies whether or not to compress
+ * the data
+ * @param {Object} options Options object
+ * @param {Boolean} [options.fin=false] Specifies whether or not to set the
+ * FIN bit
+ * @param {Function} [options.generateMask] The function used to generate the
+ * masking key
+ * @param {Boolean} [options.mask=false] Specifies whether or not to mask
+ * `data`
+ * @param {Buffer} [options.maskBuffer] The buffer used to store the masking
+ * key
+ * @param {Number} options.opcode The opcode
+ * @param {Boolean} [options.readOnly=false] Specifies whether `data` can be
+ * modified
+ * @param {Boolean} [options.rsv1=false] Specifies whether or not to set the
+ * RSV1 bit
+ * @param {Function} [cb] Callback
+ * @private
+ */
+ getBlobData(blob, compress, options, cb) {
+ this._bufferedBytes += options[kByteLength];
+ this._state = GET_BLOB_DATA;
+
+ blob
+ .arrayBuffer()
+ .then((arrayBuffer) => {
+ if (this._socket.destroyed) {
+ const err = new Error(
+ 'The socket was closed while the blob was being read'
+ );
+
+ //
+ // `callCallbacks` is called in the next tick to ensure that errors
+ // that might be thrown in the callbacks behave like errors thrown
+ // outside the promise chain.
+ //
+ process.nextTick(callCallbacks, this, err, cb);
+ return;
+ }
+
+ this._bufferedBytes -= options[kByteLength];
+ const data = toBuffer(arrayBuffer);
+
+ if (!compress) {
+ this._state = DEFAULT;
+ this.sendFrame(Sender.frame(data, options), cb);
+ this.dequeue();
+ } else {
+ this.dispatch(data, compress, options, cb);
+ }
+ })
+ .catch((err) => {
+ //
+ // `onError` is called in the next tick for the same reason that
+ // `callCallbacks` above is.
+ //
+ process.nextTick(onError, this, err, cb);
+ });
+ }
+
  /**
  * Dispatches a message.
  *
@@ -423,27 +504,19 @@ class Sender {
  const perMessageDeflate = this._extensions[PerMessageDeflate.extensionName];
 
  this._bufferedBytes += options[kByteLength];
- this._deflating = true;
+ this._state = DEFLATING;
  perMessageDeflate.compress(data, options.fin, (_, buf) => {
  if (this._socket.destroyed) {
  const err = new Error(
  'The socket was closed while data was being compressed'
  );
 
- if (typeof cb === 'function') cb(err);
-
- for (let i = 0; i < this._queue.length; i++) {
- const params = this._queue[i];
- const callback = params[params.length - 1];
-
- if (typeof callback === 'function') callback(err);
- }
-
+ callCallbacks(this, err, cb);
  return;
  }
 
  this._bufferedBytes -= options[kByteLength];
- this._deflating = false;
+ this._state = DEFAULT;
  options.readOnly = false;
  this.sendFrame(Sender.frame(buf, options), cb);
  this.dequeue();
@@ -456,7 +529,7 @@ class Sender {
  * @private
  */
  dequeue() {
- while (!this._deflating && this._queue.length) {
+ while (this._state === DEFAULT && this._queue.length) {
  const params = this._queue.shift();
 
  this._bufferedBytes -= params[3][kByteLength];
@@ -495,3 +568,35 @@ class Sender {
 }
 
 module.exports = Sender;
+
+/**
+ * Calls queued callbacks with an error.
+ *
+ * @param {Sender} sender The `Sender` instance
+ * @param {Error} err The error to call the callbacks with
+ * @param {Function} [cb] The first callback
+ * @private
+ */
+function callCallbacks(sender, err, cb) {
+ if (typeof cb === 'function') cb(err);
+
+ for (let i = 0; i < sender._queue.length; i++) {
+ const params = sender._queue[i];
+ const callback = params[params.length - 1];
+
+ if (typeof callback === 'function') callback(err);
+ }
+}
+
+/**
+ * Handles a `Sender` error.
+ *
+ * @param {Sender} sender The `Sender` instance
+ * @param {Error} err The error
+ * @param {Function} [cb] The first pending callback
+ * @private
+ */
+function onError(sender, err, cb) {
+ callCallbacks(sender, err, cb);
+ sender.onerror(err);
+}