Peering updates

README.md

@@ -1,4 +1,4 @@
-#The Zetta Runtime
+#Zetta 
 
 [![Build Status](https://travis-ci.org/zettajs/zetta.svg?branch=master)](https://travis-ci.org/zettajs/zetta) 
 

lib/api_resources/peer_management.js

@@ -3,6 +3,7 @@ var PeerClient = require('../peer_client');
 var rels = require('zetta-rels');
 var MediaType = require('api-media-type');
 var Query = require('calypso').Query;
+var http = require('http');
 
 var PeerManagementResource = module.exports = function(server) {
   this.server = server;
@@ -16,11 +17,17 @@ PeerManagementResource.prototype.init = function(config) {
     .consumes(MediaType.FORM_URLENCODED)
     .get('/', this.list)
     .post('/', this.link)
-    .get('/{id}', this.show);
+    .get('/{id}', this.show)
+    .del('/{id}', this.deletePeer)
+    .put('/{id}', this.updatePeer);
 };
 
 PeerManagementResource.prototype.list = function(env, next) {
+  var params = env.route.query;
   var allQuery = Query.of('peers');
+  if(params) {
+    allQuery.ql(params.ql);
+  }  
 
   this.registry.find(allQuery, function(err, results) {
     var builder = new PeerManagementBuilder(results, env.helpers.url);
@@ -55,7 +62,7 @@ PeerManagementResource.prototype.link = function(env, next) {
         return next(env);
       }
 
-      self._connect(peer);
+      self.server._runPeer(peer);
 
       env.response.statusCode = 202;
       env.response.setHeader('Location', env.helpers.url.join(peer.id));
@@ -65,47 +72,133 @@ PeerManagementResource.prototype.link = function(env, next) {
   });
 };
 
-PeerManagementResource.prototype._connect = function(peer) {
+//This is conditonal based on where the request is coming from.
+//From acceptor we'll add additonal info to the request, and proxy to initiator
+//From intiator we'll perform the disconnection
+PeerManagementResource.prototype.deletePeer = function(env, next) {
+  var id = env.route.params.id;
   var self = this;
-  var client = new PeerClient(peer.url, this.server);
+  var query = Query.of('peers').where({connectionId: id});
+  this.registry.find(query, function(err, results) {
+    if(results.length) {
+      var peer = results[0];
+      if(peer.direction === 'initiator') {
+        self._disconnect(peer);
+        env.response.statusCode = 200;
+        next(env);
+      } else if(peer.direction === 'acceptor') {
+        self._proxyDisconnect(env, next, id);
+      } else {
+        env.response.statusCode = 500;
+        next(env);  
+      }
+    } else {
+      env.response.statusCode = 404;
+      next(env);
+    }
+  });
+};
 
-  client.on('connected', function() {
-    self.registry.get(peer.id, function(err, result) {
-      result = JSON.parse(result);
-      result.status = 'connected';
-      self.registry.save(result);
+//Updating a peer is a two part process.
+//First we'll determine whether or not the API call should be proxied
+//Then we'll connect to the new peer.
+//Then we'll disconnect from the old peer.
+PeerManagementResource.prototype.updatePeer = function(env, next) {
+  var self = this;
+  env.request.getBody(function(err, body) {
+    var params = querystring.parse(body.toString());
+    var id = env.route.params.id;
+    var url = params.url;
+    var query = Query.of('peers').where({connectionId: id});
+    self.registry.find(query, function(err, results) {
+      if(results.length) {
+        var peer = results[0];
+        if(peer.direction === 'initiator') {
+
+          self.server._peers.push(url);
+
+          var newPeer = {
+            url: url,
+            direction: 'initiator',
+            status: 'connecting'
+          }; 
+
+
+          self.registry.add(newPeer, function(err, newPeer) {
+            if (err) {
+              env.response.statusCode = 500;
+              return next(env);
+            }
+
+            self.server._runPeer(newPeer);
+            self._disconnect(peer);
+            env.response.statusCode = 200;
+            next(env);
+          });
+
+        } else if(peer.direction === 'acceptor') {
+          self._proxyDisconnect(env, next, id);
+        } else {
+          env.response.statusCode = 500;
+          next(env);
+        } 
+      } else {
+        env.response.statusCode = 404;
+        next(env);
+      }
     });
   });
+};
 
-  client.on('error', function(error) {
-    self.registry.get(peer.id, function(err, result) {
-      result = JSON.parse(result);
-      result.status = 'failed';
-      result.error = error;
-      self.registry.save(result);
-    });
+PeerManagementResource.prototype._disconnect = function(peer) {
+  var wsUrl = PeerClient.calculatePeerUrl(peer.url, this.server._name);
+  var peers = this.server._peerClients.filter(function(peer) {
+    return peer.url === wsUrl;  
   });
+  var client = peers[0];
+  client.close();     
+};
 
-  client.on('closed', function() {
-    self.registry.get(peer.id, function(err, result) {
-      result = JSON.parse(result);
-      result.status = 'connecting';
-      self.registry.save(result, function() {
-        client.start();
-      });
-    });
+PeerManagementResource.prototype._proxyDisconnect = function(env, next, id) { 
+  var self = this;
+  var peerSocket;
+  var sockets = Object.keys(this.server.httpServer.peers).forEach(function(socketId){
+    var peers = self.server.httpServer.peers;
+    var socket = peers[socketId];
+    if(socket.connectionId === id) {
+      peerSocket = socket;  
+    }
   });
 
-  peer.status = 'connecting';
-  self.registry.save(peer, function() {
-    client.start();
+  if (!peerSocket) {
+    env.response.statusCode = 404;
+    return next(env);
+  }
+
+  peerSocket.on('end', function() {
+    env.response.statusCode = 200;
+    next(env);  
   });
-  // setTimeout to change status to failed if connection not made.
+
+  this._proxyToPeer(peerSocket, env);
+
+};
+
+PeerManagementResource.prototype._proxyToPeer = function(peer, env) {
+  var req = env.request;
+  var res = env.response;
+  var agent = env.zettaAgent || peer.agent;
+  var opts = { method: req.method, headers: req.headers, path: req.url, agent: agent };
+  var request = http.request(opts);
+  if(req.body) {
+    request.end(req.body);  
+  } else {
+    req.pipe(request);
+  }
 };
 
 PeerManagementResource.prototype.show = function(env, next) {
   var id = env.route.params.id;
-
   this.registry.get(id, function(err, result) {
     if (err) {
       env.response.statusCode = 404;
@@ -149,18 +242,33 @@ PeerManagementBuilder.prototype.entities = function() {
           direction: peer.direction,
           status: peer.status,
           error: peer.error,
-          updated: peer.updated
+          updated: peer.updated,
+          connectionId: peer.connectionId
         }
       };
 
+      entity.actions = [];
+      entity.actions.push({
+        name: 'disconnect',
+        method: 'DELETE',
+        href: self.urlHelper.path('/peer-management/'+peer.connectionId)
+      },{
+        name: 'update',
+        method: 'PUT',
+        href: self.urlHelper.path('/peer-management/'+peer.connectionId),
+        fields: [{name: 'url', type: 'url'}]
+      }); 
+
+      /* API action does not work wait till we fix it
       if (peer.direction === 'initiator') {
-        entity.actions = [{
+        entity.actions.push({
           name: 'reconnect',
           method: 'POST',
           href: self.urlHelper.current(),
           fields: [{ name: 'url', type: 'url', value: peerUrl }]
-        }];
+        });
       }
+      */
 
       var peerUrlRel = peer.direction === 'initiator' ? rels.root : rels.server;
       entity.links = [
@@ -215,20 +323,34 @@ PeerItemBuilder.prototype.properties = function() {
     direction: this.data.direction,
     status: this.data.status,
     error: this.data.error,
-    updated: this.data.updated
+    updated: this.data.updated,
+    connectionId: this.data.connectionId
   };
 
   return this;
 };
 
 PeerItemBuilder.prototype.actions = function() {
+  this.base.actions = [];
+
+  this.base.actions.push({
+    name: 'disconnect',
+    method: 'DELETE',
+    href: this.urlHelper.path('/peer-management/'+this.data.connectionId),
+  },{
+    name: 'update',
+    method: 'PUT',
+    href: this.urlHelper.path('/peer-management/'+this.data.connectionId),
+    fields: [{name: 'url', type: 'url'}]
+  });  
+
   if (this.data.direction === 'initiator') {
-    this.base.actions = [{
+    this.base.actions.push({
       name: 'reconnect',
       method: 'POST',
       href: this.urlHelper.path('/peer-management'),
       fields: [{ name: 'url', type: 'url', value: this.data.url }]
-    }];
+    });
   }
 
   return this;

lib/peer_client.js

@@ -5,30 +5,28 @@ var uuid = require('node-uuid');
 var Logger = require('./logger');
 var WebSocket = require('./web_socket');
 
-var RETRY_INTERVAL = 3000;
-var RETRY_MAX = 9;
-var CONNECTION_BACKOFF_MAX = 50;
-
-function backoffTime() {
-  return Math.floor(0 + CONNECTION_BACKOFF_MAX * Math.random());
-}
-
-var PeerClient = module.exports = function(url, server) {
-  var wsUrl = url.replace(/^http/, 'ws');
-  var peerPath = '/peers/' + server._name;
+function calculatePeerUrl(url, name){
+ var wsUrl = url.replace(/^http/, 'ws');
+  var peerPath = '/peers/' + name;
   if(wsUrl.indexOf('/', wsUrl.length - 1) === -1)  {
     wsUrl = wsUrl + peerPath;
   } else {
     wsUrl = wsUrl.slice(0, wsUrl.length - 1) + peerPath;
-  }
+  } 
+  return wsUrl;
+}
+
 
+var PeerClient = module.exports = function(url, server) {
+  var wsUrl = calculatePeerUrl(url, server._name); 
   this.reconnect = {
     min: 100,
     max: 30000, // max amount of time allowed to backoff
     maxRandomOffset: 1000, // max amount of time
   };
 
   this.url = wsUrl;
+
   this.server = server.httpServer.spdyServer;
   this.connected = false;
   this.retryCount = 0;
@@ -44,6 +42,7 @@ var PeerClient = module.exports = function(url, server) {
 };
 util.inherits(PeerClient, EventEmitter);
 
+PeerClient.calculatePeerUrl = calculatePeerUrl;
 
 PeerClient.prototype.start = function() {
   this._createSocket();
@@ -68,7 +67,7 @@ PeerClient.prototype._createSocket = function() {
   var backoff = this.generateBackoff(this.retryCount);
   this._backoffTimer = setTimeout(function(){
     // create a new connection id
-    this.connectionId = uuid.v4();
+    self.connectionId = uuid.v4();
     self._ws = new WebSocket(self.url + '?connectionId=' + self.connectionId, {});
     self._ws.on('open', function(socket) {
       self.checkServerReq();
@@ -133,7 +132,6 @@ PeerClient.prototype.generateBackoff = function(attempt) {
 
   var random = parseInt(Math.random() * this.reconnect.maxRandomOffset);
   var backoff = (Math.pow(2, attempt) * this.reconnect.min);
-
   if (backoff > this.reconnect.max) {
     return this.reconnect.max + random;
   } else {

lib/peer_socket.js

@@ -167,6 +167,7 @@ PeerSocket.prototype._setPeerStatus = function(status, err, cb) {
     }
 
     peer.status = status;
+    peer.connectionId = self.connectionId;
     if (err) {
       peer.error = err;
     }

lib/registry.js

@@ -138,7 +138,17 @@ Registry.prototype._find = function(q, cb) {
 
 
 Registry.prototype.get = function(id, cb) {
-  this.db.get(id, { valueEncoding: 'json' }, cb);
+  this.db.get(id, { valueEncoding: 'json' }, function(err, result){
+    if(err) {
+      cb(err);
+    } else {
+      if(typeof result === 'object') {
+        cb(null, result);
+      } else {
+        cb(null, JSON.parse(result));
+      }
+    }
+  });
 };
 
 Registry.prototype.close = function() {

lib/web_socket.js

@@ -51,11 +51,13 @@ var WebSocket = module.exports = function(address, httpOptions) {
       return;
     }
 
-    req.connection.on('close', function() {
-      self.emit('close');
-    });
+    self.onClose = function() {
+      self.emit('close');  
+    };
+    req.connection.on('close', self.onClose);
 
     self.emit('open', req.connection);
+    self.socket = req.connection;
   });
 
   req.on('error', function(e) { self.emit('error', e); });
@@ -68,3 +70,11 @@ var WebSocket = module.exports = function(address, httpOptions) {
 };
 
 util.inherits(WebSocket, EventEmitter);
+
+WebSocket.prototype.close = function() {
+  this.socket.removeListener('close', this.onClose);
+  if(this.socket) {
+    this.socket.end();   
+    this.emit('close', null, null, true);
+  } 
+};