Fix bugs in the Worker, add tests.

This commit is contained in:
André Cruz
2013-04-20 01:39:39 +01:00
parent 05d974578d
commit 99af52ac2f
4 changed files with 286 additions and 441 deletions

View File

@@ -4,7 +4,7 @@ var events = require('events');
var mout = require('mout');
var Worker = function (defaultConcurrency, types) {
this._defaultConcurrency = defaultConcurrency != null ? defaultConcurrency : 10;
this._defaultConcurrency = typeof defaultConcurrency === 'number' ? defaultConcurrency : 10;
// Initialize some needed properties
this._queue = {};
@@ -32,7 +32,7 @@ Worker.prototype.enqueue = function (func, type) {
// Add the entry to all the types queues
types.forEach(function (type) {
this._queue[type] = this._queue[type] || [];
this._queue.push(entry);
this._queue[type].push(entry);
}, this);
// Process the entry shortly later so that handlers can be attached to the returned promise
@@ -41,20 +41,15 @@ Worker.prototype.enqueue = function (func, type) {
return deferred.promise;
};
Worker.prototype.abort = function (type) {
var promises = [],
types;
Worker.prototype.abort = function () {
var promises;
type = type || '';
types = Array.isArray(type) ? type : [type];
// Empty the queue of each specified types
types.forEach(function (type) {
// Empty the whole queue
Object.keys(this._queue).forEach(function (type) {
this._queue[type] = [];
}, this);
// Wait for all pending functions of the specified type to finish
// Wait for all pending functions to finish
promises = this._executing.map(function (entry) {
return entry.deferred.promise;
});
@@ -65,34 +60,40 @@ Worker.prototype.abort = function (type) {
// -----------------
Worker.prottoype._processQueue = function (type) {
Worker.prototype._processQueue = function (type) {
var queue = this._queue[type],
length = queue ? queue.length : 0,
x;
for (x = 0; x < length; x += 1) {
if (!this._processEntry(this._queue[x])) {
if (this._processEntry(queue[x])) {
break;
}
}
};
Worker.prototype._processEntry = function (entry) {
var allFree = entry.types.every(this._hasSlot, this);
var allFree = entry.types.every(this._hasSlot, this),
promise;
// If there is a free slot for every tag
if (allFree) {
// Foreach type
entry.types.forEach(function (type) {
// Remove entry from the queue
mout.utils.remove(this._queue[type], entry);
mout.array.remove(this._queue[type], entry);
// Take slot
this._takeSlot(type);
}, this);
// Execute the function
this._executing.push(entry);
entry.func().then(
promise = entry.func();
if (typeof promise.then === 'undefined') {
promise = Q.resolve(promise);
}
promise.then(
this._onResolve.bind(this, entry, true),
this._onResolve.bind(this, entry, false)
);