Merge pull request #269 from Hornwitser/link-refactor-2

Another link refactor

.gitignore

@@ -32,10 +32,9 @@ Factorio*
 /instances
 /sharedMods
 /database
-database/items.db
-config.json
-config-control.json
-config-slave.json
+/config.json
+/config-control.json
+/config-slave.json
 package-lock.json
 secret-api-token.txt
 mod_cache.json

CHANGELOG.md

@@ -16,6 +16,9 @@ Version 2.0.0
   instances, starting and stopping them individually.  To manage this the
   local command line interface has been replaced with a remote interface on
   the master server that can be accessed through the clusterctl cli tool.
+- Rewritten the communication between slaves and the master from scratch.  The
+  new system is based on a WebSocket connection between the slaves and the
+  master server and provides efficient validated bi-directonal communication.
 
 ### Changes
 

client.js

@@ -18,6 +18,7 @@ const hashFile = require('lib/hash').hashFile;
 const factorio = require("lib/factorio");
 const schema = require("lib/schema");
 const link = require("lib/link");
+const errors = require("lib/errors");
 
 // Uhm...
 var global = {};
@@ -25,8 +26,10 @@ var global = {};
 /**
  * Keeps track of the runtime parameters of an instance
  */
-class Instance {
-	constructor(dir, factorioDir, instanceConfig) {
+class Instance extends link.Link{
+	constructor(connector, dir, factorioDir, instanceConfig) {
+		super('instance', 'slave', connector);
+		link.attachAllMessages(this);
 		this._dir = dir;
 
 		// This is expected to change with the config system rewrite
@@ -47,12 +50,29 @@ class Instance {
 		this.server = new factorio.FactorioServer(
 			path.join(factorioDir, "data"), this._dir, serverOptions
 		);
+
+		this.server.on('output', (output) => {
+			link.messages.instanceOutput.send(this, { instance_id: this.config.id, output })
+		});
 	}
 
 	async init() {
 		await this.server.init();
 	}
 
+	/**
+	 * Creates a new empty instance directory
+	 *
+	 * Creates the neccessary files for starting up a new instance into the
+	 * provided instance directory.
+	 *
+	 * @param {Number} id -
+	 *     ID of the new instance.  Must be unique to the cluster.
+	 * @param {String} instanceDir -
+	 *     Directory to create the new instance into.
+	 * @param {String} factorioDir - Path to factorio installation.
+	 * @param {Object} options - Options for new instance.
+	 */
 	static async create(id, instanceDir, factorioDir, options) {
 		let instanceConfig = {
 			id,
@@ -62,23 +82,18 @@ class Instance {
 			clientPassword: null,
 		}
 
-		let instance = new this(instanceDir, factorioDir, instanceConfig);
-		await instance.init();
-		console.log(`Creating ${instance.path()}`);
-		await fs.ensureDir(instance.path());
-		await fs.ensureDir(instance.path("script-output"));
-
-		await symlinkMods(instance, "sharedMods", console);
-		console.log("Clusterio | Created instance with settings:")
-		console.log(instanceConfig);
+		console.log(`Clusterio | Creating ${instanceDir}`);
+		await fs.ensureDir(instanceDir);
+		await fs.ensureDir(path.join(instanceDir, "script-output"));
+		await fs.ensureDir(path.join(instanceDir, "saves"));
 
 		// save instance config
-		await fs.outputFile(instance.path("config.json"), JSON.stringify(instanceConfig, null, 4));
+		await fs.outputFile(path.join(instanceDir, "config.json"), JSON.stringify(instanceConfig, null, 4));
 
 		let serverSettings = await factorio.FactorioServer.exampleSettings(path.join(factorioDir, "data"));
-		let gameName = "Clusterio instance: " + instance.name;
+		let gameName = "Clusterio instance: " + options.name;
 		if (options.username) {
-			gameName = options.username + "'s clusterio " + instance.name;
+			gameName = options.username + "'s clusterio " + options.name;
 		}
 
 		let overrides = {
@@ -101,33 +116,18 @@ class Instance {
 			serverSettings[key] = value;
 		}
 
-		await fs.writeFile(instance.path("server-settings.json"), JSON.stringify(serverSettings, null, 4));
-
-		return instance;
+		await fs.writeFile(
+			path.join(instanceDir, "server-settings.json"),
+			JSON.stringify(serverSettings, null, 4)
+		);
 	}
 
-	async createSave() {
-		console.log("Creating save .....");
-
-		await this.server.create("world");
-		console.log("Clusterio | Successfully created save");
-	}
-
-	async start(slaveConfig, socket) {
-		console.log("Deleting .tmp.zip files");
-		let savefiles = fs.readdirSync(this.path("saves"));
-		for(let i = 0; i < savefiles.length; i++){
-			if(savefiles[i].substr(savefiles[i].length - 8, 8) == ".tmp.zip") {
-				fs.unlinkSync(this.path("saves", savefiles[i]));
-			}
-		}
+	async start(saveName, slaveConfig, socket) {
 		console.log("Clusterio | Rotating old logs...");
 		// clean old log file to avoid crash
 		try{
 			let logPath = this.path("factorio-current.log");
 			let stat = await fs.stat(logPath);
-			console.log(stat)
-			console.log(stat.isFile())
 			if(stat.isFile()){
 				let logFilename = `factorio-${Math.floor(Date.parse(stat.mtime)/1000)}.log`;
 				await fs.rename(logPath, this.path(logFilename));
@@ -137,17 +137,6 @@ class Instance {
 
 		await symlinkMods(this, "sharedMods", console);
 
-		// Spawn factorio server
-		let latestSave = await fileOps.getNewestFile(this.path("saves"));
-		if (latestSave === null) {
-			throw new Error(
-				"Your savefile seems to be missing. This might because you created an\n"+
-				"instance without having factorio installed and configured properly.\n"+
-				"Try installing factorio and adding your savefile to\n"+
-				"instances/[instancename]/saves/"
-			);
-		}
-
 		// Patch save with lua modules from plugins
 		console.log("Clusterio | Patching save");
 
@@ -181,7 +170,7 @@ class Instance {
 
 			modules.push(module);
 		}
-		await factorio.patch(this.path("saves", latestSave), modules);
+		await factorio.patch(this.path("saves", saveName), modules);
 
 		this.server.on('rcon-ready', () => {
 			console.log("Clusterio | RCON connection established");
@@ -200,7 +189,7 @@ class Instance {
 			//instanceManagement(slaveConfig, this, compatConfig, this.server, socket); // XXX async function
 		});
 
-		await this.server.start(latestSave);
+		await this.server.start(saveName);
 	}
 
 	/**
@@ -213,6 +202,36 @@ class Instance {
 		}
 	}
 
+	async startInstanceRequestHandler() {
+		// Find save to start
+		let latestSave = await fileOps.getNewestFile(
+			this.path("saves"), (name) => !name.endsWith('.tmp.zip')
+		);
+		if (latestSave === null) {
+			throw new errors.RequestError(
+				"No savefile was found to start the instance with."
+			);
+		}
+
+		await this.start(latestSave, this.config, this.socket);
+	}
+
+	async createSaveRequestHandler() {
+		console.log("Creating save .....");
+
+		await this.server.create("world");
+		console.log("Clusterio | Successfully created save");
+	}
+
+	async stopInstanceRequestHandler() {
+		await this.stop();
+	}
+
+	async sendRconRequestHandler(message) {
+		let result = await this.server.sendRcon(message.data.command);
+		return { result };
+	}
+
 	/**
 	 * Name of the instance
 	 *
@@ -235,6 +254,67 @@ class Instance {
 	}
 }
 
+/**
+ * Searches for instances in the provided directory
+ *
+ * Looks through all sub-dirs of the provided directory for valid
+ * instance definitions and returns a mapping of instance id to
+ * instance info objects.
+ *
+ * @returns {Map<integer, Object>} mapping between instance
+ */
+async function discoverInstances(instancesDir, logger) {
+	let instanceInfos = new Map();
+	for (let entry of await fs.readdir(instancesDir, { withFileTypes: true })) {
+		if (entry.isDirectory()) {
+			let instanceConfig;
+			let configPath = path.join(instancesDir, entry.name, "config.json");
+			try {
+				instanceConfig = JSON.parse(await fs.readFile(configPath));
+			} catch (err) {
+				if (err.code === "ENOENT") {
+					continue; // Ignore folders without config.json
+				}
+
+				logger.error(`Error occured while parsing ${configPath}: ${err}`);
+				continue;
+			}
+
+			// XXX should probably validate the entire config with a JSON Schema.
+			if (typeof instanceConfig.id !== "number" || isNaN(instanceConfig.id)) {
+				logger.error(`${configPath} is missing id`);
+				continue;
+			}
+
+			if (typeof instanceConfig.name !== "string") {
+				logger.error(`${configPath} is missing name`);
+				continue;
+			}
+
+			let instancePath = path.join(instancesDir, entry.name);
+			logger.log(`found instance ${instanceConfig.name} in ${instancePath}`);
+			instanceInfos.set(instanceConfig.id, {
+				path: instancePath,
+				config: instanceConfig,
+			});
+		}
+	}
+
+	return instanceInfos;
+}
+
+class InstanceConnection extends link.Link {
+	constructor(connector, slave) {
+		super('slave', 'instance', connector);
+		this.slave = slave;
+		link.attachAllMessages(this);
+	}
+
+	async forwardEventToMaster(message, event) {
+		event.send(this.slave, message.data);
+	}
+}
+
 class SlaveConnector extends link.SocketIOClientConnector {
 	constructor(slaveConfig) {
 		super(slaveConfig.masterURL, slaveConfig.masterAuthToken);
@@ -276,6 +356,8 @@ class Slave extends link.Link {
 			publicAddress: slaveConfig.publicIP,
 		}
 
+		this.instanceConnections = new Map();
+		this.instanceInfos = new Map();
 	}
 
 	async _findNewInstanceDir(name) {
@@ -294,125 +376,116 @@ class Slave extends link.Link {
 		return dir;
 	}
 
-	/**
-	 * Looks up the instance for an instance request handler
-	 *
-	 * Called by the handler for InstanceRequest.
-	 */
-	async forwardInstanceRequest(handler, message) {
-		let instance = this.instances.get(message.data.instance_id);
-		if (!instance) {
+	async forwardRequestToInstance(message, request) {
+		let instanceId = message.data.instance_id;
+		if (!this.instanceInfos.has(instanceId)) {
 			throw new errors.RequestError(`Instance with ID ${instanceId} does not exist`);
 		}
 
-		return await handler.call(this, instance, message);
+		let instanceConnection = this.instanceConnections.get(instanceId);
+		if (!instanceConnection) {
+			throw new errors.RequestError(`Instance ID ${instanceId} is not active`);
+		}
+
+		return await request.send(instanceConnection, message.data);
 	}
 
 	async createInstanceRequestHandler(message) {
 		let { id, options } = message.data;
-		if (this.instances.has(id)) {
+		if (this.instanceInfos.has(id)) {
 			throw new Error(`Instance with ID ${id} already exists`);
 		}
 
-		let instanceDir = await this._findNewInstanceDir(options.name);
 		// XXX: race condition on multiple simultanious calls
-		let instance = await Instance.create(id, instanceDir, this.config.factorioDir, options);
-		this.instances.set(id, instance);
-		this.hookInstance(instance);
-		this.updateInstances();
+		let instanceDir = await this._findNewInstanceDir(options.name);
+
+		await Instance.create(id, instanceDir, this.config.factorioDir, options);
+		await this.updateInstances();
 	}
 
-	async createSaveInstanceRequestHandler(instance, message) {
-		await instance.createSave();
-	}
-
-	async startInstanceInstanceRequestHandler(instance, message) {
-		await instance.start(this.config, this.socket);
-	}
-
-	async stopInstanceInstanceRequestHandler(instance, message) {
-		await instance.stop();
-	}
-
-	async deleteInstanceInstanceRequestHandler(instance, message) {
-		await instance.stop();
-		this.instances.delete(instance.config.id);
-
-		await fs.remove(instance.path());
-		this.updateInstances();
-	}
-
-	async sendRconInstanceRequestHandler(instance, message) {
-		let result = await instance.server.sendRcon(message.data.command);
-		return { result };
-	}
-
-	async findInstances() {
-		let instances = new Map();
-		for (let entry of await fs.readdir(this.config.instancesDir, { withFileTypes: true })) {
-			if (entry.isDirectory()) {
-				let instanceConfig;
-				let configPath = path.join(this.config.instancesDir, entry.name, "config.json");
-				try {
-					instanceConfig = JSON.parse(await fs.readFile(configPath));
-				} catch (err) {
-					if (err.code === "ENOENT") {
-						continue; // Ignore folders without config.json
-					}
-
-					console.error(`Error occured while parsing ${configPath}: ${err}`);
-				}
-
-				// XXX should probably validate the entire config with a JSON Schema.
-				if (typeof instanceConfig.id !== "number" || isNaN(instanceConfig.id)) {
-					console.error(`${configPath} is missing id`);
-					continue;
-				}
-
-				if (typeof instanceConfig.name !== "string") {
-					console.error(`${configPath} is missing name`);
-					continue;
-				}
-
-				let instancePath = path.join(this.config.instancesDir, entry.name);
-				console.log(`found instance ${instanceConfig.name} in ${instancePath}`);
-				let instance = new Instance(instancePath, this.config.factorioDir, instanceConfig);
-				await instance.init();
-				this.hookInstance(instance); // XXX this is the wrong place for this
-				instances.set(instanceConfig.id, instance);
-			}
+	/**
+	 * Initialize and connect an unloaded instance
+	 */
+	async _connectInstance(instanceId) {
+		let instanceInfo = this.instanceInfos.get(instanceId);
+		if (!instanceInfo) {
+			throw new errors.RequestError(`Instance with ID ${instanceId} does not exist`);
 		}
 
-		return instances;
+		if (this.instanceConnections.has(instanceId)) {
+			throw new errors.RequestError(`Instance with ID ${instanceId} is active`);
+		}
+
+		let [connectionClient, connectionServer] = link.VirtualConnector.makePair();
+		let instanceConnection = new InstanceConnection(connectionServer, this);
+		let instance = new Instance(
+			connectionClient, instanceInfo.path, this.config.factorioDir, instanceInfo.config
+		);
+		await instance.init();
+
+		// XXX: race condition on multiple simultanious calls
+		this.instanceConnections.set(instanceId, instanceConnection);
+		return instanceConnection;
 	}
 
-	hookInstance(instance) {
-		instance.server.on('output', (output) => {
-			link.messages.instanceOutput.send(this, { instance_id: instance.config.id, output })
-		});
+	async startInstanceRequestHandler(message, request) {
+		let instanceConnection = await this._connectInstance(message.data.instance_id);
+		return await request.send(instanceConnection, message.data);
 	}
 
+	async createSaveRequestHandler(message, request) {
+		let instanceId = message.data.instance_id;
+		let instanceConnection = await this._connectInstance(instanceId);
+		await request.send(instanceConnection, message.data);
+		this.instanceConnections.delete(instanceId);
+	}
 
-	updateInstances() {
+	async stopInstanceRequestHandler(message, request) {
+		await this.forwardRequestToInstance(message, request);
+		this.instanceConnections.delete(message.data.instance_id);
+	}
+
+	async deleteInstanceRequestHandler(message) {
+		let instanceId = message.data.instance_id;
+		if (this.instanceConnections.has(instanceId)) {
+			throw new errors.RequestError(`Instance with ID ${instanceId} is active`);
+		}
+
+		let instanceInfo = this.instanceInfos.get(instanceId);
+		if (!instanceInfo) {
+			throw new errors.RequestError(`Instance with ID ${instanceId} does not exist`);
+		}
+
+		await fs.remove(instanceInfo.path);
+		await this.updateInstances();
+	}
+
+	/**
+	 * Discover available instances
+	 *
+	 * Looks through the instances directory for instances and updates
+	 * the slave and master server with the new list of instances.
+	 */
+	async updateInstances() {
+		this.instanceInfos = await discoverInstances(this.config.instancesDir, console);
 		let list = [];
-		for (let instance of this.instances.values()) {
+		for (let instanceInfo of this.instanceInfos.values()) {
 			list.push({
-				id: instance.config.id,
-				name: instance.config.name,
+				id: instanceInfo.config.id,
+				name: instanceInfo.config.name,
 			});
 		}
 		link.messages.updateInstances.send(this, { instances: list });
 	}
 
 	async start() {
-		this.instances = await this.findInstances();
-		await this.connect();
-		this.updateInstances();
+		await this.updateInstances();
 	}
 
 	async stop() {
-		for (let instance of this.instances.values()) {
-			await instance.stop();
+		for (let instanceConnection of this.instanceConnections.values()) {
+			// XXX this neeeds more thought to it
+			// await instance.stop();
 		}
 	}
 }
@@ -820,6 +893,7 @@ module.exports = {
 	_Instance: Instance,
 	_checkFilename: checkFilename,
 	_symlinkMods: symlinkMods,
+	_discoverInstances: discoverInstances,
 	_Slave: Slave,
 };
 

clusterctl.js

@@ -139,7 +139,6 @@ commands.push(new Command({
 			name: args.name,
 			slave_id: slaveId,
 		});
-		console.log(response);
 	},
 }));
 
@@ -155,7 +154,6 @@ commands.push(new Command({
 		let response = await link.messages.createSave.send(control, {
 			instance_id: instanceId,
 		});
-		console.log(response);
 	},
 }));
 
@@ -171,7 +169,6 @@ commands.push(new Command({
 		let response = await link.messages.startInstance.send(control, {
 			instance_id: instanceId,
 		});
-		console.log(response);
 	},
 }));
 
@@ -187,7 +184,6 @@ commands.push(new Command({
 		let response = await link.messages.stopInstance.send(control, {
 			instance_id: instanceId,
 		});
-		console.log(response);
 	},
 }));
 
@@ -201,7 +197,6 @@ commands.push(new Command({
 		let response = await link.messages.deleteInstance.send(control, {
 			instance_id: await resolveInstance(control, args.instance),
 		});
-		console.log(response);
 	},
 }));
 
@@ -396,7 +391,7 @@ async function startControl() {
 			}
 		}
 
-		control.close("quit");
+		controlConnector.close("quit");
 		return; // ??
 	}
 

docs/communication.md

@@ -0,0 +1,58 @@
+Communication Layer
+===================
+
+<sub>This document describes the communication layer in Clusterio and
+unless you're developing Clusterio it's probably not going to be very
+useful for you.</sub>
+
+Communication is facilitated by the lib/link library and is based on
+three core abstractions: links, connectors and messages.  Links
+represents one side of a two way communication pipe, connectors deal
+with connecting and establishing communication, and messages are the
+content sent over the link.
+
+
+Link
+----
+
+The Link class represents one side of a two way communication channel
+where the actual communication is done by a connector associated with
+the link.  The link also knows what the endpoint types it's connected to
+are via the source and target attributes.
+
+The endpoint types currently implemented are master, slave, instance,
+and control.  Although slave and instances run in the same Node.js
+program they use virtual links to communicate between them in order to
+simplify the communication architecture.
+
+Upon creation the link will register with the connector in order to
+receive and process messages from it.  The messages received are
+validated using the validator registered for the type of the message.  A
+message for which there is no validator for is considered an error.
+
+
+Connector
+---------
+
+The connector used with a link is responsible for establishing and/or
+maintain the connection that messages are sent over.  Reconnect logic is
+expected to be implemented in the connector, though this has not yet
+been done.
+
+There's also the virtual connector which lets two links in the same
+program be connected to each other without having to establish a
+network connection.
+
+
+Message
+-------
+
+There's two types of messages currently implemented apart from the
+Link builtin close message and that's Request and Event.  The builtin
+messages are defined in [lib/link/messages.js](../lib/link/messages.js)
+and are attached to links via the attachAllMessages function, which is
+usally called in the Link subclass constructor.
+
+The messages define which links they are valid on and only attaches
+handlers for those links.  There's also a forwarding mechanism for when
+a request or event is to be forwarded another hop.

lib/fileOps/fileOps.js

@@ -5,17 +5,20 @@ const path = require("path");
  * Returns the newest file in a directory
  *
  * @param {string} directory - The directory to look for the newest file in.
- * @returns {string|null}
+ * @param {function(string): boolean} filter -
+ *     Optional function to filter out files to ignore.  Should return true
+ *     if the file is to be considered.
+ * @returns {?string}
  *     Name of the file with the newest timestamp or null if the
  *     directory contains no files.
  */
-async function getNewestFile(directory) {
+async function getNewestFile(directory, filter = (name) => true) {
 	let newestTime = new Date(0);
 	let newestFile = null;
 	for (let entry of await fs.readdir(directory, { withFileTypes: true })) {
 		if (entry.isFile()) {
 			let stat = await fs.stat(path.join(directory, entry.name));
-			if (stat.mtime > newestTime) {
+			if (filter(entry.name) && stat.mtime > newestTime) {
 				newestTime = stat.mtime;
 				newestFile = entry.name;
 			}

lib/link/link.js

@@ -34,7 +34,7 @@ class Link {
 				if (err instanceof errors.InvalidMessage) {
 					this.connector.close(`Invalid message: ${err.message}`);
 				} else {
-					throw err;
+					this.connector.emit('error', err);
 				}
 			}
 		});
@@ -57,12 +57,12 @@ class Link {
 	 */
 	processMessage(message) {
 		if (!schema.message(message)) {
-			throw new errors.InvalidMessage("Malformed message");
+			throw new errors.InvalidMessage("Malformed");
 		}
 
 		let validator = this._validators.get(message.type);
 		if (!validator) {
-			throw new errors.InvalidMessage(`No validator for ${message.type}`);
+			throw new errors.InvalidMessage(`No validator for ${message.type} on ${this.source}-${this.target}`);
 		}
 
 		if (!validator(message)) {
@@ -180,7 +180,7 @@ class Link {
 	 */
 	async waitFor(type, data) {
 		if (!this._validators.has(type)) {
-			throw new Error(`no validator for ${type}`);
+			throw new Error(`No validator for ${type} on ${this.source}-${this.target}`);
 		}
 
 		return new Promise((resolve, reject) => {

lib/link/messages.js

@@ -9,15 +9,47 @@ const errors = require("lib/errors");
  * link.
  */
 class Request {
+
+	/**
+	 * Creates a request
+	 *
+	 * @param {string} type -
+	 *     message type used.  Will have suffix '_request' for the request
+	 *     message and suffix '_response' for the response message.
+	 * @param {Array<string>} links -
+	 *     Links this request is valid on.  Takes an array of strings
+	 *     containing "<source>-<target>" specifications.
+	 * @param {?string} forwardTo -
+	 *     Optional target to forward this request to.  Only 'instance' is
+	 *     currently supported and adds instance_id into the
+	 *     requestProperties.
+	 * @param {Object<string, Object>} requestProperties -
+	 *     Mapping of property values to JSON schema specifications for the
+	 *     properties that are valid in the data payload of this requst.
+	 * @param {Object<string, Object>} responseProperties -
+	 *     Mapping of property values to JSON schema specifications for the
+	 *     properties that are valid in the data payload of the response to
+	 *     this requst.
+	 */
 	constructor({
-		type, links, requestProperties = {}, responseProperties = {}
+		type, links, forwardTo = null, requestProperties = {}, responseProperties = {}
 	}) {
 		this.type = type;
 		this.links = links;
+		this.forwardTo = forwardTo;
 
 		this.requestType = type + '_request';
 		this.responseType = type + '_response';
 
+		if (forwardTo === 'instance') {
+			requestProperties = Object.assign(
+				{"instance_id": { type: "integer" }}, requestProperties
+			);
+
+		} else if (forwardTo !== null) {
+			throw new Error(`Invalid forwardTo value ${forwardTo}`);
+		}
+
 		this._requestValidator = schema.compile({
 			$schema: "http://json-schema.org/draft-07/schema#",
 			properties: {
@@ -64,6 +96,9 @@ class Request {
 	 * Set a the validator for the response on the given link if it's a
 	 * source and sets the handler for the request if it is a target.
 	 *
+	 * If forwardTo was set to 'instance' the `forwardRequestToInstance`
+	 * method on the link is used as the default handler.
+	 *
 	 * Does nothing if the link is neither a source nor a target for this
 	 * request.
 	 *
@@ -81,21 +116,26 @@ class Request {
 
 		// Target side.  Note: Source and target is reversed here.
 		if (this.links.includes(`${link.target}-${link.source}`)) {
+
+			// Use forwarder if handler is not present.
 			if (!handler) {
-				throw new Error(`Missing handler for ${this.requestType} on ${link.target}-${link.source} link`);
+				if (this.forwardTo === 'instance') {
+					handler = link.forwardRequestToInstance;
+				}
+			}
+
+			if (!handler) {
+				throw new Error(`Missing handler for ${this.requestType} on ${link.source}-${link.target} link`);
 			}
 
 			link.setHandler(this.requestType, message => {
-				handler.call(link, message).then(response => {
+				handler.call(link, message, this).then(response => {
 					if (response === undefined) {
 						// XXX Should we allow implicit responses like this?
 						response = {};
 					}
 
-					if (Object.prototype.hasOwnProperty.call(response, 'seq')) {
-						throw new errors.RequestError("response contains reserved property 'seq'");
-					}
-					link.connector.send(this.responseType, { seq: message.seq, ...response });
+					link.connector.send(this.responseType, { ...response, seq: message.seq });
 
 				}).catch(err => {
 					if (!(err instanceof errors.RequestError)) {
@@ -126,95 +166,6 @@ class Request {
 	}
 }
 
-/**
- * Represents a request sent to the master that is forwarded to an instance
- *
- * This is a two part request with a master part and a slave part.  The
- * master part is done by the forwardInstanceRequest method on the master
- * side of link which finds the slave the requuested instance is on and
- * forwards it there using the foward method.
- *
- * The slave part is done by the forwardInstanceRequest on the slave side of
- * the link and finds the instance and then calls the handler with it and
- * the message.
- */
-class InstanceRequest {
-	constructor({
-		type, sources, requestProperties = {}, responseProperties = {}
-	}) {
-		this._masterRequest = new Request({
-			type: type + '_master',
-			links: sources.map(source => `${source}-master`),
-			requestProperties: {
-				"instance_id": { type: "integer" },
-				...requestProperties,
-			},
-			responseProperties,
-		});
-		this._slaveRequest = new Request({
-			type: type + '_slave',
-			links: ['master-slave'],
-			requestProperties: {
-				"instance_id": { type: "integer" },
-				...requestProperties,
-			},
-			responseProperties,
-		});
-	}
-
-	/**
-	 * Attach to a link
-	 *
-	 * Attach master/slave part of the instance request.  If this is the
-	 * slave side the handler argument is required.
-	 *
-	 * Does nothing if the link is neither a source nor a target for the
-	 * slave or master request.
-	 *
-	 * @param {Link} link - The link to attach to.
-	 * @param {Function} handler -
-	 *    Async function to to handle the instance request.  It is passed on
-	 *    to forwardInstanceRequest on the slave side.
-	 * @throws {Error} if the handler is needed and not defined.
-	 */
-	attach(link, handler) {
-		this._masterRequest.attach(link, async (message) => {
-			return await link.forwardInstanceRequest(this, message);
-		});
-		if (handler) {
-			this._slaveRequest.attach(link, async (message) => {
-				return await link.forwardInstanceRequest(handler, message);
-			});
-		} else {
-			this._slaveRequest.attach(link);
-		}
-	}
-
-	/**
-	 * Forward the message over the link
-	 *
-	 * This method is expected to be called by forwardInstanceRequest on the
-	 * master with the slave link that the instance is on.
-	 */
-	async forward(link, message) {
-		let response = await this._slaveRequest.send(link, message.data);
-		delete response.seq;
-		return response;
-	}
-
-	/**
-	 * Send request over the given link
-	 *
-	 * Sends the given data over the link to the master and waits for it to
-	 * forward the request to the slave with the instance on and return the
-	 * result of it.
-	 *
-	 * @returns {object} response data
-	 */
-	async send(link, data = {}) {
-		return await this._masterRequest.send(link, data);
-	}
-}
 
 let messages = {}
 
@@ -305,30 +256,34 @@ messages.createInstance = new Request({
 	},
 });
 
-
-messages.createSave = new InstanceRequest({
-	type: 'create_save',
-	sources: ['control'],
-});
-
-messages.startInstance = new InstanceRequest({
+messages.startInstance = new Request({
 	type: 'start_instance',
-	sources: ['control'],
+	links: ['control-master', 'master-slave', 'slave-instance'],
+	forwardTo: 'instance',
 });
 
-messages.stopInstance = new InstanceRequest({
+messages.createSave = new Request({
+	type: 'create_save',
+	links: ['control-master', 'master-slave', 'slave-instance'],
+	forwardTo: 'instance',
+});
+
+messages.stopInstance = new Request({
 	type: 'stop_instance',
-	sources: ['control'],
+	links: ['control-master', 'master-slave', 'slave-instance'],
+	forwardTo: 'instance',
 });
 
-messages.deleteInstance = new InstanceRequest({
+messages.deleteInstance = new Request({
 	type: 'delete_instance',
-	sources: ['control'],
+	links: ['control-master', 'master-slave'],
+	forwardTo: 'instance',
 });
 
-messages.sendRcon = new InstanceRequest({
+messages.sendRcon = new Request({
 	type: 'send_rcon',
-	sources: ['control'],
+	links: ['control-master', 'master-slave', 'slave-instance'],
+	forwardTo: 'instance',
 	requestProperties: {
 		"command": { type: "string" },
 	},
@@ -338,10 +293,38 @@ messages.sendRcon = new InstanceRequest({
 });
 
 
+/**
+ * Represents an event sent over the link
+ *
+ * A one way message without any response or recipt confirmation that
+ * can be innitiated by either party of the link.
+ */
 class Event {
-	constructor({ type, links, eventProperties = {} }) {
+
+	/**
+	 * Creates an event
+	 *
+	 * @param {string} type -
+	 *     message type used.  Will have suffix '_event' appended to it for
+	 *     the messages sent through this event.
+	 * @param {Array<string>} links -
+	 *     Links this event is valid on.  Takes an array of strings
+	 *     containing "<source>-<target>" specifications.
+	 * @param {?string} forwardTo -
+	 *     Optional target to forward this request to.  Only 'master' is
+	 *     currently supported.
+	 * @param {Object<string, Object>} eventProperties -
+	 *     Mapping of property values to JSON schema specifications for the
+	 *     properties that are valid in the data payload of this event.
+	 */
+	constructor({ type, links, forwardTo = null, eventProperties = {} }) {
 		this.type = type;
 		this.links = links;
+		this.forwardTo = forwardTo;
+
+		if (forwardTo !== 'master' && forwardTo !== null) {
+			throw new Error(`Invalid forwardTo value ${forwardTo}`);
+		}
 
 		this.eventType = type + '_event';
 		this._eventValidator = schema.compile({
@@ -357,16 +340,40 @@ class Event {
 		});
 	}
 
+	/**
+	 * Attach to a link
+	 *
+	 * Set a the handler for the event on the given link if it's a target
+	 * for the event.
+	 *
+	 * If forwardTo was set to 'master' the `forwardEventToMaster` method on
+	 * the link is used as the default handler.
+	 *
+	 * Does nothing if the link is not a target for this event.
+	 *
+	 * @param {Link} link - The link to attach to.
+	 * @param {Function} handler -
+	 *    Async function to invoke with link set as this to handle the
+	 *    event.  Only used on the target side.
+	 * @throws {Error} if the handler is needed and not defined.
+	 */
 	attach(link, handler) {
 		// Target side.  Note: Source and target is reversed here.
 		if (this.links.includes(`${link.target}-${link.source}`)) {
+			// Use forwarder if handler is not present.
 			if (!handler) {
-				throw new Error(`Missing handler for ${this.eventType} on ${link.target}-${link.source} link`);
+				if (this.forwardTo === 'master') {
+					handler = link.forwardEventToMaster;
+				}
+			}
+
+			if (!handler) {
+				throw new Error(`Missing handler for ${this.eventType} on ${link.source}-${link.target} link`);
 			}
 
 			link.setHandler(this.eventType, message => {
 				// XXX Should event handlers even be allowed to be async?
-				handler.call(link, message).catch(err => {
+				handler.call(link, message, this).catch(err => {
 					console.error(`Unexpected error while handling ${this.eventType}`, err);
 				});
 			}, this._eventValidator);
@@ -404,7 +411,8 @@ messages.updateInstances = new Event({
 
 messages.instanceOutput = new Event({
 	type: 'instance_output',
-	links: ['slave-master', 'master-control'],
+	links: ['instance-slave', 'slave-master', 'master-control'],
+	forwardTo: 'master',
 	eventProperties: {
 		"instance_id": { type: "integer" },
 		"output": {
@@ -444,7 +452,6 @@ function attachAllMessages(link) {
 
 module.exports = {
 	Request,
-	InstanceRequest,
 	Event,
 
 	attachAllMessages,

master.js

@@ -255,7 +255,7 @@ class BaseConnection extends link.Link {
 		link.attachAllMessages(this);
 	}
 
-	async forwardInstanceRequest(request, message) {
+	async forwardRequestToInstance(message, request) {
 		let instance = db.instances.get(message.data.instance_id);
 		if (!instance) {
 			throw new errors.RequestError(`Instance with ID ${instance_id} does not exist`);
@@ -266,7 +266,7 @@ class BaseConnection extends link.Link {
 			throw new errors.RequestError("Slave containing instance is not connected");
 		}
 
-		return await request.forward(connection, message);
+		return await request.send(connection, message.data);
 	}
 }
 
@@ -542,9 +542,6 @@ async function getPlugins(){
 	return plugins;
 }
 
-// Errror class for known errors occuring during startup
-class StartupError extends Error { }
-
 /**
  * Calls listen on server capturing any errors that occurs
  * binding to the port.
@@ -552,7 +549,7 @@ class StartupError extends Error { }
 function listen(server, ...args) {
 	return new Promise((resolve, reject) => {
 		function wrapError(err) {
-			reject(new StartupError(
+			reject(new errors.StartupError(
 				`Server listening failed: ${err.message}`
 			));
 		}
@@ -640,7 +637,7 @@ async function startServer() {
 			privateKey = await fs.readFile(config.sslPrivKey);
 
 		} catch (err) {
-			throw new StartupError(
+			throw new errors.StartupError(
 				`Error loading ssl certificate: ${err.message}`
 			);
 		}
@@ -680,19 +677,19 @@ I           version of clusterio.  Expect things to break. I
 `
 	);
 	startServer().catch(err => {
-		if (!(err instanceof StartupError)) {
+		if (err instanceof errors.StartupError) {
+			console.error(`
++----------------------------------+
+| Unable to to start master server |
++----------------------------------+`
+			);
+		} else {
 			console.error(`
 +---------------------------------------------------------------+
 | Unexpected error occured while starting master, please report |
 | it to https://github.com/clusterio/factorioClusterio/issues   |
 +---------------------------------------------------------------+`
 			);
-		} else {
-			console.error(`
-+----------------------------------+
-| Unable to to start master server |
-+----------------------------------+`
-			);
 		}
 
 		console.error(err);

test/client.js

@@ -2,11 +2,12 @@ const assert = require('assert').strict;
 const path = require('path');
 const fs = require('fs-extra');
 
+const link = require("lib/link");
 const client = require("../client");
 
 describe("Client testing", function() {
 	describe("class Instance", function() {
-		let instance = new client._Instance("dir", "factorioDir", { name:"foo" })
+		let instance = new client._Instance(new link.VirtualConnector(), "dir", "factorioDir", { name:"foo" });
 		it("should give the name on .name", function() {
 			assert.equal(instance.name, "foo");
 		})
@@ -81,7 +82,9 @@ describe("Client testing", function() {
 		fs.outputFileSync(path.join(testDir, "shared", "mod_a.zip"), "a");
 		fs.outputFileSync(path.join(testDir, "shared", "mod_b.zip"), "b");
 		fs.outputFileSync(path.join(testDir, "shared", "mod.dat"), "c");
-		let instance = new client._Instance(path.join(testDir, "instance"), "factorioDir", { name: "test" });
+		let instance = new client._Instance(
+			new link.VirtualConnector(), path.join(testDir, "instance"), "factorioDir", { name: "test" }
+		);
 		fs.outputFileSync(instance.path("mods", "mod_i.zip"), "i");
 
 		let discardingLogger = {
@@ -127,4 +130,24 @@ describe("Client testing", function() {
 			assert.equal(await fs.readFile(instance.path("mods", "mod_i.zip"), "utf-8"), "i");
 		});
 	});
+
+	describe("discoverInstances()", function() {
+		it("should discover test instance", async function() {
+			let logger = { log: () => {}, error: () => {} };
+			let instancePath = path.join("test", "file", "instances");
+			let instanceInfos = await client._discoverInstances(instancePath, logger);
+			assert.deepEqual(instanceInfos, new Map([
+				[1, {
+					config: {
+						id: 1,
+						name: "test",
+						factorioPort: null,
+						clientPort: null,
+						clientPassword: null
+					},
+					path: path.join(instancePath, "test"),
+				}],
+			]));
+		});
+	});
 });

test/file/instances/test/config.json

@@ -0,0 +1,7 @@
+{
+    "id": 1,
+    "name": "test",
+    "factorioPort": null,
+    "clientPort": null,
+    "clientPassword": null
+}

test/integration/clusterio.js

@@ -63,7 +63,6 @@ describe("Integration of Clusterio", function() {
 			masterAuthToken: "invalid",
 			publicIP: "invalid",
 		});
-		testSlave.instances = await testSlave.findInstances();
 		testSlaveConnection = new master._SlaveConnection({ agent: "test", version, name: "slave", id: 4}, slaveServer);
 		master._slaveConnections.set(4, testSlaveConnection);
 	});

test/lib/fileOps/fileOps.js

@@ -21,6 +21,10 @@ describe("fileOps.js", function(){
 			let newest = await fileOps.getNewestFile(path.join(baseDir, "test"));
 			assert.equal(typeof newest, "string");
 		});
+		it("returns null if all entries were filtered out", async function() {
+			let newest = await fileOps.getNewestFile(path.join(baseDir, "test"), (name) => !name.endsWith(".txt"));
+			assert.equal(newest, null);
+		});
 		it("returns null if directory is empty", async function() {
 			let newest = await fileOps.getNewestFile(path.join(baseDir, "test", "folder"));
 			assert.equal(newest, null);

test/lib/link/link.js

@@ -9,17 +9,49 @@ const mock = require("../../mock");
 
 describe("lib/link/link", function() {
 	describe("class Link", function() {
-		let testLink = new link.Link('source', 'target', new mock.MockConnector());
-		testLink.setValidator('simple', schema.compile({
-			properties: {
-				"data": {
-					type: 'object',
-					properties: {
-						"string": { type: 'string' },
+		let testConnector;
+		let testLink;
+
+		before(function() {
+			testConnector = new mock.MockConnector();
+			testLink = new link.Link('source', 'target', testConnector);
+			testLink.setValidator('simple', schema.compile({
+				properties: {
+					"data": {
+						type: 'object',
+						properties: {
+							"string": { type: 'string' },
+						},
 					},
 				},
-			},
-		}));
+			}));
+		});
+
+		it("should close connector on invalid messages", function() {
+			testConnector.sentMessages = [];
+			testConnector.emit('message', { invalid: true });
+			assert.deepEqual(
+				testConnector.sentMessages,
+				[{
+					seq: testConnector._seq - 1,
+					type: 'close',
+					data: { reason: "Invalid message: Malformed" },
+				}]
+			);
+		});
+
+		it("should disconnect on close", function() {
+			testConnector.disconnectCalled = false;
+			testConnector.emit('message', { seq: 1, type: 'close', data: { reason: "test" }});
+			assert(testConnector.disconnectCalled, "disconnect was not called");
+		});
+
+		it("should emitt error processing message to connector", async function() {
+			testLink.setHandler('throws', () => { throw new Error("Handler error"); }, () => true);
+			let result = events.once(testConnector, 'message');
+			testConnector.emit('message', { seq: 1, type: 'throws', data: {}});
+			await assert.rejects(result, new Error("Handler error"));
+		});
 
 		describe(".setValidator()", function() {
 			it("should set the validator", function() {
@@ -58,13 +90,13 @@ describe("lib/link/link", function() {
 			it("should throw on invalid message", function() {
 				assert.throws(
 					() => testLink.processMessage({ data: "invalid message" }),
-					new errors.InvalidMessage("Malformed message")
+					new errors.InvalidMessage("Malformed")
 				);
 			});
 			it("should throw on message without validator", function() {
 				assert.throws(
 					() => testLink.processMessage({ seq: 1, type: 'no_validator', data: {} }),
-					new errors.InvalidMessage("No validator for no_validator")
+					new errors.InvalidMessage("No validator for no_validator on source-target")
 				);
 			});
 			it("should throw on message failing validation", function() {
@@ -130,11 +162,14 @@ describe("lib/link/link", function() {
 		});
 
 		describe(".waitFor()", function() {
-			testLink.setValidator('waiter', (message) => true);
+			before(function() {
+				testLink.setValidator('waiter', (message) => true);
+			});
+
 			it("should throw on missing validator", async function() {
 				await assert.rejects(
 					testLink.waitFor('no_validator', {}),
-					new Error("no validator for no_validator")
+					new Error("No validator for no_validator on source-target")
 				);
 			});
 			it("should wait for a given message", async function() {

test/lib/link/messages.js

@@ -18,6 +18,15 @@ describe("lib/link/messages", function() {
 			links: ['source-target'],
 		});
 
+		describe("constructor", function() {
+			it("should throw on invalid forwardTo", function() {
+				assert.throws(
+					() => new link.Request({ forwardTo: 'invalid'}),
+					new Error("Invalid forwardTo value invalid")
+				);
+			});
+		});
+
 		describe(".attach()", function() {
 			it("should attach validator to a source link", function() {
 				testRequest.attach(testSourceLink);
@@ -26,7 +35,7 @@ describe("lib/link/messages", function() {
 			it("should throw if missing handler on target link", function() {
 				assert.throws(
 					() => testRequest.attach(testTargetLink),
-					new Error("Missing handler for test_request on source-target link")
+					new Error("Missing handler for test_request on target-source link")
 				);
 			});
 
@@ -35,20 +44,6 @@ describe("lib/link/messages", function() {
 				testRequest.attach(testTargetLink, async (message) => handlerResult );
 				assert(testTargetLink._handlers.has('test_request'), "Handler was not set");
 			});
-			it("should throw on handler returning object with seq", async function() {
-				handlerResult = { seq: 3 };
-				testTargetLink._handlers.get('test_request')({ seq: 2 });
-				let result = await new Promise(resolve => {
-					testTargetLink.connector.send = (type, data) => resolve({ type, data });
-				});
-				assert.deepEqual(result, {
-					type: 'test_response',
-					data: {
-						error: "response contains reserved property 'seq'",
-						seq: 2,
-					},
-				});
-			});
 			it("should send result of calling the handler", async function() {
 				handlerResult = { test: "handler" };
 				testTargetLink._handlers.get('test_request')({ seq: 2 });
@@ -107,20 +102,26 @@ describe("lib/link/messages", function() {
 		});
 	});
 
-	describe("class InstanceRequest", function() {
-	});
-
 	describe("class Event", function() {
 		let testEvent = new link.Event({
 			type: 'test',
 			links: ['source-target'],
 		});
 
+		describe("constructor", function() {
+			it("should throw on invalid forwardTo", function() {
+				assert.throws(
+					() => new link.Event({ forwardTo: 'invalid'}),
+					new Error("Invalid forwardTo value invalid")
+				);
+			});
+		});
+
 		describe(".attach()", function() {
 			it("should throw if missing handler on target link", function() {
 				assert.throws(
 					() => testEvent.attach(testTargetLink),
-					new Error("Missing handler for test_event on source-target link")
+					new Error("Missing handler for test_event on target-source link")
 				);
 			});
 			let called = false;