From ee460d677d07a90e6f008fb8bcc794bfd1fb8145 Mon Sep 17 00:00:00 2001 From: Cooldude2606 <25043174+Cooldude2606@users.noreply.github.com> Date: Wed, 27 May 2026 23:10:59 +0100 Subject: [PATCH 01/13] Rename `libAddress.index` to `requestIndex` and add `addressIndex` --- packages/lib/src/data/messages_core.ts | 12 ++++++++++-- packages/lib/src/link/link.ts | 4 ++-- 2 files changed, 12 insertions(+), 4 deletions(-) diff --git a/packages/lib/src/data/messages_core.ts b/packages/lib/src/data/messages_core.ts index 36b85720..305905e4 100644 --- a/packages/lib/src/data/messages_core.ts +++ b/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; } /** diff --git a/packages/lib/src/link/link.ts b/packages/lib/src/link/link.ts index ac15d5c6..b7bc8c02 100644 --- a/packages/lib/src/link/link.ts +++ b/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); } From 1f27617a27983e908ff9e00e4323beeac80ce966 Mon Sep 17 00:00:00 2001 From: Cooldude2606 <25043174+Cooldude2606@users.noreply.github.com> Date: Wed, 27 May 2026 23:35:31 +0100 Subject: [PATCH 02/13] Use `addressIndex` and a map for subscriptions --- packages/lib/src/subscriptions.ts | 39 +++++++++++++++---------------- 1 file changed, 19 insertions(+), 20 deletions(-) diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index f73374a1..6d0af71a 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -82,7 +82,7 @@ type Subscriber = { type EventData = { subscriptionUpdate?: SubscriptionRequestHandler, - subscriptions: Array, + subscriptions: Map, Subscriber>, }; /** @@ -116,7 +116,7 @@ export class SubscriptionController { } this._events.set(entry.name, { subscriptionUpdate: subscriptionUpdate, - subscriptions: [], + subscriptions: new Map(), }); } @@ -133,12 +133,11 @@ 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 filtersToMatch = SubscriptionController.normalizeFilters(filter); + for (const [addressIndex, subscriber] of eventData.subscriptions) { if ((subscriber.link.connector as WebSocketBaseConnector).closing) { - toRemove.push(subscriber); + eventData.subscriptions.delete(addressIndex); continue; } @@ -150,7 +149,6 @@ export class SubscriptionController { subscriber.link.sendTo(subscriber.dst, event); } } - eventData.subscriptions = eventData.subscriptions.filter(subscriber => !toRemove.includes(subscriber)); } /** @@ -159,10 +157,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,10 +169,12 @@ 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); + } + } } } @@ -194,29 +193,29 @@ export class SubscriptionController { if (!eventData) { throw new Error(`Event ${event.eventName} is not a registered as subscribable`); } + const addressIndex = src.addressIndex(); + const subscriber = eventData.subscriptions.get(addressIndex); 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)); + eventData.subscriptions.delete(addressIndex); } 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); + eventData.subscriptions.delete(addressIndex); } } } 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 }); + eventData.subscriptions.set(addressIndex, { 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; From 4735f0725d39c5df7476f5c066b0c50550523441 Mon Sep 17 00:00:00 2001 From: Cooldude2606 <25043174+Cooldude2606@users.noreply.github.com> Date: Thu, 28 May 2026 00:25:29 +0100 Subject: [PATCH 03/13] Add `SubscriptionFilters` class --- packages/lib/src/subscriptions.ts | 245 ++++++++++++++++++++++-------- 1 file changed, 181 insertions(+), 64 deletions(-) diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index 6d0af71a..0403a7ee 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -8,6 +8,120 @@ import { logger } from "./logging"; export type SubscriptionRequestHandler = RequestHandler | null>; export type EventSubscriberCallback = (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, + ) {} + + static jsonSchema = Type.Array(Type.String()); + + static fromJSON(json: Static) { + 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() { + return `[SubscriptionFilters ${this._all ? "All" : String(this._filters)}]`; + } + + /** Returns true if this filter accepts this string */ + accepts(value: string) { + return this._all || 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 (this._all) { + return; // Cannot subtract from all + } + if (other._all) { + this._all = false; + this._filters.clear(); + } + for (const filter of other._filters) { + this._filters.delete(filter); + } + } +} + /** * 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 +134,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; + const data = message.data as Static; const entry = Link._eventsByName.get(data[0]); if (entry && entry.Event.permission) { if (typeof entry.Event.permission === "string") { @@ -34,16 +148,37 @@ export class SubscriptionRequest { /** Indicates if updates have been sent */ static Response = JsonBoolean; + public filters: SubscriptionFilters; + + /** @deprecated Use SubscriptionFilters instead */ + constructor( + eventName: string, + subscribe: boolean, + lastRequestTimeMs: number | undefined, + filters?: string | string[], + ); + + constructor( + eventName: string, + subscribe: boolean, + lastRequestTimeMs: number | undefined, + filters?: SubscriptionFilters, + ); constructor( public eventName: string, public subscribe: boolean, 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); + } } static jsonSchema = Type.Union([ @@ -56,28 +191,32 @@ export class SubscriptionRequest { Type.String(), Type.Boolean(), 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.subscribe, this.lastRequestTimeMs, this.filters.toJSON()]; } return [this.eventName, this.subscribe, this.lastRequestTimeMs]; } - static fromJSON(json: Static): SubscriptionRequest { + static fromJSON(json: Static) { const [eventName, subscribe, lastRequestTimeMs, filters] = json; - return new this(eventName, subscribe, lastRequestTimeMs, filters); + return new this( + eventName, + subscribe, + 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 = { @@ -92,13 +231,6 @@ type EventData = { export class SubscriptionController { _events = new Map(); - 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 @@ -124,7 +256,7 @@ export class SubscriptionController { * Broadcast an event to all subscribers of that event * @param event - Event to broadcast. */ - broadcast(event: Event, filter?: string | string[]) { + broadcast(event: Event & { filters?: string | string[] }, filters?: string | string[]) { const entry = Link._eventsByClass.get(event.constructor); if (!entry) { throw new Error(`Unregistered Event class ${event.constructor.name}`); @@ -134,18 +266,13 @@ export class SubscriptionController { throw new Error(`Event ${entry.name} is not a registered as subscribable`); } - const filtersToMatch = SubscriptionController.normalizeFilters(filter); + const broadcastFilters = SubscriptionFilters.fromShorthand(filters); for (const [addressIndex, subscriber] of eventData.subscriptions) { if ((subscriber.link.connector as WebSocketBaseConnector).closing) { 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); } } @@ -181,58 +308,48 @@ export class SubscriptionController { /** * 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`); } const addressIndex = src.addressIndex(); const subscriber = eventData.subscriptions.get(addressIndex); - if (event.subscribe === false) { - // Unsubscribe logic - if (!subscriber) { - return false; - } - if (event.filters === undefined || subscriber.filters === undefined) { - // Remove entire subscription for this address - eventData.subscriptions.delete(addressIndex); - } 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 + switch (request.subscribe) { + case false: + if (!subscriber) { + return false; + } + subscriber.filters.subtract(request.filters); + if (subscriber.filters.isEmpty()) { eventData.subscriptions.delete(addressIndex); } - } - } else { - // Subscribe logic - const newFilters = SubscriptionController.normalizeFilters(event.filters); - if (!subscriber) { - eventData.subscriptions.set(addressIndex, { 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); + break; + + case true: + if (!subscriber) { + eventData.subscriptions.set(addressIndex, { link: link, dst: src, filters: request.filters }); + } else { + subscriber.filters.union(request.filters); + } + if (eventData.subscriptionUpdate) { + const eventReplay = await eventData.subscriptionUpdate(request, src, dst); + if (eventReplay) { + link.sendTo(src, eventReplay); + return true; } } - } - 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.subscribe)}`); } return false; } From 9bed1223fd023988753cc6945b9941f5d78cce66 Mon Sep 17 00:00:00 2001 From: Cooldude2606 <25043174+Cooldude2606@users.noreply.github.com> Date: Thu, 28 May 2026 00:33:45 +0100 Subject: [PATCH 04/13] Add option to replace filters --- packages/lib/src/subscriptions.ts | 51 ++++++++++++++++++++++--------- 1 file changed, 36 insertions(+), 15 deletions(-) diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index 0403a7ee..9bf8865d 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -1,6 +1,6 @@ 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, plainJson, JsonBoolean, StringEnum } from "./data"; import isDeepStrictEqual from "./is_deep_strict_equal"; import { RequestError, SessionLost } from "./errors"; import { logger } from "./logging"; @@ -122,6 +122,9 @@ export class SubscriptionFilters { } } +const SubscriptionActionSchema = StringEnum(["subscribe", "unsubscribe", "replace"]); +type SubscriptionAction = Static; + /** * 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 @@ -149,8 +152,9 @@ export class SubscriptionRequest { /** Indicates if updates have been sent */ static Response = JsonBoolean; public filters: SubscriptionFilters; + public action: SubscriptionAction; - /** @deprecated Use SubscriptionFilters instead */ + /** @deprecated Use SubscriptionFilters and SubscriptionAction instead */ constructor( eventName: string, subscribe: boolean, @@ -160,14 +164,14 @@ export class SubscriptionRequest { constructor( eventName: string, - subscribe: boolean, + action: SubscriptionAction, lastRequestTimeMs: number | undefined, filters?: SubscriptionFilters, ); constructor( public eventName: string, - public subscribe: boolean, + action: boolean | SubscriptionAction, public lastRequestTimeMs: number = 0, filters: undefined | string | string[] | SubscriptionFilters, ) { @@ -179,17 +183,22 @@ export class SubscriptionRequest { } 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(), SubscriptionFilters.jsonSchema, ]), @@ -197,16 +206,16 @@ export class SubscriptionRequest { toJSON() { if (!this.filters.isAll()) { - return [this.eventName, this.subscribe, this.lastRequestTimeMs, this.filters.toJSON()]; + 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) { - const [eventName, subscribe, lastRequestTimeMs, filters] = json; + const [eventName, action, lastRequestTimeMs, filters] = json; return new this( eventName, - subscribe, + action, lastRequestTimeMs, filters ? SubscriptionFilters.fromJSON(filters) : SubscriptionFilters.all() ); @@ -322,8 +331,8 @@ export class SubscriptionController { } const addressIndex = src.addressIndex(); const subscriber = eventData.subscriptions.get(addressIndex); - switch (request.subscribe) { - case false: + switch (request.action) { + case "unsubscribe": if (!subscriber) { return false; } @@ -333,7 +342,7 @@ export class SubscriptionController { } break; - case true: + case "subscribe": if (!subscriber) { eventData.subscriptions.set(addressIndex, { link: link, dst: src, filters: request.filters }); } else { @@ -348,8 +357,20 @@ export class SubscriptionController { } 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; + } + default: - throw new Error(`unreachable case: ${String(request.subscribe)}`); + throw new Error(`unreachable case: ${String(request.action)}`); } return false; } @@ -515,7 +536,7 @@ export class EventSubscriber< this._syncing = this._callbacks.length > 0; const updatesSent = await this.control.sendTo("controller", new SubscriptionRequest( entry.name, - this._syncing, + this._syncing ? "subscribe" : "unsubscribe", this.lastResponseTimeMs )); if (!updatesSent) { From 21b050b22025f1172803f2ee65e0551455b3bb4f Mon Sep 17 00:00:00 2001 From: Cooldude2606 <25043174+Cooldude2606@users.noreply.github.com> Date: Thu, 28 May 2026 00:38:15 +0100 Subject: [PATCH 05/13] Call `subscriptionUpdate` for all update cases --- packages/controller/src/Controller.ts | 28 ++++++++++++++++++++++++++- packages/lib/src/subscriptions.ts | 16 ++++++++------- 2 files changed, 36 insertions(+), 8 deletions(-) diff --git a/packages/controller/src/Controller.ts b/packages/controller/src/Controller.ts index 49a18ebe..a62cb353 100644 --- a/packages/controller/src/Controller.ts +++ b/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, ); diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index 9bf8865d..ef391494 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -348,13 +348,6 @@ export class SubscriptionController { } else { subscriber.filters.union(request.filters); } - if (eventData.subscriptionUpdate) { - const eventReplay = await eventData.subscriptionUpdate(request, src, dst); - if (eventReplay) { - link.sendTo(src, eventReplay); - return true; - } - } break; case "replace": @@ -372,6 +365,15 @@ export class SubscriptionController { 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; } } From e0a39e5630516dcf8456e7232ca2a5b3ff5e8eaf Mon Sep 17 00:00:00 2001 From: Cooldude2606 <25043174+Cooldude2606@users.noreply.github.com> Date: Thu, 28 May 2026 00:39:24 +0100 Subject: [PATCH 06/13] Add option for events to define `get filters` --- packages/lib/src/subscriptions.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index ef391494..7b53289e 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -265,7 +265,7 @@ export class SubscriptionController { * Broadcast an event to all subscribers of that event * @param event - Event to broadcast. */ - broadcast(event: Event & { filters?: string | string[] }, filters?: string | string[]) { + broadcast>(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}`); @@ -275,7 +275,7 @@ export class SubscriptionController { throw new Error(`Event ${entry.name} is not a registered as subscribable`); } - const broadcastFilters = SubscriptionFilters.fromShorthand(filters); + const broadcastFilters = SubscriptionFilters.fromShorthand(filters ?? event.filters); for (const [addressIndex, subscriber] of eventData.subscriptions) { if ((subscriber.link.connector as WebSocketBaseConnector).closing) { eventData.subscriptions.delete(addressIndex); From e49b82842d8847e58432d3e1d87d14f51ce77d70 Mon Sep 17 00:00:00 2001 From: Cooldude2606 <25043174+Cooldude2606@users.noreply.github.com> Date: Thu, 28 May 2026 01:12:05 +0100 Subject: [PATCH 07/13] Add `ValueSubscriber` and `MapSubscriber` --- packages/lib/src/subscriptions.ts | 148 +++++++++++++++++--------- packages/web_ui/src/util/websocket.ts | 16 +-- 2 files changed, 107 insertions(+), 57 deletions(-) diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index 7b53289e..14f6067c 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -384,35 +384,25 @@ export interface SubscribableValue { isDeleted: boolean, } -export type EventSubscribable = Event & 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, - K extends string | number = NonNullable[number]["id"], - V extends SubscribableValue = NonNullable[number], -> { +export class EventSubscriber { /** 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(); + 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; + private _syncing = false; /** Repeat calls to getSnapshot will return the same readonly copy unless values has updated */ - _snapshot: readonly [ReadonlyMap>, boolean] = [new Map(), false]; - _snapshotLastUpdatedMs = -1; + private _snapshot: readonly [S, boolean] = [null as S, false]; + private _snapshotLastUpdatedMs = -1; /** Callbacks will be called when an event is received or the synced state changes */ - _callbacks = new Array>(); + private _callbacks = new Array>(); constructor( - private Event: EventClass, + protected Event: EventClass, public control: Link, ) { control.handle(Event, this._handleEvent.bind(this)); @@ -436,9 +426,7 @@ export class EventSubscriber< this._updateSubscription(); } else if (this.synced) { this.synced = false; - for (let callback of this._callbacks) { - callback(null, false); - } + this.notify(null); } } @@ -447,35 +435,38 @@ 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); - } - } + private async _handleEvent(event: E) { // Event handling logic if (this._syncing && !this.synced) { this._snapshotLastUpdatedMs = -1; this.synced = true; } + const eventTimeMs = this.getLastUpdatedTimeMs(event); + if (eventTimeMs > this.lastUpdatedMs) { + this.lastUpdatedMs = eventTimeMs; + } + this.processEvent(event); + this.notify(event); + } + + private notify(event: E | null) { for (const callback of this._callbacks) { callback(event, this.synced); } } + protected getLastUpdatedTimeMs(event: E) { + return Date.now(); + } + + protected processEvent(event: E) { + // Noop for EventSubscriber + } + + protected makeSnapshot(): S { + return null as S; + } + /** * Subscribe to receive all event notifications * @param handler - @@ -483,7 +474,7 @@ export class EventSubscriber< * synced property changes, in which case the event will be null. * @returns function that will unsubscribe from notifications */ - subscribe(handler: EventSubscriberCallback) { + subscribe(handler: EventSubscriberCallback) { this._callbacks.push(handler); if (this._callbacks.length === 1) { this._updateSubscription(); @@ -517,12 +508,12 @@ export class EventSubscriber< * 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; } /** @@ -539,13 +530,11 @@ export class EventSubscriber< const updatesSent = await this.control.sendTo("controller", new SubscriptionRequest( entry.name, this._syncing ? "subscribe" : "unsubscribe", - this.lastResponseTimeMs + this.lastUpdatedMs )); if (!updatesSent) { this.synced = this._syncing; - for (const callback of this._callbacks) { - callback(null, this.synced); - } + this.notify(null); } } catch (err: any) { if (!(err instanceof RequestError) || err.code !== "SessionLost") { @@ -556,3 +545,64 @@ export class EventSubscriber< } } } + +export class ValueSubscriber< + E extends { value: V }, + V extends SubscribableValue = E["value"], +> extends EventSubscriber | 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>> { + /** Current value of the subscribed resource */ + values = new Map(); + + 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); + } +} diff --git a/packages/web_ui/src/util/websocket.ts b/packages/web_ui/src/util/websocket.ts index 8d2d657c..05747892 100644 --- a/packages/web_ui/src/util/websocket.ts +++ b/packages/web_ui/src/util/websocket.ts @@ -60,14 +60,14 @@ export class Control extends lib.Link { logHandlers: Map = 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)); From 4676f1c820ed6e9be86671fa660d32ff7023989d Mon Sep 17 00:00:00 2001 From: Cooldude2606 <25043174+Cooldude2606@users.noreply.github.com> Date: Fri, 29 May 2026 14:08:09 +0100 Subject: [PATCH 08/13] Add `filteredHandler` to `EventSubscriber` --- packages/lib/src/subscriptions.ts | 93 ++++++++++++++++++++++--------- 1 file changed, 66 insertions(+), 27 deletions(-) diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index 14f6067c..05bb1c86 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -1,8 +1,8 @@ import { Type, Static } from "@sinclair/typebox"; import { Link, Event, EventClass, RequestHandler, WebSocketBaseConnector } from "./link"; -import { Address, MessageRequest, IUser, plainJson, JsonBoolean, StringEnum } 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 = RequestHandler | null>; @@ -67,11 +67,16 @@ export class SubscriptionFilters { return `[SubscriptionFilters ${this._all ? "All" : String(this._filters)}]`; } - /** Returns true if this filter accepts this string */ + /** Returns true if this filter accepts all the given filters */ 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) { @@ -393,17 +398,18 @@ export class EventSubscriber { 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 */ - private _syncing = false; /** Repeat calls to getSnapshot will return the same readonly copy unless values has updated */ private _snapshot: readonly [S, boolean] = [null as S, false]; private _snapshotLastUpdatedMs = -1; /** Callbacks will be called when an event is received or the synced state changes */ private _callbacks = new Array>(); + /** Filters applied to the filtered handler, if empty the filtered handler will not be called */ + private _filters = SubscriptionFilters.empty(); constructor( protected Event: EventClass, public control: Link, + protected filteredHandler?: EventSubscriberCallback, ) { control.handle(Event, this._handleEvent.bind(this)); // The events below exist only on websocket connectors, we can't early return because of test mocking @@ -423,10 +429,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; - this.notify(null); + this._notify(null); } } @@ -436,23 +442,22 @@ export class EventSubscriber { * @internal */ private async _handleEvent(event: E) { - // Event handling logic - if (this._syncing && !this.synced) { - this._snapshotLastUpdatedMs = -1; - this.synced = true; - } const eventTimeMs = this.getLastUpdatedTimeMs(event); if (eventTimeMs > this.lastUpdatedMs) { this.lastUpdatedMs = eventTimeMs; } + this.synced = this._hasSubscriptions(); this.processEvent(event); - this.notify(event); + this._notify(event); } - private notify(event: E | null) { + 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) { @@ -477,7 +482,7 @@ export class EventSubscriber { subscribe(handler: EventSubscriberCallback) { 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 @@ -498,12 +503,46 @@ 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. @@ -516,31 +555,31 @@ export class EventSubscriber { 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 ? "subscribe" : "unsubscribe", - this.lastUpdatedMs + entry.name, action, this.lastUpdatedMs, filters )); if (!updatesSent) { - this.synced = this._syncing; - this.notify(null); + 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}`); } } } From 41222343e39a50f02b620cbcb8c6fe4043dd4a8b Mon Sep 17 00:00:00 2001 From: Cooldude2606 <25043174+Cooldude2606@users.noreply.github.com> Date: Fri, 29 May 2026 17:45:14 +0100 Subject: [PATCH 09/13] Add tests --- packages/lib/src/subscriptions.ts | 8 +- test/lib/subscriptions.js | 703 ----------------- test/lib/subscriptions.test.js | 1227 +++++++++++++++++++++++++++++ 3 files changed, 1232 insertions(+), 706 deletions(-) delete mode 100644 test/lib/subscriptions.js create mode 100644 test/lib/subscriptions.test.js diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index 05bb1c86..8d8d5120 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -114,12 +114,13 @@ export class SubscriptionFilters { * Subtracting a filter accepting all produces an empty filter. */ subtract(other: this) { - if (this._all) { - return; // Cannot subtract from all - } 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); @@ -366,6 +367,7 @@ export class SubscriptionController { } else { subscriber.filters = request.filters; } + break; default: throw new Error(`unreachable case: ${String(request.action)}`); diff --git a/test/lib/subscriptions.js b/test/lib/subscriptions.js deleted file mode 100644 index d8748f28..00000000 --- a/test/lib/subscriptions.js +++ /dev/null @@ -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]); - }); - }); - }); -}); diff --git a/test/lib/subscriptions.test.js b/test/lib/subscriptions.test.js new file mode 100644 index 00000000..33dcfca8 --- /dev/null +++ b/test/lib/subscriptions.test.js @@ -0,0 +1,1227 @@ +"use strict"; +const events = require("events"); +const assert = require("assert").strict; +const lib = require("@clusterio/lib"); +const { MockController, MockConnector, MockControl } = require("../mock"); +const { testMatrix, testRoundTripJsonSerialisable } = require("../common"); + +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; + } + + get filters() { + return "Filters"; + } + } + if (!lib.Link._eventsByName.has(RegisteredEvent.name)) { + lib.Link.register(RegisteredEvent); + } + + class StringPermissionEvent { + static type = "event"; + static src = ["controller", "control"]; + static dst = ["controller", "control"]; + static permission = "StringPermission"; + } + if (!lib.Link._eventsByName.has(StringPermissionEvent.name)) { + lib.Link.register(StringPermissionEvent); + } + + class FunctionPermissionEvent { + static type = "event"; + static src = ["controller", "control"]; + static dst = ["controller", "control"]; + static permission(user, message) { + user.checkPermission("FunctionPermission"); + }; + } + if (!lib.Link._eventsByName.has(FunctionPermissionEvent.name)) { + lib.Link.register(FunctionPermissionEvent); + } + + class MockUser { + checkPermission(permission) { + this.lastPermissionCheck = permission; + } + } + + describe("class SubscriptionFilters", function() { + it("should be round trip serialisable", function() { + const f = lib.SubscriptionFilters.fromShorthand(["foo", "bar"]); + const reconstructed = lib.SubscriptionFilters.fromJSON(JSON.parse(JSON.stringify(f))); + assert.deepEqual(new Set(reconstructed.toJSON()), new Set(["foo", "bar"])); + }); + + it("should not preserve 'all' through JSON", function() { + // This is because we use 'undefined' to indicate all + const f = lib.SubscriptionFilters.all(); + const reconstructed = lib.SubscriptionFilters.fromJSON(JSON.parse(JSON.stringify(f))); + assert.equal(reconstructed.isAll(), false); + assert.equal(reconstructed.isEmpty(), true); + }); + + describe("construction", function() { + it("should create an 'all' filter", function() { + const f = lib.SubscriptionFilters.all(); + assert.equal(f.isAll(), true); + assert.equal(f.isEmpty(), false); + }); + + it("should create an 'empty' filter", function() { + const f = lib.SubscriptionFilters.empty(); + assert.equal(f.isAll(), false); + assert.equal(f.isEmpty(), true); + }); + + it("should create from shorthand undefined as all", function() { + const f = lib.SubscriptionFilters.fromShorthand(undefined); + assert.equal(f.isAll(), true); + }); + + it("should create from shorthand string", function() { + const f = lib.SubscriptionFilters.fromShorthand("foo"); + assert.equal(f.isAll(), false); + assert.deepEqual(new Set(f.toJSON()), new Set(["foo"])); + }); + + it("should create from shorthand array", function() { + const f = lib.SubscriptionFilters.fromShorthand(["foo", "bar"]); + assert.deepEqual(new Set(f.toJSON()), new Set(["foo", "bar"])); + }); + }); + + describe(".accepts()", function() { + it("should accept any value when all", function() { + const f = lib.SubscriptionFilters.all(); + assert.equal(f.accepts("foo"), true); + assert.equal(f.accepts("bar"), true); + }); + + it("should reject all values when empty", function() { + const f = lib.SubscriptionFilters.empty(); + assert.equal(f.accepts("foo"), false); + assert.equal(f.accepts("bar"), false); + }); + + it("should match exact values", function() { + const f = lib.SubscriptionFilters.fromShorthand(["foo"]); + assert.equal(f.accepts("foo"), true); + assert.equal(f.accepts("bar"), false); + }); + }); + + describe(".extends()", function() { + it("all should extend any filter", function() { + const all = lib.SubscriptionFilters.all(); + const foo = lib.SubscriptionFilters.fromShorthand("foo"); + assert.equal(all.extends(foo), true); + }); + + it("should return true when superset", function() { + const a = lib.SubscriptionFilters.fromShorthand(["foo", "bar"]); + const b = lib.SubscriptionFilters.fromShorthand(["foo"]); + assert.equal(a.extends(b), true); + }); + + it("should return false when not superset", function() { + const a = lib.SubscriptionFilters.fromShorthand(["foo"]); + const b = lib.SubscriptionFilters.fromShorthand(["foo", "bar"]); + assert.equal(a.extends(b), false); + }); + }); + + describe(".intersects()", function() { + it("should intersect when sharing values", function() { + const a = lib.SubscriptionFilters.fromShorthand(["foo"]); + const b = lib.SubscriptionFilters.fromShorthand(["foo", "bar"]); + assert.equal(a.intersects(b), true); + assert.equal(b.intersects(a), true); + }); + + it("should not intersect when disjoint", function() { + const a = lib.SubscriptionFilters.fromShorthand(["foo"]); + const b = lib.SubscriptionFilters.fromShorthand(["bar"]); + assert.equal(a.intersects(b), false); + assert.equal(b.intersects(a), false); + }); + + it("all should intersect non-empty", function() { + const all = lib.SubscriptionFilters.all(); + const foo = lib.SubscriptionFilters.fromShorthand("foo"); + assert.equal(all.intersects(foo), true); + assert.equal(foo.intersects(all), true); + }); + + it("all should not intersect empty", function() { + const all = lib.SubscriptionFilters.all(); + const empty = lib.SubscriptionFilters.empty(); + assert.equal(all.intersects(empty), false); + assert.equal(empty.intersects(all), false); + }); + }); + + describe(".union()", function() { + it("should merge filters", function() { + const a = lib.SubscriptionFilters.fromShorthand(["foo"]); + const b = lib.SubscriptionFilters.fromShorthand(["bar"]); + a.union(b); + assert.deepEqual(new Set(a.toJSON()), new Set(["foo", "bar"])); + }); + + it("should become all if union with all", function() { + const a = lib.SubscriptionFilters.fromShorthand(["foo"]); + const all = lib.SubscriptionFilters.all(); + a.union(all); + assert.equal(a.isAll(), true); + }); + + it("should remain all if already all", function() { + const a = lib.SubscriptionFilters.all(); + const b = lib.SubscriptionFilters.fromShorthand(["foo"]); + a.union(b); + assert.equal(a.isAll(), true); + }); + }); + + describe(".subtract()", function() { + it("should remove matching filters", function() { + const a = lib.SubscriptionFilters.fromShorthand(["foo", "bar"]); + const b = lib.SubscriptionFilters.fromShorthand("foo"); + a.subtract(b); + assert.deepEqual(new Set(a.toJSON()), new Set(["bar"])); + }); + + it("should become empty when subtracting all", function() { + const a = lib.SubscriptionFilters.fromShorthand(["foo"]); + const all = lib.SubscriptionFilters.all(); + a.subtract(all); + assert.equal(a.isEmpty(), true); + + const b = lib.SubscriptionFilters.all(); + b.subtract(all); + assert.equal(b.isEmpty(), true); + }); + + it("should do nothing when subtracting from all", function() { + const a = lib.SubscriptionFilters.all(); + const b = lib.SubscriptionFilters.fromShorthand("foo"); + a.subtract(b); + assert.equal(a.isAll(), true); + }); + }); + }); + + describe("class SubscriptionRequest", function() { + it("should be round trip json serialisable", function() { + testRoundTripJsonSerialisable(lib.SubscriptionRequest, testMatrix( + [RegisteredEvent.name], + ["subscribe", "unsubscribe", "replace"], + [0, 123], + [ + undefined, + lib.SubscriptionFilters.all(), + lib.SubscriptionFilters.empty(), + lib.SubscriptionFilters.fromShorthand("foo"), + lib.SubscriptionFilters.fromShorthand(["foo", "bar"]), + ] + )); + }); + + it("should throw when constructed via fromJSON with unregistered event", function() { + assert.throws( + () => lib.SubscriptionRequest.fromJSON([UnregisteredEvent.name,, "subscribe", 0]), + new Error(`Unregistered Event class ${UnregisteredEvent.name}`) + ); + }); + + describe("static 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, "subscribe"); + 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, "subscribe"); + 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, "subscribe"); + lib.SubscriptionRequest.permission(mockUser, newSubscriptionRequestMessage(request)); + assert.equal(mockUser.lastPermissionCheck, "FunctionPermission"); + }); + }); + + describe("construction", function() { + describe("compatibility 'action'", function() { + it("should map true to subscribe", function() { + const req = new lib.SubscriptionRequest(RegisteredEvent.name, true); + assert.equal(req.action, "subscribe"); + }); + + it("should map false to unsubscribe", function() { + const req = new lib.SubscriptionRequest(RegisteredEvent.name, false); + assert.equal(req.action, "unsubscribe"); + }); + }); + + describe("compatibility 'filters'", function() { + it("should convert string filters to SubscriptionFilters", function() { + const req = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, "foo"); + assert(req.filters instanceof lib.SubscriptionFilters); + assert.equal(req.filters.accepts("foo"), true); + }); + + it("should convert array filters to SubscriptionFilters", function() { + const req = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, ["foo", "bar"]); + assert.equal(req.filters.accepts("foo"), true); + assert.equal(req.filters.accepts("bar"), true); + }); + + it("should default to 'all' when filters undefined", function() { + const req = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0); + assert.equal(req.filters.isAll(), true); + }); + + it("should accept SubscriptionFilters directly", function() { + const filters = lib.SubscriptionFilters.fromShorthand("foo"); + const req = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, filters); + assert.equal(req.filters, filters); + }); + }); + + it("should throw when given an unregistered event", function() { + assert.throws( + () => new lib.SubscriptionRequest(UnregisteredEvent.name, "subscribe"), + new Error(`Unregistered Event class ${UnregisteredEvent.name}`) + ); + }); + }); + }); + + describe("class SubscriptionController", function() { + let mockController, subscriptions; + + const connectorSetupDate = [{ + id: 0, + dst: addr({ controlId: 0 }), + src: addr("controller"), + }, { + id: 1, + 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.deepEqual(link.connector.sentMessages, []); + } + + 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) { + const link = getLink(connectorData.id); + const request = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe"); + subscriptions.handleRequest(link, request, connectorData.dst, connectorData.src); + link.connector.sentMessages.pop(); + } + }); + + 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 respect filters of subscribers", async function() { + const request0 = new lib.SubscriptionRequest(RegisteredEvent.name, "replace", 0, "foo"); + await subscriptions.handleRequest( + getLink(0), request0, connectorSetupDate[0].dst, connectorSetupDate[0].src + ); + + const request1 = new lib.SubscriptionRequest(RegisteredEvent.name, "replace", 0, "bar"); + await subscriptions.handleRequest( + getLink(1), request1, connectorSetupDate[1].dst, connectorSetupDate[1].src + ); + + subscriptions.broadcast(new RegisteredEvent(), "foo"); + await onceConnectorSend(0); + assertNoEvent(1); + getLink(0).connector.sentMessages.pop(); + + subscriptions.broadcast(new RegisteredEvent(), ["bar", "baz"]); + await onceConnectorSend(1); + assertNoEvent(0); + }); + + it("should always notify subscribers without filters", async function() { + const request = new lib.SubscriptionRequest(RegisteredEvent.name, "replace"); + await subscriptions.handleRequest( + getLink(0), request, connectorSetupDate[0].dst, connectorSetupDate[0].src + ); + + subscriptions.broadcast(new RegisteredEvent(), "foo"); + await onceConnectorSend(0); + assertLastEvent(0, RegisteredEvent); + }); + + it("should use event filters property if present", async function() { + const request0 = new lib.SubscriptionRequest(RegisteredEvent.name, "replace", 0, "Filters"); + await subscriptions.handleRequest( + getLink(0), request0, connectorSetupDate[0].dst, connectorSetupDate[0].src + ); + + const request1 = new lib.SubscriptionRequest(RegisteredEvent.name, "replace", 0, "bar"); + await subscriptions.handleRequest( + getLink(1), request1, connectorSetupDate[1].dst, connectorSetupDate[1].src + ); + + subscriptions.broadcast(new RegisteredEvent()); + await onceConnectorSend(0); + assertNoEvent(1); + }); + + it("should not notify a link who unsubscribed", async function() { + const connectorData = connectorSetupDate[0]; + const request = new lib.SubscriptionRequest(RegisteredEvent.name, "unsubscribe"); + 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(".unsubscribeLink() / .unsubscribeAddress()", function() { + beforeEach(function() { + subscriptions.handle(RegisteredEvent); + for (let connectorData of connectorSetupDate) { + const request = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe"); + subscriptions.handleRequest( + getLink(connectorData.id), 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, "subscribe"); + 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, "subscribe"); + 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 response with an event replay when returned by the handler", async function() { + subscriptions.handle(StringPermissionEvent, async function() { + return new StringPermissionEvent(); + }); + + const link = getLink(0); + const request = new lib.SubscriptionRequest(StringPermissionEvent.name, "subscribe"); + await subscriptions.handleRequest( + link, request, connectorSetupDate[0].dst, connectorSetupDate[0].src + ); + await onceConnectorSend(0); + assertLastEvent(0, StringPermissionEvent); + + 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, "subscribe"); + await subscriptions.handleRequest( + getLink(0), request, connectorSetupDate[0].dst, connectorSetupDate[0].src + ); + await new Promise(r => setImmediate(r)); + assertNoEvent(0); + }); + + it("should throw if action is not supported", async function() { + const request = new lib.SubscriptionRequest(StringPermissionEvent.name, "invalid action"); + await assert.rejects(subscriptions.handleRequest( + getLink(0), request, connectorSetupDate[0].dst, connectorSetupDate[0].src + ), /unreachable case: invalid action/); + }); + + describe("subscribe", function() { + it("should create a new subscription", async function() { + const request = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe"); + await subscriptions.handleRequest( + getLink(0), request, connectorSetupDate[0].dst, connectorSetupDate[0].src + ); + + subscriptions.broadcast(new RegisteredEvent()); + await onceConnectorSend(0); + assertLastEvent(0, RegisteredEvent); + }); + + it("should union filters for existing subscriber", async function() { + const link = getLink(0); + const connectorData = connectorSetupDate[0]; + + const request0 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, "foo"); + await subscriptions.handleRequest(link, request0, connectorData.dst, connectorData.src); + + const request1 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, "bar"); + await subscriptions.handleRequest(link, request1, connectorData.dst, connectorData.src); + + subscriptions.broadcast(new RegisteredEvent(), "foo"); + await onceConnectorSend(0); + assertLastEvent(0, RegisteredEvent); + link.connector.sentMessages.pop(); + + subscriptions.broadcast(new RegisteredEvent(), "bar"); + await onceConnectorSend(0); + assertLastEvent(0, RegisteredEvent); + link.connector.sentMessages.pop(); + + subscriptions.broadcast(new RegisteredEvent(), "baz"); + await new Promise(r => setImmediate(r)); + assertNoEvent(0); + }); + + it("should union filters with all when not specified", async function() { + const link = getLink(0); + const connectorData = connectorSetupDate[0]; + + const request0 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, "foo"); + await subscriptions.handleRequest(link, request0, connectorData.dst, connectorData.src); + + const request1 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe"); + await subscriptions.handleRequest(link, request1, connectorData.dst, connectorData.src); + + subscriptions.broadcast(new RegisteredEvent(), "baz"); + await onceConnectorSend(0); + assertLastEvent(0, RegisteredEvent); + }); + }); + + describe("unsubscribe", function() { + it("should remove a subscription", async function() { + const request = new lib.SubscriptionRequest(RegisteredEvent.name, "unsubscribe"); + await subscriptions.handleRequest( + getLink(0), request, connectorSetupDate[0].dst, connectorSetupDate[0].src + ); + + subscriptions.broadcast(new RegisteredEvent()); + await new Promise(r => setImmediate(r)); + assertNoEvent(0); + }); + + it("should subtract filters", async function() { + const link = getLink(0); + const connectorData = connectorSetupDate[0]; + + const request0 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, ["foo", "bar"]); + await subscriptions.handleRequest(link, request0, connectorData.dst, connectorData.src); + + const request1 = new lib.SubscriptionRequest(RegisteredEvent.name, "unsubscribe", 0, "foo"); + await subscriptions.handleRequest(link, request1, connectorData.dst, connectorData.src); + + subscriptions.broadcast(new RegisteredEvent(), "foo"); + await new Promise(r => setImmediate(r)); + assertNoEvent(0); + + subscriptions.broadcast(new RegisteredEvent(), "bar"); + await onceConnectorSend(0); + assertLastEvent(0, RegisteredEvent); + }); + + it("should remove subscription when empty", async function() { + const link = getLink(0); + const connectorData = connectorSetupDate[0]; + + const request0 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, "foo"); + await subscriptions.handleRequest(link, request0, connectorData.dst, connectorData.src); + + const request1 = new lib.SubscriptionRequest(RegisteredEvent.name, "unsubscribe", 0, "foo"); + await subscriptions.handleRequest(link, request1, connectorData.dst, connectorData.src); + + subscriptions.broadcast(new RegisteredEvent(), "foo"); + await new Promise(r => setImmediate(r)); + assertNoEvent(0); + }); + + it("should remain subscribed to all if subtracting a filter", async function() { + const link = getLink(0); + const connectorData = connectorSetupDate[0]; + + const request0 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe"); + await subscriptions.handleRequest(link, request0, connectorData.dst, connectorData.src); + + const request1 = new lib.SubscriptionRequest(RegisteredEvent.name, "unsubscribe", 0, "foo"); + await subscriptions.handleRequest(link, request1, connectorData.dst, connectorData.src); + + subscriptions.broadcast(new RegisteredEvent()); + await onceConnectorSend(0); + assertLastEvent(0, RegisteredEvent); + }); + }); + + describe("replace", function() { + it("should replace existing filters", async function() { + const link = getLink(0); + const connectorData = connectorSetupDate[0]; + + const request0 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, "foo"); + await subscriptions.handleRequest(link, request0, connectorData.dst, connectorData.src); + + const request1 = new lib.SubscriptionRequest(RegisteredEvent.name, "replace", 0, "bar"); + await subscriptions.handleRequest(link, request1, connectorData.dst, connectorData.src); + + subscriptions.broadcast(new RegisteredEvent(), "foo"); + await new Promise(r => setImmediate(r)); + assertNoEvent(0); + + subscriptions.broadcast(new RegisteredEvent(), "bar"); + await onceConnectorSend(0); + assertLastEvent(0, RegisteredEvent); + }); + + it("should remove subscription when replacing with empty", async function() { + const link = getLink(0); + const connectorData = connectorSetupDate[0]; + + const request0 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, "foo"); + await subscriptions.handleRequest(link, request0, connectorData.dst, connectorData.src); + + const request1 = new lib.SubscriptionRequest(RegisteredEvent.name, "replace", 0, []); + await subscriptions.handleRequest(link, request1, connectorData.dst, connectorData.src); + + subscriptions.broadcast(new RegisteredEvent(), "foo"); + await new Promise(r => setImmediate(r)); + 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 onceConnectorSend() { + return events.once(mockControl.connector, "send"); + } + + 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.deepEqual(mockControl.connector.sentMessages, []); + } + + 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 send subscribe(all) on first subscription", async function() { + registeredEvent.subscribe(() => {}); + await onceConnectorSend(); + + assertLastMessage(new lib.SubscriptionRequest( + RegisteredEvent.name, "subscribe", -1, lib.SubscriptionFilters.all() + )); + }); + + it("should send replace(filters) when last callback unsubscribes", async function() { + const unsubscribe = registeredEvent.subscribe(() => {}); + await onceConnectorSend(); + + unsubscribe(); + await onceConnectorSend(); + + assertLastMessage(new lib.SubscriptionRequest( + RegisteredEvent.name, "replace", -1, lib.SubscriptionFilters.empty() + )); + }); + + it("should notify all callbacks", async function() { + let called1, called2; + const event = new RegisteredEvent(); + + const unsubscribe = registeredEvent.subscribe((e, s) => { called1 = [e, s]; }); + registeredEvent.subscribe((e, s) => { called2 = [e, s]; }); + + await registeredEvent._handleEvent(event); + + assert.deepEqual(called1, [event, true]); + assert.deepEqual(called2, [event, true]); + }); + + it("should not notify unsubscribed callbacks", async function() { + let called1, called2; + const event = new RegisteredEvent(); + + const unsubscribe = registeredEvent.subscribe((e, s) => { called1 = [e, s]; }); + registeredEvent.subscribe((e, s) => { called2 = [e, s]; }); + + unsubscribe(); + await new Promise(r => setImmediate(r)); + await registeredEvent._handleEvent(event); + + assert.deepEqual(called1, undefined); + assert.deepEqual(called2, [event, true]); + }); + + it("should not throw when unsubscribing multiple times", async function() { + const unsubscribe = registeredEvent.subscribe(() => {}); + unsubscribe(); + await onceConnectorSend(); + unsubscribe(); + }); + }); + + describe("filters", function() { + it("should add filters and send subscribe", async function() { + registeredEvent.addFilters("foo"); + await onceConnectorSend(); + + assertLastMessage(new lib.SubscriptionRequest( + RegisteredEvent.name, "subscribe", -1, lib.SubscriptionFilters.fromShorthand("foo") + )); + }); + + it("should remove filters and send unsubscribe", async function() { + registeredEvent.addFilters("foo"); + await onceConnectorSend(); + + registeredEvent.removeFilters("foo"); + await onceConnectorSend(); + + assertLastMessage(new lib.SubscriptionRequest( + RegisteredEvent.name, "unsubscribe", -1, lib.SubscriptionFilters.fromShorthand("foo") + )); + }); + + it("should clear filters and unsubscribe when no callbacks", async function() { + registeredEvent.addFilters("foo"); + await onceConnectorSend(); + + registeredEvent.clearFilters(); + await onceConnectorSend(); + + assertLastMessage(new lib.SubscriptionRequest( + RegisteredEvent.name, "unsubscribe", -1, lib.SubscriptionFilters.all() + )); + }); + + it("should not unsubscribe on clearFilters if callbacks exist", async function() { + registeredEvent.subscribe(() => {}); + await onceConnectorSend(); + + registeredEvent.clearFilters(); + await new Promise(r => setImmediate(r)); + + // No new message + assert.equal(mockControl.connector.sentMessages.length, 1); + }); + + it("should correctly report hasFilters", function() { + registeredEvent.addFilters(["foo", "bar"]); + assert.equal(registeredEvent.hasFilters("foo"), true); + assert.equal(registeredEvent.hasFilters("baz"), false); + }); + + it("should only fire when filters are not empty", async function() { + let called = false; + registeredEvent.filteredHandler = () => { called = true; }; + + await registeredEvent._handleEvent(new RegisteredEvent()); + assert.equal(called, false); + + registeredEvent.addFilters("foo"); + await onceConnectorSend(); + + await registeredEvent._handleEvent(new RegisteredEvent()); + assert.equal(called, true); + }); + }); + + describe("getSnapshot()", function() { + it("should cache snapshot until state changes", async function() { + const snap1 = registeredEvent.getSnapshot(); + const snap2 = registeredEvent.getSnapshot(); + assert.strictEqual(snap1, snap2); + + await registeredEvent._handleEvent(new RegisteredEvent()); + const snap3 = registeredEvent.getSnapshot(); + assert.notStrictEqual(snap1, snap3); + + registeredEvent.synced = !registeredEvent.synced; + const snap4 = registeredEvent.getSnapshot(); + assert.notStrictEqual(snap3, snap4); + }); + }); + + describe("handleConnectionEvent", function() { + it("should send replace(all) on reconnect if subscribed", async function() { + registeredEvent.subscribe(() => {}); + await onceConnectorSend(); + + registeredEvent.control.connector.emit("connect"); + await onceConnectorSend(); + assertLastMessage(new lib.SubscriptionRequest( + RegisteredEvent.name, "replace", -1, lib.SubscriptionFilters.all() + )); + + mockControl.connector.sentMessages.pop(); + registeredEvent.handleConnectionEvent("resume"); + await onceConnectorSend(); + assertLastMessage(new lib.SubscriptionRequest( + RegisteredEvent.name, "replace", -1, lib.SubscriptionFilters.all() + )); + }); + + it("should send replace(filters) when only filters exist", async function() { + registeredEvent.addFilters("foo"); + await onceConnectorSend(); + + registeredEvent.control.connector.emit("connect"); + await onceConnectorSend(); + assertLastMessage(new lib.SubscriptionRequest( + RegisteredEvent.name, "replace", -1, lib.SubscriptionFilters.fromShorthand("foo") + )); + + mockControl.connector.sentMessages.pop(); + registeredEvent.handleConnectionEvent("resume"); + await onceConnectorSend(); + assertLastMessage(new lib.SubscriptionRequest( + RegisteredEvent.name, "replace", -1, lib.SubscriptionFilters.fromShorthand("foo") + )); + }); + + it("should set synced false on close", async function() { + registeredEvent.synced = true; + registeredEvent.control.connector.emit("close"); + assert.equal(registeredEvent.synced, false); + + registeredEvent.synced = true; + registeredEvent.handleConnectionEvent("drop"); + assert.equal(registeredEvent.synced, false); + + registeredEvent.synced = true; + registeredEvent.handleConnectionEvent("close"); + assert.equal(registeredEvent.synced, false); + }); + + it("should invoke callbacks on close", async function() { + let calledWith; + registeredEvent.subscribe((v, s) => { calledWith = [v, s]; }); + const [msg] = await onceConnectorSend(); + + mockControl.connector.emit("message", new lib.MessageResponse(0, msg.dst, msg.src, + lib.SubscriptionRequest.Response.fromJSON(false) + )); + await new Promise(r => setImmediate(r)); + assert.equal(registeredEvent.synced, true); + calledWith = undefined; + + registeredEvent.control.connector.emit("close"); + assert.equal(registeredEvent.synced, false); + assert.deepEqual(calledWith, [null, false]); + }); + }); + + describe("._sendRequest()", function() { + it("should do nothing if connection is not valid", async function() { + mockControl.connector.valid = false; + registeredEvent.subscribe(() => {}); + await new Promise(r => setImmediate(r)); + assertNoMessageSent(); + }); + + it("should set synced when no updates returned", async function() { + let calledWith; + registeredEvent.subscribe((e, s) => { calledWith = [e, s]; }); + const [msg] = await onceConnectorSend(); + + mockControl.connector.emit("message", new lib.MessageResponse(0, msg.dst, msg.src, + lib.SubscriptionRequest.Response.fromJSON(false) + )); + + await new Promise(r => setImmediate(r)); + assert.deepEqual(calledWith, [null, true]); + assert.equal(registeredEvent.synced, true); + }); + + it("should set synced when updates returned via _handleEvent", async function() { + let calledWith; + registeredEvent.subscribe((e, s) => { calledWith = [e, s]; }); + const [msg] = await onceConnectorSend(); + + mockControl.connector.emit("message", new lib.MessageResponse(0, msg.dst, msg.src, + lib.SubscriptionRequest.Response.fromJSON(true) + )); + + await new Promise(r => setImmediate(r)); + assert.equal(calledWith, undefined); + assert.equal(registeredEvent.synced, false); + + const eventUpdate = new RegisteredEvent(); + registeredEvent._handleEvent(eventUpdate); + + assert.deepEqual(calledWith, [eventUpdate, true]); + assert.equal(registeredEvent.synced, true); + }); + }); + }); + + describe("class ValueSubscriber", function() { + let mockControl, subscriber; + + class ValueEvent { + static type = "event"; + static src = ["controller", "control"]; + static dst = ["controller", "control"]; + static permission = null; + constructor(value) { + this.value = value; + } + } + if (!lib.Link._eventsByName.has(ValueEvent.name)) { + lib.Link.register(ValueEvent); + } + + beforeEach(function() { + mockControl = new MockControl(new MockConnector( + addr({ controlId: 0 }), + addr("controller"), + )); + subscriber = new lib.ValueSubscriber(ValueEvent, mockControl); + }); + + it("should update value on event", async function() { + const value = { id: 1, updatedAtMs: 100, isDeleted: false }; + await subscriber._handleEvent(new ValueEvent(value)); + + assert.deepEqual(subscriber.value, value); + }); + + it("should update lastUpdatedMs from value", async function() { + const value = { id: 1, updatedAtMs: 123, isDeleted: false }; + await subscriber._handleEvent(new ValueEvent(value)); + + assert.equal(subscriber.lastUpdatedMs, 123); + }); + + it("should return value in snapshot", async function() { + const value = { id: 1, updatedAtMs: 50, isDeleted: false }; + await subscriber._handleEvent(new ValueEvent(value)); + + const [snap, synced] = subscriber.getSnapshot(); + assert.deepEqual(snap, value); + assert.equal(synced, false); // no active subscription + }); + + it("should replace value on newer event", async function() { + await subscriber._handleEvent(new ValueEvent({ + id: 1, updatedAtMs: 10, isDeleted: false, + })); + + await subscriber._handleEvent(new ValueEvent({ + id: 1, updatedAtMs: 20, isDeleted: false, + })); + + assert.equal(subscriber.value.updatedAtMs, 20); + }); + }); + + describe("class MapSubscriber", function() { + let mockControl, subscriber; + + class MapEvent { + static type = "event"; + static src = ["controller", "control"]; + static dst = ["controller", "control"]; + static permission = null; + constructor(updates) { + this.updates = updates; + } + } + if (!lib.Link._eventsByName.has(MapEvent.name)) { + lib.Link.register(MapEvent); + } + + beforeEach(function() { + mockControl = new MockControl(new MockConnector( + addr({ controlId: 0 }), + addr("controller"), + )); + subscriber = new lib.MapSubscriber(MapEvent, mockControl); + }); + + it("should add values from updates", async function() { + await subscriber._handleEvent(new MapEvent([ + { id: 1, updatedAtMs: 10, isDeleted: false }, + ])); + + assert.equal(subscriber.values.size, 1); + assert.equal(subscriber.values.get(1).updatedAtMs, 10); + }); + + it("should update existing values", async function() { + await subscriber._handleEvent(new MapEvent([ + { id: 1, updatedAtMs: 10, isDeleted: false }, + ])); + + await subscriber._handleEvent(new MapEvent([ + { id: 1, updatedAtMs: 20, isDeleted: false }, + ])); + + assert.equal(subscriber.values.get(1).updatedAtMs, 20); + }); + + it("should delete values when isDeleted is true", async function() { + await subscriber._handleEvent(new MapEvent([ + { id: 1, updatedAtMs: 10, isDeleted: false }, + ])); + + await subscriber._handleEvent(new MapEvent([ + { id: 1, updatedAtMs: 20, isDeleted: true }, + ])); + + assert.equal(subscriber.values.has(1), false); + }); + + it("should compute lastUpdatedMs as max of updates", async function() { + await subscriber._handleEvent(new MapEvent([ + { id: 1, updatedAtMs: 10, isDeleted: false }, + { id: 2, updatedAtMs: 50, isDeleted: false }, + ])); + + assert.equal(subscriber.lastUpdatedMs, 50); + }); + + it("should return snapshot as a copy", async function() { + await subscriber._handleEvent(new MapEvent([ + { id: 1, updatedAtMs: 10, isDeleted: false }, + ])); + + const [snap] = subscriber.getSnapshot(); + + assert.deepEqual(snap, subscriber.values); + assert.notStrictEqual(snap, subscriber.values); // must be new Map + }); + + it("should not mutate previous snapshots", async function() { + await subscriber._handleEvent(new MapEvent([ + { id: 1, updatedAtMs: 10, isDeleted: false }, + ])); + + const [snap1] = subscriber.getSnapshot(); + + await subscriber._handleEvent(new MapEvent([ + { id: 2, updatedAtMs: 20, isDeleted: false }, + ])); + + const [snap2] = subscriber.getSnapshot(); + + assert.equal(snap1.has(2), false); + assert.equal(snap2.has(2), true); + }); + + it("should handle multiple updates correctly", async function() { + await subscriber._handleEvent(new MapEvent([ + { id: 1, updatedAtMs: 10, isDeleted: false }, + { id: 2, updatedAtMs: 20, isDeleted: false }, + ])); + + await subscriber._handleEvent(new MapEvent([ + { id: 1, updatedAtMs: 30, isDeleted: false }, + { id: 2, updatedAtMs: 40, isDeleted: true }, + ])); + + assert.equal(subscriber.values.get(1).updatedAtMs, 30); + assert.equal(subscriber.values.has(2), false); + }); + + it("should warn on same timestamp but different content", async function() { + let warned = false; + const originalWarn = lib.logger.warn; + lib.logger.warn = () => { warned = true; }; + + await subscriber._handleEvent(new MapEvent([ + { id: 1, updatedAtMs: 10, isDeleted: false, foo: "a" }, + ])); + + await subscriber._handleEvent(new MapEvent([ + { id: 1, updatedAtMs: 10, isDeleted: false, foo: "b" }, + ])); + + lib.logger.warn = originalWarn; + assert.equal(warned, true); + }); + }); +}); From e54b292fce27a2914ad97eca9d3ee94393b3ed0c Mon Sep 17 00:00:00 2001 From: Cooldude2606 <25043174+Cooldude2606@users.noreply.github.com> Date: Fri, 29 May 2026 18:38:07 +0100 Subject: [PATCH 10/13] Fix tests relating to permissions --- packages/lib/src/data/Permission.ts | 5 +++-- test/integration/clusterio.js | 9 +++++---- test/lib/config/validators.test.js | 18 ++++++++++-------- test/lib/subscriptions.test.js | 2 +- 4 files changed, 19 insertions(+), 15 deletions(-) diff --git a/packages/lib/src/data/Permission.ts b/packages/lib/src/data/Permission.ts index 28ace0ca..2fc3363c 100644 --- a/packages/lib/src/data/Permission.ts +++ b/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) { - return new this(json.name, json.title, json.description); + return new this(json.name, json.title, json.description, Boolean(json.grantByDefault)); } } diff --git a/test/integration/clusterio.js b/test/integration/clusterio.js index 480f4b46..e44f673d 100644 --- a/test/integration/clusterio.js +++ b/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)); }); }); diff --git a/test/lib/config/validators.test.js b/test/lib/config/validators.test.js index 85bce149..949f3d0f 100644 --- a/test/lib/config/validators.test.js +++ b/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 }; diff --git a/test/lib/subscriptions.test.js b/test/lib/subscriptions.test.js index 33dcfca8..0af0dd35 100644 --- a/test/lib/subscriptions.test.js +++ b/test/lib/subscriptions.test.js @@ -586,7 +586,7 @@ describe("lib/subscriptions", function() { }); it("should throw if action is not supported", async function() { - const request = new lib.SubscriptionRequest(StringPermissionEvent.name, "invalid action"); + const request = new lib.SubscriptionRequest(RegisteredEvent.name, "invalid action"); await assert.rejects(subscriptions.handleRequest( getLink(0), request, connectorSetupDate[0].dst, connectorSetupDate[0].src ), /unreachable case: invalid action/); From 11b8a0ff971817c107ba84364bf381267ae2ab11 Mon Sep 17 00:00:00 2001 From: Cooldude2606 <25043174+Cooldude2606@users.noreply.github.com> Date: Fri, 29 May 2026 19:15:27 +0100 Subject: [PATCH 11/13] Fix bad snapshot initialisation --- packages/lib/src/subscriptions.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index 8d8d5120..d6761dd9 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -401,7 +401,7 @@ export class EventSubscriber { /** True if this subscriber is currently synced with the source */ synced = false; /** Repeat calls to getSnapshot will return the same readonly copy unless values has updated */ - private _snapshot: readonly [S, boolean] = [null as S, false]; + 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 */ private _callbacks = new Array>(); From e3d5f434a474609d8fe92519a62a5303049f7a22 Mon Sep 17 00:00:00 2001 From: Cooldude2606 <25043174+Cooldude2606@users.noreply.github.com> Date: Fri, 29 May 2026 20:14:23 +0100 Subject: [PATCH 12/13] Better `toString` and test coverage --- packages/lib/src/subscriptions.ts | 8 +- test/lib/subscriptions.test.js | 223 +++++++++++++++++++++--------- 2 files changed, 162 insertions(+), 69 deletions(-) diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index d6761dd9..f8dc48c5 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -64,7 +64,13 @@ export class SubscriptionFilters { } toString() { - return `[SubscriptionFilters ${this._all ? "All" : String(this._filters)}]`; + 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 all the given filters */ diff --git a/test/lib/subscriptions.test.js b/test/lib/subscriptions.test.js index 0af0dd35..93f330b7 100644 --- a/test/lib/subscriptions.test.js +++ b/test/lib/subscriptions.test.js @@ -105,6 +105,28 @@ describe("lib/subscriptions", function() { }); }); + describe(".toString()", function() { + it("should return 'All' for all filters", function() { + const filters = lib.SubscriptionFilters.all(); + assert.equal(filters.toString(), "[SubscriptionFilters All]"); + }); + + it("should return empty set representation for empty filters", function() { + const filters = lib.SubscriptionFilters.empty(); + assert.equal(filters.toString(), "[SubscriptionFilters Empty]"); + }); + + it("should include single filter value", function() { + const filters = lib.SubscriptionFilters.fromShorthand("foo"); + assert.equal(filters.toString(), "[SubscriptionFilters Set<1>]"); + }); + + it("should include multiple filter values", function() { + const filters = lib.SubscriptionFilters.fromShorthand(["foo", "bar"]); + assert.equal(filters.toString(), "[SubscriptionFilters Set<2>]"); + }); + }); + describe(".accepts()", function() { it("should accept any value when all", function() { const f = lib.SubscriptionFilters.all(); @@ -727,6 +749,18 @@ describe("lib/subscriptions", function() { assertLastEvent(0, RegisteredEvent); }); + it("should subscribe when no existing subscriber exists", async function() { + const link = getLink(0); + const connectorData = connectorSetupDate[0]; + + const request1 = new lib.SubscriptionRequest(RegisteredEvent.name, "replace", 0, "foo"); + await subscriptions.handleRequest(link, request1, connectorData.dst, connectorData.src); + + subscriptions.broadcast(new RegisteredEvent(), "foo"); + await onceConnectorSend(0); + assertLastEvent(0, RegisteredEvent); + }); + it("should remove subscription when replacing with empty", async function() { const link = getLink(0); const connectorData = connectorSetupDate[0]; @@ -741,19 +775,31 @@ describe("lib/subscriptions", function() { await new Promise(r => setImmediate(r)); assertNoEvent(0); }); + + it("should do nothing when replacing with empty if no subscriber exists", async function() { + const link = getLink(0); + const connectorData = connectorSetupDate[0]; + + const request1 = new lib.SubscriptionRequest(RegisteredEvent.name, "replace", 0, []); + await subscriptions.handleRequest(link, request1, connectorData.dst, connectorData.src); + + subscriptions.broadcast(new RegisteredEvent()); + await new Promise(r => setImmediate(r)); + assertNoEvent(0); + }); }); }); }); describe("class EventSubscriber", function() { - let mockControl, registeredEvent; + let mockControl, eventSubscriber; beforeEach(function() { mockControl = new MockControl(new MockConnector( addr({ controlId: 0 }), addr("controller"), )); - registeredEvent = new lib.EventSubscriber(RegisteredEvent, mockControl); + eventSubscriber = new lib.EventSubscriber(RegisteredEvent, mockControl); }); function onceConnectorSend() { @@ -787,7 +833,7 @@ describe("lib/subscriptions", function() { describe("subscribe()", function() { it("should send subscribe(all) on first subscription", async function() { - registeredEvent.subscribe(() => {}); + eventSubscriber.subscribe(() => {}); await onceConnectorSend(); assertLastMessage(new lib.SubscriptionRequest( @@ -796,7 +842,7 @@ describe("lib/subscriptions", function() { }); it("should send replace(filters) when last callback unsubscribes", async function() { - const unsubscribe = registeredEvent.subscribe(() => {}); + const unsubscribe = eventSubscriber.subscribe(() => {}); await onceConnectorSend(); unsubscribe(); @@ -811,10 +857,10 @@ describe("lib/subscriptions", function() { let called1, called2; const event = new RegisteredEvent(); - const unsubscribe = registeredEvent.subscribe((e, s) => { called1 = [e, s]; }); - registeredEvent.subscribe((e, s) => { called2 = [e, s]; }); + const unsubscribe = eventSubscriber.subscribe((e, s) => { called1 = [e, s]; }); + eventSubscriber.subscribe((e, s) => { called2 = [e, s]; }); - await registeredEvent._handleEvent(event); + await eventSubscriber._handleEvent(event); assert.deepEqual(called1, [event, true]); assert.deepEqual(called2, [event, true]); @@ -824,19 +870,19 @@ describe("lib/subscriptions", function() { let called1, called2; const event = new RegisteredEvent(); - const unsubscribe = registeredEvent.subscribe((e, s) => { called1 = [e, s]; }); - registeredEvent.subscribe((e, s) => { called2 = [e, s]; }); + const unsubscribe = eventSubscriber.subscribe((e, s) => { called1 = [e, s]; }); + eventSubscriber.subscribe((e, s) => { called2 = [e, s]; }); unsubscribe(); await new Promise(r => setImmediate(r)); - await registeredEvent._handleEvent(event); + await eventSubscriber._handleEvent(event); assert.deepEqual(called1, undefined); assert.deepEqual(called2, [event, true]); }); it("should not throw when unsubscribing multiple times", async function() { - const unsubscribe = registeredEvent.subscribe(() => {}); + const unsubscribe = eventSubscriber.subscribe(() => {}); unsubscribe(); await onceConnectorSend(); unsubscribe(); @@ -845,7 +891,7 @@ describe("lib/subscriptions", function() { describe("filters", function() { it("should add filters and send subscribe", async function() { - registeredEvent.addFilters("foo"); + eventSubscriber.addFilters("foo"); await onceConnectorSend(); assertLastMessage(new lib.SubscriptionRequest( @@ -854,10 +900,10 @@ describe("lib/subscriptions", function() { }); it("should remove filters and send unsubscribe", async function() { - registeredEvent.addFilters("foo"); + eventSubscriber.addFilters("foo"); await onceConnectorSend(); - registeredEvent.removeFilters("foo"); + eventSubscriber.removeFilters("foo"); await onceConnectorSend(); assertLastMessage(new lib.SubscriptionRequest( @@ -866,10 +912,10 @@ describe("lib/subscriptions", function() { }); it("should clear filters and unsubscribe when no callbacks", async function() { - registeredEvent.addFilters("foo"); + eventSubscriber.addFilters("foo"); await onceConnectorSend(); - registeredEvent.clearFilters(); + eventSubscriber.clearFilters(); await onceConnectorSend(); assertLastMessage(new lib.SubscriptionRequest( @@ -878,10 +924,10 @@ describe("lib/subscriptions", function() { }); it("should not unsubscribe on clearFilters if callbacks exist", async function() { - registeredEvent.subscribe(() => {}); + eventSubscriber.subscribe(() => {}); await onceConnectorSend(); - registeredEvent.clearFilters(); + eventSubscriber.clearFilters(); await new Promise(r => setImmediate(r)); // No new message @@ -889,55 +935,55 @@ describe("lib/subscriptions", function() { }); it("should correctly report hasFilters", function() { - registeredEvent.addFilters(["foo", "bar"]); - assert.equal(registeredEvent.hasFilters("foo"), true); - assert.equal(registeredEvent.hasFilters("baz"), false); + eventSubscriber.addFilters(["foo", "bar"]); + assert.equal(eventSubscriber.hasFilters("foo"), true); + assert.equal(eventSubscriber.hasFilters("baz"), false); }); it("should only fire when filters are not empty", async function() { let called = false; - registeredEvent.filteredHandler = () => { called = true; }; + eventSubscriber.filteredHandler = () => { called = true; }; - await registeredEvent._handleEvent(new RegisteredEvent()); + await eventSubscriber._handleEvent(new RegisteredEvent()); assert.equal(called, false); - registeredEvent.addFilters("foo"); + eventSubscriber.addFilters("foo"); await onceConnectorSend(); - await registeredEvent._handleEvent(new RegisteredEvent()); + await eventSubscriber._handleEvent(new RegisteredEvent()); assert.equal(called, true); }); }); describe("getSnapshot()", function() { it("should cache snapshot until state changes", async function() { - const snap1 = registeredEvent.getSnapshot(); - const snap2 = registeredEvent.getSnapshot(); + const snap1 = eventSubscriber.getSnapshot(); + const snap2 = eventSubscriber.getSnapshot(); assert.strictEqual(snap1, snap2); - await registeredEvent._handleEvent(new RegisteredEvent()); - const snap3 = registeredEvent.getSnapshot(); + await eventSubscriber._handleEvent(new RegisteredEvent()); + const snap3 = eventSubscriber.getSnapshot(); assert.notStrictEqual(snap1, snap3); - registeredEvent.synced = !registeredEvent.synced; - const snap4 = registeredEvent.getSnapshot(); + eventSubscriber.synced = !eventSubscriber.synced; + const snap4 = eventSubscriber.getSnapshot(); assert.notStrictEqual(snap3, snap4); }); }); describe("handleConnectionEvent", function() { it("should send replace(all) on reconnect if subscribed", async function() { - registeredEvent.subscribe(() => {}); + eventSubscriber.subscribe(() => {}); await onceConnectorSend(); - registeredEvent.control.connector.emit("connect"); + eventSubscriber.control.connector.emit("connect"); await onceConnectorSend(); assertLastMessage(new lib.SubscriptionRequest( RegisteredEvent.name, "replace", -1, lib.SubscriptionFilters.all() )); mockControl.connector.sentMessages.pop(); - registeredEvent.handleConnectionEvent("resume"); + eventSubscriber.handleConnectionEvent("resume"); await onceConnectorSend(); assertLastMessage(new lib.SubscriptionRequest( RegisteredEvent.name, "replace", -1, lib.SubscriptionFilters.all() @@ -945,17 +991,17 @@ describe("lib/subscriptions", function() { }); it("should send replace(filters) when only filters exist", async function() { - registeredEvent.addFilters("foo"); + eventSubscriber.addFilters("foo"); await onceConnectorSend(); - registeredEvent.control.connector.emit("connect"); + eventSubscriber.control.connector.emit("connect"); await onceConnectorSend(); assertLastMessage(new lib.SubscriptionRequest( RegisteredEvent.name, "replace", -1, lib.SubscriptionFilters.fromShorthand("foo") )); mockControl.connector.sentMessages.pop(); - registeredEvent.handleConnectionEvent("resume"); + eventSubscriber.handleConnectionEvent("resume"); await onceConnectorSend(); assertLastMessage(new lib.SubscriptionRequest( RegisteredEvent.name, "replace", -1, lib.SubscriptionFilters.fromShorthand("foo") @@ -963,33 +1009,33 @@ describe("lib/subscriptions", function() { }); it("should set synced false on close", async function() { - registeredEvent.synced = true; - registeredEvent.control.connector.emit("close"); - assert.equal(registeredEvent.synced, false); + eventSubscriber.synced = true; + eventSubscriber.control.connector.emit("close"); + assert.equal(eventSubscriber.synced, false); - registeredEvent.synced = true; - registeredEvent.handleConnectionEvent("drop"); - assert.equal(registeredEvent.synced, false); + eventSubscriber.synced = true; + eventSubscriber.handleConnectionEvent("drop"); + assert.equal(eventSubscriber.synced, false); - registeredEvent.synced = true; - registeredEvent.handleConnectionEvent("close"); - assert.equal(registeredEvent.synced, false); + eventSubscriber.synced = true; + eventSubscriber.handleConnectionEvent("close"); + assert.equal(eventSubscriber.synced, false); }); it("should invoke callbacks on close", async function() { let calledWith; - registeredEvent.subscribe((v, s) => { calledWith = [v, s]; }); + eventSubscriber.subscribe((v, s) => { calledWith = [v, s]; }); const [msg] = await onceConnectorSend(); mockControl.connector.emit("message", new lib.MessageResponse(0, msg.dst, msg.src, lib.SubscriptionRequest.Response.fromJSON(false) )); await new Promise(r => setImmediate(r)); - assert.equal(registeredEvent.synced, true); + assert.equal(eventSubscriber.synced, true); calledWith = undefined; - registeredEvent.control.connector.emit("close"); - assert.equal(registeredEvent.synced, false); + eventSubscriber.control.connector.emit("close"); + assert.equal(eventSubscriber.synced, false); assert.deepEqual(calledWith, [null, false]); }); }); @@ -997,14 +1043,14 @@ describe("lib/subscriptions", function() { describe("._sendRequest()", function() { it("should do nothing if connection is not valid", async function() { mockControl.connector.valid = false; - registeredEvent.subscribe(() => {}); + eventSubscriber.subscribe(() => {}); await new Promise(r => setImmediate(r)); assertNoMessageSent(); }); it("should set synced when no updates returned", async function() { let calledWith; - registeredEvent.subscribe((e, s) => { calledWith = [e, s]; }); + eventSubscriber.subscribe((e, s) => { calledWith = [e, s]; }); const [msg] = await onceConnectorSend(); mockControl.connector.emit("message", new lib.MessageResponse(0, msg.dst, msg.src, @@ -1013,12 +1059,12 @@ describe("lib/subscriptions", function() { await new Promise(r => setImmediate(r)); assert.deepEqual(calledWith, [null, true]); - assert.equal(registeredEvent.synced, true); + assert.equal(eventSubscriber.synced, true); }); it("should set synced when updates returned via _handleEvent", async function() { let calledWith; - registeredEvent.subscribe((e, s) => { calledWith = [e, s]; }); + eventSubscriber.subscribe((e, s) => { calledWith = [e, s]; }); const [msg] = await onceConnectorSend(); mockControl.connector.emit("message", new lib.MessageResponse(0, msg.dst, msg.src, @@ -1027,13 +1073,52 @@ describe("lib/subscriptions", function() { await new Promise(r => setImmediate(r)); assert.equal(calledWith, undefined); - assert.equal(registeredEvent.synced, false); + assert.equal(eventSubscriber.synced, false); const eventUpdate = new RegisteredEvent(); - registeredEvent._handleEvent(eventUpdate); + eventSubscriber._handleEvent(eventUpdate); assert.deepEqual(calledWith, [eventUpdate, true]); - assert.equal(registeredEvent.synced, true); + assert.equal(eventSubscriber.synced, true); + }); + + it("should log error when sendTo throws", async function() { + let logged; + const originalError = lib.logger.error; + lib.logger.error = msg => { logged = msg; }; + + // Mock sendTo to throw generic error + mockControl.sendTo = async function() { + throw new Error("boom"); + }; + + try { + eventSubscriber.subscribe(() => {}); + await new Promise(r => setImmediate(r)); + } finally { + lib.logger.error = originalError; + } + assert.notEqual(logged, undefined); + }); + + it("should not log error when sendTo throws SessionLost RequestError", async function() { + let logged; + const originalError = lib.logger.error; + lib.logger.error = msg => { logged = msg; }; + + mockControl.sendTo = async function() { + const err = new lib.RequestError("Session lost"); + err.code = "SessionLost"; + throw err; + }; + + try { + eventSubscriber.subscribe(() => {}); + await new Promise(r => setImmediate(r)); + } finally { + lib.logger.error = originalError; + } + assert.equal(logged, undefined); }); }); }); @@ -1208,20 +1293,22 @@ describe("lib/subscriptions", function() { }); it("should warn on same timestamp but different content", async function() { - let warned = false; + let logged = false; const originalWarn = lib.logger.warn; - lib.logger.warn = () => { warned = true; }; + lib.logger.warn = msg => { logged = msg; }; - await subscriber._handleEvent(new MapEvent([ - { id: 1, updatedAtMs: 10, isDeleted: false, foo: "a" }, - ])); + try { + await subscriber._handleEvent(new MapEvent([ + { id: 1, updatedAtMs: 10, isDeleted: false, foo: "a" }, + ])); - await subscriber._handleEvent(new MapEvent([ - { id: 1, updatedAtMs: 10, isDeleted: false, foo: "b" }, - ])); - - lib.logger.warn = originalWarn; - assert.equal(warned, true); + await subscriber._handleEvent(new MapEvent([ + { id: 1, updatedAtMs: 10, isDeleted: false, foo: "b" }, + ])); + } finally { + lib.logger.warn = originalWarn; + } + assert.notEqual(logged, undefined); }); }); }); From b5c99fad2a0dfe6cd783b98ac19da80e2d178f33 Mon Sep 17 00:00:00 2001 From: Cooldude2606 <25043174+Cooldude2606@users.noreply.github.com> Date: Tue, 2 Jun 2026 22:50:41 +0100 Subject: [PATCH 13/13] Fix incorrect doc comment --- packages/lib/src/subscriptions.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index f8dc48c5..aeb08827 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -73,7 +73,7 @@ export class SubscriptionFilters { return `[SubscriptionFilters Set<${this._filters.size}>]`; } - /** Returns true if this filter accepts all the given filters */ + /** Returns true if this filter accepts the given string */ accepts(value: string) { return this._all || this._filters.has(value); }