www.aleph1.co.uk Git - yaffs-website/blob - node_modules/orchestrator/index.js

   1 /*jshint node:true */\r
   2 \r
   3 "use strict";\r
   4 \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
   9 \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
  16 };\r
  17 util.inherits(Orchestrator, EventEmitter);\r
  18 \r
  19         Orchestrator.prototype.reset = function () {\r
  20                 if (this.isRunning) {\r
  21                         this.stop(null);\r
  22                 }\r
  23                 this.tasks = {};\r
  24                 this.seq = [];\r
  25                 this.isRunning = false;\r
  26                 this.doneCallback = undefined;\r
  27                 return this;\r
  28         };\r
  29         Orchestrator.prototype.add = function (name, dep, fn) {\r
  30                 if (!fn && typeof dep === 'function') {\r
  31                         fn = dep;\r
  32                         dep = undefined;\r
  33                 }\r
  34                 dep = dep || [];\r
  35                 fn = fn || function () {}; // no-op\r
  36                 if (!name) {\r
  37                         throw new Error('Task requires a name');\r
  38                 }\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
  42                 }\r
  43                 if (typeof fn !== 'function') {\r
  44                         throw new Error('Task '+name+' requires a function that is a function');\r
  45                 }\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
  48                 }\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
  52                         }\r
  53                 });\r
  54                 this.tasks[name] = {\r
  55                         fn: fn,\r
  56                         dep: dep,\r
  57                         name: name\r
  58                 };\r
  59                 return this;\r
  60         };\r
  61         Orchestrator.prototype.task = function (name, dep, fn) {\r
  62                 if (dep || fn) {\r
  63                         // alias for add, return nothing rather than this\r
  64                         this.add(name, dep, fn);\r
  65                 } else {\r
  66                         return this.tasks[name];\r
  67                 }\r
  68         };\r
  69         Orchestrator.prototype.hasTask = function (name) {\r
  70                 return !!this.tasks[name];\r
  71         };\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
  76                 if (args.length) {\r
  77                         lastTask = args[args.length-1];\r
  78                         if (typeof lastTask === 'function') {\r
  79                                 this.doneCallback = lastTask;\r
  80                                 args.pop();\r
  81                         }\r
  82                         for (i = 0; i < args.length; i++) {\r
  83                                 arg = args[i];\r
  84                                 if (typeof arg === 'string') {\r
  85                                         names.push(arg);\r
  86                                 } else if (Array.isArray(arg)) {\r
  87                                         names = names.concat(arg); // FRAGILE: ASSUME: it's an array of strings\r
  88                                 } else {\r
  89                                         throw new Error('pass strings or arrays of strings');\r
  90                                 }\r
  91                         }\r
  92                 }\r
  93                 if (this.isRunning) {\r
  94                         // reset specified tasks (and dependencies) as not run\r
  95                         this._resetSpecificTasks(names);\r
  96                 } else {\r
  97                         // reset all tasks as not run\r
  98                         this._resetAllTasks();\r
  99                 }\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
 104                 }\r
 105                 if (names.length < 1) {\r
 106                         // run all tasks\r
 107                         for (i in this.tasks) {\r
 108                                 if (this.tasks.hasOwnProperty(i)) {\r
 109                                         names.push(this.tasks[i].name);\r
 110                                 }\r
 111                         }\r
 112                 }\r
 113                 seq = [];\r
 114                 try {\r
 115                         this.sequence(this.tasks, names, seq, []);\r
 116                 } catch (err) {\r
 117                         // Is this a known error?\r
 118                         if (err) {\r
 119                                 if (err.missingTask) {\r
 120                                         this.emit('task_not_found', {message: err.message, task:err.missingTask, err: err});\r
 121                                 }\r
 122                                 if (err.recursiveTasks) {\r
 123                                         this.emit('task_recursion', {message: err.message, recursiveTasks:err.recursiveTasks, err: err});\r
 124                                 }\r
 125                         }\r
 126                         this.stop(err);\r
 127                         return this;\r
 128                 }\r
 129                 this.seq = seq;\r
 130                 this.emit('start', {message:'seq: '+this.seq.join(',')});\r
 131                 if (!this.isRunning) {\r
 132                         this.isRunning = true;\r
 133                 }\r
 134                 this._runStep();\r
 135                 return this;\r
 136         };\r
 137         Orchestrator.prototype.stop = function (err, successfulFinish) {\r
 138                 this.isRunning = false;\r
 139                 if (err) {\r
 140                         this.emit('err', {message:'orchestration failed', err:err});\r
 141                 } else if (successfulFinish) {\r
 142                         this.emit('stop', {message:'orchestration succeeded'});\r
 143                 } else {\r
 144                         // ASSUME\r
 145                         err = 'orchestration aborted';\r
 146                         this.emit('err', {message:'orchestration aborted', err: err});\r
 147                 }\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
 153                         throw err;\r
 154                 }\r
 155         };\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
 161                         if (!task.done) {\r
 162                                 allDone = false;\r
 163                                 break;\r
 164                         }\r
 165                 }\r
 166                 return allDone;\r
 167         };\r
 168         Orchestrator.prototype._resetTask = function(task) {\r
 169                 if (task) {\r
 170                         if (task.done) {\r
 171                                 task.done = false;\r
 172                         }\r
 173                         delete task.start;\r
 174                         delete task.stop;\r
 175                         delete task.duration;\r
 176                         delete task.hrDuration;\r
 177                         delete task.args;\r
 178                 }\r
 179         };\r
 180         Orchestrator.prototype._resetAllTasks = function() {\r
 181                 var task;\r
 182                 for (task in this.tasks) {\r
 183                         if (this.tasks.hasOwnProperty(task)) {\r
 184                                 this._resetTask(this.tasks[task]);\r
 185                         }\r
 186                 }\r
 187         };\r
 188         Orchestrator.prototype._resetSpecificTasks = function (names) {\r
 189                 var i, name, t;\r
 190 \r
 191                 if (names && names.length) {\r
 192                         for (i = 0; i < names.length; i++) {\r
 193                                 name = names[i];\r
 194                                 t = this.tasks[name];\r
 195                                 if (t) {\r
 196                                         this._resetTask(t);\r
 197                                         if (t.dep && t.dep.length) {\r
 198                                                 this._resetSpecificTasks(t.dep); // recurse\r
 199                                         }\r
 200                                 //} else {\r
 201                                         // FRAGILE: ignore that the task doesn't exist\r
 202                                 }\r
 203                         }\r
 204                 }\r
 205         };\r
 206         Orchestrator.prototype._runStep = function () {\r
 207                 var i, task;\r
 208                 if (!this.isRunning) {\r
 209                         return; // user aborted, ASSUME: stop called previously\r
 210                 }\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
 215                         }\r
 216                         if (!this.isRunning) {\r
 217                                 return; // task failed or user aborted, ASSUME: stop called previously\r
 218                         }\r
 219                 }\r
 220                 if (this.allDone()) {\r
 221                         this.stop(null, true);\r
 222                 }\r
 223         };\r
 224         Orchestrator.prototype._readyToRunTask = function (task) {\r
 225                 var ready = true, // no one disproved it yet\r
 226                         i, name, t;\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
 231                                 if (!t) {\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
 234                                         ready = false;\r
 235                                         break;\r
 236                                 }\r
 237                                 if (!t.done) {\r
 238                                         ready = false;\r
 239                                         break;\r
 240                                 }\r
 241                         }\r
 242                 }\r
 243                 return ready;\r
 244         };\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
 249                 task.done = true;\r
 250         };\r
 251         Orchestrator.prototype._emitTaskDone = function (task, message, err) {\r
 252                 if (!task.args) {\r
 253                         task.args = {task:task.name};\r
 254                 }\r
 255                 task.args.duration = task.duration;\r
 256                 task.args.hrDuration = task.hrDuration;\r
 257                 task.args.message = task.name+' '+message;\r
 258                 var evt = 'stop';\r
 259                 if (err) {\r
 260                         task.args.err = err;\r
 261                         evt = 'err';\r
 262                 }\r
 263                 // 'task_stop' or 'task_err'\r
 264                 this.emit('task_'+evt, task.args);\r
 265         };\r
 266         Orchestrator.prototype._runTask = function (task) {\r
 267                 var that = this;\r
 268 \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
 272 \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
 276                         if (err) {\r
 277                                 return that.stop.call(that, err);\r
 278                         }\r
 279                         that._runStep.call(that);\r
 280                 });\r
 281         };\r
 282 \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
 285 \r
 286 var listenToEvent = function (target, event, callback) {\r
 287         target.on(event, function (e) {\r
 288                 e.src = event;\r
 289                 callback(e);\r
 290         });\r
 291 };\r
 292 \r
 293         Orchestrator.prototype.onAll = function (callback) {\r
 294                 var i;\r
 295                 if (typeof callback !== 'function') {\r
 296                         throw new Error('No callback specified');\r
 297                 }\r
 298 \r
 299                 for (i = 0; i < events.length; i++) {\r
 300                         listenToEvent(this, events[i], callback);\r
 301                 }\r
 302         };\r
 303 \r
 304 module.exports = Orchestrator;\r