Merge pull request #510 from Cooldude2606/refactor/subscriptions

Refactor/subscriptions - Create a generalized event subscription class

packages/controller/src/ControlConnection.ts

@@ -11,8 +11,6 @@ const { logFilter, logger } = lib;
 import BaseConnection from "./BaseConnection";
 import * as routes from "./routes";
 import Controller from "./Controller";
-import { promises } from "dns";
-import InstanceInfo from "./InstanceInfo";
 
 const strcmp = new Intl.Collator(undefined, { numeric: true, sensitivity: "base" }).compare;
 
@@ -30,12 +28,6 @@ const queryLogTime = new lib.Summary(
 export default class ControlConnection extends BaseConnection {
 	private _agent: string;
 	private _version: string;
-	hostSubscriptions: { all:boolean, hostIds:number[] } = { all:false, hostIds:[] };
-	instanceSubscriptions: { all:boolean, instanceIds:number[] } = { all:false, instanceIds:[] };
-	saveListSubscriptions: { all:boolean, instanceIds:number[] } = { all:false, instanceIds:[] };
-	modPackSubscriptions: { all:boolean, modPackIds:number[] }  = { all:false, modPackIds:[] };
-	modSubscriptions: { all:boolean, modNames:string[] } = { all:false, modNames:[] };
-	userSubscriptions: { all:boolean, names:string[]} = { all:false, names:[] };
 	logTransport: lib.LinkTransport | null = null;
 	logSubscriptions: {
 		all: boolean,
@@ -89,33 +81,28 @@ export default class ControlConnection extends BaseConnection {
 		this.handle(lib.ControllerConfigSetPropRequest, this.handleControllerConfigSetPropRequest.bind(this));
 		this.handle(lib.HostRevokeTokensRequest, this.handleHostRevokeTokensRequest.bind(this));
 		this.handle(lib.HostListRequest, this.handleHostListRequest.bind(this));
-		this.handle(lib.HostSetSubscriptionsRequest, this.handleHostSetSubscriptionsRequest.bind(this));
 		this.handle(lib.HostGenerateTokenRequest, this.handleHostGenerateTokenRequest.bind(this));
 		this.handle(lib.HostConfigCreateRequest, this.handleHostConfigCreateRequest.bind(this));
 		this.handle(lib.InstanceDetailsGetRequest, this.handleInstanceDetailsGetRequest.bind(this));
 		this.handle(lib.InstanceDetailsListRequest, this.handleInstanceDetailsListRequest.bind(this));
-		this.handle(lib.InstanceDetailsSetSubscriptionsRequest, this.handleInstanceDetailsSetSubscriptionsRequest.bind(this));
 		this.handle(lib.InstanceCreateRequest, this.handleInstanceCreateRequest.bind(this));
 		this.handle(lib.InstanceDeleteRequest, this.handleInstanceDeleteRequest.bind(this));
 		this.handle(lib.InstanceConfigGetRequest, this.handleInstanceConfigGetRequest.bind(this));
 		this.handle(lib.InstanceConfigSetFieldRequest, this.handleInstanceConfigSetFieldRequest.bind(this));
 		this.handle(lib.InstanceConfigSetPropRequest, this.handleInstanceConfigSetPropRequest.bind(this));
 		this.handle(lib.InstanceAssignRequest, this.handleInstanceAssignRequest.bind(this));
-		this.handle(lib.InstanceSetSaveListSubscriptionsRequest, this.handleInstanceSetSaveListSubscriptionsRequest.bind(this));
 		this.handle(lib.InstanceRenameSaveRequest, this._controller.sendToHostByInstanceId.bind(this._controller));
 		this.handle(lib.InstanceCopySaveRequest, this._controller.sendToHostByInstanceId.bind(this._controller));
 		this.handle(lib.InstanceDeleteSaveRequest, this._controller.sendToHostByInstanceId.bind(this._controller));
 		this.handle(lib.InstanceDownloadSaveRequest, this.handleInstanceDownloadSaveRequest.bind(this));
 		this.handle(lib.InstanceTransferSaveRequest, this.handleInstanceTransferSaveRequest.bind(this));
 		this.handle(lib.ModPackListRequest, this.handleModPackListRequest.bind(this));
-		this.handle(lib.ModPackSetSubscriptionsRequest, this.handleModPackSetSubscriptionsRequest.bind(this));
 		this.handle(lib.ModPackCreateRequest, this.handleModPackCreateRequest.bind(this));
 		this.handle(lib.ModPackUpdateRequest, this.handleModPackUpdateRequest.bind(this));
 		this.handle(lib.ModPackDeleteRequest, this.handleModPackDeleteRequest.bind(this));
 		this.handle(lib.ModGetRequest, this.handleModGetRequest.bind(this));
 		this.handle(lib.ModListRequest, this.handleModListRequest.bind(this));
 		this.handle(lib.ModSearchRequest, this.handleModSearchRequest.bind(this));
-		this.handle(lib.ModSetSubscriptionsRequest, this.handleModSetSubscriptionsRequest.bind(this));
 		this.handle(lib.ModDownloadRequest, this.handleModDownloadRequest.bind(this));
 		this.handle(lib.ModDeleteRequest, this.handleModDeleteRequest.bind(this));
 		this.handle(lib.LogSetSubscriptionsRequest, this.handleLogSetSubscriptionsRequest.bind(this));
@@ -128,7 +115,6 @@ export default class ControlConnection extends BaseConnection {
 		this.handle(lib.RoleDeleteRequest, this.handleRoleDeleteRequest.bind(this));
 		this.handle(lib.UserGetRequest, this.handleUserGetRequest.bind(this));
 		this.handle(lib.UserListRequest, this.handleUserListRequest.bind(this));
-		this.handle(lib.UserSetSubscriptionsRequest, this.handleUserSetSubscriptionsRequest.bind(this));
 		this.handle(lib.UserCreateRequest, this.handleUserCreateRequest.bind(this));
 		this.handle(lib.UserRevokeTokenRequest, this.handleUserRevokeTokenRequest.bind(this));
 		this.handle(lib.UserUpdateRolesRequest, this.handleUserUpdateRolesRequest.bind(this));
@@ -234,22 +220,6 @@ export default class ControlConnection extends BaseConnection {
 		return list;
 	}
 
-	async handleHostSetSubscriptionsRequest(request: lib.HostSetSubscriptionsRequest) {
-		this.hostSubscriptions = { ...request };
-	}
-
-	hostUpdated(
-		host: HostInfo,
-		update: lib.HostDetails
-	) {
-		if (
-			this.hostSubscriptions.all
-			|| this.hostSubscriptions.hostIds.includes(host.id)
-		) {
-			this.send(new lib.HostUpdateEvent(update));
-		}
-	}
-
 	async handleHostGenerateTokenRequest(message: lib.HostGenerateTokenRequest) {
 		let hostId = message.hostId;
 		if (hostId === undefined) {
@@ -322,38 +292,6 @@ export default class ControlConnection extends BaseConnection {
 		return list;
 	}
 
-	async handleInstanceDetailsSetSubscriptionsRequest(request: lib.InstanceDetailsSetSubscriptionsRequest) {
-		this.instanceSubscriptions = { ...request };
-	}
-
-	instanceUpdated(instance: InstanceInfo) {
-		if (
-			this.instanceSubscriptions.all
-			|| this.instanceSubscriptions.instanceIds.includes(instance.id)
-		) {
-			let assigned_host: number|null|undefined = instance.config.get("instance.assigned_host");
-			if (assigned_host === null) {
-				assigned_host = undefined;
-			}
-	
-			let game_port: number|null|undefined = instance.game_port
-			if (game_port === null) {
-				game_port = undefined;
-			}
-
-
-			this.send(new lib.InstanceDetailsUpdateEvent(
-				new lib.InstanceDetails(
-					instance.config.get("instance.name"),
-					instance.id,
-					assigned_host,
-					game_port,
-					instance.status,
-				)
-			));
-		}
-	}
-
 	// XXX should probably add a hook for host reuqests?
 	async handleInstanceCreateRequest(request: lib.InstanceCreateRequest) {
 		let instanceConfig = new lib.InstanceConfig("controller");
@@ -396,19 +334,6 @@ export default class ControlConnection extends BaseConnection {
 		await this._controller.instanceAssign(request.instanceId, request.hostId);
 	}
 
-	async handleInstanceSetSaveListSubscriptionsRequest(request: lib.InstanceSetSaveListSubscriptionsRequest) {
-		this.saveListSubscriptions = { ...request };
-	}
-
-	saveListUpdate(instanceId: number, saves: lib.SaveDetails[]) {
-		if (
-			this.saveListSubscriptions.all
-			|| this.saveListSubscriptions.instanceIds.includes(instanceId)
-		) {
-			this.send(new lib.InstanceSaveListUpdateEvent(instanceId, saves));
-		}
-	}
-
 	async handleInstanceDownloadSaveRequest(request: lib.InstanceDownloadSaveRequest) {
 		let { instanceId, name } = request;
 		let stream = await routes.createProxyStream(this._controller.app);
@@ -506,10 +431,6 @@ export default class ControlConnection extends BaseConnection {
 		return [...this._controller.modPacks!.values()];
 	}
 
-	async handleModPackSetSubscriptionsRequest(request: lib.ModPackSetSubscriptionsRequest) {
-		this.modPackSubscriptions = { ...request };
-	}
-
 	async handleModPackCreateRequest(request: lib.ModPackCreateRequest) {
 		let modPack = request.modPack;
 		if (modPack.id === undefined) {
@@ -542,15 +463,6 @@ export default class ControlConnection extends BaseConnection {
 		this._controller.modPackUpdated(modPack);
 	}
 
-	modPackUpdated(modPack: lib.ModPack) {
-		if (
-			this.modPackSubscriptions.all
-			|| this.modPackSubscriptions.modPackIds.includes(modPack.id!)
-		) {
-			this.send(new lib.ModPackUpdateEvent(modPack));
-		}
-	}
-
 	async handleModGetRequest(request: lib.ModGetRequest) {
 		let { name, version } = request;
 		let filename = `${name}_${version}.zip`;
@@ -654,10 +566,6 @@ export default class ControlConnection extends BaseConnection {
 		};
 	}
 
-	async handleModSetSubscriptionsRequest(request: lib.ModSetSubscriptionsRequest) {
-		this.modSubscriptions = { ...request };
-	}
-
 	async handleModDownloadRequest(request: lib.ModDownloadRequest) {
 		let { name, version } = request;
 		let filename = `${name}_${version}.zip`;
@@ -690,15 +598,6 @@ export default class ControlConnection extends BaseConnection {
 		this.updateLogSubscriptions();
 	}
 
-	modUpdated(mod: lib.ModInfo) {
-		if (
-			this.modSubscriptions.all
-			|| this.modSubscriptions.modNames.includes(mod.name)
-		) {
-			this.send(new lib.ModUpdateEvent(mod));
-		}
-	}
-
 	updateLogSubscriptions() {
 		let { all, controller, hostIds, instanceIds } = this.logSubscriptions;
 		if (all || controller || hostIds.length || instanceIds.length) {
@@ -850,32 +749,6 @@ export default class ControlConnection extends BaseConnection {
 		return list;
 	}
 
-	async handleUserSetSubscriptionsRequest(request: lib.UserSetSubscriptionsRequest) {
-		this.userSubscriptions = { ...request };
-	}
-
-	userUpdated(user: any) {
-		if (
-			this.userSubscriptions.all
-			|| this.userSubscriptions.names.includes(user.name)
-		) {
-			this.send(new lib.UserUpdateEvent(
-				new lib.RawUser(
-					user.name,
-					[...user.roles].map(role => role.id),
-					[...user.instances],
-					user.isAdmin,
-					user.isBanned,
-					user.isWhitelisted,
-					user.banReason,
-					user.isDeleted,
-					user.playerStats,
-					user.instanceStats,
-				)
-			));
-		}
-	}
-
 	async handleUserCreateRequest(request: lib.UserCreateRequest) {
 		let user = this._controller.userManager.createUser(request.name);
 		this._controller.userUpdated(user);

packages/controller/src/Controller.ts

@@ -74,6 +74,9 @@ export default class Controller {
 	wsServer: WsServer;
 	debugEvents: events.EventEmitter = new events.EventEmitter();
 	private _events: events.EventEmitter = new events.EventEmitter();
+
+	/** Event subscription controller */
+	subscriptions: lib.SubscriptionController;
 	
 	// Possible states are new, starting, running, stopping, stopped
 	private _state: string = "new";
@@ -106,6 +109,7 @@ export default class Controller {
 
 		this.wsServer = new WsServer(this);
 		this.userManager = new UserManager(this.config);
+		this.subscriptions = new lib.SubscriptionController(this);
 	}
 
 	async start(args: ControllerArgs) {
@@ -236,6 +240,14 @@ export default class Controller {
 			this.app.locals.mainBundle = manifest["main.js"] || "no_web_build";
 		}
 
+		// Handle subscriptions for all internal properties
+		this.subscriptions.handle(lib.HostUpdateEvent);
+		this.subscriptions.handle(lib.InstanceDetailsUpdateEvent);
+		this.subscriptions.handle(lib.InstanceSaveListUpdateEvent);
+		this.subscriptions.handle(lib.ModPackUpdateEvent);
+		this.subscriptions.handle(lib.ModUpdateEvent);
+		this.subscriptions.handle(lib.UserUpdateEvent);
+
 		// Load plugins
 		await this.loadPlugins();
 
@@ -576,13 +588,7 @@ export default class Controller {
 			host.token_valid_after,
 		);
 
-		for (let controlConnection of this.wsServer.controlConnections.values()) {
-			if (controlConnection.connector.closing) {
-				continue;
-			}
-
-			controlConnection.hostUpdated(host, update);
-		}
+		this.subscriptions.broadcast(new lib.HostUpdateEvent(update));
 	}
 
 	/**
@@ -762,56 +768,56 @@ export default class Controller {
 	}
 
 	instanceUpdated(instance: InstanceInfo) {
-		for (let controlConnection of this.wsServer.controlConnections.values()) {
-			if (controlConnection.connector.closing) {
-				continue;
-			}
-
-			controlConnection.instanceUpdated(instance);
+		let assigned_host: number|null|undefined = instance.config.get("instance.assigned_host");
+		if (assigned_host === null) {
+			assigned_host = undefined;
 		}
+
+		let game_port: number|null|undefined = instance.game_port
+		if (game_port === null) {
+			game_port = undefined;
+		}
+
+		this.subscriptions.broadcast(new lib.InstanceDetailsUpdateEvent(
+			new lib.InstanceDetails(
+				instance.config.get("instance.name"),
+				instance.id,
+				assigned_host,
+				game_port,
+				instance.status,
+			)
+		));
 	}
 
 	saveListUpdate(instanceId: number, saves: lib.SaveDetails[]) {
-		for (let controlConnection of this.wsServer.controlConnections.values()) {
-			if (controlConnection.connector.closing) {
-				continue;
-			}
-
-			controlConnection.saveListUpdate(instanceId, saves);
-		}
+		this.subscriptions.broadcast(new lib.InstanceSaveListUpdateEvent(instanceId, saves));
 	}
 
 	modPackUpdated(modPack: lib.ModPack) {
-		for (let controlConnection of this.wsServer.controlConnections.values()) {
-			if (controlConnection.connector.closing) {
-				continue;
-			}
-			controlConnection.modPackUpdated(modPack);
-		}
-
+		this.subscriptions.broadcast(new lib.ModPackUpdateEvent(modPack));
 		lib.invokeHook(this.plugins, "onModPackUpdated", modPack);
 	}
 
 	modUpdated(mod: lib.ModInfo) {
-		for (let controlConnection of this.wsServer.controlConnections.values()) {
-			if (controlConnection.connector.closing) {
-				continue;
-			}
-
-			controlConnection.modUpdated(mod);
-		}
-
+		this.subscriptions.broadcast(new lib.ModUpdateEvent(mod));
 		lib.invokeHook(this.plugins, "onModUpdated", mod);
 	}
 
 	userUpdated(user: lib.User) {
-		for (let controlConnection of this.wsServer.controlConnections.values()) {
-			if (controlConnection.connector.closing) {
-				continue;
-			}
-
-			controlConnection.userUpdated(user);
-		}
+		this.subscriptions.broadcast(new lib.UserUpdateEvent(
+			new lib.RawUser(
+				user.name,
+				[...user.roles].map(role => role.id),
+				[...user.instances],
+				user.isAdmin,
+				user.isBanned,
+				user.isWhitelisted,
+				user.banReason,
+				user.isDeleted,
+				user.playerStats,
+				user.instanceStats,
+			)
+		));
 	}
 
 	/**

packages/lib/browser.js

@@ -9,6 +9,7 @@ export * from "./dist/src/logging";
 export * from "./dist/src/plugin";
 export * from "./dist/src/schema";
 export * from "./dist/src/users";
+export * from "./dist/src/subscriptions";
 
 export { default as ExponentialBackoff } from "./dist/src/ExponentialBackoff";
 export { default as PlayerStats } from "./dist/src/PlayerStats";

packages/lib/index.ts

@@ -25,6 +25,7 @@ export * from "./src/shared_commands";
 export * from "./src/stream";
 export * from "./src/users";
 export * from "./src/zip_ops";
+export * from "./src/subscriptions"
 
 export { default as ExponentialBackoff } from "./src/ExponentialBackoff";
 export { default as PlayerStats } from "./src/PlayerStats";

packages/lib/src/data/messages_host.ts

@@ -36,34 +36,12 @@ export class HostListRequest {
 	static Response = jsonArray(HostDetails);
 }
 
-export class HostSetSubscriptionsRequest {
-	declare ["constructor"]: typeof HostSetSubscriptionsRequest;
-	static type = "request" as const;
-	static src = "control" as const;
-	static dst = "controller" as const;
-	static permission = "core.host.subscribe" as const;
-
-	constructor(
-		public all: boolean,
-		public hostIds: number[],
-	) { }
-
-	static jsonSchema = Type.Object({
-		"all": Type.Boolean(),
-		"hostIds": Type.Array(Type.Integer()),
-	});
-
-	static fromJSON(json: Static<typeof this.jsonSchema>) {
-		return new this(json.all, json.hostIds);
-	}
-}
-
-
 export class HostUpdateEvent {
 	declare ["constructor"]: typeof HostUpdateEvent;
 	static type = "event" as const;
 	static src = "controller" as const;
 	static dst = "control" as const;
+	static permission = "core.host.subscribe" as const;
 
 	constructor(
 		public update: HostDetails,
@@ -76,6 +54,10 @@ export class HostUpdateEvent {
 	static fromJSON(json: Static<typeof this.jsonSchema>) {
 		return new this(HostDetails.fromJSON(json.update));
 	}
+
+	get subscriptionChannel() {
+		return this.update.id;
+	}
 }
 
 export class HostMetricsRequest {

packages/lib/src/data/messages_instance.ts

@@ -67,33 +67,12 @@ export class InstanceDetailsListRequest {
 	static Response = jsonArray(InstanceDetails);
 };
 
-export class InstanceDetailsSetSubscriptionsRequest {
-	declare ["constructor"]: typeof InstanceDetailsSetSubscriptionsRequest;
-	static type = "request" as const;
-	static src = "control" as const;
-	static dst = "controller" as const;
-	static permission = "core.instance.subscribe" as const;
-
-	constructor(
-		public all: boolean = false,
-		public instanceIds: number[] = [],
-	) { }
-
-	static jsonSchema = Type.Object({
-		"all": Type.Boolean(),
-		"instanceIds": Type.Array(Type.Integer()),
-	});
-
-	static fromJSON(json: Static<typeof this.jsonSchema>) {
-		return new this(json.all, json.instanceIds);
-	}
-};
-
 export class InstanceDetailsUpdateEvent {
 	declare ["constructor"]: typeof InstanceDetailsUpdateEvent;
 	static type = "event" as const;
 	static src = "controller" as const;
 	static dst = "control" as const;
+	static permission = "core.instance.subscribe" as const;
 
 	constructor(
 		public details: InstanceDetails,
@@ -108,6 +87,10 @@ export class InstanceDetailsUpdateEvent {
 	static fromJSON(json: Static<typeof this.jsonSchema>) {
 		return new this(InstanceDetails.fromJSON(json));
 	}
+
+	get subscriptionChannel() {
+		return this.details.id;
+	}
 };
 
 export class InstanceCreateRequest {
@@ -299,33 +282,12 @@ export class InstanceListSavesRequest {
 	static Response = jsonArray(SaveDetails);
 }
 
-export class InstanceSetSaveListSubscriptionsRequest {
-	declare ["constructor"]: typeof InstanceSetSaveListSubscriptionsRequest;
-	static type = "request" as const;
-	static src = "control" as const;
-	static dst = "controller" as const;
-	static permission = "core.instance.save.list_subscribe" as const;
-
-	constructor(
-		public all: boolean = false,
-		public instanceIds: number[] = [],
-	) { }
-
-	static jsonSchema = Type.Object({
-		"all": Type.Boolean(),
-		"instanceIds": Type.Array(Type.Integer()),
-	});
-
-	static fromJSON(json: Static<typeof this.jsonSchema>) {
-		return new this(json.all, json.instanceIds);
-	}
-};
-
 export class InstanceSaveListUpdateEvent {
 	declare ["constructor"]: typeof InstanceSaveListUpdateEvent;
 	static type = "event" as const;
 	static src = ["instance", "host", "controller"] as const;
 	static dst = ["controller", "control"] as const;
+	static permission = "core.instance.save.list_subscribe" as const;
 
 	constructor(
 		public instanceId: number,
@@ -340,6 +302,10 @@ export class InstanceSaveListUpdateEvent {
 	static fromJSON(json: Static<typeof this.jsonSchema>) {
 		return new this(json.instanceId, json.saves.map(i => SaveDetails.fromJSON(i)));
 	}
+
+	get subscriptionChannel() {
+		return this.instanceId;
+	}
 }
 
 export class InstanceCreateSaveRequest {

packages/lib/src/data/messages_mod.ts

@@ -104,28 +104,6 @@ export class ModPackDeleteRequest {
 	}
 }
 
-export class ModPackSetSubscriptionsRequest {
-	declare ["constructor"]: typeof ModPackSetSubscriptionsRequest;
-	static type = "request" as const;
-	static src = "control" as const;
-	static dst = "controller" as const;
-	static permission = "core.mod_pack.subscribe" as const;
-
-	constructor(
-		public all: boolean,
-		public modPackIds: number[],
-	) { }
-
-	static jsonSchema = Type.Object({
-		"all": Type.Boolean(),
-		"modPackIds": Type.Array(Type.Integer()),
-	});
-
-	static fromJSON(json: Static<typeof this.jsonSchema>) {
-		return new this(json.all, json.modPackIds);
-	}
-}
-
 export class ModGetRequest {
 	declare ["constructor"]: typeof ModGetRequest;
 	static type = "request" as const;
@@ -217,28 +195,6 @@ export class ModSearchRequest {
 	};
 }
 
-export class ModSetSubscriptionsRequest {
-	declare ["constructor"]: typeof ModSetSubscriptionsRequest;
-	static type = "request" as const;
-	static src = "control" as const;
-	static dst = "controller" as const;
-	static permission = "core.mod.subscribe" as const;
-
-	constructor(
-		public all: boolean,
-		public modNames: string[],
-	) { }
-
-	static jsonSchema = Type.Object({
-		"all": Type.Boolean(),
-		"modNames": Type.Array(Type.String()),
-	});
-
-	static fromJSON(json: Static<typeof this.jsonSchema>) {
-		return new this(json.all, json.modNames);
-	}
-}
-
 export class ModDownloadRequest {
 	declare ["constructor"]: typeof ModDownloadRequest;
 	static type = "request" as const;
@@ -290,6 +246,7 @@ export class ModPackUpdateEvent {
 	static type = "event" as const;
 	static src = "controller" as const;
 	static dst = "control" as const;
+	static permission = "core.mod_pack.subscribe" as const;
 
 	constructor(
 		public modPack: ModPack,
@@ -302,6 +259,10 @@ export class ModPackUpdateEvent {
 	static fromJSON(json: Static<typeof this.jsonSchema>) {
 		return new this(ModPack.fromJSON(json.modPack));
 	}
+
+	get subscriptionChannel() {
+		return this.modPack.id;
+	}
 }
 
 export class ModUpdateEvent {
@@ -309,6 +270,7 @@ export class ModUpdateEvent {
 	static type = "event" as const;
 	static src = "controller" as const;
 	static dst = "control" as const;
+	static permission = "core.mod.subscribe" as const;
 
 	constructor(
 		public mod: ModInfo,
@@ -321,4 +283,8 @@ export class ModUpdateEvent {
 	static fromJSON(json: Static<typeof this.jsonSchema>) {
 		return new this(ModInfo.fromJSON(json.mod));
 	}
+
+	get subscriptionChannel() {
+		return this.mod.name;
+	}
 }

packages/lib/src/data/messages_user.ts

@@ -251,28 +251,6 @@ export class UserListRequest {
 	static Response = jsonArray(RawUser);
 }
 
-export class UserSetSubscriptionsRequest {
-	declare ["constructor"]: typeof UserSetSubscriptionsRequest;
-	static type = "request" as const;
-	static src = "control" as const;
-	static dst = "controller" as const;
-	static permission = "core.user.subscribe" as const;
-
-	constructor(
-		public all: boolean,
-		public names: string[],
-	) { }
-
-	static jsonSchema = Type.Object({
-		"all": Type.Boolean(),
-		"names": Type.Array(Type.String()),
-	});
-
-	static fromJSON(json: Static<typeof this.jsonSchema>) {
-		return new this(json.all, json.names);
-	}
-}
-
 export class UserCreateRequest {
 	declare ["constructor"]: typeof UserCreateRequest;
 	static type = "request" as const;
@@ -460,6 +438,7 @@ export class UserUpdateEvent {
 	static type = "event" as const;
 	static src = "controller" as const;
 	static dst = "control" as const;
+	static permission = "core.user.subscribe" as const;
 
 	constructor(
 		public user: RawUser,
@@ -472,4 +451,8 @@ export class UserUpdateEvent {
 	static fromJSON(json: Static<typeof this.jsonSchema>) {
 		return new this(RawUser.fromJSON(json.user));
 	}
+
+	get subscriptionChannel() {
+		return this.user.name;
+	}
 }

packages/lib/src/link/messages.ts

@@ -5,6 +5,7 @@ import * as host from "../data/messages_host";
 import * as instance from "../data/messages_instance";
 import * as mod from "../data/messages_mod";
 import * as user from "../data/messages_user";
+import * as subscriptions from "../subscriptions"
 import type { RequestClass, EventClass } from "./link";
 
 /**
@@ -27,8 +28,9 @@ export const dataClasses: (RequestClass<unknown, unknown> | EventClass<unknown>)
 	controller.DebugDumpWsRequest,
 	controller.DebugWsMessageEvent,
 
+	subscriptions.SubscriptionRequest,
+
 	host.HostListRequest,
-	host.HostSetSubscriptionsRequest,
 	host.HostUpdateEvent,
 	host.HostMetricsRequest,
 	host.ControllerConnectionEvent,
@@ -38,7 +40,6 @@ export const dataClasses: (RequestClass<unknown, unknown> | EventClass<unknown>)
 
 	instance.InstanceDetailsGetRequest,
 	instance.InstanceDetailsListRequest,
-	instance.InstanceDetailsSetSubscriptionsRequest,
 	instance.InstanceDetailsUpdateEvent,
 	instance.InstanceCreateRequest,
 	instance.InstanceConfigGetRequest,
@@ -48,7 +49,6 @@ export const dataClasses: (RequestClass<unknown, unknown> | EventClass<unknown>)
 	instance.InstanceMetricsRequest,
 	instance.InstanceStartRequest,
 	instance.InstanceListSavesRequest,
-	instance.InstanceSetSaveListSubscriptionsRequest,
 	instance.InstanceSaveListUpdateEvent,
 	instance.InstanceCreateSaveRequest,
 	instance.InstanceRenameSaveRequest,
@@ -83,11 +83,9 @@ export const dataClasses: (RequestClass<unknown, unknown> | EventClass<unknown>)
 	mod.ModPackCreateRequest,
 	mod.ModPackUpdateRequest,
 	mod.ModPackDeleteRequest,
-	mod.ModPackSetSubscriptionsRequest,
 	mod.ModGetRequest,
 	mod.ModListRequest,
 	mod.ModSearchRequest,
-	mod.ModSetSubscriptionsRequest,
 	mod.ModDownloadRequest,
 	mod.ModDeleteRequest,
 	mod.ModPackUpdateEvent,
@@ -101,7 +99,6 @@ export const dataClasses: (RequestClass<unknown, unknown> | EventClass<unknown>)
 	user.RoleDeleteRequest,
 	user.UserGetRequest,
 	user.UserListRequest,
-	user.UserSetSubscriptionsRequest,
 	user.UserCreateRequest,
 	user.UserRevokeTokenRequest,
 	user.UserUpdateRolesRequest,

packages/lib/src/subscriptions.ts

@@ -0,0 +1,332 @@
+import { Type, Static } from "@sinclair/typebox";
+import { Link, Event, EventClass, RequestHandler, WebSocketClientConnector, WebSocketBaseConnector } from "./link";
+import { Address, MessageRequest } from "./data";
+import { User } from "./users";
+
+export type SubscriptionRequestHandler<T> = RequestHandler<SubscriptionRequest, Event<T> | null>;
+export type EventSubscriberCallback<T> = (value: T) => void;
+
+export type ChannelEvent<T> = Event<T> & {
+	get subscriptionChannel(): number | string;
+}
+
+export type ChannelEventClass<T> = EventClass<T> & {
+	new(...args: any): ChannelEvent<T>
+}
+
+/**
+ * Response received by the subscriber after a request
+ * It can contain an eventReplay value if the event sender implements a subscription handler
+ * This replay will be send to the handlers as if an update had just occurred
+ */
+export class SubscriptionResponse {
+	constructor(
+		public readonly eventReplay: Event<unknown> | null = null,
+	) {
+		if (eventReplay && !Link._eventsByClass.has(eventReplay.constructor)) {
+			throw new Error(`Unregistered Event class ${eventReplay.constructor.name}`);
+		}
+	}
+
+	static jsonSchema = Type.Union([
+		Type.Tuple([
+			Type.String(),
+			Type.Unknown(),
+		]),
+		Type.Null()
+	])
+
+	toJSON() {
+		if (this.eventReplay) {
+			const entry = Link._eventsByClass.get(this.eventReplay.constructor)!;
+			return [entry.name, this.eventReplay];
+		} else {
+			return null;
+		}
+	}
+
+	static fromJSON(json: Static<typeof SubscriptionResponse.jsonSchema>): SubscriptionResponse {
+		if (json) {
+			const entry = Link._eventsByName.get(json[0]);
+			if (!entry) {
+				throw new Error(`Unregistered Event class ${json[0]}`);
+			} else {
+				return new SubscriptionResponse(entry.eventFromJSON(json[1]));
+			}
+		} else {
+			return new SubscriptionResponse();
+		}
+	}
+}
+
+/**
+ * A subscription request sent by a subscriber, this updates what events the subscriber will be sent
+ * The permission for this request copies the permission from the event being subscribed to
+ * allChannels: false, channels: [] will unsubscribe the subscriber from all notifications
+ */
+export class SubscriptionRequest {
+	declare ["constructor"]: typeof SubscriptionRequest;
+	static type = "request" as const;
+	static src =  ["control", "instance"] as const;
+	static dst = "controller" as const;
+	static Response = SubscriptionResponse;
+	static permission(user: User, message: MessageRequest) {
+		if (typeof message.data === "object" && message.data !== null) {
+			const data = message.data as Static<typeof SubscriptionRequest.jsonSchema>;
+			const entry = Link._eventsByName.get(data[0]);
+			if (entry && entry.Event.permission) {
+				if (typeof entry.Event.permission === "string") {
+					user.checkPermission(entry.Event.permission);
+				} else {
+					entry.Event.permission(user, message);
+				}
+			}
+		}
+	}
+
+	constructor(
+		public eventName: string,
+		public allChannels: boolean,
+		public channels: Array<string | number> = [],
+		public lastRequestTime: number = 0,
+	) {
+		if (!Link._eventsByName.has(eventName)) {
+			throw new Error(`Unregistered Event class ${eventName}`);
+		}
+	}
+
+	static jsonSchema = Type.Tuple([
+		Type.String(),
+		Type.Boolean(),
+		Type.Array(Type.Union([Type.String(), Type.Number()])),
+		Type.Number(),
+	])
+
+	toJSON() {
+		return [this.eventName, this.allChannels, this.channels, this.lastRequestTime];
+	}
+
+	static fromJSON(json: Static<typeof SubscriptionRequest.jsonSchema>): SubscriptionRequest {
+		return new this(...json);
+	}
+}
+
+/**
+ * A class component to handle incoming subscription requests and offers a method to broadcast events to subscribers
+ * After creation, no other handler can be registered for SubscriptionRequest on the controller
+ */
+export class SubscriptionController {
+	_events = new Map<string, {
+		subscriptionUpdate?: SubscriptionRequestHandler<unknown>,
+		subscriptions: Map<Link, { all: boolean, channels: Array<string | number>, onceClose: () => void }>,
+	}>();
+
+	constructor(
+		private controller: any, // Controller | Link
+	) {
+		this.controller.handle(SubscriptionRequest, this._handleRequest.bind(this));
+	}
+
+	/**
+	 * Allow clients to subscribe to an event by telling the subscription controller to accept them
+	 * Has an optional subscription update handler which is called when any client updates their subscription
+	 */
+	handle<T>(Event: EventClass<T>, subscriptionUpdate?: SubscriptionRequestHandler<T>) {
+		const entry = Link._eventsByClass.get(Event);
+		if (!entry) {
+			throw new Error(`Unregistered Event class ${Event.name}`);
+		}
+		if (this._events.has(entry.name)) {
+			throw new Error(`Event ${entry.name} is already registered`);
+		}
+		this._events.set(entry.name, {
+			subscriptionUpdate: subscriptionUpdate,
+			subscriptions: new Map(),
+		});
+	}
+
+	/**
+	 * Broadcast an event to all subscribers of that event, will be filtered by channel if a ChannelEvent is provided
+	 */
+	broadcast<T>(event: Event<T> | ChannelEvent<T>) {
+		const entry = Link._eventsByClass.get(event.constructor);
+		if (!entry) {
+			throw new Error(`Unregistered Event class ${event.constructor.name}`);
+		}
+		const eventData = this._events.get(entry.name);
+		if (!eventData) {
+			throw new Error(`Event ${entry.name} is not a registered as subscribable`);
+		}
+		const channel = "subscriptionChannel" in event ? event.subscriptionChannel : null;
+		for (let [link, subscription] of eventData.subscriptions) {
+			if ((link.connector as WebSocketBaseConnector).closing) {
+				eventData.subscriptions.delete(link);
+			} else if ((subscription.all || (channel && subscription.channels.includes(channel)))) {
+				link.send(event);
+			}
+		}
+	}
+
+	/**
+	 * Handle incoming event subscription requests
+	 */
+	async _handleRequest(event: SubscriptionRequest, src: Address, dst: Address) {
+		if (!Link._eventsByName.has(event.eventName)) {
+			throw new Error(`Event ${event.eventName} is not a registered event`);
+		}
+		const eventData = this._events.get(event.eventName);
+		if (!eventData) {
+			throw new Error(`Event ${event.eventName} is not a registered as subscribable`);
+		}
+		const eventReplay = eventData.subscriptionUpdate ? await eventData.subscriptionUpdate(event, src, dst) : null;
+		const link: Link = this.controller.wsServer.controlConnections.get(src.id);
+		if (event.allChannels === false && event.channels.length === 0) {
+			let onceClose = eventData.subscriptions.get(link)?.onceClose;
+			if (onceClose) {
+				link.connector.off("close", onceClose);
+				eventData.subscriptions.delete(link);
+			}
+		} else {
+			let onceClose = eventData.subscriptions.get(link)?.onceClose;
+			if (!onceClose) {
+				onceClose = () => eventData.subscriptions.delete(link);
+				link.connector.once("close", onceClose);
+			}
+			eventData.subscriptions.set(link, { all: event.allChannels, channels: event.channels, onceClose: onceClose });
+		}
+		return new SubscriptionResponse(eventReplay);
+	}
+}
+
+/**
+ * A class component to allow subscribing and unsubscribing to/from an event and its channels
+ * Channels are an optional feature which can be implemented to only get notifications of reliant updates
+ * Multiple handlers can be subscribed at the same time, on the same or different channels
+ */
+export class EventSubscriber<T, V=T> {
+	_callbacks = new Array<EventSubscriberCallback<V>>()
+	_channelCallbacks = new Map<string | number, Array<EventSubscriberCallback<V>>>()
+	lastResponse: Event<T> | null = null;
+	lastResponseTime = 0;
+	control?: Link
+
+	constructor(
+		private event: EventClass<T> | ChannelEventClass<T>,
+		private prehandler?: (event: T) => V,
+		control?: Link
+	) {
+		const entry = Link._eventsByClass.get(this.event);
+		if (!entry) {
+			throw new Error(`Unregistered Event class ${this.event.name}`);
+		}
+		if (control) {
+			this.connectControl(control);
+		}
+	}
+
+	/**
+	 * Handle incoming events and distribute it to the correct callbacks
+	 */
+	async _handle(response: Event<T> | ChannelEvent<T>) {
+		this.lastResponse = response;
+		this.lastResponseTime = Date.now();
+		const value = this.prehandler ? this.prehandler(response as T) : response as any;
+		for (let callback of this._callbacks) {
+			callback(value);
+		}
+		const channel = "subscriptionChannel" in response ? response.subscriptionChannel : null;
+		if (channel) {
+			const callbacks = this._channelCallbacks.get(channel);
+			if (callbacks) {
+				for (let callback of callbacks) {
+					callback(value);
+				}
+			}
+		}
+	}
+
+	/**
+	 * If a control link was not connected at creation, it can be connected here (normally during onControllerConnectionEvent)
+	 */
+	async connectControl(control: Link) {
+		if (this.control === control) return;
+		this.control = control;
+		this.control.handle(this.event, this._handle.bind(this));
+		await this._updateSubscription();
+	}
+
+	/**
+	 * Subscribe to receive all event notifications
+	 */
+	async subscribe(handler: EventSubscriberCallback<V>) {
+		this._callbacks.push(handler);
+		await this._updateSubscription();
+	}
+
+	/**
+	 * Subscribe to receive event notifications for the given channel
+	 */
+	async subscribeToChannel(channel: string | number, handler: EventSubscriberCallback<V>) {
+		if (!this._channelCallbacks.get(channel)) {
+			this._channelCallbacks.set(channel, []);
+		}
+		this._channelCallbacks.get(channel)!.push(handler);
+		await this._updateSubscription();
+	}
+
+	/**
+	 * Unsubscribe from receiving all event notifications, does not effect the subscriptions of other handlers
+	 * Will throw an error if your handler is not subscribed to all channels, does not accept anonymous functions
+	 */
+	async unsubscribe(handler: EventSubscriberCallback<V>) {
+		let index = this._callbacks.lastIndexOf(handler);
+		if (index === -1) {
+			throw new Error("handler is not registered");
+		}
+
+		this._callbacks.splice(index, 1);
+		await this._updateSubscription();
+	}
+
+	/**
+	 * Unsubscribe from receiving event notifications for a given channel, does not effect the subscriptions of other handlers
+	 * Will throw an error if your handler is not subscribed to the channel, does not accept anonymous functions
+	 */
+	async unsubscribeFromChannel(channel: string | number, handler: EventSubscriberCallback<V>) {
+		const channelCallbacks = this._channelCallbacks.get(channel);
+		if (!channelCallbacks) {
+			throw new Error("handler is not registered");
+		}
+
+		let index = channelCallbacks.lastIndexOf(handler);
+		if (index === -1) {
+			throw new Error("handler is not registered");
+		}
+
+		if (channelCallbacks.length === 1) {
+			this._channelCallbacks.delete(channel);
+		} else {
+			channelCallbacks.splice(index, 1);
+		}
+		await this._updateSubscription();
+	}
+
+	/**
+	 * Update the subscription with the controller based on current handler counts
+	 */
+	async _updateSubscription() {
+		if (!this.control || !(this.control.connector as WebSocketClientConnector).connected) return;
+		const entry = Link._eventsByClass.get(this.event)!;
+
+		const response = await this.control.send(new SubscriptionRequest(
+			entry.name,
+			this._callbacks.length > 0,
+			[...this._channelCallbacks.keys()],
+			this.lastResponseTime
+		));
+
+		if (response.eventReplay) {
+			await this._handle(response.eventReplay);
+		}
+	}
+}

packages/web_ui/src/model/host.ts

@@ -38,9 +38,9 @@ export function useHost(id: number): [ HostState, () => void ] {
 			setHost({ ...newHost, loading:false, missing:false, present:true });
 		}
 
-		control.onHostUpdate(id, updateHandler);
+		control.hostUpdate.subscribeToChannel(id, updateHandler);
 		return () => {
-			control.offHostUpdate(id, updateHandler);
+			control.hostUpdate.unsubscribeFromChannel(id, updateHandler);
 		};
 	}, [id]);
 
@@ -74,9 +74,9 @@ export function useHostList() {
 			});
 		}
 
-		control.onHostUpdate(null, updateHandler);
+		control.hostUpdate.subscribe(updateHandler);
 		return () => {
-			control.offHostUpdate(null, updateHandler);
+			control.hostUpdate.unsubscribe(updateHandler);
 		};
 	}, []);
 

packages/web_ui/src/model/instance.ts

@@ -35,9 +35,9 @@ export function useInstance(id: number): [InstanceState, ()=>void] {
 			setInstance({ ...newInstance, present: true });
 		}
 
-		control.onInstanceUpdate(id, updateHandler);
+		control.instanceUpdate.subscribeToChannel(id, updateHandler);
 		return () => {
-			control.offInstanceUpdate(id, updateHandler);
+			control.instanceUpdate.unsubscribeFromChannel(id, updateHandler);
 		};
 	}, [id]);
 
@@ -76,9 +76,9 @@ export function useInstanceList() {
 			});
 		}
 
-		control.onInstanceUpdate(null, updateHandler);
+		control.instanceUpdate.subscribe(updateHandler);
 		return () => {
-			control.offInstanceUpdate(null, updateHandler);
+			control.instanceUpdate.unsubscribe(updateHandler);
 		};
 	}, []);
 

packages/web_ui/src/model/mod_pack.ts

@@ -33,9 +33,9 @@ export function useModPack(id: number) {
 		}
 		updateModPack();
 
-		control.onModPackUpdate(id, setModPack);
+		control.modPackUpdate.subscribeToChannel(id, setModPack);
 		return () => {
-			control.offModPackUpdate(id, setModPack);
+			control.modPackUpdate.unsubscribeFromChannel(id, setModPack);
 		};
 	}, [id]);
 
@@ -74,9 +74,9 @@ export function useModPackList() {
 			});
 		}
 
-		control.onModPackUpdate(null, updateHandler);
+		control.modPackUpdate.subscribe(updateHandler);
 		return () => {
-			control.offModPackUpdate(null, updateHandler);
+			control.modPackUpdate.unsubscribe(updateHandler);
 		};
 	}, []);
 

packages/web_ui/src/model/mods.ts

@@ -38,9 +38,9 @@ export function useModList() {
 			});
 		}
 
-		control.onModUpdate(null, updateHandler);
+		control.modUpdate.subscribe(updateHandler);
 		return () => {
-			control.offModUpdate(null, updateHandler);
+			control.modUpdate.unsubscribe(updateHandler);
 		};
 	}, []);
 

packages/web_ui/src/model/saves.ts

@@ -29,9 +29,9 @@ export function useSaves(instanceId?: number): lib.SaveDetails[] {
 			setSaves(data.saves);
 		}
 
-		control.onSaveListUpdate(instanceId!, updateHandler);
+		control.saveListUpdate.subscribeToChannel(instanceId!, updateHandler);
 		return () => {
-			control.offSaveListUpdate(instanceId!, updateHandler);
+			control.saveListUpdate.unsubscribeFromChannel(instanceId!, updateHandler);
 		};
 	}, [instanceId]);
 

packages/web_ui/src/model/user.tsx

@@ -82,9 +82,9 @@ export function useUser(name: string): [RawUserState, () => void] {
 			setUser({ ...newUser, present: true });
 		}
 
-		control.onUserUpdate(name, updateHandler);
+		control.userUpdate.subscribeToChannel(name, updateHandler);
 		return () => {
-			control.offUserUpdate(name, updateHandler);
+			control.userUpdate.unsubscribeFromChannel(name, updateHandler);
 		};
 	}, [name]);
 
@@ -123,9 +123,9 @@ export function useUserList() {
 			});
 		}
 
-		control.onUserUpdate(null, updateHandler);
+		control.userUpdate.subscribe(updateHandler);
 		return () => {
-			control.offUserUpdate(null, updateHandler);
+			control.userUpdate.unsubscribe(updateHandler);
 		};
 	}, []);
 

packages/web_ui/src/util/websocket.ts

@@ -50,15 +50,19 @@ export class Control extends lib.Link {
 	accountName: string | null = null;
 	/** Roles of the account this control link is connected as. */
 	accountRoles: lib.AccountRole[] | null = null;
+
+	/** Updates not handled by the subscription service */
 	accountUpdateHandlers: Function[] = [];
-	hostUpdateHandlers: Map<number|null, hostHandler[]> = new Map();
-	instanceUpdateHandlers: Map<number|null, instanceHandler[]> = new Map();
-	saveListUpdateHandlers: Map<number|null, saveListHandler[]> = new Map();
-	modPackUpdateHandlers: Map<number|null, modPackHandler[]> = new Map();
-	modUpdateHandlers: Map<string|null, modInfoHandler[]> = new Map();
-	userUpdateHandlers: Map<string|null, userHandler[]> = new Map();
 	logHandlers: Map<lib.LogFilter, logHandler[]> = new Map();
 
+	/** Updates handled by the subscription service */
+	hostUpdate = new lib.EventSubscriber<lib.HostUpdateEvent, lib.HostDetails>(lib.HostUpdateEvent, event => event.update);
+	instanceUpdate = new lib.EventSubscriber<lib.InstanceDetailsUpdateEvent, lib.InstanceDetails>(lib.InstanceDetailsUpdateEvent, event => event.details);
+	saveListUpdate = new lib.EventSubscriber<lib.InstanceSaveListUpdateEvent>(lib.InstanceSaveListUpdateEvent);
+	modPackUpdate = new lib.EventSubscriber<lib.ModPackUpdateEvent, lib.ModPack>(lib.ModPackUpdateEvent, event => event.modPack);
+	modUpdate = new lib.EventSubscriber<lib.ModUpdateEvent, lib.ModInfo>(lib.ModUpdateEvent, event => event.mod);
+	userUpdate = new lib.EventSubscriber<lib.UserUpdateEvent, lib.RawUser>(lib.UserUpdateEvent, event => event.user);
+
 	declare connector: ControlConnector;
 
 	constructor(
@@ -75,22 +79,22 @@ export class Control extends lib.Link {
 			this.accountName = data.account.name;
 			this.accountRoles = data.account.roles;
 			this.emitAccountUpdate();
-			this.updateHostSubscriptions().catch(err => logger.error(
+			this.hostUpdate.connectControl(this).catch(err => logger.error(
 				`Unexpected error updating host subscriptions:\n${err.stack}`
-			));
-			this.updateInstanceSubscriptions().catch(err => logger.error(
+			));;
+			this.instanceUpdate.connectControl(this).catch(err => logger.error(
 				`Unexpected error updating instance subscriptions:\n${err.stack}`
 			));
-			this.updateSaveListSubscriptions().catch(err => logger.error(
+			this.saveListUpdate.connectControl(this).catch(err => logger.error(
 				`Unexpected error updating save list subscriptions:\n${err.stack}`
 			));
-			this.updateModPackSubscriptions().catch(err => logger.error(
+			this.modPackUpdate.connectControl(this).catch(err => logger.error(
 				`Unexpected error updating mod pack subscriptions:\n${err.stack}`
 			));
-			this.updateModSubscriptions().catch(err => logger.error(
+			this.modUpdate.connectControl(this).catch(err => logger.error(
 				`Unexpected error updating mod subscriptions:\n${err.stack}`
 			));
-			this.updateUserSubscriptions().catch(err => logger.error(
+			this.userUpdate.connectControl(this).catch(err => logger.error(
 				`Unexpected error updating user subscriptions:\n${err.stack}`
 			));
 			this.updateLogSubscriptions().catch(err => logger.error(
@@ -113,12 +117,6 @@ export class Control extends lib.Link {
 		}
 
 		this.handle(lib.AccountUpdateEvent, this.handleAccountUpdateEvent.bind(this));
-		this.handle(lib.HostUpdateEvent, this.handleHostUpdateEvent.bind(this));
-		this.handle(lib.InstanceDetailsUpdateEvent, this.handleInstanceDetailsUpdateEvent.bind(this));
-		this.handle(lib.InstanceSaveListUpdateEvent, this.handleInstanceSaveListUpdateEvent.bind(this));
-		this.handle(lib.ModPackUpdateEvent, this.handleModPackUpdateEvent.bind(this));
-		this.handle(lib.ModUpdateEvent, this.handleModUpdateEvent.bind(this));
-		this.handle(lib.UserUpdateEvent, this.handleUserUpdateEvent.bind(this));
 		this.handle(lib.LogMessageEvent, this.handleLogMessageEvent.bind(this));
 		this.handle(lib.DebugWsMessageEvent, this.handleDebugWsMessageEvent.bind(this));
 	}
@@ -149,349 +147,6 @@ export class Control extends lib.Link {
 		this.accountUpdateHandlers.splice(index, 1);
 	}
 
-	async handleHostUpdateEvent(event: lib.HostUpdateEvent) {
-		let handlers: Function[] = ([] as Function[]).concat(
-			this.hostUpdateHandlers.get(null) || [],
-			this.hostUpdateHandlers.get(event.update.id) || [],
-		);
-		for (let handler of handlers) {
-			handler(event.update);
-		}
-	}
-
-	async onHostUpdate(id: number|null, handler: hostHandler) {
-		if (id !== null && !Number.isInteger(id)) {
-			throw new Error("Invalid host id");
-		}
-
-		let handlers = this.hostUpdateHandlers.get(id);
-		if (!handlers) {
-			handlers = [];
-			this.hostUpdateHandlers.set(id, handlers);
-		}
-
-		handlers.push(handler);
-
-		if (handlers.length === 1) {
-			await this.updateHostSubscriptions();
-		}
-	}
-
-	async offHostUpdate(id: number|null, handler: hostHandler) {
-		let handlers = this.hostUpdateHandlers.get(id);
-		if (!handlers || !handlers.length) {
-			throw new Error(`No handlers for host ${id} exists`);
-		}
-
-		let index = handlers.lastIndexOf(handler);
-		if (index === -1) {
-			throw new Error(`Given handler is not registered for host ${id}`);
-		}
-
-		handlers.splice(index, 1);
-		if (!handlers.length) {
-			this.hostUpdateHandlers.delete(id);
-			await this.updateHostSubscriptions();
-		}
-	}
-
-	async updateHostSubscriptions() {
-		if (!this.connector.connected) {
-			return;
-		}
-
-		await this.send(new lib.HostSetSubscriptionsRequest(
-			this.hostUpdateHandlers.has(null),
-			[...this.hostUpdateHandlers.keys()].filter(e => e !== null) as number[],
-		));
-	}
-
-	async handleInstanceDetailsUpdateEvent(event: lib.InstanceDetailsUpdateEvent) {
-		let handlers = ([] as Function[]).concat(
-			this.instanceUpdateHandlers.get(null) || [],
-			this.instanceUpdateHandlers.get(event.details.id) || [],
-		);
-		for (let handler of handlers) {
-			handler(event.details);
-		}
-	};
-
-	async onInstanceUpdate(id: number|null, handler: instanceHandler) {
-		if (id !== null && !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: number|null, handler: instanceHandler) {
-		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 this.send(new lib.InstanceDetailsSetSubscriptionsRequest(
-			this.instanceUpdateHandlers.has(null),
-			[...this.instanceUpdateHandlers.keys()].filter(e => e !== null) as number[],
-		));
-	}
-
-	async handleInstanceSaveListUpdateEvent(event: lib.InstanceSaveListUpdateEvent) {
-		let handlers = this.saveListUpdateHandlers.get(event.instanceId);
-		for (let handler of handlers || []) {
-			handler(event);
-		}
-	};
-
-	async onSaveListUpdate(id: number, handler: saveListHandler) {
-		if (!Number.isInteger(id)) {
-			throw new Error("Invalid instance id");
-		}
-
-		let handlers = this.saveListUpdateHandlers.get(id);
-		if (!handlers) {
-			handlers = [];
-			this.saveListUpdateHandlers.set(id, handlers);
-		}
-
-		handlers.push(handler);
-
-		if (handlers.length === 1) {
-			await this.updateSaveListSubscriptions();
-		}
-	}
-
-	async offSaveListUpdate(id: number, handler: saveListHandler) {
-		let handlers = this.saveListUpdateHandlers.get(id);
-		if (!handlers) {
-			throw new Error(`No handlers for instance ${id} exists`);
-		}
-
-		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.saveListUpdateHandlers.delete(id);
-			await this.updateSaveListSubscriptions();
-		}
-	}
-
-	async updateSaveListSubscriptions() {
-		if (!this.connector.connected) {
-			return;
-		}
-
-		await this.send(new lib.InstanceSetSaveListSubscriptionsRequest(
-			false,
-			[...this.saveListUpdateHandlers.keys()].filter(e => e !== null) as number[],
-		));
-	}
-
-	async handleModPackUpdateEvent(event: lib.ModPackUpdateEvent) {
-		let modPack = event.modPack;
-
-		let handlers = ([] as modPackHandler[]).concat(
-			this.modPackUpdateHandlers.get(null) || [],
-			this.modPackUpdateHandlers.get(modPack.id??null) || [],
-		);
-		for (let handler of handlers) {
-			handler(modPack);
-		}
-	}
-
-	async onModPackUpdate(id: number|null, handler: modPackHandler) {
-		if (id !== null && typeof id !== "number") {
-			throw new Error("Invalid mod pack id");
-		}
-
-		let handlers = this.modPackUpdateHandlers.get(id);
-		if (!handlers) {
-			handlers = [];
-			this.modPackUpdateHandlers.set(id, handlers);
-		}
-
-		handlers.push(handler);
-
-		if (handlers.length === 1) {
-			await this.updateModPackSubscriptions();
-		}
-	}
-
-	async offModPackUpdate(id: number|null, handler: modPackHandler) {
-		let handlers = this.modPackUpdateHandlers.get(id);
-		if (!handlers || !handlers.length) {
-			throw new Error(`No handlers for mod pack ${id} exist`);
-		}
-
-		let index = handlers.lastIndexOf(handler);
-		if (index === -1) {
-			throw new Error(`Given handler is not registered for mod pack ${id}`);
-		}
-
-		handlers.splice(index, 1);
-		if (!handlers.length) {
-			this.modPackUpdateHandlers.delete(id);
-			await this.updateModPackSubscriptions();
-		}
-	}
-
-	async updateModPackSubscriptions() {
-		if (!this.connector.connected) {
-			return;
-		}
-
-		await this.send(new lib.ModPackSetSubscriptionsRequest(
-			this.modPackUpdateHandlers.has(null),
-			[...this.modPackUpdateHandlers.keys()]
-				.filter(k => k !== null && k !== undefined) as number[],
-		));
-	}
-
-	async handleModUpdateEvent(event: lib.ModUpdateEvent) {
-		let mod = event.mod;
-		let handlers = ([] as modInfoHandler[]).concat(
-			this.modUpdateHandlers.get(null) || [],
-			this.modUpdateHandlers.get(mod.name) || []
-		);
-		for (let handler of handlers) {
-			handler(mod);
-		}
-	}
-
-	async onModUpdate(name: string|null, handler: modInfoHandler) {
-		if (name !== null && typeof name !== "string") {
-			throw new Error("Invalid mod name");
-		}
-
-		let handlers = this.modUpdateHandlers.get(name);
-		if (!handlers) {
-			handlers = [];
-			this.modUpdateHandlers.set(name, handlers);
-		}
-
-		handlers.push(handler);
-
-		if (handlers.length === 1) {
-			await this.updateModSubscriptions();
-		}
-	}
-
-	async offModUpdate(name: string|null, handler: modInfoHandler) {
-		let handlers = this.modUpdateHandlers.get(name);
-		if (!handlers || !handlers.length) {
-			throw new Error(`No handlers for mod ${name} exist`);
-		}
-
-		let index = handlers.lastIndexOf(handler);
-		if (index === -1) {
-			throw new Error(`Given handler is not registered for mod ${name}`);
-		}
-
-		handlers.splice(index, 1);
-		if (!handlers.length) {
-			this.modUpdateHandlers.delete(name);
-			await this.updateModSubscriptions();
-		}
-	}
-
-	async updateModSubscriptions() {
-		if (!this.connector.connected) {
-			return;
-		}
-
-		await this.send(new lib.ModSetSubscriptionsRequest(
-			this.modUpdateHandlers.has(null),
-			[...this.modUpdateHandlers.keys()].filter(k => k !== null) as string[],
-		));
-	}
-
-	async handleUserUpdateEvent(event: lib.UserUpdateEvent) {
-		let handlers = ([] as userHandler[]).concat(
-			this.userUpdateHandlers.get(null) || [],
-			this.userUpdateHandlers.get(event.user.name) || [],
-		);
-		for (let handler of handlers) {
-			handler(event.user);
-		}
-	};
-
-	async onUserUpdate(name: string|null, handler: userHandler) {
-		if (name !== null && typeof name !== "string") {
-			throw new Error("Invalid user name");
-		}
-
-		let handlers = this.userUpdateHandlers.get(name);
-		if (!handlers) {
-			handlers = [];
-			this.userUpdateHandlers.set(name, handlers);
-		}
-
-		handlers.push(handler);
-
-		if (handlers.length === 1) {
-			await this.updateUserSubscriptions();
-		}
-	}
-
-	async offUserUpdate(name: string|null, handler: userHandler) {
-		let handlers = this.userUpdateHandlers.get(name);
-		if (!handlers || !handlers.length) {
-			throw new Error(`No handlers for user ${name} exist`);
-		}
-
-		let index = handlers.lastIndexOf(handler);
-		if (index === -1) {
-			throw new Error(`Given handler is not registered for user ${name}`);
-		}
-
-		handlers.splice(index, 1);
-		if (!handlers.length) {
-			this.userUpdateHandlers.delete(name);
-			await this.updateUserSubscriptions();
-		}
-	}
-
-	async updateUserSubscriptions() {
-		if (!this.connector.connected) {
-			return;
-		}
-
-		await this.send(new lib.UserSetSubscriptionsRequest(
-			this.userUpdateHandlers.has(null),
-			[...this.userUpdateHandlers.keys()].filter(e => e !== null) as string[],
-		));
-	}
-
 	async handleLogMessageEvent(event: lib.LogMessageEvent) {
 		let info = event.info;
 

test/integration/index.js

@@ -34,19 +34,22 @@ class TestControl extends lib.Link {
 				return;
 			}
 			this.send(
-				new lib.HostSetSubscriptionsRequest(true, [])
+				new lib.SubscriptionRequest(lib.HostUpdateEvent.name, true)
 			).catch(err => logger.error(`Error setting host subscriptions:\n${err.stack}`));
 			this.send(
-				new lib.InstanceDetailsSetSubscriptionsRequest(true, [])
+				new lib.SubscriptionRequest(lib.InstanceDetailsUpdateEvent.name, true)
 			).catch(err => logger.error(`Error setting instance subscriptions:\n${err.stack}`));
 			this.send(
-				new lib.InstanceSetSaveListSubscriptionsRequest(true, [])
+				new lib.SubscriptionRequest(lib.InstanceSaveListUpdateEvent.name, true)
 			).catch(err => logger.error(`Error setting save list subscriptions:\n${err.stack}`));
 			this.send(
-				new lib.ModSetSubscriptionsRequest(true, [])
+				new lib.SubscriptionRequest(lib.ModUpdateEvent.name, true)
 			).catch(err => logger.error(`Error setting mod subscriptions:\n${err.stack}`));
 			this.send(
-				new lib.UserSetSubscriptionsRequest(true, [])
+				new lib.SubscriptionRequest(lib.ModPackUpdateEvent.name, true)
+			).catch(err => logger.error(`Error setting mod pack subscriptions:\n${err.stack}`));
+			this.send(
+				new lib.SubscriptionRequest(lib.UserUpdateEvent.name, true)
 			).catch(err => logger.error(`Error setting user subscriptions:\n${err.stack}`));
 		});
 

test/lib/subscriptions.js

@@ -0,0 +1,585 @@
+"use strict";
+const events = require("events");
+const assert = require("assert").strict;
+const lib = require("@clusterio/lib");
+const { MockController, MockConnector, MockControl } = require("../mock");
+
+const addr = lib.Address.fromShorthand;
+
+describe("lib/subscriptions", function() {
+	class UnregisteredEvent {
+		static type = "event";
+		static src = ["controller", "control"];
+		static dst = ["controller", "control"];
+		static permission = null;
+	}
+
+	class RegisteredEvent {
+		static type = "event";
+		static src = ["controller", "control"];
+		static dst = ["controller", "control"];
+		static permission = null;
+	}
+	lib.Link.register(RegisteredEvent);
+
+	class ChannelEvent {
+		static type = "event";
+		static src = ["controller", "control"];
+		static dst = ["controller", "control"];
+		static permission = null;
+
+		get subscriptionChannel() {
+			return "channelOne";
+		}
+	}
+	lib.Link.register(ChannelEvent);
+
+	class StringPermissionEvent {
+		static type = "event";
+		static src = ["controller", "control"];
+		static dst = ["controller", "control"];
+		static permission = "StringPermission";
+	}
+	lib.Link.register(StringPermissionEvent);
+
+	class FunctionPermissionEvent {
+		static type = "event";
+		static src = ["controller", "control"];
+		static dst = ["controller", "control"];
+		static permission(user, message) {
+			user.checkPermission("FunctionPermission");
+		};
+	}
+	lib.Link.register(FunctionPermissionEvent);
+
+	class MockUser {
+		checkPermission(permission) {
+			this.lastPermissionCheck = permission;
+		}
+	}
+
+	describe("class SubscriptionResponse", function() {
+		it("should be round trip json serialisable without an event replay", function() {
+			const response = new lib.SubscriptionResponse();
+			const json = JSON.stringify(response);
+			assert.equal(typeof json, "string");
+			const reconstructed = lib.SubscriptionResponse.fromJSON(JSON.parse(json));
+			assert.deepEqual(reconstructed, response);
+		});
+
+		it("should be round trip json serialisable with an event replay", function() {
+			const response = new lib.SubscriptionResponse(new RegisteredEvent());
+			const json = JSON.stringify(response);
+			assert.equal(typeof json, "string");
+			const reconstructed = lib.SubscriptionResponse.fromJSON(JSON.parse(json));
+			assert.deepEqual(reconstructed, response);
+		});
+
+		it("should be round trip json serialisable with a null event replay", function() {
+			const response = new lib.SubscriptionResponse(null);
+			const json = JSON.stringify(response);
+			assert.equal(typeof json, "string");
+			const reconstructed = lib.SubscriptionResponse.fromJSON(JSON.parse(json));
+			assert.deepEqual(reconstructed, response);
+		});
+
+		it("should be throw when given an unregistered event in a json", function() {
+			assert.throws(
+				() => lib.SubscriptionResponse.fromJSON([UnregisteredEvent.name, {}]),
+				new Error(`Unregistered Event class ${UnregisteredEvent.name}`)
+			);
+		});
+
+		it("should be throw when given an unregistered event", function() {
+			assert.throws(
+				() => new lib.SubscriptionResponse(new UnregisteredEvent()),
+				new Error(`Unregistered Event class ${UnregisteredEvent.name}`)
+			);
+		});
+	});
+
+	describe("class SubscriptionRequest", function() {
+		describe("permission()", function() {
+			function newSubscriptionRequestMessage(request) {
+				return new lib.MessageRequest(
+					0,
+					addr({ controlId: 0 }),
+					addr("controller"),
+					lib.SubscriptionRequest.name,
+					request.toJSON()
+				);
+			}
+			it("should do nothing when the event has no permission property", function() {
+				const mockUser = new MockUser();
+				const request = new lib.SubscriptionRequest(RegisteredEvent.name, true);
+				lib.SubscriptionRequest.permission(mockUser, newSubscriptionRequestMessage(request));
+				assert.equal(mockUser.lastPermissionCheck, undefined);
+			});
+			it("should check user permission when the permission property is a string", function() {
+				const mockUser = new MockUser();
+				const request = new lib.SubscriptionRequest(StringPermissionEvent.name, true);
+				lib.SubscriptionRequest.permission(mockUser, newSubscriptionRequestMessage(request));
+				assert.equal(mockUser.lastPermissionCheck, "StringPermission");
+			});
+			it("should call the permission property when it is a function", function() {
+				const mockUser = new MockUser();
+				const request = new lib.SubscriptionRequest(FunctionPermissionEvent.name, true);
+				lib.SubscriptionRequest.permission(mockUser, newSubscriptionRequestMessage(request));
+				assert.equal(mockUser.lastPermissionCheck, "FunctionPermission");
+			});
+		});
+
+		it("should be round trip json serialisable", function() {
+			const request = new lib.SubscriptionRequest(RegisteredEvent.name, true, ["channelOne"], 123);
+			const json = JSON.stringify(request);
+			assert.equal(typeof json, "string");
+			const reconstructed = lib.SubscriptionRequest.fromJSON(JSON.parse(json));
+			assert.deepEqual(reconstructed, request);
+		});
+
+		it("should be throw when given an unregistered event", function() {
+			assert.throws(
+				() => new lib.SubscriptionRequest(UnregisteredEvent.name, true),
+				new Error(`Unregistered Event class ${UnregisteredEvent.name}`)
+			);
+		});
+	});
+
+	describe("class SubscriptionController", function() {
+		let mockController, subscriptions;
+		const connectorSetupDate = [{
+			id: 0,
+			request: new lib.SubscriptionRequest(RegisteredEvent.name, true),
+			src: addr({ controlId: 0 }),
+			dst: addr("controller"),
+		}, {
+			id: 1,
+			request: new lib.SubscriptionRequest(RegisteredEvent.name, true),
+			src: addr({ controlId: 1 }),
+			dst: addr("controller"),
+		}, {
+			id: 2,
+			request: new lib.SubscriptionRequest(ChannelEvent.name, true),
+			src: addr({ controlId: 2 }),
+			dst: addr("controller"),
+		}, {
+			id: 3,
+			request: new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelOne"]),
+			src: addr({ controlId: 3 }),
+			dst: addr("controller"),
+		}, {
+			id: 4,
+			request: new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelTwo"]),
+			src: addr({ controlId: 4 }),
+			dst: addr("controller"),
+		}];
+
+		function onceConnectorSend(id) {
+			const link = mockController.wsServer.controlConnections.get(id);
+			return events.once(link.connector, "send");
+		}
+
+		function assertLastEvent(connectorId, Event) {
+			const link = mockController.wsServer.controlConnections.get(connectorId);
+			const lastMessage = link.connector.sentMessages.at(-1);
+			const name = Event.plugin ? `${Event.plugin}:${Event.name}` : Event.name;
+			assert.equal(lastMessage.name, name);
+		}
+
+		function assertNoEvent(connectorId) {
+			const link = mockController.wsServer.controlConnections.get(connectorId);
+			assert.equal(link.connector.sentMessages.length, 0);
+		}
+
+		beforeEach(function() {
+			mockController = new MockController();
+			mockController.wsServer = {
+				controlConnections: new Map(connectorSetupDate.map(function (connectorData) {
+					return [connectorData.id, new lib.Link(new MockConnector(connectorData.src, connectorData.dst))];
+				})),
+			};
+			subscriptions = new lib.SubscriptionController(mockController);
+		});
+
+		it("should handle the SubscriptionRequest event", function() {
+			assert.equal(mockController.handles.has(lib.SubscriptionRequest), true);
+		});
+
+		describe("handle()", function() {
+			it("should accept registered events", function() {
+				subscriptions.handle(RegisteredEvent);
+			});
+			it("should not accept unregistered events", function() {
+				assert.throws(
+					() => subscriptions.handle(UnregisteredEvent),
+					new Error(`Unregistered Event class ${UnregisteredEvent.name}`)
+				);
+			});
+			it("should not accept events already handled by the class", function() {
+				subscriptions.handle(RegisteredEvent);
+				assert.throws(
+					() => subscriptions.handle(RegisteredEvent),
+					new Error(`Event ${RegisteredEvent.name} is already registered`)
+				);
+			});
+		});
+
+		describe("broadcast()", function() {
+			beforeEach(function() {
+				subscriptions.handle(RegisteredEvent);
+				subscriptions.handle(ChannelEvent);
+				for (let connectorData of connectorSetupDate) {
+					subscriptions._handleRequest(connectorData.request, connectorData.src, connectorData.dst);
+				}
+			});
+
+			it("should not accept unregistered events", function() {
+				assert.throws(
+					() => subscriptions.broadcast(new UnregisteredEvent()),
+					new Error(`Unregistered Event class ${UnregisteredEvent.name}`)
+				);
+			});
+			it("should not accept events not handled by the class", function() {
+				assert.throws(
+					() => subscriptions.broadcast(new StringPermissionEvent()),
+					new Error(`Event ${StringPermissionEvent.name} is not a registered as subscribable`)
+				);
+			});
+			it("should notify all links who subscribed to all channels", async function() {
+				subscriptions.broadcast(new RegisteredEvent());
+				await Promise.all([onceConnectorSend(0), onceConnectorSend(1)]);
+				assertLastEvent(0, RegisteredEvent); // RegisteredEvent: All
+				assertLastEvent(1, RegisteredEvent); // RegisteredEvent: All
+				assertNoEvent(2); // ChannelEvent: All
+				assertNoEvent(3); // ChannelEvent: channelOne
+				assertNoEvent(4); // ChannelEvent: channelTwo
+			});
+			it("should not notify a link who unsubscribed from all all channels", async function() {
+				const connectorData = connectorSetupDate[0];
+				const request = new lib.SubscriptionRequest(RegisteredEvent.name, false);
+				await subscriptions._handleRequest(request, connectorData.src, connectorData.dst);
+				mockController.wsServer.controlConnections.get(0).connector.sentMessages.pop(); // Response to request
+				subscriptions.broadcast(new RegisteredEvent());
+				await onceConnectorSend(1);
+				assertNoEvent(0);
+				assertLastEvent(1, RegisteredEvent);
+				assertNoEvent(2);
+				assertNoEvent(3);
+				assertNoEvent(4);
+			});
+			it("should notify all links who subscribed the specific channel", async function() {
+				subscriptions.broadcast(new ChannelEvent());
+				await Promise.all([onceConnectorSend(2), onceConnectorSend(3)]);
+				assertNoEvent(0);
+				assertNoEvent(1);
+				assertLastEvent(2, ChannelEvent);
+				assertLastEvent(3, ChannelEvent);
+				assertNoEvent(4);
+			});
+			it("should not notify a link who unsubscribed from the specific channel", async function() {
+				const connectorData = connectorSetupDate[3];
+				const request = new lib.SubscriptionRequest(ChannelEvent.name, false, []);
+				await subscriptions._handleRequest(request, connectorData.src, connectorData.dst);
+				mockController.wsServer.controlConnections.get(3).connector.sentMessages.pop(); // Response to request
+				subscriptions.broadcast(new ChannelEvent());
+				await onceConnectorSend(2);
+				assertNoEvent(0);
+				assertNoEvent(1);
+				assertLastEvent(2, ChannelEvent);
+				assertNoEvent(3);
+				assertNoEvent(4);
+			});
+			it("should not notify links which are closed or closing", async function() {
+				const link = mockController.wsServer.controlConnections.get(0);
+				link.connector.closing = true;
+				subscriptions.broadcast(new RegisteredEvent());
+				await onceConnectorSend(1);
+				assertNoEvent(0);
+				assertLastEvent(1, RegisteredEvent);
+				assertNoEvent(2);
+				assertNoEvent(3);
+				assertNoEvent(4);
+			});
+		});
+
+		describe("_handleRequest()", function() {
+			beforeEach(function() {
+				subscriptions.handle(RegisteredEvent);
+				subscriptions.handle(ChannelEvent);
+			});
+
+			it("should not accept subscriptions to unregistered events", function() {
+				const request = new lib.SubscriptionRequest(RegisteredEvent.name, true);
+				request.eventName = UnregisteredEvent.name;
+				assert.rejects(
+					subscriptions._handleRequest(request, addr({ controlId: 0 }), addr("controller")),
+					new Error(`Unregistered Event class ${UnregisteredEvent.name}`)
+				);
+			});
+			it("should not accept subscriptions to events not handled by the class", function() {
+				const request = new lib.SubscriptionRequest(RegisteredEvent.name, true);
+				request.eventName = StringPermissionEvent.name;
+				assert.rejects(
+					subscriptions._handleRequest(request, addr({ controlId: 0 }), addr("controller")),
+					new Error(`Event ${StringPermissionEvent.eventName} is not a registered as subscribable`)
+				);
+			});
+			it("should accept a subscription to all channels", async function() {
+				const request = new lib.SubscriptionRequest(RegisteredEvent.name, true);
+				await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst);
+				subscriptions.broadcast(new RegisteredEvent());
+				await onceConnectorSend(0);
+				assertLastEvent(0, RegisteredEvent);
+			});
+			it("should accept a unsubscription from all channels", async function() {
+				const request = new lib.SubscriptionRequest(RegisteredEvent.name, true);
+				await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst);
+				subscriptions.broadcast(new RegisteredEvent());
+				await onceConnectorSend(0);
+				assertLastEvent(0, RegisteredEvent);
+
+				const link = mockController.wsServer.controlConnections.get(0);
+				link.connector.sentMessages.pop();
+
+				const unsubRequest = new lib.SubscriptionRequest(RegisteredEvent.name, false);
+				await subscriptions._handleRequest(unsubRequest, connectorSetupDate[0].src, connectorSetupDate[0].dst);
+				subscriptions.broadcast(new RegisteredEvent());
+				assertNoEvent(0);
+			});
+			it("should accept a subscription to a specific channel", async function() {
+				const request = new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelOne"]);
+				await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst);
+				subscriptions.broadcast(new ChannelEvent());
+				await onceConnectorSend(0);
+				assertLastEvent(0, ChannelEvent);
+			});
+			it("should accept a unsubscription from a specific channel", async function() {
+				const request = new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelOne"]);
+				await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst);
+				subscriptions.broadcast(new ChannelEvent());
+				await onceConnectorSend(0);
+				assertLastEvent(0, ChannelEvent);
+
+				const link = mockController.wsServer.controlConnections.get(0);
+				link.connector.sentMessages.pop();
+
+				const unsubRequest = new lib.SubscriptionRequest(ChannelEvent.name, false, []);
+				await subscriptions._handleRequest(unsubRequest, connectorSetupDate[0].src, connectorSetupDate[0].dst);
+				subscriptions.broadcast(new ChannelEvent());
+				assertNoEvent(0);
+			});
+			it("should accept a respond with an event replay when returned by the handler", async function() {
+				subscriptions.handle(StringPermissionEvent, undefined, async function() {
+					return new StringPermissionEvent();
+				});
+
+				const request = new lib.SubscriptionRequest(StringPermissionEvent.name, true);
+				await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst);
+				subscriptions.broadcast(new StringPermissionEvent());
+				await onceConnectorSend(0);
+				assertLastEvent(0, StringPermissionEvent);
+
+				const link = mockController.wsServer.controlConnections.get(0);
+				const lastMessage = link.connector.sentMessages.pop();
+				const eventEntry = lib.Link._eventsByClass.get(StringPermissionEvent);
+				const response = eventEntry.eventFromJSON(lastMessage.data);
+				assert.deepEqual(response, new StringPermissionEvent());
+			});
+		});
+	});
+
+	describe("class EventSubscriber", function() {
+		let channelEvent, mockControl, registeredEvent;
+		beforeEach(function() {
+			mockControl = new MockControl(new MockConnector(
+				addr({ controlId: 0 }),
+				addr("controller"),
+			));
+			registeredEvent = new lib.EventSubscriber(RegisteredEvent, undefined, mockControl);
+			channelEvent = new lib.EventSubscriber(ChannelEvent, undefined, mockControl);
+		});
+
+		function assertLastRequest(request) {
+			const Request = request.constructor;
+			const lastMessage = mockControl.connector.sentMessages.at(-1);
+			const name = Request.plugin ? `${Request.plugin}:${Request.name}` : Request.name;
+			assert.equal(lastMessage.name, name);
+			assert.deepEqual(lastMessage.data, request);
+		}
+
+		describe("constructor()", function() {
+			it("should not accept unregistered events", function() {
+				assert.throws(
+					() => new lib.EventSubscriber(UnregisteredEvent),
+					new Error(`Unregistered Event class ${UnregisteredEvent.name}`)
+				);
+			});
+			it("should handle the provided event, if a control link is provided", function() {
+				assert.equal(mockControl._eventHandlers.has(RegisteredEvent), true);
+				assert.equal(mockControl._eventHandlers.has(ChannelEvent), true);
+			});
+			it("should call and use the return of a pre-handler, if provided", function() {
+				let calledWith = null;
+				function prehandler() { return "PreHandlerReturn"; }
+				const eventSubscriber = new lib.EventSubscriber(StringPermissionEvent, prehandler, mockControl);
+				eventSubscriber.subscribe(function(value) { calledWith = value; });
+				eventSubscriber._handle(new StringPermissionEvent());
+				assert.equal(calledWith, "PreHandlerReturn");
+			});
+		});
+
+		describe("connectControl()", function() {
+			it("should handle the provided event", function() {
+				const control = new MockControl(new MockConnector(addr({ controlId: 0 }), addr("controller")));
+				registeredEvent.connectControl(control);
+				assert.equal(control._eventHandlers.has(RegisteredEvent), true);
+			});
+			it("should do nothing if the control is already connected", function() {
+				assert.equal(mockControl._eventHandlers.has(RegisteredEvent), true);
+				registeredEvent.connectControl(mockControl);
+				assert.equal(mockControl._eventHandlers.has(RegisteredEvent), true);
+			});
+		});
+
+		describe("subscribe()", function() {
+			it("should allow subscriptions to an event", function() {
+				let calledWith = null;
+				let calledWithTwo = null;
+				const event = new RegisteredEvent();
+				registeredEvent.subscribe(function(value) { calledWith = value; });
+				registeredEvent.subscribe(function(value) { calledWithTwo = value; });
+				registeredEvent._handle(event);
+				assert.deepEqual(calledWith, event);
+				assert.deepEqual(calledWithTwo, event);
+			});
+		});
+
+		describe("subscribeToChannel()", function() {
+			it("should allow subscriptions to a channel for an event", function() {
+				let calledWith = null;
+				let calledWithTwo = null;
+				const event = new ChannelEvent();
+				channelEvent.subscribeToChannel("channelOne", function(value) { calledWith = value; });
+				channelEvent.subscribeToChannel("channelTwo", function(value) { calledWithTwo = value; });
+				channelEvent._handle(event);
+				assert.deepEqual(calledWith, event);
+				assert.deepEqual(calledWithTwo, null);
+			});
+		});
+
+		describe("unsubscribe()", function() {
+			it("should allow unsubscribing from an event", function() {
+				let calledWith = null;
+				let event = new RegisteredEvent();
+				function callback(value) { calledWith = value; }
+				registeredEvent.subscribe(callback);
+				registeredEvent._handle(event);
+				assert.deepEqual(calledWith, event);
+
+				calledWith = null;
+				event = new RegisteredEvent();
+				registeredEvent.unsubscribe(callback);
+				registeredEvent._handle(event);
+				assert.equal(calledWith, null);
+			});
+			it("should throw an error if the handler was not subscribed", function() {
+				function callback(value) { calledWith = value; }
+				assert.rejects(
+					() => registeredEvent.unsubscribe(callback),
+					new Error("handler is not registered")
+				);
+			});
+		});
+
+		describe("unsubscribeFromChannel()", function() {
+			it("should allow unsubscribing from a channel for an event", function() {
+				let calledWith = null;
+				let event = new ChannelEvent();
+				function callback(value) { calledWith = value; }
+				channelEvent.subscribeToChannel("channelOne", callback);
+				channelEvent._handle(event);
+				assert.deepEqual(calledWith, event);
+
+				calledWith = null;
+				event = new ChannelEvent();
+				channelEvent.unsubscribeFromChannel("channelOne", callback);
+				channelEvent._handle(event);
+				assert.equal(calledWith, null);
+
+				let calledWithTwo = null;
+				function callbackTwo(value) { calledWithTwo = value; }
+				channelEvent.subscribeToChannel("channelOne", callback);
+				channelEvent.subscribeToChannel("channelOne", callbackTwo);
+				channelEvent._handle(event);
+				assert.equal(calledWith, event);
+				assert.equal(calledWithTwo, event);
+
+				calledWith = null;
+				calledWithTwo = null;
+				event = new ChannelEvent();
+				channelEvent.unsubscribeFromChannel("channelOne", callback);
+				channelEvent._handle(event);
+				assert.equal(calledWith, null);
+				assert.equal(calledWithTwo, event);
+			});
+			it("should throw an error if the handler was not subscribed", function() {
+				function callback() {}
+				assert.rejects(
+					() => channelEvent.unsubscribeFromChannel("channelOne", callback),
+					new Error("handler is not registered")
+				);
+
+				let calledWith = null;
+				function callbackTwo(value) { calledWith = value; }
+				channelEvent.subscribeToChannel("channelOne", callbackTwo);
+				assert.rejects(
+					() => channelEvent.unsubscribeFromChannel("channelOne", callback),
+					new Error("handler is not registered")
+				);
+				const event = new ChannelEvent();
+				channelEvent._handle(event);
+				assert.equal(calledWith, event);
+			});
+		});
+
+		describe("_updateSubscription()", function() {
+			it("should correctly request a subscription for all channels", async function() {
+				const expected = new lib.SubscriptionRequest(RegisteredEvent.name, true);
+				registeredEvent.subscribe(() => true);
+				await events.once(mockControl.connector, "send");
+				assertLastRequest(expected);
+			});
+			it("should correctly request a subscription for some channels", async function() {
+				const expected = new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelOne"]);
+				channelEvent.subscribeToChannel("channelOne", () => true);
+				await events.once(mockControl.connector, "send");
+				assertLastRequest(expected);
+			});
+			it("should correctly request a subscription for no channels", async function() {
+				const expected = new lib.SubscriptionRequest(RegisteredEvent.name, false);
+				function callback() {}
+				registeredEvent.subscribe(callback);
+				await events.once(mockControl.connector, "send");
+				registeredEvent.unsubscribe(callback);
+				await events.once(mockControl.connector, "send");
+				assertLastRequest(expected);
+			});
+			it("should call handlers when a replay event is returned", async function() {
+				let calledWith = null;
+				const event = new RegisteredEvent();
+				const request = registeredEvent.subscribe(function(value) { calledWith = value; });
+				await events.once(mockControl.connector, "send");
+
+				mockControl.connector.emit("message", new lib.MessageResponse(
+					0,
+					mockControl.connector.dst,
+					new lib.Address(lib.Address.control, 0, mockControl._nextRequestId - 1),
+					new lib.SubscriptionResponse(event).toJSON()
+				));
+
+				await request;
+				assert.deepEqual(calledWith, event);
+			});
+		});
+	});
+});

test/mock.js

@@ -178,9 +178,11 @@ class MockController {
 		]);
 		this.instances = new Map();
 		this.hosts = new Map();
+		this.handles = new Map();
 	}
 
-	handle() {
+	handle(eventClass, handler) {
+		this.handles.set(eventClass, handler);
 	}
 
 	getControllerUrl() {