server.js 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397
  1. var util = require('./util');
  2. var express = require('express');
  3. var http = require('http');
  4. var EventEmitter = require('events').EventEmitter;
  5. var WebSocketServer = require('ws').Server;
  6. var url = require('url');
  7. function PeerServer(options) {
  8. if (!(this instanceof PeerServer)) return new PeerServer(options);
  9. EventEmitter.call(this);
  10. this._app = express();
  11. this._httpServer = http.createServer(this._app);
  12. this._options = util.extend({
  13. port: 80,
  14. debug: false,
  15. timeout: 5000,
  16. key: 'peerjs',
  17. ip_limit: 5000,
  18. concurrent_limit: 5000
  19. }, options);
  20. util.debug = this._options.debug;
  21. // Connected clients
  22. this._clients = {};
  23. // Messages waiting for another peer.
  24. this._outstanding = {};
  25. // Initailize WebSocket server handlers.
  26. this._initializeWSS();
  27. // Initialize HTTP routes. This is only used for the first few milliseconds
  28. // before a socket is opened for a Peer.
  29. this._initializeHTTP();
  30. // Mark concurrent users per ip
  31. this._ips = {};
  32. var self = this;
  33. setInterval(function(){
  34. self._pruneOutstanding();
  35. }, 5000);
  36. };
  37. util.inherits(PeerServer, EventEmitter);
  38. /** Initialize WebSocket server. */
  39. PeerServer.prototype._initializeWSS = function() {
  40. var self = this;
  41. // Create WebSocket server as well.
  42. this._wss = new WebSocketServer({ path: '/peerjs', server: this._httpServer });
  43. this._wss.on('connection', function(socket) {
  44. var query = url.parse(socket.upgradeReq.url, true).query;
  45. var id = query.id;
  46. var token = query.token;
  47. var key = query.key;
  48. var ip = socket.upgradeReq.socket.remoteAddress;
  49. if (!id || !token || !key) {
  50. socket.send(JSON.stringify({ type: 'ERROR', payload: { msg: 'No id, token, or key supplied to websocket server' } }));
  51. socket.close();
  52. return;
  53. }
  54. if (!self._clients[key] || !self._clients[key][id]) {
  55. self._checkKey(key, ip, function(err) {
  56. if (!err) {
  57. self._clients[key][id] = { token: token, ip: ip };
  58. self._ips[ip]++;
  59. socket.send(JSON.stringify({ type: 'OPEN' }));
  60. self._configureWS(socket, key, id, token);
  61. } else {
  62. socket.send(JSON.stringify({ type: 'ERROR', payload: { msg: err } }));
  63. }
  64. });
  65. } else {
  66. self._configureWS(socket, key, id, token);
  67. }
  68. });
  69. };
  70. PeerServer.prototype._configureWS = function(socket, key, id, token) {
  71. var self = this;
  72. var client = this._clients[key][id];
  73. if (token === client.token) {
  74. // res 'close' event will delete client.res for us
  75. client.socket = socket;
  76. // Client already exists
  77. if (client.res) {
  78. client.res.end();
  79. }
  80. } else {
  81. // ID-taken, invalid token
  82. socket.send(JSON.stringify({ type: 'ID-TAKEN', payload: { msg: 'ID is taken' } }));
  83. socket.close();
  84. return;
  85. }
  86. this._processOutstanding(key, id);
  87. // Cleanup after a socket closes.
  88. socket.on('close', function() {
  89. util.log('Socket closed:', id);
  90. if (client.socket == socket) {
  91. self._removePeer(key, id);
  92. }
  93. });
  94. // Handle messages from peers.
  95. socket.on('message', function(data) {
  96. try {
  97. var message = JSON.parse(data);
  98. util.log(message);
  99. switch (message.type) {
  100. case 'LEAVE':
  101. // Clean up if a Peer sends a LEAVE.
  102. if (!message.dst) {
  103. self._removePeer(key, id);
  104. break;
  105. }
  106. // ICE candidates
  107. case 'CANDIDATE':
  108. // Offer or answer between peers.
  109. case 'OFFER':
  110. case 'ANSWER':
  111. // Firefoxism (connectDataConnection ports)
  112. // case 'PORT':
  113. // Use the ID we know to be correct to prevent spoofing.
  114. self._handleTransmission(key, {
  115. type: message.type,
  116. src: id,
  117. dst: message.dst,
  118. payload: message.payload
  119. });
  120. break;
  121. default:
  122. util.prettyError('Message unrecognized');
  123. }
  124. } catch(e) {
  125. throw e;
  126. util.log('Invalid message', data);
  127. }
  128. });
  129. }
  130. PeerServer.prototype._checkKey = function(key, ip, cb) {
  131. if (key == this._options.key) {
  132. if (!this._clients[key]) {
  133. this._clients[key] = {};
  134. }
  135. if (!this._outstanding[key]) {
  136. this._outstanding[key] = {};
  137. }
  138. if (!this._ips[ip]) {
  139. this._ips[ip] = 0;
  140. }
  141. // Check concurrent limit
  142. if (Object.keys(this._clients[key]).length >= this._options.concurrent_limit) {
  143. cb('Server has reached its concurrent user limit');
  144. return;
  145. }
  146. if (this._ips[ip] >= this._options.ip_limit) {
  147. cb(ip + ' has reached its concurrent user limit');
  148. return;
  149. }
  150. cb(null);
  151. } else {
  152. cb('Invalid key provided');
  153. }
  154. }
  155. /** Initialize HTTP server routes. */
  156. PeerServer.prototype._initializeHTTP = function() {
  157. var self = this;
  158. this._app.use(express.bodyParser());
  159. this._app.use(util.allowCrossDomain);
  160. this._app.options('/*', function(req, res, next) {
  161. res.send(200);
  162. });
  163. // Retrieve guaranteed random ID.
  164. this._app.get('/:key/id', function(req, res) {
  165. res.send(self._generateClientId(req.params.key));
  166. });
  167. // Server sets up HTTP streaming when you get post an ID.
  168. this._app.post('/:key/:id/:token/id', function(req, res) {
  169. var id = req.params.id;
  170. var token = req.params.token;
  171. var key = req.params.key;
  172. var ip = req.ip;
  173. if (!self._clients[key] || !self._clients[key][id]) {
  174. self._checkKey(key, ip, function(err) {
  175. if (!err) {
  176. self._clients[key][id] = { token: token, ip: ip };
  177. self._ips[ip]++;
  178. self._startStreaming(res, key, id, token, true);
  179. } else {
  180. res.send(JSON.stringify({ type: 'HTTP-ERROR' }));
  181. }
  182. });
  183. } else {
  184. self._startStreaming(res, key, id, token);
  185. }
  186. });
  187. var handle = function(req, res){
  188. var key = req.params.key;
  189. var id = req.params.id;
  190. var client = self._clients[key][id];
  191. // Auth the req
  192. if (!client || req.params.token !== client.token) {
  193. res.send(401);
  194. return;
  195. } else {
  196. self._handleTransmission(key, {
  197. type: req.body.type,
  198. src: id,
  199. dst: req.body.dst,
  200. payload: req.body.payload
  201. });
  202. res.send(200);
  203. }
  204. };
  205. this._app.post('/:key/:id/:token/offer', handle);
  206. this._app.post('/:key/:id/:token/candidate', handle);
  207. this._app.post('/:key/:id/:token/answer', handle);
  208. this._app.post('/:key/:id/:token/leave', handle);
  209. //this._app.post('/port', handle);
  210. // Listen on user-specified port and
  211. this._httpServer.listen(this._options.port);
  212. };
  213. /** Saves a streaming response and takes care of timeouts and headers. */
  214. PeerServer.prototype._startStreaming = function(res, key, id, token, open) {
  215. var self = this;
  216. res.writeHead(200, {'Content-Type': 'application/octet-stream'});
  217. var pad = '00';
  218. for (var i = 0; i < 10; i++) {
  219. pad += pad;
  220. }
  221. res.write(pad + '\n');
  222. if (open) {
  223. res.write(JSON.stringify({ type: 'OPEN' }) + '\n');
  224. }
  225. var client = this._clients[key][id];
  226. if (token === client.token) {
  227. // Client already exists
  228. res.on('close', function(){
  229. if (client.res === res) {
  230. if (!client.socket) {
  231. // No new request yet, peer dead
  232. self._removePeer(key, id);
  233. return;
  234. }
  235. delete client.res;
  236. }
  237. });
  238. client.res = res;
  239. this._processOutstanding(key, id);
  240. } else {
  241. // ID-taken, invalid token
  242. res.end(JSON.stringify({ type: 'HTTP-ERROR' }));
  243. }
  244. };
  245. PeerServer.prototype._pruneOutstanding = function() {
  246. var keys = Object.keys(this._outstanding);
  247. for (var k = 0, kk = keys.length; k < kk; k++) {
  248. var key = keys[k];
  249. var dsts = Object.keys(this._outstanding[key]);
  250. for (var i = 0, ii = dsts.length; i < ii; i++) {
  251. var offers = this._outstanding[key][dsts[i]];
  252. var seen = {};
  253. for (var j = 0, jj = offers.length; j < jj; j++) {
  254. var message = offers[j];
  255. if (!seen[message.src]) {
  256. this._handleTransmission(key, { type: 'EXPIRE', src: message.dst, dst: message.src });
  257. seen[message.src] = true;
  258. }
  259. }
  260. }
  261. this._outstanding[key] = {};
  262. }
  263. }
  264. /** Process outstanding peer offers. */
  265. PeerServer.prototype._processOutstanding = function(key, id) {
  266. var offers = this._outstanding[key][id];
  267. if (!offers) {
  268. return;
  269. }
  270. for (var j = 0, jj = offers.length; j < jj; j += 1) {
  271. this._handleTransmission(key, offers[j]);
  272. }
  273. delete this._outstanding[key][id];
  274. };
  275. PeerServer.prototype._removePeer = function(key, id) {
  276. if (this._clients[key][id]) {
  277. this._ips[this._clients[key][id].ip]--;
  278. delete this._clients[key][id];
  279. }
  280. };
  281. /** Handles passing on a message. */
  282. PeerServer.prototype._handleTransmission = function(key, message) {
  283. var type = message.type;
  284. var src = message.src;
  285. var dst = message.dst;
  286. var data = JSON.stringify(message);
  287. var destination = this._clients[key][dst];
  288. // User is connected!
  289. if (destination) {
  290. try {
  291. if (destination.socket) {
  292. destination.socket.send(data);
  293. } else if (destination.res) {
  294. data += '\n';
  295. destination.res.write(data);
  296. } else {
  297. // Neither socket no res available. Peer dead?
  298. throw "Peer dead"
  299. }
  300. } catch (e) {
  301. // This happens when a peer disconnects without closing connections and
  302. // the associated WebSocket has not closed.
  303. util.prettyError(e);
  304. // Tell other side to stop trying.
  305. this._removePeer(key, dst);
  306. this._handleTransmission(key, {
  307. type: 'LEAVE',
  308. src: dst,
  309. dst: src
  310. });
  311. }
  312. } else {
  313. // Wait for this client to connect/reconnect (XHR) for important
  314. // messages.
  315. if (type !== 'LEAVE' && type !== 'EXPIRE' && !!dst) {
  316. var self = this;
  317. if (!this._outstanding[key][dst]) {
  318. this._outstanding[key][dst] = [];
  319. }
  320. this._outstanding[key][dst].push(message);
  321. } else if (type === 'LEAVE' && !dst) {
  322. this._removePeer(key, src);
  323. } else {
  324. // Unavailable destination specified with message LEAVE or EXPIRE
  325. // Ignore
  326. }
  327. }
  328. };
  329. PeerServer.prototype._generateClientId = function(key) {
  330. var clientId = util.randomId();
  331. while (!!this._clients[key][clientId]) {
  332. clientId = util.randomId();
  333. }
  334. return clientId;
  335. };
  336. exports.PeerServer = PeerServer;