Queue

queue.js

/*jslint node: true, maxerr: 50, indent: 4 */

(function () {
    "use strict";
    var EventEmitter = require('events').EventEmitter,
        Queue = module.exports = function Queue() {
        /*  Constructor is doing work..?? 
        */
        var _emitter    = this._emitter     = new EventEmitter(),
            _awaiting   = this._awaiting    = [],
            _queue      = this._queue       = [];
        
        _emitter.on('offer', function (data) {
            _awaiting.shift()(data);
        });
        
    };
    
    /**
     * Delete queue.
     */
    Queue.prototype.clear = function () {
        this._queue = [];
        this._awaiting = [];
        this._emitter.removeAllListeners();
    };
        
    /**
     * Return number of elements in queue.
     */
    Queue.prototype.size = function () {
        return this._queue.length;
    };
    
    /**
     *  Returns number of awaiting request for data from queue.
     */
    Queue.prototype.awaiting = function () {
        return this._awaiting.length;
    };
    
    /**
     *  Insert element into queue
     */    
    Queue.prototype.offer = function (data) {
        this._queue.push(data);
        
        if (this._awaiting.length > 0) {
            this._emitter.emit('offer', data);                
        }
    };
    
    /**
     *  Look at element in front of queue, but don't remove it
     */    
    Queue.prototype.peek = function () {
        return this._queue[0];
    };
    
    /**
     *  Retrieve element in front of queue and removes it.
     */
    Queue.prototype.poll = function () {
        return this._queue.shift();
    };
    
    /**
     *  Is blocking.
     */
    Queue.prototype.take = function (callback) {
        var element = this.poll();
        if (element) {
            callback(element);
        } else { // No element so wait for it.
            this._awaiting.push(callback);
        }
    };

})();

queue.test.js

/*jslint node: true, maxerr: 50, indent: 4 */
/*global describe, it, beforeEach*/
(function () {
    "use strict";

    var BASE_PATH = process.env.MOCHA_COV ?
        './../lib-cov/' //: 
        : './../lib/',
        should  = require('should'),
        _       = require('underscore'),
        Queue   = require(BASE_PATH + 'queue.js');
        
    describe("Queue", function () {
        var queue = null,
            firstElement    = "First Element",
            secondElement   = "Second Element";
            
        beforeEach(function () {
            queue = new Queue();
        });
        
        it("Should be empty when constructed", function () {
            queue.size().should.equal(0);
        });
        
        it("Should increase size by one when adding item", function () {
            queue.offer(firstElement);
            queue.size().should.equal(1);
        });
        
        it("Should be able to clear queue", function () {
            queue.offer(firstElement);
            queue.clear();
            queue.size().should.equal(0);
        });
        
        it("Should not retrieve element when inserted afterwards", function () {
            should.not.exist(queue.poll());
            queue.offer(firstElement);
        });
        
        it("Should decrease size when element has been popped", function () {
            queue.offer(firstElement);
            queue.poll();
            queue.size().should.equal(0);
        });
        
        it("Should be able to pop element which has been inserted", function () {
            queue.offer(firstElement);
            queue.poll().should.equal(firstElement);
        });
        
        it("should retrieve first elements inserted on multiple insertions(FIFO)", function () {
            queue.offer(firstElement);
            queue.offer(secondElement);
            queue.poll().should.equal(firstElement);
        });
        
        it("should retrieve second element next", function () {
            queue.offer(firstElement);
            queue.offer(secondElement);
            queue.poll();
            queue.poll().should.equal(secondElement);
        });
        
        it("Peeking should not remove element, but retrieve first element", function () {
            queue.offer(firstElement);
            queue.peek().should.equal(firstElement);
            queue.size().should.equal(1);
        });
        
        it("Peeking should return same first element on multiple calls", function () {
            queue.offer(firstElement);
            queue.offer(secondElement);
            
            queue.peek();
            queue.peek().should.equal(firstElement);
        });
             
        it("should have blocking peek which returns immediately if contains element", function (done) {
            queue.offer(firstElement);
            
            queue.take(function (data) {
                data.should.equal(firstElement);
                done();
            });
        });
        
        it("Take should not return element when queue is empty", function (done) {
            queue.take(function (data) {
                should.fail();
            });
            
            process.nextTick(function () {
                done();
            });
        });
        
        it("should have waiting awaiting clients when no element is available", function (done) {
            queue.take(function (data) {
            
            });
            
            process.nextTick(function () {
                queue.awaiting().should.equal(1);
                done();        
            });
        });
        
        it("should have blocking peek which `blocks` if no element available", function (done) {
            queue.take(function (data) {
                data.should.equal(firstElement);
                done();
            });
            
            process.nextTick(function () {
                queue.offer(firstElement);
            });
        });
        
        it("multiple blocking fifo order", function (done) {
            queue.take(function (data) {
                data.should.equal(firstElement);
            });
            
            queue.take(function (data) {
                data.should.equal(secondElement);
                done();
            });
            
            process.nextTick(function () {
                queue.offer(firstElement);
                queue.offer(secondElement);
            });
        });
    });
})();