305 lines
8.3 KiB
JavaScript
305 lines
8.3 KiB
JavaScript
/*jshint node:true */
|
|
|
|
"use strict";
|
|
|
|
var util = require('util');
|
|
var events = require('events');
|
|
var EventEmitter = events.EventEmitter;
|
|
var runTask = require('./lib/runTask');
|
|
|
|
var Orchestrator = function () {
|
|
EventEmitter.call(this);
|
|
this.doneCallback = undefined; // call this when all tasks in the queue are done
|
|
this.seq = []; // the order to run the tasks
|
|
this.tasks = {}; // task objects: name, dep (list of names of dependencies), fn (the task to run)
|
|
this.isRunning = false; // is the orchestrator running tasks? .start() to start, .stop() to stop
|
|
};
|
|
util.inherits(Orchestrator, EventEmitter);
|
|
|
|
Orchestrator.prototype.reset = function () {
|
|
if (this.isRunning) {
|
|
this.stop(null);
|
|
}
|
|
this.tasks = {};
|
|
this.seq = [];
|
|
this.isRunning = false;
|
|
this.doneCallback = undefined;
|
|
return this;
|
|
};
|
|
Orchestrator.prototype.add = function (name, dep, fn) {
|
|
if (!fn && typeof dep === 'function') {
|
|
fn = dep;
|
|
dep = undefined;
|
|
}
|
|
dep = dep || [];
|
|
fn = fn || function () {}; // no-op
|
|
if (!name) {
|
|
throw new Error('Task requires a name');
|
|
}
|
|
// validate name is a string, dep is an array of strings, and fn is a function
|
|
if (typeof name !== 'string') {
|
|
throw new Error('Task requires a name that is a string');
|
|
}
|
|
if (typeof fn !== 'function') {
|
|
throw new Error('Task '+name+' requires a function that is a function');
|
|
}
|
|
if (!Array.isArray(dep)) {
|
|
throw new Error('Task '+name+' can\'t support dependencies that is not an array of strings');
|
|
}
|
|
dep.forEach(function (item) {
|
|
if (typeof item !== 'string') {
|
|
throw new Error('Task '+name+' dependency '+item+' is not a string');
|
|
}
|
|
});
|
|
this.tasks[name] = {
|
|
fn: fn,
|
|
dep: dep,
|
|
name: name
|
|
};
|
|
return this;
|
|
};
|
|
Orchestrator.prototype.task = function (name, dep, fn) {
|
|
if (dep || fn) {
|
|
// alias for add, return nothing rather than this
|
|
this.add(name, dep, fn);
|
|
} else {
|
|
return this.tasks[name];
|
|
}
|
|
};
|
|
Orchestrator.prototype.hasTask = function (name) {
|
|
return !!this.tasks[name];
|
|
};
|
|
// tasks and optionally a callback
|
|
Orchestrator.prototype.start = function() {
|
|
var args, arg, names = [], lastTask, i, seq = [];
|
|
args = Array.prototype.slice.call(arguments, 0);
|
|
if (args.length) {
|
|
lastTask = args[args.length-1];
|
|
if (typeof lastTask === 'function') {
|
|
this.doneCallback = lastTask;
|
|
args.pop();
|
|
}
|
|
for (i = 0; i < args.length; i++) {
|
|
arg = args[i];
|
|
if (typeof arg === 'string') {
|
|
names.push(arg);
|
|
} else if (Array.isArray(arg)) {
|
|
names = names.concat(arg); // FRAGILE: ASSUME: it's an array of strings
|
|
} else {
|
|
throw new Error('pass strings or arrays of strings');
|
|
}
|
|
}
|
|
}
|
|
if (this.isRunning) {
|
|
// reset specified tasks (and dependencies) as not run
|
|
this._resetSpecificTasks(names);
|
|
} else {
|
|
// reset all tasks as not run
|
|
this._resetAllTasks();
|
|
}
|
|
if (this.isRunning) {
|
|
// if you call start() again while a previous run is still in play
|
|
// prepend the new tasks to the existing task queue
|
|
names = names.concat(this.seq);
|
|
}
|
|
if (names.length < 1) {
|
|
// run all tasks
|
|
for (i in this.tasks) {
|
|
if (this.tasks.hasOwnProperty(i)) {
|
|
names.push(this.tasks[i].name);
|
|
}
|
|
}
|
|
}
|
|
seq = [];
|
|
try {
|
|
this.sequence(this.tasks, names, seq, []);
|
|
} catch (err) {
|
|
// Is this a known error?
|
|
if (err) {
|
|
if (err.missingTask) {
|
|
this.emit('task_not_found', {message: err.message, task:err.missingTask, err: err});
|
|
}
|
|
if (err.recursiveTasks) {
|
|
this.emit('task_recursion', {message: err.message, recursiveTasks:err.recursiveTasks, err: err});
|
|
}
|
|
}
|
|
this.stop(err);
|
|
return this;
|
|
}
|
|
this.seq = seq;
|
|
this.emit('start', {message:'seq: '+this.seq.join(',')});
|
|
if (!this.isRunning) {
|
|
this.isRunning = true;
|
|
}
|
|
this._runStep();
|
|
return this;
|
|
};
|
|
Orchestrator.prototype.stop = function (err, successfulFinish) {
|
|
this.isRunning = false;
|
|
if (err) {
|
|
this.emit('err', {message:'orchestration failed', err:err});
|
|
} else if (successfulFinish) {
|
|
this.emit('stop', {message:'orchestration succeeded'});
|
|
} else {
|
|
// ASSUME
|
|
err = 'orchestration aborted';
|
|
this.emit('err', {message:'orchestration aborted', err: err});
|
|
}
|
|
if (this.doneCallback) {
|
|
// Avoid calling it multiple times
|
|
this.doneCallback(err);
|
|
} else if (err && !this.listeners('err').length) {
|
|
// No one is listening for the error so speak louder
|
|
throw err;
|
|
}
|
|
};
|
|
Orchestrator.prototype.sequence = require('sequencify');
|
|
Orchestrator.prototype.allDone = function () {
|
|
var i, task, allDone = true; // nothing disputed it yet
|
|
for (i = 0; i < this.seq.length; i++) {
|
|
task = this.tasks[this.seq[i]];
|
|
if (!task.done) {
|
|
allDone = false;
|
|
break;
|
|
}
|
|
}
|
|
return allDone;
|
|
};
|
|
Orchestrator.prototype._resetTask = function(task) {
|
|
if (task) {
|
|
if (task.done) {
|
|
task.done = false;
|
|
}
|
|
delete task.start;
|
|
delete task.stop;
|
|
delete task.duration;
|
|
delete task.hrDuration;
|
|
delete task.args;
|
|
}
|
|
};
|
|
Orchestrator.prototype._resetAllTasks = function() {
|
|
var task;
|
|
for (task in this.tasks) {
|
|
if (this.tasks.hasOwnProperty(task)) {
|
|
this._resetTask(this.tasks[task]);
|
|
}
|
|
}
|
|
};
|
|
Orchestrator.prototype._resetSpecificTasks = function (names) {
|
|
var i, name, t;
|
|
|
|
if (names && names.length) {
|
|
for (i = 0; i < names.length; i++) {
|
|
name = names[i];
|
|
t = this.tasks[name];
|
|
if (t) {
|
|
this._resetTask(t);
|
|
if (t.dep && t.dep.length) {
|
|
this._resetSpecificTasks(t.dep); // recurse
|
|
}
|
|
//} else {
|
|
// FRAGILE: ignore that the task doesn't exist
|
|
}
|
|
}
|
|
}
|
|
};
|
|
Orchestrator.prototype._runStep = function () {
|
|
var i, task;
|
|
if (!this.isRunning) {
|
|
return; // user aborted, ASSUME: stop called previously
|
|
}
|
|
for (i = 0; i < this.seq.length; i++) {
|
|
task = this.tasks[this.seq[i]];
|
|
if (!task.done && !task.running && this._readyToRunTask(task)) {
|
|
this._runTask(task);
|
|
}
|
|
if (!this.isRunning) {
|
|
return; // task failed or user aborted, ASSUME: stop called previously
|
|
}
|
|
}
|
|
if (this.allDone()) {
|
|
this.stop(null, true);
|
|
}
|
|
};
|
|
Orchestrator.prototype._readyToRunTask = function (task) {
|
|
var ready = true, // no one disproved it yet
|
|
i, name, t;
|
|
if (task.dep.length) {
|
|
for (i = 0; i < task.dep.length; i++) {
|
|
name = task.dep[i];
|
|
t = this.tasks[name];
|
|
if (!t) {
|
|
// FRAGILE: this should never happen
|
|
this.stop("can't run "+task.name+" because it depends on "+name+" which doesn't exist");
|
|
ready = false;
|
|
break;
|
|
}
|
|
if (!t.done) {
|
|
ready = false;
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
return ready;
|
|
};
|
|
Orchestrator.prototype._stopTask = function (task, meta) {
|
|
task.duration = meta.duration;
|
|
task.hrDuration = meta.hrDuration;
|
|
task.running = false;
|
|
task.done = true;
|
|
};
|
|
Orchestrator.prototype._emitTaskDone = function (task, message, err) {
|
|
if (!task.args) {
|
|
task.args = {task:task.name};
|
|
}
|
|
task.args.duration = task.duration;
|
|
task.args.hrDuration = task.hrDuration;
|
|
task.args.message = task.name+' '+message;
|
|
var evt = 'stop';
|
|
if (err) {
|
|
task.args.err = err;
|
|
evt = 'err';
|
|
}
|
|
// 'task_stop' or 'task_err'
|
|
this.emit('task_'+evt, task.args);
|
|
};
|
|
Orchestrator.prototype._runTask = function (task) {
|
|
var that = this;
|
|
|
|
task.args = {task:task.name, message:task.name+' started'};
|
|
this.emit('task_start', task.args);
|
|
task.running = true;
|
|
|
|
runTask(task.fn.bind(this), function (err, meta) {
|
|
that._stopTask.call(that, task, meta);
|
|
that._emitTaskDone.call(that, task, meta.runMethod, err);
|
|
if (err) {
|
|
return that.stop.call(that, err);
|
|
}
|
|
that._runStep.call(that);
|
|
});
|
|
};
|
|
|
|
// FRAGILE: ASSUME: this list is an exhaustive list of events emitted
|
|
var events = ['start','stop','err','task_start','task_stop','task_err','task_not_found','task_recursion'];
|
|
|
|
var listenToEvent = function (target, event, callback) {
|
|
target.on(event, function (e) {
|
|
e.src = event;
|
|
callback(e);
|
|
});
|
|
};
|
|
|
|
Orchestrator.prototype.onAll = function (callback) {
|
|
var i;
|
|
if (typeof callback !== 'function') {
|
|
throw new Error('No callback specified');
|
|
}
|
|
|
|
for (i = 0; i < events.length; i++) {
|
|
listenToEvent(this, events[i], callback);
|
|
}
|
|
};
|
|
|
|
module.exports = Orchestrator;
|