StreamQueue.js

var StreamQueue = (function () {

    function StreamQueue() {
        // The factory
        this.nextTaskFactory = null;
        // The current task
        this.currentTask = null;

        // Tasks which must have end() called on them.
        this.tasksToEnd = [];
    }
    
    /**
     * 
     * @param taskFactory A function which should return the next task.
     */
    StreamQueue.prototype.currentTaskEnded = function () {

        gUtil.log('Finished');

        this.currentTask = null;
        if (this.nextTaskFactory) {
            var nextTask = this.nextTaskFactory();
            
            this.nextTaskFactory = null;
            this.currentTask = nextTask;

            // When we were queuing tasks, we were returning
            // streams. We must signify 'end' on all those streams, when
            // the now current task has finished.
            var tasksToEnd = [].concat(this.tasksToEnd);
            this.tasksToEnd = [];

            var _this = this;
            this.currentTask.on('end', function () {
                // Signify to everything that was listening to the 'next stream' that it's finished now.
                _.each(tasksToEnd, function (task) {
                    task.end();
                });
                tasksToEnd = [];

                _this.currentTaskEnded();
            });
        }
    };

    /**
     * 
     * @param taskFactory A function which should return the next task.
     */
    StreamQueue.prototype.queueTask = function (taskFactory) {
        var _this = this;
        if (!this.currentTask) {
            gUtil.log('Starting');
            this.currentTask = taskFactory();
            var currentTaskEnded = function () {
                _this.currentTaskEnded();
            };
            this.currentTask
                .on('end', currentTaskEnded);
            return this.currentTask;
        } else {
            gUtil.log('Task already in progress - queued next task.');
            if (!this.nextTaskFactory) {
                this.nextTaskFactory = taskFactory;
            }

            var ws = Writable();
            this.tasksToEnd.push(ws);

            return ws;
        }
    };

    return StreamQueue;
})();