Merge pull request #913 from Cooldude2606/refactor/subscriptions-2

Refactor subscription handling and add new filter classes

packages/controller/src/Controller.ts

@@ -875,6 +875,9 @@ export default class Controller {
 	}
 
 	async handleSystemInfoSubscription(request: lib.SubscriptionRequest) {
+		if (request.action === "unsubscribe") {
+			return null;
+		}
 		const systems = [...this.systems.values()].filter(
 			metric => metric.updatedAtMs > request.lastRequestTimeMs,
 		);
@@ -897,6 +900,9 @@ export default class Controller {
 	}
 
 	async handleHostSubscription(request: lib.SubscriptionRequest) {
+		if (request.action === "unsubscribe") {
+			return null;
+		}
 		const hosts = [...this.hosts.values()].filter(
 			host => host.updatedAtMs > request.lastRequestTimeMs,
 		).map(host => host.toHostDetails());
@@ -944,6 +950,9 @@ export default class Controller {
 	}
 
 	async handleInstanceDetailsSubscription(request: lib.SubscriptionRequest) {
+		if (request.action === "unsubscribe") {
+			return null;
+		}
 		const instances = [...this.instances.values()].filter(
 			instance => instance.updatedAtMs > request.lastRequestTimeMs,
 		).map(instance => instance.toInstanceDetails());
@@ -955,7 +964,12 @@ export default class Controller {
 	}
 
 	async handleInstanceSaveDetailsSubscription(request: lib.SubscriptionRequest) {
-		const saves = [...this.saves.values()].filter(save => save.updatedAtMs > request.lastRequestTimeMs);
+		if (request.action === "unsubscribe") {
+			return null;
+		}
+		const saves = [...this.saves.values()].filter(
+			save => save.updatedAtMs > request.lastRequestTimeMs
+		);
 		return saves.length ? new lib.InstanceSaveDetailsUpdatesEvent(saves) : null;
 	}
 
@@ -966,6 +980,9 @@ export default class Controller {
 	}
 
 	async handleModPackSubscription(request: lib.SubscriptionRequest) {
+		if (request.action === "unsubscribe") {
+			return null;
+		}
 		const modPacks = [...this.modPacks.values()].filter(
 			modPack => modPack.updatedAtMs > request.lastRequestTimeMs,
 		);
@@ -979,6 +996,9 @@ export default class Controller {
 	}
 
 	async handleModSubscription(request: lib.SubscriptionRequest) {
+		if (request.action === "unsubscribe") {
+			return null;
+		}
 		const mods = [...this.modStore.files.values()].filter(
 			mod => mod.updatedAtMs > request.lastRequestTimeMs,
 		);
@@ -990,6 +1010,9 @@ export default class Controller {
 	}
 
 	async handleUserSubscription(request: lib.SubscriptionRequest) {
+		if (request.action === "unsubscribe") {
+			return null;
+		}
 		const users = [...this.users.records.values()].filter(
 			user => user.updatedAtMs > request.lastRequestTimeMs,
 		);
@@ -1037,6 +1060,9 @@ export default class Controller {
 	}
 
 	async handleRoleSubscription(request: lib.SubscriptionRequest) {
+		if (request.action === "unsubscribe") {
+			return null;
+		}
 		const roles = [...this.roles.values()].filter(
 			role => role.updatedAtMs > request.lastRequestTimeMs,
 		);

packages/lib/src/data/Permission.ts

@@ -8,16 +8,17 @@ export default class Permission {
 		public name: string,
 		public title: string,
 		public description: string,
-		public grantByDefault?: boolean,
+		public grantByDefault: boolean = false,
 	) { }
 
 	static jsonSchema = Type.Object({
 		name: Type.String(),
 		title: Type.String(),
 		description: Type.String(),
+		grantByDefault: Type.Optional(Type.Boolean()),
 	});
 
 	static fromJSON(json: Static<typeof this.jsonSchema>) {
-		return new this(json.name, json.title, json.description);
+		return new this(json.name, json.title, json.description, Boolean(json.grantByDefault));
 	}
 }

packages/lib/src/data/messages_core.ts

@@ -102,8 +102,16 @@ export class Address {
 	 * Returns a string suitable for indexing the address in a Map
 	 * @returns unique string for this address
 	 */
-	index() {
-		return `${this.type}:${this.id}:${this.requestId}`;
+	requestIndex() {
+		return `${this.type}:${this.id}:${this.requestId}` as const;
+	}
+
+	/**
+	 * Returns a string suitable for indexing the address in a Map
+	 * @returns unique string for this address, excluding required id
+	 */
+	addressIndex() {
+		return `${this.type}:${this.id}` as const;
 	}
 
 	/**

packages/lib/src/link/link.ts

@@ -219,7 +219,7 @@ export class Link {
 
 		if (!message.dst.addressedTo(this.connector.src)) {
 			if (message.type === "response" || message.type === "responseError") {
-				this._forwardedRequests.delete(message.dst.index());
+				this._forwardedRequests.delete(message.dst.requestIndex());
 			}
 			this._routeMessage(message, entry);
 			return;
@@ -566,7 +566,7 @@ export class Link {
 			src: message.src,
 			dst: message.dst,
 		};
-		this._forwardedRequests.set(message.src.index(), pending);
+		this._forwardedRequests.set(message.src.requestIndex(), pending);
 		this.connector.forward(message);
 	}
 

packages/lib/src/subscriptions.ts

@@ -1,13 +1,142 @@
 import { Type, Static } from "@sinclair/typebox";
 import { Link, Event, EventClass, RequestHandler, WebSocketBaseConnector } from "./link";
-import { Address, MessageRequest, IUser, plainJson, JsonBoolean } from "./data";
+import { Address, MessageRequest, IUser, JsonBoolean, StringEnum } from "./data";
 import isDeepStrictEqual from "./is_deep_strict_equal";
-import { RequestError, SessionLost } from "./errors";
+import { RequestError } from "./errors";
 import { logger } from "./logging";
 
 export type SubscriptionRequestHandler<T> = RequestHandler<SubscriptionRequest, Event<T> | null>;
 export type EventSubscriberCallback<T> = (event: T | null, synced: boolean) => void
 
+/**
+ * Represents a set of string subscription filters.
+ * When isAll() is true, all strings match this filter.
+ * When isEmpty() is true, no strings match this filter.
+ * Otherwise, literal string matching is performed against a set.
+ */
+export class SubscriptionFilters {
+	private constructor(
+		private _all: boolean,
+		private _filters: Set<string>,
+	) {}
+
+	static jsonSchema = Type.Array(Type.String());
+
+	static fromJSON(json: Static<typeof this.jsonSchema>) {
+		return new this(false, new Set(json));
+	}
+
+	toJSON() {
+		return [...this._filters];
+	}
+
+	/** Creates a subscription filter accepting all strings */
+	static all() {
+		return new this(true, new Set());
+	}
+
+	isAll() {
+		return this._all;
+	}
+
+	/** Creates a subscription filter rejecting all strings */
+	static empty() {
+		return new this(false, new Set());
+	}
+
+	isEmpty() {
+		return !this._all && this._filters.size === 0;
+	}
+
+	/**
+	 * Creates a subscription filter from undefined, string, or array of strings.
+	 * Strings work as expected, undefined will produce a filter accepting all values.
+	 * @param filters Shorthand representation
+	 */
+	static fromShorthand(filters?: string | string[]) {
+		if (filters === undefined) {
+			return this.all();
+		}
+		if (Array.isArray(filters)) {
+			return new this(false, new Set(filters));
+		}
+		return new this(false, new Set([filters]));
+	}
+
+	toString() {
+		if (this._all) {
+			return "[SubscriptionFilters All]";
+		}
+		if (this._filters.size === 0) {
+			return "[SubscriptionFilters Empty]";
+		}
+		return `[SubscriptionFilters Set<${this._filters.size}>]`;
+	}
+
+	/** Returns true if this filter accepts the given string */
+	accepts(value: string) {
+		return this._all || this._filters.has(value);
+	}
+
+	/** Returns true if this filter extends the other */
+	extends(other: this) {
+		return this._all || (!other._all && [...other._filters.values()].every(value => this._filters.has(value)));
+	}
+
+	/** Returns true if there is overlap between two filters */
+	intersects(other: this) {
+		if (this._all) {
+			return !other.isEmpty();
+		}
+		if (other._all) {
+			return !this.isEmpty();
+		}
+		for (const filter of other._filters) {
+			if (this._filters.has(filter)) {
+				return true;
+			}
+		}
+		return false;
+	}
+
+	/** Add filters from a filter into this one. */
+	union(other: this) {
+		if (this._all) {
+			return; // Already all
+		}
+		if (other._all) {
+			this._all = true;
+			this._filters.clear();
+			return;
+		}
+		for (const filter of other._filters) {
+			this._filters.add(filter);
+		}
+	}
+
+	/**
+	 * Remove filters in a filter from this one.
+	 * You cannot subtract filters from a filter accepting all.
+	 * Subtracting a filter accepting all produces an empty filter.
+	 */
+	subtract(other: this) {
+		if (other._all) {
+			this._all = false;
+			this._filters.clear();
+			return;
+		}
+		if (this._all) {
+			return; // Cannot subtract from all
+		}
+		for (const filter of other._filters) {
+			this._filters.delete(filter);
+		}
+	}
+}
+
+const SubscriptionActionSchema = StringEnum(["subscribe", "unsubscribe", "replace"]);
+type SubscriptionAction = Static<typeof SubscriptionActionSchema>;
+
 /**
  * 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
@@ -20,7 +149,7 @@ export class SubscriptionRequest {
 	static dst = "controller" as const;
 	static permission(user: IUser, message: MessageRequest) {
 		if (typeof message.data === "object" && message.data !== null) {
-			const data = message.data as Static<typeof SubscriptionRequest.jsonSchema>;
+			const data = message.data as Static<typeof this.jsonSchema>;
 			const entry = Link._eventsByName.get(data[0]);
 			if (entry && entry.Event.permission) {
 				if (typeof entry.Event.permission === "string") {
@@ -34,55 +163,86 @@ export class SubscriptionRequest {
 
 	/** Indicates if updates have been sent */
 	static Response = JsonBoolean;
+	public filters: SubscriptionFilters;
+	public action: SubscriptionAction;
+
+	/** @deprecated Use SubscriptionFilters and SubscriptionAction instead */
+	constructor(
+		eventName: string,
+		subscribe: boolean,
+		lastRequestTimeMs: number | undefined,
+		filters?: string | string[],
+	);
+
+	constructor(
+		eventName: string,
+		action: SubscriptionAction,
+		lastRequestTimeMs: number | undefined,
+		filters?: SubscriptionFilters,
+	);
 
 	constructor(
 		public eventName: string,
-		public subscribe: boolean,
+		action: boolean | SubscriptionAction,
 		public lastRequestTimeMs: number = 0,
-		public filters: string | string[] | undefined = undefined,
+		filters: undefined | string | string[] | SubscriptionFilters,
 	) {
 		if (!Link._eventsByName.has(eventName)) {
 			throw new Error(`Unregistered Event class ${eventName}`);
 		}
+		if (filters instanceof SubscriptionFilters) { // Easier migration for plugins
+			this.filters = filters;
+		} else {
+			this.filters = SubscriptionFilters.fromShorthand(filters);
+		}
+		if (typeof action === "boolean") { // Easier migration for plugins
+			this.action = action ? "subscribe" : "unsubscribe";
+		} else {
+			this.action = action;
+		}
 	}
 
 	static jsonSchema = Type.Union([
 		Type.Tuple([
 			Type.String(),
-			Type.Boolean(),
+			SubscriptionActionSchema,
 			Type.Number(),
 		]),
 		Type.Tuple([
 			Type.String(),
-			Type.Boolean(),
+			SubscriptionActionSchema,
 			Type.Number(),
-			Type.Union([Type.String(), Type.Array(Type.String())]),
+			SubscriptionFilters.jsonSchema,
 		]),
 	]);
 
 	toJSON() {
-		if (this.filters !== undefined) {
-			return [this.eventName, this.subscribe, this.lastRequestTimeMs, this.filters];
+		if (!this.filters.isAll()) {
+			return [this.eventName, this.action, this.lastRequestTimeMs, this.filters.toJSON()];
 		}
-		return [this.eventName, this.subscribe, this.lastRequestTimeMs];
+		return [this.eventName, this.action, this.lastRequestTimeMs];
 	}
 
-	static fromJSON(json: Static<typeof SubscriptionRequest.jsonSchema>): SubscriptionRequest {
-		const [eventName, subscribe, lastRequestTimeMs, filters] = json;
-		return new this(eventName, subscribe, lastRequestTimeMs, filters);
+	static fromJSON(json: Static<typeof this.jsonSchema>) {
+		const [eventName, action, lastRequestTimeMs, filters] = json;
+		return new this(
+			eventName,
+			action,
+			lastRequestTimeMs,
+			filters ? SubscriptionFilters.fromJSON(filters) : SubscriptionFilters.all()
+		);
 	}
 }
 
 type Subscriber = {
 	link: Link,
 	dst: Address,
-	/** Optional filters this subscriber is interested in. Undefined means all events */
-	filters?: string[],
+	filters: SubscriptionFilters,
 }
 
 type EventData = {
 	subscriptionUpdate?: SubscriptionRequestHandler<unknown>,
-	subscriptions: Array<Subscriber>,
+	subscriptions: Map<ReturnType<Address["addressIndex"]>, Subscriber>,
 };
 
 /**
@@ -92,13 +252,6 @@ type EventData = {
 export class SubscriptionController {
 	_events = new Map<string, EventData>();
 
-	private static normalizeFilters(value?: string | string[]): string[] | undefined {
-		if (value === undefined) {
-			return undefined;
-		}
-		return Array.isArray(value) ? value : [value];
-	}
-
 	/**
 	 * 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 a client subscribes
@@ -116,7 +269,7 @@ export class SubscriptionController {
 		}
 		this._events.set(entry.name, {
 			subscriptionUpdate: subscriptionUpdate,
-			subscriptions: [],
+			subscriptions: new Map(),
 		});
 	}
 
@@ -124,7 +277,7 @@ export class SubscriptionController {
 	 * Broadcast an event to all subscribers of that event
 	 * @param event - Event to broadcast.
 	 */
-	broadcast<T>(event: Event<T>, filter?: string | string[]) {
+	broadcast<T extends Event<T>>(event: T & { filters?: string | string[] }, filters?: string | string[]) {
 		const entry = Link._eventsByClass.get(event.constructor);
 		if (!entry) {
 			throw new Error(`Unregistered Event class ${event.constructor.name}`);
@@ -133,24 +286,17 @@ export class SubscriptionController {
 		if (!eventData) {
 			throw new Error(`Event ${entry.name} is not a registered as subscribable`);
 		}
-		const toRemove: Subscriber[] = [];
-		const filtersToMatch = SubscriptionController.normalizeFilters(filter);
 
-		for (const subscriber of eventData.subscriptions) {
+		const broadcastFilters = SubscriptionFilters.fromShorthand(filters ?? event.filters);
+		for (const [addressIndex, subscriber] of eventData.subscriptions) {
 			if ((subscriber.link.connector as WebSocketBaseConnector).closing) {
-				toRemove.push(subscriber);
+				eventData.subscriptions.delete(addressIndex);
 				continue;
 			}
-
-			if (
-				filtersToMatch === undefined
-				|| subscriber.filters === undefined
-				|| subscriber.filters.some(f => filtersToMatch.includes(f))
-			) {
+			if (subscriber.filters.intersects(broadcastFilters)) {
 				subscriber.link.sendTo(subscriber.dst, event);
 			}
 		}
-		eventData.subscriptions = eventData.subscriptions.filter(subscriber => !toRemove.includes(subscriber));
 	}
 
 	/**
@@ -159,10 +305,9 @@ export class SubscriptionController {
 	 * @param address - Address to stop sending events to.
 	 */
 	unsubscribeAddress(address: Address) {
-		for (let eventData of this._events.values()) {
-			eventData.subscriptions = eventData.subscriptions.filter(
-				subscriber => !subscriber.dst.addressedTo(address)
-			);
+		const addressIndex = address.addressIndex();
+		for (const eventData of this._events.values()) {
+			eventData.subscriptions.delete(addressIndex);
 		}
 	}
 
@@ -172,69 +317,76 @@ export class SubscriptionController {
 	 * @param link - Link to stop sending events to.
 	 */
 	unsubscribeLink(link: Link) {
-		for (let eventData of this._events.values()) {
-			eventData.subscriptions = eventData.subscriptions.filter(
-				subscriber => subscriber.link !== link
-			);
+		for (const eventData of this._events.values()) {
+			for (const [addressIndex, subscriber] of eventData.subscriptions) {
+				if (subscriber.link === link) {
+					eventData.subscriptions.delete(addressIndex);
+				}
+			}
 		}
 	}
 
 	/**
 	 * Handle incoming subscription requests on a link
 	 * @param link - Link message was received on
-	 * @param event - incomming event.
+	 * @param request - incoming request.
 	 * @param src - Source address of incomming request.
 	 * @param dst - destination address of incomming request.
 	 */
-	async handleRequest(link: Link, event: SubscriptionRequest, src: Address, dst: Address) {
-		if (!Link._eventsByName.has(event.eventName)) {
-			throw new Error(`Event ${event.eventName} is not a registered event`);
+	async handleRequest(link: Link, request: SubscriptionRequest, src: Address, dst: Address) {
+		if (!Link._eventsByName.has(request.eventName)) {
+			throw new Error(`Event ${request.eventName} is not a registered event`);
 		}
-		const eventData = this._events.get(event.eventName);
+		const eventData = this._events.get(request.eventName);
 		if (!eventData) {
-			throw new Error(`Event ${event.eventName} is not a registered as subscribable`);
+			throw new Error(`Event ${request.eventName} is not a registered as subscribable`);
 		}
-		if (event.subscribe === false) {
-			// Unsubscribe logic
-			const subscriber = eventData.subscriptions.find(sub => sub.dst.addressedTo(src));
-			if (!subscriber) {
-				return false;
-			}
-			if (event.filters === undefined || subscriber.filters === undefined) {
-				// Remove entire subscription for this address
-				eventData.subscriptions = eventData.subscriptions.filter(sub => !sub.dst.addressedTo(src));
-			} else {
-				const filtersToRemove = Array.isArray(event.filters) ? event.filters : [event.filters];
-				subscriber.filters = subscriber.filters.filter(f => !filtersToRemove.includes(f));
-				if (subscriber.filters.length === 0) {
-					// No filters left, remove subscription
-					eventData.subscriptions = eventData.subscriptions.filter(sub => sub !== subscriber);
+		const addressIndex = src.addressIndex();
+		const subscriber = eventData.subscriptions.get(addressIndex);
+		switch (request.action) {
+			case "unsubscribe":
+				if (!subscriber) {
+					return false;
 				}
-			}
-		} else {
-			// Subscribe logic
-			let subscriber = eventData.subscriptions.find(sub => sub.dst.addressedTo(src));
-			const newFilters = SubscriptionController.normalizeFilters(event.filters);
-			if (!subscriber) {
-				eventData.subscriptions.push({ link: link, dst: src, filters: newFilters });
-			} else if (subscriber.filters === undefined || newFilters === undefined) {
-				// Either was already subscribed to all or new request wants all
-				subscriber.filters = undefined;
-			} else {
-				for (const f of newFilters) {
-					if (!subscriber.filters.includes(f)) {
-						subscriber.filters.push(f);
+				subscriber.filters.subtract(request.filters);
+				if (subscriber.filters.isEmpty()) {
+					eventData.subscriptions.delete(addressIndex);
+				}
+				break;
+
+			case "subscribe":
+				if (!subscriber) {
+					eventData.subscriptions.set(addressIndex, { link: link, dst: src, filters: request.filters });
+				} else {
+					subscriber.filters.union(request.filters);
+				}
+				break;
+
+			case "replace":
+				if (request.filters.isEmpty()) {
+					if (!subscriber) {
+						return false;
 					}
+					eventData.subscriptions.delete(addressIndex);
+				} else if (!subscriber) {
+					eventData.subscriptions.set(addressIndex, { link: link, dst: src, filters: request.filters });
+				} else {
+					subscriber.filters = request.filters;
 				}
-			}
-			if (eventData.subscriptionUpdate) {
-				const eventReplay = await eventData.subscriptionUpdate(event, src, dst);
-				if (eventReplay) {
-					link.sendTo(src, eventReplay);
-					return true;
-				}
+				break;
+
+			default:
+				throw new Error(`unreachable case: ${String(request.action)}`);
+		}
+
+		if (eventData.subscriptionUpdate) {
+			const eventReplay = await eventData.subscriptionUpdate(request, src, dst);
+			if (eventReplay) {
+				link.sendTo(src, eventReplay);
+				return true;
 			}
 		}
+
 		return false;
 	}
 }
@@ -245,36 +397,27 @@ export interface SubscribableValue {
 	isDeleted: boolean,
 }
 
-export type EventSubscribable<T, V extends SubscribableValue> = Event<T> & Partial<{
-	updates: V[],
-}>
-
 /**
  * Component for subscribing to and tracking updates of a remote resource
  * Multiple handlers can be subscribed at the same time
  */
-export class EventSubscriber<
-	T extends EventSubscribable<T, V>,
-	K extends string | number = NonNullable<T["updates"]>[number]["id"],
-	V extends SubscribableValue = NonNullable<T["updates"]>[number],
-> {
+export class EventSubscriber<E, S = null> {
 	/** The time at which an event was last received, is less than 0 when there have been no events */
-	lastResponseTimeMs = -1;
-	/** Values of the subscribed resource */
-	values = new Map<K, V>();
+	lastUpdatedMs = -1;
 	/** True if this subscriber is currently synced with the source */
 	synced = false;
-	/** True if this subscriber is expecting to receive updates from the source */
-	_syncing = false;
 	/** Repeat calls to getSnapshot will return the same readonly copy unless values has updated */
-	_snapshot: readonly [ReadonlyMap<K, Readonly<V>>, boolean] = [new Map<K, V>(), false];
-	_snapshotLastUpdatedMs = -1;
+	private _snapshot: readonly [S, boolean] = [this.makeSnapshot(), false];
+	private _snapshotLastUpdatedMs = -1;
 	/** Callbacks will be called when an event is received or the synced state changes */
-	_callbacks = new Array<EventSubscriberCallback<T>>();
+	private _callbacks = new Array<EventSubscriberCallback<E>>();
+	/** Filters applied to the filtered handler, if empty the filtered handler will not be called */
+	private _filters = SubscriptionFilters.empty();
 
 	constructor(
-		private Event: EventClass<T>,
+		protected Event: EventClass<E>,
 		public control: Link,
+		protected filteredHandler?: EventSubscriberCallback<E>,
 	) {
 		control.handle(Event, this._handleEvent.bind(this));
 		// The events below exist only on websocket connectors, we can't early return because of test mocking
@@ -294,12 +437,10 @@ export class EventSubscriber<
 	 */
 	handleConnectionEvent(event: "connect" | "drop" | "resume" | "close") {
 		if (event === "connect" || event === "resume") {
-			this._updateSubscription();
+			this._sendRequest("replace", this._callbacks.length > 0 ? SubscriptionFilters.all() : this._filters);
 		} else if (this.synced) {
 			this.synced = false;
-			for (let callback of this._callbacks) {
-				callback(null, false);
-			}
+			this._notify(null);
 		}
 	}
 
@@ -308,33 +449,35 @@ export class EventSubscriber<
 	 * @param event - event from subscribed resource
 	 * @internal
 	 */
-	async _handleEvent(event: T) {
-		// We support the automatic maintaining of a map for events with the updates property
-		const updates = event.updates ?? [];
-		for (const value of updates) {
-			this.lastResponseTimeMs = Math.max(this.lastResponseTimeMs, value.updatedAtMs);
-			// Warn about updates which changes content but don't update updatedAtMs
-			const existing = this.values.get(value.id as K);
-			if (existing && existing.updatedAtMs === value.updatedAtMs && !isDeepStrictEqual(existing, value)) {
-				logger.warn(
-					`${this.Event.name} contains update for value with id ${value.id} that has an identical ` +
-					"updatedAtMs timestamp but differing content"
-				);
-			}
-			if (value.isDeleted) {
-				this.values.delete(value.id as K);
-			} else {
-				this.values.set(value.id as K, value);
-			}
-		}
-		// Event handling logic
-		if (this._syncing && !this.synced) {
-			this._snapshotLastUpdatedMs = -1;
-			this.synced = true;
+	private async _handleEvent(event: E) {
+		const eventTimeMs = this.getLastUpdatedTimeMs(event);
+		if (eventTimeMs > this.lastUpdatedMs) {
+			this.lastUpdatedMs = eventTimeMs;
 		}
+		this.synced = this._hasSubscriptions();
+		this.processEvent(event);
+		this._notify(event);
+	}
+
+	private _notify(event: E | null) {
 		for (const callback of this._callbacks) {
 			callback(event, this.synced);
 		}
+		if (this.filteredHandler && !this._filters.isEmpty()) {
+			this.filteredHandler(event, this.synced);
+		}
+	}
+
+	protected getLastUpdatedTimeMs(event: E) {
+		return Date.now();
+	}
+
+	protected processEvent(event: E) {
+		// Noop for EventSubscriber
+	}
+
+	protected makeSnapshot(): S {
+		return null as S;
 	}
 
 	/**
@@ -344,10 +487,10 @@ export class EventSubscriber<
 	 *     synced property changes, in which case the event will be null.
 	 * @returns function that will unsubscribe from notifications
 	 */
-	subscribe(handler: EventSubscriberCallback<T>) {
+	subscribe(handler: EventSubscriberCallback<E>) {
 		this._callbacks.push(handler);
 		if (this._callbacks.length === 1) {
-			this._updateSubscription();
+			this._sendRequest("subscribe", SubscriptionFilters.all());
 		}
 		return () => {
 			// During a page transition the components currently rendered
@@ -368,52 +511,145 @@ export class EventSubscriber<
 				}
 				this._callbacks.splice(index, 1);
 				if (this._callbacks.length === 0) {
-					this._updateSubscription();
+					this._sendRequest("replace", this._filters);
 				}
 			});
 		};
 	}
 
+	/** Returns true if all filters are present */
+	hasFilters(filters: string | string[] | SubscriptionFilters) {
+		if (!(filters instanceof SubscriptionFilters)) {
+			filters = SubscriptionFilters.fromShorthand(filters);
+		}
+		return this._filters.extends(filters);
+	}
+
+	/** Add filters to the filtered handler */
+	addFilters(filters: string | string[] | SubscriptionFilters) {
+		if (!(filters instanceof SubscriptionFilters)) {
+			filters = SubscriptionFilters.fromShorthand(filters);
+		}
+		this._filters.union(filters);
+		this._sendRequest("subscribe", filters);
+	}
+
+	/** Remove filters from the filtered handler */
+	removeFilters(filters: string | string[] | SubscriptionFilters) {
+		if (!(filters instanceof SubscriptionFilters)) {
+			filters = SubscriptionFilters.fromShorthand(filters);
+		}
+		this._filters.subtract(filters);
+		this._sendRequest("unsubscribe", filters);
+	}
+
+	/** Clear all filters from the filtered handler, it will no longer be called */
+	clearFilters() {
+		this._filters = SubscriptionFilters.empty();
+		if (!this._hasSubscriptions()) {
+			this._sendRequest("unsubscribe", SubscriptionFilters.all());
+		}
+	}
+
 	/**
 	 * Obtain a snapshot of the current state of the tracked resource
 	 * @returns tuple of values map snapshot and synced property.
 	 */
-	getSnapshot() {
-		if (this._snapshotLastUpdatedMs !== this.lastResponseTimeMs) {
-			this._snapshotLastUpdatedMs = this.lastResponseTimeMs;
-			this._snapshot = [new Map(this.values), this.synced];
+	getSnapshot(): readonly [S, boolean] {
+		if (this._snapshotLastUpdatedMs !== this.lastUpdatedMs || this._snapshot[1] !== this.synced) {
+			this._snapshotLastUpdatedMs = this.lastUpdatedMs;
+			this._snapshot = [this.makeSnapshot(), this.synced];
 		}
-		return this._snapshot!;
+		return this._snapshot;
+	}
+
+	/** Returns true if there are any active subscription callbacks */
+	private _hasSubscriptions() {
+		return this._callbacks.length > 0 || !this._filters.isEmpty();
 	}
 
 	/**
 	 * Update the subscription with the controller based on current handler counts
 	 */
-	async _updateSubscription() {
+	private async _sendRequest(action: SubscriptionAction, filters: SubscriptionFilters) {
 		if (!this.control.connector.valid) {
 			return;
 		}
-		const entry = Link._eventsByClass.get(this.Event)!;
 
+		const entry = Link._eventsByClass.get(this.Event)!;
 		try {
-			this._syncing = this._callbacks.length > 0;
 			const updatesSent = await this.control.sendTo("controller", new SubscriptionRequest(
-				entry.name,
-				this._syncing,
-				this.lastResponseTimeMs
+				entry.name, action, this.lastUpdatedMs, filters
 			));
 			if (!updatesSent) {
-				this.synced = this._syncing;
-				for (const callback of this._callbacks) {
-					callback(null, this.synced);
-				}
+				this.synced = this._hasSubscriptions();
+				this._notify(null);
 			}
 		} catch (err: any) {
 			if (!(err instanceof RequestError) || err.code !== "SessionLost") {
-				logger.error(
-					`Unexpected error updating ${entry.name} subscription:\n${err.stack}`
-				);
+				logger.error(`Unexpected error updating ${entry.name} subscription:\n${err.stack}`);
 			}
 		}
 	}
 }
+
+export class ValueSubscriber<
+	E extends { value: V },
+	V extends SubscribableValue = E["value"],
+> extends EventSubscriber<E, Readonly<V> | null> {
+	/** Current value of the subscribed resource */
+	value: V | null = null;
+
+	protected override getLastUpdatedTimeMs(event: E) {
+		return event.value.updatedAtMs;
+	}
+
+	protected override processEvent(event: E) {
+		this.value = event.value;
+	}
+
+	protected override makeSnapshot() {
+		return this.value;
+	}
+}
+
+export class MapSubscriber<
+	E extends { updates: V[] },
+	K extends string | number = E["updates"][number]["id"],
+	V extends SubscribableValue = E["updates"][number],
+> extends EventSubscriber<E, ReadonlyMap<K, Readonly<V>>> {
+	/** Current value of the subscribed resource */
+	values = new Map<K, V>();
+
+	protected override getLastUpdatedTimeMs(event: E) {
+		let maxTime = this.lastUpdatedMs;
+		for (const update of event.updates) {
+			if (update.updatedAtMs > maxTime) {
+				maxTime = update.updatedAtMs;
+			}
+		}
+		return maxTime;
+	}
+
+	protected override processEvent(event: E) {
+		for (const value of event.updates) {
+			// Warn about updates which changes content but don't update updatedAtMs
+			const existing = this.values.get(value.id as K);
+			if (existing && existing.updatedAtMs === value.updatedAtMs && !isDeepStrictEqual(existing, value)) {
+				logger.warn(
+					`${this.Event.name} contains update for value with id ${value.id} that has an identical ` +
+					"updatedAtMs timestamp but differing content"
+				);
+			}
+			if (value.isDeleted) {
+				this.values.delete(value.id as K);
+			} else {
+				this.values.set(value.id as K, value);
+			}
+		}
+	}
+
+	protected override makeSnapshot() {
+		return new Map(this.values);
+	}
+}

packages/web_ui/src/util/websocket.ts

@@ -60,14 +60,14 @@ export class Control extends lib.Link {
 	logHandlers: Map<lib.LogFilter, logHandler[]> = new Map();
 
 	/** Updates handled by the subscription service */
-	systems = new lib.EventSubscriber(lib.SystemInfoUpdateEvent, this);
-	hosts = new lib.EventSubscriber(lib.HostUpdatesEvent, this);
-	instances = new lib.EventSubscriber(lib.InstanceDetailsUpdatesEvent, this);
-	saves = new lib.EventSubscriber(lib.InstanceSaveDetailsUpdatesEvent, this);
-	modPacks = new lib.EventSubscriber(lib.ModPackUpdatesEvent, this);
-	mods = new lib.EventSubscriber(lib.ModUpdatesEvent, this);
-	users = new lib.EventSubscriber(lib.UserUpdatesEvent, this);
-	roles = new lib.EventSubscriber(lib.RoleUpdatesEvent, this);
+	systems = new lib.MapSubscriber(lib.SystemInfoUpdateEvent, this);
+	hosts = new lib.MapSubscriber(lib.HostUpdatesEvent, this);
+	instances = new lib.MapSubscriber(lib.InstanceDetailsUpdatesEvent, this);
+	saves = new lib.MapSubscriber(lib.InstanceSaveDetailsUpdatesEvent, this);
+	modPacks = new lib.MapSubscriber(lib.ModPackUpdatesEvent, this);
+	mods = new lib.MapSubscriber(lib.ModUpdatesEvent, this);
+	users = new lib.MapSubscriber(lib.UserUpdatesEvent, this);
+	roles = new lib.MapSubscriber(lib.RoleUpdatesEvent, this);
 
 	/** Cache for factorio versions to avoid repeat calls to the controller */
 	factorioVersions = new lib.ValueCache(this.requestFactorioVersions.bind(this));

test/integration/clusterio.js

@@ -1655,12 +1655,13 @@ describe("Integration of Clusterio", function() {
 			});
 			it("should grant default permissions with --grant-default", async function() {
 				await execCtl("role edit new --grant-default");
-				let roles = await getControl().send(new lib.RoleListRequest());
+				const control = getControl();
+				let roles = await control.send(new lib.RoleListRequest());
 				let newRole = roles.find(role => role.name === "new");
-				let defaultPermissions = [...lib.permissions.values()]
+				const permissions = await control.send(new lib.PermissionListRequest());
+				const defaultPermissions = permissions
 					.filter(p => p.grantByDefault)
-					.map(p => p.name)
-				;
+					.map(p => p.name);
 				assert.deepEqual(new Set(newRole.permissions), new Set(defaultPermissions));
 			});
 		});

test/lib/config/validators.test.js

@@ -240,14 +240,16 @@ describe("lib/config/definitions/validators", function() {
 		});
 
 		describe("plugin.load_plugin", function() {
-			const initialConfigFields = {
-				"global_chat.load_plugin": false,
-				"inventory_sync.load_plugin": false,
-				"player_auth.load_plugin": false,
-				"research_sync.load_plugin": false,
-				"statistics_exporter.load_plugin": false,
-				"subspace_storage.load_plugin": false,
-			};
+			const initialConfigFields = {};
+
+			before(function() {
+				for (const fieldName of Object.keys(InstanceConfig.fieldDefinitions)) {
+					if (fieldName.endsWith("load_plugin")) {
+						initialConfigFields[fieldName] = false;
+					}
+				}
+			});
+
 			it("should validate against factorio.enable_save_patching", function() {
 				class TestInstanceConfig extends InstanceConfig {
 					static fieldDefinitions = { ...InstanceConfig.fieldDefinitions };

test/lib/subscriptions.js

@@ -1,703 +0,0 @@
-"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;
-		constructor(updates = []) {
-			this.updates = updates;
-		}
-	}
-	lib.Link.register(RegisteredEvent);
-
-	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 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, 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 round trip json serialisable with a single filter", function() {
-			const request = new lib.SubscriptionRequest(RegisteredEvent.name, true, 456, "foo");
-			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 round trip json serialisable with multiple filters", function() {
-			const request = new lib.SubscriptionRequest(RegisteredEvent.name, true, 789, ["foo", "bar"]);
-			const json = JSON.stringify(request);
-			assert.equal(typeof json, "string");
-			const reconstructed = lib.SubscriptionRequest.fromJSON(JSON.parse(json));
-			assert.deepEqual(reconstructed, request);
-		});
-
-		it("should 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),
-			dst: addr({ controlId: 0 }),
-			src: addr("controller"),
-		}, {
-			id: 1,
-			request: new lib.SubscriptionRequest(RegisteredEvent.name, true),
-			dst: addr({ controlId: 1 }),
-			src: addr("controller"),
-		}];
-
-		function getLink(id) {
-			return mockController.wsServer.controlConnections.get(id);
-		}
-		function onceConnectorSend(id) {
-			return events.once(getLink(id).connector, "send");
-		}
-
-		function assertLastEvent(connectorId, Event) {
-			const link = getLink(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 = getLink(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);
-		});
-
-		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);
-				for (let connectorData of connectorSetupDate) {
-					subscriptions.handleRequest(
-						getLink(connectorData.id), connectorData.request, connectorData.dst, connectorData.src
-					);
-				}
-			});
-
-			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", async function() {
-				subscriptions.broadcast(new RegisteredEvent());
-				await Promise.all([onceConnectorSend(0), onceConnectorSend(1)]);
-				assertLastEvent(0, RegisteredEvent); // RegisteredEvent: All
-				assertLastEvent(1, RegisteredEvent); // RegisteredEvent: All
-			});
-			it("should not notify a link who unsubscribed", async function() {
-				const connectorData = connectorSetupDate[0];
-				const request = new lib.SubscriptionRequest(RegisteredEvent.name, false);
-				await subscriptions.handleRequest(
-					getLink(connectorData.id), request, connectorData.dst, connectorData.src
-				);
-				subscriptions.broadcast(new RegisteredEvent());
-				await onceConnectorSend(1);
-				assertNoEvent(0);
-				assertLastEvent(1, RegisteredEvent);
-			});
-			it("should not notify links which are closed or closing", async function() {
-				const link = getLink(0);
-				link.connector.closing = true;
-				subscriptions.broadcast(new RegisteredEvent());
-				await onceConnectorSend(1);
-				assertNoEvent(0);
-				assertLastEvent(1, RegisteredEvent);
-			});
-		});
-
-		describe("unsubscribe()", function() {
-			beforeEach(function() {
-				subscriptions.handle(RegisteredEvent);
-				for (let connectorData of connectorSetupDate) {
-					subscriptions.handleRequest(
-						getLink(connectorData.id), connectorData.request, connectorData.dst, connectorData.src
-					);
-				}
-			});
-
-			it("should remove an active subscription link", async function() {
-				const link = getLink(0);
-				subscriptions.unsubscribeLink(link);
-				subscriptions.broadcast(new RegisteredEvent());
-				await onceConnectorSend(1);
-				assertNoEvent(0);
-				assertLastEvent(1, RegisteredEvent);
-			});
-			it("should remove an active subscription address", async function() {
-				const link = getLink(0);
-				subscriptions.unsubscribeAddress(connectorSetupDate[0].dst);
-				subscriptions.broadcast(new RegisteredEvent());
-				await onceConnectorSend(1);
-				assertNoEvent(0);
-				assertLastEvent(1, RegisteredEvent);
-			});
-		});
-
-		describe("handleRequest()", function() {
-			beforeEach(function() {
-				subscriptions.handle(RegisteredEvent);
-			});
-
-			it("should not accept subscriptions to unregistered events", async function() {
-				const request = new lib.SubscriptionRequest(RegisteredEvent.name, true);
-				request.eventName = UnregisteredEvent.name;
-				await assert.rejects(
-					subscriptions.handleRequest(getLink(0), request, addr({ controlId: 0 }), addr("controller")),
-					new Error(`Event ${UnregisteredEvent.name} is not a registered event`)
-				);
-			});
-			it("should not accept subscriptions to events not handled by the class", async function() {
-				const request = new lib.SubscriptionRequest(RegisteredEvent.name, true);
-				request.eventName = StringPermissionEvent.name;
-				await assert.rejects(
-					subscriptions.handleRequest(getLink(0), request, addr({ controlId: 0 }), addr("controller")),
-					new Error(`Event ${StringPermissionEvent.name} is not a registered as subscribable`)
-				);
-			});
-			it("should accept a subscription", async function() {
-				const request = new lib.SubscriptionRequest(RegisteredEvent.name, true);
-				await subscriptions.handleRequest(
-					getLink(0), request, connectorSetupDate[0].dst, connectorSetupDate[0].src
-				);
-				subscriptions.broadcast(new RegisteredEvent());
-				await onceConnectorSend(0);
-				assertLastEvent(0, RegisteredEvent);
-			});
-			it("should accept an unsubscription", async function() {
-				const request = new lib.SubscriptionRequest(RegisteredEvent.name, true);
-				await subscriptions.handleRequest(
-					getLink(0), request, connectorSetupDate[0].dst, connectorSetupDate[0].src
-				);
-				subscriptions.broadcast(new RegisteredEvent());
-				await onceConnectorSend(0);
-				assertLastEvent(0, RegisteredEvent);
-
-				const link = getLink(0);
-				link.connector.sentMessages.pop();
-
-				const unsubRequest = new lib.SubscriptionRequest(RegisteredEvent.name, false);
-				await subscriptions.handleRequest(
-					getLink(0), unsubRequest, connectorSetupDate[0].dst, connectorSetupDate[0].src
-				);
-				subscriptions.broadcast(new RegisteredEvent());
-				assertNoEvent(0);
-			});
-			it("should accept a response with an event replay when returned by the handler", async function() {
-				subscriptions.handle(StringPermissionEvent, async function() {
-					return new StringPermissionEvent();
-				});
-
-				const request = new lib.SubscriptionRequest(StringPermissionEvent.name, true);
-				await subscriptions.handleRequest(
-					getLink(0), request, connectorSetupDate[0].dst, connectorSetupDate[0].src
-				);
-				await onceConnectorSend(0);
-				assertLastEvent(0, StringPermissionEvent);
-
-				const link = getLink(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());
-			});
-			it("should accept a response with null when returned by the handler", async function() {
-				subscriptions.handle(StringPermissionEvent, async function() {
-					return null;
-				});
-
-				const request = new lib.SubscriptionRequest(StringPermissionEvent.name, true);
-				await subscriptions.handleRequest(
-					getLink(0), request, connectorSetupDate[0].dst, connectorSetupDate[0].src
-				);
-				await new Promise(resolve => setImmediate(resolve));
-				assertNoEvent(0);
-			});
-
-			describe("filters()", function() {
-				it("should notify only subscribers matching the broadcast filter", async function() {
-					const reqFoo = new lib.SubscriptionRequest(RegisteredEvent.name, true, 0, "foo");
-					await subscriptions.handleRequest(
-						getLink(0), reqFoo, connectorSetupDate[0].dst, connectorSetupDate[0].src
-					);
-					const reqBar = new lib.SubscriptionRequest(RegisteredEvent.name, true, 0, "bar");
-					await subscriptions.handleRequest(
-						getLink(1), reqBar, connectorSetupDate[1].dst, connectorSetupDate[1].src
-					);
-
-					const before0 = getLink(0).connector.sentMessages.length;
-					const before1 = getLink(1).connector.sentMessages.length;
-
-					subscriptions.broadcast(new RegisteredEvent(), "foo");
-					await onceConnectorSend(0);
-					assert.equal(getLink(0).connector.sentMessages.length, before0 + 1);
-					assert.equal(getLink(1).connector.sentMessages.length, before1);
-
-					subscriptions.broadcast(new RegisteredEvent(), "bar");
-					await onceConnectorSend(1);
-					assert.equal(getLink(0).connector.sentMessages.length, before0 + 1);
-					assert.equal(getLink(1).connector.sentMessages.length, before1 + 1);
-
-					const beforeBaz0 = getLink(0).connector.sentMessages.length;
-					const beforeBaz1 = getLink(1).connector.sentMessages.length;
-					subscriptions.broadcast(new RegisteredEvent(), "baz");
-					await new Promise(resolve => setImmediate(resolve));
-					assert.equal(getLink(0).connector.sentMessages.length, beforeBaz0);
-					assert.equal(getLink(1).connector.sentMessages.length, beforeBaz1);
-				});
-
-				it("should notify subscribers with no filters for any broadcast filter", async function() {
-					const reqAll = new lib.SubscriptionRequest(RegisteredEvent.name, true);
-					await subscriptions.handleRequest(
-						getLink(0), reqAll, connectorSetupDate[0].dst, connectorSetupDate[0].src
-					);
-					const reqBaz = new lib.SubscriptionRequest(RegisteredEvent.name, true, 0, "baz");
-					await subscriptions.handleRequest(
-						getLink(1), reqBaz, connectorSetupDate[1].dst, connectorSetupDate[1].src
-					);
-
-					const before0 = getLink(0).connector.sentMessages.length;
-					const before1 = getLink(1).connector.sentMessages.length;
-
-					subscriptions.broadcast(new RegisteredEvent(), "foo");
-					await onceConnectorSend(0);
-					assert.equal(getLink(0).connector.sentMessages.length, before0 + 1);
-					assert.equal(getLink(1).connector.sentMessages.length, before1);
-
-					subscriptions.broadcast(new RegisteredEvent(), "baz");
-					await Promise.all([onceConnectorSend(0), onceConnectorSend(1)]);
-					assert.equal(getLink(0).connector.sentMessages.length, before0 + 2);
-					assert.equal(getLink(1).connector.sentMessages.length, before1 + 1);
-				});
-
-				it("should merge multiple filter subscriptions for the same subscriber", async function() {
-					const reqFoo = new lib.SubscriptionRequest(RegisteredEvent.name, true, 0, "alpha");
-					await subscriptions.handleRequest(
-						getLink(0), reqFoo, connectorSetupDate[0].dst, connectorSetupDate[0].src
-					);
-					const reqBar = new lib.SubscriptionRequest(RegisteredEvent.name, true, 0, "beta");
-					await subscriptions.handleRequest(
-						getLink(0), reqBar, connectorSetupDate[0].dst, connectorSetupDate[0].src
-					);
-
-					const before0 = getLink(0).connector.sentMessages.length;
-
-					subscriptions.broadcast(new RegisteredEvent(), "alpha");
-					await onceConnectorSend(0);
-					assert.equal(getLink(0).connector.sentMessages.length, before0 + 1);
-
-					subscriptions.broadcast(new RegisteredEvent(), "beta");
-					await onceConnectorSend(0);
-					assert.equal(getLink(0).connector.sentMessages.length, before0 + 2);
-
-					subscriptions.broadcast(new RegisteredEvent(), "gamma");
-					await new Promise(resolve => setImmediate(resolve));
-					assert.equal(getLink(0).connector.sentMessages.length, before0 + 2);
-				});
-
-				it("should reset to all when subscribing with no filters after having filters", async function() {
-					const reqFoo = new lib.SubscriptionRequest(RegisteredEvent.name, true, 0, "foo");
-					await subscriptions.handleRequest(
-						getLink(0), reqFoo, connectorSetupDate[0].dst, connectorSetupDate[0].src
-					);
-					const reqAll = new lib.SubscriptionRequest(RegisteredEvent.name, true);
-					await subscriptions.handleRequest(
-						getLink(0), reqAll, connectorSetupDate[0].dst, connectorSetupDate[0].src
-					);
-
-					subscriptions.broadcast(new RegisteredEvent(), "baz");
-					await onceConnectorSend(0);
-				});
-
-				it("should remove only specified filters on unsubscribe with filters", async function() {
-					const reqFilters = new lib.SubscriptionRequest(RegisteredEvent.name, true, 0, ["foo", "bar"]);
-					await subscriptions.handleRequest(
-						getLink(0), reqFilters, connectorSetupDate[0].dst, connectorSetupDate[0].src
-					);
-
-					const unsubFoo = new lib.SubscriptionRequest(RegisteredEvent.name, false, 0, "foo");
-					await subscriptions.handleRequest(
-						getLink(0), unsubFoo, connectorSetupDate[0].dst, connectorSetupDate[0].src
-					);
-
-					const before0 = getLink(0).connector.sentMessages.length;
-
-					subscriptions.broadcast(new RegisteredEvent(), "foo");
-					await new Promise(resolve => setImmediate(resolve));
-					assert.equal(getLink(0).connector.sentMessages.length, before0);
-
-					subscriptions.broadcast(new RegisteredEvent(), "bar");
-					await onceConnectorSend(0);
-					assert.equal(getLink(0).connector.sentMessages.length, before0 + 1);
-				});
-
-				it("should remove entire subscription when last filter is unsubscribed", async function() {
-					const reqFoo = new lib.SubscriptionRequest(RegisteredEvent.name, true, 0, "foo");
-					await subscriptions.handleRequest(
-						getLink(0), reqFoo, connectorSetupDate[0].dst, connectorSetupDate[0].src
-					);
-
-					const unsubFoo = new lib.SubscriptionRequest(RegisteredEvent.name, false, 0, "foo");
-					await subscriptions.handleRequest(
-						getLink(0), unsubFoo, connectorSetupDate[0].dst, connectorSetupDate[0].src
-					);
-
-					subscriptions.broadcast(new RegisteredEvent(), "foo");
-					await new Promise(resolve => setImmediate(resolve));
-					assertNoEvent(0);
-				});
-
-				// eslint-disable-next-line max-len
-				it("should remove entire subscription when unsubscribing any filter from an all-subscription", async function() {
-					const reqAll = new lib.SubscriptionRequest(RegisteredEvent.name, true);
-					await subscriptions.handleRequest(
-						getLink(0), reqAll, connectorSetupDate[0].dst, connectorSetupDate[0].src
-					);
-					const unsubAny = new lib.SubscriptionRequest(RegisteredEvent.name, false, 0, "foo");
-					await subscriptions.handleRequest(
-						getLink(0), unsubAny, connectorSetupDate[0].dst, connectorSetupDate[0].src
-					);
-
-					subscriptions.broadcast(new RegisteredEvent());
-					await new Promise(resolve => setImmediate(resolve));
-					assertNoEvent(0);
-				});
-
-				it("should match any of multiple broadcast filters", async function() {
-					const reqFoo = new lib.SubscriptionRequest(RegisteredEvent.name, true, 0, "foo");
-					await subscriptions.handleRequest(
-						getLink(0), reqFoo, connectorSetupDate[0].dst, connectorSetupDate[0].src
-					);
-					const reqBar = new lib.SubscriptionRequest(RegisteredEvent.name, true, 0, "bar");
-					await subscriptions.handleRequest(
-						getLink(1), reqBar, connectorSetupDate[1].dst, connectorSetupDate[1].src
-					);
-
-					subscriptions.broadcast(new RegisteredEvent(), ["bar", "baz"]);
-					await onceConnectorSend(1);
-					assertNoEvent(0);
-				});
-			});
-		});
-	});
-
-	describe("class EventSubscriber", function() {
-		let mockControl, registeredEvent;
-		beforeEach(function() {
-			mockControl = new MockControl(new MockConnector(
-				addr({ controlId: 0 }),
-				addr("controller"),
-			));
-			registeredEvent = new lib.EventSubscriber(RegisteredEvent, mockControl);
-		});
-
-		function assertLastMessage(message) {
-			const Message = message.constructor;
-			const lastMessage = mockControl.connector.sentMessages.at(-1);
-			const name = Message.plugin ? `${Message.plugin}:${Message.name}` : Message.name;
-			assert.equal(lastMessage.name, name);
-			assert.deepEqual(lastMessage.data, message);
-		}
-
-		function assertNoMessageSent() {
-			assert.equal(mockControl.connector.sentMessages.length, 0, "messages were sent");
-		}
-
-		describe("constructor()", function() {
-			it("should not accept unregistered events", function() {
-				assert.throws(
-					() => new lib.EventSubscriber(UnregisteredEvent, mockControl),
-					new Error(`Unregistered Event class ${UnregisteredEvent.name}`)
-				);
-			});
-			it("should handle the provided event", function() {
-				assert.equal(mockControl._eventHandlers.has(RegisteredEvent), true);
-			});
-		});
-
-		describe("subscribe()", function() {
-			it("should allow subscriptions to an event", async function() {
-				let calledWith = null;
-				let calledWithTwo = null;
-				const item = { id: 1, updatedAtMs: 1, isDeleted: false };
-				const eventUpdate = new RegisteredEvent([item]);
-				registeredEvent.subscribe(function(event, synced) { calledWith = [event, synced]; });
-				registeredEvent.subscribe(function(event, synced) { calledWithTwo = [event, synced]; });
-				await registeredEvent._handleEvent(eventUpdate);
-				assert.deepEqual(calledWith, [eventUpdate, true]);
-				assert.deepEqual(calledWithTwo, [eventUpdate, true]);
-			});
-			it("should allow unsubscribing from an event", async function() {
-				let called = false;
-				let event = new RegisteredEvent();
-				function callback() { called = true; }
-				const unsubscribe = registeredEvent.subscribe(callback);
-				await registeredEvent._handleEvent(event);
-				assert.deepEqual(called, true);
-
-				called = false;
-				event = new RegisteredEvent();
-				unsubscribe();
-				// Wait for unsubscribe to happen.
-				await new Promise(resolve => setImmediate(resolve));
-				await registeredEvent._handleEvent(event);
-				assert.equal(called, false);
-			});
-			it("should not throw error on incorrectly unsubscribing multiple times", async function() {
-				const unsubscribe = registeredEvent.subscribe(() => {});
-				unsubscribe();
-				unsubscribe();
-				// Wait for unsubscribe to happen.
-				await new Promise(resolve => setImmediate(resolve));
-			});
-		});
-
-		describe("getSnapshot()", function() {
-			it("should give an immutable copy of the tracked values", async function() {
-				const update1 = [{ id: 1, updatedAtMs: 1, isDeleted: false }];
-				const update2 = [
-					{ id: 1, updatedAtMs: 2, isDeleted: false },
-					{ id: 2, updatedAtMs: 3, isDeleted: false },
-				];
-				const update3 = [{ id: 1, updatedAtMs: 4, isDeleted: true }];
-				const snap0 = registeredEvent.getSnapshot();
-				assert.deepEqual(snap0, [new Map(), false]);
-				await registeredEvent._handleEvent(new RegisteredEvent(update1));
-				const snap1 = registeredEvent.getSnapshot();
-				assert.deepEqual(snap1, [new Map([[1, update1[0]]]), false]);
-				await registeredEvent._handleEvent(new RegisteredEvent(update2));
-				const snap2 = registeredEvent.getSnapshot();
-				assert.deepEqual(snap2, [new Map([[1, update2[0]], [2, update2[1]]]), false]);
-				await registeredEvent._handleEvent(new RegisteredEvent(update3));
-				const snap3 = registeredEvent.getSnapshot();
-				assert.deepEqual(snap3, [new Map([[2, update2[1]]]), false]);
-				// validate previous snapshots were not modified
-				assert.deepEqual(snap2, [new Map([[1, update2[0]], [2, update2[1]]]), false]);
-				assert.deepEqual(snap1, [new Map([[1, update1[0]]]), false]);
-				assert.deepEqual(snap0, [new Map(), false]);
-			});
-		});
-
-		describe("handleConnectionEvent", function() {
-			it("should be called on connector connect event", async function() {
-				registeredEvent.subscribe(() => {});
-				await events.once(mockControl.connector, "send");
-				mockControl.connector.sentMessages.pop();
-
-				registeredEvent.control.connector.emit("connect");
-				await events.once(mockControl.connector, "send");
-				assertLastMessage(new lib.SubscriptionRequest(RegisteredEvent.name, true, -1));
-				mockControl.connector.sentMessages.pop();
-
-				registeredEvent.handleConnectionEvent("connect");
-				await events.once(mockControl.connector, "send");
-				assertLastMessage(new lib.SubscriptionRequest(RegisteredEvent.name, true, -1));
-				mockControl.connector.sentMessages.pop();
-
-				registeredEvent.handleConnectionEvent("resume");
-				await events.once(mockControl.connector, "send");
-				assertLastMessage(new lib.SubscriptionRequest(RegisteredEvent.name, true, -1));
-			});
-			it("should set synced to false on connection close", async function() {
-				registeredEvent.synced = true;
-				registeredEvent.control.connector.emit("close");
-				assert(!registeredEvent.synced, "synced was not set to false during close event");
-
-				registeredEvent.synced = true;
-				registeredEvent.handleConnectionEvent("drop");
-				assert(!registeredEvent.synced, "synced was not set to false during drop");
-
-				registeredEvent.synced = true;
-				registeredEvent.handleConnectionEvent("close");
-				assert(!registeredEvent.synced, "synced was not set to false during close");
-			});
-			it("should invoke callbacks on connector close", async function() {
-				let calledWith;
-				registeredEvent.subscribe((values, synced) => { calledWith = [values, synced]; });
-				const [msg] = await events.once(mockControl.connector, "send");
-				mockControl.connector.emit("message", new lib.MessageResponse(0, msg.dst, msg.src,
-					lib.SubscriptionRequest.Response.fromJSON(false)
-				));
-				await new Promise(resolve => setImmediate(resolve));
-				assert(registeredEvent.synced, "synced was not set to true before close");
-				calledWith = undefined;
-
-				registeredEvent.control.connector.emit("close");
-				assert(!registeredEvent.synced, "synced was not set to false");
-				assert(calledWith !== undefined, "callback not called");
-				assert.deepEqual(calledWith, [null, false]);
-			});
-		});
-
-		describe("_updateSubscription()", function() {
-			it("should do nothing if connection is not valid", async function() {
-				registeredEvent.control.connector.valid = false;
-				registeredEvent.subscribe(() => true);
-				await new Promise(resolve => setImmediate(resolve));
-				assertNoMessageSent();
-			});
-			it("should correctly request a subscription", async function() {
-				registeredEvent.subscribe(() => true);
-				await events.once(mockControl.connector, "send");
-				assertLastMessage(new lib.SubscriptionRequest(RegisteredEvent.name, true, -1));
-			});
-			it("should correctly request an unsubscription", async function() {
-				function callback() {}
-				const unsubscribe = registeredEvent.subscribe(callback);
-				await events.once(mockControl.connector, "send");
-				unsubscribe();
-				await events.once(mockControl.connector, "send");
-				assertLastMessage(new lib.SubscriptionRequest(RegisteredEvent.name, false, -1));
-			});
-			it("should set synced and invoke callbacks when request is responded to", async function() {
-				let calledWith;
-				registeredEvent.subscribe((values, synced) => { calledWith = [values, synced]; });
-				const [msg] = await events.once(mockControl.connector, "send");
-				assert(!registeredEvent.synced, "synced was set before response");
-				mockControl.connector.emit("message", new lib.MessageResponse(0, msg.dst, msg.src,
-					lib.SubscriptionRequest.Response.fromJSON(false)
-				));
-				await new Promise(resolve => setImmediate(resolve));
-				assert(calledWith !== undefined, "callback not called");
-				assert.deepEqual(calledWith, [null, true]);
-				assert(registeredEvent.synced, "synced was not set");
-			});
-			it("should set synced and invoke callbacks when request is responded to with data", async function() {
-				let calledWith;
-				registeredEvent.subscribe((values, synced) => { calledWith = [values, synced]; });
-				const [msg] = await events.once(mockControl.connector, "send");
-				assert(!registeredEvent.synced, "synced was set before response");
-				mockControl.connector.emit("message", new lib.MessageResponse(0, msg.dst, msg.src,
-					lib.SubscriptionRequest.Response.fromJSON(true)
-				));
-				await new Promise(resolve => setImmediate(resolve));
-
-				assert(!registeredEvent.synced, "synced was set before update");
-				assert(calledWith === undefined, "callback called before update");
-				const eventUpdate = new RegisteredEvent();
-				registeredEvent._handleEvent(eventUpdate);
-
-				assert(registeredEvent.synced, "synced was not set");
-				assert(calledWith !== undefined, "callback not called");
-				assert.deepEqual(calledWith, [eventUpdate, true]);
-			});
-		});
-	});
-});