Deletes old jobs, addresses search term race conditions, and adds "indexing"

.gitignore

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

History.md

@@ -1,4 +1,22 @@
 
+2013-08-10
+==================
+  * Allows isOutstanding check for jobs
+  * Exposes ability to get job by id
+  * Stores outstanding jobs by 'key'
+  * Removes default text indexing
+  * Added "outstanding" jobs search
+  * Added multi-state search
+
+2013-07-13
+==================
+  * Added expiration options
+
+2013-07-12
+==================
+  * Added indexing options
+
+
 0.6.2 / 2013-04-03 
 ==================
 

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`.
  *
@@ -92,8 +108,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);
 };
 
 /**
@@ -196,6 +212,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;
+};
+
+
 /**
  * Get the job types present and callback `fn(err, types)`.
  *
@@ -209,6 +258,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)`.
  *
@@ -218,11 +345,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)`.
  *
@@ -248,8 +387,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;
 };
 
@@ -285,6 +434,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.
  */
@@ -317,6 +474,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

@@ -63,17 +63,21 @@ 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);
-    // TODO: abstract this out
-    if ('progress' != msg.event) delete exports.jobs[job.id];
+  if ('indexed' == msg.events) {
+    this.emit(msg.id, 'indexed');
+  } else {
+    var job = exports.jobs[msg.id];
+    if (job) {
+      job.emit.apply(job, msg.args);
+      // TODO: abstract this out
+      if ('progress' != msg.event) delete exports.jobs[job.id];
+    }
+
+    // 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);
 };
 
 /**
@@ -92,5 +96,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);
 };