This commit is contained in:
ericz 2013-02-13 00:04:07 -08:00
parent 08df4056da
commit 6f22c4225e

View File

@ -18,7 +18,9 @@ function PeerServer(options) {
port: 80, port: 80,
debug: false, debug: false,
timeout: 5000, timeout: 5000,
path: 'peerjs' key: 'peerjs',
ip_limit: 5000,
concurrent_limit: 5000
}, options); }, options);
util.debug = this._options.debug; util.debug = this._options.debug;
@ -50,7 +52,7 @@ PeerServer.prototype._initializeWSS = function() {
var self = this; var self = this;
// Create WebSocket server as well. // Create WebSocket server as well.
this._wss = new WebSocketServer({ path: '/' + this._options.path, server: this._httpServer }); this._wss = new WebSocketServer({ path: '/peerjs', server: this._httpServer });
this._wss.on('connection', function(socket) { this._wss.on('connection', function(socket) {
@ -58,8 +60,7 @@ PeerServer.prototype._initializeWSS = function() {
var id = query.id; var id = query.id;
var token = query.token; var token = query.token;
var key = query.key; var key = query.key;
var ip = socket.upgradeReq.socket.remoteAddress;
var client = self._clients[id];
if (!id || !token || !key) { if (!id || !token || !key) {
socket.send(JSON.stringify({ type: 'ERROR', payload: { msg: 'No id, token, or key supplied to websocket server' } })); socket.send(JSON.stringify({ type: 'ERROR', payload: { msg: 'No id, token, or key supplied to websocket server' } }));
@ -67,75 +68,105 @@ PeerServer.prototype._initializeWSS = function() {
return; return;
} }
if (!client) { if (!self._clients[key] || !self._clients[key][id]) {
client = { token: token }; self._checkKey(key, ip, function(err) {
self._clients[id] = client; if (!err) {
socket.send(JSON.stringify({ type: 'OPEN' })); self._clients[key][id] = { token: token };
} socket.send(JSON.stringify({ type: 'OPEN' }));
self._configureWS(socket, key, id, token);
if (token === client.token) { } else {
// res 'close' event will delete client.res for us socket.send({ type: 'ERROR', payload: { msg: err } });
client.socket = socket;
// Client already exists
if (client.res) {
client.res.end();
}
} else {
// ID-taken, invalid token
socket.send(JSON.stringify({ type: 'ID-TAKEN', payload: { msg: 'ID is taken' } }));
socket.close();
return;
}
self._processOutstanding(id);
// Cleanup after a socket closes.
socket.on('close', function() {
util.log('Socket closed:', id);
if (client.socket == socket) {
self._removePeer(id);
}
});
// Handle messages from peers.
socket.on('message', function(data) {
try {
var message = JSON.parse(data);
util.log(message);
switch (message.type) {
case 'LEAVE':
// Clean up if a Peer sends a LEAVE.
if (!message.dst) {
self._removePeer(id);
break;
}
// ICE candidates
case 'CANDIDATE':
// Offer or answer between peers.
case 'OFFER':
case 'ANSWER':
// Firefoxism (connectDataConnection ports)
// case 'PORT':
// Use the ID we know to be correct to prevent spoofing.
self._handleTransmission({
type: message.type,
src: id,
dst: message.dst,
payload: message.payload
});
break;
default:
util.prettyError('Message unrecognized');
} }
} catch(e) { });
throw e; } else {
util.log('Invalid message', data); self._configureWS(socket, key, id, token);
} }
});
}); });
}; };
PeerServer.prototype._configureWS = function(socket, key, id, token) {
var self = this;
var client = this._clients[key][id];
if (token === client.token) {
// res 'close' event will delete client.res for us
client.socket = socket;
// Client already exists
if (client.res) {
client.res.end();
}
} else {
// ID-taken, invalid token
socket.send(JSON.stringify({ type: 'ID-TAKEN', payload: { msg: 'ID is taken' } }));
socket.close();
return;
}
this._processOutstanding(key, id);
// Cleanup after a socket closes.
socket.on('close', function() {
util.log('Socket closed:', id);
if (client.socket == socket) {
self._removePeer(key, id);
}
});
// Handle messages from peers.
socket.on('message', function(data) {
try {
var message = JSON.parse(data);
util.log(message);
switch (message.type) {
case 'LEAVE':
// Clean up if a Peer sends a LEAVE.
if (!message.dst) {
self._removePeer(key, id);
break;
}
// ICE candidates
case 'CANDIDATE':
// Offer or answer between peers.
case 'OFFER':
case 'ANSWER':
// Firefoxism (connectDataConnection ports)
// case 'PORT':
// Use the ID we know to be correct to prevent spoofing.
self._handleTransmission(key, {
type: message.type,
src: id,
dst: message.dst,
payload: message.payload
});
break;
default:
util.prettyError('Message unrecognized');
}
} catch(e) {
throw e;
util.log('Invalid message', data);
}
});
}
PeerServer.prototype._checkKey = function(key, ip, cb) {
if (key == 'peerjs') {
if (!this._clients[key]) {
this._clients[key] = {};
}
if (!this._outstanding[key]) {
this._outstanding[key] = {};
}
cb(null);
} else {
cb('Bad key!');
}
}
/** Initialize HTTP server routes. */ /** Initialize HTTP server routes. */
PeerServer.prototype._initializeHTTP = function() { PeerServer.prototype._initializeHTTP = function() {
@ -151,25 +182,42 @@ PeerServer.prototype._initializeHTTP = function() {
// Retrieve guaranteed random ID. // Retrieve guaranteed random ID.
this._app.get('/:key/id', function(req, res) { this._app.get('/:key/id', function(req, res) {
res.send(self._generateClientId()); res.send(self._generateClientId(req.params.key));
}); });
// Server sets up HTTP streaming when you get post an ID. // Server sets up HTTP streaming when you get post an ID.
this._app.post('/:key/:id/:token/id', function(req, res) { this._app.post('/:key/:id/:token/id', function(req, res) {
self._startStreaming(res, req.params.id, req.params.token); var id = req.params.id;
var token = req.params.token;
var key = req.params.key;
if (!self._clients[key] || !self._clients[key][id]) {
self._checkKey(key, req.ip, function(err) {
if (!err) {
self._clients[key][id] = { token: token };
self._startStreaming(res, key, id, token, true);
} else {
res.send({ type: 'ERROR', payload: { msg: err } });
}
});
} else {
self._startStreaming(res, key, id, token);
}
}); });
var handle = function(req, res){ var handle = function(req, res){
var client = self._clients[req.params.id]; var key = req.params.key;
var id = req.params.id;
var client = self._clients[key][id];
// Auth the req // Auth the req
if (!client || req.params.token !== client.token) { if (!client || req.params.token !== client.token) {
res.send(401); res.send(401);
return; return;
} else { } else {
self._handleTransmission({ self._handleTransmission(key, {
type: req.body.type, type: req.body.type,
src: req.params.id, src: id,
dst: req.body.dst, dst: req.body.dst,
payload: req.body.payload payload: req.body.payload
}); });
@ -193,7 +241,7 @@ PeerServer.prototype._initializeHTTP = function() {
}; };
/** Saves a streaming response and takes care of timeouts and headers. */ /** Saves a streaming response and takes care of timeouts and headers. */
PeerServer.prototype._startStreaming = function(res, id, token) { PeerServer.prototype._startStreaming = function(res, key, id, token, open) {
var self = this; var self = this;
res.writeHead(200, {'Content-Type': 'application/octet-stream'}); res.writeHead(200, {'Content-Type': 'application/octet-stream'});
@ -204,30 +252,26 @@ PeerServer.prototype._startStreaming = function(res, id, token) {
} }
res.write(pad + '\n'); res.write(pad + '\n');
var client = this._clients[id]; if (open) {
// Save res so we can write to it.
if (!client) {
// New client, save info
client = { token: token };
this._clients[id] = client;
res.write(JSON.stringify({ type: 'OPEN' }) + '\n'); res.write(JSON.stringify({ type: 'OPEN' }) + '\n');
} }
var client = this._clients[key][id];
if (token === client.token) { if (token === client.token) {
// Client already exists // Client already exists
res.on('close', function(){ res.on('close', function(){
if (client.res === res) { if (client.res === res) {
if (!client.socket) { if (!client.socket) {
// No new request yet, peer dead // No new request yet, peer dead
self._removePeer(id); self._removePeer(key, id);
return; return;
} }
delete client.res; delete client.res;
} }
}); });
client.res = res; client.res = res;
this._processOutstanding(id); this._processOutstanding(key, id);
} else { } else {
// ID-taken, invalid token // ID-taken, invalid token
res.end(JSON.stringify({ type: 'HTTP-ERROR' })); res.end(JSON.stringify({ type: 'HTTP-ERROR' }));
@ -235,45 +279,50 @@ PeerServer.prototype._startStreaming = function(res, id, token) {
}; };
PeerServer.prototype._pruneOutstanding = function() { PeerServer.prototype._pruneOutstanding = function() {
var dsts = Object.keys(this._outstanding); console.log('before prune', this._outstanding);
for (var i = 0, ii = dsts.length; i < ii; i++) { var keys = Object.keys(this._outstanding);
var offers = this._outstanding[dsts[i]]; for (var k = 0, kk = keys.length; k < kk; k++) {
var seen = {}; var key = keys[k];
for (var j = 0, jj = offers.length; j < jj; j++) { var dsts = Object.keys(this._outstanding[key]);
var message = offers[j]; for (var i = 0, ii = dsts.length; i < ii; i++) {
if (!seen[message.src]) { var offers = this._outstanding[key][dsts[i]];
this._handleTransmission({ type: 'EXPIRE', src: message.dst, dst: message.src }); var seen = {};
seen[message.src] = true; for (var j = 0, jj = offers.length; j < jj; j++) {
var message = offers[j];
if (!seen[message.src]) {
this._handleTransmission(key, { type: 'EXPIRE', src: message.dst, dst: message.src });
seen[message.src] = true;
}
} }
} }
this._outstanding[key] = {};
} }
this._outstanding = [];
} }
/** Process outstanding peer offers. */ /** Process outstanding peer offers. */
PeerServer.prototype._processOutstanding = function(id) { PeerServer.prototype._processOutstanding = function(key, id) {
var offers = this._outstanding[id]; var offers = this._outstanding[key][id];
if (!offers) { if (!offers) {
return; return;
} }
for (var j = 0, jj = offers.length; j < jj; j += 1) { for (var j = 0, jj = offers.length; j < jj; j += 1) {
this._handleTransmission(offers[j]); this._handleTransmission(key, offers[j]);
} }
delete this._outstanding[id]; delete this._outstanding[key][id];
}; };
PeerServer.prototype._removePeer = function(id) { PeerServer.prototype._removePeer = function(key, id) {
delete this._clients[id]; delete this._clients[key][id];
}; };
/** Handles passing on a message. */ /** Handles passing on a message. */
PeerServer.prototype._handleTransmission = function(message) { PeerServer.prototype._handleTransmission = function(key, message) {
var type = message.type; var type = message.type;
var src = message.src; var src = message.src;
var dst = message.dst; var dst = message.dst;
var data = JSON.stringify(message); var data = JSON.stringify(message);
var destination = this._clients[dst]; var destination = this._clients[key][dst];
// User is connected! // User is connected!
if (destination) { if (destination) {
@ -292,8 +341,8 @@ PeerServer.prototype._handleTransmission = function(message) {
// the associated WebSocket has not closed. // the associated WebSocket has not closed.
util.prettyError(e); util.prettyError(e);
// Tell other side to stop trying. // Tell other side to stop trying.
this._removePeer(dst); this._removePeer(key, dst);
this._handleTransmission({ this._handleTransmission(key, {
type: 'LEAVE', type: 'LEAVE',
src: dst, src: dst,
dst: src dst: src
@ -304,12 +353,12 @@ PeerServer.prototype._handleTransmission = function(message) {
// messages. // messages.
if (type !== 'LEAVE' && type !== 'EXPIRE' && !!dst) { if (type !== 'LEAVE' && type !== 'EXPIRE' && !!dst) {
var self = this; var self = this;
if (!this._outstanding[dst]) { if (!this._outstanding[key][dst]) {
this._outstanding[dst] = []; this._outstanding[key][dst] = [];
} }
this._outstanding[dst].push(message); this._outstanding[key][dst].push(message);
} else if (type === 'LEAVE' && !dst) { } else if (type === 'LEAVE' && !dst) {
this._removePeer(src); this._removePeer(key, src);
} else { } else {
// Unavailable destination specified with message LEAVE or EXPIRE // Unavailable destination specified with message LEAVE or EXPIRE
// Ignore // Ignore
@ -317,9 +366,9 @@ PeerServer.prototype._handleTransmission = function(message) {
} }
}; };
PeerServer.prototype._generateClientId = function() { PeerServer.prototype._generateClientId = function(key) {
var clientId = util.randomId(); var clientId = util.randomId();
while (!!this._clients[clientId]) { while (!!this._clients[key][clientId]) {
clientId = util.randomId(); clientId = util.randomId();
} }
return clientId; return clientId;