Merge pull request #382 from Hornwitser/slave-updates

Add slave update subscriptions

packages/ctl/ctl.js

@@ -918,6 +918,8 @@ class Control extends libLink.Link {
 		this.keepOpen = false;
 	}
 
+	async slaveUpdateEventHandler() { }
+
 	async instanceUpdateEventHandler() { }
 
 	async saveListUpdateEventHandler() { }

packages/lib/link/messages.js

@@ -331,6 +331,14 @@ messages.setMasterConfigProp = new Request({
 	},
 });
 
+let slaveProperties = {
+	"agent": { type: "string" },
+	"version": { type: "string" },
+	"name": { type: "string" },
+	"id": { type: "integer" },
+	"connected": { type: "boolean" },
+};
+
 messages.listSlaves = new Request({
 	type: "list_slaves",
 	links: ["control-master"],
@@ -341,18 +349,25 @@ messages.listSlaves = new Request({
 			items: {
 				additionalProperties: false,
 				required: ["agent", "version", "name", "id", "connected"],
-				properties: {
-					"agent": { type: "string" },
-					"version": { type: "string" },
-					"name": { type: "string" },
-					"id": { type: "integer" },
-					"connected": { type: "boolean" },
-				},
+				properties: slaveProperties,
 			},
 		},
 	},
 });
 
+messages.setSlaveSubscriptions = new Request({
+	type: "set_slave_subscriptions",
+	links: ["control-master"],
+	permission: "core.slave.subscribe",
+	requestProperties: {
+		"all": { type: "boolean" },
+		"slave_ids": {
+			type: "array",
+			items: { type: "integer" },
+		},
+	},
+});
+
 let instanceProperties = {
 	"name": { type: "string" },
 	"id": { type: "integer" },
@@ -1083,6 +1098,13 @@ messages.logMessage = new Event({
 	},
 });
 
+messages.slaveUpdate = new Event({
+	type: "slave_update",
+	links: ["master-control"],
+	eventProperties: slaveProperties,
+});
+
+
 messages.instanceInitialized = new Event({
 	type: "instance_initialized",
 	links: ["instance-slave"],

packages/lib/users.js

@@ -322,6 +322,12 @@ definePermission({
 	description: "Get the full list of slaves in the cluster.",
 	grantByDefault: true,
 });
+definePermission({
+	name: "core.slave.subscribe",
+	title: "Subscribe to slave updates",
+	description: "Subscribe to be notified on updates on the details of slaves.",
+	grantByDefault: true,
+});
 definePermission({
 	name: "core.slave.generate_token",
 	title: "Generate slave token",

packages/master/src/ControlConnection.js

@@ -39,6 +39,11 @@ class ControlConnection extends BaseConnection {
 		 */
 		this.user = user;
 
+		this.slaveSubscriptions = {
+			all: false,
+			slave_ids: [],
+		};
+
 		this.instanceSubscriptions = {
 			all: false,
 			instance_ids: [],
@@ -107,6 +112,19 @@ class ControlConnection extends BaseConnection {
 		return { list };
 	}
 
+	async setSlaveSubscriptionsRequestHandler(message) {
+		this.slaveSubscriptions = message.data;
+	}
+
+	slaveUpdated(slave, update) {
+		if (
+			this.slaveSubscriptions.all
+			|| this.slaveSubscriptions.slave_ids.includes(slave.id)
+		) {
+			libLink.messages.slaveUpdate.send(this, update);
+		}
+	}
+
 	generateSlaveToken(slaveId) {
 		return jwt.sign({ aud: "slave", slave: slaveId }, this._master.config.get("master.auth_secret"));
 	}

packages/master/src/Master.js

@@ -360,6 +360,24 @@ class Master {
 		}
 	}
 
+	slaveUpdated(slave) {
+		let update = {
+			agent: slave.agent,
+			version: slave.version,
+			id: slave.id,
+			name: slave.name,
+			connected: this.wsServer.slaveConnections.has(slave.id),
+		};
+
+		for (let controlConnection of this.wsServer.controlConnections) {
+			if (controlConnection.connector.closing) {
+				continue;
+			}
+
+			controlConnection.slaveUpdated(slave, update);
+		}
+	}
+
 	addInstanceHooks(instance) {
 		instance.config.on("fieldChanged", (group, field, prev) => {
 			if (group.name === "instance" && field === "name") {

packages/master/src/WsServer.js

@@ -274,9 +274,11 @@ ${err.stack}`
 			connector.on("close", () => {
 				if (this.slaveConnections.get(data.id) === connection) {
 					this.slaveConnections.delete(data.id);
+					this.master.slaveUpdated(this.master.slaves.get(data.id));
 				}
 			});
 			this.slaveConnections.set(data.id, connection);
+			this.master.slaveUpdated(this.master.slaves.get(data.id));
 
 
 		} else if (type === "register_control") {

packages/web_ui/src/components/AssignInstanceModal.jsx

@@ -5,24 +5,20 @@ import libLink from "@clusterio/lib/link";
 
 import ControlContext from "./ControlContext";
 import { notifyErrorHandler } from "../util/notify";
+import { useSlaveList } from "../model/slave";
 
 const { Paragraph } = Typography;
 
 
 export default function AssignInstanceModal(props) {
 	let [visible, setVisible] = useState(false);
-	let [slaves, setSlaves] = useState([]);
-	let [loading, setLoading] = useState(true);
+	let [slaveList] = useSlaveList();
 	let [applying, setApplying] = useState(false);
 	let [form] = Form.useForm();
 	let control = useContext(ControlContext);
 
 	function open() {
 		setVisible(true);
-		libLink.messages.listSlaves.send(control).then((result) => {
-			setSlaves(result.list);
-			setLoading(false);
-		});
 	}
 
 	function handleAssign() {
@@ -73,8 +69,8 @@ export default function AssignInstanceModal(props) {
 			</Paragraph>
 			<Form form={form} initialValues={{ slave: props.slaveId }}>
 				<Form.Item name="slave" label="Slave">
-					<Select loading={loading}>
-						{slaves.map((slave) => <Select.Option
+					<Select>
+						{slaveList.map((slave) => <Select.Option
 							key={slave["id"]}
 							value={slave["id"]}
 							disabled={!slave["connected"]}

packages/web_ui/src/components/SlavesPage.jsx

@@ -1,28 +1,42 @@
 import React, { useContext } from "react";
+import { Table } from "antd";
 
 import libLink from "@clusterio/lib/link";
 
-import DataTable from "./data-table";
 import ControlContext from "./ControlContext";
 import PageLayout from "./PageLayout";
+import { useSlaveList } from "../model/slave";
 
 
 export default function SlavesPage() {
 	let control = useContext(ControlContext);
-
-	async function listSlaves() {
-		let result = await libLink.messages.listSlaves.send(control);
-		return result["list"].map((item) => ({
-			key: item["id"],
-			"Name": item["name"],
-			"Agent": item["agent"],
-			"Version": item["version"],
-			"Connected": item["connected"] && "Yes",
-		}));
-	}
+	let [slaveList] = useSlaveList();
 
 	return <PageLayout nav={[{ name: "Slaves" }]}>
 		<h2>Slaves</h2>
-		<DataTable DataFunction={listSlaves} />
+		<Table
+			columns={[
+				{
+					title: "Name",
+					dataIndex: "name",
+				},
+				{
+					title: "Agent",
+					dataIndex: "agent",
+				},
+				{
+					title: "Version",
+					dataIndex: "version",
+				},
+				{
+					title: "Connected",
+					key: "connected",
+					render: slave => slave["connected"] && "Yes",
+				},
+			]}
+			dataSource={slaveList}
+			rowKey={slave => slave["id"]}
+			pagination={false}
+		/>
 	</PageLayout>;
 };

packages/web_ui/src/model/slave.jsx

@@ -27,7 +27,52 @@ export function useSlave(id) {
 
 	useEffect(() => {
 		updateSlave();
+
+		function updateHandler(newSlave) {
+			setSlave({ ...newSlave, present: true });
+		}
+
+		control.onSlaveUpdate(id, updateHandler);
+		return () => {
+			control.offSlaveUpdate(id, updateHandler);
+		};
 	}, [id]);
 
 	return [slave, updateSlave];
 }
+
+export function useSlaveList() {
+	let control = useContext(ControlContext);
+	let [slaveList, setSlaveList] = useState([]);
+
+	function updateSlaveList() {
+		libLink.messages.listSlaves.send(control).then(result => {
+			setSlaveList(result.list);
+		});
+	}
+
+	useEffect(() => {
+		updateSlaveList();
+
+		function updateHandler(newSlave) {
+			setSlaveList(oldList => {
+				let newList = oldList.concat();
+				let index = newList.findIndex(s => s.id === newSlave.id);
+				if (index !== -1) {
+					newList[index] = newSlave;
+				} else {
+					newList.push(newSlave);
+				}
+				return newList;
+			});
+		}
+
+		control.onSlaveUpdate(null, updateHandler);
+		return () => {
+			control.offSlaveUpdate(null, updateHandler);
+		};
+	}, []);
+
+
+	return [slaveList];
+}

packages/web_ui/src/util/websocket.jsx

@@ -62,12 +62,16 @@ export class Control extends libLink.Link {
 		 */
 		this.accountName = null;
 
+		this.slaveUpdateHandlers = new Map();
 		this.instanceUpdateHandlers = new Map();
 		this.saveListUpdateHandlers = new Map();
 		this.instanceLogHandlers = new Map();
 
 		this.connector.on("connect", data => {
 			this.accountName = data.account.name;
+			this.updateSlaveSubscriptions().catch(err => logger.error(
+				`Unexpected error updating slave subscriptions:\n${err.stack}`
+			));
 			this.updateInstanceSubscriptions().catch(err => logger.error(
 				`Unexpected error updating instance subscriptions:\n${err.stack}`
 			));
@@ -92,6 +96,63 @@ export class Control extends libLink.Link {
 		}
 	}
 
+	async slaveUpdateEventHandler(message) {
+		let handlers = [].concat(
+			this.slaveUpdateHandlers.get(null) || [],
+			this.slaveUpdateHandlers.get(message.data.id) || [],
+		);
+		for (let handler of handlers) {
+			handler(message.data);
+		}
+	}
+
+	async onSlaveUpdate(id, handler) {
+		if (id !== null && !Number.isInteger(id)) {
+			throw new Error("Invalid slave id");
+		}
+
+		let handlers = this.slaveUpdateHandlers.get(id);
+		if (!handlers) {
+			handlers = [];
+			this.slaveUpdateHandlers.set(id, handlers);
+		}
+
+		handlers.push(handler);
+
+		if (handlers.length === 1) {
+			await this.updateSlaveSubscriptions();
+		}
+	}
+
+	async offSlaveUpdate(id, handler) {
+		let handlers = this.slaveUpdateHandlers.get(id);
+		if (!handlers || !handlers.length) {
+			throw new Error(`No handlers for slave ${id} exists`);
+		}
+
+		let index = handlers.lastIndexOf(handler);
+		if (index === -1) {
+			throw new Error(`Given handler is not registered for slave ${id}`);
+		}
+
+		handlers.slice(index, 1);
+		if (!handlers.length) {
+			this.slaveUpdateHandlers.delete(id);
+			await this.updateSlaveSubscriptions();
+		}
+	}
+
+	async updateSlaveSubscriptions() {
+		if (!this.connector.connected) {
+			return;
+		}
+
+		await libLink.messages.setSlaveSubscriptions.send(this, {
+			all: this.slaveUpdateHandlers.has(null),
+			slave_ids: [...this.slaveUpdateHandlers.keys()].filter(e => e !== null),
+		});
+	}
+
 	async instanceUpdateEventHandler(message) {
 		let handlers = this.instanceUpdateHandlers.get(message.data.id);
 		for (let handler of handlers || []) {

test/integration/clusterio.js

@@ -10,7 +10,7 @@ const libUsers = require("@clusterio/lib/users");
 const testStrings = require("../lib/factorio/test_strings");
 const {
 	TestControl, TestControlConnector, url, controlToken, slowTest,
-	get, exec, execCtl, sendRcon, getControl, instancesDir,
+	get, exec, execCtl, sendRcon, getControl, spawn, instancesDir,
 } = require("./index");
 
 
@@ -62,6 +62,59 @@ describe("Integration of Clusterio", function() {
 		});
 	});
 
+	describe("clusterioslave", function() {
+		describe("slaveUpdateEventHandler()", function() {
+			it("should trigger when a new slave is added", async function() {
+				slowTest(this);
+				getControl().slaveUpdates = [];
+				let config = "alt-slave-config.json";
+				let configPath = path.join("temp", "test", config);
+				await fs.remove(configPath);
+				await execCtl(`slave create-config --id 5 --name alt-slave --generate-token --output ${config}`);
+				await exec(
+					`node ../../packages/slave --config ${config} config set slave.tls_ca ../../test/file/tls/cert.pem`
+				);
+
+				let slaveProcess;
+				try {
+					slaveProcess = await spawn(
+						"alt-slave:", `node ../../packages/slave run --config ${config}`, /SOCKET \| registering slave/
+					);
+				} finally {
+					if (slaveProcess) {
+						slaveProcess.kill("SIGINT");
+						await events.once(slaveProcess, "exit");
+					}
+				}
+
+				let sawUpdate = false;
+				let sawConnected = false;
+				let sawDisconnected = false;
+
+				for (let update of getControl().slaveUpdates) {
+					if (update.name !== "alt-slave") {
+						continue;
+					}
+
+					sawUpdate = true;
+					if (update.connected) {
+						sawConnected = true;
+					} else {
+						sawDisconnected = true;
+					}
+				}
+
+				assert(sawUpdate, "No slave update was sent");
+				assert(sawConnected, "No slave update with status connected was sent");
+				assert(sawDisconnected, "No slave update with status disconnected was sent");
+
+				let result = await libLink.messages.listSlaves.send(getControl());
+				let slaves = new Map(result.list.map(instance => [instance.id, instance]));
+				assert(slaves.has(5), "Slave list was not updated");
+			});
+		});
+	});
+
 	describe("clusterioctl", function() {
 		describe("master config list", function() {
 			it("runs", async function() {

test/integration/index.js

@@ -25,10 +25,14 @@ class TestControl extends libLink.Link {
 	constructor(connector) {
 		super("control", "master", connector);
 		libLink.attachAllMessages(this);
+		this.slaveUpdates = [];
 		this.instanceUpdates = [];
 		this.saveListUpdates = [];
 
 		this.connector.on("connect", () => {
+			libLink.messages.setSlaveSubscriptions.send(
+				this, { all: true, slave_ids: [] }
+			).catch(err => logger.error(`Error setting slave subscriptions:\n${err.stack}`));
 			libLink.messages.setInstanceSubscriptions.send(
 				this, { all: true, instance_ids: [] }
 			).catch(err => logger.error(`Error setting instance subscriptions:\n${err.stack}`));
@@ -45,6 +49,10 @@ class TestControl extends libLink.Link {
 
 	async debugWsMessageEventHandler() { }
 
+	async slaveUpdateEventHandler(message) {
+		this.slaveUpdates.push(message.data);
+	}
+
 	async instanceUpdateEventHandler(message) {
 		this.instanceUpdates.push(message.data);
 	}
@@ -223,6 +231,7 @@ module.exports = {
 	execCtl,
 	sendRcon,
 	getControl,
+	spawn,
 
 	url,
 	controlToken,