Websocket streams

lib/api_resources/root.js

@@ -214,6 +214,13 @@ RootResource.prototype._queryDevices = function(env, next) {
 };
 //This is called when ql and server isn't supplied
 RootResource.prototype._renderRoot = function(env, next) {
+
+  var isForwardedProtocol = env.request.headers.hasOwnProperty('x-forwarded-proto') && ['http', 'https'].indexOf(env.request.headers['x-forwarded-proto']) !== -1;
+  var isSpdy = !!env.request.isSpdy && !isForwardedProtocol;
+
+
+  var eventsPath = env.helpers.url.path('/events');
+  var wsEventsPath = eventsPath.replace(/^http/, 'ws');
   env.response.body = {
     class: ['root'],
     links: [
@@ -225,7 +232,8 @@ RootResource.prototype._renderRoot = function(env, next) {
         title: this.server._name,
         rel: [rels.server],
         href: env.helpers.url.path('/servers/' + encodeURI(this.server.id) )
-      }
+      },
+
     ], 
     actions: [
       {
@@ -247,6 +255,13 @@ RootResource.prototype._renderRoot = function(env, next) {
     ]
   };
 
+
+  if(!isSpdy) {
+    env.response.body.links.push({
+        rel: [rels.events],
+        href: wsEventsPath  
+    });  
+  }
   var peerQuery = Query.of('peers').ql('where direction = "acceptor" and status = "connected"');
   this.server.peerRegistry.find(peerQuery, function(err, results) {
     if (results) {

lib/event_broker.js

@@ -1,8 +1,6 @@
+var JSCompiler = require('caql-js-compiler');
 var querytopic = require('./query_topic');
-
-function topicMatch(a, b) {
-  return a === b;
-}
+var StreamTopic = require('./stream_topic');
 
 var EventBroker = module.exports = function(zetta) {
   this.peers = {};
@@ -14,6 +12,8 @@ var EventBroker = module.exports = function(zetta) {
 
   this._publishListeners = {}; // {<server_name>: {<topic>: _listner } }
   this._deviceQueries = {};
+
+  this._queryCache = {};
 };
 
 EventBroker.prototype.peer = function(peer) {
@@ -38,7 +38,10 @@ EventBroker.prototype._setupPublishListener = function(peer, topic) {
   }
 
   this._publishListeners[peer.name][topic] = function(data) {
-    self._publish(topic, data);
+    streamTopic = StreamTopic.parse(topic);
+
+
+    self._publish(streamTopic, null, data, peer.name);
   };
 
   peer.on(topic, this._publishListeners[peer.name][topic]);
@@ -70,12 +73,68 @@ EventBroker.prototype.client = function(client) {
     return stillValid && client.ws === cl.ws;
   });
 
-  if (c.length > 0) {
-    return;
-  }
 
-  this.clients.push(client);
+  if (client.streamEnabled) {
+    function generateQuery(topic) {
+      return {
+        name: topic.serverName,
+        topic: topic.pubsubIdentifier()
+      };
+    }
+    client.on('subscribe', function(subscription) {
+      var query = generateQuery(subscription.topic);
+      query.subscriptionId = subscription.subscriptionId;
+      query.limit = subscription.limit;
+      query.count = 0;
+      query.caql = subscription.topic.streamQuery;
+      client.query.push(query);
+      var connectedPeers = [];
+
+      var subscribeToPeer = function(peerName) {
+        var copiedQuery = {};
+        Object.keys(query).forEach(function(key) {
+          copiedQuery[key] = query[key];
+        });
+
+        if(peerName) {
+          copiedQuery.name = peerName;
+        }
+
+        if(connectedPeers.indexOf(copiedQuery.name) === -1) {
+          connectedPeers.push(copiedQuery.name);
+
+          if(query.name instanceof RegExp && !query.name.exec(copiedQuery.name)) {
+            return;
+          } 
+          self._subscribe(copiedQuery);
+        }
+
+      }
+
+      if(query.name instanceof RegExp || query.name === '*') {        
+        self.zetta.pubsub.subscribe('_peer/connect', function(topic, data) {
+          subscribeToPeer(data.peer.name);    
+        });
 
+        Object.keys(self.peers).forEach(subscribeToPeer);
+        subscribeToPeer(self.zetta._name);
+      } else {
+        subscribeToPeer();
+      } 
+
+      //listen peer connect events
+      //iterate through current peers
+      //array of peers that have been given the topic
+
+    });
+
+    client.on('unsubscribe', function(subscription) {
+      var query = generateQuery(subscription.topic);
+      self._unsubscribe(query);
+    });
+  }
+
+  this.clients.push(client);
   client.on('close', function() {
     client.query.forEach(self._unsubscribe.bind(self));
     var idx = self.clients.indexOf(client);
@@ -92,7 +151,7 @@ EventBroker.prototype._subscribe = function(query) {
   var topic = query.topic;
 
   // is local
-  if (query.name === this.zetta.id) {
+  if (query.name === this.zetta.id || !query.name) {
     if (!this.subscriptions[topic]) {
       this.subscriptions[topic] = { count: 0, listener: null };
     }
@@ -110,7 +169,6 @@ EventBroker.prototype._subscribe = function(query) {
 
     this.subscriptions[topic].count++;
   } else {
-
     if (!this._peerSubscriptions[query.name]) {
       this._peerSubscriptions[query.name] = {};
     }
@@ -129,11 +187,12 @@ EventBroker.prototype._subscribe = function(query) {
 EventBroker.prototype._unsubscribe = function(query) {
   var topic = query.topic;
 
-  if (query.name === this.zetta.id) {
+  if (query.name === this.zetta.id || !query.name) {
     this.subscriptions[topic].count--;
     if (this.subscriptions[topic].count > 0) {
       return;
     }
+
     // unsubscribe locally
     this.zetta.pubsub.unsubscribe(topic, this.subscriptions[topic].listener);
     delete this.subscriptions[topic];
@@ -172,25 +231,58 @@ EventBroker.prototype._unsubscribe = function(query) {
   }
 };
 
+EventBroker.prototype._publish = function(topic, sourceTopic, msg, peerName) {
+  var self = this;
+  var originalTopic = msg.topic;
 
-EventBroker.prototype._publish = function(topic, data) {
   this.clients.forEach(function(client) {
     client.query.forEach(function(query) {
-      if (!topicMatch(topic, query.topic)) {
-        return;
+      if (topic.match(query.topic)) {
+
+          if (client.streamEnabled) {
+            msg.type = 'event';
+            if(topic._useComponents) {
+              msg.topic = peerName + '/' + originalTopic;
+            } else {
+              if(topic._original.indexOf('*') === -1) {
+                msg.topic = topic._original;
+              } else {
+                msg.topic = sourceTopic;
+              }
+            }
+            msg.subscriptionId = query.subscriptionId;
+
+            query.count++;
+            if (typeof query.limit === 'number' && query.count > query.limit) {
+              // unsubscribe broker
+              self._unsubscribe(query);
+              client._unsubscribe(query.subscriptionId)
+              return;
+            }
+            if (query.caql) {
+
+              var compiled = client._queryCache[query.caql];
+              var result = compiled.filterOne({ data: msg.data });
+              if (result) {
+                msg.data = result[Object.keys(result)[0]];
+              } else {
+                return;
+              }
+            }
+          }
+
+          client.send(topic, msg, function(err){
+            if (err) {
+              console.error('ws error: '+err);
+            }
+          });
       }
-
-      client.send(topic, data, function(err){
-        if (err) {
-          console.error('ws error: '+err);
-        }
-      });
     });
   });
 };
 
-EventBroker.prototype._onLocalPubsub = function(topic, data) {
-  this._publish(topic, data);
+EventBroker.prototype._onLocalPubsub = function(topic, data, sourceTopic) {
+  this._publish(topic, sourceTopic, data, this.zetta._name);
 };