zlib stream compression (#2028)

README.md

@@ -41,6 +41,7 @@ Using opusscript is only recommended for development environments where node-opu
 For production bots, using node-opus should be considered a necessity, especially if they're going to be running on multiple servers.
 
 ### Optional packages
+- [zlib-sync](https://www.npmjs.com/package/zlib-sync) for significantly faster WebSocket data inflation (`npm install zlib-sync`) 
 - [bufferutil](https://www.npmjs.com/package/bufferutil) to greatly speed up the WebSocket when *not* using uws (`npm install bufferutil --save`)
 - [erlpack](https://github.com/hammerandchisel/erlpack) for significantly faster WebSocket data (de)serialisation (`npm install discordapp/erlpack --save`)
 - One of the following packages can be installed for faster voice packet encryption and decryption:

package.json

@@ -33,6 +33,7 @@
   "runkitExampleFilename": "./docs/examples/ping.js",
   "dependencies": {
     "long": "^3.0.0",
+    "pako": "^1.0.0",
     "prism-media": "^0.0.2",
     "snekfetch": "^3.0.0",
     "tweetnacl": "^1.0.0",
@@ -45,7 +46,8 @@
     "opusscript": "^0.0.4",
     "sodium": "^2.0.0",
     "libsodium-wrappers": "^0.7.0",
-    "uws": "^8.14.0"
+    "uws": "^8.14.0",
+    "zlib-sync": "^0.1.0"
   },
   "devDependencies": {
     "@types/node": "^8.0.0",
@@ -70,6 +72,7 @@
     "node-opus": false,
     "tweetnacl": false,
     "sodium": false,
+    "zlib-sync": false,
     "src/sharding/Shard.js": false,
     "src/sharding/ShardClientUtil.js": false,
     "src/sharding/ShardingManager.js": false,

src/WebSocket.js

@@ -1,6 +1,9 @@
 const { browser } = require('./util/Constants');
-const zlib = require('zlib');
 const querystring = require('querystring');
+try {
+  var erlpack = require('erlpack');
+  if (!erlpack.pack) erlpack = null;
+} catch (err) {} // eslint-disable-line no-empty
 
 if (browser) {
   exports.WebSocket = window.WebSocket; // eslint-disable-line no-undef
@@ -12,25 +15,14 @@ if (browser) {
   }
 }
 
-try {
-  var erlpack = require('erlpack');
-  if (!erlpack.pack) erlpack = null;
-} catch (err) {} // eslint-disable-line no-empty
-
 exports.encoding = erlpack ? 'etf' : 'json';
 
 exports.pack = erlpack ? erlpack.pack : JSON.stringify;
 
 exports.unpack = data => {
-  if (Array.isArray(data)) data = Buffer.concat(data);
+  if (!erlpack || data[0] === '{') return JSON.parse(data);
-  if (!browser && data instanceof ArrayBuffer) data = Buffer.from(new Uint8Array(data));
+  if (!(data instanceof Buffer)) data = Buffer.from(new Uint8Array(data));
-
+  return erlpack.unpack(data);
-  if (erlpack && typeof data !== 'string') {
-    return erlpack.unpack(data);
-  } else if (data instanceof ArrayBuffer || (!browser && data instanceof Buffer)) {
-    data = zlib.inflateSync(data).toString();
-  }
-  return JSON.parse(data);
 };
 
 exports.create = (gateway, query = {}, ...args) => {

src/client/websocket/WebSocketConnection.js

@@ -1,7 +1,13 @@
 const EventEmitter = require('events');
-const { DefaultOptions, Events, OPCodes, Status, WSCodes } = require('../../util/Constants');
+const { Events, OPCodes, Status, WSCodes } = require('../../util/Constants');
 const PacketManager = require('./packets/WebSocketPacketManager');
 const WebSocket = require('../../WebSocket');
+try {
+  var zlib = require('zlib-sync');
+  if (!zlib.Inflate) zlib = require('pako');
+} catch (err) {
+  zlib = require('pako');
+}
 
 /**
  * Abstracts a WebSocket connection with decoding/encoding for the Discord gateway.
@@ -67,13 +73,13 @@ class WebSocketConnection extends EventEmitter {
       time: 60e3,
       resetTimer: null,
     };
-    this.connect(gateway);
 
     /**
      * Events that are disabled (will not be processed)
      * @type {Object}
      */
     this.disabledEvents = {};
+    for (const event of this.client.options.disabledEvents) this.disabledEvents[event] = true;
 
     /**
      * The sequence on WebSocket close
@@ -86,7 +92,9 @@ class WebSocketConnection extends EventEmitter {
      * @type {boolean}
      */
     this.expectingClose = false;
-    for (const event of this.client.options.disabledEvents) this.disabledEvents[event] = true;
+
+    this.inflate = null;
+    this.connect(gateway);
   }
 
   /**
@@ -206,10 +214,18 @@ class WebSocketConnection extends EventEmitter {
       this.debug(`Tried to connect to an invalid gateway: ${gateway}`);
       return false;
     }
+    this.inflate = new zlib.Inflate({
+      chunkSize: 65535,
+      flush: zlib.Z_SYNC_FLUSH,
+      to: WebSocket.encoding === 'json' ? 'string' : '',
+    });
     this.expectingClose = false;
     this.gateway = gateway;
     this.debug(`Connecting to ${gateway}`);
-    const ws = this.ws = WebSocket.create(gateway, { v: DefaultOptions.ws.version });
+    const ws = this.ws = WebSocket.create(gateway, {
+      v: this.client.options.ws.version,
+      compress: 'zlib-stream',
+    });
     ws.onmessage = this.onMessage.bind(this);
     ws.onopen = this.onOpen.bind(this);
     ws.onerror = this.onError.bind(this);
@@ -241,18 +257,25 @@ class WebSocketConnection extends EventEmitter {
   /**
    * Called whenever a message is received.
    * @param {Event} event Event received
-   * @returns {boolean}
    */
-  onMessage(event) {
+  onMessage({ data }) {
-    let data;
+    if (data instanceof ArrayBuffer) data = new Uint8Array(data);
+    const l = data.length;
+    const flush = l >= 4 &&
+      data[l - 4] === 0x00 &&
+      data[l - 3] === 0x00 &&
+      data[l - 2] === 0xFF &&
+      data[l - 1] === 0xFF;
+
+    this.inflate.push(data, flush && zlib.Z_SYNC_FLUSH);
+    if (!flush) return;
     try {
-      data = WebSocket.unpack(event.data);
+      const packet = WebSocket.unpack(this.inflate.result);
+      this.onPacket(packet);
+      if (this.client.listenerCount('raw')) this.client.emit('raw', data);
     } catch (err) {
       this.client.emit('debug', err);
     }
-    const ret = this.onPacket(data);
-    this.client.emit('raw', data);
-    return ret;
   }
 
   /**

src/util/Constants.js

@@ -57,7 +57,7 @@ exports.DefaultOptions = {
    */
   ws: {
     large_threshold: 250,
-    compress: !browser,
+    compress: false,
     properties: {
       $os: browser ? 'browser' : process.platform,
       $browser: 'discord.js',

test/webpack.html

@@ -18,7 +18,6 @@
       client.on('debug', console.log);
 
       client.on('error', console.error);
-      client.on('debug', console.info);
 
       client.ws.on('close', (event) => console.log('[CLIENT] Disconnect!', event));
 

webpack.config.js

@@ -55,11 +55,11 @@ const createConfig = options => {
       tls: 'mock',
       child_process: 'empty',
       dgram: 'empty',
-      zlib: 'empty',
       __dirname: true,
       process: false,
       path: 'empty',
       Buffer: false,
+      zlib: 'empty',
     },
     plugins,
   };