Experimental zlib-stream support part 3: Pako

So apparently the NodeJS built-in zlib package uses too many CPUs. Let's see if the pako package is better. Pako does have a nice synchronous API though. Pako is currently listed as an optional dependency, and Eris will use zlib as a fallback.
abalabahaha · Oct 21, 2017 · a105af3 · a105af3
1 parent cdf4a85
commit a105af3
Show file tree

Hide file tree

Showing 2 changed files with 48 additions and 14 deletions.
diff --git a/lib/gateway/Shard.js b/lib/gateway/Shard.js
@@ -25,6 +25,11 @@ try {
     Erlpack = require("erlpack");
 } catch(err) { // eslint-disable no-empty
 }
+var Pako;
+try {
+    Pako = require("pako");
+} catch(err) { // eslint-disable no-empty
+}
 try {
     WebSocket = require("uws");
 } catch(err) { // eslint-disable no-empty
@@ -1489,17 +1494,22 @@ class Shard extends EventEmitter {
     initializeWS() {
         this.status = "connecting";
         if(this.client.options.compress) {
-            this._zlib = Zlib.createUnzip({
-                flush: Z_SYNC_FLUSH,
-                chunkSize: 128 * 1024
-            }).on("error", (err) => {
-                this.emit("error", err);
-            });
-            this._zlibChunks = [];
-            this._zlibIncomingChunks = [];
-            this._zlib.on("data", (data) => {
-                this._zlibChunks.push(data);
-            });
+            if(Pako) {
+                this._pako = new Pako.Inflate({
+                    chunkSize: 128 * 1024
+                });
+            } else {
+                this._zlibIncomingChunks = [];
+                this._zlibChunks = [];
+                this._zlib = Zlib.createUnzip({
+                    flush: Z_SYNC_FLUSH,
+                    chunkSize: 128 * 1024
+                }).on("error", (err) => {
+                    this.emit("error", err);
+                }).on("data", (data) => {
+                    this._zlibChunks.push(data);
+                });
+            }
         }
         this.ws = new WebSocket(this.client.gatewayURL, this.client.options.ws);
         this.ws.onopen = () => {
@@ -1519,12 +1529,35 @@ class Shard extends EventEmitter {
             try {
                 var data = m.data;
                 if(data instanceof ArrayBuffer) {
-                    data = new Buffer(data);
+                    if(!this.client.options.compress) {
+                        data = new Buffer(data);
+                    }
                 } else if(Array.isArray(data)) { // Fragmented messages
                     data = Buffer.concat(data); // Copyfull concat is slow, but no alternative
                 }
                 if(this.client.options.compress) {
-                    if(this._zlibFlushing) {
+                    if(Pako) {
+                        if(data.length >= 4 && data.readUInt32BE(data.length - 4) === 0xFFFF) {
+                            this._pako.push(data, Z_SYNC_FLUSH);
+                            if(this._pako.err) {
+                                this.emit("error", new Error(`Pako error ${this._pako.err}: ${this._pako.msg}`));
+                                return;
+                            }
+
+                            data = Buffer.alloc ? Buffer.from(this._pako.result) : new Buffer(this._pako.result);
+                            if(Erlpack) {
+                                this.onWSMessage(Erlpack.unpack(data));
+                            } else {
+                                try {
+                                    return this.onWSMessage(JSON.parse(data.toString()));
+                                } catch(err) {
+                                    throw new Error(err.message + "\n\n" + data.toString());
+                                }
+                            }
+                        } else {
+                            this._pako.push(data, false);
+                        }
+                    } else if(this._zlibFlushing) {
                         this._zlibIncomingChunks.push(data);
                     } else if(this._zlib) {
                         this._zlib.write(data);
@@ -1540,7 +1573,7 @@ class Shard extends EventEmitter {
                     try {
                         return this.onWSMessage(JSON.parse(data.toString()));
                     } catch(err) {
-                        throw new Error(err.message + "\n\n" + data);
+                        throw new Error(err.message + "\n\n" + data.toString());
                     }
                 }
             } catch(err) {

diff --git a/package.json b/package.json
@@ -31,6 +31,7 @@
   },
   "optionalDependencies": {
     "opusscript": "^0.0.4",
+    "pako": "^1.0.6",
     "tweetnacl": "^1.0.0"
   },
   "browser": {