Follow the DB

Author: jhs

lib/transports/couchdb.js

@@ -336,7 +336,7 @@ Server.prototype.start = function(callback) {
       return callback(er);
 
     if(body.couchdb !== 'Welcome')
-      retur callback(new Error('Cannot confirm CouchDB server: ' + self.url));
+      return callback(new Error('Cannot confirm CouchDB server: ' + self.url));
 
     self.log.debug('Confirming CouchDB admin credentials');
     self.req('/_session', function(er, resp, body) {
@@ -406,10 +406,47 @@ Server.prototype.prep = function(callback) {
 
 Server.prototype.listen = function(callback) {
   var self = this;
-  self.log.debug('Listening for CouchDB connections');
 
-  // TODO
-};
+  if(self.feed)
+    return callback(new Error('Server feed already exists'));
+
+  self.feed = new follow.Feed;
+
+  self.feed.db            = self.url + self.resource;
+  self.feed.since         = 0;
+  self.feed.heartbeat     = 30 * 1000;        // 30 seconds
+  self.feed.inactivity_ms = 3 * 86400 * 1000; // 3 days
+  self.feed.include_docs  = true;
+  self.feed.filter        = message_filter;
+
+  self.feed.on('change', on_change);
+  self.feed.on('error' , on_error);
+
+  self.log.debug('Listening for CouchDB transport connections');
+  self.feed.follow();
+
+  var id_re = /^socket\.io\/1\/([0-9a-f]{32})$/;
+  var connections = {};
+
+  function message_filter(doc, req) {
+    var match = doc._id.match(id_re);
+    if(!match)
+      return false;
+
+    var transport_id = doc._id.match(id_re)[1];
+
+    return (transport_id in connections) || doc.state === 'opening';
+  }
+
+  function on_error(er) {
+    self.emit('error', er);
+  }
+
+  function on_change(change) {
+    self.log.debug('Change ' + change.seq + ' is ' + change.id + ': ' + JSON.stringify(change.doc));
+  }
+}
+
 
 /**
  * Data validator for the public connection DB