orchestrator模块源码

回调函数视作为任务，配置任务名和依赖任务两个属性，通过任务名执行回调函数。

个中实现可以同模块化加载器requirejs、seajs相比较。

 

orchestrator.js

"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);// 继承EventEmitter对象的属性
	this.doneCallback = undefined; // 所有任务执行完成后的回调函数
	this.seq = []; // 执行中的任务或等待执行的父任务
	this.tasks = {}; // 存储任务的name任务名、dep依赖的任务、fn任务执行函数
	this.isRunning = false; // 当前有任务在执行过程中
};
util.inherits(Orchestrator, EventEmitter);// 继承EventEmitter对象原型中的方法

	Orchestrator.prototype.reset = function () {
		if (this.isRunning) {
			this.stop(null);
		}
		this.tasks = {};
		this.seq = [];
		this.isRunning = false;
		this.doneCallback = undefined;
		return this;
	};

	// 添加任务，参数name为任务名，dep为依赖的任务，fn为执行函数
	// gulp中，赋值给Gulp.prototype.task，用于添加任务
	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];
	};

	// 将待执行的任务添加到this.seq中，校验添加的任务是否符合规范，调用_runStep执行任务
	// this.seq中，通过running属性判断任务是否已执行，避免再次执行
	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);
				} else {
					throw new Error('pass strings or arrays of strings');
				}
			}
		}
		if (this.isRunning) {
			// 嵌套重设任务及其依赖的状态
			this._resetSpecificTasks(names);
		} else {
			// 重设所有任务的状态
			this._resetAllTasks();
		}
		if (this.isRunning) {
			names = names.concat(this.seq);
		}
		if (names.length < 1) {
			// 传参为空且this.seq中没有任务，遍历this.tasks，执行所有已添加的任务
			for (i in this.tasks) {
				if (this.tasks.hasOwnProperty(i)) {
					names.push(this.tasks[i].name);
				}
			}
		}
		seq = [];
		try {
			// 通过this.tasks校验任务名是否正确，尾参nest校验父子任务不构成相互依赖
			// 通过校验后，将names中未存在于seq的任务推送到seq中
			this.sequence(this.tasks, names, seq, []);
		} catch (err) {
			if (err) {
				// 以任务名name从this.tasks中获取任务失败时，报错
				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 {
			err = 'orchestration aborted';
			this.emit('err', {message:'orchestration aborted', err: err});
		}
		if (this.doneCallback) {
			this.doneCallback(err);

		// 用户监听"err"事件的绑定函数为空，主动抛出错误
		} else if (err && !this.listeners('err').length) {
			throw err;
		}
	};

	// sequence(tasks,names,results,nest)
	// 通过tasks校验任务名是否正确，尾参nest校验父子任务不构成相互依赖
	// 通过校验后，将names中未存在于seq的任务推送到seq中
	Orchestrator.prototype.sequence = require('sequencify');

	// 判断this.seq中待执行或执行中的所有任务是否执行完成
	Orchestrator.prototype.allDone = function () {
		var i, task, allDone = true;
		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);
					}
				}
			}
		}
	};

	// 逐个判断任务的依赖执行完成后，调用_runTask执行当前任务
	Orchestrator.prototype._runStep = function () {
		var i, task;
		if (!this.isRunning) {
			return;
		}
		for (i = 0; i < this.seq.length; i++) {
			task = this.tasks[this.seq[i]];
			// this._readyToRunTask(task) 当前任务的依赖已执行完成，当前任务可执行
			if (!task.done && !task.running && this._readyToRunTask(task)) {
				this._runTask(task);
			}
			// 子任务执行完成，_runTask回调中调用_runStep方法，这时调用stop方法，终止后续任务执行
			if (!this.isRunning) {
				return;
			}
		}
		if (this.allDone()) {
			this.stop(null, true);
		}
	};

	// 判断依赖的任务在this.tasks中存在，且执行完成，则当前任务的可执行状态为true
	Orchestrator.prototype._readyToRunTask = function (task) {
		var ready = true,
			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) {
					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;
	};

	// 任务执行完成或报错时，更新duration执行时间（作为监听事件的参数）、running、done执行状态属性
	Orchestrator.prototype._stopTask = function (task, meta) {
		task.duration = meta.duration;
		task.hrDuration = meta.hrDuration;
		task.running = false;
		task.done = true;
	};

	// 任务执行完成或报错时，触发'task_stop'、'task_err'事件
	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';
		}

		this.emit('task_'+evt, task.args);
	};

	// task任务的running属性设为true，通过runTask模块执行用户设置的task.fn任务函数
	// 任务执行完成或报错时，通过触发回调，执行_stopTask，更改task任务的duration、running、done属性
	// 		执行_emitTaskDone，触发'task_stop'、'task_err'事件
	// 		报错时，执行stop，触发错误事件，执行用户设置的任务回调函数
	// 		执行_runStep，间接执行依赖当前任务的父任务函数
	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);
		});
	};

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);
	});
};

	// 绑定callback回调函数，监听events列表中的事件
	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;

 

sequencify.js

// sequence(tasks,names,results,nest)
// 通过tasks校验任务名是否正确，尾参nest校验父子任务不构成相互依赖
// 通过校验后，将names中未存在于seq的任务推送到seq中
"use strict";

var sequence = function (tasks, names, results, nest) {
	var i, name, node, e, j;
	nest = nest || [];
	for (i = 0; i < names.length; i++) {
		name = names[i];

		if (results.indexOf(name) === -1) {
			node = tasks[name];
			if (!node) {
				e = new Error('task "'+name+'" is not defined');
				e.missingTask = name;
				e.taskList = [];
				for (j in tasks) {
					if (tasks.hasOwnProperty(j)) {
						e.taskList.push(tasks[j].name);
					}
				}
				throw e;
			}
			if (nest.indexOf(name) > -1) {
				nest.push(name);
				e = new Error('Recursive dependencies detected: '+nest.join(' -> '));
				e.recursiveTasks = nest;
				e.taskList = [];
				for (j in tasks) {
					if (tasks.hasOwnProperty(j)) {
						e.taskList.push(tasks[j].name);
					}
				}
				throw e;
			}
			if (node.dep.length) {
				nest.push(name);
				sequence(tasks, node.dep, results, nest);
				nest.pop(name);
			}
			results.push(name);
		}
	}
};

module.exports = sequence;

 

runTask.js

"use strict";

var eos = require('end-of-stream');
var consume = require('stream-consume');

// task任务执行函数，done任务执行完成后的回调函数（Orchestrator模块中设置，调用用户设置的回调）
// 回调的处理，其一通过task传入回调，手动触发执行，其二通过Promise模块自动判断执行状态触发执行
// 		其三通过'end-of-stream‘模块 ？？？
module.exports = function (task, done) {
	var that = this, finish, cb, isDone = false, start, r;

	finish = function (err, runMethod) {
		var hrDuration = process.hrtime(start);// 计算任务执行时间

		if (isDone && !err) {
			err = new Error('task completion callback called too many times');
		}
		isDone = true;

		var duration = hrDuration[0] + (hrDuration[1] / 1e9); // seconds

		done.call(that, err, {
			duration: duration, // seconds
			hrDuration: hrDuration, // [seconds,nanoseconds]
			runMethod: runMethod
		});
	};

	cb = function (err) {
		finish(err, 'callback');
	};

	try {
		start = process.hrtime();
		r = task(cb);
	} catch (err) {
		return finish(err, 'catch');
	}

	if (r && typeof r.then === 'function') {
		r.then(function () {
			finish(null, 'promise');
		}, function(err) {
			finish(err, 'promise');
		});
	} else if (r && typeof r.pipe === 'function') {
		// wait for stream to end

		eos(r, { error: true, readable: r.readable, writable: r.writable && !r.readable }, function(err){
			finish(err, 'stream');
		});

		// Ensure that the stream completes
        consume(r);

	} else if (task.length === 0) {
		finish(null, 'sync');
	}
};