text: /*!
   * kue
   * Copyright (c) 2013 Automattic <behradz@gmail.com>
   * Copyright (c) 2011 LearnBoost <tj@learnboost.com>
   * MIT Licensed
   */
  
  /**
   * Module dependencies.
   */
  
  var EventEmitter = require('events').EventEmitter
    , Worker       = require('./queue/worker')
    , events       = require('./queue/events')
    , Job          = require('./queue/job')
    , Warlock      = require('node-redis-warlock')
    , _            = require('lodash')
    , redis        = require('./redis')
    , noop         = function(){};
  
  /**
   * Expose `Queue`.
   */
  
  exports = module.exports = Queue;
  
  /**
   * Library version.
   */
  
  exports.version = require('../package.json').version;
  
  /**
   * Expose `Job`.
   */
  
  exports.Job = Job;
  
  /**
   * Server instance (that is lazily required)
   */
  
  var app;
  
  /**
   * Expose the server.
   */
  
  Object.defineProperty(exports, 'app', {
    get: function() {
      return app || (app = require('./http'));
    }
  });
  
  /**
   * Expose the RedisClient factory.
   */
  
  exports.redis = redis;
  
  /**
   * Create a new `Queue`.
   *
   * @return {Queue}
   * @api public
   */
  
  exports.createQueue = function( options ) {
    if( !Queue.singleton ) {
      Queue.singleton = new Queue(options);
    }
    events.subscribe();
    return Queue.singleton;
  
  };
  
  /**
   * Store workers
   */
  exports.workers = [];
  
  /**
   * Initialize a new job `Queue`.
   *
   * @api public
   */
  
  function Queue( options ) {
    options     = options || {};
    this.name   = options.name || 'kue';
    this.id = [ 'kue', require('os').hostname(), process.pid ].join(':');
    this._options   = options;
    this.promoter     = null;
    this.workers      = exports.workers;
    this.shuttingDown = false;
    Job.disableSearch = options.disableSearch !== false;
    options.jobEvents !== undefined ? Job.jobEvents = options.jobEvents : '';
    redis.configureFactory(options, this);
    this.client = Worker.client = Job.client = redis.createClient();
  }
  
  /**
   * Inherit from `EventEmitter.prototype`.
   */
  
  Queue.prototype.__proto__ = EventEmitter.prototype;
  
  /**
   * Create a `Job` with the given `type` and `data`.
   *
   * @param {String} type
   * @param {Object} data
   * @return {Job}
   * @api public
   */
  
  Queue.prototype.create =
    Queue.prototype.createJob = function( type, data ) {
      return new Job(type, data);
    };
  
  /**
   * Proxy to auto-subscribe to events.
   *
   * @api public
   */
  
  var on             = EventEmitter.prototype.on;
  Queue.prototype.on = function( event ) {
    if( 0 == event.indexOf('job') ) events.subscribe();
    return on.apply(this, arguments);
  };
  
  /**
   * Promote delayed jobs, checking every `ms`,
   * defaulting to 1 second.
   *
   * @params {Number} ms
   * @deprecated
   */
  
  Queue.prototype.promote = function( ms, l ) {
    console.warn('promote method is deprecated, you don\'t need to call this anymore. You can safely remove it from your code now.');
  };
  
  /**
   * sets up promotion & ttl timers
   */
  
  Queue.prototype.setupTimers = function() {
    if( this.warlock === undefined ) {
      this.lockClient = redis.createClient();
      this.warlock    = new Warlock(this.lockClient);
    }
    this.checkJobPromotion(this._options.promotion);
    this.checkActiveJobTtl(this._options.promotion);
  };
  
  /**
   * This new method is called by Kue when created
   *
   * Promote delayed jobs, checking every `ms`,
   * defaulting to 1 second.
   *
   * @params {Number} ms
   */
  
  Queue.prototype.checkJobPromotion = function( promotionOptions ) {
    promotionOptions = promotionOptions || {};
    var client       = this.client
      , self         = this
      , timeout      = promotionOptions.interval || 1000
      , lockTtl      = promotionOptions.lockTtl || 2000
        //, lockTtl = timeout
      , limit        = promotionOptions.limit || 1000;
    clearInterval(this.promoter);
    this.promoter    = setInterval(function() {
      self.warlock.lock('promotion', lockTtl, function( err, unlock ) {
        if( err ) {
          // Something went wrong and we weren't able to set a lock
          self.emit('error', err);
          return;
        }
        if( typeof unlock === 'function' ) {
          // If the lock is set successfully by this process, an unlock function is passed to our callback.
          client.zrangebyscore(client.getKey('jobs:delayed'), 0, Date.now(), 'LIMIT', 0, limit, function( err, ids ) {
            if( err || !ids.length ) return unlock();
            //TODO do a ZREMRANGEBYRANK jobs:delayed 0 ids.length-1
            var doUnlock = _.after(ids.length, unlock);
            ids.forEach(function( id ) {
              id = client.stripFIFO(id);
              Job.get(id, function( err, job ) {
                if( err ) return doUnlock();
                events.emit(id, 'promotion');
                if( job ) job.inactive(doUnlock);
              });
            });
          });
        } else {
          // The lock was not established by us, be silent
        }
      });
    }, timeout);
  };
  
  
  Queue.prototype.checkActiveJobTtl = function( ttlOptions ) {
    ttlOptions              = ttlOptions || {};
    var client              = this.client
      , self                = this
      , timeout             = ttlOptions.interval || 1000
      , lockTtl             = 2000
      , limit               = ttlOptions.limit || 1000;
    clearInterval(this.activeJobsTtlTimer);
    this.activeJobsTtlTimer = setInterval(function() {
      self.warlock.lock('activeJobsTTL', lockTtl, function( err, unlock ) {
        if( err ) {
          // Something went wrong and we weren't able to set a lock
          self.emit('error', err);
          return;
        }
        if( typeof unlock === 'function' ) {
          // If the lock is set successfully by this process, an unlock function is passed to our callback.
          // filter only jobs set with a ttl (timestamped) between a large number and current time
          client.zrangebyscore(client.getKey('jobs:active'), 100000, Date.now(), 'LIMIT', 0, limit, function( err, ids ) {
            if( err || !ids.length ) return unlock();
  
            var idsRemaining = ids.slice();
            var doUnlock = _.after(ids.length, function(){
              self.removeAllListeners( 'job ttl exceeded ack' );
              waitForAcks && clearTimeout( waitForAcks );
              unlock && unlock();
            });
  
            self.on( 'job ttl exceeded ack', function( id ) {
              idsRemaining.splice( idsRemaining.indexOf( id ), 1 );
              doUnlock();
            });
  
            var waitForAcks = setTimeout( function(){
              idsRemaining.forEach( function( id ){
                id = client.stripFIFO(id);
                Job.get(id, function( err, job ) {
                  if( err ) return doUnlock();
                  job.failedAttempt( { error: true, message: 'TTL exceeded' }, doUnlock );
                });
              });
            }, 1000 );
  
            ids.forEach(function( id ) {
              id = client.stripFIFO(id);
              events.emit(id, 'ttl exceeded');
            });
          });
        } else {
          // The lock was not established by us, be silent
        }
      });
    }, timeout);
  };
  
  /**
   * Runs a LUA script to diff inactive jobs ZSET cardinality
   * and helper pop LIST length each `ms` milliseconds and syncs helper LIST.
   *
   * @param {Number} ms interval for periodical script runs
   * @api public
   */
  
  Queue.prototype.watchStuckJobs = function( ms ) {
    var client = this.client
      , self   = this
      , ms     = ms || 1000;
    var prefix = this.client.prefix;
  
    if( this.client.constructor.name == 'Redis'  || this.client.constructor.name == 'Cluster') {
      // {prefix}:jobs format is needed in using ioredis cluster to keep they keys in same node
      prefix = '{' + prefix + '}';
    }
    var script =
          'local msg = redis.call( "keys", "' + prefix + ':jobs:*:inactive" )\n\
          local need_fix = 0\n\
          for i,v in ipairs(msg) do\n\
            local queue = redis.call( "zcard", v )\n\
            local jt = string.match(v, "' + prefix + ':jobs:(.*):inactive")\n\
            local pending = redis.call( "LLEN", "' + prefix + ':" .. jt .. ":jobs" )\n\
            if queue > pending then\n\
              need_fix = need_fix + 1\n\
              for j=1,(queue-pending) do\n\
                redis.call( "lpush", "' + prefix + ':"..jt..":jobs", 1 )\n\
              end\n\
            end\n\
          end\n\
          return need_fix';
    clearInterval(this.stuck_job_watch);
    client.script('LOAD', script, function( err, sha ) {
      if( err ) {
        return self.emit('error', err);
      }
      this.stuck_job_watch = setInterval(function() {
        client.evalsha(sha, 0, function( err, fixes ) {
          if( err ) return clearInterval(this.stuck_job_watch);
        }.bind(this));
      }.bind(this), ms);
  
    }.bind(this));
  };
  
  /**
   * Get setting `name` and invoke `fn(err, res)`.
   *
   * @param {String} name
   * @param {Function} fn
   * @return {Queue} for chaining
   * @api public
   */
  
  Queue.prototype.setting = function( name, fn ) {
    fn = fn || noop;
    this.client.hget(this.client.getKey('settings'), name, fn);
    return this;
  };
  
  /**
   * Process jobs with the given `type`, invoking `fn(job)`.
   *
   * @param {String} type
   * @param {Number|Function} n
   * @param {Function} fn
   * @api public
   */
  
  Queue.prototype.process = function( type, n, fn ) {
    var self = this;
  
    if( 'function' == typeof n ) fn = n, n = 1;
  
    while( n-- ) {
      var worker = new Worker(this, type).start(fn);
      worker.id  = [ self.id, type, self.workers.length + 1 ].join(':');
      worker.on('error', function( err ) {
        self.emit('error', err);
      });
      worker.on('job complete', function( job ) {
        // guard against emit after shutdown
        if( self.client ) {
          self.client.incrby(self.client.getKey('stats:work-time'), job.duration, noop);
        }
      });
      // Save worker so we can access it later
      self.workers.push(worker);
    }
    this.setupTimers();
  };
  
  /**
   * Graceful shutdown
   *
   * @param {Number} timeout in milliseconds to wait for workers to finish
   * @param {String} type specific worker type to shutdown
   * @param {Function} fn callback
   * @return {Queue} for chaining
   * @api public
   */
  
  Queue.prototype.shutdown = function( timeout, type, fn ) {
    var self = this
      , n    = self.workers.length;
    if( arguments.length === 1 ) {
      fn      = timeout;
      type    = '';
      timeout = null;
    } else if( arguments.length === 2 ) {
      fn   = type;
      type = '';
    }
    var origFn = fn || function() {
      };
  
    if( this.shuttingDown && type === '' ) { // a global shutdown already has been called
      return fn(new Error('Shutdown already in progress'));
    }
  
    if( type === '' ) { // this is a global shutdown call
      this.shuttingDown = true;
    }
  
    var cleanup = function() {
      if( self.shuttingDown ) {
        self.workers    = [];
        exports.workers = [];
        self.removeAllListeners();
        Queue.singleton = null;
        events.unsubscribe();
        // destroy redis client and pubsub
        redis.reset();
        self.client && self.client.quit();
        self.client = null;
        self.lockClient && self.lockClient.quit();
        self.lockClient = null;
      }
    };
  
    // Wrap `fn` to only call after all workers finished
    fn = function( err ) {
      if( err ) {
        return origFn(err);
      }
      if( !--n ) {
        cleanup();
        origFn.apply(null, arguments);
      }
    };
  
    // shut down promoter interval
    if( self.shuttingDown ) {
      if( self.promoter ) {
        clearInterval(self.promoter);
        self.promoter = null;
      }
      if( self.activeJobsTtlTimer ) {
        clearInterval(self.activeJobsTtlTimer);
        self.activeJobsTtlTimer = null;
      }
  
    }
  
    if( !self.workers.length ) {
      cleanup();
      origFn();
    } else {
      // Shut down workers 1 by 1
      self.workers.forEach(function( worker ) {
        if( self.shuttingDown || worker.type == type ) {
          worker.shutdown(timeout, fn);
        } else {
          fn && fn();
        }
      });
    }
  
    return this;
  };
  
  /**
   * Get the job types present and callback `fn(err, types)`.
   *
   * @param {Function} fn
   * @return {Queue} for chaining
   * @api public
   */
  
  Queue.prototype.types = function( fn ) {
    fn = fn || noop;
    this.client.smembers(this.client.getKey('job:types'), fn);
    return this;
  };
  
  /**
   * Return job ids with the given `state`, and callback `fn(err, ids)`.
   *
   * @param {String} state
   * @param {Function} fn
   * @return {Queue} for chaining
   * @api public
   */
  
  Queue.prototype.state = function( state, fn ) {
    var self = this;
    this.client.zrange(this.client.getKey('jobs:' + state), 0, -1, function(err,ids){
      var fixedIds = [];
      ids.forEach(function(id){
          fixedIds.push(self.client.stripFIFO(id));
        });
      fn(err,fixedIds);
    });
    return this;
  };
  
  /**
   * Get queue work time in milliseconds and invoke `fn(err, ms)`.
   *
   * @param {Function} fn
   * @return {Queue} for chaining
   * @api public
   */
  
  Queue.prototype.workTime = function( fn ) {
    this.client.get(this.client.getKey('stats:work-time'), function( err, n ) {
      if( err ) return fn(err);
      fn(null, parseInt(n, 10));
    });
    return this;
  };
  
  /**
   * Get cardinality of jobs with given `state` and `type` and callback `fn(err, n)`.
   *
   * @param {String} type
   * @param {String} state
   * @param {Function} fn
   * @return {Queue} for chaining
   * @api public
   */
  
  Queue.prototype.cardByType = function( type, state, fn ) {
    fn = fn || noop;
    this.client.zcard(this.client.getKey('jobs:' + type + ':' + state), fn);
    return this;
  };
  
  /**
   * Get cardinality of `state` and callback `fn(err, n)`.
   *
   * @param {String} state
   * @param {Function} fn
   * @return {Queue} for chaining
   * @api public
   */
  
  Queue.prototype.card = function( state, fn ) {
    fn = fn || noop;
    this.client.zcard(this.client.getKey('jobs:' + state), fn);
    return this;
  };
  
  /**
   * Completed jobs.
   * @param {Function} fn
   * @return {Queue} for chaining
   * @api public
   */
  
  Queue.prototype.complete = function( fn ) {
    return this.state('complete', fn);
  };
  
  /**
   * Failed jobs.
   * @param {Function} fn
   * @return {Queue} for chaining
   * @api public
   */
  
  Queue.prototype.failed = function( fn ) {
    return this.state('failed', fn);
  };
  
  /**
   * Inactive jobs (queued).
   * @param {Function} fn
   * @return {Queue} for chaining
   * @api public
   */
  
  Queue.prototype.inactive = function( fn ) {
    return this.state('inactive', fn);
  };
  
  /**
   * Active jobs (mid-process).
   * @param {Function} fn
   * @return {Queue} for chaining
   * @api public
   */
  
  Queue.prototype.active = function( fn ) {
    return this.state('active', fn);
  };
  
  /**
   * Delayed jobs.
   * @param {Function} fn
   * @return {Queue} for chaining
   * @api public
   */
  
  Queue.prototype.delayed = function( fn ) {
    return this.state('delayed', fn);
  };
  
  /**
   * Completed jobs of type `type` count.
   * @param {String} type is optional
   * @param {Function} fn
   * @return {Queue} for chaining
   * @api public
   */
  
  Queue.prototype.completeCount = function( type, fn ) {
    if( 1 == arguments.length ) {
      fn = type;
      return this.card('complete', fn);
    }
    return this.cardByType(type, 'complete', fn);
  };
  
  
  /**
   * Failed jobs of type `type` count.
   * @param {String} type is optional
   * @param {Function} fn
   * @return {Queue} for chaining
   * @api public
   */
  
  Queue.prototype.failedCount = function( type, fn ) {
    if( 1 == arguments.length ) {
      fn = type;
      return this.card('failed', fn);
    }
    return this.cardByType(type, 'failed', fn);
  };
  
  /**
   * Inactive jobs (queued) of type `type` count.
   * @param {String} type is optional
   * @param {Function} fn
   * @return {Queue} for chaining
   * @api public
   */
  
  Queue.prototype.inactiveCount = function( type, fn ) {
    if( 1 == arguments.length ) {
      fn = type;
      return this.card('inactive', fn);
    }
    return this.cardByType(type, 'inactive', fn);
  };
  
  /**
   * Active jobs (mid-process) of type `type` count.
   * @param {String} type is optional
   * @param {Function} fn
   * @return {Queue} for chaining
   * @api public
   */
  
  Queue.prototype.activeCount = function( type, fn ) {
    if( 1 == arguments.length ) {
      fn = type;
      return this.card('active', fn);
    }
    return this.cardByType(type, 'active', fn);
  };
  
  /**
   * Delayed jobs of type `type` count.
   * @param {String} type is optional
   * @param {Function} fn
   * @return {Queue} for chaining
   * @api public
   */
  
  Queue.prototype.delayedCount = function( type, fn ) {
    if( 1 == arguments.length ) {
      fn = type;
      return this.card('delayed', fn);
    }
    return this.cardByType(type, 'delayed', fn);
  };
  
  /**
   * Test mode for convenience in test suites
   * @api public
   */
  
  Queue.prototype.testMode = require('./queue/test_mode');
  