/** * Module dependencies. */ var qs = require('querystring') , parse = require('url').parse , readFileSync = require('fs').readFileSync , crypto = require('crypto') , base64id = require('base64id') , transports = require('./transports') , EventEmitter = require('events').EventEmitter , Socket = require('./socket') , WebSocketServer = require('ws').Server , debug = require('debug')('engine'); /** * Module exports. */ module.exports = Server; /** * Server constructor. * * @param {Object} options * @api public */ function Server(opts){ if (!(this instanceof Server)) { return new Server(opts); } this.clients = {}; this.clientsCount = 0; opts = opts || {}; this.pingTimeout = opts.pingTimeout || 60000; this.pingInterval = opts.pingInterval || 25000; this.upgradeTimeout = opts.upgradeTimeout || 10000; this.maxHttpBufferSize = opts.maxHttpBufferSize || 10E7; this.transports = opts.transports || Object.keys(transports); this.allowUpgrades = false !== opts.allowUpgrades; this.allowRequest = opts.allowRequest; this.cookie = false !== opts.cookie ? (opts.cookie || 'io') : false; // initialize websocket server if (~this.transports.indexOf('websocket')) { this.ws = new WebSocketServer({ noServer: true, clientTracking: false }); } } /** * Protocol errors mappings. */ Server.errors = { UNKNOWN_TRANSPORT: 0, UNKNOWN_SID: 1, BAD_HANDSHAKE_METHOD: 2, BAD_REQUEST: 3 }; Server.errorMessages = { 0: 'Transport unknown', 1: 'Session ID unknown', 2: 'Bad handshake method', 3: 'Bad request' }; /** * Inherits from EventEmitter. */ Server.prototype.__proto__ = EventEmitter.prototype; /** * Hash of open clients. * * @api public */ Server.prototype.clients; /** * Returns a list of available transports for upgrade given a certain transport. * * @return {Array} * @api public */ Server.prototype.upgrades = function(transport){ if (!this.allowUpgrades) return []; return transports[transport].upgradesTo || []; }; /** * Verifies a request. * * @param {http.ServerRequest} * @return {Boolean} whether the request is valid * @api private */ Server.prototype.verify = function(req, upgrade, fn){ // transport check var transport = req._query.transport; if (!~this.transports.indexOf(transport)) { debug('unknown transport "%s"', transport); return fn(Server.errors.UNKNOWN_TRANSPORT, false); } // sid check var sid = req._query.sid; if (sid) { if (!this.clients.hasOwnProperty(sid)) return fn(Server.errors.UNKNOWN_SID, false); if (!upgrade && this.clients[sid].transport.name !== transport) { debug('bad request: unexpected transport without upgrade'); return fn(Server.errors.BAD_REQUEST, false); } } else { // handshake is GET only if ('GET' != req.method) return fn(Server.errors.BAD_HANDSHAKE_METHOD, false); if (!this.allowRequest) return fn(null, true); return this.allowRequest(req, fn); } fn(null, true); }; /** * Prepares a request by processing the query string. * * @api private */ Server.prototype.prepare = function(req){ // try to leverage pre-existing `req._query` (e.g: from connect) if (!req._query) { req._query = ~req.url.indexOf('?') ? qs.parse(parse(req.url).query) : {}; } }; /** * Closes all clients. * * @api public */ Server.prototype.close = function(){ debug('closing all open clients'); for (var i in this.clients) { this.clients[i].close(); } return this; }; /** * Handles an Engine.IO HTTP request. * * @param {http.ServerRequest} request * @param {http.ServerResponse|http.OutgoingMessage} response * @api public */ Server.prototype.handleRequest = function(req, res){ debug('handling "%s" http request "%s"', req.method, req.url); this.prepare(req); req.res = res; var self = this; this.verify(req, false, function(err, success) { if (!success) { sendErrorMessage(req, res, err); return; } if (req._query.sid) { debug('setting new request for existing client'); self.clients[req._query.sid].transport.onRequest(req); } else { self.handshake(req._query.transport, req); } }); }; /** * Sends an Engine.IO Error Message * * @param {http.ServerResponse} response * @param {code} error code * @api private */ function sendErrorMessage(req, res, code) { var headers = { 'Content-Type': 'application/json' }; if (req.headers.origin) { headers['Access-Control-Allow-Credentials'] = 'true'; headers['Access-Control-Allow-Origin'] = req.headers.origin; } else { headers['Access-Control-Allow-Origin'] = '*'; } res.writeHead(400, headers); res.end(JSON.stringify({ code: code, message: Server.errorMessages[code] })); } /** * Handshakes a new client. * * @param {String} transport name * @param {Object} request object * @api private */ Server.prototype.handshake = function(transport, req){ var id = base64id.generateId(); debug('handshaking client "%s"', id); var transportName = transport; try { var transport = new transports[transport](req); if ('polling' == transportName) { transport.maxHttpBufferSize = this.maxHttpBufferSize; } if (req._query && req._query.b64) { transport.supportsBinary = false; } else { transport.supportsBinary = true; } } catch (e) { sendErrorMessage(req, req.res, Server.errors.BAD_REQUEST); return; } var socket = new Socket(id, this, transport, req); var self = this; if (false !== this.cookie) { transport.on('headers', function(headers){ headers['Set-Cookie'] = self.cookie + '=' + id; }); } transport.onRequest(req); this.clients[id] = socket; this.clientsCount++; socket.once('close', function(){ delete self.clients[id]; self.clientsCount--; }); this.emit('connection', socket); }; /** * Handles an Engine.IO HTTP Upgrade. * * @api public */ Server.prototype.handleUpgrade = function(req, socket, upgradeHead){ this.prepare(req); var self = this; this.verify(req, true, function(err, success) { if (!success) { socket.end(); return; } var head = new Buffer(upgradeHead.length); upgradeHead.copy(head); upgradeHead = null; // delegate to ws self.ws.handleUpgrade(req, socket, head, function(conn){ self.onWebSocket(req, conn); }); }); }; /** * Called upon a ws.io connection. * * @param {ws.Socket} websocket * @api private */ Server.prototype.onWebSocket = function(req, socket){ if (!transports[req._query.transport].prototype.handlesUpgrades) { debug('transport doesnt handle upgraded requests'); socket.close(); return; } // get client id var id = req._query.sid; // keep a reference to the ws.Socket req.websocket = socket; if (id) { if (!this.clients[id]) { debug('upgrade attempt for closed client'); socket.close(); } else if (this.clients[id].upgraded) { debug('transport had already been upgraded'); socket.close(); } else { debug('upgrading existing transport'); var transport = new transports[req._query.transport](req); if (req._query && req._query.b64) { transport.supportsBinary = false; } else { transport.supportsBinary = true; } this.clients[id].maybeUpgrade(transport); } } else { this.handshake(req._query.transport, req); } }; /** * Captures upgrade requests for a http.Server. * * @param {http.Server} server * @param {Object} options * @api public */ Server.prototype.attach = function(server, options){ var self = this; var options = options || {}; var path = (options.path || '/engine.io').replace(/\/$/, ''); var destroyUpgrade = (options.destroyUpgrade !== undefined) ? options.destroyUpgrade : true; var destroyUpgradeTimeout = options.destroyUpgradeTimeout || 1000; // normalize path path += '/'; function check (req) { return path == req.url.substr(0, path.length); } // cache and clean up listeners var listeners = server.listeners('request').slice(0); server.removeAllListeners('request'); server.on('close', self.close.bind(self)); // add request handler server.on('request', function(req, res){ if (check(req)) { debug('intercepting request for path "%s"', path); self.handleRequest(req, res); } else { for (var i = 0, l = listeners.length; i < l; i++) { listeners[i].call(server, req, res); } } }); if(~self.transports.indexOf('websocket')) { server.on('upgrade', function (req, socket, head) { if (check(req)) { self.handleUpgrade(req, socket, head); } else if (false !== options.destroyUpgrade) { // default node behavior is to disconnect when no handlers // but by adding a handler, we prevent that // and if no eio thing handles the upgrade // then the socket needs to die! setTimeout(function() { if (socket.writable && socket.bytesWritten <= 0) { return socket.end(); } }, options.destroyUpgradeTimeout); } }); } };