Merge pull request #294 from Hornwitser/plugin-track

Integrate plugins closer with connections

clusterctl.js

@@ -1,3 +1,7 @@
+/**
+ * Command line interface for controlling a Clusterio cluster
+ * @module
+ */
 const jwt = require("jsonwebtoken");
 const fs = require("fs-extra");
 const yargs = require("yargs");
@@ -16,6 +20,8 @@ const config = require("lib/config");
  *
  * Formats a parsed Factorio output from lib/factorio into a readable
  * colorized output using terminal escape codes that can be printed.
+ *
+ * @private
  */
 function formatOutputColored(output) {
 	let time = "";
@@ -73,6 +79,7 @@ class Command {
  * @param client - link to master server to query instance on.
  * @param instanceName - string with name or id of instance.
  * @returns {number} instance ID.
+ * @private
  */
 async function resolveInstance(client, instanceName) {
 	let instanceId;
@@ -104,6 +111,7 @@ async function resolveInstance(client, instanceName) {
  * @param client - link to master server to query slave on.
  * @param slaveName - string with name or id of slave.
  * @returns {number} slave ID.
+ * @private
  */
 async function resolveSlave(client, slaveName) {
 	let slaveId;
@@ -324,6 +332,7 @@ commands = new Map([...commands.map(command => [command.name, command])]);
 
 /**
  * Connector for control connection to master server
+ * @private
  */
 class ControlConnector extends link.WebSocketClientConnector {
 	constructor(url, reconnectDelay, token) {

docs/jsdoc.json

@@ -18,6 +18,7 @@
             "plugins/",
             "routes/",
             "routes.js"
-        ]
+        ],
+        "excludePattern": "node_modules"
     }
 }

docs/socket.md

@@ -114,6 +114,9 @@ Handshake messages
 Sent by the master server when the WebSocket connection has been opened.
 
 - version - string - The version of the master, e.g. "2.0.0".
+- plugins - Object<string, string> -
+    Object mapping plugin names to plugin version for plugins that are
+    loaded on the master server.
 
 ### `register_slave`
 
@@ -124,6 +127,9 @@ Slave handshake for establishing a new connection session.
 - version - string - The protocol version of the slave, e.g. "2.0.0".
 - name - string - Name of the slave.
 - id - integer - ID of the slave.
+- plugins - Object<string, string> -
+    Object mapping plugin names to plugin version for plugins that are
+    available on the slave.
 
 ### `register_control`
 

docs/writing-plugins.md

@@ -17,6 +17,7 @@ Contents
     - [Defining Events](#defining-events)
     - [Defining Requests](#defining-requests)
 - [Sending Link Messages](#sending-link-messages)
+    - [Handling connection events](#handling-connection-events)
 - [Collecting Statistics](#collecting-statistics)
 
 
@@ -73,9 +74,9 @@ The following properties are recognized:
 **instanceEntrypoint**:
     Path to a Node.js module relative to the plugin directory which
     contains the InstancePlugin class definition for this plugin.  This
-    is an optional paramater.  A plugin can be for instances only, but
-    for it to be able to send any messages to other instances it will
-    still have to be loaded on the master server.
+    is an optional paramater.  A plugin may have code only for instances
+    but it must still be loaded on the master in order for it to be
+    possible to load it on an instance.
 
 **InstanceConfigGroup**:
     Subclass of `PluginConfigGroup` for defining the per instance
@@ -310,10 +311,10 @@ It will also be forwarded by slaves to a specific instance.
 The following properties are recognized by the Event constructor:
 
 **type**:
-    The message type sent over the wire.  This can be any string, but
-    it must be unique across all plugins, and it's recommended that it is
-    of format `plugin_name:message_name`.  The suffix `_event` will be
-    appended to the type.
+    The message type sent over the wire.  This must start with the name
+    of the plugin followed by colon and and be unique for the plugin.
+    The type of the message sent over the socket will have the suffix
+    `_event` appended to it.
 
 **links**:
     An array of strings describing which links this event can be sent
@@ -390,11 +391,11 @@ instance specified by `instance_id`.
 The following properties are recognized by the Request constructor:
 
 **type**:
-    The message type sent over the wire.  This can be any string, but
-    it must be unique across all plugins, and it's recommended that it is
-    of format `plugin_name:message_name`.  The suffix `_request` will be
-    appended to the type for the request message sent, and the suffix
-    `_response` will be appended to the type for the response.
+    The message type sent over the wire.  This must start with the name
+    of the plugin followed by colon and and be unique for the plugin.
+    The type of the message sent over the socket will have the suffix
+    `_request` appended to it for the request and `_response` appended
+    to it for the response.
 
 **links**:
     An array of strings describing which links this request can be sent
@@ -444,6 +445,51 @@ data received from the target it was sent to, or throws an error if the
 request failed.
 
 
+### Handling connection events
+
+There are a few connection related events that plugins neeed to repsond
+to in order to avoid data loss and connection problems.  The most
+important is the prepare disconnect for the link between master and
+slave.  This is signaled to `MasterPlugin` classes via the
+`onPrepareSlaveDisconnect` hook and to `InstancePlugin` classes via the
+`onPrepareMasterDisconnect` hook.
+
+After the prepare disconnect the connection will be closed, which will
+result in pending requests and events being dropped.  Plugins must
+respond to the prepare disconnect by stopping any processess it does
+that send events or requests over the link in question.  This can be
+accomplished either through listening for the prepare disconnect hook,
+or by checking the `connected` or `closing` property of the connector
+for the master/slave connection.  For example the sending of an event
+from an `InstancePlugin` class can be stopped while the connection is
+closing by using the following code:
+
+    if (!this.slave.connector.closing) {
+        this.info.messages.frobnicate.send(this.instance, { foo: "bar" });
+    }
+
+If the event or request needs to be sent to the master it can be put
+into a queue stored on the plugin instance and sent out when the
+connection is established again.  The re-establishement of the
+connection is  notified to plugins via the `connect` event to the
+`onMasterConnectionEvent` and `onSlaveConnectionEvent` hooks.
+
+The second connection event which is of lesser importance to respond to
+is the `drop` connection event served through `onMasterConnectionEvent`
+for `InstancePlugin` classes and through `onSlaveConnectionEvent` for
+`MasterPlugin` classes.  This is raised when the connection between the
+master and slave in question is lost, most likely due to networking
+issues.  When in the dropped state the slave will keep trying to
+reconnect to the master server in order to re-establish it, and if
+successful no events or requests will be lost.  However while in the
+dropped state any requests and events sent gets queued up in memory
+until the connection is re-established.  This means that if your plugin
+sends a lot of events or requests, they can end up being queued up in a
+buffer for a long time and sent out all at once.  To avoid this you
+should be throtteling and/or stopping your requests/events after `drop`
+has been raised, and continue back as normal when `connect` is raised.
+
+
 Collecting Statistics
 ---------------------
 

lib/config/classes.js

@@ -1,7 +1,4 @@
-/**
- * @file Configuration classes
- * @author Hornwitser
- */
+// Configuration classes
 "use strict";
 
 const { basicType } = require("lib/helpers");
@@ -12,6 +9,9 @@ const { basicType } = require("lib/helpers");
  *
  * Exception class for when invalid values are attempted to be set on a
  * field.
+ *
+ * @extends Error
+ * @memberof module:lib/config
  */
 class InvalidValue extends Error { };
 
@@ -20,18 +20,26 @@ class InvalidValue extends Error { };
  *
  * Exception class for when a group or field that does not exist is
  * attempted to be accessed.
+ *
+ * @extends Error
+ * @memberof module:lib/config
  */
 class InvalidField extends Error { };
 
 /**
  * Split the given string on the first instance of separator
  *
- * Splits @p string on the first instance of @p separator and returns an
+ * Splits `string` on the first instance of `separator` and returns an
  * array consisting of the string up to the separator and the string
  * after the separator.  Returns an array with the string and an empty
  * string if the separator is not present.
  *
+ * @param {string} separator - Separator to split string by.
+ * @param {string} string - String to split
  * @returns {Array} string split on separator.
+ * @memberof module:lib/config
+ * @private
+ * @inner
  */
 function splitOn(separator, string) {
 	let index = string.indexOf(separator);
@@ -46,6 +54,8 @@ function splitOn(separator, string) {
  *
  * Represents a collection of ConfigGroup instances that can be
  * managed as one.
+ *
+ * @memberof module:lib/config
  */
 class Config {
 	constructor() {
@@ -215,6 +225,8 @@ class Config {
 
 /**
  * Collection of related config entries
+ *
+ * @memberof module:lib/config
  */
 class ConfigGroup {
 	/**
@@ -285,7 +297,7 @@ class ConfigGroup {
 	 * The field must be defined for the config group and the value is
 	 * type checked against the field definition.
 	 *
-	 * @throws {InvalidValue} if value is not allowed for the field.
+	 * @throws {module:lib/config.InvalidValue} if value is not allowed for the field.
 	 */
 	set(name, value) {
 		let def = this.constructor._definitions.get(name);
@@ -553,6 +565,9 @@ class ConfigGroup {
  * Grouping that includes per plugin fields clusterio uses to manage
  * plugins.  This is currently only the enabled field.  It is otherwise
  * identical to the ConfigGroup class.
+ *
+ * @extends module:lib/config.ConfigGroup
+ * @memberof module:lib/config
  */
 class PluginConfigGroup extends ConfigGroup {
 	static _initSubclass() {

lib/config/definitions.js

@@ -1,7 +1,4 @@
-/**
- * @file Core definitions for the configuration system
- * @author Hornwitser
- */
+// Core definitions for the configuration system
 "use strict";
 
 const util = require("util");
@@ -12,6 +9,11 @@ const fs = require("fs-extra");
 const classes = require("./classes");
 
 
+/**
+ * Master config group for {@link module:lib/config.MasterConfig}
+ * @extends module:lib/config.ConfigGroup
+ * @memberof module:lib/config
+ */
 class MasterGroup extends classes.ConfigGroup { }
 MasterGroup.groupName = "master"
 MasterGroup.define({
@@ -103,10 +105,20 @@ MasterGroup.define({
 });
 MasterGroup.finalize();
 
+/**
+ * Master Config class
+ * @extends module:lib/config.Config
+ * @memberof module:lib/config
+ */
 class MasterConfig extends classes.Config { }
 MasterConfig.registerGroup(MasterGroup);
 
 
+/**
+ * Slave config group for {@link module:lib/config.SlaveConfig}
+ * @extends module:lib/config.ConfigGroup
+ * @memberof module:lib/config
+ */
 class SlaveGroup extends classes.ConfigGroup {}
 SlaveGroup.groupName = "slave";
 SlaveGroup.define({
@@ -160,10 +172,20 @@ SlaveGroup.define({
 });
 SlaveGroup.finalize();
 
+/**
+ * Slave Config class
+ * @extends module:lib/config.Config
+ * @memberof module:lib/config
+ */
 class SlaveConfig extends classes.Config { }
 SlaveConfig.registerGroup(SlaveGroup);
 
 
+/**
+ * Instance config group for {@link module:lib/config.InstanceConfig}
+ * @extends module:lib/config.ConfigGroup
+ * @memberof module:lib/config
+ */
 class InstanceGroup extends classes.ConfigGroup { }
 InstanceGroup.groupName = "instance"
 InstanceGroup.define({
@@ -184,6 +206,11 @@ InstanceGroup.define({
 });
 InstanceGroup.finalize();
 
+/**
+ * Factorio config group for {@link module:lib/config.InstanceConfig}
+ * @extends module:lib/config.ConfigGroup
+ * @memberof module:lib/config
+ */
 class FactorioGroup extends classes.ConfigGroup { }
 FactorioGroup.groupName = "factorio";
 FactorioGroup.define({
@@ -221,6 +248,11 @@ FactorioGroup.define({
 });
 FactorioGroup.finalize();
 
+/**
+ * Instance config class
+ * @extends module:lib/config.Config
+ * @memberof module:lib/config
+ */
 class InstanceConfig extends classes.Config { }
 InstanceConfig.registerGroup(InstanceGroup);
 InstanceConfig.registerGroup(FactorioGroup);
@@ -242,6 +274,8 @@ function validateGroup(pluginInfo, groupName) {
 
 /**
  * Registers the config groups for the provided plugin infos
+ *
+ * @memberof module:lib/config
  */
 function registerPluginConfigGroups(pluginInfos) {
 	for (let pluginInfo of pluginInfos) {
@@ -273,6 +307,8 @@ function registerPluginConfigGroups(pluginInfos) {
 
 /**
  * Lock configs from adding more groups and make them usable
+ *
+ * @memberof module:lib/config
  */
 function finalizeConfigs() {
 	MasterConfig.finalize();
@@ -285,6 +321,8 @@ function finalizeConfigs() {
  *
  * Can be passed to yargs.command to implement a config command.  Use
  * handleConfigCommand to do the requested action.
+ *
+ * @memberof module:lib/config
  */
 function configCommand(yargs) {
 	yargs
@@ -300,6 +338,8 @@ function configCommand(yargs) {
  * Handle yargs command
  *
  * Handle the actions that are made available by configCommand.
+ *
+ * @memberof module:lib/config
  */
 async function handleConfigCommand(args, instance, configPath) {
 	let command = args._[1];

lib/config/index.js

@@ -1,8 +1,8 @@
 /**
- * @file Configuration System
+ * Configuration System
  * @author Hornwitser
+ * @module lib/config
  */
-
 module.exports = {
 	...require("./classes"),
 	...require("./definitions"),

lib/errors.js

@@ -3,10 +3,22 @@
  * @module
  */
 
+/**
+ * Thrown by commands when they fail
+ * @static
+ */
 class CommandError extends Error { }
+
+/**
+ * Thrown from requests sent when an error occured handling it
+ * @static
+ */
 class RequestError extends Error { }
 
-// Signal for messages that fail validation
+/**
+ * Signal for messages that fail validation
+ * @static
+ */
 class InvalidMessage extends Error {
 	constructor(msg, errors) {
 		super(msg);
@@ -14,19 +26,34 @@ class InvalidMessage extends Error {
 	}
 }
 
-// Thrown from requests when the session it was sent on was lost.
+/**
+ * Thrown from requests when the session it was sent on was lost
+ * @static
+ */
 class SessionLost extends Error { }
 
-// Thrown when WebSocket authentication failed
+/**
+ * Thrown when WebSocket authentication failed
+ * @static
+ */
 class AuthenticationFailed extends Error { }
 
-// Errror class for known errors occuring during startup
+/**
+ * Errror class for known errors occuring during startup
+ * @static
+ */
 class StartupError extends Error { }
 
-// Errors outside of our control.
+/**
+ * Errors outside of our control
+ * @static
+ */
 class EnvironmentError extends Error { }
 
-// Errors caused by plugins
+/**
+ * Errors caused by plugins
+ * @static
+ */
 class PluginError extends Error {
 	constructor(pluginName, original) {
 		super(`PluginError: ${original.message}`);

lib/factorio/index.js

@@ -1,3 +1,8 @@
+/**
+ * Library for interfacing with Factorio Servers and saves.
+ * @module lib/factorio
+ * @author Hornwitser
+ */
 module.exports = {
 	...require('lib/factorio/locale'),
 	...require('lib/factorio/patch'),

lib/factorio/locale.js

@@ -1,7 +1,4 @@
-/**
- * Function for parsing factorio locale from a local factorio install
- */
-
+// Function for parsing factorio locale from a local factorio install
 "use strict";
 const fs = require("fs-extra");
 const path = require("path");
@@ -13,6 +10,7 @@ const ini = require("ini");
  * @param {string} factorioDirectory
  * @param {string} languageCode
  * @param {function} callback
+ * @memberof module:lib/factorio
  */
 async function getLocale(dataDir, languageCode) {
 	if (typeof dataDir !== "string") throw new TypeError("dataDir must be a string");

lib/factorio/patch.js

@@ -1,7 +1,4 @@
-/**
- * Library for patching Factorio saves with scenario code.
- * @author Hornwitser
- */
+// Library for patching Factorio saves with scenario code.
 
 "use strict";
 const events = require("events");
@@ -31,6 +28,9 @@ const knownScenarios = {
  * contained in.  Throws an error if there are multiple such folders.
  *
  * @returns {string} name of the root folder.
+ * @memberof module:lib/factorio
+ * @private
+ * @inner
  */
 function findRoot(zip) {
 	let root = null;
@@ -60,6 +60,9 @@ function findRoot(zip) {
  * Generates control.lua code for loading the Clusterio modules
  *
  * @param {Object} patchInfo - The patch info files's json content
+ * @memberof module:lib/factorio
+ * @private
+ * @inner
  */
 function generateLoader(patchInfo) {
 	let lines = [
@@ -107,6 +110,9 @@ function generateLoader(patchInfo) {
  * their dependents.  Throws an error if this is not possible.
  *
  * @param {Array<Object>} modules - Array of modules to reorder
+ * @memberof module:lib/factorio
+ * @private
+ * @inner
  */
 function reorderDependencies(modules) {
 	let index = 0;
@@ -202,6 +208,7 @@ function reorderDependencies(modules) {
  *
  * @param {string} savePath - Path to the Factorio save to patch.
  * @param {Array<Object>} modules - Description of the modules to patch.
+ * @memberof module:lib/factorio
  */
 async function patch(savePath, modules) {
 	let zip = await jszip.loadAsync(await fs.readFile(savePath));

lib/factorio/server.js

@@ -1,9 +1,4 @@
-/**
- * Factorio interfaces
- *
- * @module
- */
-
+// Factorio Server interface
 "use strict";
 const fs = require("fs-extra");
 const child_process = require("child_process");
@@ -20,6 +15,10 @@ const errors = require("lib/errors");
 /**
  * Determines the version of Factorio the datadir is pointing to by
  * reading the changelog.txt in it.
+ *
+ * @memberof module:lib/factorio
+ * @private
+ * @inner
  */
 async function getVersion(changelogPath) {
 	let changelog;
@@ -47,6 +46,10 @@ async function getVersion(changelogPath) {
 
 /**
  * Comparison function for sorting Factorio version strings
+ *
+ * @memberof module:lib/factorio
+ * @private
+ * @inner
  */
 function versionOrder(a, b) {
 	let aParts = a.split(".").map(s => Number.parseInt(s, 10));
@@ -70,6 +73,10 @@ function versionOrder(a, b) {
  *
  * Searches the given factorio dir for an installation of Factorio of
  * the target version.
+ *
+ * @memberof module:lib/factorio
+ * @private
+ * @inner
  */
 async function findVersion(factorioDir, targetVersion) {
 
@@ -125,6 +132,9 @@ async function findVersion(factorioDir, targetVersion) {
  * RFC 6335.
  *
  * @return {number} a number in the range 49152 to 65535.
+ * @memberof module:lib/factorio
+ * @private
+ * @inner
  */
 function randomDynamicPort() {
 	const start = 49152;
@@ -141,6 +151,9 @@ function randomDynamicPort() {
  *
  * @param {number} length - the length of the password to generate.
  * @return {string} password of the given length
+ * @memberof module:lib/factorio
+ * @private
+ * @inner
  */
 async function generatePassword(length) {
 	function validChar(byte) {
@@ -179,6 +192,10 @@ async function generatePassword(length) {
  * Data is streamed via the data method, and once the stream ends the
  * end method should be called.  This ensures an unterminated line at
  * the end of the stream is passed to the callback.
+ *
+ * @memberof module:lib/factorio
+ * @private
+ * @inner
  */
 class LineSplitter {
 
@@ -253,6 +270,9 @@ class LineSplitter {
  * @param {string} source - Passed into the output structure as source
  *
  * @returns {object} - An object with interpeted data.
+ * @memberof module:lib/factorio
+ * @private
+ * @inner
  */
 function parseOutput(line, source) {
 	let output = {
@@ -425,15 +445,17 @@ const outputHeuristics = [
  * stopping and sending commands to the server.  It does not deal with
  * creating or managing servers, or downloading Factorio.
  *
- * This is an events emitter with the following events:
- * stdout - invoked when Factorio outputs a line to stdout
- * stderr - invoked when Factorio outputs a line to stderr
- * output - invoked with parsed output from Factorio
- * rcon-ready - invoked when the RCON client has connected
- * game-ready - invoked when the server is finished starting up
- * autosave-start - invoked when the server starts an autosave
- * autosave-fnished - invoked when the autosave finished
- * exit - invoked when the sterver has exited
+ * This is an EventEmitter with the following events:
+ * - stdout - invoked when Factorio outputs a line to stdout
+ * - stderr - invoked when Factorio outputs a line to stderr
+ * - output - invoked with parsed output from Factorio
+ * - rcon-ready - invoked when the RCON client has connected
+ * - game-ready - invoked when the server is finished starting up
+ * - autosave-start - invoked when the server starts an autosave
+ * - autosave-fnished - invoked when the autosave finished
+ * - exit - invoked when the sterver has exited
+ * @extends events.EventEmitter
+ * @memberof module:lib/factorio
  */
 class FactorioServer extends events.EventEmitter {
 	/**
@@ -442,6 +464,12 @@ class FactorioServer extends events.EventEmitter {
 	 * @param {string} factorioDir - Directory of the Factorio install(s).
 	 * @param {string} writeDir - Directory to write runtime data to.
 	 * @param {object} options - Optional parameters.
+	 * @param {string} options.version -
+	 *     Version of Factorio to use.  Can also be the string "latest" to
+	 *     use the latest version found in `factorioDir`.
+	 * @param {number} options.gamePort - UDP port to host game on.
+	 * @param {number} options.rconPort - TCP port to use for RCON.
+	 * @param {string} options.rconPassword - Password use for RCON.
 	 */
 	constructor(factorioDir, writeDir, options) {
 		super();
@@ -454,8 +482,11 @@ class FactorioServer extends events.EventEmitter {
 		this._dataDir = null;
 
 		this._targetVersion = options.version || "latest";
+		/** UDP port used for hosting the Factorio game server on */
 		this.gamePort = options.gamePort || randomDynamicPort();
+		/** TCP port used for RCON on the Factorio game server */
 		this.rconPort = options.rconPort || randomDynamicPort();
+		/** Password used for RCON on the Factorio game server */
 		this.rconPassword = options.rconPassword;
 		this._state = "new";
 		this._server = null;
@@ -603,6 +634,7 @@ class FactorioServer extends events.EventEmitter {
 	 * Initialize class instance
 	 *
 	 * Must be called before instances of this class can be used.
+	 * @throws {Error} if the requested version of Factorio was not found.
 	 */
 	async init() {
 		this._check(["new"]);
@@ -612,7 +644,9 @@ class FactorioServer extends events.EventEmitter {
 	}
 
 	/**
-	 * The version of Factorio that was detected
+	 * The version of Factorio in use.  This will be the actual version if
+	 * "latest" (the default) was specified as `factorioVersion` to the
+	 * constructor.
 	 */
 	get version() {
 		return this._version;

lib/fileOps/fileOps.js

@@ -1,3 +1,6 @@
+/**
+ * @module lib/fileOps
+ */
 const fs = require("fs-extra");
 const path = require("path");
 

lib/hash.js

@@ -1,3 +1,7 @@
+/**
+ * Hashing functions
+ * @module
+ */
 const crypto = require('crypto');
 const fs = require('fs');
 
@@ -6,6 +10,7 @@ const fs = require('fs');
  * Returns a promise that resolves to the SHA1 hash of the stream given
  *
  * @param stream - Node stream of the content to hash.
+ * @returns {Promise<string>} hash of the stream.
  */
 function hashStream(stream) {
 	return new Promise(function(resolve, reject) {
@@ -28,6 +33,7 @@ function hashStream(stream) {
  * Returns a promise that resolves to the SHA1 hash of the file given by path
  *
  * @param {string} path - Path to the file to hash.
+ * @returns {Promise<string>} hash of the file given.
  */
 function hashFile(path) {
 	return hashStream(fs.createReadStream(path));

lib/link/connectors.js

@@ -1,3 +1,4 @@
+// Connection adapters for Links
 "use strict";
 const events = require("events");
 const WebSocket = require("ws");
@@ -6,6 +7,12 @@ const schema = require("lib/schema");
 const errors = require("lib/errors");
 
 
+/**
+ * Base connector for links
+ *
+ * @extends events.EventEmitter
+ * @memberof module:lib/link
+ */
 class WebSocketBaseConnector extends events.EventEmitter {
 	constructor() {
 		super();
@@ -36,7 +43,7 @@ class WebSocketBaseConnector extends events.EventEmitter {
 	}
 
 	_doHeartbeat() {
-		this._check(["connected"]);
+		this._check(["connected", "closing"]);
 		if (Date.now() - this._lastHeartbeat > 2000 * this._heartbeatInterval) {
 			console.log("SOCKET | closing after heartbeat timed out");
 			this._socket.close(1008, "Heartbeat timeout");
@@ -108,10 +115,27 @@ class WebSocketBaseConnector extends events.EventEmitter {
 	setClosing() {
 		this._state = "closing";
 	}
+
+	/**
+	 * True if the connection is established and active
+	 */
+	get connected() {
+		return this._state === "connected";
+	}
+
+	/**
+	 * True if the connection is closing down or closed
+	 */
+	get closing() {
+		return ["closing", "closed"].includes(this._state);
+	}
 }
 
 /**
  * Connector for master server clients
+ *
+ * @extends module:lib/link.WebSocketBaseConnector
+ * @memberof module:lib/link
  */
 class WebSocketClientConnector extends WebSocketBaseConnector {
 	constructor(url, reconnectDelay) {
@@ -192,7 +216,7 @@ class WebSocketClientConnector extends WebSocketBaseConnector {
 		this._doConnect();
 
 		// Wait for the connection to be ready
-		await events.once(this, "connected");
+		await events.once(this, "connect");
 	}
 
 	async _doConnect() {
@@ -229,11 +253,8 @@ class WebSocketClientConnector extends WebSocketBaseConnector {
 			this._lastReceivedSeq = null;
 			this._sessionToken = null;
 			this._sendBuffer.length = 0;
-			if (this._connected) {
-				this._connected = false;
-				this.emit("close");
-			}
-			this.emit("invalidate");
+			this._connected = false;
+			this.emit("close");
 			return;
 		}
 
@@ -259,16 +280,16 @@ class WebSocketClientConnector extends WebSocketBaseConnector {
 				this._sessionToken = null;
 				this._sendBuffer.length = 0;
 				this._state = "new";
-
-				if (this._connected) {
-					this._connected = false;
-					this.emit("close");
-				}
+				this._connected = false;
+				this.emit("close");
 
 			} else {
 				this._state = "handshake";
 				this.reconnect();
-				this.emit("dropped");
+				if (this._connected) {
+					this._connected = false;
+					this.emit("drop");
+				}
 			}
 
 			this.stopHeartbeat();
@@ -335,6 +356,7 @@ class WebSocketClientConnector extends WebSocketBaseConnector {
 		let { seq, type, data } = message;
 		if (type == 'hello') {
 			console.log(`SOCKET | received hello from master version ${data.version}`);
+			this.emit("hello", data);
 			if (this._sessionToken) {
 				console.log("SOCKET | Attempting resume");
 				this.sendHandshake("resume", {
@@ -351,9 +373,12 @@ class WebSocketClientConnector extends WebSocketBaseConnector {
 			this._sessionToken = data.session_token;
 			this._heartbeatInterval = data.heartbeat_interval;
 			this.startHeartbeat();
+			for (let message of this._sendBuffer) {
+				this._socket.send(JSON.stringify(message));
+			}
 			this._startedReconnect = null;
 			this._connected = true;
-			this.emit("connected");
+			this.emit("connect");
 
 		} else if (type === "continue") {
 			console.log("SOCKET | resuming existing session");
@@ -366,7 +391,7 @@ class WebSocketClientConnector extends WebSocketBaseConnector {
 			}
 			this._startedReconnect = null;
 			this._connected = true;
-			this.emit("connected");
+			this.emit("connect");
 
 		} else if (type === "invalidate") {
 			this._heartbeatInterval = null;
@@ -389,6 +414,12 @@ class WebSocketClientConnector extends WebSocketBaseConnector {
 	}
 }
 
+/**
+ * Connector for in-memory local links
+ *
+ * @extends events.EventEmitter
+ * @memberof module:lib/link
+ */
 class VirtualConnector extends events.EventEmitter {
 	constructor() {
 		super();

lib/link/index.js

@@ -1,3 +1,8 @@
+/**
+ * Library for message based communication between nodes
+ * @module lib/link
+ * @author Hornwitser
+ */
 module.exports = {
 	...require('lib/link/link'),
 	...require('lib/link/messages'),

lib/link/link.js

@@ -1,3 +1,4 @@
+// Implementation of Link class
 "use strict";
 
 const schema = require("lib/schema");
@@ -15,6 +16,8 @@ const messages = require("lib/link/messages");
 
 /**
  * Common interface for server and client connections
+ *
+ * @memberof module:lib/link
  */
 class Link {
 	constructor(source, target, connector) {
@@ -51,6 +54,16 @@ class Link {
 
 			this._waiters.clear();
 		});
+
+		connector.on("close", () => {
+			for (let waiterType of this._waiters.values()) {
+				for (let waiter of waiterType) {
+					waiter.reject(new errors.SessionLost("Session Closed"));
+				}
+			}
+
+			this._waiters.clear();
+		});
 	}
 
 	/**
@@ -59,7 +72,7 @@ class Link {
 	 * Validates and invokes the handler and/or waiters for a message that has
 	 * been received.  An unhandled message is considered to be an error.
 	 *
-	 * @throws {InvalidMessage}
+	 * @throws {module:lib/errors.InvalidMessage}
 	 *     if validation failed, is missing, the message is invalid, or no
 	 *     handlers/waiters were invoked to handle the message.
 	 */

lib/link/messages.js

@@ -1,14 +1,38 @@
+// Message definitions for links
 "use strict";
 const schema = require("lib/schema");
 const errors = require("lib/errors");
 
+/**
+ * Represents a message that can be sent over the link
+ *
+ * @memberof module:lib/link
+ */
+class Message {
+	/**
+	 * Name of the plugin this message belongs to, or null.
+	 * @returns {?string}
+	 **/
+	get plugin() {
+		if (this._plugin === undefined) {
+			let index = this.type.indexOf(":");
+			this._plugin = index === -1 ? null : this.type.slice(0, index);
+		}
+
+		return this._plugin;
+	}
+}
+
 /**
  * Represents a request sent over the link
  *
  * A request-response message that can be innitated by either party of the
  * link.
+ *
+ * @extends module:lib/link.Message
+ * @memberof module:lib/link
  */
-class Request {
+class Request extends Message {
 
 	/**
 	 * Creates a request
@@ -34,6 +58,7 @@ class Request {
 	constructor({
 		type, links, forwardTo = null, requestProperties = {}, responseProperties = {}
 	}) {
+		super();
 		this.type = type;
 		this.links = links;
 		this.forwardTo = forwardTo;
@@ -190,6 +215,11 @@ messages.prepareDisconnect = new Request({
 	links: wsLinks,
 });
 
+messages.prepareMasterDisconnect = new Request({
+	type: "prepare_master_disconnect",
+	links: ["slave-instance"],
+});
+
 messages.ping = new Request({
 	type: "ping",
 	links: wsLinks,
@@ -353,8 +383,11 @@ messages.getMetrics = new Request({
  *
  * A one way message without any response or recipt confirmation that
  * can be innitiated by either party of the link.
+ *
+ * @extends module:lib/link.Message
+ * @memberof module:lib/link
  */
-class Event {
+class Event extends Message {
 
 	/**
 	 * Creates an event
@@ -380,6 +413,7 @@ class Event {
 	 *     properties that are valid in the data payload of this event.
 	 */
 	constructor({ type, links, forwardTo = null, broadcastTo = null, eventProperties = {} }) {
+		super();
 		this.type = type;
 		this.links = links;
 		this.forwardTo = forwardTo;
@@ -499,9 +533,30 @@ messages.debugWsMessage = new Event({
 	},
 });
 
+messages.instanceInitialized = new Event({
+	type: "instance_initialized",
+	links: ["instance-slave", "slave-master"],
+	eventProperties: {
+		"instance_id": { type: "integer" },
+		"plugins": {
+			type: "object",
+			additionalProperties: { type: "string" },
+		}
+	},
+});
+messages.instanceStarted = new Event({
+	type: "instance_started",
+	links: ["instance-slave", "slave-master"],
+	eventProperties: {
+		"instance_id": { type: "integer" },
+	},
+});
 messages.instanceStopped = new Event({
 	type: 'instance_stopped',
-	links: ['instance-slave'],
+	links: ["instance-slave", "slave-master"],
+	eventProperties: {
+		"instance_id": { type: "integer" },
+	},
 });
 
 messages.updateInstances = new Event({
@@ -513,9 +568,10 @@ messages.updateInstances = new Event({
 			items: {
 				type: "object",
 				additionalProperties: false,
-				required: ["serialized_config"],
+				required: ["serialized_config", "status"],
 				properties: {
 					"serialized_config": { type: "object" },
+					"status": { enum: ["stopped", "initialized", "running"] },
 				},
 			},
 		},
@@ -547,6 +603,13 @@ messages.instanceOutput = new Event({
 	},
 });
 
+messages.masterConnectionEvent = new Event({
+	type: "master_connection_event",
+	links: ["slave-instance"],
+	eventProperties: {
+		"event": { type: "string" },
+	},
+});
 
 /**
  * Attaches all requests and events to the given link
@@ -555,6 +618,8 @@ messages.instanceOutput = new Event({
  * 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'.
+ *
+ * @memberof module:lib/link
  */
 function attachAllMessages(link) {
 	for (let [name, message] of Object.entries(messages)) {
@@ -564,6 +629,7 @@ function attachAllMessages(link) {
 
 
 module.exports = {
+	Message,
 	Request,
 	Event,
 

lib/luaTools.js

@@ -1,3 +1,8 @@
+/**
+ * Utilities for dealing with Lua
+ * @module
+ */
+
 /**
  * Escapes a string for inclusion into a lua string
  */

lib/plugin.js

@@ -11,6 +11,7 @@ const errors = require("lib/errors");
 
 /**
  * Base class for instance plugins
+ * @static
  */
 class BaseInstancePlugin {
 	constructor(info, instance, slave) {
@@ -21,7 +22,7 @@ class BaseInstancePlugin {
 
 		/**
 		 * Instance the plugin started for
-		 * @type {Instance}
+		 * @type {module:slave~Instance}
 		 */
 		this.instance = instance;
 
@@ -31,7 +32,7 @@ class BaseInstancePlugin {
 		 * With the exepction of accessing the slave's config you should
 		 * avoid ineracting with the slave object directly.
 		 *
-		 * @type {Slave}
+		 * @type {module:slave~Slave}
 		 */
 		this.slave = slave;
 	}
@@ -72,7 +73,11 @@ class BaseInstancePlugin {
 	async onStop() { }
 
 	/**
-	 * Called when the Factorio server exited
+	 * Called when the instance exits
+	 *
+	 * Invoked when the instance is shut down.  This may occur after init
+	 * has been called if an error occurs during startup.  Note that if
+	 * the plugin's init() throws this method will still be invoked.
 	 */
 	onExit() { }
 
@@ -93,8 +98,59 @@ class BaseInstancePlugin {
 	 * - message: Main content of the line.
 	 */
 	async onOutput(output) { }
+
+	/**
+	 * Called when an event on the master connection happens
+	 *
+	 * The event param may be one of connect, drop and close and has the
+	 * following meaning:
+	 *
+	 * ##### connect
+	 *
+	 * Invoked when a new connection to the master has been established, or
+	 * the existing one that have previously been dropped is re-established.
+	 *
+	 * ##### drop
+	 *
+	 * Invoked when a connection loss is detected between the slave and the
+	 * master.  Plugins should respond to this event by throtteling messages
+	 * it is sending to the master to an absolute minimum.
+	 *
+	 * Messages sent over a dropped master connection will get queued up in
+	 * memory on the slave and sent all in one go when the connection is
+	 * re-established again.
+	 *
+	 * ##### close
+	 *
+	 * Invoked when the connection to the master has been closed.  This
+	 * typically means the master server has shut down.  Plugins should not
+	 * send any messages that goes to or via the master server after the
+	 * connection has been closed and before a new one is established.
+	 *
+	 * @param {string} event - one of connect, drop, and close
+	 */
+	onMasterConnectionEvent(event) { }
+
+	/**
+	 * Called when the master is preparing to disconnect from the slave
+	 *
+	 * Invoked when the master server has requested the disconnection from
+	 * the slave.  This typically happens when it is in the process of
+	 * shutting down.
+	 *
+	 * Plugins must stop sending messages to the master, or are forwarded
+	 * via the master server after the prepare disconnect has been handled.
+	 *
+	 * @param {module:master~SlaveConnection} connection -
+	 *     The connection to the slave preparing to disconnect.
+	 */
+	async onPrepareMasterDisconnect(connection) { }
 }
 
+/**
+ * Base class for master plugins
+ * @static
+ */
 class BaseMasterPlugin {
 	constructor(info, master, metrics) {
 		this.info = info;
@@ -107,6 +163,21 @@ class BaseMasterPlugin {
 	 */
 	async init() { }
 
+	/**
+	 * Called when the status of an instance changes
+	 *
+	 * Invoked when the master server has received notice from a slave that
+	 * the running status of an instance has changed.  The possible statuses
+	 * are `stopped` meaning it is not loaded or running, `initialized`
+	 * meaning the instance has initialized and is starting up the Factorio
+	 * server, and `running` which means the Factorio server it manages is
+	 * up and running.
+	 *
+	 * @param {Object} instance - the instance that changed.
+	 * @param {string} prev - the previous status of the instance.
+	 */
+	async onInstanceStatusChanged(instance, prev) { }
+
 	/**
 	 * Called before collecting Prometheus metrics
 	 *
@@ -128,6 +199,72 @@ class BaseMasterPlugin {
 	 * Called when the master server is shutting down
 	 */
 	async onShutdown() { }
+
+	/**
+	 * Called when an event on a slave connection happens
+	 *
+	 * The event param may be one of connect, drop and close and has the
+	 * following meaning:
+	 *
+	 * ##### connect
+	 *
+	 * Invoked when a new connection to the slave has been established, or
+	 * an existing one that had previously been dropped is re-established.
+	 *
+	 * ##### drop
+	 *
+	 * Invoked when a connection loss is detected between the master and a
+	 * slave.  Plugins should respond to this event by throtteling messages
+	 * it is sending to the given slave connection to an absolute minimum.
+	 *
+	 * Messages sent over a dropped slave connection will get queued up in
+	 * memory on the master and sent all in one go when the connection is
+	 * re-established again.
+	 *
+	 * ##### close
+	 *
+	 * Invoked when the connection to the slave has been closed.
+	 *
+	 * @param {string} event - one of connect, drop, and close
+	 * @param {module:master~SlaveConnection} connection -
+	 *     The connection the event occured on.
+	 */
+	onSlaveConnectionEvent(connection, event) { }
+
+	/**
+	 * Called when a slave is preparing to disconnect from the master
+	 *
+	 * Invoked when a slave has requested the disconnection from the master.
+	 * This typically happens when it is in the process of shutting down.
+	 *
+	 * Plugins must stop sending messages to the slave in question after the
+	 * prepare disconnect has been handled.
+	 *
+	 * @param {module:master~SlaveConnection} connection -
+	 *     The connection to the slave preparing to disconnect.
+	 */
+	async onPrepareSlaveDisconnect(connection) { }
+
+	/**
+	 * Broadcast event to all connected slaves
+	 *
+	 * Sends the given event to all slaves connected to the master server.
+	 * This does not include slaves that are in the process of closing the
+	 * connection, which typically happens when they are shutting down.
+	 *
+	 * @param {module:lib/link.Event} event - Event to send
+	 * @param {Object} data - Data ta pass with the event.
+	 */
+	broadcastEventToSlaves(event, data={}) {
+		for (let slaveConnection of this.master.slaveConnections.values()) {
+			if (
+				!slaveConnection.connector.closing
+				&& (!event.plugin || slaveConnection.plugins.has(event.plugin))
+			) {
+				event.send(slaveConnection, data);
+			}
+		}
+	}
 }
 
 /**
@@ -138,6 +275,7 @@ class BaseMasterPlugin {
  * function be called again.
  *
  * @returns {Map<string, Object>} mapping of plugin name to info module.
+ * @static
  */
 async function loadPluginInfos(baseDir) {
 	let plugins = [];
@@ -167,9 +305,23 @@ async function loadPluginInfos(baseDir) {
 	return plugins;
 }
 
+/**
+ * Attach plugin messages
+ *
+ * Attaches all messages defined in the `.message` property of the
+ * `pluginInfo` passed with handlers taken from `pluin`.
+ *
+ * @param {module:lib/link.Link} link - Link to attach handlers to.
+ * @param {object} pluginInfo - Plugin info object.
+ * @param plugin - The instance of the plugin to use handlers from.
+ */
 function attachPluginMessages(link, pluginInfo, plugin) {
 	let messageDefinitions = pluginInfo.messages || [];
 	for (let [name, messageFormat] of Object.entries(messageDefinitions)) {
+		if (messageFormat.plugin !== pluginInfo.name) {
+			throw new Error(`Type of ${name} message must start with "${pluginInfo.name}:"`);
+		}
+
 		let handler = name + messageFormat.constructor.name + 'Handler';
 		if (plugin === null || !plugin[handler]) {
 			messageFormat.attach(link);

lib/schema.js

@@ -1,3 +1,7 @@
+/**
+ * JSON schemas used for validating link messages
+ * @module
+ */
 const ajv = new (require('ajv'))({
 	verbose: true,
 	format: "full",
@@ -24,8 +28,20 @@ const messageSchema = {
 	},
 };
 
+/**
+ * Validator for message schemas
+ * @function
+ * @param {Object} obj - Object to validate
+ * @returns {boolean} True if object passed validation.
+ */
 const message = ajv.compile(messageSchema);
 
+/**
+ * Validator for heartbeat messages
+ * @function
+ * @param {Object} obj - Object to validate
+ * @returns {boolean} True if object passed validation.
+ */
 const heartbeat = ajv.compile({
 	$schema: "http://json-schema.org/draft-07/schema#",
 	title: "Heartbeat",
@@ -47,6 +63,12 @@ const heartbeat = ajv.compile({
 	},
 });
 
+/**
+ * Validator for server sent part of handshake
+ * @function
+ * @param {Object} obj - Object to validate
+ * @returns {boolean} True if object passed validation.
+ */
 const serverHandshake = ajv.compile({
 	$schema: "http://json-schema.org/draft-07/schema#",
 	title: "Server Handshake",
@@ -58,9 +80,13 @@ const serverHandshake = ajv.compile({
 				"type": { const: "hello" },
 				"data": {
 					additionalProperties: false,
-					required: ["version"],
+					required: ["version", "plugins"],
 					properties: {
 						"version": { type: "string" },
+						"plugins": {
+							type: "object",
+							additionalProperties: { type: "string" },
+						},
 					}
 				},
 			},
@@ -100,6 +126,12 @@ const serverHandshake = ajv.compile({
 	],
 });
 
+/**
+ * Validator for client sent part of handshake
+ * @function
+ * @param {Object} obj - Object to validate
+ * @returns {boolean} True if object passed validation.
+ */
 const clientHandshake = ajv.compile({
 	$schema: "http://json-schema.org/draft-07/schema#",
 	title: "Client Message",
@@ -111,13 +143,17 @@ const clientHandshake = ajv.compile({
 				"type": { const: "register_slave" },
 				"data": {
 					additionalProperties: false,
-					required: ["token", "agent", "version", "name", "id"],
+					required: ["token", "agent", "version", "name", "id", "plugins"],
 					properties: {
 						"token": { type: "string" },
 						"agent": { type: "string" },
 						"version": { type: "string" },
 						"name": { type: "string" },
 						"id": { type: "integer" },
+						"plugins": {
+							type: "object",
+							additionalProperties: { type: "string" },
+						},
 					},
 				},
 			},
@@ -154,6 +190,12 @@ const clientHandshake = ajv.compile({
 
 
 module.exports = {
+	/**
+	 * Compile JSON schema into validator
+	 * @function
+	 * @param {Object} schema - JSON schema to create validator for.
+	 * @returns {Function} Validator for the schema.
+	 */
 	compile: ajv.compile.bind(ajv),
 
 	message,

master.js

@@ -1,13 +1,16 @@
 /**
-Clusterio master server. Facilitates communication between slaves through
-a webserver, storing data related to slaves like production graphs.
-
-@module
-@author Danielv123
-
-@example
-node master
-*/
+ * Clusterio master server
+ *
+ * Facilitates communication between slaves and control of the cluster
+ * through WebSocet connections, and hosts a webserver for browser
+ * interfaces and Prometheus statistics export.  It is remotely controlled
+ * by {@link module:clusterctl}.
+ *
+ * @module
+ * @author Danielv123, Hornwitser
+ * @example
+ * node master run
+ */
 
 // Attempt updating
 // const updater = require("./updater");
@@ -31,6 +34,7 @@ let masterConfig;
 let masterConfigPath;
 let stopAcceptingNewSessions = false;
 let debugEvents = new events.EventEmitter();
+let pluginList = {};
 
 // homebrew modules
 const generateSSLcert = require("lib/generateSSLcert");
@@ -205,7 +209,7 @@ async function loadInstances(databaseDirectory, file) {
 		for (let serializedConfig of serialized) {
 			let instanceConfig = new config.InstanceConfig();
 			await instanceConfig.load(serializedConfig);
-			instances.set(instanceConfig.get("instance.id"), instanceConfig);
+			instances.set(instanceConfig.get("instance.id"), { config: instanceConfig });
 		}
 
 	} catch (err) {
@@ -229,8 +233,8 @@ async function saveMap(databaseDirectory, file, map) {
 async function saveInstances(databaseDirectory, file, instances) {
 	let filePath = path.join(databaseDirectory, file);
 	let serialized = [];
-	for (let instanceConfig of instances.values()) {
-		serialized.push(instanceConfig.serialize());
+	for (let instance of instances.values()) {
+		serialized.push(instance.config.serialize());
 	}
 
 	await fs.outputFile(filePath, JSON.stringify(serialized, null, 4));
@@ -344,6 +348,11 @@ app.get("/api/getFactorioLocale", function(req,res){
 	}
 });
 
+/**
+ * Base class for master server connections
+ *
+ * @extends module:lib/link.Link
+ */
 class BaseConnection extends link.Link {
 	constructor(target, connector) {
 		super('master', target, connector);
@@ -354,12 +363,12 @@ class BaseConnection extends link.Link {
 	}
 
 	async forwardRequestToInstance(message, request) {
-		let instanceConfig = db.instances.get(message.data.instance_id);
-		if (!instanceConfig) {
+		let instance = db.instances.get(message.data.instance_id);
+		if (!instance) {
 			throw new errors.RequestError(`Instance with ID ${message.data.instance_id} does not exist`);
 		}
 
-		let slaveId = instanceConfig.get('instance.assigned_slave');
+		let slaveId = instance.config.get('instance.assigned_slave');
 		if (slaveId === null) {
 			throw new errors.RequestError("Instance is not assigned to a slave");
 		}
@@ -368,34 +377,42 @@ class BaseConnection extends link.Link {
 		if (!connection) {
 			throw new errors.RequestError("Slave containing instance is not connected");
 		}
+		if (request.plugin && !connection.plugins.has(request.plugin)) {
+			throw new errors.RequestError(`Slave containing instance does not have ${request.plugin} plugin`)
+		}
 
 		return await request.send(connection, message.data);
 	}
 
 	async forwardEventToInstance(message, event) {
-		let instanceConfig = db.instances.get(message.data.instance_id);
-		if (!instanceConfig) { return; }
+		let instance = db.instances.get(message.data.instance_id);
+		if (!instance) { return; }
 
-		let slaveId = instanceConfig.get('instance.assigned_slave');
+		let slaveId = instance.config.get('instance.assigned_slave');
 		if (slaveId === null) { return; }
 
-		let connection = slaveConnection.get(slaveId);
-		if (!connection) { return; }
+		let connection = slaveConnections.get(slaveId);
+		if (!connection || connection.closing) { return; }
+		if (event.plugin && !connection.plugins.has(event.plugin)) { return; }
 
 		event.send(connection, message.data);
 	}
 
 	async broadcastEventToInstance(message, event) {
 		for (let slaveConnection of slaveConnections.values()) {
-			if (slaveConnection === this) {
-				continue; // Do not broadcast back to the source
-			}
+			// Do not broadcast back to the source
+			if (slaveConnection === this) { continue; }
+			if (slaveConnection.connector.closing) { continue; }
+			if (event.plugin && !slaveConnection.plugins.has(event.plugin)) { continue; }
 
 			event.send(slaveConnection, message.data);
 		}
 	}
 
 	async prepareDisconnectRequestHandler(message, request) {
+		for (let masterPlugin of masterPlugins.values()) {
+			await masterPlugin.onPrepareSlaveDisconnect(this);
+		}
 		this.connector.setClosing();
 		return await super.prepareDisconnectRequestHandler(message, request);
 	}
@@ -432,7 +449,7 @@ class ControlConnection extends BaseConnection {
 		this.instanceOutputSubscriptions = new Set();
 
 		this.ws_dumper = null;
-		this.connector.on("connected", () => {
+		this.connector.on("connect", () => {
 			this.connector._socket.clusterio_ignore_dump = !!this.ws_dumper;
 		});
 		this.connector.on("close", () => {
@@ -458,11 +475,11 @@ class ControlConnection extends BaseConnection {
 
 	async listInstancesRequestHandler(message) {
 		let list = [];
-		for (let instanceConfig of db.instances.values()) {
+		for (let instance of db.instances.values()) {
 			list.push({
-				id: instanceConfig.get("instance.id"),
-				name: instanceConfig.get("instance.name"),
-				assigned_slave: instanceConfig.get("instance.assigned_slave"),
+				id: instance.config.get("instance.id"),
+				name: instance.config.get("instance.name"),
+				assigned_slave: instance.config.get("instance.assigned_slave"),
 			});
 		}
 		return { list };
@@ -477,35 +494,35 @@ class ControlConnection extends BaseConnection {
 		if (db.instances.has(instanceId)) {
 			throw new errors.RequestError(`Instance with ID ${instanceId} already exists`);
 		}
-		db.instances.set(instanceId, instanceConfig);
+		db.instances.set(instanceId, { config: instanceConfig });
 	}
 
 	async deleteInstanceRequestHandler(message, request) {
-		let instanceConfig = db.instances.get(message.data.instance_id);
-		if (!instanceConfig) {
+		let instance = db.instances.get(message.data.instance_id);
+		if (!instance) {
 			throw new errors.RequestError(`Instance with ID ${message.data.instance_id} does not exist`);
 		}
 
-		if (instanceConfig.get('instance.assigned_slave') !== null) {
+		if (instance.config.get('instance.assigned_slave') !== null) {
 			await this.forwardRequestToInstance(message, request);
 		}
 		db.instances.delete(message.data.instance_id);
 	}
 
 	async getInstanceConfigRequestHandler(message) {
-		let instanceConfig = db.instances.get(message.data.instance_id);
-		if (!instanceConfig) {
+		let instance = db.instances.get(message.data.instance_id);
+		if (!instance) {
 			throw new errors.RequestError(`Instance with ID ${message.data.instance_id} does not exist`);
 		}
 
 		return {
-			serialized_config: instanceConfig.serialize(),
+			serialized_config: instance.config.serialize(),
 		}
 	}
 
 	async setInstanceConfigFieldRequestHandler(message) {
-		let instanceConfig = db.instances.get(message.data.instance_id);
-		if (!instanceConfig) {
+		let instance = db.instances.get(message.data.instance_id);
+		if (!instance) {
 			throw new errors.RequestError(`Instance with ID ${message.data.instance_id} does not exist`);
 		}
 
@@ -518,24 +535,24 @@ class ControlConnection extends BaseConnection {
 			throw new errors.RequestError("Setting instance.id is not supported");
 		}
 
-		instanceConfig.set(message.data.field, message.data.value);
+		instance.config.set(message.data.field, message.data.value);
 
 		// Push updated config to slave
-		let slaveId = instanceConfig.get('instance.assigned_slave');
+		let slaveId = instance.config.get('instance.assigned_slave');
 		if (slaveId) {
 			let connection = slaveConnections.get(slaveId);
 			if (connection) {
 				await link.messages.assignInstance.send(connection, {
-					instance_id: instanceConfig.get("instance.id"),
-					serialized_config: instanceConfig.serialize(),
+					instance_id: instance.config.get("instance.id"),
+					serialized_config: instance.config.serialize(),
 				});
 			}
 		}
 	}
 
 	async assignInstanceCommandRequestHandler(message, request) {
-		let instanceConfig = db.instances.get(message.data.instance_id);
-		if (!instanceConfig) {
+		let instance = db.instances.get(message.data.instance_id);
+		if (!instance) {
 			throw new errors.RequestError(`Instance with ID ${message.data.instance_id} does not exist`);
 		}
 
@@ -551,11 +568,11 @@ class ControlConnection extends BaseConnection {
 			throw new errors.RequestError("Target slave is not connected to the master server");
 		}
 
-		instanceConfig.set("instance.assigned_slave", message.data.slave_id)
+		instance.config.set("instance.assigned_slave", message.data.slave_id)
 
 		return await link.messages.assignInstance.send(connection, {
-			instance_id: instanceConfig.get("instance.id"),
-			serialized_config: instanceConfig.serialize(),
+			instance_id: instance.config.get("instance.id"),
+			serialized_config: instance.config.serialize(),
 		});
 	}
 
@@ -565,7 +582,9 @@ class ControlConnection extends BaseConnection {
 
 	async debugDumpWsRequestHandler(message) {
 		this.ws_dumper = data => {
-			link.messages.debugWsMessage.send(this, data);
+			if (this.connector.connected) {
+				link.messages.debugWsMessage.send(this, data);
+			}
 		};
 		this.connector._socket.clusterio_ignore_dump = true;
 		debugEvents.on("message", this.ws_dumper)
@@ -573,6 +592,11 @@ class ControlConnection extends BaseConnection {
 }
 
 var slaveConnections = new Map();
+/**
+ * Represents the connection to a slave
+ *
+ * @extends module:master~BaseConnection
+ */
 class SlaveConnection extends BaseConnection {
 	constructor(registerData, connector) {
 		super('slave', connector);
@@ -581,12 +605,14 @@ class SlaveConnection extends BaseConnection {
 		this._id = registerData.id;
 		this._name = registerData.name;
 		this._version = registerData.version;
+		this.plugins = new Map(Object.entries(registerData.plugins));
 
 		db.slaves.set(this._id, {
 			agent: this._agent,
 			id: this._id,
 			name: this._name,
 			version: this._version,
+			plugins: registerData.plugins,
 		});
 
 		this.connector.on("close", () => {
@@ -594,15 +620,53 @@ class SlaveConnection extends BaseConnection {
 				slaveConnections.delete(this._id);
 			}
 		});
+
+		for (let event of ["connect", "drop", "close"]) {
+			this.connector.on(event, () => {
+				for (let masterPlugin of masterPlugins.values()) {
+					masterPlugin.onSlaveConnectionEvent(this, event);
+				}
+			});
+		}
+	}
+
+	async instanceInitializedEventHandler(message, event) {
+		let instance = db.instances.get(message.data.instance_id);
+		let prev = instance.status;
+		instance.status = "initialized";
+		console.log(`Clusterio | Instance ${instance.config.get("instance.name")} Initialized`)
+		for (let masterPlugin of masterPlugins.values()) {
+			await masterPlugin.onInstanceStatusChanged(instance, prev);
+		}
+	}
+
+	async instanceStartedEventHandler(message, event) {
+		let instance = db.instances.get(message.data.instance_id);
+		let prev = instance.status;
+		instance.status = "running";
+		console.log(`Clusterio | Instance ${instance.config.get("instance.name")} Started`)
+		for (let masterPlugin of masterPlugins.values()) {
+			await masterPlugin.onInstanceStatusChanged(instance, prev);
+		}
+	}
+
+	async instanceStoppedEventHandler(message, event) {
+		let instance = db.instances.get(message.data.instance_id);
+		let prev = instance.status;
+		instance.status = "stopped";
+		console.log(`Clusterio | Instance ${instance.config.get("instance.name")} Stopped`)
+		for (let masterPlugin of masterPlugins.values()) {
+			await masterPlugin.onInstanceStatusChanged(instance, prev);
+		}
 	}
 
 	async updateInstancesEventHandler(message) {
 		// Push updated instance configs
-		for (let instanceConfig of db.instances.values()) {
-			if (instanceConfig.get("instance.assigned_slave") === this._id) {
+		for (let instance of db.instances.values()) {
+			if (instance.config.get("instance.assigned_slave") === this._id) {
 				await link.messages.assignInstance.send(this, {
-					instance_id: instanceConfig.get("instance.id"),
-					serialized_config: instanceConfig.serialize(),
+					instance_id: instance.config.get("instance.id"),
+					serialized_config: instance.config.serialize(),
 				});
 			}
 		}
@@ -611,12 +675,24 @@ class SlaveConnection extends BaseConnection {
 		for (let instance of message.data.instances) {
 			let instanceConfig = new config.InstanceConfig();
 			await instanceConfig.load(instance.serialized_config);
-			if (db.instances.has(instanceConfig.get("instance.id"))) {
+
+			let masterInstance = db.instances.get(instanceConfig.get("instance.id"))
+			if (masterInstance) {
+				// Already have this instance, update state instead
+				if (masterInstance.status !== instance.status) {
+					let prev = masterInstance.status;
+					masterInstance.status = instance.status;
+					if (prev !== undefined) {
+						for (let masterPlugin of masterPlugins.values()) {
+							await masterPlugin.onInstanceStatusChanged(instance, prev);
+						}
+					}
+				}
 				continue;
 			}
 
 			instanceConfig.set("instance.assigned_slave", this._id);
-			db.instances.set(instanceConfig.get("instance.id"), instanceConfig);
+			db.instances.set(instanceConfig.get("instance.id"), { config: instanceConfig });
 			await link.messages.assignInstance.send(this, {
 				instance_id: instanceConfig.get("instance.id"),
 				serialized_config: instanceConfig.serialize(),
@@ -644,7 +720,8 @@ const wss = new WebSocket.Server({
  *
  * @param value - value to test.
  * @returns {Boolean}
- *     true if value is an integer between -2**31 and 2**31-1.
+ *     true if value is an integer between -2<sup>31</sup> and
+ *     2<sup>31</sup>-1.
  */
 function isInteger(value) {
 	return value | 0 === value;
@@ -653,6 +730,8 @@ function isInteger(value) {
 
 /**
  * Connector for master server connections
+ *
+ * @extends module:lib/link.WebSocketBaseConnector
  */
 class WebSocketServerConnector extends link.WebSocketBaseConnector {
 	constructor(socket, sessionId) {
@@ -698,7 +777,7 @@ class WebSocketServerConnector extends link.WebSocketBaseConnector {
 		this._state = "connected";
 		this._connected = true;
 		this._attachSocketHandlers();
-		this.emit("connected");
+		this.emit("connect");
 	}
 
 	/**
@@ -733,7 +812,7 @@ class WebSocketServerConnector extends link.WebSocketBaseConnector {
 		for (let message of this._sendBuffer) {
 			this._socket.send(JSON.stringify(message));
 		}
-		this.emit("connected");
+		this.emit("connect");
 	}
 
 	setTimeout(timeout) {
@@ -779,7 +858,7 @@ class WebSocketServerConnector extends link.WebSocketBaseConnector {
 
 			} else {
 				this._state = "handshake";
-				this.emit("dropped");
+				this.emit("drop");
 				this._timeoutId = setTimeout(() => { this._timedOut() }, this._timeout * 1000);
 			}
 
@@ -863,7 +942,10 @@ wss.on("connection", function (socket, req) {
 	};
 
 	// Start connection handshake.
-	socket.send(JSON.stringify({ seq: null, type: "hello", data: { version }}));
+	socket.send(JSON.stringify({ seq: null, type: "hello", data: {
+		version,
+		plugins: pluginList
+	}}));
 
 	function attachHandler() {
 		pendingSockets.add(socket);
@@ -971,6 +1053,7 @@ async function handleHandshake(message, socket, req, attachHandler) {
 		let connection = slaveConnections.get(data.id);
 		if (connection) {
 			console.log(`SOCKET | disconnecting existing connection for slave ${data.id}`);
+			connection.connector.setTimeout(15); // Slave connection is likely stalled
 			await connection.disconnect(1008, "Registered from another connection");
 		}
 
@@ -1000,6 +1083,8 @@ async function loadPlugins(pluginInfos) {
 			continue;
 		}
 
+		pluginList[pluginInfo.name] = pluginInfo.version;
+
 		let pluginLoadStarted = Date.now();
 		let MasterPlugin = plugin.BaseMasterPlugin;
 		try {

plugins/global_chat/instance.js

@@ -1,3 +1,6 @@
+/**
+ * @module
+ */
 const plugin = require("lib/plugin");
 const luaTools = require("lib/luaTools");
 
@@ -10,12 +13,17 @@ function removeTags(content) {
 }
 
 class InstancePlugin extends plugin.BaseInstancePlugin {
-	constructor(...args) {
-		super(...args);
+	async init() {
+		this.messageQueue = [];
 	}
 
-	async init() {
-		// XXX Nothing?
+	onMasterConnectionEvent(event) {
+		if (event === "connect") {
+			for (let message of this.messageQueue) {
+				this.sendChat(message);
+			}
+			this.messageQueue = [];
+		}
 	}
 
 	async chatEventHandler(message) {
@@ -24,12 +32,20 @@ class InstancePlugin extends plugin.BaseInstancePlugin {
 		await this.instance.server.sendRcon(`/sc game.print('${luaTools.escapeString(content)}')`, true);
 	}
 
+	sendChat(message) {
+		this.info.messages.chat.send(this.instance, {
+			instance_name: this.instance.name,
+			content: message,
+		});
+	}
+
 	async onOutput(output) {
 		if (output.type === "action" && output.action === "CHAT") {
-			this.info.messages.chat.send(this.instance, {
-				instance_name: this.instance.name,
-				content: output.message,
-			});
+			if (this.slave.connector.connected) {
+				this.sendChat(output.message);
+			} else {
+				this.messageQueue.push(output.message);
+			}
 		}
 	}
 

plugins/global_chat/test/plugin.js

@@ -23,7 +23,7 @@ describe("global_chat plugin", function() {
 		let instancePlugin;
 
 		before(async function() {
-			instancePlugin = new instance.InstancePlugin(info, new mock.MockInstance());
+			instancePlugin = new instance.InstancePlugin(info, new mock.MockInstance(), new mock.MockSlave());
 			await instancePlugin.init();
 		});
 

plugins/research_sync/master.js

@@ -49,9 +49,7 @@ class MasterPlugin extends plugin.BaseMasterPlugin {
 
 		this.lastProgressBroadcast = Date.now();
 		if (techs.length) {
-			for (let slaveConnection of this.master.slaveConnections.values()) {
-				this.info.messages.progress.send(slaveConnection, { technologies: techs });
-			}
+			this.broadcastEventToSlaves(this.info.messages.progress, { technologies: techs });
 		}
 	}
 
@@ -89,12 +87,7 @@ class MasterPlugin extends plugin.BaseMasterPlugin {
 			tech.progress = null;
 			this.progressToBroadcast.delete(name);
 
-			for (let slaveConnection of this.master.slaveConnections.values()) {
-				this.info.messages.finished.send(slaveConnection, {
-					name,
-					level: tech.level
-				});
-			}
+			this.broadcastEventToSlaves(this.info.messages.finished, { name, level: tech.level });
 		}
 	}
 
@@ -125,9 +118,7 @@ class MasterPlugin extends plugin.BaseMasterPlugin {
 				if (progress) {
 					this.progressToBroadcast.add(name);
 				} else if (researched || baseLevel(name) != level) {
-					for (let slaveConnection of this.master.slaveConnections.values()) {
-						this.info.messages.finished.send(slaveConnection, { name, level });
-					}
+					this.broadcastEventToSlaves(this.info.messages.finished, { name, level });
 				}
 
 			} else {
@@ -138,9 +129,7 @@ class MasterPlugin extends plugin.BaseMasterPlugin {
 				if (tech.level < level || researched) {
 					// Send update if the unlocked level is greater
 					if (level - !researched > tech.level - !tech.researched) {
-						for (let slaveConnection of this.master.slaveConnections.values()) {
-							this.info.messages.finished.send(slaveConnection, { name, level: level - !researched });
-						}
+						this.broadcastEventToSlaves(this.info.messages.finished, { name, level: level - !researched });
 					}
 					tech.level = level;
 					tech.progress = progress;

plugins/subspace_storage/master.js

@@ -77,9 +77,7 @@ class MasterPlugin extends plugin.BaseMasterPlugin {
 		}
 
 		itemsToUpdate = [...itemsToUpdate.entries()];
-		for (let slaveConnection of this.master.slaveConnections.values()) {
-			this.info.messages.updateStorage.send(slaveConnection, { items: itemsToUpdate });
-		}
+		this.broadcastEventToSlaves(this.info.messages.updateStorage, { items: itemsToUpdate });
 		this.itemsLastUpdate = new Map(this.items._items.entries());
 	}
 

slave.js

@@ -1,8 +1,21 @@
+/**
+ * Clusterio slave
+ *
+ * Connects to the master server and hosts Factorio servers that can
+ * communicate with the cluster.  It is remotely controlled by {@link
+ * module:master}.
+ *
+ * @module
+ * @author Danielv123, Hornwitser
+ * @example
+ * node slave run
+ */
 const fs = require('fs-extra');
 const path = require('path');
 const yargs = require("yargs");
 const events = require("events");
 const pidusage = require("pidusage");
+const setBlocking = require("set-blocking");
 const version = require("./package").version;
 
 // internal libraries
@@ -99,7 +112,7 @@ class Instance extends link.Link{
 
 	notifyExit() {
 		this._running = false;
-		link.messages.instanceStopped.send(this);
+		link.messages.instanceStopped.send(this, { instance_id: this.config.get("instance.id") });
 
 		// Clear metrics this instance is exporting
 		for (let collector of prometheus.defaultRegistry.collectors) {
@@ -117,25 +130,43 @@ class Instance extends link.Link{
 		}
 	}
 
+	async _loadPlugin(pluginInfo, slave) {
+		let pluginLoadStarted = Date.now();
+		let { InstancePlugin } = require(`./plugins/${pluginInfo.name}/${pluginInfo.instanceEntrypoint}`);
+		let instancePlugin = new InstancePlugin(pluginInfo, this, slave);
+		this.plugins.set(pluginInfo.name, instancePlugin);
+		await instancePlugin.init();
+		plugin.attachPluginMessages(this, pluginInfo, instancePlugin);
+
+		console.log(`Clusterio | Loaded plugin ${pluginInfo.name} in ${Date.now() - pluginLoadStarted}ms`);
+	}
+
 	async init(pluginInfos, slave) {
 		await this.server.init();
 
 		// load plugins
 		for (let pluginInfo of pluginInfos) {
-			if (!pluginInfo.instanceEntrypoint || !this.config.group(pluginInfo.name).get("enabled")) {
+			if (
+				!pluginInfo.instanceEntrypoint
+				|| !slave.serverPlugins.has(pluginInfo.name)
+				|| !this.config.group(pluginInfo.name).get("enabled")
+			) {
 				continue;
 			}
 
-			// require plugin class and initialize it
-			let pluginLoadStarted = Date.now();
-			let { InstancePlugin } = require(`./plugins/${pluginInfo.name}/${pluginInfo.instanceEntrypoint}`);
-			let instancePlugin = new InstancePlugin(pluginInfo, this, slave);
-			await instancePlugin.init();
-			this.plugins.set(pluginInfo.name, instancePlugin);
-			plugin.attachPluginMessages(this, pluginInfo, instancePlugin);
-
-			console.log(`Clusterio | Loaded plugin ${pluginInfo.name} in ${Date.now() - pluginLoadStarted}ms`);
+			try {
+				await this._loadPlugin(pluginInfo, slave);
+			} catch (err) {
+				this.notifyExit();
+				throw err;
+			}
 		}
+
+		let plugins = {}
+		for (let [name, plugin] of this.plugins) {
+			plugins[name] = plugin.info.version;
+		}
+		link.messages.instanceInitialized.send(this, { instance_id: this.config.get("instance.id"), plugins });
 	}
 
 	/**
@@ -309,6 +340,7 @@ class Instance extends link.Link{
 		}
 
 		this._running = true;
+		link.messages.instanceStarted.send(this, { instance_id: this.config.get("instance.id") });
 	}
 
 	/**
@@ -336,6 +368,18 @@ class Instance extends link.Link{
 		}
 	}
 
+	async masterConnectionEventEventHandler(message) {
+		for (let pluginInstance of this.plugins.values()) {
+			pluginInstance.onMasterConnectionEvent(message.data.event);
+		}
+	}
+
+	async prepareMasterDisconnectRequestHandler() {
+		for (let pluginInstance of this.plugins.values()) {
+			await pluginInstance.onPrepareMasterDisconnect();
+		}
+	}
+
 	async getMetricsRequestHandler() {
 		let results = []
 		if (this._running) {
@@ -470,6 +514,8 @@ class InstanceConnection extends link.Link {
 		super('slave', 'instance', connector);
 		this.slave = slave;
 		this.instanceId = instanceId;
+		this.plugins = new Map();
+		this.status = "stopped";
 		link.attachAllMessages(this);
 
 		for (let pluginInfo of slave.pluginInfos) {
@@ -489,6 +535,7 @@ class InstanceConnection extends link.Link {
 			await this.slave.forwardEventToMaster(message, event);
 			return;
 		}
+		if (event.plugin && !instanceConnection.plugins.has(event.plugin)) { return; }
 
 		event.send(instanceConnection, message.data);
 	}
@@ -499,33 +546,53 @@ class InstanceConnection extends link.Link {
 
 	async broadcastEventToInstance(message, event) {
 		for (let instanceConnection of this.slave.instanceConnections.values()) {
-			if (instanceConnection === this) {
-				continue; // Do not broadcast back to the source
-			}
+			// Do not broadcast back to the source
+			if (instanceConnection === this) { continue; }
+			if (event.plugin && !instanceConnection.plugins.has(event.plugin)) { continue; }
 
 			event.send(instanceConnection, message.data);
 		}
 	}
 
-	async instanceStoppedEventHandler(message) {
+	async instanceInitializedEventHandler(message, event) {
+		this.status = "initialized";
+		this.plugins = new Map(Object.entries(message.data.plugins));
+		this.forwardEventToMaster(message, event);
+	}
+
+	async instanceStartedEventHandler(message, event) {
+		this.status = "running";
+		this.forwardEventToMaster(message, event);
+	}
+
+	async instanceStoppedEventHandler(message, event) {
+		this.status = "stopped";
 		this.slave.instanceConnections.delete(this.instanceId);
+		this.forwardEventToMaster(message, event);
 	}
 }
 
 class SlaveConnector extends link.WebSocketClientConnector {
-	constructor(slaveConfig) {
+	constructor(slaveConfig, pluginInfos) {
 		super(slaveConfig.get("slave.master_url"), slaveConfig.get("slave.reconnect_delay"));
 		this.slaveConfig = slaveConfig;
+		this.pluginInfos = pluginInfos;
 	}
 
 	register() {
 		console.log("SOCKET | registering slave");
+		let plugins = {};
+		for (let pluginInfo of this.pluginInfos) {
+			plugins[pluginInfo.name] = pluginInfo.version;
+		}
+
 		this.sendHandshake("register_slave", {
 			token: this.slaveConfig.get("slave.master_token"),
 			agent: "Clusterio Slave",
 			version,
 			id: this.slaveConfig.get("slave.id"),
 			name: this.slaveConfig.get("slave.name"),
+			plugins,
 		});
 	}
 }
@@ -551,6 +618,68 @@ class Slave extends link.Link {
 
 		this.instanceConnections = new Map();
 		this.instanceInfos = new Map();
+
+		this.connector.on("hello", data => {
+			this.serverVersion = data.version;
+			this.serverPlugins = new Map(Object.entries(data.plugins));
+		});
+
+		this._disconnecting = false;
+		this._shuttingDown = false;
+
+		this.connector.on("connect", () => {
+			if (this._shuttingDown) {
+				return;
+			}
+
+			this.updateInstances().catch((err) => {
+				console.error("ERROR: Unexpected error updating instances");
+				console.error(err);
+				this.shutdown().catch((err) => {
+					setBlocking(true);
+					console.error("ERROR: Unexpected error during shutdown");
+					console.error(err);
+					process.exit(1)
+				});
+			});;
+		});
+
+		this.connector.on("close", () => {
+			if (this._shuttingDown) {
+				return;
+			}
+
+			if (this._disconnecting) {
+				this._disconnecting = false;
+				this.connector.connect().catch((err) => {
+					console.error("ERROR: Unexpected error reconnecting to master");
+					console.error(err);
+					this.shutdown().catch((err) => {
+						setBlocking(true);
+						console.error("ERROR: Unexpected error during shutdown");
+						console.error(err);
+						process.exit(1)
+					});
+				});
+
+			} else {
+				console.error("ERROR: Master connection was unexpectedly closed");
+				this.shutdown().catch((err) => {
+					setBlocking(true);
+					console.error("ERROR: Unexpected error during shutdown");
+					console.error(err);
+					process.exit(1)
+				});
+			}
+		});
+
+		for (let event of ["connect", "drop", "close"]) {
+			this.connector.on(event, () => {
+				for (let instanceConnection of this.instanceConnections.values()) {
+					link.messages.masterConnectionEvent.send(instanceConnection, { event });
+				}
+			});
+		}
 	}
 
 	async _findNewInstanceDir(name) {
@@ -580,6 +709,10 @@ class Slave extends link.Link {
 			throw new errors.RequestError(`Instance ID ${instanceId} is not running`);
 		}
 
+		if (request.plugin && !instanceConnection.plugins.has(request.plugin)) {
+			throw new errors.RequestError(`Instance ID ${instanceId} does not have ${request.plugin} plugin loaded`)
+		}
+
 		return await request.send(instanceConnection, message.data);
 	}
 
@@ -589,12 +722,15 @@ class Slave extends link.Link {
 
 		let instanceConnection = this.instanceConnections.get(instanceId);
 		if (!instanceConnection) { return; }
+		if (event.plugin && !instanceConnection.plugins.has(event.plugin)) { return; }
 
 		event.send(instanceConnection, message.data);
 	}
 
 	async broadcastEventToInstance(message, event) {
 		for (let instanceConnection of this.instanceConnections.values()) {
+			if (event.plugin && !instanceConnection.plugins.has(event.plugin)) { continue; }
+
 			event.send(instanceConnection, message.data);
 		}
 	}
@@ -727,22 +863,30 @@ class Slave extends link.Link {
 	async updateInstances() {
 		this.instanceInfos = await discoverInstances(this.config.get("slave.instances_directory"), console);
 		let list = [];
-		for (let instanceInfo of this.instanceInfos.values()) {
+		for (let [instanceId, instanceInfo] of this.instanceInfos) {
+			let instanceConnection = this.instanceConnections.get(instanceId);
 			list.push({
 				serialized_config: instanceInfo.config.serialize(),
+				status: instanceConnection ? instanceConnection.status : "stopped",
 			});
 		}
 		link.messages.updateInstances.send(this, { instances: list });
 	}
 
-	async start() {
-		await this.updateInstances();
+	async prepareDisconnectRequestHandler(message, request) {
+		this._disconnecting = true;
+		for (let instanceConnection of this.instanceConnections.values()) {
+			await link.messages.prepareMasterDisconnect.send(instanceConnection);
+		}
+		this.connector.setClosing();
+		return await super.prepareDisconnectRequestHandler(message, request);
 	}
 
 	/**
 	 * Stops all instances and closes the connection
 	 */
 	async shutdown() {
+		this._shuttingDown = true;
 		this.connector.setTimeout(30);
 
 		try {
@@ -945,7 +1089,7 @@ async function startSlave() {
 		return;
 	}
 
-	let slaveConnector = new SlaveConnector(slaveConfig);
+	let slaveConnector = new SlaveConnector(slaveConfig, pluginInfos);
 	let slave = new Slave(slaveConnector, slaveConfig, pluginInfos);
 
 	// Handle interrupts
@@ -971,7 +1115,6 @@ async function startSlave() {
 	});
 
 	await slaveConnector.connect();
-	await slave.start();
 
 	/*
 	} else if (command == "manage"){

test/integration/link.js

@@ -12,8 +12,12 @@ class TestControl extends link.Link {
 		link.attachAllMessages(this);
 	}
 
+	async prepareDisconnectRequestHandler(message, request) {
+		this.connector.setClosing();
+		return await super.prepareDisconnectRequestHandler(message, request);
+	}
+
 	async debugWsMessageEventHandler() { }
-	async prepareDisconnectRequestHandler() { }
 	async instanceOutputEventHandler() { }
 }
 
@@ -45,27 +49,23 @@ describe("Integration of lib/link", function() {
 
 	it("should reconnect on connection lost", async function() {
 		controlConnector._socket.close(1008, "Test");
-		await events.once(controlConnector, "connected");
+		await events.once(controlConnector, "connect");
 	});
 
 	it("should reconnect on connection terminated", async function() {
 		controlConnector._socket.terminate();
-		await events.once(controlConnector, "connected");
+		await events.once(controlConnector, "connect");
 	});
 
 	it("should reconnect on connection heartbeat timeout", async function() {
 		controlConnector.stopHeartbeat();
-		await events.once(controlConnector, "connected");
+		await events.once(controlConnector, "connect");
 	});
 
 	it("should invalidate on reconnection with bad session token", async function() {
 		controlConnector._sessionToken = "blah";
 		controlConnector.stopHeartbeat();
 		await events.once(controlConnector, "invalidate");
-		await events.once(controlConnector, "connected");
-	});
-
-	after(async function() {
-		controlConnector.close(1001, "Test Quit");
+		await events.once(controlConnector, "connect");
 	});
 });

test/lib/link/connector.js

@@ -86,11 +86,11 @@ describe("lib/link/connectors", function() {
 			it("should call register on hello", function() {
 				let called = false;
 				testConnector.register = () => { called = true; };
-				testConnector._processHandshake({ seq: 1, type: 'hello', data: { version: "test" }});
+				testConnector._processHandshake({ seq: 1, type: 'hello', data: { version: "test", plugins: {} }});
 				assert(called, "register was not called");
 			});
-			it("should emit connected on ready", async function() {
-				let result = events.once(testConnector, 'connected');
+			it("should emit connect on ready", async function() {
+				let result = events.once(testConnector, "connect");
 				testConnector._processHandshake(
 					{ seq: 1, type: 'ready', data: { session_token: "x", heartbeat_interval: 10 }}
 				);

test/lib/plugin.js

@@ -21,6 +21,8 @@ describe("lib/plugin", function() {
 			await instancePlugin.onStop();
 			instancePlugin.onExit();
 			await instancePlugin.onOutput({});
+			instancePlugin.onMasterConnectionEvent("connect");
+			await instancePlugin.onPrepareMasterDisconnect();
 		})
 	});
 
@@ -31,8 +33,11 @@ describe("lib/plugin", function() {
 			await masterPlugin.init();
 		})
 		it("should define defaults for hooks", async function() {
+			await masterPlugin.onInstanceStatusChanged({}, "running", "initialized");
 			await masterPlugin.onMetrics();
 			await masterPlugin.onShutdown();
+			masterPlugin.onSlaveConnectionEvent({}, "connect");
+			await masterPlugin.onPrepareSlaveDisconnect({});
 		})
 	});
 
@@ -92,19 +97,25 @@ describe("lib/plugin", function() {
 
 	describe("attachPluginMessages()", function() {
 		let mockLink = new link.Link('source', 'target', new mock.MockConnector());
-		let mockEvent = new link.Event({ type: 'test', links: ['target-source'] });
+		let mockEvent = new link.Event({ type: "test:test", links: ["target-source"] });
 		it("should accept pluginInfo without messages", function() {
 			plugin.attachPluginMessages(mockLink, {}, null);
 		});
 		it("should attach handler for the given message", function() {
 			let mockEventEventHandler = function() {};
-			plugin.attachPluginMessages(mockLink, { messages: { mockEvent }}, { mockEventEventHandler });
-			assert(mockLink._handlers.get('test_event'), "handler was not registered");
+			plugin.attachPluginMessages(mockLink, { name: "test", messages: { mockEvent }}, { mockEventEventHandler });
+			assert(mockLink._handlers.get("test:test_event"), "handler was not registered");
 		});
 		it("should throw if missing handler for the given message", function() {
 			assert.throws(
-				() => plugin.attachPluginMessages(mockLink, { messages: { mockEvent }}, {}),
-				new Error("Missing handler for test_event on source-target link")
+				() => plugin.attachPluginMessages(mockLink, { name: "test", messages: { mockEvent }}, {}),
+				new Error("Missing handler for test:test_event on source-target link")
+			);
+		});
+		it("should throw if message starts with the wrong prefix", function() {
+			assert.throws(
+				() => plugin.attachPluginMessages(mockLink, { name: "foo", messages: { mockEvent }}, {}),
+				new Error('Type of mockEvent message must start with "foo:"')
 			);
 		});
 	});

test/mock.js

@@ -34,6 +34,9 @@ class MockConnector extends events.EventEmitter {
 		this.sentMessages = [];
 		this.events = new Map();
 		this.handshake = { address: "socket.test" };
+
+		this.connected = true;
+		this.closing = false;
 	}
 
 	send(type, data) {
@@ -70,9 +73,16 @@ class MockInstance extends link.Link {
 	}
 }
 
+class MockSlave extends link.Link {
+	constructor() {
+		super('slave', 'master', new MockConnector());
+	}
+}
+
 
 module.exports = {
 	MockSocket,
 	MockConnector,
 	MockInstance,
+	MockSlave,
 };