server.js 10 KB

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