socket close handler, message unification

lib/server.js

@@ -63,12 +63,12 @@ PeerServer.prototype._initializeWSS = function() {
       if (!!self._timeouts[id]) {
         clearTimeout(self._timeouts[id]);
         delete self._timeouts[id];
-        self._clients[id].end('socket');
+        self._clients[id].end(JSON.stringify({ type: 'HTTP-SOCKET' }));
       } else {
         socket.send(JSON.stringify({ type: 'ERROR', msg: 'ID is taken' }));
         return;
       }
-    } else if (!id) {
+    } else if (id === undefined) {
       id = self._generateClientId();
       socket.send(JSON.stringify({ type: 'ID', id: id }));
     }
@@ -78,6 +78,12 @@ PeerServer.prototype._initializeWSS = function() {
 
     self._processOutstandingOffers(id);
 
+    // Cleanup after a socket closes.
+    socket.on('close', function() {
+      util.log('Socket closed:', id);
+      self._removePeer(id);
+    });
+    // Handle messages from peers.
     socket.on('message', function(data) {
       try {
         var message = JSON.parse(data);
@@ -87,8 +93,7 @@ PeerServer.prototype._initializeWSS = function() {
           case 'LEAVE':
             // Clean up if a Peer sends a LEAVE.
             if (!message.dst) {
-              delete self._clients[message.src];
-              delete self._timeouts[message.src];
+              self._removePeer(message.src);
               break;
             }
           // ICE candidates
@@ -98,7 +103,7 @@ PeerServer.prototype._initializeWSS = function() {
           case 'ANSWER':
           // Firefoxism (connectDataConnection ports)
           case 'PORT':
-            self._handleTransmission(message.type, message.src, message.dst, data);
+            self._handleTransmission(message);
             break;
           default:
             util.prettyError('message unrecognized');
@@ -113,7 +118,6 @@ PeerServer.prototype._initializeWSS = function() {
 
 /** Process outstanding peer offers. */
 PeerServer.prototype._processOutstandingOffers = function(id) {
-  console.log('processing outstanding offers');
   var offers = this._outstandingOffers[id];
   if (offers === undefined)
     return;
@@ -125,7 +129,6 @@ PeerServer.prototype._processOutstandingOffers = function(id) {
 
     delete this._outstandingOffers[id][sources[i]];
   }
-  console.log(this._outstandingOffers[id]);
 };
 
 /** Initialize HTTP server routes. */
@@ -155,36 +158,34 @@ PeerServer.prototype._initializeHTTP = function() {
   });
 
   this._app.post('/offer', function(req, res) {
-    var src = req.body.src;
-    var dst = req.body.dst;
-    self._handleTransmission('OFFER', src, dst, JSON.stringify(req.body), res);
+    self._handleTransmission(req.body, res);
   });
 
   this._app.post('/ice', function(req, res) {
-    var src = req.body.src;
-    var dst = req.body.dst;
-    self._handleTransmission('CANDIDATE', src, dst, JSON.stringify(req.body), res);
+    self._handleTransmission(req.body, res);
   });
 
   this._app.post('/answer', function(req, res) {
-    var src = req.body.src;
-    var dst = req.body.dst;
-    self._handleTransmission('ANSWER', src, dst, JSON.stringify(req.body), res);
+    self._handleTransmission(req.body, res);
   });
 
   this._app.post('/leave', function(req, res) {
-    var src = req.body.src;
-    var dst = req.body.dst;
-    self._handleTransmission('LEAVE', src, dst, JSON.stringify(req.body), res);
+    self._handleTransmission(req.body, res);
   });
 
   this._app.post('/port', function(req, res) {
-    var src = req.body.src;
-    var dst = req.body.dst;
-    self._handleTransmission('PORT', src, dst, JSON.stringify(req.body), res);
+    self._handleTransmission(req.body, res);
   });
 };
 
+PeerServer.prototype._removePeer = function(id) {
+  delete this._clients[id];
+  if (this._timeouts[id]) {
+    clearTimeout(this._timeouts[id]);
+    delete this._timeouts[id];
+  }
+};
+
 /** Saves a streaming response and takes care of timeouts and headers. */
 PeerServer.prototype._startStreaming = function(res, id, write) {
   res.writeHead(200, {'Content-Type': 'application/octet-stream'});
@@ -203,59 +204,75 @@ PeerServer.prototype._startStreaming = function(res, id, write) {
   if (!this._clients[id]) {
     this._clients[id] = res;
     // Set timeout to expire.
-    this._timeouts[id] = setTimeout(function() { res.end('end') }, 10000);
+    var self = this;
+    this._timeouts[id] = setTimeout(function() {
+      self._removePeer(id);
+      res.end(JSON.stringify({ type: 'HTTP-END' }));
+    }, 30000);
+    this._processOutstandingOffers(id);
   } else {
     res.write(JSON.stringify({ type: 'ERROR', msg: 'ID is taken' }) + '\n');
-    res.end('error');
+    res.end(JSON.stringify({ type: 'HTTP-ERROR' }));
   }
 
 };
 
 /** Handles passing on a message. */
-PeerServer.prototype._handleTransmission = function(type, src, dst, data, res) {
+PeerServer.prototype._handleTransmission = function(message, res) {
+  var type = message.type;
+  var src = message.src;
+  var dst = message.dst;
+  var data = JSON.stringify(message);
+
   var destination = this._clients[dst];
 
   if (!!destination) {
     try {
       if (this._timeouts[dst]) {
         data += '\n';
+        destination.write(data);
+      } else {
+        destination.send(data);
       }
       // We have to let the source peer know that the offer was sent
       // successfully so that ice can start being processed.
-      if (type === 'OFFER') {
-        if (!!res) {
-          res.send(200);
-        } else if (!this._timeouts[src] && !!this._clients[src]) {
-          this._clients[src].send(JSON.stringify({ type: 'PEER_READY', src: dst, dst: src }));
-        }
+      if (!!res) {
+        res.send(200);
       }
-      destination.send(data);
     } catch (e) {
+      // This happens when a peer disconnects without closing connections and
+      // the associated WebSocket has not closed.
       util.prettyError(e);
-      // This really shouldn't happen given correct client browsers.
-      // 501: Server does not support this functionality.
+      // Tell other side to stop trying.
+      this._removePeer(dst);
+      this._handleTransmission({
+        type: 'LEAVE',
+        src: dst,
+        dst: src
+      });
       if (!!res) res.send(501);
     }
   } else {
-    if (type === 'OFFER' && (!this._outstandingOffers[dst] || !this._outstandingOffers[dst][src])) {
+    if (type !== 'LEAVE') {
       // Wait 5 seconds for this client to connect.
       var self = this;
       if (!this._outstandingOffers[dst])
         this._outstandingOffers[dst] = {};
-      this._outstandingOffers[dst][src] = [];
-      this._outstandingOffers[dst][src].push(Array.prototype.slice.apply(arguments));
-
-      console.log('offer on queue');
-      setTimeout(function() {
-        delete self._outstandingOffers[dst][src]
-      }, 30000);
-    } else if (type === 'CANDIDATE' && !!this._outstandingOffers[dst][src]) {
+      if (!this._outstandingOffers[dst][src])
+        this._outstandingOffers[dst][src] = [];
+        setTimeout(function() {
+          delete self._outstandingOffers[dst][src]
+        }, 30000);
       this._outstandingOffers[dst][src].push(Array.prototype.slice.apply(arguments));
-      console.log('ice on queue');
+    } else if (type === 'LEAVE' && !dst) {
+      this._removePeer(src);
     } else {
       // Assume a disconnect if the client no longer exists.
-      util.log('destination does not exist');
-      this._handleTransmission('LEAVE', dst, src, JSON.stringify({ type: 'LEAVE', dst: src, src: dst }));
+      this._handleTransmission({
+        type: 'LEAVE',
+        src: dst,
+        dst: src
+      });
       // 410: Resource not available.
       if (!!res) res.send(410);
     }
@@ -264,9 +281,10 @@ PeerServer.prototype._handleTransmission = function(type, src, dst, data, res) {
 
 PeerServer.prototype._generateClientId = function() {
   var clientId = util.randomId();
-  while (!!self._clients[clientId]) {
+  while (!!this._clients[clientId]) {
     clientId = util.randomId();
   }
+  return clientId;
 };
 
 exports.PeerServer = PeerServer;