From d11e17c8d7c09b9f0dbc4b0df72c6f6914aa68bc Mon Sep 17 00:00:00 2001 From: Naoyuki Kanezawa Date: Tue, 6 Jan 2015 02:11:39 +0900 Subject: [PATCH] support compression --- README.md | 8 ++ lib/server.js | 14 ++- lib/socket.js | 25 +++-- lib/transports/polling-jsonp.js | 18 +--- lib/transports/polling-xhr.js | 30 ------ lib/transports/polling.js | 119 ++++++++++++++++++++++-- lib/transports/websocket.js | 5 +- package.json | 5 +- test/server.js | 158 ++++++++++++++++++++++++++++++++ 9 files changed, 316 insertions(+), 66 deletions(-) diff --git a/README.md b/README.md index 3ad198a16..da7a3ac16 100644 --- a/README.md +++ b/README.md @@ -212,6 +212,11 @@ to a single process. to (`['polling', 'websocket']`) - `allowUpgrades` (`Boolean`): whether to allow transport upgrades (`true`) + - `perMessageDeflate` (`Object|Boolean`): parameters of the WebSocket permessage-deflate extension + (see [ws module](https://github.com/einaros/ws) api docs). Set to `false` to disable. (`true`) + - `httpCompression` (`Object|Boolean`): parameters of the http compression for the polling transports + (see [zlib](http://nodejs.org/api/zlib.html#zlib_options) api docs). Set to `false` to disable. (`true`) + - `threshold` (`Number`): data is compressed only if the byte size is above this value (`1024`) - `cookie` (`String|Boolean`): name of the HTTP cookie that contains the client sid to send as part of handshake response headers. Set to `false` to not send one. (`io`) @@ -297,7 +302,10 @@ A representation of a client. _Inherits from EventEmitter_. sending binary data, which is sent as is. - **Parameters** - `String` | `Buffer` | `ArrayBuffer` | `ArrayBufferView`: a string or any object implementing `toString()`, with outgoing data, or a Buffer or ArrayBuffer with binary data. Also any ArrayBufferView can be sent as is. + - `Object`: optional, options object - `Function`: optional, a callback executed when the message gets flushed out by the transport + - **Options** + - `compress` (`Boolean`): whether to compress sending data. This option might be ignored and forced to be `true` when using polling. (`true`) - **Returns** `Socket` for chaining - `close` - Disconnects the client diff --git a/lib/server.js b/lib/server.js index b2a0b7f5c..2afd64897 100644 --- a/lib/server.js +++ b/lib/server.js @@ -44,10 +44,21 @@ function Server(opts){ this.allowUpgrades = false !== opts.allowUpgrades; this.allowRequest = opts.allowRequest; this.cookie = false !== opts.cookie ? (opts.cookie || 'io') : false; + this.perMessageDeflate = false !== opts.perMessageDeflate ? (opts.perMessageDeflate || true) : false; + this.httpCompression = false !== opts.httpCompression ? (opts.httpCompression || {}) : false; + + if (true === this.httpCompression) this.httpCompression = {}; + if (this.httpCompression && null == this.httpCompression.threshold) { + this.httpCompression.threshold = 1024; + } // initialize websocket server if (~this.transports.indexOf('websocket')) { - this.ws = new WebSocketServer({ noServer: true, clientTracking: false }); + this.ws = new WebSocketServer({ + noServer: true, + clientTracking: false, + perMessageDeflate: this.perMessageDeflate + }); } } @@ -228,6 +239,7 @@ Server.prototype.handshake = function(transport, req){ var transport = new transports[transport](req); if ('polling' == transportName) { transport.maxHttpBufferSize = this.maxHttpBufferSize; + transport.httpCompression = this.httpCompression; } if (req._query && req._query.b64) { diff --git a/lib/socket.js b/lib/socket.js index 08b2b8b20..71a95ee8c 100644 --- a/lib/socket.js +++ b/lib/socket.js @@ -173,7 +173,7 @@ Socket.prototype.maybeUpgrade = function (transport) { function onPacket(packet){ if ('ping' == packet.type && 'probe' == packet.data) { - transport.send([{ type: 'pong', data: 'probe' }]); + transport.send([{ type: 'pong', data: 'probe', options: { compress: true } }]); clearInterval(self.checkIntervalTimer); self.checkIntervalTimer = setInterval(check, 100); } else if ('upgrade' == packet.type && self.readyState != 'closed') { @@ -202,7 +202,7 @@ Socket.prototype.maybeUpgrade = function (transport) { function check(){ if ('polling' == self.transport.name && self.transport.writable) { debug('writing a noop packet to polling for fast upgrade'); - self.transport.send([{ type: 'noop' }]); + self.transport.send([{ type: 'noop', options: { compress: true } }]); } } @@ -280,14 +280,15 @@ Socket.prototype.setupSendCallback = function () { * Sends a message packet. * * @param {String} message + * @param {Object} options * @param {Function} callback * @return {Socket} for chaining * @api public */ Socket.prototype.send = -Socket.prototype.write = function(data, callback){ - this.sendPacket('message', data, callback); +Socket.prototype.write = function(data, options, callback){ + this.sendPacket('message', data, options, callback); return this; }; @@ -296,14 +297,26 @@ Socket.prototype.write = function(data, callback){ * * @param {String} packet type * @param {String} optional, data + * @param {Object} options * @api private */ -Socket.prototype.sendPacket = function (type, data, callback) { +Socket.prototype.sendPacket = function (type, data, options, callback) { + if ('function' == typeof options) { + callback = options; + options = null; + } + + options = options || {}; + options.compress = false !== options.compress; + if ('closing' != this.readyState) { debug('sending packet "%s" (%s)', type, data); - var packet = { type: type }; + var packet = { + type: type, + options: options + }; if (data) packet.data = data; // exports packetCreate event diff --git a/lib/transports/polling-jsonp.js b/lib/transports/polling-jsonp.js index e9e416bce..ede1a0397 100644 --- a/lib/transports/polling-jsonp.js +++ b/lib/transports/polling-jsonp.js @@ -60,7 +60,7 @@ JSONP.prototype.onData = function (data) { * @api private */ -JSONP.prototype.doWrite = function (data) { +JSONP.prototype.doWrite = function (data, options, callback) { // we must output valid javascript, not valid json // see: http://timelessrepo.com/json-isnt-a-javascript-subset var js = JSON.stringify(data) @@ -70,21 +70,7 @@ JSONP.prototype.doWrite = function (data) { // prepare response data = this.head + js + this.foot; - // explicit UTF-8 is required for pages not served under utf - var headers = { - 'Content-Type': 'text/javascript; charset=UTF-8', - 'Content-Length': Buffer.byteLength(data) - }; - - // prevent XSS warnings on IE - // https://github.com/LearnBoost/socket.io/pull/1333 - var ua = this.req.headers['user-agent']; - if (ua && (~ua.indexOf(';MSIE') || ~ua.indexOf('Trident/'))) { - headers['X-XSS-Protection'] = '0'; - } - - this.res.writeHead(200, this.headers(this.req, headers)); - this.res.end(data); + Polling.prototype.doWrite.call(this, data, options, callback); }; /** diff --git a/lib/transports/polling-xhr.js b/lib/transports/polling-xhr.js index 2a8293c31..5784b0af6 100644 --- a/lib/transports/polling-xhr.js +++ b/lib/transports/polling-xhr.js @@ -48,36 +48,6 @@ XHR.prototype.onRequest = function (req) { } }; -/** - * Frames data prior to write. - * - * @api private - */ - -XHR.prototype.doWrite = function(data){ - // explicit UTF-8 is required for pages not served under utf - var isString = typeof data == 'string'; - var contentType = isString - ? 'text/plain; charset=UTF-8' - : 'application/octet-stream'; - var contentLength = '' + (isString ? Buffer.byteLength(data) : data.length); - - var headers = { - 'Content-Type': contentType, - 'Content-Length': contentLength - }; - - // prevent XSS warnings on IE - // https://github.com/LearnBoost/socket.io/pull/1333 - var ua = this.req.headers['user-agent']; - if (ua && (~ua.indexOf(';MSIE') || ~ua.indexOf('Trident/'))) { - headers['X-XSS-Protection'] = '0'; - } - - this.res.writeHead(200, this.headers(this.req, headers)); - this.res.end(data); -}; - /** * Returns headers for a response. * diff --git a/lib/transports/polling.js b/lib/transports/polling.js index 1c5f1f6c5..ee504e9f1 100644 --- a/lib/transports/polling.js +++ b/lib/transports/polling.js @@ -5,8 +5,15 @@ var Transport = require('../transport') , parser = require('engine.io-parser') + , zlib = require('zlib') + , accepts = require('accepts') , debug = require('debug')('engine:polling'); +var compressionMethods = { + gzip: zlib.createGzip, + deflate: zlib.createDeflate +}; + /** * Exports the constructor. */ @@ -99,7 +106,7 @@ Polling.prototype.onPollRequest = function (req, res) { // if we're still writable but had a pending close, trigger an empty send if (this.writable && this.shouldClose) { debug('triggering empty send to append close packet'); - this.send([{ type: 'noop' }]); + this.send([{ type: 'noop', options: { compress: true } }]); } }; @@ -215,7 +222,7 @@ Polling.prototype.onData = function (data) { Polling.prototype.onClose = function () { if (this.writable) { // close pending poll request - this.send([{ type: 'noop' }]); + this.send([{ type: 'noop', options: { compress: true } }]); } Transport.prototype.onClose.call(this); }; @@ -228,16 +235,21 @@ Polling.prototype.onClose = function () { */ Polling.prototype.send = function (packets) { + this.writable = false; + if (this.shouldClose) { debug('appending close packet to payload'); - packets.push({ type: 'close' }); + packets.push({ type: 'close', options: { compress: true } }); this.shouldClose(); this.shouldClose = null; } var self = this; parser.encodePayload(packets, this.supportsBinary, function(data) { - self.write(data); + var compress = packets.some(function(packet) { + return packet.options && packet.options.compress; + }); + self.write(data, { compress: compress }); }); }; @@ -245,14 +257,103 @@ Polling.prototype.send = function (packets) { * Writes data as response to poll request. * * @param {String} data + * @param {Object} options * @api private */ -Polling.prototype.write = function (data) { +Polling.prototype.write = function (data, options) { debug('writing "%s"', data); - this.doWrite(data); - this.req.cleanup(); - this.writable = false; + var self = this; + this.doWrite(data, options, function() { + self.req.cleanup(); + }); +}; + +/** + * Performs the write. + * + * @api private + */ + +Polling.prototype.doWrite = function (data, options, callback) { + var self = this; + + // explicit UTF-8 is required for pages not served under utf + var isString = typeof data == 'string'; + var contentType = isString + ? 'text/plain; charset=UTF-8' + : 'application/octet-stream'; + + var headers = { + 'Content-Type': contentType + }; + + // prevent XSS warnings on IE + // https://github.com/LearnBoost/socket.io/pull/1333 + var ua = this.req.headers['user-agent']; + if (ua && (~ua.indexOf(';MSIE') || ~ua.indexOf('Trident/'))) { + headers['X-XSS-Protection'] = '0'; + } + + if (!this.httpCompression || !options.compress) { + respond(data); + return; + } + + var len = isString ? Buffer.byteLength(data) : data.length; + if (len < this.httpCompression.threshold) { + respond(data); + return; + } + + var encoding = accepts(this.req).encodings(['gzip', 'deflate']); + if (!encoding) { + respond(data); + return; + } + + this.compress(data, encoding, function(err, data) { + if (err) { + self.res.writeHead(500); + self.res.end(); + callback(err); + return; + } + + headers['Content-Encoding'] = encoding; + respond(data); + }); + + function respond(data) { + headers['Content-Length'] = 'string' == typeof data ? Buffer.byteLength(data) : data.length; + self.res.writeHead(200, self.headers(self.req, headers)); + self.res.end(data); + callback(); + } +}; + +/** + * Comparesses data. + * + * @api private + */ + +Polling.prototype.compress = function (data, encoding, callback) { + debug('compressing'); + + var buffers = []; + var nread = 0; + + compressionMethods[encoding](this.httpCompression) + .on('error', callback) + .on('data', function(chunk) { + buffers.push(chunk); + nread += chunk.length; + }) + .on('end', function() { + callback(null, Buffer.concat(buffers, nread)); + }) + .end(data); }; /** @@ -272,7 +373,7 @@ Polling.prototype.doClose = function (fn) { if (this.writable) { debug('transport writable - closing right away'); - this.send([{ type: 'close' }]); + this.send([{ type: 'close', options: { compress: true } }]); fn(); } else { debug('transport not writable - buffering orderly close'); diff --git a/lib/transports/websocket.js b/lib/transports/websocket.js index 59b8eb7b1..7c0ad34e5 100644 --- a/lib/transports/websocket.js +++ b/lib/transports/websocket.js @@ -85,10 +85,11 @@ WebSocket.prototype.onData = function (data) { WebSocket.prototype.send = function (packets) { var self = this; for (var i = 0, l = packets.length; i < l; i++) { - parser.encodePacket(packets[i], this.supportsBinary, function(data) { + var packet = packets[i]; + parser.encodePacket(packet, this.supportsBinary, function(data) { debug('writing "%s"', data); self.writable = false; - self.socket.send(data, function (err){ + self.socket.send(data, packet.options, function (err){ if (err) return self.onError('write error', err.stack); self.writable = true; self.emit('drain'); diff --git a/package.json b/package.json index 9c934f2be..a1bee74e1 100644 --- a/package.json +++ b/package.json @@ -17,9 +17,10 @@ ], "dependencies": { "debug": "1.0.3", - "ws": "0.5.0", + "ws": "0.6.5", "engine.io-parser": "1.1.0", - "base64id": "0.1.0" + "base64id": "0.1.0", + "accepts": "1.1.4" }, "devDependencies": { "mocha": "1.12.0", diff --git a/test/server.js b/test/server.js index e6d8356e1..da6a1b83b 100644 --- a/test/server.js +++ b/test/server.js @@ -6,6 +6,7 @@ var http = require('http'); var https = require('https'); var fs = require('fs'); +var zlib = require('zlib'); var eio = require('..'); var eioc = require('engine.io-client'); var listen = require('./common').listen; @@ -1986,4 +1987,161 @@ describe('server', function () { }); }); + describe('http compression', function () { + it('should compress by default', function (done) { + var engine = listen({ transports: ['polling'] }, function (port) { + engine.on('connection', function (conn) { + var buf = new Buffer(1024); + for (var i = 0; i < buf.length; i++) buf[i] = i % 0xff; + conn.send(buf); + }); + + http.get({ + port: port, + path: '/engine.io/default/?transport=polling' + }, function(res) { + var sid = res.headers['set-cookie'][0].split('=')[1]; + http.get({ + port: port, + path: '/engine.io/default/?transport=polling&sid=' + sid, + headers: { 'Accept-Encoding': 'gzip, deflate' } + }, function(res) { + expect(res.headers['content-encoding']).to.equal('gzip'); + res.pipe(zlib.createGunzip()) + .on('error', done) + .on('end', done) + .resume(); + }); + }); + }); + }); + + it('should compress using deflate', function (done) { + var engine = listen({ transports: ['polling'] }, function (port) { + engine.on('connection', function (conn) { + var buf = new Buffer(1024); + for (var i = 0; i < buf.length; i++) buf[i] = i % 0xff; + conn.send(buf); + }); + + http.get({ + port: port, + path: '/engine.io/default/?transport=polling' + }, function(res) { + var sid = res.headers['set-cookie'][0].split('=')[1]; + http.get({ + port: port, + path: '/engine.io/default/?transport=polling&sid=' + sid, + headers: { 'Accept-Encoding': 'deflate' } + }, function(res) { + expect(res.headers['content-encoding']).to.equal('deflate'); + res.pipe(zlib.createDeflate()) + .on('error', done) + .on('end', done) + .resume(); + }); + }); + }); + }); + + it('should set threshold', function (done) { + var engine = listen({ transports: ['polling'], httpCompression: { threshold: 0 } }, function (port) { + engine.on('connection', function (conn) { + var buf = new Buffer(10); + for (var i = 0; i < buf.length; i++) buf[i] = i % 0xff; + conn.send(buf); + }); + + http.get({ + port: port, + path: '/engine.io/default/?transport=polling' + }, function(res) { + var sid = res.headers['set-cookie'][0].split('=')[1]; + http.get({ + port: port, + path: '/engine.io/default/?transport=polling&sid=' + sid, + headers: { 'Accept-Encoding': 'gzip, deflate' } + }, function(res) { + expect(res.headers['content-encoding']).to.equal('gzip'); + done(); + }); + }); + }); + }); + + it('should disable compression', function (done) { + var engine = listen({ transports: ['polling'], httpCompression: false }, function (port) { + engine.on('connection', function (conn) { + var buf = new Buffer(1024); + for (var i = 0; i < buf.length; i++) buf[i] = i % 0xff; + conn.send(buf); + }); + + http.get({ + port: port, + path: '/engine.io/default/?transport=polling' + }, function(res) { + var sid = res.headers['set-cookie'][0].split('=')[1]; + http.get({ + port: port, + path: '/engine.io/default/?transport=polling&sid=' + sid, + headers: { 'Accept-Encoding': 'gzip, deflate' } + }, function(res) { + expect(res.headers['content-encoding']).to.be(undefined); + done(); + }); + }); + }); + }); + + it('should disable compression per message', function (done) { + var engine = listen({ transports: ['polling'] }, function (port) { + engine.on('connection', function (conn) { + var buf = new Buffer(1024); + for (var i = 0; i < buf.length; i++) buf[i] = i % 0xff; + conn.send(buf, { compress: false }); + }); + + http.get({ + port: port, + path: '/engine.io/default/?transport=polling' + }, function(res) { + var sid = res.headers['set-cookie'][0].split('=')[1]; + http.get({ + port: port, + path: '/engine.io/default/?transport=polling&sid=' + sid, + headers: { 'Accept-Encoding': 'gzip, deflate' } + }, function(res) { + expect(res.headers['content-encoding']).to.be(undefined); + done(); + }); + }); + }); + }); + + it('should not compress when the byte size is below threshold', function (done) { + var engine = listen({ transports: ['polling'] }, function (port) { + engine.on('connection', function (conn) { + var buf = new Buffer(100); + for (var i = 0; i < buf.length; i++) buf[i] = i % 0xff; + conn.send(buf); + }); + + http.get({ + port: port, + path: '/engine.io/default/?transport=polling' + }, function(res) { + var sid = res.headers['set-cookie'][0].split('=')[1]; + http.get({ + port: port, + path: '/engine.io/default/?transport=polling&sid=' + sid, + headers: { 'Accept-Encoding': 'gzip, deflate' } + }, function(res) { + expect(res.headers['content-encoding']).to.be(undefined); + done(); + }); + }); + }); + }); + }); });