Freestyle Sparring

// interval processor's job is to run a batch of callback items, one at a time. It runs these batches
// at a given interval, and will continue to run batches either forever or until a maxBatches value is reached.
var Processor = function(interval)
{
    this.nextTick = true;
    this.interval = interval;
    this.timeout = 50;
    this.items = [];
    this.maxBatch = null;
    this.events = {error: [], batchStart: [], batchEnd: [], stopped: [], timeout: []};
    console.log('Interval Processor created with ' + interval + 'ms interval');
}

Processor.prototype.on = function on(event, callback)
{
    this.events[event].push(callback);
}

// returns a new function that ensures that the passed in fn only gets called once no matter
// how many times the returned function is called.
function once(fn)
{
    var called = false;
    return function()
    {
        if (!called)
        {
            if (fn) fn();
            called = true;
        }
    }
}

function fire(processor, event, arg)
{
    processor.events[event].forEach(function(fn)
    {
        fn(processor, arg);
    });
}

function nextTick(run)
{
    return function(i)
    {
        process.nextTick(function()
        {
            run(i);
        });
    }
};

// add a new callback to the processor chain. 
Processor.prototype.add = function add(item)
{
    var self = this;
    this.items.push(function(next)
    {
        var _next = once(function(){
          clearTimeout(timeout);
          next();
        });

        // set a timeout so that if for some reason the item fails and doesn't call its callback,
        // we make sure that the chain doesn't break.
        var timeout = setTimeout(function()
        {
            fire(self, 'timeout')
            _next();
        }, self.timeout);
        try
        {
            item(_next);
        }
        catch(ex)
        {
            fire(self, 'error', ex);
            _next();
        }
    });
}

// stop any additional iterations from happening
Processor.prototype.stop = function stop()
{
    this.started = false;
    clearTimeout(this._timeout);
    fire(this, 'stopped');
}

// start the loop
Processor.prototype.start = function start()
{
    // if not started yet and there are items to process
    if (!this.started && this.items.length > 0)
    {
        var self = this;
        self.started = true;
        self.batch = 0;

        function run(i)
        {
            i = i || 0;
            // call the runStart event
            if (i == 0)
            {
                self.batch++;
                fire(self, 'batchStart');
            }

            // if not the end of the loop
            if (i < self.items.length)
            {
                // run the current item, pass in the next run as the next function
                self.items[i](function(){ run(++i);})
            }
            // if no more items to process then check if still running and if so then schedule the next batch
            else if (self.started)
            {
                fire(self, 'batchEnd');

                if (self.maxBatches && self.maxBatches == self.batch)
                {
                    self.stop();
                }
                else
                {
                    self._timeout = setTimeout(run, self.interval);
                }
            }
        }

        // if nextTick is true then wrap each run within a nextTick call
        if (this.nextTick)
        {
            run = nextTick(run);
        }

        // run the first batch now
        run();
        return true;
    }

    return false;
}



module.exports = Processor;