123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304 |
- /*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;
|