Merge of high priority issue fixes and pull requests on top of version 0.6.2

.gitignore

@@ -1,5 +1,7 @@
+.idea
 .DS_Store
 node_modules
 *.sock
-testing
 lib/http/public/stylesheets/main.css
+*.rdb
+test/incomplete

lib/kue.js

@@ -13,7 +13,10 @@ var EventEmitter = require('events').EventEmitter
  , Worker = require('./queue/worker')
  , events = require('./queue/events')
  , Job = require('./queue/job')
- , redis = require('./redis');
+ , redis = require('./redis')
+ , reds = require('reds')
+ , async = require('async');
+
 
 /**
  * Expose `Queue`.
@@ -55,6 +58,19 @@ Object.defineProperty(exports, 'app', {
 
 exports.redis = redis;
 
+
+/**
+ * Search instance.
+ */
+
+var search;
+function getSearch() {
+ if (search) return search;
+ reds.createClient = require('./redis').createClient;
+ return search = reds.createSearch('q:search');
+};
+
+
 /**
  * Create a new `Queue`.
  *
@@ -98,8 +114,8 @@ Queue.prototype.__proto__ = EventEmitter.prototype;
  */
 
 Queue.prototype.create =
-Queue.prototype.createJob = function(type, data){
- return new Job(type, data);
+Queue.prototype.createJob = function(type, data, options){
+ return new Job(type, data, options);
 };
 
 /**
@@ -205,6 +221,39 @@ Queue.prototype.process = function(type, n, fn){
  }
 };
 
+
+/**
+ * Initializes garbage collection for jobs of `type` after `n` milliseconds.
+ * Currently, there is a 1000 millisecond minimum (this should not be primary form of removal)
+ *
+ * @param {String} type
+ * @param {Number} n
+ * @api public
+ */
+
+Queue.prototype.setExpiration = function(type, n){
+ var self = this,
+ client = this.client,
+ n = Math.max(n, 1000);
+
+
+ setInterval(function() {
+
+ var cutoff = Math.floor((new Date().getTime() - n) / 1000);
+
+ self.client.zrangebyscore('q:expiring:' + type, 0, cutoff, function(err, results) {
+
+ if (!err && results.length)
+ for (var i = 0; i < results.length; ++i)
+ Job.remove(results[i]);
+
+ });
+ }, n);
+
+ return this;
+};
+
+
 /**
  * Graceful shutdown
  *
@@ -248,6 +297,84 @@ Queue.prototype.types = function(fn){
  return this;
 };
 
+
+/**
+ * Enables job retrieval by id
+ *
+ * @param {Number} id
+ * @param {Function} fn
+ * @api public
+ */
+Queue.prototype.get =
+Queue.prototype.getById = function(id, fn){
+ Job.get(id, fn);
+};
+
+
+/**
+ * Determines whether a job is outstanding. If input is number,
+ * taken to be a Job ID. If a string, taken to be a type + key. If an object
+ * the key should be either id or key
+ *
+ * @param {Number} args or {Object}
+ * @param {Function} fn
+ * @api public
+ */
+
+Queue.prototype.isOutstanding = function(args, fn){
+ var isKey = false,
+ val = null;
+
+ if ("object" == typeof args) {
+ if ("undefined" != typeof args.id) {
+ val = args.id;
+ } else if ("undefined" != typeof args.key && "undefined" != typeof args.type) {
+ isKey = true;
+ } else {
+ return fn(new Error("No key or id specified for outstanding check."))
+ }
+ } else if ("number" == typeof args) {
+ val = args;
+ } else {
+ return fn(new Error("No key or id specified for outstanding check."))
+ }
+
+ if (isKey)
+ this.client.sismember("q:outstanding:" + args.type, args.key, function(err, isMember) {
+ if (err)
+ return fn(err, null);
+ else
+ return fn(null, 1 == isMember); 
+ });
+ else {
+ this.get(val, function(err, job) {
+ if (err)
+ return fn(err, false);
+ else {
+ var state = job.state();
+ return fn(err, ('active' == state || 'inactive' == state));
+ }
+ });
+ }
+};
+
+
+/**
+ * Find jobs by a series of identifiers and callback `fn(err)`.
+ *
+ * @param {String} lookup
+ * @param {Function} fn
+ * @api public
+ */
+
+Queue.prototype.find = function(query, fn){
+ getSearch().query(query).end(function(err, ids){
+ fn(err, ids);
+ }, 'and');
+};
+
+
+
 /**
  * Return job ids with the given `state`, and callback `fn(err, ids)`.
  *
@@ -257,11 +384,23 @@ Queue.prototype.types = function(fn){
  * @api public
  */
 
-Queue.prototype.state = function(state, fn){
- this.client.zrange('q:jobs:' + state, 0, -1, fn);
+Queue.prototype.state = 
+Queue.prototype.states = function(states, fn){
+ if ('string' == typeof states)
+ this.client.zrange('q:jobs:' + states, 0, -1, fn);
+ else {
+ var self = this;
+ function getByState(state, cb) { self.client.zrange('q:jobs:' + state, 0, -1, cb); }
+ async.map(states, getByState, function(err, ids) {
+ self = null;
+ fn(err, Array.prototype.concat.apply([], ids));
+ });
+ }
  return this;
 };
 
+
+
 /**
  * Get queue work time in milliseconds and invoke `fn(err, ms)`.
  *
@@ -287,8 +426,18 @@ Queue.prototype.workTime = function(fn){
  * @api public
  */
 
-Queue.prototype.card = function(state, fn){
- this.client.zcard('q:jobs:' + state, fn);
+Queue.prototype.card = 
+Queue.prototype.cards = function(states, fn){
+ if ('string' == typeof states)
+ this.client.zcard('q:jobs:' + states, fn);
+ else {
+ var self = this;
+ function countByState(state, cb) { self.client.zcard('q:jobs:' + state, cb); }
+ async.map(states, countByState, function(err, counts) {
+ self = null;
+ fn(err, counts.reduce(function(memo, num) { return memo + num; }, 0));
+ });
+ }
  return this;
 };
 
@@ -324,6 +473,14 @@ Queue.prototype.active = function(fn){
  return this.state('active', fn);
 };
 
+/**
+ * Oustanding jobs (inactive or active).
+ */
+
+Queue.prototype.outstanding = function(fn){
+ return this.state(['active','inactive'], fn);
+};
+
 /**
  * Completed jobs count.
  */
@@ -356,6 +513,14 @@ Queue.prototype.activeCount = function(fn){
  return this.card('active', fn);
 };
 
+/**
+ * Outstanding jobs (active or inactive).
+ */
+
+Queue.prototype.outstandingCount = function(fn){
+ return this.card(['active','inactive'], fn);
+};
+
 /**
  * Delayed jobs.
  */

lib/queue/events.js

@@ -74,16 +74,20 @@ exports.onMessage = function(channel, msg){
  var msg = JSON.parse(msg);
 
  // map to Job when in-process
- var job = exports.jobs[msg.id];
- if (job) {
- job.emit.apply(job, msg.args);
- if (['complete', 'failed'].indexOf(msg.event) !== -1) exports.remove(job);
+ if ('indexed' == msg.events) {
+ this.emit(msg.id, 'indexed');
+ } else {
+ var job = exports.jobs[msg.id];
+ if (job) {
+ job.emit.apply(job, msg.args);
+ if (['complete', 'failed'].indexOf(msg.event) !== -1) exports.remove(job);
+ }
+
+ // emit args on Queues
+ msg.args[0] = 'job ' + msg.args[0];
+ msg.args.push(msg.id);
+ exports.queue.emit.apply(exports.queue, msg.args);
  }
-
- // emit args on Queues
- msg.args[0] = 'job ' + msg.args[0];
- msg.args.push(msg.id);
- exports.queue.emit.apply(exports.queue, msg.args);
 };
 
 /**
@@ -102,5 +106,9 @@ exports.emit = function(id, event) {
  , event: event
  , args: [].slice.call(arguments, 1)
  });
- client.publish(exports.key, msg);
+
+ if ('indexed' == event)
+ client.publish('q:indexing', msg);
+ else
+ client.publish(exports.key, msg);
 };