1 /*jshint node:true */
\r
5 var util = require('util');
\r
6 var events = require('events');
\r
7 var EventEmitter = events.EventEmitter;
\r
8 var runTask = require('./lib/runTask');
\r
10 var Orchestrator = function () {
\r
11 EventEmitter.call(this);
\r
12 this.doneCallback = undefined; // call this when all tasks in the queue are done
\r
13 this.seq = []; // the order to run the tasks
\r
14 this.tasks = {}; // task objects: name, dep (list of names of dependencies), fn (the task to run)
\r
15 this.isRunning = false; // is the orchestrator running tasks? .start() to start, .stop() to stop
\r
17 util.inherits(Orchestrator, EventEmitter);
\r
19 Orchestrator.prototype.reset = function () {
\r
20 if (this.isRunning) {
\r
25 this.isRunning = false;
\r
26 this.doneCallback = undefined;
\r
29 Orchestrator.prototype.add = function (name, dep, fn) {
\r
30 if (!fn && typeof dep === 'function') {
\r
35 fn = fn || function () {}; // no-op
\r
37 throw new Error('Task requires a name');
\r
39 // validate name is a string, dep is an array of strings, and fn is a function
\r
40 if (typeof name !== 'string') {
\r
41 throw new Error('Task requires a name that is a string');
\r
43 if (typeof fn !== 'function') {
\r
44 throw new Error('Task '+name+' requires a function that is a function');
\r
46 if (!Array.isArray(dep)) {
\r
47 throw new Error('Task '+name+' can\'t support dependencies that is not an array of strings');
\r
49 dep.forEach(function (item) {
\r
50 if (typeof item !== 'string') {
\r
51 throw new Error('Task '+name+' dependency '+item+' is not a string');
\r
54 this.tasks[name] = {
\r
61 Orchestrator.prototype.task = function (name, dep, fn) {
\r
63 // alias for add, return nothing rather than this
\r
64 this.add(name, dep, fn);
\r
66 return this.tasks[name];
\r
69 Orchestrator.prototype.hasTask = function (name) {
\r
70 return !!this.tasks[name];
\r
72 // tasks and optionally a callback
\r
73 Orchestrator.prototype.start = function() {
\r
74 var args, arg, names = [], lastTask, i, seq = [];
\r
75 args = Array.prototype.slice.call(arguments, 0);
\r
77 lastTask = args[args.length-1];
\r
78 if (typeof lastTask === 'function') {
\r
79 this.doneCallback = lastTask;
\r
82 for (i = 0; i < args.length; i++) {
\r
84 if (typeof arg === 'string') {
\r
86 } else if (Array.isArray(arg)) {
\r
87 names = names.concat(arg); // FRAGILE: ASSUME: it's an array of strings
\r
89 throw new Error('pass strings or arrays of strings');
\r
93 if (this.isRunning) {
\r
94 // reset specified tasks (and dependencies) as not run
\r
95 this._resetSpecificTasks(names);
\r
97 // reset all tasks as not run
\r
98 this._resetAllTasks();
\r
100 if (this.isRunning) {
\r
101 // if you call start() again while a previous run is still in play
\r
102 // prepend the new tasks to the existing task queue
\r
103 names = names.concat(this.seq);
\r
105 if (names.length < 1) {
\r
107 for (i in this.tasks) {
\r
108 if (this.tasks.hasOwnProperty(i)) {
\r
109 names.push(this.tasks[i].name);
\r
115 this.sequence(this.tasks, names, seq, []);
\r
117 // Is this a known error?
\r
119 if (err.missingTask) {
\r
120 this.emit('task_not_found', {message: err.message, task:err.missingTask, err: err});
\r
122 if (err.recursiveTasks) {
\r
123 this.emit('task_recursion', {message: err.message, recursiveTasks:err.recursiveTasks, err: err});
\r
130 this.emit('start', {message:'seq: '+this.seq.join(',')});
\r
131 if (!this.isRunning) {
\r
132 this.isRunning = true;
\r
137 Orchestrator.prototype.stop = function (err, successfulFinish) {
\r
138 this.isRunning = false;
\r
140 this.emit('err', {message:'orchestration failed', err:err});
\r
141 } else if (successfulFinish) {
\r
142 this.emit('stop', {message:'orchestration succeeded'});
\r
145 err = 'orchestration aborted';
\r
146 this.emit('err', {message:'orchestration aborted', err: err});
\r
148 if (this.doneCallback) {
\r
149 // Avoid calling it multiple times
\r
150 this.doneCallback(err);
\r
151 } else if (err && !this.listeners('err').length) {
\r
152 // No one is listening for the error so speak louder
\r
156 Orchestrator.prototype.sequence = require('sequencify');
\r
157 Orchestrator.prototype.allDone = function () {
\r
158 var i, task, allDone = true; // nothing disputed it yet
\r
159 for (i = 0; i < this.seq.length; i++) {
\r
160 task = this.tasks[this.seq[i]];
\r
168 Orchestrator.prototype._resetTask = function(task) {
\r
175 delete task.duration;
\r
176 delete task.hrDuration;
\r
180 Orchestrator.prototype._resetAllTasks = function() {
\r
182 for (task in this.tasks) {
\r
183 if (this.tasks.hasOwnProperty(task)) {
\r
184 this._resetTask(this.tasks[task]);
\r
188 Orchestrator.prototype._resetSpecificTasks = function (names) {
\r
191 if (names && names.length) {
\r
192 for (i = 0; i < names.length; i++) {
\r
194 t = this.tasks[name];
\r
196 this._resetTask(t);
\r
197 if (t.dep && t.dep.length) {
\r
198 this._resetSpecificTasks(t.dep); // recurse
\r
201 // FRAGILE: ignore that the task doesn't exist
\r
206 Orchestrator.prototype._runStep = function () {
\r
208 if (!this.isRunning) {
\r
209 return; // user aborted, ASSUME: stop called previously
\r
211 for (i = 0; i < this.seq.length; i++) {
\r
212 task = this.tasks[this.seq[i]];
\r
213 if (!task.done && !task.running && this._readyToRunTask(task)) {
\r
214 this._runTask(task);
\r
216 if (!this.isRunning) {
\r
217 return; // task failed or user aborted, ASSUME: stop called previously
\r
220 if (this.allDone()) {
\r
221 this.stop(null, true);
\r
224 Orchestrator.prototype._readyToRunTask = function (task) {
\r
225 var ready = true, // no one disproved it yet
\r
227 if (task.dep.length) {
\r
228 for (i = 0; i < task.dep.length; i++) {
\r
229 name = task.dep[i];
\r
230 t = this.tasks[name];
\r
232 // FRAGILE: this should never happen
\r
233 this.stop("can't run "+task.name+" because it depends on "+name+" which doesn't exist");
\r
245 Orchestrator.prototype._stopTask = function (task, meta) {
\r
246 task.duration = meta.duration;
\r
247 task.hrDuration = meta.hrDuration;
\r
248 task.running = false;
\r
251 Orchestrator.prototype._emitTaskDone = function (task, message, err) {
\r
253 task.args = {task:task.name};
\r
255 task.args.duration = task.duration;
\r
256 task.args.hrDuration = task.hrDuration;
\r
257 task.args.message = task.name+' '+message;
\r
260 task.args.err = err;
\r
263 // 'task_stop' or 'task_err'
\r
264 this.emit('task_'+evt, task.args);
\r
266 Orchestrator.prototype._runTask = function (task) {
\r
269 task.args = {task:task.name, message:task.name+' started'};
\r
270 this.emit('task_start', task.args);
\r
271 task.running = true;
\r
273 runTask(task.fn.bind(this), function (err, meta) {
\r
274 that._stopTask.call(that, task, meta);
\r
275 that._emitTaskDone.call(that, task, meta.runMethod, err);
\r
277 return that.stop.call(that, err);
\r
279 that._runStep.call(that);
\r
283 // FRAGILE: ASSUME: this list is an exhaustive list of events emitted
\r
284 var events = ['start','stop','err','task_start','task_stop','task_err','task_not_found','task_recursion'];
\r
286 var listenToEvent = function (target, event, callback) {
\r
287 target.on(event, function (e) {
\r
293 Orchestrator.prototype.onAll = function (callback) {
\r
295 if (typeof callback !== 'function') {
\r
296 throw new Error('No callback specified');
\r
299 for (i = 0; i < events.length; i++) {
\r
300 listenToEvent(this, events[i], callback);
\r
304 module.exports = Orchestrator;
\r