diff --git a/clusterctl.js b/clusterctl.js index d02594db..0b5ecc43 100644 --- a/clusterctl.js +++ b/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) { diff --git a/docs/jsdoc.json b/docs/jsdoc.json index 88b1ded8..54216916 100644 --- a/docs/jsdoc.json +++ b/docs/jsdoc.json @@ -18,6 +18,7 @@ "plugins/", "routes/", "routes.js" - ] + ], + "excludePattern": "node_modules" } } diff --git a/docs/socket.md b/docs/socket.md index 87d4718a..99bd49ef 100644 --- a/docs/socket.md +++ b/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` diff --git a/docs/writing-plugins.md b/docs/writing-plugins.md index 2b57862d..94b32f17 100644 --- a/docs/writing-plugins.md +++ b/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 --------------------- diff --git a/lib/config/classes.js b/lib/config/classes.js index c43ec6be..13196729 100644 --- a/lib/config/classes.js +++ b/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() { diff --git a/lib/config/definitions.js b/lib/config/definitions.js index 9c1dbc14..c20c61ed 100644 --- a/lib/config/definitions.js +++ b/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]; diff --git a/lib/config/index.js b/lib/config/index.js index bbe1c364..f591436e 100644 --- a/lib/config/index.js +++ b/lib/config/index.js @@ -1,8 +1,8 @@ /** - * @file Configuration System + * Configuration System * @author Hornwitser + * @module lib/config */ - module.exports = { ...require("./classes"), ...require("./definitions"), diff --git a/lib/errors.js b/lib/errors.js index 19c9d27f..2b9a6bdd 100644 --- a/lib/errors.js +++ b/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}`); diff --git a/lib/factorio/index.js b/lib/factorio/index.js index 1e03bc70..b99f8803 100644 --- a/lib/factorio/index.js +++ b/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'), diff --git a/lib/factorio/locale.js b/lib/factorio/locale.js index 36ee8a1e..0bc6cef2 100644 --- a/lib/factorio/locale.js +++ b/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"); diff --git a/lib/factorio/patch.js b/lib/factorio/patch.js index 2d5c1957..e7545be1 100644 --- a/lib/factorio/patch.js +++ b/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} 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} 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)); diff --git a/lib/factorio/server.js b/lib/factorio/server.js index 8f7a6540..bf31d557 100644 --- a/lib/factorio/server.js +++ b/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; diff --git a/lib/fileOps/fileOps.js b/lib/fileOps/fileOps.js index 251991a3..ed42b794 100644 --- a/lib/fileOps/fileOps.js +++ b/lib/fileOps/fileOps.js @@ -1,3 +1,6 @@ +/** + * @module lib/fileOps + */ const fs = require("fs-extra"); const path = require("path"); diff --git a/lib/hash.js b/lib/hash.js index 9cf9c4bf..9bb694ed 100644 --- a/lib/hash.js +++ b/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} 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} hash of the file given. */ function hashFile(path) { return hashStream(fs.createReadStream(path)); diff --git a/lib/link/connectors.js b/lib/link/connectors.js index 5a420323..d48b1a58 100644 --- a/lib/link/connectors.js +++ b/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(); diff --git a/lib/link/index.js b/lib/link/index.js index 132ff5d0..e5c18c5f 100644 --- a/lib/link/index.js +++ b/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'), diff --git a/lib/link/link.js b/lib/link/link.js index d36d8b56..67d2fc37 100644 --- a/lib/link/link.js +++ b/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. */ diff --git a/lib/link/messages.js b/lib/link/messages.js index 365bfa6a..f1fba2b5 100644 --- a/lib/link/messages.js +++ b/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, diff --git a/lib/luaTools.js b/lib/luaTools.js index 8165beb5..7566e03e 100644 --- a/lib/luaTools.js +++ b/lib/luaTools.js @@ -1,3 +1,8 @@ +/** + * Utilities for dealing with Lua + * @module + */ + /** * Escapes a string for inclusion into a lua string */ diff --git a/lib/plugin.js b/lib/plugin.js index 41661db3..81a66330 100644 --- a/lib/plugin.js +++ b/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} 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); diff --git a/lib/schema.js b/lib/schema.js index fec79da6..84fc6de0 100644 --- a/lib/schema.js +++ b/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, diff --git a/master.js b/master.js index a144945f..f624d279 100644 --- a/master.js +++ b/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 -231 and + * 231-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 { diff --git a/plugins/global_chat/instance.js b/plugins/global_chat/instance.js index e44ed210..cf6cfbee 100644 --- a/plugins/global_chat/instance.js +++ b/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); + } } } diff --git a/plugins/global_chat/test/plugin.js b/plugins/global_chat/test/plugin.js index 28590068..1652282e 100644 --- a/plugins/global_chat/test/plugin.js +++ b/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(); }); diff --git a/plugins/research_sync/master.js b/plugins/research_sync/master.js index 4e141be9..fa114503 100644 --- a/plugins/research_sync/master.js +++ b/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; diff --git a/plugins/subspace_storage/master.js b/plugins/subspace_storage/master.js index 64a10a3b..928757b7 100644 --- a/plugins/subspace_storage/master.js +++ b/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()); } diff --git a/slave.js b/slave.js index a08d56bd..eabefd02 100644 --- a/slave.js +++ b/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"){ diff --git a/test/integration/link.js b/test/integration/link.js index 5836d507..a5f2b95a 100644 --- a/test/integration/link.js +++ b/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"); }); }); diff --git a/test/lib/link/connector.js b/test/lib/link/connector.js index b800a679..6f0abb7c 100644 --- a/test/lib/link/connector.js +++ b/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 }} ); diff --git a/test/lib/plugin.js b/test/lib/plugin.js index 92cd772a..5e915e8a 100644 --- a/test/lib/plugin.js +++ b/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:"') ); }); }); diff --git a/test/mock.js b/test/mock.js index 1fcefdab..f2354628 100644 --- a/test/mock.js +++ b/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, };