--- /dev/null
+'use strict';
+
+const Limiter = require('async-limiter');
+const zlib = require('zlib');
+
+const bufferUtil = require('./buffer-util');
+const constants = require('./constants');
+
+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');
+
+//
+// We limit zlib concurrency, which prevents severe memory fragmentation
+// as documented in https://github.com/nodejs/node/issues/8871#issuecomment-250915913
+// and https://github.com/websockets/ws/issues/1202
+//
+// Intentionally global; it's the global thread pool that's an issue.
+//
+let zlibLimiter;
+
+/**
+ * permessage-deflate implementation.
+ */
+class PerMessageDeflate {
+ /**
+ * Creates a PerMessageDeflate instance.
+ *
+ * @param {Object} options Configuration options
+ * @param {Boolean} options.serverNoContextTakeover Request/accept disabling
+ * of server context takeover
+ * @param {Boolean} options.clientNoContextTakeover Advertise/acknowledge
+ * disabling of client context takeover
+ * @param {(Boolean|Number)} options.serverMaxWindowBits Request/confirm the
+ * use of a custom server window size
+ * @param {(Boolean|Number)} options.clientMaxWindowBits Advertise support
+ * for, or request, a custom client window size
+ * @param {Object} options.zlibDeflateOptions Options to pass to zlib on deflate
+ * @param {Object} options.zlibInflateOptions Options to pass to zlib on inflate
+ * @param {Number} options.threshold Size (in bytes) below which messages
+ * should not be compressed
+ * @param {Number} options.concurrencyLimit The number of concurrent calls to
+ * zlib
+ * @param {Boolean} isServer Create the instance in either server or client
+ * mode
+ * @param {Number} maxPayload The maximum allowed message length
+ */
+ constructor (options, isServer, maxPayload) {
+ this._maxPayload = maxPayload | 0;
+ this._options = options || {};
+ this._threshold = this._options.threshold !== undefined
+ ? this._options.threshold
+ : 1024;
+ this._isServer = !!isServer;
+ this._deflate = null;
+ this._inflate = null;
+
+ this.params = null;
+
+ if (!zlibLimiter) {
+ const concurrency = this._options.concurrencyLimit !== undefined
+ ? this._options.concurrencyLimit
+ : 10;
+ zlibLimiter = new Limiter({ concurrency });
+ }
+ }
+
+ /**
+ * @type {String}
+ */
+ static get extensionName () {
+ return 'permessage-deflate';
+ }
+
+ /**
+ * Create an extension negotiation offer.
+ *
+ * @return {Object} Extension parameters
+ * @public
+ */
+ offer () {
+ const params = {};
+
+ if (this._options.serverNoContextTakeover) {
+ params.server_no_context_takeover = true;
+ }
+ if (this._options.clientNoContextTakeover) {
+ params.client_no_context_takeover = true;
+ }
+ if (this._options.serverMaxWindowBits) {
+ params.server_max_window_bits = this._options.serverMaxWindowBits;
+ }
+ if (this._options.clientMaxWindowBits) {
+ params.client_max_window_bits = this._options.clientMaxWindowBits;
+ } else if (this._options.clientMaxWindowBits == null) {
+ params.client_max_window_bits = true;
+ }
+
+ return params;
+ }
+
+ /**
+ * Accept an extension negotiation offer/response.
+ *
+ * @param {Array} configurations The extension negotiation offers/reponse
+ * @return {Object} Accepted configuration
+ * @public
+ */
+ accept (configurations) {
+ configurations = this.normalizeParams(configurations);
+
+ this.params = this._isServer
+ ? this.acceptAsServer(configurations)
+ : this.acceptAsClient(configurations);
+
+ return this.params;
+ }
+
+ /**
+ * Releases all resources used by the extension.
+ *
+ * @public
+ */
+ cleanup () {
+ if (this._inflate) {
+ if (this._inflate[kWriteInProgress]) {
+ this._inflate[kPendingClose] = true;
+ } else {
+ this._inflate.close();
+ this._inflate = null;
+ }
+ }
+ if (this._deflate) {
+ if (this._deflate[kWriteInProgress]) {
+ this._deflate[kPendingClose] = true;
+ } else {
+ this._deflate.close();
+ this._deflate = null;
+ }
+ }
+ }
+
+ /**
+ * Accept an extension negotiation offer.
+ *
+ * @param {Array} offers The extension negotiation offers
+ * @return {Object} Accepted configuration
+ * @private
+ */
+ acceptAsServer (offers) {
+ const opts = this._options;
+ const accepted = offers.find((params) => {
+ if (
+ (opts.serverNoContextTakeover === false &&
+ params.server_no_context_takeover) ||
+ (params.server_max_window_bits &&
+ (opts.serverMaxWindowBits === false ||
+ (typeof opts.serverMaxWindowBits === 'number' &&
+ opts.serverMaxWindowBits > params.server_max_window_bits))) ||
+ (typeof opts.clientMaxWindowBits === 'number' &&
+ !params.client_max_window_bits)
+ ) {
+ return false;
+ }
+
+ return true;
+ });
+
+ if (!accepted) {
+ throw new Error('None of the extension offers can be accepted');
+ }
+
+ if (opts.serverNoContextTakeover) {
+ accepted.server_no_context_takeover = true;
+ }
+ if (opts.clientNoContextTakeover) {
+ accepted.client_no_context_takeover = true;
+ }
+ if (typeof opts.serverMaxWindowBits === 'number') {
+ accepted.server_max_window_bits = opts.serverMaxWindowBits;
+ }
+ if (typeof opts.clientMaxWindowBits === 'number') {
+ accepted.client_max_window_bits = opts.clientMaxWindowBits;
+ } else if (
+ accepted.client_max_window_bits === true ||
+ opts.clientMaxWindowBits === false
+ ) {
+ delete accepted.client_max_window_bits;
+ }
+
+ return accepted;
+ }
+
+ /**
+ * Accept the extension negotiation response.
+ *
+ * @param {Array} response The extension negotiation response
+ * @return {Object} Accepted configuration
+ * @private
+ */
+ acceptAsClient (response) {
+ const params = response[0];
+
+ if (
+ this._options.clientNoContextTakeover === false &&
+ params.client_no_context_takeover
+ ) {
+ throw new Error('Unexpected parameter "client_no_context_takeover"');
+ }
+
+ if (!params.client_max_window_bits) {
+ if (typeof this._options.clientMaxWindowBits === 'number') {
+ params.client_max_window_bits = this._options.clientMaxWindowBits;
+ }
+ } else if (
+ this._options.clientMaxWindowBits === false ||
+ (typeof this._options.clientMaxWindowBits === 'number' &&
+ params.client_max_window_bits > this._options.clientMaxWindowBits)
+ ) {
+ throw new Error(
+ 'Unexpected or invalid parameter "client_max_window_bits"'
+ );
+ }
+
+ return params;
+ }
+
+ /**
+ * Normalize parameters.
+ *
+ * @param {Array} configurations The extension negotiation offers/reponse
+ * @return {Array} The offers/response with normalized parameters
+ * @private
+ */
+ normalizeParams (configurations) {
+ configurations.forEach((params) => {
+ Object.keys(params).forEach((key) => {
+ var value = params[key];
+
+ if (value.length > 1) {
+ throw new Error(`Parameter "${key}" must have only a single value`);
+ }
+
+ value = value[0];
+
+ if (key === 'client_max_window_bits') {
+ if (value !== true) {
+ const num = +value;
+ if (!Number.isInteger(num) || num < 8 || num > 15) {
+ throw new TypeError(
+ `Invalid value for parameter "${key}": ${value}`
+ );
+ }
+ value = num;
+ } else if (!this._isServer) {
+ throw new TypeError(
+ `Invalid value for parameter "${key}": ${value}`
+ );
+ }
+ } else if (key === 'server_max_window_bits') {
+ const num = +value;
+ if (!Number.isInteger(num) || num < 8 || num > 15) {
+ throw new TypeError(
+ `Invalid value for parameter "${key}": ${value}`
+ );
+ }
+ value = num;
+ } else if (
+ key === 'client_no_context_takeover' ||
+ key === 'server_no_context_takeover'
+ ) {
+ if (value !== true) {
+ throw new TypeError(
+ `Invalid value for parameter "${key}": ${value}`
+ );
+ }
+ } else {
+ throw new Error(`Unknown parameter "${key}"`);
+ }
+
+ params[key] = value;
+ });
+ });
+
+ return configurations;
+ }
+
+ /**
+ * Decompress data. Concurrency limited by async-limiter.
+ *
+ * @param {Buffer} data Compressed data
+ * @param {Boolean} fin Specifies whether or not this is the last fragment
+ * @param {Function} callback Callback
+ * @public
+ */
+ decompress (data, fin, callback) {
+ zlibLimiter.push((done) => {
+ this._decompress(data, fin, (err, result) => {
+ done();
+ callback(err, result);
+ });
+ });
+ }
+
+ /**
+ * Compress data. Concurrency limited by async-limiter.
+ *
+ * @param {Buffer} data Data to compress
+ * @param {Boolean} fin Specifies whether or not this is the last fragment
+ * @param {Function} callback Callback
+ * @public
+ */
+ compress (data, fin, callback) {
+ zlibLimiter.push((done) => {
+ this._compress(data, fin, (err, result) => {
+ done();
+ callback(err, result);
+ });
+ });
+ }
+
+ /**
+ * Decompress data.
+ *
+ * @param {Buffer} data Compressed data
+ * @param {Boolean} fin Specifies whether or not this is the last fragment
+ * @param {Function} callback Callback
+ * @private
+ */
+ _decompress (data, fin, callback) {
+ const endpoint = this._isServer ? 'client' : 'server';
+
+ if (!this._inflate) {
+ const key = `${endpoint}_max_window_bits`;
+ const windowBits = typeof this.params[key] !== 'number'
+ ? zlib.Z_DEFAULT_WINDOWBITS
+ : this.params[key];
+
+ this._inflate = zlib.createInflateRaw(
+ Object.assign({}, this._options.zlibInflateOptions, { windowBits })
+ );
+ this._inflate[kPerMessageDeflate] = this;
+ this._inflate[kTotalLength] = 0;
+ this._inflate[kBuffers] = [];
+ this._inflate.on('error', inflateOnError);
+ this._inflate.on('data', inflateOnData);
+ }
+
+ this._inflate[kCallback] = callback;
+ this._inflate[kWriteInProgress] = true;
+
+ this._inflate.write(data);
+ if (fin) this._inflate.write(TRAILER);
+
+ this._inflate.flush(() => {
+ const err = this._inflate[kError];
+
+ if (err) {
+ this._inflate.close();
+ this._inflate = null;
+ callback(err);
+ return;
+ }
+
+ const data = bufferUtil.concat(
+ this._inflate[kBuffers],
+ this._inflate[kTotalLength]
+ );
+
+ if (
+ (fin && this.params[`${endpoint}_no_context_takeover`]) ||
+ this._inflate[kPendingClose]
+ ) {
+ this._inflate.close();
+ this._inflate = null;
+ } else {
+ this._inflate[kWriteInProgress] = false;
+ this._inflate[kTotalLength] = 0;
+ this._inflate[kBuffers] = [];
+ }
+
+ callback(null, data);
+ });
+ }
+
+ /**
+ * Compress data.
+ *
+ * @param {Buffer} data Data to compress
+ * @param {Boolean} fin Specifies whether or not this is the last fragment
+ * @param {Function} callback Callback
+ * @private
+ */
+ _compress (data, fin, callback) {
+ if (!data || data.length === 0) {
+ process.nextTick(callback, null, EMPTY_BLOCK);
+ return;
+ }
+
+ const endpoint = this._isServer ? 'server' : 'client';
+
+ if (!this._deflate) {
+ const key = `${endpoint}_max_window_bits`;
+ const windowBits = typeof this.params[key] !== 'number'
+ ? zlib.Z_DEFAULT_WINDOWBITS
+ : this.params[key];
+
+ this._deflate = zlib.createDeflateRaw(
+ Object.assign(
+ // TODO deprecate memLevel/level and recommend zlibDeflateOptions instead
+ {
+ memLevel: this._options.memLevel,
+ level: this._options.level
+ },
+ this._options.zlibDeflateOptions,
+ { windowBits }
+ )
+ );
+
+ this._deflate[kTotalLength] = 0;
+ this._deflate[kBuffers] = [];
+
+ //
+ // `zlib.DeflateRaw` emits an `'error'` event only when an attempt to use
+ // it is made after it has already been closed. This cannot happen here,
+ // so we only add a listener for the `'data'` event.
+ //
+ this._deflate.on('data', deflateOnData);
+ }
+
+ this._deflate[kWriteInProgress] = true;
+
+ this._deflate.write(data);
+ this._deflate.flush(zlib.Z_SYNC_FLUSH, () => {
+ var data = bufferUtil.concat(
+ this._deflate[kBuffers],
+ this._deflate[kTotalLength]
+ );
+
+ if (fin) data = data.slice(0, data.length - 4);
+
+ if (
+ (fin && this.params[`${endpoint}_no_context_takeover`]) ||
+ this._deflate[kPendingClose]
+ ) {
+ this._deflate.close();
+ this._deflate = null;
+ } else {
+ this._deflate[kWriteInProgress] = false;
+ this._deflate[kTotalLength] = 0;
+ this._deflate[kBuffers] = [];
+ }
+
+ callback(null, data);
+ });
+ }
+}
+
+module.exports = PerMessageDeflate;
+
+/**
+ * The listener of the `zlib.DeflateRaw` stream `'data'` event.
+ *
+ * @param {Buffer} chunk A chunk of data
+ * @private
+ */
+function deflateOnData (chunk) {
+ this[kBuffers].push(chunk);
+ this[kTotalLength] += chunk.length;
+}
+
+/**
+ * The listener of the `zlib.InflateRaw` stream `'data'` event.
+ *
+ * @param {Buffer} chunk A chunk of data
+ * @private
+ */
+function inflateOnData (chunk) {
+ this[kTotalLength] += chunk.length;
+
+ if (
+ this[kPerMessageDeflate]._maxPayload < 1 ||
+ this[kTotalLength] <= this[kPerMessageDeflate]._maxPayload
+ ) {
+ this[kBuffers].push(chunk);
+ return;
+ }
+
+ this[kError] = new RangeError('Max payload size exceeded');
+ this[kError][constants.kStatusCode] = 1009;
+ this.removeListener('data', inflateOnData);
+ this.reset();
+}
+
+/**
+ * The listener of the `zlib.InflateRaw` stream `'error'` event.
+ *
+ * @param {Error} err The emitted error
+ * @private
+ */
+function inflateOnError (err) {
+ //
+ // There is no need to call `Zlib#close()` as the handle is automatically
+ // closed when an error is emitted.
+ //
+ this[kPerMessageDeflate]._inflate = null;
+ err[constants.kStatusCode] = 1007;
+ this[kCallback](err);
+}
projects / VSoRC / .git / blobdiff