text: var util = require('util'),
    Lock = require('../base'),
    _ = require('lodash'),
    async = require('async'),
    redis = Lock.use('redis');
  
  function Redis(options) {
    Lock.call(this, options);
  
    var defaults = {
      host: 'localhost',
      port: 6379,
      prefix: 'aggregatelock',
      retry_strategy: function (options) {
        return undefined;
      }
      // heartbeat: 60 * 1000
    };
  
    _.defaults(options, defaults);
  
    if (options.url) {
      var url = require('url').parse(options.url);
      if (url.protocol === 'redis:') {
        if (url.auth) {
          var userparts = url.auth.split(':');
          options.user = userparts[0];
          if (userparts.length === 2) {
            options.password = userparts[1];
          }
        }
        options.host = url.hostname;
        options.port = url.port;
        if (url.pathname) {
          options.db = url.pathname.replace('/', '', 1);
        }
      }
    }
  
    this.options = options;
  }
  
  util.inherits(Redis, Lock);
  
  _.extend(Redis.prototype, {
  
    connect: function (callback) {
      var self = this;
  
      var options = this.options;
  
      this.client = new redis.createClient(options.port || options.socket, options.host, _.omit(options, 'prefix'));
  
      this.prefix = options.prefix;
  
      var calledBack = false;
  
      if (options.password) {
        this.client.auth(options.password, function(err) {
          if (err && !calledBack && callback) {
            calledBack = true;
            if (callback) callback(err, self);
            return;
          }
          if (err) throw err;
        });
      }
  
      if (options.db) {
        this.client.select(options.db);
      }
  
      this.client.on('end', function () {
        self.disconnect();
        self.stopHeartbeat();
      });
  
      this.client.on('error', function (err) {
        console.log(err);
  
        if (calledBack) return;
        calledBack = true;
        if (callback) callback(null, self);
      });
  
      this.client.on('connect', function () {
        if (options.db) {
          self.client.send_anyways = true;
          self.client.select(options.db);
          self.client.send_anyways = false;
        }
  
        self.emit('connect');
  
        if (self.options.heartbeat) {
          self.startHeartbeat();
        }
  
        if (calledBack) return;
        calledBack = true;
        if (callback) callback(null, self);
      });
    },
  
    stopHeartbeat: function () {
      if (this.heartbeatInterval) {
        clearInterval(this.heartbeatInterval);
        delete this.heartbeatInterval;
      }
    },
  
    startHeartbeat: function () {
      var self = this;
  
      var gracePeriod = Math.round(this.options.heartbeat / 2);
      this.heartbeatInterval = setInterval(function () {
        var graceTimer = setTimeout(function () {
          if (self.heartbeatInterval) {
            console.error((new Error ('Heartbeat timeouted after ' + gracePeriod + 'ms (redis)')).stack);
            self.disconnect();
          }
        }, gracePeriod);
  
        self.client.ping(function (err) {
          if (graceTimer) clearTimeout(graceTimer);
          if (err) {
            console.error(err.stack || err);
            self.disconnect();
          }
        });
      }, this.options.heartbeat);
    },
  
    disconnect: function (callback) {
      this.stopHeartbeat();
  
      if (this.client) {
        this.client.end(true);
      }
      this.emit('disconnect');
      if (callback) callback(null, this);
    },
  
    getNewId: function(callback) {
      this.client.incr('nextItemId:' + this.prefix, function(err, id) {
        if (err) {
          return callback(err);
        }
        callback(null, id.toString());
      });
    },
  
    reserve: function(workerId, aggregateId, callback) {
      var prefixedId = this.prefix + ':' + aggregateId;
  
      this.client.rpush(prefixedId, workerId, function (err) {
        if (callback) callback(err);
      });
    },
  
    getAll: function(aggregateId, callback) {
      var prefixedId = this.prefix + ':' + aggregateId;
  
      this.client.lrange(prefixedId, 0, -1, callback);
    },
  
    resolve: function(aggregateId, callback) {
      var prefixedId = this.prefix + ':' + aggregateId;
  
      this.client.del(prefixedId, function (err) {
        if (callback) callback(err);
      });
    },
  
    clear: function (callback) {
      var self = this;
      async.parallel([
        function (callback) {
          self.client.del('nextItemId:' + self.options.prefix, callback);
        },
        function (callback) {
          self.client.keys(self.options.prefix + ':*', function(err, keys) {
            if (err) {
              return callback(err);
            }
            async.each(keys, function (key, callback) {
              self.client.del(key, callback);
            }, callback);
          });
        }
      ], function (err) {
        if (callback) callback(err);
      });
    }
  
  });
  
  module.exports = Redis;
  