No Description

index.js 8.0KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304
  1. /*jshint node:true */
  2. "use strict";
  3. var util = require('util');
  4. var events = require('events');
  5. var EventEmitter = events.EventEmitter;
  6. var runTask = require('./lib/runTask');
  7. var Orchestrator = function () {
  8. EventEmitter.call(this);
  9. this.doneCallback = undefined; // call this when all tasks in the queue are done
  10. this.seq = []; // the order to run the tasks
  11. this.tasks = {}; // task objects: name, dep (list of names of dependencies), fn (the task to run)
  12. this.isRunning = false; // is the orchestrator running tasks? .start() to start, .stop() to stop
  13. };
  14. util.inherits(Orchestrator, EventEmitter);
  15. Orchestrator.prototype.reset = function () {
  16. if (this.isRunning) {
  17. this.stop(null);
  18. }
  19. this.tasks = {};
  20. this.seq = [];
  21. this.isRunning = false;
  22. this.doneCallback = undefined;
  23. return this;
  24. };
  25. Orchestrator.prototype.add = function (name, dep, fn) {
  26. if (!fn && typeof dep === 'function') {
  27. fn = dep;
  28. dep = undefined;
  29. }
  30. dep = dep || [];
  31. fn = fn || function () {}; // no-op
  32. if (!name) {
  33. throw new Error('Task requires a name');
  34. }
  35. // validate name is a string, dep is an array of strings, and fn is a function
  36. if (typeof name !== 'string') {
  37. throw new Error('Task requires a name that is a string');
  38. }
  39. if (typeof fn !== 'function') {
  40. throw new Error('Task '+name+' requires a function that is a function');
  41. }
  42. if (!Array.isArray(dep)) {
  43. throw new Error('Task '+name+' can\'t support dependencies that is not an array of strings');
  44. }
  45. dep.forEach(function (item) {
  46. if (typeof item !== 'string') {
  47. throw new Error('Task '+name+' dependency '+item+' is not a string');
  48. }
  49. });
  50. this.tasks[name] = {
  51. fn: fn,
  52. dep: dep,
  53. name: name
  54. };
  55. return this;
  56. };
  57. Orchestrator.prototype.task = function (name, dep, fn) {
  58. if (dep || fn) {
  59. // alias for add, return nothing rather than this
  60. this.add(name, dep, fn);
  61. } else {
  62. return this.tasks[name];
  63. }
  64. };
  65. Orchestrator.prototype.hasTask = function (name) {
  66. return !!this.tasks[name];
  67. };
  68. // tasks and optionally a callback
  69. Orchestrator.prototype.start = function() {
  70. var args, arg, names = [], lastTask, i, seq = [];
  71. args = Array.prototype.slice.call(arguments, 0);
  72. if (args.length) {
  73. lastTask = args[args.length-1];
  74. if (typeof lastTask === 'function') {
  75. this.doneCallback = lastTask;
  76. args.pop();
  77. }
  78. for (i = 0; i < args.length; i++) {
  79. arg = args[i];
  80. if (typeof arg === 'string') {
  81. names.push(arg);
  82. } else if (Array.isArray(arg)) {
  83. names = names.concat(arg); // FRAGILE: ASSUME: it's an array of strings
  84. } else {
  85. throw new Error('pass strings or arrays of strings');
  86. }
  87. }
  88. }
  89. if (this.isRunning) {
  90. // reset specified tasks (and dependencies) as not run
  91. this._resetSpecificTasks(names);
  92. } else {
  93. // reset all tasks as not run
  94. this._resetAllTasks();
  95. }
  96. if (this.isRunning) {
  97. // if you call start() again while a previous run is still in play
  98. // prepend the new tasks to the existing task queue
  99. names = names.concat(this.seq);
  100. }
  101. if (names.length < 1) {
  102. // run all tasks
  103. for (i in this.tasks) {
  104. if (this.tasks.hasOwnProperty(i)) {
  105. names.push(this.tasks[i].name);
  106. }
  107. }
  108. }
  109. seq = [];
  110. try {
  111. this.sequence(this.tasks, names, seq, []);
  112. } catch (err) {
  113. // Is this a known error?
  114. if (err) {
  115. if (err.missingTask) {
  116. this.emit('task_not_found', {message: err.message, task:err.missingTask, err: err});
  117. }
  118. if (err.recursiveTasks) {
  119. this.emit('task_recursion', {message: err.message, recursiveTasks:err.recursiveTasks, err: err});
  120. }
  121. }
  122. this.stop(err);
  123. return this;
  124. }
  125. this.seq = seq;
  126. this.emit('start', {message:'seq: '+this.seq.join(',')});
  127. if (!this.isRunning) {
  128. this.isRunning = true;
  129. }
  130. this._runStep();
  131. return this;
  132. };
  133. Orchestrator.prototype.stop = function (err, successfulFinish) {
  134. this.isRunning = false;
  135. if (err) {
  136. this.emit('err', {message:'orchestration failed', err:err});
  137. } else if (successfulFinish) {
  138. this.emit('stop', {message:'orchestration succeeded'});
  139. } else {
  140. // ASSUME
  141. err = 'orchestration aborted';
  142. this.emit('err', {message:'orchestration aborted', err: err});
  143. }
  144. if (this.doneCallback) {
  145. // Avoid calling it multiple times
  146. this.doneCallback(err);
  147. } else if (err && !this.listeners('err').length) {
  148. // No one is listening for the error so speak louder
  149. throw err;
  150. }
  151. };
  152. Orchestrator.prototype.sequence = require('sequencify');
  153. Orchestrator.prototype.allDone = function () {
  154. var i, task, allDone = true; // nothing disputed it yet
  155. for (i = 0; i < this.seq.length; i++) {
  156. task = this.tasks[this.seq[i]];
  157. if (!task.done) {
  158. allDone = false;
  159. break;
  160. }
  161. }
  162. return allDone;
  163. };
  164. Orchestrator.prototype._resetTask = function(task) {
  165. if (task) {
  166. if (task.done) {
  167. task.done = false;
  168. }
  169. delete task.start;
  170. delete task.stop;
  171. delete task.duration;
  172. delete task.hrDuration;
  173. delete task.args;
  174. }
  175. };
  176. Orchestrator.prototype._resetAllTasks = function() {
  177. var task;
  178. for (task in this.tasks) {
  179. if (this.tasks.hasOwnProperty(task)) {
  180. this._resetTask(this.tasks[task]);
  181. }
  182. }
  183. };
  184. Orchestrator.prototype._resetSpecificTasks = function (names) {
  185. var i, name, t;
  186. if (names && names.length) {
  187. for (i = 0; i < names.length; i++) {
  188. name = names[i];
  189. t = this.tasks[name];
  190. if (t) {
  191. this._resetTask(t);
  192. if (t.dep && t.dep.length) {
  193. this._resetSpecificTasks(t.dep); // recurse
  194. }
  195. //} else {
  196. // FRAGILE: ignore that the task doesn't exist
  197. }
  198. }
  199. }
  200. };
  201. Orchestrator.prototype._runStep = function () {
  202. var i, task;
  203. if (!this.isRunning) {
  204. return; // user aborted, ASSUME: stop called previously
  205. }
  206. for (i = 0; i < this.seq.length; i++) {
  207. task = this.tasks[this.seq[i]];
  208. if (!task.done && !task.running && this._readyToRunTask(task)) {
  209. this._runTask(task);
  210. }
  211. if (!this.isRunning) {
  212. return; // task failed or user aborted, ASSUME: stop called previously
  213. }
  214. }
  215. if (this.allDone()) {
  216. this.stop(null, true);
  217. }
  218. };
  219. Orchestrator.prototype._readyToRunTask = function (task) {
  220. var ready = true, // no one disproved it yet
  221. i, name, t;
  222. if (task.dep.length) {
  223. for (i = 0; i < task.dep.length; i++) {
  224. name = task.dep[i];
  225. t = this.tasks[name];
  226. if (!t) {
  227. // FRAGILE: this should never happen
  228. this.stop("can't run "+task.name+" because it depends on "+name+" which doesn't exist");
  229. ready = false;
  230. break;
  231. }
  232. if (!t.done) {
  233. ready = false;
  234. break;
  235. }
  236. }
  237. }
  238. return ready;
  239. };
  240. Orchestrator.prototype._stopTask = function (task, meta) {
  241. task.duration = meta.duration;
  242. task.hrDuration = meta.hrDuration;
  243. task.running = false;
  244. task.done = true;
  245. };
  246. Orchestrator.prototype._emitTaskDone = function (task, message, err) {
  247. if (!task.args) {
  248. task.args = {task:task.name};
  249. }
  250. task.args.duration = task.duration;
  251. task.args.hrDuration = task.hrDuration;
  252. task.args.message = task.name+' '+message;
  253. var evt = 'stop';
  254. if (err) {
  255. task.args.err = err;
  256. evt = 'err';
  257. }
  258. // 'task_stop' or 'task_err'
  259. this.emit('task_'+evt, task.args);
  260. };
  261. Orchestrator.prototype._runTask = function (task) {
  262. var that = this;
  263. task.args = {task:task.name, message:task.name+' started'};
  264. this.emit('task_start', task.args);
  265. task.running = true;
  266. runTask(task.fn.bind(this), function (err, meta) {
  267. that._stopTask.call(that, task, meta);
  268. that._emitTaskDone.call(that, task, meta.runMethod, err);
  269. if (err) {
  270. return that.stop.call(that, err);
  271. }
  272. that._runStep.call(that);
  273. });
  274. };
  275. // FRAGILE: ASSUME: this list is an exhaustive list of events emitted
  276. var events = ['start','stop','err','task_start','task_stop','task_err','task_not_found','task_recursion'];
  277. var listenToEvent = function (target, event, callback) {
  278. target.on(event, function (e) {
  279. e.src = event;
  280. callback(e);
  281. });
  282. };
  283. Orchestrator.prototype.onAll = function (callback) {
  284. var i;
  285. if (typeof callback !== 'function') {
  286. throw new Error('No callback specified');
  287. }
  288. for (i = 0; i < events.length; i++) {
  289. listenToEvent(this, events[i], callback);
  290. }
  291. };
  292. module.exports = Orchestrator;