/*! * ws: a node.js websocket client * Copyright(c) 2011 Einar Otto Stangvik * MIT Licensed */ var util = require('util') , events = require('events') , http = require('http') , https = require('https') , crypto = require('crypto') , url = require('url') , fs = require('fs') , Options = require('options') , Sender = require('./Sender') , Receiver = require('./Receiver') , SenderHixie = require('./Sender.hixie') , ReceiverHixie = require('./Receiver.hixie'); /** * Constants */ // Default protocol version var protocolVersion = 13; // Close timeout var closeTimeout = 30000; // Allow 5 seconds to terminate the connection cleanly /** * Node version 0.4 and 0.6 compatibility */ var isNodeV4 = /^v0\.4/.test(process.version); /** * WebSocket implementation */ function WebSocket(address, protocols, options) { if (protocols && !Array.isArray(protocols) && 'object' == typeof protocols) { // accept the "options" Object as the 2nd argument options = protocols; protocols = null; } if ('string' == typeof protocols) { protocols = [ protocols ]; } if (!Array.isArray(protocols)) { protocols = []; } // TODO: actually handle the `Sub-Protocols` part of the WebSocket client this._socket = null; this.bytesReceived = 0; this.readyState = null; this.supports = {}; if (Array.isArray(address)) { initAsServerClient.apply(this, address.concat(options)); } else { initAsClient.apply(this, [address, protocols, options]); } } /** * Inherits from EventEmitter. */ util.inherits(WebSocket, events.EventEmitter); /** * Ready States */ ["CONNECTING", "OPEN", "CLOSING", "CLOSED"].forEach(function (state, index) { WebSocket.prototype[state] = WebSocket[state] = index; }); /** * Gracefully closes the connection, after sending a description message to the server * * @param {Object} data to be sent to the server * @api public */ WebSocket.prototype.close = function(code, data) { if (this.readyState == WebSocket.CLOSING || this.readyState == WebSocket.CLOSED) return; if (this.readyState == WebSocket.CONNECTING) { this.readyState = WebSocket.CLOSED; return; } try { this.readyState = WebSocket.CLOSING; this._closeCode = code; this._closeMessage = data; var mask = !this._isServer; this._sender.close(code, data, mask); } catch (e) { this.emit('error', e); } finally { this.terminate(); } } /** * Pause the client stream * * @api public */ WebSocket.prototype.pause = function() { if (this.readyState != WebSocket.OPEN) throw new Error('not opened'); return this._socket.pause(); } /** * Sends a ping * * @param {Object} data to be sent to the server * @param {Object} Members - mask: boolean, binary: boolean * @param {boolean} dontFailWhenClosed indicates whether or not to throw if the connection isnt open * @api public */ WebSocket.prototype.ping = function(data, options, dontFailWhenClosed) { if (this.readyState != WebSocket.OPEN) { if (dontFailWhenClosed === true) return; throw new Error('not opened'); } options = options || {}; if (typeof options.mask == 'undefined') options.mask = !this._isServer; this._sender.ping(data, options); } /** * Sends a pong * * @param {Object} data to be sent to the server * @param {Object} Members - mask: boolean, binary: boolean * @param {boolean} dontFailWhenClosed indicates whether or not to throw if the connection isnt open * @api public */ WebSocket.prototype.pong = function(data, options, dontFailWhenClosed) { if (this.readyState != WebSocket.OPEN) { if (dontFailWhenClosed === true) return; throw new Error('not opened'); } options = options || {}; if (typeof options.mask == 'undefined') options.mask = !this._isServer; this._sender.pong(data, options); } /** * Resume the client stream * * @api public */ WebSocket.prototype.resume = function() { if (this.readyState != WebSocket.OPEN) throw new Error('not opened'); return this._socket.resume(); } /** * Sends a piece of data * * @param {Object} data to be sent to the server * @param {Object} Members - mask: boolean, binary: boolean * @param {function} Optional callback which is executed after the send completes * @api public */ WebSocket.prototype.send = function(data, options, cb) { if (typeof options == 'function') { cb = options; options = {}; } if (this.readyState != WebSocket.OPEN) { if (typeof cb == 'function') cb(new Error('not opened')); else throw new Error('not opened'); return; } if (!data) data = ''; if (this._queue) { var self = this; this._queue.push(function() { self.send(data, options, cb); }); return; } options = options || {}; options.fin = true; if (typeof options.binary == 'undefined') { options.binary = (data instanceof ArrayBuffer || data instanceof Buffer || data instanceof Uint8Array || data instanceof Uint16Array || data instanceof Uint32Array || data instanceof Int8Array || data instanceof Int16Array || data instanceof Int32Array || data instanceof Float32Array || data instanceof Float64Array); } if (typeof options.mask == 'undefined') options.mask = !this._isServer; if (data instanceof fs.ReadStream) { startQueue(this); var self = this; sendStream(this, data, options, function(error) { process.nextTick(function() { executeQueueSends(self); }); if (typeof cb == 'function') cb(error); }); } else this._sender.send(data, options, cb); } /** * Streams data through calls to a user supplied function * * @param {Object} Members - mask: boolean, binary: boolean * @param {function} 'function (error, send)' which is executed on successive ticks of which send is 'function (data, final)'. * @api public */ WebSocket.prototype.stream = function(options, cb) { if (typeof options == 'function') { cb = options; options = {}; } var self = this; if (typeof cb != 'function') throw new Error('callback must be provided'); if (this.readyState != WebSocket.OPEN) { if (typeof cb == 'function') cb(new Error('not opened')); else throw new Error('not opened'); return; } if (this._queue) { this._queue.push(function() { self.stream(options, cb); }); return; } options = options || {}; if (typeof options.mask == 'undefined') options.mask = !this._isServer; startQueue(this); var send = function(data, final) { try { if (self.readyState != WebSocket.OPEN) throw new Error('not opened'); options.fin = final === true; self._sender.send(data, options); if (!final) process.nextTick(cb.bind(null, null, send)); else executeQueueSends(self); } catch (e) { if (typeof cb == 'function') cb(e); else { delete self._queue; self.emit('error', e); } } } process.nextTick(cb.bind(null, null, send)); } /** * Immediately shuts down the connection * * @api public */ WebSocket.prototype.terminate = function() { if (this.readyState == WebSocket.CLOSED) return; if (this._socket) { try { // End the connection this._socket.end(); } catch (e) { // Socket error during end() call, so just destroy it right now cleanupWebsocketResources.call(this, true); return; } // Add a timeout to ensure that the connection is completely // cleaned up within 30 seconds, even if the clean close procedure // fails for whatever reason this._closeTimer = setTimeout(cleanupWebsocketResources.bind(this, true), closeTimeout); } else if (this.readyState == WebSocket.CONNECTING) { cleanupWebsocketResources.call(this, true); } }; /** * Expose bufferedAmount * * @api public */ Object.defineProperty(WebSocket.prototype, 'bufferedAmount', { get: function get() { var amount = 0; if (this._socket) { amount = this._socket.bufferSize || 0; } return amount; } }); /** * Emulates the W3C Browser based WebSocket interface using function members. * * @see http://dev.w3.org/html5/websockets/#the-websocket-interface * @api public */ ['open', 'error', 'close', 'message'].forEach(function(method) { Object.defineProperty(WebSocket.prototype, 'on' + method, { /** * Returns the current listener * * @returns {Mixed} the set function or undefined * @api public */ get: function get() { var listener = this.listeners(method)[0]; return listener ? (listener._listener ? listener._listener : listener) : undefined; }, /** * Start listening for events * * @param {Function} listener the listener * @returns {Mixed} the set function or undefined * @api public */ set: function set(listener) { this.removeAllListeners(method); this.addEventListener(method, listener); } }); }); /** * Emulates the W3C Browser based WebSocket interface using addEventListener. * * @see https://developer.mozilla.org/en/DOM/element.addEventListener * @see http://dev.w3.org/html5/websockets/#the-websocket-interface * @api public */ WebSocket.prototype.addEventListener = function(method, listener) { var target = this; if (typeof listener === 'function') { if (method === 'message') { function onMessage (data, flags) { listener.call(this, new MessageEvent(data, flags.binary ? 'Binary' : 'Text', target)); } // store a reference so we can return the original function from the addEventListener hook onMessage._listener = listener; this.on(method, onMessage); } else if (method === 'close') { function onClose (code, message) { listener.call(this, new CloseEvent(code, message, target)); } // store a reference so we can return the original function from the addEventListener hook onClose._listener = listener; this.on(method, onClose); } else if (method === 'error') { function onError (event) { event.target = target; listener.call(this, event); } // store a reference so we can return the original function from the addEventListener hook onError._listener = listener; this.on(method, onError); } else if (method === 'open') { function onOpen () { listener.call(this, new OpenEvent(target)); } // store a reference so we can return the original function from the addEventListener hook onOpen._listener = listener; this.on(method, onOpen); } else { this.on(method, listener); } } } module.exports = WebSocket; /** * W3C MessageEvent * * @see http://www.w3.org/TR/html5/comms.html * @api private */ function MessageEvent(dataArg, typeArg, target) { this.data = dataArg; this.type = typeArg; this.target = target; } /** * W3C CloseEvent * * @see http://www.w3.org/TR/html5/comms.html * @api private */ function CloseEvent(code, reason, target) { this.wasClean = (typeof code == 'undefined' || code == 1000); this.code = code; this.reason = reason; this.target = target; } /** * W3C OpenEvent * * @see http://www.w3.org/TR/html5/comms.html * @api private */ function OpenEvent(target) { this.target = target; } /** * Entirely private apis, * which may or may not be bound to a sepcific WebSocket instance. */ function initAsServerClient(req, socket, upgradeHead, options) { options = new Options({ protocolVersion: protocolVersion, protocol: null }).merge(options); // expose state properties this.protocol = options.value.protocol; this.protocolVersion = options.value.protocolVersion; this.supports.binary = (this.protocolVersion != 'hixie-76'); this.upgradeReq = req; this.readyState = WebSocket.CONNECTING; this._isServer = true; // establish connection if (options.value.protocolVersion == 'hixie-76') establishConnection.call(this, ReceiverHixie, SenderHixie, socket, upgradeHead); else establishConnection.call(this, Receiver, Sender, socket, upgradeHead); } function initAsClient(address, protocols, options) { options = new Options({ origin: null, protocolVersion: protocolVersion, host: null, headers: null, protocol: null, agent: null, // ssl-related options pfx: null, key: null, passphrase: null, cert: null, ca: null, ciphers: null, rejectUnauthorized: null }).merge(options); if (options.value.protocolVersion != 8 && options.value.protocolVersion != 13) { throw new Error('unsupported protocol version'); } // verify url and establish http class var serverUrl = url.parse(address); var isUnixSocket = serverUrl.protocol === 'ws+unix:'; if (!serverUrl.host && !isUnixSocket) throw new Error('invalid url'); var isSecure = serverUrl.protocol === 'wss:' || serverUrl.protocol === 'https:'; var httpObj = isSecure ? https : http; var port = serverUrl.port || (isSecure ? 443 : 80); var auth = serverUrl.auth; // expose state properties this._isServer = false; this.url = address; this.protocolVersion = options.value.protocolVersion; this.supports.binary = (this.protocolVersion != 'hixie-76'); // begin handshake var key = new Buffer(options.value.protocolVersion + '-' + Date.now()).toString('base64'); var shasum = crypto.createHash('sha1'); shasum.update(key + '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'); var expectedServerKey = shasum.digest('base64'); var agent = options.value.agent; // node<=v0.4.x compatibility if (!agent && isNodeV4) { isNodeV4 = true; agent = new httpObj.Agent({ host: serverUrl.hostname, port: port }); } var headerHost = serverUrl.hostname; // Append port number to Host and Origin header, only if specified in the url and non-default if(serverUrl.port) { if((isSecure && (port != 443)) || (!isSecure && (port != 80))){ headerHost = headerHost + ':' + port; } } var requestOptions = { port: port, host: serverUrl.hostname, headers: { 'Connection': 'Upgrade', 'Upgrade': 'websocket', 'Host': headerHost, 'Origin': headerHost, 'Sec-WebSocket-Version': options.value.protocolVersion, 'Sec-WebSocket-Key': key } }; // If we have basic auth. if (auth) { requestOptions.headers['Authorization'] = 'Basic ' + new Buffer(auth).toString('base64'); } if (options.value.protocol) { requestOptions.headers['Sec-WebSocket-Protocol'] = options.value.protocol; } if (options.value.host) { requestOptions.headers['Host'] = options.value.host; } if (options.value.headers) { for (var header in options.value.headers) { if (options.value.headers.hasOwnProperty(header)) { requestOptions.headers[header] = options.value.headers[header]; } } } if (options.isDefinedAndNonNull('pfx') || options.isDefinedAndNonNull('key') || options.isDefinedAndNonNull('passphrase') || options.isDefinedAndNonNull('cert') || options.isDefinedAndNonNull('ca') || options.isDefinedAndNonNull('ciphers') || options.isDefinedAndNonNull('rejectUnauthorized')) { if (isNodeV4) { throw new Error('Client side certificates are not supported on Node 0.4.x'); } if (options.isDefinedAndNonNull('pfx')) requestOptions.pfx = options.value.pfx; if (options.isDefinedAndNonNull('key')) requestOptions.key = options.value.key; if (options.isDefinedAndNonNull('passphrase')) requestOptions.passphrase = options.value.passphrase; if (options.isDefinedAndNonNull('cert')) requestOptions.cert = options.value.cert; if (options.isDefinedAndNonNull('ca')) requestOptions.ca = options.value.ca; if (options.isDefinedAndNonNull('ciphers')) requestOptions.ciphers = options.value.ciphers; if (options.isDefinedAndNonNull('rejectUnauthorized')) requestOptions.rejectUnauthorized = options.value.rejectUnauthorized; if (!agent) { // global agent ignores client side certificates agent = new httpObj.Agent(requestOptions); } } if (isNodeV4) { requestOptions.path = (serverUrl.pathname || '/') + (serverUrl.search || ''); } else requestOptions.path = serverUrl.path || '/'; if (agent) { requestOptions.agent = agent; } if (isUnixSocket) { requestOptions.socketPath = serverUrl.pathname; } if (options.value.origin) { if (options.value.protocolVersion < 13) requestOptions.headers['Sec-WebSocket-Origin'] = options.value.origin; else requestOptions.headers['Origin'] = options.value.origin; } var self = this; var req = httpObj.request(requestOptions); (isNodeV4 ? agent : req).on('error', function(error) { self.emit('error', error); cleanupWebsocketResources.call(this, error); }); (isNodeV4 ? agent : req).once('response', function(res) { var error = new Error('unexpected server response (' + res.statusCode + ')'); self.emit('error', error); cleanupWebsocketResources.call(this, error); }); (isNodeV4 ? agent : req).once('upgrade', function(res, socket, upgradeHead) { if (self.readyState == WebSocket.CLOSED) { // client closed before server accepted connection self.emit('close'); removeAllListeners(self); socket.end(); return; } var serverKey = res.headers['sec-websocket-accept']; if (typeof serverKey == 'undefined' || serverKey !== expectedServerKey) { self.emit('error', 'invalid server key'); removeAllListeners(self); socket.end(); return; } var serverProt = res.headers['sec-websocket-protocol']; var protList = (options.value.protocol || "").split(/, */); var protError = null; if (!options.value.protocol && serverProt) { protError = 'server sent a subprotocol even though none requested'; } else if (options.value.protocol && !serverProt) { protError = 'server sent no subprotocol even though requested'; } else if (serverProt && protList.indexOf(serverProt) === -1) { protError = 'server responded with an invalid protocol'; } if (protError) { self.emit('error', protError); removeAllListeners(self); socket.end(); return; } else if (serverProt) { self.protocol = serverProt; } establishConnection.call(self, Receiver, Sender, socket, upgradeHead); // perform cleanup on http resources removeAllListeners(isNodeV4 ? agent : req); req = null; agent = null; }); req.end(); this.readyState = WebSocket.CONNECTING; } function establishConnection(ReceiverClass, SenderClass, socket, upgradeHead) { this._socket = socket; socket.setTimeout(0); socket.setNoDelay(true); var self = this; this._receiver = new ReceiverClass(); // socket cleanup handlers socket.on('end', cleanupWebsocketResources.bind(this)); socket.on('close', cleanupWebsocketResources.bind(this)); socket.on('error', cleanupWebsocketResources.bind(this)); // ensure that the upgradeHead is added to the receiver function firstHandler(data) { if (self.readyState != WebSocket.OPEN) return; if (upgradeHead && upgradeHead.length > 0) { self.bytesReceived += upgradeHead.length; var head = upgradeHead; upgradeHead = null; self._receiver.add(head); } dataHandler = realHandler; if (data) { self.bytesReceived += data.length; self._receiver.add(data); } } // subsequent packets are pushed straight to the receiver function realHandler(data) { if (data) self.bytesReceived += data.length; self._receiver.add(data); } var dataHandler = firstHandler; // if data was passed along with the http upgrade, // this will schedule a push of that on to the receiver. // this has to be done on next tick, since the caller // hasn't had a chance to set event handlers on this client // object yet. process.nextTick(firstHandler); // receiver event handlers self._receiver.ontext = function (data, flags) { flags = flags || {}; self.emit('message', data, flags); }; self._receiver.onbinary = function (data, flags) { flags = flags || {}; flags.binary = true; self.emit('message', data, flags); }; self._receiver.onping = function(data, flags) { flags = flags || {}; self.pong(data, {mask: !self._isServer, binary: flags.binary === true}, true); self.emit('ping', data, flags); }; self._receiver.onpong = function(data, flags) { self.emit('pong', data, flags); }; self._receiver.onclose = function(code, data, flags) { flags = flags || {}; self.close(code, data); }; self._receiver.onerror = function(reason, errorCode) { // close the connection when the receiver reports a HyBi error code self.close(typeof errorCode != 'undefined' ? errorCode : 1002, ''); self.emit('error', reason, errorCode); }; // finalize the client this._sender = new SenderClass(socket); this._sender.on('error', function(error) { self.close(1002, ''); self.emit('error', error); }); this.readyState = WebSocket.OPEN; this.emit('open'); socket.on('data', dataHandler); } function startQueue(instance) { instance._queue = instance._queue || []; } function executeQueueSends(instance) { var queue = instance._queue; if (typeof queue == 'undefined') return; delete instance._queue; for (var i = 0, l = queue.length; i < l; ++i) { queue[i](); } } function sendStream(instance, stream, options, cb) { stream.on('data', function(data) { if (instance.readyState != WebSocket.OPEN) { if (typeof cb == 'function') cb(new Error('not opened')); else { delete instance._queue; instance.emit('error', new Error('not opened')); } return; } options.fin = false; instance._sender.send(data, options); }); stream.on('end', function() { if (instance.readyState != WebSocket.OPEN) { if (typeof cb == 'function') cb(new Error('not opened')); else { delete instance._queue; instance.emit('error', new Error('not opened')); } return; } options.fin = true; instance._sender.send(null, options); if (typeof cb == 'function') cb(null); }); } function cleanupWebsocketResources(error) { if (this.readyState == WebSocket.CLOSED) return; var emitClose = this.readyState != WebSocket.CONNECTING; this.readyState = WebSocket.CLOSED; clearTimeout(this._closeTimer); this._closeTimer = null; if (emitClose) this.emit('close', this._closeCode || 1000, this._closeMessage || ''); if (this._socket) { removeAllListeners(this._socket); // catch all socket error after removing all standard handlers var socket = this._socket; this._socket.on('error', function() { try { socket.destroy(); } catch (e) {} }); try { if (!error) this._socket.end(); else this._socket.destroy(); } catch (e) { /* Ignore termination errors */ } this._socket = null; } if (this._sender) { removeAllListeners(this._sender); this._sender = null; } if (this._receiver) { this._receiver.cleanup(); this._receiver = null; } removeAllListeners(this); this.on('error', function() {}); // catch all errors after this delete this._queue; } function removeAllListeners(instance) { if (isNodeV4) { // node v4 doesn't *actually* remove all listeners globally, // so we do that instead instance._events = {}; } else instance.removeAllListeners(); }