dispatcher.js

var _          = require('underscore'),
    redis      = require('redis'),
    redis_conf = require('../../servers/config/redis.json'),
    cached     = {},

    Dispatcher = function(key) {
        var self = this,
            key_base = 'app:dispatcher:';

        this.key = key_base + key;

        this._pub = redis.createClient(redis_conf.port, redis_conf.host, redis_conf.options),
        this._sub = redis.createClient(redis_conf.port, redis_conf.host, redis_conf.options);
        this._sub.setMaxListeners(0);
        this._sub.subscribe(this.key);

        /* --- Subscribe to Redis only once ---*/

        this._sub.on('message', function(ch, data) {
            var payload = JSON.parse(data),
                key = payload.evt + ':' + String(payload.id);

            if (cached[key] && cached[key].length) {
                for (var i = 0, j = cached[key].length; i < j; i++) {
                    cached[key][i](payload.data);
                }
            }
        });
    };

Dispatcher.prototype.emit = function(evt, id, data) {
    var payload = {
        evt  : evt,
        id   : id,
        data : data
    };
    this._pub.publish(this.key, JSON.stringify(payload));
}

Dispatcher.prototype.on = function(evt, id, callback) {
    var key = evt + ':' + String(id);
    cached[key] = cached[key] || [];
    cached[key].push(callback);
}

Dispatcher.prototype.once = function(evt, id, callback) {
    var key = evt + ':' + String(id);
    delete cached[key];
    cached[key] = [];
    cached[key].push(callback);
}

Dispatcher.prototype.off = function(evt, id) {
    var key = evt + ':' + String(id);
    delete cached[key];
}

Dispatcher.prototype.flush = function() {
    cached = {};
}

module.exports.devices  = new Dispatcher('devices');
module.exports.sputniks = new Dispatcher('sputniks');