greenlock-express.js/master.js

"use strict";

require("./main.js");

var Master = module.exports;

var cluster = require("cluster");
var os = require("os");
var msgPrefix = "greenlock:";

Master.create = function(opts) {
	var resolveCb;
	var _readyCb;
	var _kicked = false;

	var greenlock = require("./greenlock.js").create(opts);

	var ready = new Promise(function(resolve) {
		resolveCb = resolve;
	}).then(function(fn) {
		_readyCb = fn;
		return fn;
	});

	function kickoff() {
		if (_kicked) {
			return;
		}
		_kicked = true;

		Master._spawnWorkers(opts, greenlock);

		ready.then(function(fn) {
			// not sure what this API should be yet
			fn();
		});
	}

	var master = {
		serve: function() {
			kickoff();
			return master;
		},
		master: function(fn) {
			if (_readyCb) {
				throw new Error("can't call master twice");
			}
			kickoff();
			resolveCb(fn);
			return master;
		}
	};
	return master;
};

function range(n) {
	n = parseInt(n, 10);
	if (!n) {
		return [];
	}
	return new Array(n).join(",").split(",");
}

Master._spawnWorkers = function(opts, greenlock) {
	var numCpus = parseInt(process.env.NUMBER_OF_PROCESSORS, 10) || os.cpus().length;

	// process rpc messages
	// start when dead
	var numWorkers = parseInt(opts.numWorkers, 10);
	if (!numWorkers) {
		if (numCpus <= 2) {
			numWorkers = 2;
		} else {
			numWorkers = numCpus - 1;
		}
	}

	cluster.on("exit", function() {
		setTimeout(function() {
			process.exit(3);
		}, 100);
	});

	var workers = range(numWorkers);
	function next() {
		if (!workers.length) {
			return;
		}
		workers.pop();

		// for a nice aesthetic
		setTimeout(function() {
			Master._spawnWorker(opts, greenlock);
			next();
		}, 250);
	}

	next();
};

Master._spawnWorker = function(opts, greenlock) {
	var w = cluster.fork();
	// automatically added to master's `cluster.workers`
	w.on("exit", function(code, signal) {
		// TODO handle failures
		// Should test if the first starts successfully
		// Should exit if failures happen too quickly

		// For now just kill all when any die
		if (signal) {
			console.error("worker was killed by signal:", signal);
		} else if (code !== 0) {
			console.error("worker exited with error code:", code);
		} else {
			console.error("worker unexpectedly quit without exit code or signal");
		}
		process.exit(2);

		//addWorker();
	});

	function handleMessage(msg) {
		if (0 !== (msg._id || "").indexOf(msgPrefix)) {
			return;
		}
		if ("string" !== typeof msg._funcname) {
			// TODO developer error
			return;
		}

		function rpc() {
			return greenlock[msg._funcname](msg._input)
				.then(function(result) {
					w.send({
						_id: msg._id,
						_result: result
					});
				})
				.catch(function(e) {
					var error = new Error(e.message);
					Object.getOwnPropertyNames(e).forEach(function(k) {
						error[k] = e[k];
					});
					w.send({
						_id: msg._id,
						_error: error
					});
				});
		}

		try {
			rpc();
		} catch (e) {
			console.error("Unexpected and uncaught greenlock." + msg._funcname + " error:");
			console.error(e);
		}
	}

	w.on("message", handleMessage);
};