queue.js

var _ = require("underscore");

var Queue = module.exports = function() {
  this._queue = new Array();
  this._waitingWorkers = new Array();
}

_.extend(Queue.prototype, {
  clear: function() {
    this._queue = new Array();
  },

  // Subscribes a function to the next available job. Returns true if work was
  // allocated immediately, false otherwise.
  ready: function(cb) {
    if (this._queue.length > 0) {
      var data = this._queue.shift();
      process.nextTick(function() { cb(data) });
      return true;
    } else {
      this._waitingWorkers.push(cb);
      return false;
    }
  },

  // Adds work to be done. If a worker was available to process this, returns
  // true, false otherwise.
  push: function(data) {
    if (this._waitingWorkers.length > 0) {
      var worker = this._waitingWorkers.shift();
      process.nextTick(function() { worker(data) });
      return true;
    } else {
      this._queue.push(data);
      return false;
    }
  },
});

queue_test.js

var should = require("chai").should(),
    sinon  = require("sinon"),
    _      = require("underscore");

var Queue  = require("../queue");

queue = new Queue();

describe("Queue", function(){
  describe("ready", function(){
    describe("for an empty queue", function(){
      it("requires a job to return", function(done){
        queue.ready(done).should.be.false;
        queue.push();
      });
    });
    describe("for a busy queue with no consumers", function(){
      it("takes the next job", function(done){
        queue.push();
        queue.ready(done).should.be.true;
      });
    });
    describe("for a busy queue with a waiting consumer", function(){
      it("takes the second job", function(done){
        queue.ready( function() {/*noop*/} );
        queue.push("foo");
        queue.push();
        queue.ready(done).should.be.true;
      });
    });
  });
});