Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

(#47) - Add support _db_updates feeds #48

Merged
merged 1 commit into from May 21, 2014
Merged

(#47) - Add support _db_updates feeds #48

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
51 changes: 34 additions & 17 deletions lib/feed.js
View file
Open in desktop
Expand Up @@ -61,10 +61,6 @@ function Feed (opts) {
if(typeof opts === 'string')
opts = {'db': opts};

opts.db = opts.db || opts.url || opts.uri
delete opts.url
delete opts.uri

Object.keys(opts).forEach(function(key) {
self[key] = opts[key];
})
Expand All @@ -78,9 +74,19 @@ Feed.prototype.start =
Feed.prototype.follow = function follow_feed() {
var self = this;

self.db = self.db || self.url || self.uri
delete self.url
delete self.uri

if(!self.db)
throw new Error('Database URL required');

if (self.db.match(/\/_db_updates$/))
self.is_db_updates = true;

if(self.is_db_updates)
delete self.since;

if(self.feed !== 'continuous' && self.feed !== 'longpoll')
throw new Error('The only valid feed options are "continuous" and "longpoll"');

Expand All @@ -99,17 +105,20 @@ Feed.prototype.confirm = function confirm_feed() {

self.db_safe = lib.scrub_creds(self.db);

self.log.debug('Checking database: ' + self.db_safe);
var endpoint = self.is_db_updates ? 'server' : 'database';

self.log.debug('Checking ' + endpoint + ': ' + self.db_safe);

var confirm_timeout = self.heartbeat * 3; // Give it time to look up the name, connect, etc.
var timeout_id = setTimeout(function() {
return self.die(new Error('Timeout confirming database: ' + self.db_safe));
return self.die(new Error('Timeout confirming ' + endpoint + ': ' + self.db_safe));
}, confirm_timeout);

var headers = lib.JP(lib.JS(self.headers));
headers.accept = 'application/json';

var req = {'uri':self.db, 'headers':headers}
var uri = self.is_db_updates ? self.db.replace(/\/_db_updates$/, '') : self.db;
var req = {'uri':uri, 'headers':headers}
Object.keys(self.request).forEach(function(key) {
req[key] = self.request[key];
})
Expand All @@ -130,11 +139,16 @@ Feed.prototype.confirm = function confirm_feed() {
return self.emit('error', json_er)
}

if(!self.dead && (!db.db_name || !db.instance_start_time))
if(!self.is_db_updates && !self.dead && (!db.db_name || !db.instance_start_time))
return self.emit('error', new Error('Bad DB response: ' + body));

self.original_db_seq = db.update_seq
self.log.debug('Confirmed db: ' + self.db_safe);
if(self.is_db_updates && !self.dead && !db.couchdb)
return self.emit('error', new Error('Bad server response: ' + body));

if (!self.is_db_updates)
self.original_db_seq = db.update_seq

self.log.debug('Confirmed ' + endpoint + ': ' + self.db_safe);
self.emit('confirm', db);

if(self.since == 'now') {
Expand Down Expand Up @@ -186,7 +200,7 @@ Feed.prototype.query = function query_feed() {
if(query_params.feed == 'longpoll' && (!query_params.limit || query_params.limit > poll_size))
query_params.limit = poll_size;

var feed_url = self.db + '/_changes?' + querystring.stringify(query_params);
var feed_url = self.db + (self.is_db_updates ? '' : '/_changes') + '?' + querystring.stringify(query_params);

self.headers.accept = self.headers.accept || 'application/json';
var req = { method : 'GET'
Expand Down Expand Up @@ -418,15 +432,15 @@ Feed.prototype.on_couch_data = function on_couch_data(change) {
change = JSON.parse(change)

//self.log.debug('Object:\n' + util.inspect(change));
if('last_seq' in change) {
if(!self.is_db_updates && 'last_seq' in change) {
self.log.warn('Stopping upon receiving a final message: ' + JSON.stringify(change))
var del_er = new Error('Database deleted after change: ' + change.last_seq)
del_er.deleted = true
del_er.last_seq = change.last_seq
return self.die(del_er)
}

if(!change.seq)
if(!self.is_db_updates && !change.seq)
return self.die(new Error('Change has no .seq field: ' + JSON.stringify(change)))

self.on_change(change)
Expand Down Expand Up @@ -531,15 +545,15 @@ Feed.prototype.die = function(er) {
Feed.prototype.on_change = function on_change(change) {
var self = this;

if(!change.seq)
if(!self.is_db_updates && !change.seq)
return self.die(new Error('No seq value in change: ' + lib.JS(change)));

if(change.seq == self.since) {
if(!self.is_db_updates && change.seq == self.since) {
self.log.debug('Bad seq value ' + change.seq + ' since=' + self.since);
return destroy_req(self.pending.request);
}

if(!self.caught_up && change.seq == self.original_db_seq) {
if(!self.is_db_updates && !self.caught_up && change.seq == self.original_db_seq) {
self.caught_up = true
self.emit('catchup', change.seq)
}
Expand Down Expand Up @@ -581,7 +595,10 @@ Feed.prototype.on_good_change = function on_good_change(change) {
self.inactivity_timer = setTimeout(function() { self.on_inactivity() }, self.inactivity_ms);

self.change_at = new Date;
self.since = change.seq;

if(!self.is_db_updates)
self.since = change.seq;

self.emit('change', change);
}

Expand Down