Merge pull request #265 from Hornwitser/link-refactor

lib/link refactor and plugin preparations

CHANGELOG.md

@@ -58,6 +58,7 @@ Version 2.0.0
   instance is now three separate commands.
 - Removed oddball limits to slaves.json size.
 - Moved slave specific configuration into its own configuration file.
+- Removed unused binary option from plugin config.
 
 ### Breaking Changes
 

client.js

@@ -75,7 +75,7 @@ class Instance {
 		// save instance config
 		await fs.outputFile(instance.path("config.json"), JSON.stringify(instanceConfig, null, 4));
 
-		let serverSettings = await instance.server.exampleSettings();
+		let serverSettings = await factorio.FactorioServer.exampleSettings(path.join(factorioDir, "data"));
 		let gameName = "Clusterio instance: " + instance.name;
 		if (options.username) {
 			gameName = options.username + "'s clusterio " + instance.name;
@@ -197,7 +197,7 @@ class Instance {
 				clientPort: this.server.rconPort,
 				clientPassword: this.server.rconPassword,
 			}
-			instanceManagement(slaveConfig, this, compatConfig, this.server, socket); // XXX async function
+			//instanceManagement(slaveConfig, this, compatConfig, this.server, socket); // XXX async function
 		});
 
 		await this.server.start(latestSave);
@@ -235,17 +235,36 @@ class Instance {
 	}
 }
 
+class SlaveConnector extends link.SocketIOClientConnector {
+	constructor(slaveConfig) {
+		super(slaveConfig.masterURL, slaveConfig.masterAuthToken);
+
+		this.id = slaveConfig.id;
+		this.name = slaveConfig.name;
+	}
+
+	register() {
+		console.log("SOCKET | registering slave");
+		this.send('register_slave', {
+			agent: 'Clusterio Slave',
+			version,
+			id: this.id,
+			name: this.name,
+		});
+	}
+}
+
 /**
  * Handles running the slave
  *
  * Connects to the master server over the socket.io connection and manages
  * intsances.
  */
-class Slave extends link.Client {
+class Slave extends link.Link {
 	// I don't like God classes, but the alternative of putting all this state
 	// into global variables is not much better.
-	constructor(slaveConfig) {
-		super('slave', slaveConfig.masterURL, slaveConfig.masterAuthToken);
+	constructor(connector, slaveConfig) {
+		super('slave', 'master', connector);
 		link.attachAllMessages(this);
 		this.config = {
 			id: slaveConfig.id,
@@ -256,16 +275,7 @@ class Slave extends link.Client {
 			masterToken: slaveConfig.masterAuthToken,
 			publicAddress: slaveConfig.publicIP,
 		}
-	}
 
-	register() {
-		console.log("SOCKET | registering slave");
-		this.send('register_slave', {
-			agent: 'Clusterio Slave',
-			version,
-			id: this.config.id,
-			name: this.config.name,
-		});
 	}
 
 	async _findNewInstanceDir(name) {
@@ -289,7 +299,7 @@ class Slave extends link.Client {
 	 *
 	 * Called by the handler for InstanceRequest.
 	 */
-	async forwardInstanceRequest(message, handler) {
+	async forwardInstanceRequest(handler, message) {
 		let instance = this.instances.get(message.data.instance_id);
 		if (!instance) {
 			throw new errors.RequestError(`Instance with ID ${instanceId} does not exist`);
@@ -312,27 +322,27 @@ class Slave extends link.Client {
 		this.updateInstances();
 	}
 
-	async createSaveRequestHandler(instance, message) {
+	async createSaveInstanceRequestHandler(instance, message) {
 		await instance.createSave();
 	}
 
-	async startInstanceRequestHandler(instance, message) {
+	async startInstanceInstanceRequestHandler(instance, message) {
 		await instance.start(this.config, this.socket);
 	}
 
-	async stopInstanceRequestHandler(instance, message) {
+	async stopInstanceInstanceRequestHandler(instance, message) {
 		await instance.stop();
 	}
 
-	async deleteInstanceRequestHandler(instance, message) {
+	async deleteInstanceInstanceRequestHandler(instance, message) {
 		await instance.stop();
 		this.instances.delete(instance.config.id);
 
-		fs.remove(instance.path());
+		await fs.remove(instance.path());
 		this.updateInstances();
 	}
 
-	async sendRconRequestHandler(instance, message) {
+	async sendRconInstanceRequestHandler(instance, message) {
 		let result = await instance.server.sendRcon(message.data.command);
 		return { result };
 	}
@@ -378,7 +388,7 @@ class Slave extends link.Client {
 
 	hookInstance(instance) {
 		instance.server.on('output', (output) => {
-			link.events.instanceOutput.send(this, { instance_id: instance.config.id, output })
+			link.messages.instanceOutput.send(this, { instance_id: instance.config.id, output })
 		});
 	}
 
@@ -391,7 +401,7 @@ class Slave extends link.Client {
 				name: instance.config.name,
 			});
 		}
-		link.events.updateInstances.send(this, { instances: list });
+		link.messages.updateInstances.send(this, { instances: list });
 	}
 
 	async start() {
@@ -626,7 +636,8 @@ async function startClient() {
 		return;
 	}
 
-	let slave = new Slave(slaveConfig);
+	let slaveConnector = new SlaveConnector(slaveConfig);
+	let slave = new Slave(slaveConnector, slaveConfig);
 
 	// Handle interrupts
 	let secondSigint = false
@@ -645,14 +656,13 @@ async function startClient() {
 		});
 	});
 
+	await slaveConnector.connect();
 	await slave.start();
 
 	/*
 	} else if (command == "manage"){
 		await manage(config, instance);
 		// process.exit(0);
-	} else if (command == "delete") {
-		await deleteInstance(instance);
 	*/
 }
 
@@ -690,7 +700,7 @@ async function instanceManagement(slaveConfig, instance, instanceconfig, server,
 		if(!global.subscribedFiles) {
 			global.subscribedFiles = {};
 		}
-		if(pluginConfig.binary == "nodePackage" && pluginConfig.enabled){
+		if (pluginConfig.enabled) {
 			// require plugin class and execute it
 			let pluginClass = require(path.resolve(pluginConfig.pluginPath, "index"));
 			plugins[i] = new pluginClass(combinedConfig, async function(data, callback){
@@ -810,6 +820,7 @@ module.exports = {
 	_Instance: Instance,
 	_checkFilename: checkFilename,
 	_symlinkMods: symlinkMods,
+	_Slave: Slave,
 };
 
 if (module === require.main) {

clusterctl.js

@@ -19,9 +19,9 @@ const errors = require("lib/errors");
 function formatOutputColored(output) {
 	let time = "";
 	if (output.format === 'seconds') {
-		time = chalk.dim(output.time.padStart(8)) + ' ';
+		time = chalk.yellow(output.time.padStart(8)) + ' ';
 	} else if (output.format === 'date') {
-		time = chalk.dim(output.time) + ' ';
+		time = chalk.yellow(output.time) + ' ';
 	}
 
 	let info = "";
@@ -35,7 +35,7 @@ function formatOutputColored(output) {
 			level = chalk.bold.redBright(level);
 		}
 
-		info = level + ' ' + chalk.dim(output.file) + ': ';
+		info = level + ' ' + chalk.gray(output.file) + ': ';
 
 	} else if (output.type === 'action') {
 		info = '[' + chalk.yellow(output.action) + '] ';
@@ -44,31 +44,6 @@ function formatOutputColored(output) {
 	return time + info + output.message;
 }
 
-/**
- * Format a parsed Factorio output message
- *
- * Formats a parsed Factorio output from lib/factorio into a readable
- * output string that can be printed
- */
-function formatOutput(output) {
-	let time = "";
-	if (output.format === 'seconds') {
-		time = output.time.padStart(8) + ' ';
-	} else if (output.format === 'date') {
-		time = output.time + ' ';
-	}
-
-	let info = "";
-	if (output.type === 'log') {
-		info = output.level + ' ' + output.file + ': ';
-
-	} else if (output.type === 'action') {
-		info = '[' + output.action + '] ';
-	}
-
-	return time + info + output.message;
-}
-
 /**
  * Represens a command that can be runned by clusterctl
  */
@@ -103,7 +78,7 @@ async function resolveInstance(control, instanceName) {
 	if (/^-?\d+$/.test(instanceName)) {
 		instanceId = parseInt(instanceName, 10);
 	} else {
-		let response = await link.requests.listInstances.send(control);
+		let response = await link.messages.listInstances.send(control);
 		for (let instance of response.list) {
 			if (instance.name === instanceName) {
 				instanceId = instance.id;
@@ -123,7 +98,7 @@ let commands = [];
 commands.push(new Command({
 	definition: ["list-slaves", "List slaves connected to the master"],
 	handler: async function(args, control) {
-		let response = await link.requests.listSlaves.send(control);
+		let response = await link.messages.listSlaves.send(control);
 		console.log(asTable(response.list));
 	},
 }));
@@ -131,7 +106,7 @@ commands.push(new Command({
 commands.push(new Command({
 	definition: ["list-instances", "List instances known to the master"],
 	handler: async function(args, control) {
-		let response = await link.requests.listInstances.send(control);
+		let response = await link.messages.listInstances.send(control);
 		console.log(asTable(response.list));
 	}
 }));
@@ -148,7 +123,7 @@ commands.push(new Command({
 		if (/^-?\d+$/.test(args.slave)) {
 			slaveId = parseInt(args.slave, 10);
 		} else {
-			let response = await link.requests.listSlaves.send(control);
+			let response = await link.messages.listSlaves.send(control);
 			for (let slave of response.list) {
 				if (slave.name === args.slave) {
 					slaveId = slave.id;
@@ -160,7 +135,7 @@ commands.push(new Command({
 				throw new errors.CommandError(`No slave named ${args.slave}`);
 			}
 		}
-		let response = await link.requests.createInstanceCommand.send(control, {
+		let response = await link.messages.createInstanceCommand.send(control, {
 			name: args.name,
 			slave_id: slaveId,
 		});
@@ -176,8 +151,8 @@ commands.push(new Command({
 	}],
 	handler: async function(args, control) {
 		let instanceId = await resolveInstance(control, args.instance);
-		await link.requests.setInstanceOutputSubscriptions.send(control, { instance_ids: [instanceId] });
-		let response = await link.requests.createSave.send(control, {
+		await link.messages.setInstanceOutputSubscriptions.send(control, { instance_ids: [instanceId] });
+		let response = await link.messages.createSave.send(control, {
 			instance_id: instanceId,
 		});
 		console.log(response);
@@ -192,8 +167,8 @@ commands.push(new Command({
 	}],
 	handler: async function(args, control) {
 		let instanceId = await resolveInstance(control, args.instance);
-		await link.requests.setInstanceOutputSubscriptions.send(control, { instance_ids: [instanceId] });
-		let response = await link.requests.startInstance.send(control, {
+		await link.messages.setInstanceOutputSubscriptions.send(control, { instance_ids: [instanceId] });
+		let response = await link.messages.startInstance.send(control, {
 			instance_id: instanceId,
 		});
 		console.log(response);
@@ -208,8 +183,8 @@ commands.push(new Command({
 	}],
 	handler: async function(args, control) {
 		let instanceId = await resolveInstance(control, args.instance);
-		await link.requests.setInstanceOutputSubscriptions.send(control, { instance_ids: [instanceId] });
-		let response = await link.requests.stopInstance.send(control, {
+		await link.messages.setInstanceOutputSubscriptions.send(control, { instance_ids: [instanceId] });
+		let response = await link.messages.stopInstance.send(control, {
 			instance_id: instanceId,
 		});
 		console.log(response);
@@ -223,7 +198,7 @@ commands.push(new Command({
 		});
 	}],
 	handler: async function(args, control) {
-		let response = await link.requests.deleteInstance.send(control, {
+		let response = await link.messages.deleteInstance.send(control, {
 			instance_id: await resolveInstance(control, args.instance),
 		});
 		console.log(response);
@@ -238,7 +213,7 @@ commands.push(new Command({
 		});
 	}],
 	handler: async function(args, control) {
-		let response = await link.requests.sendRcon.send(control, {
+		let response = await link.messages.sendRcon.send(control, {
 			instance_id: await resolveInstance(control, args.instance),
 			command: args.command
 		})
@@ -253,20 +228,9 @@ commands = new Map([...commands.map(command => [command.name, command])]);
 
 
 /**
- * Handles running the control
- *
- * Connects to the master server over the socket.io connection and sends
- * commands to it.
+ * Connector for control connection to master server
  */
-class Control extends link.Client {
-
-	// I don't like God classes, but the alternative of putting all this state
-	// into global variables is not much better.
-	constructor(controlConfig) {
-		super('control', controlConfig.url, controlConfig.token);
-		link.attachAllMessages(this);
-	}
-
+class ControlConnector extends link.SocketIOClientConnector {
 	register() {
 		console.log("SOCKET | registering control");
 		this.send('register_control', {
@@ -274,14 +238,26 @@ class Control extends link.Client {
 			version: version,
 		});
 	}
+}
+
+/**
+ * Handles running the control
+ *
+ * Connects to the master server over the socket.io connection and sends
+ * commands to it.
+ */
+class Control extends link.Link {
+
+	// I don't like God classes, but the alternative of putting all this state
+	// into global variables is not much better.
+	constructor(connector) {
+		super('control', 'master', connector);
+		link.attachAllMessages(this);
+	}
 
 	async instanceOutputEventHandler(message) {
 		let { instance_id, output } = message.data;
-		if (process.platform === "win32") {
-			console.log(formatOutput(output));
-		} else {
-			console.log(formatOutputColored(output));
-		}
+		console.log(formatOutputColored(output));
 	}
 }
 
@@ -395,8 +371,9 @@ async function startControl() {
 		}
 	}
 
-	let control = new Control(controlConfig);
-	await control.connect();
+	let controlConnector = new ControlConnector(controlConfig.url, controlConfig.token);
+	let control = new Control(controlConnector);
+	await controlConnector.connect();
 
 	if (commands.has(commandName)) {
 		command = commands.get(commandName);
@@ -426,6 +403,15 @@ async function startControl() {
 	//XXX control.close("done");
 }
 
+module.exports = {
+	// for testing only
+	_formatOutputColored: formatOutputColored,
+	_resolveInstance: resolveInstance,
+	_Control: Control,
+
+	_commands: commands,
+}
+
 
 if (module === require.main) {
 	console.warn(`

docs/socket.md

@@ -80,6 +80,19 @@ with an empty data payload.
 ```
 
 
+Events
+------
+
+Events are messages send in one direction over the link, that the
+receiver is expected to act upon, but not make any replies (for that
+there's [Requests](#requests).  By convention a message types ending in
+`*_event` is an event.  The data payload of an event consists solely of
+event specific properties.
+
+See [lib/link/messages.js](lib/link/messages.js) for the recognized
+events and their contents.
+
+
 Requests
 --------
 
@@ -100,8 +113,8 @@ error property in the data payload is sent instead.
 - seq - integer - The seq number of the request message.
 - error - string - Human readable text message describing the error.
 
-See [lib/link/requests.js](lib/link/requests.js) for the recognized
-requests and their implementation.
+See [lib/link/messages.js](lib/link/messages.js) for the recognized
+requests and their contents.
 
 
 Closing the connection

lib/factorio/server.js

@@ -753,10 +753,11 @@ class FactorioServer extends events.EventEmitter {
 	 *
 	 * Loads server-settings.example.json from the data dir.
 	 *
+	 * @param {string} dataDir - The data directory for Factorio
 	 * @returns {object} the parsed server-settings.
 	 */
-	async exampleSettings() {
-		return JSON.parse(await fs.readFile(this.dataPath("server-settings.example.json"), "utf-8"));
+	static async exampleSettings(dataDir) {
+		return JSON.parse(await fs.readFile(path.join(dataDir, "server-settings.example.json"), "utf-8"));
 	}
 
 	async _writeConfigIni() {

lib/link/connectors.js

@@ -0,0 +1,229 @@
+"use strict";
+const events = require("events");
+const io = require("socket.io-client");
+
+const schema = require("lib/schema");
+
+
+/**
+ * Connector for master server clients
+ */
+class SocketIOClientConnector extends events.EventEmitter {
+	constructor(url, token) {
+		super();
+
+		this._url = url;
+		this._token = token;
+
+		this._seq = 1;
+		this._state = "new";
+	}
+
+	_check(expectedStates) {
+		if (!expectedStates.includes(this._state)) {
+			throw new Error(
+				`Expected state ${expectedStates} but state is ${this._state}`
+			);
+		}
+	}
+
+	/**
+	 * Send a message over the socket
+	 *
+	 * @returns the sequence number of the message sent
+	 */
+	send(type, data) {
+		this._socket.send({ seq: this._seq, type, data });
+		return this._seq++;
+	}
+
+	/**
+	 * Close the connection with the given reason.
+	 *
+	 * Sends a close message and disconnects the connector.
+	 */
+	close(reason) {
+		this.send('close', { reason });
+		this.disconnect();
+	}
+
+	/**
+	 * Connect to the master server
+	 */
+	async connect() {
+		this._check(["new"]);
+		this._state = "handshake";
+
+		// Open socket.io connection to master
+		console.log(`SOCKET | connecting to ${this._url}`)
+		let url = new URL(this._url);
+		url.searchParams.append('token', this._token);
+
+		// I really have to question the choice of having the path connected
+		// to be passed as separate parameter instead of the namespace.
+		let urlPath = url.pathname + "socket.io";
+		url.pathname = "";
+		this._socket = io(url.href, {
+			path: urlPath,
+			reconnectionAttempts: 2,
+
+			// For now we do not verify TLS certificates since the default setup is
+			// to create a self-signed certificate.
+			rejectUnauthorized: false,
+		});
+
+		this._attachSocketHandlers();
+
+		// Wait for the connection to be ready
+		await events.once(this, 'ready');
+	}
+
+	_attachSocketHandlers() {
+		this._socket.on("error", err => {
+			console.error("SOCKET | Error:", err);
+			this._socket.close();
+		});
+
+		/*this._socket.on("connect_error", err => {
+			console.log("SOCKET | Connect error:", err);
+		});*/
+		this._socket.on("connect_timeout", () => {
+			console.log("SOCKET | Connect timeout");
+		});
+		this._socket.on("reconnect", attempt => {
+			console.log(`SOCKET | Reconnect no. ${attempt} succeeded`);
+		});
+		this._socket.on("reconnect_attempt", () => {
+			console.log("SOCKET | Attempting reconnect");
+		});
+		this._socket.on("reconnecting", attempt => {
+			console.log(`SOCKET | Attempting reconnect no. ${attempt}`);
+		});
+		/*this._socket.on("reconnect_error", err => {
+			console.log("SOCKET | Reconnect error:", err);
+		});*/
+		this._socket.on("reconnect_failed", () => {
+			console.log("SOCKET | Reconnecting failed");
+		});
+		/*this._socket.on("ping", () => {
+			console.log("SOCKET | Ping");
+		});
+		this._socket.on("pong", (ms) => {
+			// XXX Latency might be useful
+			console.log(`SOCKET | Pong latency ${ms}ms`);
+		});*/
+		this._socket.on("connect", () => {
+			console.log("SOCKET | connected");
+		});
+
+		// Handle messages
+		this._socket.on("message", message => {
+			if (this._state === "handshake") {
+				this._processHandshake(message);
+			} else if (this._state === "ready") {
+				this.emit("message", message);
+			} else {
+				throw new Error(`Received message in unexpected state ${this._state}`);
+			}
+		});
+	}
+
+	/**
+	 * Immediatly close the connection
+	 */
+	disconnect() {
+		this._socket.close();
+	}
+
+	_processHandshake(message) {
+		if (!schema.serverHandshake(message)) {
+			console.log("SOCKET | closing after received invalid handshake:", message);
+			this.close("Invalid handshake");
+			return;
+		}
+
+		let { seq, type, data } = message;
+		if (type == 'hello') {
+			console.log(`SOCKET | received hello from master version ${data.version}`);
+			this.register();
+
+		} else if (type === 'ready') {
+			console.log("SOCKET | received ready from master");
+			this._state = "ready";
+			this.emit('ready');
+
+		} else if (type === 'close') {
+			console.log(`SOCKET | received close from server: ${data.reason}`);
+			this.emit('error', new Error(`server closed during handshake: ${data.reason}`));
+			this._socket.close();
+		}
+	}
+
+	/**
+	 * Register the link with the server
+	 *
+	 * This function is expected to be overriden by sub-classes and send the
+	 * register message over the socket when invoked.
+	 */
+	register() {
+		throw Error("Abstract function");
+	}
+}
+
+class VirtualConnector extends events.EventEmitter {
+	constructor() {
+		super();
+
+		this.other = this;
+		this._seq = 1;
+	}
+
+	/**
+	 * Create a pair of virtual connector hooked into each other
+	 *
+	 * Creates two virtual connectors that are liked to each othes such that
+	 * a message sent on one is received by the other.
+	 *
+	 * @returns {Array} two virtuarl connectors.
+	 */
+	static makePair() {
+		let first = new this();
+		let second = new this();
+		first.other = second;
+		second.other = first;
+		return [first, second];
+	}
+
+	/**
+	 * Send a message to the other end of the connector
+	 *
+	 * @returns the sequence number of the message sent
+	 */
+	send(type, data) {
+		this.other.emit('message', { seq: this._seq, type, data });
+		return this._seq++;
+	}
+
+	/**
+	 * Close the connection with the given reason.
+	 *
+	 * Sends a close message to the other end of the connector with the
+	 * given reason.  Doesn't actually disconnect as virtual connectors
+	 * doen't have a concept of disconnecting.
+	 */
+	close(reason) {
+		this.send('close', { reason });
+	}
+
+	/**
+	 * Does nothing
+	 *
+	 * Virtual connectors do not have a concept of disconnecting.
+	 */
+	disconnect() { }
+}
+
+module.exports = {
+	SocketIOClientConnector,
+	VirtualConnector,
+};

lib/link/index.js

@@ -1,4 +1,5 @@
 module.exports = {
 	...require('lib/link/link'),
 	...require('lib/link/messages'),
+	...require('lib/link/connectors'),
 };

lib/link/link.js

@@ -1,6 +1,4 @@
 "use strict";
-const events = require("events");
-const io = require("socket.io-client");
 
 const schema = require("lib/schema");
 const errors = require("lib/errors");
@@ -8,6 +6,7 @@ const messages = require("lib/link/messages");
 
 // Some definitions for the terminology used here:
 // link: Either side of a master - client connection
+// connector: Adapter between a link and a socket connection
 // connection: the master server side of a link
 // client: the side that is not the master server of a link
 // message: the complete object sent using the 'message' event
@@ -18,15 +17,32 @@ const messages = require("lib/link/messages");
  * Common interface for server and client connections
  */
 class Link {
-	constructor(source, target, socket) {
+	constructor(source, target, connector) {
 		this.source = source;
 		this.target = target;
-		this.socket = socket;
+		this.connector = connector;
 
-		this._seq = 1;
 		this._waiters = new Map();
 		this._handlers = new Map();
 		this._validators = new Map();
+
+		// Process messages received by the connector
+		connector.on('message', payload => {
+			try {
+				this.processMessage(payload);
+			} catch (err) {
+				if (err instanceof errors.InvalidMessage) {
+					this.connector.close(`Invalid message: ${err.message}`);
+				} else {
+					throw err;
+				}
+			}
+		});
+
+		this.setHandler('close', message => {
+			console.log(`SOCKET | Received close: ${message.data.reason}`);
+			this.connector.disconnect();
+		}, schema.close);
 	}
 
 	/**
@@ -154,16 +170,6 @@ class Link {
 		this._validators.set(type, validator);
 	}
 
-	/**
-	 * Send a message over the link
-	 *
-	 * @returns the sequence number of the message sent
-	 */
-	send(type, data) {
-		this.socket.send({ seq: this._seq, type, data });
-		return this._seq++;
-	}
-
 	/**
 	 * Wait for a message matching given type and data
 	 *
@@ -185,215 +191,9 @@ class Link {
 			}
 		});
 	}
-
-	/**
-	 * Close the connection with the given reason.
-	 *
-	 * Sends a close message and disconnects from the socket.
-	 */
-	close(reason) {
-		this.send('close', { reason });
-		this.disconnect();
-	}
-
-
-	/**
-	 * Disconnect from the socket
-	 *
-	 * Abstract function that subclases override to implement disconnection
-	 * from the socket.
-	 */
-	disconnect() {
-		throw new Error("Abstract function");
-	}
 }
 
-class Connection extends Link {
-	constructor(target, socket) {
-		super('master', target, socket);
-
-		this._seq = 3; // XXX this is private to Link
-		this.socket.on('message', payload => {
-			try {
-				this.processMessage(payload);
-			} catch (err) {
-				if (err instanceof errors.InvalidMessage) {
-					this.close(`Invalid message: ${err.message}`);
-				} else {
-					throw err;
-				}
-			}
-		});
-		this.socket.on('disconnect', this.disconnect.bind(this));
-
-		this.setHandler('close', payload => {
-			let address = this.socket.handshake.address
-			console.log(`SOCKET | received close from control ${address}: ${payload.data.reason}`);
-			this.socket.disconnect(true);
-		}, schema.close);
-
-		socket.send({ seq: 2, type: 'ready', data: {} });
-	}
-
-	disconnect() {
-		this.socket.disconnect(true);
-	}
-}
-
-
-class Client extends Link {
-	constructor(source, url, token) {
-		super(source, 'master', null);
-
-		this._url = url;
-		this._token = token;
-
-		this._state = "new";
-		this._events = new events.EventEmitter();
-
-		this.setHandler('close', message => {
-			console.log(`SOCKET | received close from server: ${message.data.reason}`);
-			this.socket.close();
-			return; // XXX what now?
-		}, schema.close);
-	}
-
-	_check(expectedStates) {
-		if (!expectedStates.includes(this._state)) {
-			throw new Error(
-				`Expected state ${expectedStates} but state is ${this._state}`
-			);
-		}
-	}
-
-	async connect() {
-		this._check(["new"]);
-		this._state = "handshake";
-
-		// Open socket.io connection to master
-		console.log(`SOCKET | connecting to ${this._url}`)
-		let url = new URL(this._url);
-		url.searchParams.append('token', this._token);
-
-		// I really have to question the choice of having the path connected
-		// to be passed as separate parameter instead of the namespace.
-		let urlPath = url.pathname + "socket.io";
-		url.pathname = "";
-		this.socket = io(url.href, {
-			path: urlPath,
-			reconnectionAttempts: 2,
-
-			// For now we do not verify TLS certificates since the default setup is
-			// to create a self-signed certificate.
-			rejectUnauthorized: false,
-		});
-
-		_attachSocketHandlers();
-
-		// Wait for the connection to be ready
-		await events.once(this._events, 'ready');
-	}
-
-	_attachSocketHandlers() {
-		this.socket.on("error", err => {
-			console.error("SOCKET | Error:", err);
-			this.socket.close();
-		});
-
-		/*this.socket.on("connect_error", err => {
-			console.log("SOCKET | Connect error:", err);
-		});*/
-		this.socket.on("connect_timeout", () => {
-			console.log("SOCKET | Connect timeout");
-		});
-		this.socket.on("reconnect", attempt => {
-			console.log(`SOCKET | Reconnect no. ${attempt} succeeded`);
-		});
-		this.socket.on("reconnect_attempt", () => {
-			console.log("SOCKET | Attempting reconnect");
-		});
-		this.socket.on("reconnecting", attempt => {
-			console.log(`SOCKET | Attempting reconnect no. ${attempt}`);
-		});
-		/*this.socket.on("reconnect_error", err => {
-			console.log("SOCKET | Reconnect error:", err);
-		});*/
-		this.socket.on("reconnect_failed", () => {
-			console.log("SOCKET | Reconnecting failed");
-		});
-		/*this.socket.on("ping", () => {
-			console.log("SOCKET | Ping");
-		});
-		this.socket.on("pong", (ms) => {
-			// XXX Latency might be useful
-			console.log(`SOCKET | Pong latency ${ms}ms`);
-		});*/
-		this.socket.on("connect", () => {
-			console.log("SOCKET | connected");
-		});
-
-		// Handle messages
-		this.socket.on("message", message => {
-			if (this._state === "handshake") {
-				this._processHandshake(message);
-			} else if (this._state === "ready") {
-				try {
-					this.processMessage(message);
-				} catch (err) {
-					if (err instanceof errors.InvalidMessage) {
-						this.close(`Invalid message: ${err.message}`);
-					} else {
-						throw err;
-					}
-				}
-			} else {
-				throw new Error(`Received message in unexpected state ${this._state}`);
-			}
-		})
-	}
-
-	disconnect(reason) {
-		this.socket.close();
-		// XXX what now?
-	}
-
-	_processHandshake(message) {
-		if (!schema.serverHandshake(message)) {
-			console.log("SOCKET | closing after received invalid handshake:", message);
-			this.close("Invalid handshake");
-			return;
-		}
-
-		let { seq, type, data } = message;
-		if (type == 'hello') {
-			console.log(`SOCKET | received hello from master version ${data.version}`);
-			this.register();
-
-		} else if (type === 'ready') {
-			console.log("SOCKET | received ready from master");
-			this._state = "ready";
-			this._events.emit('ready');
-
-		} else if (type === 'close') {
-			console.log(`SOCKET | received close from server: ${data.reason}`);
-			this._events.emit('error', new Error(`server closed during handshake: ${data.reason}`));
-			this.socket.close();
-		}
-	}
-
-	/**
-	 * Register the link with the server
-	 *
-	 * This function is expected to be overriden by sub-classes and send the
-	 * register message over the socket when invoked.
-	 */
-	register() {
-		throw Error("Abstract function");
-	}
-}
 
 module.exports = {
 	Link,
-	Connection,
-	Client,
 }

lib/link/messages.js

@@ -10,11 +10,10 @@ const errors = require("lib/errors");
  */
 class Request {
 	constructor({
-		type, sources, targets, requestProperties = {}, responseProperties = {}
+		type, links, requestProperties = {}, responseProperties = {}
 	}) {
 		this.type = type;
-		this.sources = sources;
-		this.targets = targets;
+		this.links = links;
 
 		this.requestType = type + '_request';
 		this.responseType = type + '_response';
@@ -76,12 +75,12 @@ class Request {
 	 */
 	attach(link, handler) {
 		// Source side.
-		if (this.sources.includes(link.source) && this.targets.includes(link.target)) {
+		if (this.links.includes(`${link.source}-${link.target}`)) {
 			link.setValidator(this.responseType, this._responseValidator);
 		}
 
 		// Target side.  Note: Source and target is reversed here.
-		if (this.targets.includes(link.source) && this.sources.includes(link.target)) {
+		if (this.links.includes(`${link.target}-${link.source}`)) {
 			if (!handler) {
 				throw new Error(`Missing handler for ${this.requestType} on ${link.target}-${link.source} link`);
 			}
@@ -96,13 +95,13 @@ class Request {
 					if (Object.prototype.hasOwnProperty.call(response, 'seq')) {
 						throw new errors.RequestError("response contains reserved property 'seq'");
 					}
-					link.send(this.responseType, { seq: message.seq, ...response });
+					link.connector.send(this.responseType, { seq: message.seq, ...response });
 
 				}).catch(err => {
 					if (!(err instanceof errors.RequestError)) {
 						console.error(`Unexpected error while responding to ${this.requestType}`, err);
 					}
-					link.send(this.responseType, { seq: message.seq, error: err.message });
+					link.connector.send(this.responseType, { seq: message.seq, error: err.message });
 				});
 			}, this._requestValidator);
 		}
@@ -118,7 +117,7 @@ class Request {
 	 */
 	async send(link, data = {}) {
 		// XXX validate link target/source?
-		let seq = link.send(this.requestType, data);
+		let seq = link.connector.send(this.requestType, data);
 		let responseMessage = await link.waitFor(this.responseType, { seq });
 		if (responseMessage.data.error) {
 			throw new errors.RequestError(responseMessage.data.error);
@@ -145,8 +144,7 @@ class InstanceRequest {
 	}) {
 		this._masterRequest = new Request({
 			type: type + '_master',
-			sources,
-			targets: ['master'],
+			links: sources.map(source => `${source}-master`),
 			requestProperties: {
 				"instance_id": { type: "integer" },
 				...requestProperties,
@@ -155,8 +153,7 @@ class InstanceRequest {
 		});
 		this._slaveRequest = new Request({
 			type: type + '_slave',
-			sources: ['master'],
-			targets: ['slave'],
+			links: ['master-slave'],
 			requestProperties: {
 				"instance_id": { type: "integer" },
 				...requestProperties,
@@ -182,11 +179,15 @@ class InstanceRequest {
 	 */
 	attach(link, handler) {
 		this._masterRequest.attach(link, async (message) => {
-			return await link.forwardInstanceRequest(message, this);
-		});
-		this._slaveRequest.attach(link, async (message) => {
-			return await link.forwardInstanceRequest(message, handler);
+			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);
+		}
 	}
 
 	/**
@@ -215,13 +216,12 @@ class InstanceRequest {
 	}
 }
 
-let requests = {}
+let messages = {}
 
 // Management requests
-requests.listSlaves = new Request({
+messages.listSlaves = new Request({
 	type: 'list_slaves',
-	sources: ['control'],
-	targets: ['master'],
+	links: ['control-master'],
 	responseProperties: {
 		"list": {
 			type: "array",
@@ -240,10 +240,9 @@ requests.listSlaves = new Request({
 	},
 });
 
-requests.listInstances = new Request({
+messages.listInstances = new Request({
 	type: 'list_instances',
-	sources: ['control'],
-	targets: ['master'],
+	links: ['control-master'],
 	responseProperties: {
 		"list": {
 			type: "array",
@@ -260,10 +259,9 @@ requests.listInstances = new Request({
 	},
 });
 
-requests.setInstanceOutputSubscriptions = new Request({
+messages.setInstanceOutputSubscriptions = new Request({
 	type: 'set_instance_output_subscriptions',
-	sources: ['control'],
-	targets: ['master'],
+	links: ['control-master'],
 	requestProperties: {
 		"instance_ids": {
 			type: "array",
@@ -272,20 +270,18 @@ requests.setInstanceOutputSubscriptions = new Request({
 	},
 });
 
-requests.createInstanceCommand = new Request({
+messages.createInstanceCommand = new Request({
 	type: 'create_instance',
-	sources: ['control'],
-	targets: ['master'],
+	links: ['control-master'],
 	requestProperties: {
 		"name": { type: "string" },
 		"slave_id": { type: "integer" },
 	},
 });
 
-requests.createInstance = new Request({
+messages.createInstance = new Request({
 	type: 'create_instance',
-	sources: ['master'],
-	targets: ['slave'],
+	links: ['master-slave'],
 	requestProperties: {
 		"id": { type: "integer" },
 		"options": {
@@ -310,27 +306,27 @@ requests.createInstance = new Request({
 });
 
 
-requests.createSave = new InstanceRequest({
+messages.createSave = new InstanceRequest({
 	type: 'create_save',
 	sources: ['control'],
 });
 
-requests.startInstance = new InstanceRequest({
+messages.startInstance = new InstanceRequest({
 	type: 'start_instance',
 	sources: ['control'],
 });
 
-requests.stopInstance = new InstanceRequest({
+messages.stopInstance = new InstanceRequest({
 	type: 'stop_instance',
 	sources: ['control'],
 });
 
-requests.deleteInstance = new InstanceRequest({
+messages.deleteInstance = new InstanceRequest({
 	type: 'delete_instance',
 	sources: ['control'],
 });
 
-requests.sendRcon = new InstanceRequest({
+messages.sendRcon = new InstanceRequest({
 	type: 'send_rcon',
 	sources: ['control'],
 	requestProperties: {
@@ -343,10 +339,9 @@ requests.sendRcon = new InstanceRequest({
 
 
 class Event {
-	constructor({ type, sources, targets, eventProperties = {} }) {
+	constructor({ type, links, eventProperties = {} }) {
 		this.type = type;
-		this.sources = sources;
-		this.targets = targets;
+		this.links = links;
 
 		this.eventType = type + '_event';
 		this._eventValidator = schema.compile({
@@ -364,7 +359,7 @@ class Event {
 
 	attach(link, handler) {
 		// Target side.  Note: Source and target is reversed here.
-		if (this.targets.includes(link.source) && this.sources.includes(link.target)) {
+		if (this.links.includes(`${link.target}-${link.source}`)) {
 			if (!handler) {
 				throw new Error(`Missing handler for ${this.eventType} on ${link.target}-${link.source} link`);
 			}
@@ -384,15 +379,13 @@ class Event {
 	 * Sends the given event data over the link.
 	 */
 	send(link, data = {}) {
-		link.send(this.eventType, data);
+		link.connector.send(this.eventType, data);
 	}
 }
 
-let events = {};
-events.updateInstances = new Event({
+messages.updateInstances = new Event({
 	type: 'update_instances',
-	sources: ['slave'],
-	targets: ['master'],
+	links: ['slave-master'],
 	eventProperties: {
 		"instances": {
 			type: "array",
@@ -409,10 +402,9 @@ events.updateInstances = new Event({
 	},
 });
 
-events.instanceOutput = new Event({
+messages.instanceOutput = new Event({
 	type: 'instance_output',
-	sources: ['slave', 'master'],
-	targets: ['master', 'control'],
+	links: ['slave-master', 'master-control'],
 	eventProperties: {
 		"instance_id": { type: "integer" },
 		"output": {
@@ -438,22 +430,18 @@ events.instanceOutput = new Event({
 /**
  * Attaches all requests and events to the given link
  *
- * Loops through all requests and events defined and attaches all of
- * them to the link.  For requests the handler used is looked up on the
- * link instance as the name of the request + 'RequestHandler'.  For
- * events the the handler used is looked up on the link instance as the
- * name of the event + 'EventHandler'.
+ * Loops through all builtin messages defined and attaches all of them
+ * to the link.  The handler used for the messageis looked up on the
+ * link instance as the concatenation of the name of the message, the
+ * name of message class, and 'Handler'.
  */
 function attachAllMessages(link) {
-	for (let [name, request] of Object.entries(requests)) {
-		request.attach(link, link[name + 'RequestHandler']);
-	}
-
-	for (let [name, event] of Object.entries(events)) {
-		event.attach(link, link[name + 'EventHandler']);
+	for (let [name, message] of Object.entries(messages)) {
+		message.attach(link, link[name + message.constructor.name + 'Handler']);
 	}
 }
 
+
 module.exports = {
 	Request,
 	InstanceRequest,
@@ -461,6 +449,5 @@ module.exports = {
 
 	attachAllMessages,
 
-	requests,
-	events,
+	messages,
 }

lib/manager/instanceManager.js

@@ -1,2 +0,0 @@
-const fs = require("fs-extra");
-const needle = require("needle");

master.js

@@ -24,6 +24,7 @@ const crypto = require('crypto');
 const base64url = require('base64url');
 const moment = require("moment");
 const request = require("request");
+const events = require("events");
 const version = require("./package").version;
 
 // constants
@@ -248,13 +249,13 @@ const ioMetrics = require("socket.io-prometheus");
 ioMetrics(io);
 
 
-class BaseConnection extends link.Connection {
-	constructor(target, socket) {
-		super(target, socket);
+class BaseConnection extends link.Link {
+	constructor(target, connector) {
+		super('master', target, connector);
 		link.attachAllMessages(this);
 	}
 
-	async forwardInstanceRequest(message, request) {
+	async forwardInstanceRequest(request, message) {
 		let instance = db.instances.get(message.data.instance_id);
 		if (!instance) {
 			throw new errors.RequestError(`Instance with ID ${instance_id} does not exist`);
@@ -271,13 +272,13 @@ class BaseConnection extends link.Connection {
 
 let controlConnections = new Array();
 class ControlConnection extends BaseConnection {
-	constructor(registerData, socket) {
-		super('control', socket)
+	constructor(registerData, connector) {
+		super('control', connector)
 
 		this._agent = registerData.agent;
 		this._version = registerData.version;
 
-		socket.on('disconnect', () => {
+		this.connector.on('disconnect', () => {
 			let index = controlConnections.indexOf(this);
 			if (index !== -1) {
 				controlConnections.splice(index, 1);
@@ -321,7 +322,7 @@ class ControlConnection extends BaseConnection {
 			throw new errors.RequestError("Slave is not connected");
 		}
 
-		await link.requests.createInstance.send(connection, {
+		await link.messages.createInstance.send(connection, {
 			id: Math.random() * 2**31 | 0, // TODO: add id option
 			options: {
 				'name': name,
@@ -344,8 +345,8 @@ class ControlConnection extends BaseConnection {
 
 var slaveConnections = new Map();
 class SlaveConnection extends BaseConnection {
-	constructor(registerData, socket) {
-		super('slave', socket);
+	constructor(registerData, connector) {
+		super('slave', connector);
 
 		this._agent = registerData.agent;
 		this._id = registerData.id;
@@ -359,37 +360,14 @@ class SlaveConnection extends BaseConnection {
 			version: this._version,
 		});
 
-		socket.on('message', () => {
+		this.connector.on('message', () => {
 			prometheusWsUsageCounter.labels('message', "other").inc();
 		});
 
-		socket.on('disconnect', () => {
+		this.connector.on('disconnect', () => {
 			if (slaveConnections.get(this._id) === this) {
 				slaveConnections.delete(this._id);
 			}
-
-			// XXX for backwards compatibility only
-			if (global.wsChatRecievers) {
-				global.wsChatRecievers.delete(this._id);
-			}
-		});
-
-		// XXX for backwards compatibility only
-		socket.on("gameChat", data => {
-			prometheusWsUsageCounter.labels('gameChat', this.instanceID).inc();
-			if(typeof data === "object"){
-				data.timestamp = Date.now();
-				if(!global.wsChatRecievers) global.wsChatRecievers = new Map();
-				for (let slave of global.wsChatRecievers.values()) {
-					slave.socket.emit("gameChat", data);
-				}
-			}
-		});
-
-		socket.on("registerChatReciever", data => {
-			prometheusWsUsageCounter.labels("registerChatReciever", "other").inc();
-			if(!global.wsChatRecievers) global.wsChatRecievers = new Map();
-			global.wsChatRecievers.set(this._id, this);
 		});
 	}
 
@@ -414,7 +392,7 @@ class SlaveConnection extends BaseConnection {
 		let { instance_id, output } = message.data;
 		for (let controlConnection of controlConnections) {
 			if (controlConnection.instanceOutputSubscriptions.has(instance_id)) {
-				link.events.instanceOutput.send(controlConnection, message.data);
+				link.messages.instanceOutput.send(controlConnection, message.data);
 			}
 		}
 	}
@@ -445,21 +423,53 @@ function isInteger(value) {
 }
 
 
+/**
+ * Connector for master server connections
+ */
+class SocketIOServerConnector extends events.EventEmitter {
+	constructor(socket) {
+		super();
+
+		this._seq = 1;
+		this._socket = socket;
+
+		this._socket.on('message', message => {
+			this.emit('message', message);
+		});
+		this._socket.on('disconnect', () => {
+			this.emit('disconnect');
+		});
+	}
+
+	/**
+	 * Send a message over the socket
+	 *
+	 * @returns the sequence number of the message sent
+	 */
+	send(type, data = {}) {
+		this._socket.send({ seq: this._seq, type, data });
+		return this._seq++;
+	}
+
+	disconnect() {
+		this._socket.disconnect(true);
+	}
+}
+
 io.on('connection', function (socket) {
 	console.log(`SOCKET | new connection from ${socket.handshake.address}`);
 
 	// Start connection handshake
-	socket.send({ seq: 1, type: 'hello', data: { version }});
+	let connector = new SocketIOServerConnector(socket);
+	connector.send('hello', { version });
 
 	// Handle socket handshake
 	socket.once('message', (payload) => {
 		prometheusWsUsageCounter.labels('message', "other").inc();
 		if (!schema.clientHandshake(payload)) {
 			console.log(`SOCKET | closing ${socket.handshake.address} after receiving invalid handshake`, payload);
-			socket.send({ type: 'close', data: {
-				reason: "Invalid handshake",
-			}});
-			socket.disconnect(true);
+			connector.send('close', { reason: "Invalid handshake" });
+			connector.disconnect(true);
 			return;
 		}
 
@@ -472,26 +482,20 @@ io.on('connection', function (socket) {
 			}
 
 			console.log(`SOCKET | registered slave ${data.id} version ${data.version}`);
-			slaveConnections.set(data.id, new SlaveConnection(data, socket));
+			slaveConnections.set(data.id, new SlaveConnection(data, connector));
+			connector.send('ready');
 
 		} else if (type === "register_control") {
 			console.log(`SOCKET | registered control from ${socket.handshake.address}`);
-			controlConnections.push(new ControlConnection(data, socket));
+			controlConnections.push(new ControlConnection(data, connector));
+			connector.send('ready');
 
 		} else if (type === "close") {
 			console.log(`SOCKET | received close from ${socket.handshake.address}: ${data.reason}`);
-			socket.disconnect(true);
+			connector.disconnect();
 			return;
 		}
 	});
-
-	// XXX TODO
-	/*
-	socket.on("registerAlertReciever", function(data){
-		prometheusWsUsageCounter.labels("registerAlertReciever", "other").inc();
-		if(!global.wsAlertRecievers) global.wsAlertRecievers = [];
-		global.wsAlertRecievers.push(socket);
-	});*/
 });
 
 // handle plugins on the master
@@ -660,6 +664,11 @@ module.exports = {
 	// For testing only
 	_db: db,
 	_config: config,
+	_SocketIOServerConnector: SocketIOServerConnector,
+	_controlConnections: controlConnections,
+	_ControlConnection: ControlConnection,
+	_slaveConnections: slaveConnections,
+	_SlaveConnection: SlaveConnection,
 }
 
 if (module === require.main) {

sharedPlugins/UPSdisplay/config.js

@@ -1,16 +1,11 @@
 /*
 	Example config file for example Clusterio plugin
-	Binary will be spawned using child_process.spawn and get input from stdin and text from stdout will be ran as
-	commands in the factorio server.
 */
 
 module.exports = {
 	// Name of package. For display somewhere I guess.
 	name: "UPSdisplay",
 	version: "2.0.0",
-	// Binary entrypoint for plugin. Don't let it crash. Stdout is sent to game as server chat (run /c commands from here for interface)
-	// Make sure its cross platform somehow.
-	binary: "nodePackage",
 	description: "Internal clusterio plugin to get the worlds UPS and display it on master",
 	// We'll send everything in this file to your stdin. Beware.
 	scriptOutputFileSubscription: "UPSdisplay.txt",

sharedPlugins/clusterioMod/config.js

@@ -5,7 +5,6 @@ module.exports = {
 	// Name of package. For display somewhere I guess.
 	name: "clusterioMod",
 	version: "1.0.0",
-	binary: "nodePackage",
 	description: "Server side handling of functionality in the clusterio lua mod",
 	masterPlugin: "masterPlugin.js",
 }

sharedPlugins/globalChat/config.js

@@ -5,6 +5,5 @@ module.exports = {
 	// Name of package. For display somewhere I guess.
 	name: "globalChat",
 	version: "1.0.0",
-	binary: "nodePackage",
 	description: "Clusterio plugin to allow for chat between instances.",
 }

sharedPlugins/inventoryImports/config.js

@@ -10,9 +10,6 @@ module.exports = {
 	// Name of package. For display somewhere I guess.
 	name: "inventoryImports",
 	version: "1.0.0",
-	// Binary entrypoint for plugin. Don't let it crash. Stdout is sent to game as server chat (run /c commands from here for interface)
-	// Make sure its cross platform somehow.
-	binary: "nodePackage",
 	arguments: ["index.js"],
 	description: "Clusterio plugin to fill character logistic slots with master imports",
 	// We'll send everything in this file to your stdin. Beware.

sharedPlugins/researchSync/config.js

@@ -4,9 +4,6 @@ module.exports = {
     // Name of package. For display somewhere I guess.
     name: "researchSync",
     version: "1.0.0",
-    // Binary entrypoint for plugin. Don't let it crash. Stdout is sent to game as server chat (run /c commands from here for interface)
-    // Make sure its cross platform somehow.
-    binary: "nodePackage",
     arguments: ["index.js"],
     description: "syncs research across servers",
     // We'll send everything in this file to your stdin. Beware.

sharedPlugins/statisticsExporter/config.js

@@ -4,7 +4,6 @@
 module.exports = {
 	name: "statisticsExporter",
 	version: "1.0.0",
-	binary: "nodePackage",
 	description: "Gathers ingame stats such as production, consumption, kills and buildings placed",
 	masterPlugin: "masterPlugin.js"
 }

test/clusterctl.js

@@ -0,0 +1,54 @@
+const assert = require('assert').strict;
+const chalk = require('chalk');
+
+const errors = require('lib/errors');
+const { testLines } = require('./lib/factorio/lines');
+const clusterctl = require('../clusterctl.js');
+const mock = require('./mock');
+
+
+describe("clusterctl", function() {
+	describe("formatOutputColored()", function() {
+		it("should pass the test lines", function() {
+			// Ensure tests get uncoloured output.
+			let old = chalk.level;
+			chalk.level = 0;
+
+			for (let [reference, output] of testLines) {
+				let line = clusterctl._formatOutputColored(output);
+				assert.deepEqual(line, reference);
+			}
+
+			chalk.level = old;
+		});
+	});
+
+	let mockConnector = new mock.MockConnector();
+	mockConnector.on('send', function(message) {
+		if (message.type === 'list_instances_request') {
+			this.emit('message', {
+				seq: 1, type: 'list_instances_response',
+				data: {
+					seq: message.seq,
+					list: [{ id: 57, slave_id: 4, name: 'Test Instance' }],
+				},
+			});
+		}
+	});
+	let testControl = new clusterctl._Control(mockConnector);
+
+	describe("resolveInstance", function() {
+		it("should pass an integer like string back", async function() {
+			assert.equal(await clusterctl._resolveInstance(null, "123"), 123);
+		});
+		it("should resolve an instance name with the master server", async function() {
+			assert.equal(await clusterctl._resolveInstance(testControl, "Test Instance"), 57);
+		});
+		it("should throw if instance is not found", async function() {
+			await assert.rejects(
+				clusterctl._resolveInstance(testControl, "invalid"),
+				new errors.CommandError("No instance named invalid")
+			);
+		});
+	});
+});

test/integration/clusterio.js

@@ -0,0 +1,129 @@
+const assert = require("assert").strict;
+const fs = require("fs-extra");
+const path = require("path");
+
+const link = require("lib/link");
+const master = require("../../master");
+const client = require("../../client");
+const clusterctl = require("../../clusterctl");
+const { version } = require("../../package");
+
+
+describe("Integration of Clusterio", function() {
+	let testControl;
+	let testControlConnection;
+
+	let testSlave;
+	let testSlaveConnection;
+
+	let instancesDir = path.join("test", "temp", "instances");
+
+	// Mark that this test takes a lot of time, or depeneds on a test
+	// that takes a lot of time.
+	function slowTest(test) {
+		if (process.env.FAST_TEST) {
+			test.skip();
+		}
+
+		test.timeout(20000);
+	}
+
+	before(async function() {
+		master._db.slaves = new Map();
+		master._db.instances = new Map([
+			[11, { id: 11, name: "foo", slaveId: 4 }],
+			[21, { id: 21, name: "bar", slaveId: 5 }],
+		]);
+
+		Object.assign(master._config, {
+			description: "test",
+			visibility: { public: false, lan: false },
+			username: "test",
+			token: "",
+			game_password: "",
+			verify_user_identity: false,
+			allow_commands: "admins-only",
+			auto_pause: false,
+		});
+
+		let [controlClient, controlServer] = link.VirtualConnector.makePair();
+		testControl = new clusterctl._Control(controlClient);
+		testControlConnection = new master._ControlConnection({ agent: "test", version }, controlServer);
+		master._controlConnections.push(testControlConnection);
+
+		await fs.remove(instancesDir);
+		await fs.ensureDir(instancesDir)
+		let [slaveClient, slaveServer] = link.VirtualConnector.makePair();
+		testSlave = new client._Slave(slaveClient, {
+			id: 4,
+			name: "slave",
+			instanceDirectory: instancesDir,
+			factorioDirectory: "factorio",
+			masterURL: "http://invalid",
+			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);
+	});
+
+	describe("clusterctl", function() {
+		describe("list-slaves", function() {
+			it("does not throw", async function() {
+				await clusterctl._commands.get("list-slaves").run({}, testControl);
+			});
+		});
+		describe("list-instances", function() {
+			it("does not throw", async function() {
+				await clusterctl._commands.get("list-instances").run({}, testControl);
+			});
+		});
+
+		describe("create-instances", function() {
+			it("creates the instance", async function() {
+				await clusterctl._commands.get("create-instance").run({name: "test", slave: "slave"}, testControl);
+				assert(await fs.exists(path.join(instancesDir, "test")), "Instance was not created");
+			});
+		});
+
+		describe("create-save", function() {
+			slowTest(this);
+			it("creates a save", async function() {
+				await clusterctl._commands.get("create-save").run({instance: "test"}, testControl);
+			});
+		});
+
+		describe("start-instance", function() {
+			slowTest(this);
+			it("starts the instance", async function() {
+				await clusterctl._commands.get("start-instance").run({instance: "test"}, testControl);
+				// TODO check that the instance actually stopped
+			});
+		});
+
+		describe("send-rcon", function() {
+			slowTest(this);
+			it("sends the command", async function() {
+				await clusterctl._commands.get("send-rcon").run({instance: "test", command: "test"}, testControl);
+				// TODO check that the command was received
+			});
+		});
+
+		describe("stop-instance", function() {
+			slowTest(this);
+			it("stops the instance", async function() {
+				await clusterctl._commands.get("stop-instance").run({instance: "test"}, testControl);
+				// TODO check that the instance actually stopped
+			});
+		});
+
+		describe("delete-instance", function() {
+			slowTest(this);
+			it("deletes the instance", async function() {
+				await clusterctl._commands.get("delete-instance").run({instance: "test"}, testControl);
+				assert(!await fs.exists(path.join(instancesDir, "test")), "Instance was not deleted");
+			});
+		});
+	});
+});

test/integration/server.js

@@ -45,7 +45,7 @@ describe("Integration of lib/factorio/server", function() {
 		describe(".exampleSettings()", function() {
 			let settings;
 			it("returns an object", async function() {
-				settings = await server.exampleSettings();
+				settings = await factorio.FactorioServer.exampleSettings(path.join("factorio", "data"));
 				assert(typeof settings === "object");
 			});
 

test/lib/factorio/lines.js

@@ -0,0 +1,116 @@
+// Lines for testing the factorio output
+
+let testLines = new Map([
+	[
+		"   1.306 Info RemoteCommandProcessor.cpp:131: Starting RCON interface at port 4000",
+		{
+			format: 'seconds',
+			time: '1.306',
+			type: 'log',
+			level: 'Info',
+			file: "RemoteCommandProcessor.cpp:131",
+			message: "Starting RCON interface at port 4000",
+		}
+	],
+	[
+		"   0.622 Warning FileUtil.cpp:527: test not found; using test.zip",
+		{
+			format: 'seconds',
+			time: '0.622',
+			type: 'log',
+			level: 'Warning',
+			file: "FileUtil.cpp:527",
+			message: "test not found; using test.zip",
+		}
+	],
+	[
+		"   0.554 Error Main.cpp:839: require_user_verification must be enabled for public games.",
+		{
+			format: 'seconds',
+			time: '0.554',
+			type: 'log',
+			level: 'Error',
+			file: "Main.cpp:839",
+			message: "require_user_verification must be enabled for public games.",
+		}
+	],
+	[
+		" 640.910 Quitting: remote-quit.",
+		{
+			format: 'seconds',
+			time: '640.910',
+			type: 'generic',
+			message: "Quitting: remote-quit.",
+		}
+	],
+	[
+		"1641.421 Goodbye",
+		{
+			format: 'seconds',
+			time: '1641.421',
+			type: 'generic',
+			message: "Goodbye",
+		}
+	],
+	[
+		"2019-01-20 10:30:04 [JOIN] User joined the game",
+		{
+			format: 'date',
+			time: '2019-01-20 10:30:04',
+			type: 'action',
+			action: "JOIN",
+			message: "User joined the game",
+		}
+	],
+	[
+		"2019-01-20 10:30:07 [CHAT] User: chat message",
+		{
+			format: 'date',
+			time: '2019-01-20 10:30:07',
+			type: 'action',
+			action: "CHAT",
+			message: "User: chat message",
+		}
+	],
+	[
+		"2019-01-20 10:30:14 [COMMAND] User (command): blah",
+		{
+			format: 'date',
+			time: '2019-01-20 10:30:14',
+			type: 'action',
+			action: "COMMAND",
+			message: "User (command): blah",
+		}
+	],
+	[
+		"2019-01-20 10:30:14 Cannot execute command. Error: [string \"blah\"]:1: syntax error near <eof>",
+		{
+			format: 'date',
+			time: '2019-01-20 10:30:14',
+			type: 'generic',
+			message: "Cannot execute command. Error: [string \"blah\"]:1: syntax error near <eof>",
+		}
+	],
+	[
+		"2019-01-20 10:30:21 [LEAVE] User left the game",
+		{
+			format: 'date',
+			time: '2019-01-20 10:30:21',
+			type: 'action',
+			action: "LEAVE",
+			message: "User left the game",
+		}
+	],
+	[
+		"Error while running event level::on_nth_tick(1)",
+		{
+			format: 'none',
+			type: 'generic',
+			message: "Error while running event level::on_nth_tick(1)",
+		}
+	],
+])
+
+module.exports = {
+	testLines,
+}

test/lib/factorio/server.js

@@ -3,6 +3,7 @@ const fs = require("fs-extra");
 const path = require("path");
 
 const factorio = require("lib/factorio");
+const { testLines } = require("./lines");
 
 
 describe("lib/factorio/server", function() {
@@ -92,95 +93,6 @@ describe("lib/factorio/server", function() {
 	});
 
 	describe("parseOutput()", function() {
-		let testLines = new Map([
-			[
-				"   1.306 Info RemoteCommandProcessor.cpp:131: Starting RCON interface at port 4000",
-				{
-					format: 'seconds',
-					time: '1.306',
-					type: 'log',
-					level: 'Info',
-					file: "RemoteCommandProcessor.cpp:131",
-					message: "Starting RCON interface at port 4000",
-				}
-			],
-			[
-				" 640.910 Quitting: remote-quit.",
-				{
-					format: 'seconds',
-					time: '640.910',
-					type: 'generic',
-					message: "Quitting: remote-quit.",
-				}
-			],
-			[
-				"1641.421 Goodbye",
-				{
-					format: 'seconds',
-					time: '1641.421',
-					type: 'generic',
-					message: "Goodbye",
-				}
-			],
-			[
-				"2019-01-20 10:30:04 [JOIN] User joined the game",
-				{
-					format: 'date',
-					time: '2019-01-20 10:30:04',
-					type: 'action',
-					action: "JOIN",
-					message: "User joined the game",
-				}
-			],
-			[
-				"2019-01-20 10:30:07 [CHAT] User: chat message",
-				{
-					format: 'date',
-					time: '2019-01-20 10:30:07',
-					type: 'action',
-					action: "CHAT",
-					message: "User: chat message",
-				}
-			],
-			[
-				"2019-01-20 10:30:14 [COMMAND] User (command): blah",
-				{
-					format: 'date',
-					time: '2019-01-20 10:30:14',
-					type: 'action',
-					action: "COMMAND",
-					message: "User (command): blah",
-				}
-			],
-			[
-				"2019-01-20 10:30:14 Cannot execute command. Error: [string \"blah\"]:1: syntax error near <eof>",
-				{
-					format: 'date',
-					time: '2019-01-20 10:30:14',
-					type: 'generic',
-					message: "Cannot execute command. Error: [string \"blah\"]:1: syntax error near <eof>",
-				}
-			],
-			[
-				"2019-01-20 10:30:21 [LEAVE] User left the game",
-				{
-					format: 'date',
-					time: '2019-01-20 10:30:21',
-					type: 'action',
-					action: "LEAVE",
-					message: "User left the game",
-				}
-			],
-			[
-				"Error while running event level::on_nth_tick(1)",
-				{
-					format: 'none',
-					type: 'generic',
-					message: "Error while running event level::on_nth_tick(1)",
-				}
-			],
-		])
-
 		it("should parse the test lines", function() {
 			for (let [line, reference] of testLines) {
 				reference.source = 'test';

test/lib/link/connector.js

@@ -0,0 +1,108 @@
+const assert = require('assert').strict;
+const events = require('events');
+
+const mock = require('../../mock');
+const link = require('lib/link');
+
+
+describe("lib/link/connectors", function() {
+	describe("class SocketIOClientConnector", function() {
+		let testConnector = new link.SocketIOClientConnector('source', 'url', 'token');
+		testConnector._socket = new mock.MockSocket();
+		describe(".disconnect()", function() {
+			it("should call close on the socket", function() {
+				testConnector._socket.closeCalled = false;
+				testConnector.disconnect();
+				assert(testConnector._socket.closeCalled, "Close was not called");
+			});
+		});
+
+		describe(".register()", function() {
+			it("is abstract", function() {
+				assert.throws(
+					() => testConnector.register(),
+					new Error("Abstract function")
+				);
+			});
+		});
+
+		describe(".send()", function() {
+			it("calls send on the socket", function() {
+				testConnector._socket.sentMessages = [];
+				let seq = testConnector.send('test', { test: true });
+				assert.deepEqual(testConnector._socket.sentMessages, [{ seq, type: 'test', data: { test: true }}]);
+			});
+		});
+
+		describe(".close()", function() {
+			it("should send close and call disconnect", function() {
+				testConnector._socket.sentMessages = [];
+				let called = false;
+				testConnector.disconnect = () => { called = true; }
+				testConnector.close("test reason");
+				assert.deepEqual(
+					testConnector._socket.sentMessages,
+					[{ seq: testConnector._seq - 1, type: 'close', data: { reason: "test reason" } }]
+				)
+				assert(called, ".disconnect() was not called");
+			});
+		});
+
+		describe(".processHandshake()", function() {
+			it("should close on invalid message", function() {
+				testConnector._socket.sentMessages = [];
+				testConnector._processHandshake({ data: "invalid message" }),
+				assert.deepEqual(
+					testConnector._socket.sentMessages,
+					[{seq: testConnector._seq - 1, type: 'close', data: {
+						reason: "Invalid handshake" }
+					}]
+				);
+			});
+			it("should call register on hello", function() {
+				let called = false;
+				testConnector.register = () => { called = true; };
+				testConnector._processHandshake({ seq: 1, type: 'hello', data: { version: "test" }}),
+				assert(called, "register was not called");
+			});
+			it("should emit ready on ready", async function() {
+				let result = events.once(testConnector, 'ready');
+				testConnector._processHandshake({ seq: 1, type: 'ready', data: {}}),
+				await result;
+			});
+			it("should emit error on close", async function() {
+				let result = events.once(testConnector, 'ready');
+				testConnector._processHandshake({ seq: 1, type: 'close', data: { reason: "test" }}),
+				await assert.rejects(result, new Error("server closed during handshake: test"));
+			});
+		});
+
+		describe("._attachSocketHandlers()", function() {
+			it("should attach handlers", function() {
+				testConnector._attachSocketHandlers();
+				assert(testConnector._socket.events.size > 0, "No handlers were attached");
+			});
+			it("should throw on message received in invalid state", function() {
+				testConnector._state = "new";
+				assert.throws(
+					() => testConnector._socket.events.get('message')(),
+					new Error("Received message in unexpected state new")
+				);
+			});
+			it("should call _processHandshake on message in handshake state", function() {
+				testConnector._state = "handshake";
+				let called = false;
+				testConnector._processHandshake = () => { called = true; };
+				testConnector._socket.events.get('message')();
+				assert(called, "_processHandshake was not called");
+			});
+			it("should emit message on message in ready state", async function() {
+				testConnector._socket.sentMessages = [];
+				testConnector._state = "ready";
+				let result = events.once(testConnector, 'message');
+				testConnector._socket.events.get('message')("message");
+				assert.deepEqual(await result, ["message"]);
+			});
+		});
+	});
+});

test/lib/link/link.js

@@ -4,35 +4,12 @@ const events = require('events');
 const link = require("lib/link");
 const errors = require("lib/errors");
 const schema = require("lib/schema");
+const mock = require("../../mock");
 
 
-class MockSocket {
-	constructor() {
-		this.sentMessages = [];
-		this.events = new Map();
-		this.handshake = { address: "socket.test" };
-	}
-
-	send(message) {
-		this.sentMessages.push(message);
-	}
-
-	on(event, fn) {
-		this.events.set(event, fn);
-	}
-
-	disconnect() {
-		this.disconnectCalled = true;
-	}
-
-	close() {
-		this.closeCalled = true;
-	}
-};
-
 describe("lib/link/link", function() {
 	describe("class Link", function() {
-		let testLink = new link.Link('source', 'target', new MockSocket);
+		let testLink = new link.Link('source', 'target', new mock.MockConnector());
 		testLink.setValidator('simple', schema.compile({
 			properties: {
 				"data": {
@@ -121,23 +98,6 @@ describe("lib/link/link", function() {
 			});
 		});
 
-		describe(".disconnect()", function() {
-			it("is abstract", function() {
-				assert.throws(
-					() => testLink.disconnect(),
-					new Error("Abstract function")
-				);
-			});
-		});
-
-		describe(".send()", function() {
-			it("calls send on the socket", function() {
-				let seq = testLink.send('test', { test: true });
-				assert.deepEqual(testLink.socket.sentMessages, [{ seq, type: 'test', data: { test: true }}]);
-				testLink.socket.sentMessages = [];
-			});
-		});
-
 		describe("._processWaiters()", function() {
 			it("should call waiters", async function() {
 				let waiter = new Promise(resolve => {
@@ -196,142 +156,5 @@ describe("lib/link/link", function() {
 				assert.deepEqual(await result2, { type: 'waiter', seq: 5, data: {} });
 			});
 		});
-
-		describe(".close()", function() {
-			it("should send close and call disconnect", function() {
-				let called = false;
-				testLink.disconnect = () => { called = true; }
-				testLink.close("test reason");
-				assert.deepEqual(
-					testLink.socket.sentMessages,
-					[{ seq: testLink._seq - 1, type: 'close', data: { reason: "test reason" } }]
-				)
-				assert(called, ".disconnect() was not called");
-			});
-		});
-	});
-
-	describe("class Connection", function() {
-		let testConnection = new link.Connection('target', new MockSocket());
-		it("should send ready after constructor", function() {
-			assert.deepEqual(
-				testConnection.socket.sentMessages,
-				[{seq: 2, type: 'ready', data: {}}]
-			);
-			testConnection.socket.sentMessages = [];
-		});
-		it("should disconnect when receiving a close message", function() {
-			testConnection.socket.disconnectCalled = false;
-			testConnection.socket.events.get('message')({
-				seq: 1, type: 'close', data: { reason: "test close" }
-			});
-			assert(testConnection.socket.disconnectCalled, "Disconnect was not called");
-		});
-		it("should close when receiving an invalid message", function() {
-			testConnection.socket.events.get('message')({ invalid: true });
-			assert.deepEqual(
-				testConnection.socket.sentMessages,
-				[{seq: testConnection._seq - 1, type: 'close', data: { reason: "Invalid message: Malformed message" }}]
-			);
-			testConnection.socket.sentMessages = [];
-		});
-
-		describe(".disconnect()", function() {
-			it("should call disconnect on the socket", function() {
-				testConnection.socket.disconnectCalled = false;
-				testConnection.disconnect();
-				assert(testConnection.socket.disconnectCalled, "Disconnect was not called");
-			});
-		});
-	});
-
-	describe("class Client", function() {
-		let testClient = new link.Client('source', 'url', 'token');
-		testClient.socket = new MockSocket();
-		describe(".disconnect()", function() {
-			it("should call close on the socket", function() {
-				testClient.socket.closeCalled = false;
-				testClient.disconnect();
-				assert(testClient.socket.closeCalled, "Close was not called");
-			});
-		});
-
-		describe(".register()", function() {
-			it("is abstract", function() {
-				assert.throws(
-					() => testClient.register(),
-					new Error("Abstract function")
-				);
-			});
-		});
-
-		describe(".processHandshake()", function() {
-			it("should close on invalid message", function() {
-				testClient._processHandshake({ data: "invalid message" }),
-				assert.deepEqual(
-					testClient.socket.sentMessages,
-					[{seq: testClient._seq - 1, type: 'close', data: {
-						reason: "Invalid handshake" }
-					}]
-				);
-				testClient.socket.sentMessages = [];
-			});
-			it("should call register on hello", function() {
-				let called = false;
-				testClient.register = () => { called = true; };
-				testClient._processHandshake({ seq: 1, type: 'hello', data: { version: "test" }}),
-				assert(called, "register was not called");
-			});
-			it("should emit ready on ready", async function() {
-				let result = events.once(testClient._events, 'ready');
-				testClient._processHandshake({ seq: 1, type: 'ready', data: {}}),
-				await result;
-			});
-			it("should emit error on close", async function() {
-				let result = events.once(testClient._events, 'ready');
-				testClient._processHandshake({ seq: 1, type: 'close', data: { reason: "test" }}),
-				await assert.rejects(result, new Error("server closed during handshake: test"));
-			});
-		});
-
-		describe("._attachSocketHandlers()", function() {
-			it("should attach handlers", function() {
-				testClient._attachSocketHandlers();
-				assert(testClient.socket.events.size > 0, "No handlers were attached");
-			});
-			it("should throw on message received in invalid state", function() {
-				testClient._state = "new";
-				assert.throws(
-					() => testClient.socket.events.get('message')(),
-					new Error("Received message in unexpected state new")
-				);
-			});
-			it("should call _processHandshake on message in handshake state", function() {
-				testClient._state = "handshake";
-				let called = false;
-				testClient._processHandshake = () => { called = true; };
-				testClient.socket.events.get('message')();
-				assert(called, "_processHandshake was not called");
-			});
-			it("should call processMessage on message in ready state", function() {
-				testClient._state = "ready";
-				let called = false;
-				testClient.processMessage = () => { called = true; };
-				testClient.socket.events.get('message')();
-				assert(called, "processMessage was not called");
-				delete testClient.processMessage;
-			});
-			it("should close on invalid message in ready state", function() {
-				testClient._state = "ready";
-				testClient.socket.events.get('message')({ data: "invalid message"});
-				assert.deepEqual(
-					testClient.socket.sentMessages,
-					[{seq: testClient._seq - 1, type: 'close', data: {
-						reason: "Invalid message: Malformed message" }
-					}]
-				);
-				testClient.socket.sentMessages = [];
-			});
-		});
 	});
 });

test/lib/link/messages.js

@@ -2,11 +2,12 @@ const assert = require('assert').strict;
 
 const link = require('lib/link');
 const errors = require('lib/errors');
+const mock = require('../../mock');
 
 
 describe("lib/link/messages", function() {
-	let testSourceLink = new link.Link('source', 'target', null);
-	let testTargetLink = new link.Link('target', 'source', null);
+	let testSourceLink = new link.Link('source', 'target', new mock.MockConnector());
+	let testTargetLink = new link.Link('target', 'source', new mock.MockConnector());
 
 	let lastTargetSent;
 	testTargetLink.send = (type, data) => { lastTargetSent = { type, data }; };
@@ -14,8 +15,7 @@ describe("lib/link/messages", function() {
 	describe("class Request", function() {
 		let testRequest = new link.Request({
 			type: 'test',
-			sources: 'source',
-			targets: 'target',
+			links: ['source-target'],
 		});
 
 		describe(".attach()", function() {
@@ -39,7 +39,7 @@ describe("lib/link/messages", function() {
 				handlerResult = { seq: 3 };
 				testTargetLink._handlers.get('test_request')({ seq: 2 });
 				let result = await new Promise(resolve => {
-					testTargetLink.send = (type, data) => resolve({ type, data });
+					testTargetLink.connector.send = (type, data) => resolve({ type, data });
 				});
 				assert.deepEqual(result, {
 					type: 'test_response',
@@ -53,7 +53,7 @@ describe("lib/link/messages", function() {
 				handlerResult = { test: "handler" };
 				testTargetLink._handlers.get('test_request')({ seq: 2 });
 				let result = await new Promise(resolve => {
-					testTargetLink.send = (type, data) => resolve({ type, data });
+					testTargetLink.connector.send = (type, data) => resolve({ type, data });
 				});
 				assert.deepEqual(result, {
 					type: 'test_response',
@@ -67,7 +67,7 @@ describe("lib/link/messages", function() {
 				handlerResult = undefined;
 				testTargetLink._handlers.get('test_request')({ seq: 2 });
 				let result = await new Promise(resolve => {
-					testTargetLink.send = (type, data) => resolve({ type, data });
+					testTargetLink.connector.send = (type, data) => resolve({ type, data });
 				});
 				assert.deepEqual(result, {
 					type: 'test_response',
@@ -79,11 +79,11 @@ describe("lib/link/messages", function() {
 		});
 
 		describe(".send()", function() {
-			let request;
-			testSourceLink.send = (type, data) => {
-				request = { type, data };
-			};
 			it("should send request with send and use waitFor to get response", async function() {
+				let request;
+				testSourceLink.connector.send = (type, data) => {
+					request = { type, data };
+				};
 				testSourceLink.waitFor = (type, condition) => {
 					return { data: { type, request }};
 				};
@@ -91,6 +91,8 @@ describe("lib/link/messages", function() {
 					await testRequest.send(testSourceLink, { test: "request" }),
 					{ type: 'test_response', request: { type: 'test_request', data: {test: "request" }}}
 				);
+				delete testSourceLink.connector.send;
+				delete testSourceLink.waitFor;
 			});
 			it("should throw error response", async function() {
 				testSourceLink.waitFor = (type, condition) => {
@@ -100,6 +102,7 @@ describe("lib/link/messages", function() {
 					testRequest.send(testSourceLink),
 					new errors.RequestError("test error")
 				);
+				delete testSourceLink.waitFor;
 			});
 		});
 	});
@@ -110,8 +113,7 @@ describe("lib/link/messages", function() {
 	describe("class Event", function() {
 		let testEvent = new link.Event({
 			type: 'test',
-			sources: 'source',
-			targets: 'target',
+			links: ['source-target'],
 		});
 
 		describe(".attach()", function() {
@@ -134,11 +136,13 @@ describe("lib/link/messages", function() {
 
 		describe(".send()", function() {
 			it("should send the event over the link", async function() {
-				let sentData = [];
-				testSourceLink.send = (type, data) => { sentData.push({ type, data }); };
-
+				testSourceLink.connector.sentMessages = [];
 				testEvent.send(testSourceLink, { test: "event" });
-				assert.deepEqual(sentData, [{ type: "test_event", data: { test: "event" }}]);
+				let seq = testSourceLink.connector._seq - 1;
+				assert.deepEqual(
+					testSourceLink.connector.sentMessages,
+					[{ seq, type: "test_event", data: { test: "event" }}]
+				);
 			});
 		});
 	});

test/master.js

@@ -3,6 +3,7 @@ const request = require('supertest');
 const validateHTML = require('html5-validator');
 const parallel = require('mocha.parallel');
 
+const mock = require('./mock');
 const master = require('../master');
 
 
@@ -43,4 +44,15 @@ describe('Master testing', function() {
 			it(`sends some HTML when accessing ${path}`, () => getValidate(path));
 		}
 	});
+
+	describe("class SocketIOServerConnector", function() {
+		let testConnector = new master._SocketIOServerConnector(new mock.MockSocket());
+		describe(".disconnect()", function() {
+			it("should call disconnect on the socket", function() {
+				testConnector._socket.disconnectCalled = false;
+				testConnector.disconnect();
+				assert(testConnector._socket.disconnectCalled, "Disconnect was not called");
+			});
+		});
+	});
 });

test/missing.js

@@ -1,7 +1,6 @@
 // These are libraries that does not have any units test, they are
 // required here in order to track coverage for them.
 
-require('lib/manager/instanceManager');
 require('lib/manager/modManager');
 require('lib/manager/pluginManager');
 require('lib/npmPostinstall.js');

test/mock.js

@@ -0,0 +1,58 @@
+const events = require("events");
+
+
+class MockSocket {
+	constructor() {
+		this.sentMessages = [];
+		this.events = new Map();
+		this.handshake = { address: "socket.test" };
+	}
+
+	send(message) {
+		this.sentMessages.push(message);
+	}
+
+	on(event, fn) {
+		this.events.set(event, fn);
+	}
+
+	disconnect() {
+		this.disconnectCalled = true;
+	}
+
+	close() {
+		this.closeCalled = true;
+	}
+};
+
+class MockConnector extends events.EventEmitter {
+	constructor() {
+		super();
+
+		this._seq = 1;
+		this.sentMessages = [];
+		this.events = new Map();
+		this.handshake = { address: "socket.test" };
+	}
+
+	send(type, data) {
+		let message = { seq: this._seq, type, data };
+		this.sentMessages.push(message);
+		setImmediate(() => this.emit('send', message));
+		return this._seq++;
+	}
+
+	close(reason) {
+		this.send('close', { reason });
+		this.disconnect();
+	}
+
+	disconnect() {
+		this.disconnectCalled = true;
+	}
+};
+
+module.exports = {
+	MockSocket,
+	MockConnector,
+};