You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
572 lines
11 KiB
572 lines
11 KiB
|
|
/**
|
|
* Module dependencies.
|
|
*/
|
|
|
|
var Emitter = require('events').EventEmitter;
|
|
var parser = require('socket.io-parser');
|
|
var hasBin = require('has-binary2');
|
|
var url = require('url');
|
|
var debug = require('debug')('socket.io:socket');
|
|
|
|
/**
|
|
* Module exports.
|
|
*/
|
|
|
|
module.exports = exports = Socket;
|
|
|
|
/**
|
|
* Blacklisted events.
|
|
*
|
|
* @api public
|
|
*/
|
|
|
|
exports.events = [
|
|
'error',
|
|
'connect',
|
|
'disconnect',
|
|
'disconnecting',
|
|
'newListener',
|
|
'removeListener'
|
|
];
|
|
|
|
/**
|
|
* Flags.
|
|
*
|
|
* @api private
|
|
*/
|
|
|
|
var flags = [
|
|
'json',
|
|
'volatile',
|
|
'broadcast',
|
|
'local'
|
|
];
|
|
|
|
/**
|
|
* `EventEmitter#emit` reference.
|
|
*/
|
|
|
|
var emit = Emitter.prototype.emit;
|
|
|
|
/**
|
|
* Interface to a `Client` for a given `Namespace`.
|
|
*
|
|
* @param {Namespace} nsp
|
|
* @param {Client} client
|
|
* @api public
|
|
*/
|
|
|
|
function Socket(nsp, client, query){
|
|
this.nsp = nsp;
|
|
this.server = nsp.server;
|
|
this.adapter = this.nsp.adapter;
|
|
this.id = nsp.name !== '/' ? nsp.name + '#' + client.id : client.id;
|
|
this.client = client;
|
|
this.conn = client.conn;
|
|
this.rooms = {};
|
|
this.acks = {};
|
|
this.connected = true;
|
|
this.disconnected = false;
|
|
this.handshake = this.buildHandshake(query);
|
|
this.fns = [];
|
|
this.flags = {};
|
|
this._rooms = [];
|
|
}
|
|
|
|
/**
|
|
* Inherits from `EventEmitter`.
|
|
*/
|
|
|
|
Socket.prototype.__proto__ = Emitter.prototype;
|
|
|
|
/**
|
|
* Apply flags from `Socket`.
|
|
*/
|
|
|
|
flags.forEach(function(flag){
|
|
Object.defineProperty(Socket.prototype, flag, {
|
|
get: function() {
|
|
this.flags[flag] = true;
|
|
return this;
|
|
}
|
|
});
|
|
});
|
|
|
|
/**
|
|
* `request` engine.io shortcut.
|
|
*
|
|
* @api public
|
|
*/
|
|
|
|
Object.defineProperty(Socket.prototype, 'request', {
|
|
get: function() {
|
|
return this.conn.request;
|
|
}
|
|
});
|
|
|
|
/**
|
|
* Builds the `handshake` BC object
|
|
*
|
|
* @api private
|
|
*/
|
|
|
|
Socket.prototype.buildHandshake = function(query){
|
|
var self = this;
|
|
function buildQuery(){
|
|
var requestQuery = url.parse(self.request.url, true).query;
|
|
//if socket-specific query exist, replace query strings in requestQuery
|
|
return Object.assign({}, query, requestQuery);
|
|
}
|
|
return {
|
|
headers: this.request.headers,
|
|
time: (new Date) + '',
|
|
address: this.conn.remoteAddress,
|
|
xdomain: !!this.request.headers.origin,
|
|
secure: !!this.request.connection.encrypted,
|
|
issued: +(new Date),
|
|
url: this.request.url,
|
|
query: buildQuery()
|
|
};
|
|
};
|
|
|
|
/**
|
|
* Emits to this client.
|
|
*
|
|
* @return {Socket} self
|
|
* @api public
|
|
*/
|
|
|
|
Socket.prototype.emit = function(ev){
|
|
if (~exports.events.indexOf(ev)) {
|
|
emit.apply(this, arguments);
|
|
return this;
|
|
}
|
|
|
|
var args = Array.prototype.slice.call(arguments);
|
|
var packet = {
|
|
type: (this.flags.binary !== undefined ? this.flags.binary : hasBin(args)) ? parser.BINARY_EVENT : parser.EVENT,
|
|
data: args
|
|
};
|
|
|
|
// access last argument to see if it's an ACK callback
|
|
if (typeof args[args.length - 1] === 'function') {
|
|
if (this._rooms.length || this.flags.broadcast) {
|
|
throw new Error('Callbacks are not supported when broadcasting');
|
|
}
|
|
|
|
debug('emitting packet with ack id %d', this.nsp.ids);
|
|
this.acks[this.nsp.ids] = args.pop();
|
|
packet.id = this.nsp.ids++;
|
|
}
|
|
|
|
var rooms = this._rooms.slice(0);
|
|
var flags = Object.assign({}, this.flags);
|
|
|
|
// reset flags
|
|
this._rooms = [];
|
|
this.flags = {};
|
|
|
|
if (rooms.length || flags.broadcast) {
|
|
this.adapter.broadcast(packet, {
|
|
except: [this.id],
|
|
rooms: rooms,
|
|
flags: flags
|
|
});
|
|
} else {
|
|
// dispatch packet
|
|
this.packet(packet, flags);
|
|
}
|
|
return this;
|
|
};
|
|
|
|
/**
|
|
* Targets a room when broadcasting.
|
|
*
|
|
* @param {String} name
|
|
* @return {Socket} self
|
|
* @api public
|
|
*/
|
|
|
|
Socket.prototype.to =
|
|
Socket.prototype.in = function(name){
|
|
if (!~this._rooms.indexOf(name)) this._rooms.push(name);
|
|
return this;
|
|
};
|
|
|
|
/**
|
|
* Sends a `message` event.
|
|
*
|
|
* @return {Socket} self
|
|
* @api public
|
|
*/
|
|
|
|
Socket.prototype.send =
|
|
Socket.prototype.write = function(){
|
|
var args = Array.prototype.slice.call(arguments);
|
|
args.unshift('message');
|
|
this.emit.apply(this, args);
|
|
return this;
|
|
};
|
|
|
|
/**
|
|
* Writes a packet.
|
|
*
|
|
* @param {Object} packet object
|
|
* @param {Object} opts options
|
|
* @api private
|
|
*/
|
|
|
|
Socket.prototype.packet = function(packet, opts){
|
|
packet.nsp = this.nsp.name;
|
|
opts = opts || {};
|
|
opts.compress = false !== opts.compress;
|
|
this.client.packet(packet, opts);
|
|
};
|
|
|
|
/**
|
|
* Joins a room.
|
|
*
|
|
* @param {String|Array} room or array of rooms
|
|
* @param {Function} fn optional, callback
|
|
* @return {Socket} self
|
|
* @api private
|
|
*/
|
|
|
|
Socket.prototype.join = function(rooms, fn){
|
|
debug('joining room %s', rooms);
|
|
var self = this;
|
|
if (!Array.isArray(rooms)) {
|
|
rooms = [rooms];
|
|
}
|
|
rooms = rooms.filter(function (room) {
|
|
return !self.rooms.hasOwnProperty(room);
|
|
});
|
|
if (!rooms.length) {
|
|
fn && fn(null);
|
|
return this;
|
|
}
|
|
this.adapter.addAll(this.id, rooms, function(err){
|
|
if (err) return fn && fn(err);
|
|
debug('joined room %s', rooms);
|
|
rooms.forEach(function (room) {
|
|
self.rooms[room] = room;
|
|
});
|
|
fn && fn(null);
|
|
});
|
|
return this;
|
|
};
|
|
|
|
/**
|
|
* Leaves a room.
|
|
*
|
|
* @param {String} room
|
|
* @param {Function} fn optional, callback
|
|
* @return {Socket} self
|
|
* @api private
|
|
*/
|
|
|
|
Socket.prototype.leave = function(room, fn){
|
|
debug('leave room %s', room);
|
|
var self = this;
|
|
this.adapter.del(this.id, room, function(err){
|
|
if (err) return fn && fn(err);
|
|
debug('left room %s', room);
|
|
delete self.rooms[room];
|
|
fn && fn(null);
|
|
});
|
|
return this;
|
|
};
|
|
|
|
/**
|
|
* Leave all rooms.
|
|
*
|
|
* @api private
|
|
*/
|
|
|
|
Socket.prototype.leaveAll = function(){
|
|
this.adapter.delAll(this.id);
|
|
this.rooms = {};
|
|
};
|
|
|
|
/**
|
|
* Called by `Namespace` upon successful
|
|
* middleware execution (ie: authorization).
|
|
* Socket is added to namespace array before
|
|
* call to join, so adapters can access it.
|
|
*
|
|
* @api private
|
|
*/
|
|
|
|
Socket.prototype.onconnect = function(){
|
|
debug('socket connected - writing packet');
|
|
this.nsp.connected[this.id] = this;
|
|
this.join(this.id);
|
|
var skip = this.nsp.name === '/' && this.nsp.fns.length === 0;
|
|
if (skip) {
|
|
debug('packet already sent in initial handshake');
|
|
} else {
|
|
this.packet({ type: parser.CONNECT });
|
|
}
|
|
};
|
|
|
|
/**
|
|
* Called with each packet. Called by `Client`.
|
|
*
|
|
* @param {Object} packet
|
|
* @api private
|
|
*/
|
|
|
|
Socket.prototype.onpacket = function(packet){
|
|
debug('got packet %j', packet);
|
|
switch (packet.type) {
|
|
case parser.EVENT:
|
|
this.onevent(packet);
|
|
break;
|
|
|
|
case parser.BINARY_EVENT:
|
|
this.onevent(packet);
|
|
break;
|
|
|
|
case parser.ACK:
|
|
this.onack(packet);
|
|
break;
|
|
|
|
case parser.BINARY_ACK:
|
|
this.onack(packet);
|
|
break;
|
|
|
|
case parser.DISCONNECT:
|
|
this.ondisconnect();
|
|
break;
|
|
|
|
case parser.ERROR:
|
|
this.onerror(new Error(packet.data));
|
|
}
|
|
};
|
|
|
|
/**
|
|
* Called upon event packet.
|
|
*
|
|
* @param {Object} packet object
|
|
* @api private
|
|
*/
|
|
|
|
Socket.prototype.onevent = function(packet){
|
|
var args = packet.data || [];
|
|
debug('emitting event %j', args);
|
|
|
|
if (null != packet.id) {
|
|
debug('attaching ack callback to event');
|
|
args.push(this.ack(packet.id));
|
|
}
|
|
|
|
this.dispatch(args);
|
|
};
|
|
|
|
/**
|
|
* Produces an ack callback to emit with an event.
|
|
*
|
|
* @param {Number} id packet id
|
|
* @api private
|
|
*/
|
|
|
|
Socket.prototype.ack = function(id){
|
|
var self = this;
|
|
var sent = false;
|
|
return function(){
|
|
// prevent double callbacks
|
|
if (sent) return;
|
|
var args = Array.prototype.slice.call(arguments);
|
|
debug('sending ack %j', args);
|
|
|
|
self.packet({
|
|
id: id,
|
|
type: hasBin(args) ? parser.BINARY_ACK : parser.ACK,
|
|
data: args
|
|
});
|
|
|
|
sent = true;
|
|
};
|
|
};
|
|
|
|
/**
|
|
* Called upon ack packet.
|
|
*
|
|
* @api private
|
|
*/
|
|
|
|
Socket.prototype.onack = function(packet){
|
|
var ack = this.acks[packet.id];
|
|
if ('function' == typeof ack) {
|
|
debug('calling ack %s with %j', packet.id, packet.data);
|
|
ack.apply(this, packet.data);
|
|
delete this.acks[packet.id];
|
|
} else {
|
|
debug('bad ack %s', packet.id);
|
|
}
|
|
};
|
|
|
|
/**
|
|
* Called upon client disconnect packet.
|
|
*
|
|
* @api private
|
|
*/
|
|
|
|
Socket.prototype.ondisconnect = function(){
|
|
debug('got disconnect packet');
|
|
this.onclose('client namespace disconnect');
|
|
};
|
|
|
|
/**
|
|
* Handles a client error.
|
|
*
|
|
* @api private
|
|
*/
|
|
|
|
Socket.prototype.onerror = function(err){
|
|
if (this.listeners('error').length) {
|
|
this.emit('error', err);
|
|
} else {
|
|
console.error('Missing error handler on `socket`.');
|
|
console.error(err.stack);
|
|
}
|
|
};
|
|
|
|
/**
|
|
* Called upon closing. Called by `Client`.
|
|
*
|
|
* @param {String} reason
|
|
* @throw {Error} optional error object
|
|
* @api private
|
|
*/
|
|
|
|
Socket.prototype.onclose = function(reason){
|
|
if (!this.connected) return this;
|
|
debug('closing socket - reason %s', reason);
|
|
this.emit('disconnecting', reason);
|
|
this.leaveAll();
|
|
this.nsp.remove(this);
|
|
this.client.remove(this);
|
|
this.connected = false;
|
|
this.disconnected = true;
|
|
delete this.nsp.connected[this.id];
|
|
this.emit('disconnect', reason);
|
|
};
|
|
|
|
/**
|
|
* Produces an `error` packet.
|
|
*
|
|
* @param {Object} err error object
|
|
* @api private
|
|
*/
|
|
|
|
Socket.prototype.error = function(err){
|
|
this.packet({ type: parser.ERROR, data: err });
|
|
};
|
|
|
|
/**
|
|
* Disconnects this client.
|
|
*
|
|
* @param {Boolean} close if `true`, closes the underlying connection
|
|
* @return {Socket} self
|
|
* @api public
|
|
*/
|
|
|
|
Socket.prototype.disconnect = function(close){
|
|
if (!this.connected) return this;
|
|
if (close) {
|
|
this.client.disconnect();
|
|
} else {
|
|
this.packet({ type: parser.DISCONNECT });
|
|
this.onclose('server namespace disconnect');
|
|
}
|
|
return this;
|
|
};
|
|
|
|
/**
|
|
* Sets the compress flag.
|
|
*
|
|
* @param {Boolean} compress if `true`, compresses the sending data
|
|
* @return {Socket} self
|
|
* @api public
|
|
*/
|
|
|
|
Socket.prototype.compress = function(compress){
|
|
this.flags.compress = compress;
|
|
return this;
|
|
};
|
|
|
|
/**
|
|
* Sets the binary flag
|
|
*
|
|
* @param {Boolean} Encode as if it has binary data if `true`, Encode as if it doesnt have binary data if `false`
|
|
* @return {Socket} self
|
|
* @api public
|
|
*/
|
|
|
|
Socket.prototype.binary = function (binary) {
|
|
this.flags.binary = binary;
|
|
return this;
|
|
};
|
|
|
|
/**
|
|
* Dispatch incoming event to socket listeners.
|
|
*
|
|
* @param {Array} event that will get emitted
|
|
* @api private
|
|
*/
|
|
|
|
Socket.prototype.dispatch = function(event){
|
|
debug('dispatching an event %j', event);
|
|
var self = this;
|
|
function dispatchSocket(err) {
|
|
process.nextTick(function(){
|
|
if (err) {
|
|
return self.error(err.data || err.message);
|
|
}
|
|
emit.apply(self, event);
|
|
});
|
|
}
|
|
this.run(event, dispatchSocket);
|
|
};
|
|
|
|
/**
|
|
* Sets up socket middleware.
|
|
*
|
|
* @param {Function} middleware function (event, next)
|
|
* @return {Socket} self
|
|
* @api public
|
|
*/
|
|
|
|
Socket.prototype.use = function(fn){
|
|
this.fns.push(fn);
|
|
return this;
|
|
};
|
|
|
|
/**
|
|
* Executes the middleware for an incoming event.
|
|
*
|
|
* @param {Array} event that will get emitted
|
|
* @param {Function} last fn call in the middleware
|
|
* @api private
|
|
*/
|
|
Socket.prototype.run = function(event, fn){
|
|
var fns = this.fns.slice(0);
|
|
if (!fns.length) return fn(null);
|
|
|
|
function run(i){
|
|
fns[i](event, function(err){
|
|
// upon error, short-circuit
|
|
if (err) return fn(err);
|
|
|
|
// if no middleware left, summon callback
|
|
if (!fns[i + 1]) return fn(null);
|
|
|
|
// go on to next
|
|
run(i + 1);
|
|
});
|
|
}
|
|
|
|
run(0);
|
|
};
|