Refactor instance communication

Change Instance class to be a subclass of Link and refactor the
communication to and establisment of instances to use a virtual
connection between the slave and the instance.  This is simplifies the
implementation of plugin communication for instances.

Replace InstanceRequest with a forwardTo option to Request which does
the task of forwarding in a much cleaner way.

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,21 @@ 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"));
+		console.log(`Creating ${instanceDir}`);
+		await fs.ensureDir(instanceDir);
+		await fs.ensureDir(path.join(instanceDir, "script-output"));
+		await fs.ensureDir(path.join(instanceDir, "saves"));
 
-		await symlinkMods(instance, "sharedMods", console);
 		console.log("Clusterio | Created instance with settings:")
 		console.log(instanceConfig);
 
 		// 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,26 +119,13 @@ 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{
@@ -137,17 +142,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 +175,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 +194,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 +207,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 +259,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 +361,8 @@ class Slave extends link.Link {
 			publicAddress: slaveConfig.publicIP,
 		}
 
+		this.instanceConnections = new Map();
+		this.instanceInfos = new Map();
 	}
 
 	async _findNewInstanceDir(name) {
@@ -294,125 +381,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 +898,7 @@ module.exports = {
 	_Instance: Instance,
 	_checkFilename: checkFilename,
 	_symlinkMods: symlinkMods,
+	_discoverInstances: discoverInstances,
 	_Slave: Slave,
 };
 

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

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/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);
@@ -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,15 +102,21 @@ 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(