cartodb/CartoDB-SQL-API
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Share redis channel config

Browse Source
This commit is contained in:
Raul Ochoa 2016-10-11 18:41:59 +02:00
parent d15c7ab0de
commit e7c4ee32df
3 changed files with 15 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
batch/pubsub/channel.js Normal file
View File

@ -0,0 +1,4 @@
module.exports = {
DB: 0,
NAME: 'batch:hosts'
};

14
batch/pubsub/job-publisher.js
View File

@ -1,31 +1,29 @@
'use strict';
var Channel = require('./channel');
var debug = require('./../util/debug')('pubsub:publisher');
var error = require('./../util/debug')('pubsub:publisher:error');
var DB = 0;
function JobPublisher(pool) {
this.pool = pool;
this.channel = 'batch:hosts';
}
JobPublisher.prototype.publish = function (host) {
var self = this;
this.pool.acquire(DB, function (err, client) {
this.pool.acquire(Channel.DB, function (err, client) {
if (err) {
return error('Error adquiring redis client: ' + err.message);
}
client.publish(self.channel, host, function (err) {
self.pool.release(DB, client);
client.publish(Channel.NAME, host, function (err) {
self.pool.release(Channel.DB, client);
if (err) {
return error('Error publishing to ' + self.channel + ':' + host + ', ' + err.message);
return error('Error publishing to ' + Channel.NAME + ':' + host + ', ' + err.message);
}
debug('publish to ' + self.channel + ':' + host);
debug('publish to ' + Channel.NAME + ':' + host);
});
});
};

11
batch/pubsub/job-subscriber.js
View File

@ -1,9 +1,9 @@
'use strict';
var Channel = require('./channel');
var debug = require('./../util/debug')('pubsub:subscriber');
var error = require('./../util/debug')('pubsub:subscriber:error');
var DB = 0;
var SUBSCRIBE_INTERVAL_IN_MILLISECONDS = 10 * 60 * 1000; // 10 minutes
function _subscribe(client, channel, queueSeeker, onMessage, callback) {
@ -35,7 +35,6 @@ function _subscribe(client, channel, queueSeeker, onMessage, callback) {
}
function JobSubscriber(pool, queueSeeker) {
this.channel = 'batch:hosts';
this.pool = pool;
this.queueSeeker = queueSeeker;
}
@ -45,7 +44,7 @@ module.exports = JobSubscriber;
JobSubscriber.prototype.subscribe = function (onMessage, callback) {
var self = this;
this.pool.acquire(DB, function (err, client) {
this.pool.acquire(Channel.DB, function (err, client) {
if (err) {
return error('Error adquiring redis client: ' + err.message);
}
@ -56,12 +55,12 @@ JobSubscriber.prototype.subscribe = function (onMessage, callback) {
_subscribe,
SUBSCRIBE_INTERVAL_IN_MILLISECONDS,
self.client,
self.channel,
Channel.NAME,
self.queueSeeker,
onMessage
);
_subscribe(self.client, self.channel, self.queueSeeker, onMessage, callback);
_subscribe(self.client, Channel.NAME, self.queueSeeker, onMessage, callback);
});
};
@ -69,6 +68,6 @@ JobSubscriber.prototype.subscribe = function (onMessage, callback) {
JobSubscriber.prototype.unsubscribe = function () {
clearInterval(this.seekerInterval);
if (this.client && this.client.connected) {
this.client.unsubscribe(this.channel);
this.client.unsubscribe(Channel.NAME);
}
};