Add instance update subscription

Provide the control link with the ability to subscribe to updates when
instance status and properties changes.  This allows the web interface
to respond in real time to instance starts and stops that are triggered
from somewhere else.

packages/ctl/ctl.js

@@ -748,6 +748,8 @@ class Control extends libLink.Link {
 		this.keepOpen = false;
 	}
 
+	async instanceUpdateEventHandler() { }
+
 	async setLogSubscriptions({
 		all = false,
 		master = false,

packages/lib/link/messages.js

@@ -353,6 +353,26 @@ messages.listSlaves = new Request({
 	},
 });
 
+let instanceProperties = {
+	"name": { type: "string" },
+	"id": { type: "integer" },
+	"assigned_slave": { type: ["null", "integer"] },
+	"status": { enum: [
+		"unknown", "unassigned", "stopped", "starting", "running", "stopping",
+		"creating_save", "exporting_data",
+	]},
+};
+
+messages.getInstance = new Request({
+	type: "get_instance",
+	links: ["control-master"],
+	permission: "core.instance.get",
+	requestProperties: {
+		"id": { type: "integer" },
+	},
+	responseProperties: instanceProperties,
+});
+
 messages.listInstances = new Request({
 	type: "list_instances",
 	links: ["control-master"],
@@ -363,20 +383,25 @@ messages.listInstances = new Request({
 			items: {
 				additionalProperties: false,
 				required: ["name", "id", "assigned_slave", "status"],
-				properties: {
-					"name": { type: "string" },
-					"id": { type: "integer" },
-					"assigned_slave": { type: ["null", "integer"] },
-					"status": { enum: [
-						"unknown", "unassigned", "stopped", "starting", "running", "stopping",
-						"creating_save", "exporting_data",
-					]},
-				},
+				properties: instanceProperties,
 			},
 		},
 	},
 });
 
+messages.setInstanceSubscriptions = new Request({
+	type: "set_instance_subscriptions",
+	links: ["control-master"],
+	permission: "core.instance.subscribe",
+	requestProperties: {
+		"all": { type: "boolean" },
+		"instance_ids": {
+			type: "array",
+			items: { type: "integer" },
+		},
+	},
+});
+
 messages.generateSlaveToken = new Request({
 	type: "generate_slave_token",
 	links: ["control-master"],
@@ -985,6 +1010,12 @@ messages.instanceStatusChanged = new Event({
 	},
 });
 
+messages.instanceUpdate = new Event({
+	type: "instance_update",
+	links: ["master-control"],
+	eventProperties: instanceProperties,
+});
+
 messages.masterConnectionEvent = new Event({
 	type: "master_connection_event",
 	links: ["slave-instance"],

packages/lib/users.js

@@ -333,12 +333,24 @@ definePermission({
 	description: "Create slave configs via the master server.",
 });
 
+definePermission({
+	name: "core.instance.get",
+	title: "Get instance",
+	description: "Get the details of an instance in the cluster.",
+	grantByDefault: true,
+});
 definePermission({
 	name: "core.instance.list",
 	title: "List instances",
 	description: "Get the full list of instances in the cluster.",
 	grantByDefault: true,
 });
+definePermission({
+	name: "core.instance.subscribe",
+	title: "Subscribe to instance updates",
+	description: "Subscribe to be notified on updates on the details and status of instances.",
+	grantByDefault: true,
+});
 definePermission({
 	name: "core.instance.create",
 	title: "Create instance",

packages/master/src/ControlConnection.js

@@ -38,6 +38,11 @@ class ControlConnection extends BaseConnection {
 		 */
 		this.user = user;
 
+		this.instanceSubscriptions = {
+			all: false,
+			instance_ids: [],
+		};
+
 		this.logTransport = null;
 		this.logSubscriptions = {
 			all: false,
@@ -122,6 +127,21 @@ class ControlConnection extends BaseConnection {
 		return { serialized_config: slaveConfig.serialize() };
 	}
 
+	async getInstanceRequestHandler(message) {
+		let id = message.data.id;
+		let instance = this._master.instances.get(id);
+		if (!instance) {
+			throw new libErrors.RequestError(`Instance with ID ${id} does not exist`);
+		}
+
+		return {
+			id,
+			name: instance.config.get("instance.name"),
+			assigned_slave: instance.config.get("instance.assigned_slave"),
+			status: instance.status,
+		};
+	}
+
 	async listInstancesRequestHandler(message) {
 		let list = [];
 		for (let instance of this._master.instances.values()) {
@@ -135,6 +155,24 @@ class ControlConnection extends BaseConnection {
 		return { list };
 	}
 
+	async setInstanceSubscriptionsRequestHandler(message) {
+		this.instanceSubscriptions = message.data;
+	}
+
+	instanceUpdated(instance) {
+		if (
+			this.instanceSubscriptions.all
+			|| this.instanceSubscriptions.instance_ids.includes(instance.config.get("instance.id"))
+		) {
+			libLink.messages.instanceUpdate.send(this, {
+				id: instance.config.get("instance.id"),
+				name: instance.config.get("instance.name"),
+				assigned_slave: instance.config.get("instance.assigned_slave"),
+				status: instance.status,
+			});
+		}
+	}
+
 	// XXX should probably add a hook for slave reuqests?
 	async createInstanceRequestHandler(message) {
 		let instanceConfig = new libConfig.InstanceConfig("master");
@@ -174,7 +212,7 @@ class ControlConnection extends BaseConnection {
 		let instance = { config: instanceConfig, status: "unassigned" };
 		this._master.instances.set(instanceId, instance);
 		await libPlugin.invokeHook(this._master.plugins, "onInstanceStatusChanged", instance, null);
-		this._master.addInstancePluginHooks(instance);
+		this._master.addInstanceHooks(instance);
 	}
 
 	async deleteInstanceRequestHandler(message, request) {
@@ -190,6 +228,7 @@ class ControlConnection extends BaseConnection {
 
 		let prev = instance.status;
 		instance.status = "deleted";
+		this.instanceUpdated(instance);
 		await libPlugin.invokeHook(this._master.plugins, "onInstanceStatusChanged", instance, prev);
 	}
 

packages/master/src/Master.js

@@ -152,7 +152,7 @@ class Master {
 		await this.userManager.load(path.join(databaseDirectory, "users.json"));
 
 		for (let instance of this.instances.values()) {
-			this.addInstancePluginHooks(instance);
+			this.addInstanceHooks(instance);
 		}
 
 		// Make sure we're actually going to listen on a port
@@ -361,10 +361,26 @@ class Master {
 		}
 	}
 
-	addInstancePluginHooks(instance) {
+	addInstanceHooks(instance) {
 		instance.config.on("fieldChanged", (group, field, prev) => {
+			if (group.name === "instance" && field === "name") {
+				this.instanceUpdated(instance);
+			}
+
 			libPlugin.invokeHook(this.plugins, "onInstanceConfigFieldChanged", instance, group, field, prev);
 		});
+
+		this.instanceUpdated(instance);
+	}
+
+	instanceUpdated(instance) {
+		for (let controlConnection of this.wsServer.controlConnections) {
+			if (controlConnection.connector.closing) {
+				continue;
+			}
+
+			controlConnection.instanceUpdated(instance);
+		}
 	}
 
 	async loadPlugins() {

packages/master/src/SlaveConnection.js

@@ -65,6 +65,7 @@ class SlaveConnection extends BaseConnection {
 		let prev = instance.status;
 		instance.status = message.data.status;
 		logger.verbose(`Instance ${instance.config.get("instance.name")} State: ${instance.status}`);
+		this._master.instanceUpdated(instance);
 		await libPlugin.invokeHook(this._master.plugins, "onInstanceStatusChanged", instance, prev);
 	}
 
@@ -99,6 +100,7 @@ class SlaveConnection extends BaseConnection {
 					let prev = masterInstance.status;
 					masterInstance.status = instanceData.status;
 					logger.verbose(`Instance ${instanceConfig.get("instance.name")} State: ${instanceData.status}`);
+					this._master.instanceUpdated(instance);
 					await libPlugin.invokeHook(this._master.plugins, "onInstanceStatusChanged", masterInstance, prev);
 				}
 				continue;
@@ -107,7 +109,7 @@ class SlaveConnection extends BaseConnection {
 			instanceConfig.set("instance.assigned_slave", this._id);
 			let newInstance = { config: instanceConfig, status: instanceData.status };
 			this._master.instances.set(instanceConfig.get("instance.id"), newInstance);
-			this._master.addInstancePluginHooks(newInstance);
+			this._master.addInstanceHooks(newInstance);
 			await libLink.messages.assignInstance.send(this, {
 				instance_id: instanceConfig.get("instance.id"),
 				serialized_config: instanceConfig.serialize("slave"),

packages/web_ui/src/components/InstanceViewPage.jsx

@@ -26,7 +26,7 @@ export default function InstanceViewPage(props) {
 	let history = useHistory();
 
 	let control = useContext(ControlContext);
-	let [instance, updateInstance] = useInstance(instanceId);
+	let [instance] = useInstance(instanceId);
 	let [slave] = useSlave(Number(instance["assigned_slave"]));
 
 	let [creatingSave, setCreatingSave] = useState(false);
@@ -44,10 +44,7 @@ export default function InstanceViewPage(props) {
 	}
 
 	let instanceButtons = <Space>
-		<StartStopInstanceButton
-			instance={instance}
-			onFinish={updateInstance}
-		/>
+		<StartStopInstanceButton instance={instance} />
 		{instance.status === "stopped" && <Button
 			loading={creatingSave}
 			onClick={() => {
@@ -55,7 +52,6 @@ export default function InstanceViewPage(props) {
 				libLink.messages.createSave.send(
 					control, { instance_id: instanceId }
 				).catch(notifyErrorHandler("Error creating save")).finally(() => {
-					updateInstance();
 					setCreatingSave(false);
 				});
 			}}
@@ -107,7 +103,6 @@ export default function InstanceViewPage(props) {
 						disabled: !["unknown", "unassigned", "stopped"].includes(instance["status"]),
 					}}
 					buttonContent={assigned ? "Reassign" : "Assign"}
-					onFinish={updateInstance}
 				/>
 			</Descriptions.Item>
 			<Descriptions.Item label="Status">{instance["status"]}</Descriptions.Item>
@@ -117,6 +112,6 @@ export default function InstanceViewPage(props) {
 		<InstanceConsole id={instanceId} />
 		<InstanceRcon id={instanceId} disabled={instance["status"] !== "running"} />
 
-		<InstanceConfigTree id={instanceId} onApply={updateInstance} />
+		<InstanceConfigTree id={instanceId} />
 	</PageLayout>;
 }

packages/web_ui/src/model/instance.jsx

@@ -2,6 +2,7 @@ import React, { useEffect, useContext, useState } from "react";
 import ControlContext from "../components/ControlContext";
 
 import libLink from "@clusterio/lib/link";
+import { logger } from "@clusterio/lib/logging";
 
 
 export function useInstance(id) {
@@ -14,19 +15,26 @@ export function useInstance(id) {
 			return;
 		}
 
-		// XXX optimize by requesting only the instance in question
-		libLink.messages.listInstances.send(control).then(result => {
-			let match = result.list.find(i => i.id === id);
-			if (!match) {
-				setInstance({ missing: true });
-			} else {
-				setInstance({ ...match, present: true });
-			}
+		libLink.messages.getInstance.send(control, { id }).then(result => {
+			setInstance({ ...result, present: true });
+		}).catch(err => {
+			logger.log(`Failed to get instance: ${err}`);
+			setInstance({ missing: true });
 		});
+
 	}
 
 	useEffect(() => {
 		updateInstance();
+
+		function updateHandler(newInstance) {
+			setInstance({ ...newInstance, present: true });
+		}
+
+		control.onInstanceUpdate(id, updateHandler);
+		return () => {
+			control.offInstanceUpdate(id, updateHandler);
+		};
 	}, [id]);
 
 	return [instance, updateInstance];

packages/web_ui/src/util/websocket.jsx

@@ -62,10 +62,14 @@ export class Control extends libLink.Link {
 		 */
 		this.accountName = null;
 
+		this.instanceUpdateHandlers = new Map();
 		this.instanceLogHandlers = new Map();
 
 		this.connector.on("connect", data => {
 			this.accountName = data.account.name;
+			this.updateInstanceSubscriptions().catch(err => logger.error(
+				`Unexpected error updating instance subscriptions:\n${err.stack}`
+			));
 			this.updateLogSubscriptions().catch(err => logger.error(
 				`Unexpected error updating log subscriptions:\n${err.stack}`
 			));
@@ -84,6 +88,60 @@ export class Control extends libLink.Link {
 		}
 	}
 
+	async instanceUpdateEventHandler(message) {
+		let handlers = this.instanceUpdateHandlers.get(message.data.id);
+		for (let handler of handlers || []) {
+			handler(message.data);
+		}
+	};
+
+	async onInstanceUpdate(id, handler) {
+		if (!Number.isInteger(id)) {
+			throw new Error("Invalid instance id");
+		}
+
+		let handlers = this.instanceUpdateHandlers.get(id);
+		if (!handlers) {
+			handlers = [];
+			this.instanceUpdateHandlers.set(id, handlers);
+		}
+
+		handlers.push(handler);
+
+		if (handlers.length === 1) {
+			await this.updateInstanceSubscriptions();
+		}
+	}
+
+	async offInstanceUpdate(id, handler) {
+		let handlers = this.instanceUpdateHandlers.get(id);
+		if (!handlers || !handlers.length) {
+			throw new Error(`No handlers for instance ${id} exist`);
+		}
+
+		let index = handlers.lastIndexOf(handler);
+		if (index === -1) {
+			throw new Error(`Given handler is not registered for instance ${id}`);
+		}
+
+		handlers.splice(index, 1);
+		if (!handlers.length) {
+			this.instanceUpdateHandlers.delete(id);
+			await this.updateInstanceSubscriptions();
+		}
+	}
+
+	async updateInstanceSubscriptions() {
+		if (!this.connector.connected) {
+			return;
+		}
+
+		await libLink.messages.setInstanceSubscriptions.send(this, {
+			all: false,
+			instance_ids: [...this.instanceUpdateHandlers.keys()],
+		});
+	}
+
 	async logMessageEventHandler(message) {
 		let info = message.data.info;
 

test/integration/clusterio.js

@@ -325,6 +325,22 @@ describe("Integration of Clusterio", function() {
 			});
 		});
 
+		describe("instanceUpdateEventHandler()", function() {
+			it("should have triggered for the previous instance status updates", function() {
+				let statusesToCheck = new Set(
+					["unassigned", "stopped", "creating_save", "exporting_data", "starting", "running", "stopping"]
+				);
+				let statusesNotSeen = new Set(statusesToCheck);
+
+				for (let update of getControl().instanceUpdates) {
+					assert(statusesToCheck.has(update.status), `Missing check for status ${update.status}`);
+					statusesNotSeen.delete(update.status);
+				}
+
+				assert(statusesNotSeen.size === 0, `Did not see the statuses ${[...statusesNotSeen]}`);
+			});
+		});
+
 		describe("permission list", function() {
 			it("runs", async function() {
 				await execCtl("permission list");

test/integration/index.js

@@ -25,6 +25,13 @@ class TestControl extends libLink.Link {
 	constructor(connector) {
 		super("control", "master", connector);
 		libLink.attachAllMessages(this);
+		this.instanceUpdates = [];
+
+		this.connector.on("connect", () => {
+			libLink.messages.setInstanceSubscriptions.send(
+				this, { all: true, instance_ids: [] }
+			).catch(err => logger.error(`Error setting instance subscriptions:\n${err.stack}`));
+		});
 	}
 
 	async prepareDisconnectRequestHandler(message, request) {
@@ -34,6 +41,10 @@ class TestControl extends libLink.Link {
 
 	async debugWsMessageEventHandler() { }
 
+	async instanceUpdateEventHandler(message) {
+		this.instanceUpdates.push(message.data);
+	}
+
 	async logMessageEventHandler() { }
 }