diff --git a/docs/socket.md b/docs/socket.md index b0544b9f..99bd49ef 100644 --- a/docs/socket.md +++ b/docs/socket.md @@ -129,7 +129,7 @@ Slave handshake for establishing a new connection session. - 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. Currently not used by the master server. + available on the slave. ### `register_control` diff --git a/docs/writing-plugins.md b/docs/writing-plugins.md index 5eddd0e5..94b32f17 100644 --- a/docs/writing-plugins.md +++ b/docs/writing-plugins.md @@ -311,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 @@ -391,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 diff --git a/lib/link/connectors.js b/lib/link/connectors.js index ce109410..d48b1a58 100644 --- a/lib/link/connectors.js +++ b/lib/link/connectors.js @@ -155,13 +155,6 @@ class WebSocketClientConnector extends WebSocketBaseConnector { this._reconnectDelay = reconnectDelay; this._timeout = 15 * 60; this._startedReconnect = null; - - /** - * Mapping of plugin name to plugin version loaded on the master - * server. Is `null` before handshake has taken place. - * @member {?Object} - */ - this.plugins = null; } _check(expectedStates) { @@ -363,7 +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.plugins = data.plugins; + this.emit("hello", data); if (this._sessionToken) { console.log("SOCKET | Attempting resume"); this.sendHandshake("resume", { diff --git a/lib/link/messages.js b/lib/link/messages.js index fdcefc3d..547d7470 100644 --- a/lib/link/messages.js +++ b/lib/link/messages.js @@ -3,15 +3,36 @@ 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 @@ -37,6 +58,7 @@ class Request { constructor({ type, links, forwardTo = null, requestProperties = {}, responseProperties = {} }) { + super(); this.type = type; this.links = links; this.forwardTo = forwardTo; @@ -362,9 +384,10 @@ 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 @@ -390,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; @@ -509,6 +533,16 @@ messages.debugWsMessage = new Event({ }, }); +messages.instanceInitialized = new Event({ + type: "instance_initialized", + links: ["instance-slave"], + eventProperties: { + "plugins": { + type: "object", + additionalProperties: { type: "string" }, + } + }, +}); messages.instanceStopped = new Event({ type: 'instance_stopped', links: ['instance-slave'], @@ -583,6 +617,7 @@ function attachAllMessages(link) { module.exports = { + Message, Request, Event, diff --git a/lib/plugin.js b/lib/plugin.js index 3b6a5440..ed907ce5 100644 --- a/lib/plugin.js +++ b/lib/plugin.js @@ -242,7 +242,10 @@ class BaseMasterPlugin { */ broadcastEventToSlaves(event, data={}) { for (let slaveConnection of this.master.slaveConnections.values()) { - if (!slaveConnection.connector.closing) { + if ( + !slaveConnection.connector.closing + && (!event.plugin || slaveConnection.plugins.has(event.plugin)) + ) { event.send(slaveConnection, data); } } @@ -300,6 +303,10 @@ async function loadPluginInfos(baseDir) { 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/master.js b/master.js index 2617cf59..0496d0d8 100644 --- a/master.js +++ b/master.js @@ -377,6 +377,9 @@ 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); } @@ -390,15 +393,17 @@ class BaseConnection extends link.Link { 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 || slaveConnection.connector.closing) { - 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); } @@ -600,14 +605,14 @@ class SlaveConnection extends BaseConnection { this._id = registerData.id; this._name = registerData.name; this._version = registerData.version; - this._plugins = registerData.plugins; + 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: this._plugins, + plugins: registerData.plugins, }); this.connector.on("close", () => { diff --git a/slave.js b/slave.js index b2ad1c7a..f4ee2692 100644 --- a/slave.js +++ b/slave.js @@ -141,14 +141,14 @@ class Instance extends link.Link{ console.log(`Clusterio | Loaded plugin ${pluginInfo.name} in ${Date.now() - pluginLoadStarted}ms`); } - async init(pluginInfos, masterPlugins, slave) { + async init(pluginInfos, slave) { await this.server.init(); // load plugins for (let pluginInfo of pluginInfos) { if ( !pluginInfo.instanceEntrypoint - || !Object.hasOwnProperty.call(masterPlugins, pluginInfo.name) + || !slave.serverPlugins.has(pluginInfo.name) || !this.config.group(pluginInfo.name).get("enabled") ) { continue; @@ -161,6 +161,12 @@ class Instance extends link.Link{ throw err; } } + + let plugins = {} + for (let [name, plugin] of this.plugins) { + plugins[name] = plugin.info.version; + } + link.messages.instanceInitialized.send(this, { plugins }); } /** @@ -507,6 +513,7 @@ class InstanceConnection extends link.Link { super('slave', 'instance', connector); this.slave = slave; this.instanceId = instanceId; + this.plugins = new Map(); link.attachAllMessages(this); for (let pluginInfo of slave.pluginInfos) { @@ -526,6 +533,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); } @@ -536,14 +544,18 @@ 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 instanceInitializedEventHandler(message) { + this.plugins = new Map(Object.entries(message.data.plugins)); + } + async instanceStoppedEventHandler(message) { this.slave.instanceConnections.delete(this.instanceId); } @@ -596,8 +608,14 @@ 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("close", () => { if (this._shuttingDown) { return; @@ -663,6 +681,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); } @@ -672,12 +694,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); } } @@ -736,7 +761,7 @@ class Slave extends link.Link { let instance = new Instance( connectionClient, instanceInfo.path, this.config.get("slave.factorio_directory"), instanceInfo.config ); - await instance.init(this.pluginInfos, this.connector.plugins, this); + await instance.init(this.pluginInfos, this); // XXX: race condition on multiple simultanious calls this.instanceConnections.set(instanceId, instanceConnection); diff --git a/test/lib/plugin.js b/test/lib/plugin.js index 4798f1a4..d7e34a90 100644 --- a/test/lib/plugin.js +++ b/test/lib/plugin.js @@ -96,19 +96,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:"') ); }); });