Sentinel aware redis client.

sentinel.js

/**
Usage example:
var redisConfig = {
    masterName: "mymaster",
    sentinels: [
        {host: "127.0.0.1", port: 26384},
        {host: "127.0.0.1", port: 26383},
        {host: "127.0.0.1", port: 26382}
    ],
};
var RedisMetaClient = require("./../../node_modules/node_redis/sentinel").RedisMetaClient;
	
var redisMetaClient = new RedisMetaClient(redisConfig.sentinels);
var client = redisMetaClient.createClient(redisConfig.masterName);

client.set ...
*/

var reply_to_object = require("./lib/utils").reply_to_object;

var RedisSingleClient = require('./index');

function RedisMetaClient(sentinels, options){
    this.options = options || {};
    this.currentGettersOfMasterConnectionInfo = {};
    this.masterClients = [];
    this.mastersInfos = {};
    this.sentinels = sentinels.slice();
}

RedisMetaClient.prototype.updateMasterConnectionInfoForExistingClients = function(masterName, masterConnectionInfo) {
    var masterClients = this.masterClients[masterName];
    if(!masterClients){
        return;
    }

    for (var i = 0; i < masterClients.length; i++) {
        var masterClient = masterClients[i];
        if(!masterConnectionInfo){
            // No master was found, inform all clients.
            masterClient.flush_and_error("No master found");
            masterClient.enable_offline_queue = false;
        }
        else {
            // If the client has stopped trying, or if the master's host and/or port were wrong, try to reconnect
            if(masterClient.connectionDropped || masterClient.host !== masterConnectionInfo.host || masterClient.port !== masterConnectionInfo.port){
                masterClient.host = masterConnectionInfo.host;
                masterClient.port = masterConnectionInfo.port;
                masterClient.forceReconnectionAttempt();
            }
        }
    }
};

RedisMetaClient.prototype.updateSentinels = function(newSentinels) {
    for (var i = 0; i < newSentinels.length; i++) {
        var newSentinel = newSentinels[i];
        var alreadyThere = false;
        for (var j = 0; j < this.sentinels.length; j++) {
            var thisSentinel = this.sentinels[j];
            if(newSentinel.host === thisSentinel.host && newSentinel.port === thisSentinel.port){
                alreadyThere = true;
            }
        }
        if(!alreadyThere){
            this.sentinels.push(newSentinel);
        }
    }
};

RedisMetaClient.prototype.registerMasterClient = function(masterName, masterClient) {
    var self = this;

    masterClient.masterName = masterName;
    if(!this.masterClients[masterName]){
        this.masterClients[masterName] = [masterClient];
    }
    else {
        this.masterClients[masterName].push(masterClient);
    }

    var errorListener = function(){
        self.getMasterConnectionInfo(masterClient.masterName);
    };

    masterClient.on('error', errorListener);

    masterClient.on('closing', function(){
        masterClient.removeListener('error', errorListener);
        var allMasterClients = self.masterClients[masterName];
        for (var i = 0; i < allMasterClients.length; i++) {
            if(masterClient === allMasterClients[i]){
                allMasterClients.splice(i, 1);
                return;
            }
        }
    });
};

RedisMetaClient.prototype.getMasterConnectionInfo = function(masterName){
    if(this.mastersInfos[masterName].gettingConnection === true){
        return;
    }

    this.mastersInfos[masterName].gettingConnection = true;

    var i = 0;
    var self = this;

    var onProcedureFinished = function(error, masterConnectionInfo){
        if(error){
            // No master was found. Error out all clients that are waiting for master's connection infos
            self.updateMasterConnectionInfoForExistingClients(masterName, false);
        }
        else {
            // A master was found. Update existing clients
            self.updateMasterConnectionInfoForExistingClients(masterName, masterConnectionInfo);
        }

        self.mastersInfos[masterName].gettingConnection = false;
    };

    var askNextSentinelForMaster = function(masterName, sentinelIndex, cb){
        var callbackCalled = false;
        var sentinel = self.sentinels[sentinelIndex];
        if(!sentinel){
            return cb("No such sentinel");
        }
        var sentinelClient = RedisSingleClient.createClient(sentinel.port, sentinel.host, {socket_timeout: 2000});

        sentinelClient.on('error', function(error){
            if(callbackCalled){
                return;
            }
            callbackCalled = true;
            if(sentinelClient.end){
                sentinelClient.end();
            }
            return cb(error);
        });

        sentinelClient.send_command("SENTINEL", ["get-master-addr-by-name", masterName], function(error, newMaster){
            if(callbackCalled){
                return;
            }
            callbackCalled = true;
            if(error){
                if(sentinelClient.end){
                    sentinelClient.end();
                }
                return cb(error);
            }

            var masterConnectionInfo = {
                host: newMaster[0],
                port: newMaster[1]
            };

            // Ask for the full list of sentinels
            sentinelClient.send_command("SENTINEL", ["sentinels", masterName], function(error, allSentinels){
                if(sentinelClient.end){
                    sentinelClient.end();
                }
                if(error){
                    console.log(error);
                }
                else {
                    var newSentinels = [];
                    for (var i = 0; i < allSentinels.length; i++) {
                        var tmpNewSentinel = reply_to_object(allSentinels[i]);
                        newSentinels.push({
                            host: tmpNewSentinel.ip,
                            port: tmpNewSentinel.port
                        });
                    }
                    self.updateSentinels(newSentinels);
                }
            });

            return cb(null, masterConnectionInfo);
        });
    };

    var next = function(){
        if(i >= self.sentinels.length){
            onProcedureFinished("No master found");
        }
        else {
            askNextSentinelForMaster(masterName, i, function(error, masterConnectionInfo){
                if(error){
                    i++;
                    return next();
                }

                // Put this sentinel as the first in the list*
                var thisSentinel = self.sentinels[i];
                self.sentinels.splice(i, 1);
                self.sentinels.unshift(thisSentinel);
                onProcedureFinished(null, masterConnectionInfo);
            });
        }
    };

    next();
};

RedisMetaClient.prototype.createClient = function(masterName, options) {
    // Initialize this.mastersInfos[masterName] if it wasn't already done
    if(this.mastersInfos[masterName] === undefined){
        this.mastersInfos[masterName] = {};
    }

    options = options || {};

    options.allowNoSocket = true;
    options.no_ready_check = true;

    // Create an 'empty client', not connected yet.
    var client = RedisSingleClient.createClient(null, null, options);
    this.registerMasterClient(masterName, client);

    this.getMasterConnectionInfo(masterName);

    return client;
};

module.exports.RedisMetaClient = RedisMetaClient;