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/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/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); } diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index f73374a1..aeb08827 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -1,13 +1,142 @@ import { Type, Static } from "@sinclair/typebox"; import { Link, Event, EventClass, RequestHandler, WebSocketBaseConnector } from "./link"; -import { Address, MessageRequest, IUser, plainJson, JsonBoolean } from "./data"; +import { Address, MessageRequest, IUser, JsonBoolean, StringEnum } from "./data"; import isDeepStrictEqual from "./is_deep_strict_equal"; -import { RequestError, SessionLost } from "./errors"; +import { RequestError } from "./errors"; import { logger } from "./logging"; export type SubscriptionRequestHandler = 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() { + if (this._all) { + return "[SubscriptionFilters All]"; + } + if (this._filters.size === 0) { + return "[SubscriptionFilters Empty]"; + } + return `[SubscriptionFilters Set<${this._filters.size}>]`; + } + + /** Returns true if this filter accepts the given string */ + accepts(value: string) { + return this._all || this._filters.has(value); + } + + /** Returns true if this filter extends the other */ + extends(other: this) { + return this._all || (!other._all && [...other._filters.values()].every(value => this._filters.has(value))); + } + + /** Returns true if there is overlap between two filters */ + intersects(other: this) { + if (this._all) { + return !other.isEmpty(); + } + if (other._all) { + return !this.isEmpty(); + } + for (const filter of other._filters) { + if (this._filters.has(filter)) { + return true; + } + } + return false; + } + + /** Add filters from a filter into this one. */ + union(other: this) { + if (this._all) { + return; // Already all + } + if (other._all) { + this._all = true; + this._filters.clear(); + return; + } + for (const filter of other._filters) { + this._filters.add(filter); + } + } + + /** + * Remove filters in a filter from this one. + * You cannot subtract filters from a filter accepting all. + * Subtracting a filter accepting all produces an empty filter. + */ + subtract(other: this) { + if (other._all) { + this._all = false; + this._filters.clear(); + return; + } + if (this._all) { + return; // Cannot subtract from all + } + for (const filter of other._filters) { + this._filters.delete(filter); + } + } +} + +const SubscriptionActionSchema = StringEnum(["subscribe", "unsubscribe", "replace"]); +type SubscriptionAction = Static; + /** * A subscription request sent by a subscriber, this updates what events the subscriber will be sent * The permission for this request copies the permission from the event being subscribed to @@ -20,7 +149,7 @@ export class SubscriptionRequest { static dst = "controller" as const; static permission(user: IUser, message: MessageRequest) { if (typeof message.data === "object" && message.data !== null) { - const data = message.data as Static; + 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,55 +163,86 @@ export class SubscriptionRequest { /** Indicates if updates have been sent */ static Response = JsonBoolean; + public filters: SubscriptionFilters; + public action: SubscriptionAction; + + /** @deprecated Use SubscriptionFilters and SubscriptionAction instead */ + constructor( + eventName: string, + subscribe: boolean, + lastRequestTimeMs: number | undefined, + filters?: string | string[], + ); + + constructor( + eventName: string, + action: SubscriptionAction, + lastRequestTimeMs: number | undefined, + filters?: SubscriptionFilters, + ); constructor( public eventName: string, - public subscribe: boolean, + action: boolean | SubscriptionAction, public lastRequestTimeMs: number = 0, - public filters: string | string[] | undefined = undefined, + filters: undefined | string | string[] | SubscriptionFilters, ) { if (!Link._eventsByName.has(eventName)) { throw new Error(`Unregistered Event class ${eventName}`); } + if (filters instanceof SubscriptionFilters) { // Easier migration for plugins + this.filters = filters; + } else { + this.filters = SubscriptionFilters.fromShorthand(filters); + } + if (typeof action === "boolean") { // Easier migration for plugins + this.action = action ? "subscribe" : "unsubscribe"; + } else { + this.action = action; + } } static jsonSchema = Type.Union([ Type.Tuple([ Type.String(), - Type.Boolean(), + SubscriptionActionSchema, Type.Number(), ]), Type.Tuple([ Type.String(), - Type.Boolean(), + SubscriptionActionSchema, Type.Number(), - Type.Union([Type.String(), Type.Array(Type.String())]), + SubscriptionFilters.jsonSchema, ]), ]); toJSON() { - if (this.filters !== undefined) { - return [this.eventName, this.subscribe, this.lastRequestTimeMs, this.filters]; + if (!this.filters.isAll()) { + return [this.eventName, this.action, this.lastRequestTimeMs, this.filters.toJSON()]; } - return [this.eventName, this.subscribe, this.lastRequestTimeMs]; + return [this.eventName, this.action, this.lastRequestTimeMs]; } - static fromJSON(json: Static): SubscriptionRequest { - const [eventName, subscribe, lastRequestTimeMs, filters] = json; - return new this(eventName, subscribe, lastRequestTimeMs, filters); + static fromJSON(json: Static) { + const [eventName, action, lastRequestTimeMs, filters] = json; + return new this( + eventName, + action, + lastRequestTimeMs, + filters ? SubscriptionFilters.fromJSON(filters) : SubscriptionFilters.all() + ); } } type Subscriber = { link: Link, dst: Address, - /** Optional filters this subscriber is interested in. Undefined means all events */ - filters?: string[], + filters: SubscriptionFilters, } type EventData = { subscriptionUpdate?: SubscriptionRequestHandler, - subscriptions: Array, + subscriptions: Map, Subscriber>, }; /** @@ -92,13 +252,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 @@ -116,7 +269,7 @@ export class SubscriptionController { } this._events.set(entry.name, { subscriptionUpdate: subscriptionUpdate, - subscriptions: [], + subscriptions: new Map(), }); } @@ -124,7 +277,7 @@ export class SubscriptionController { * Broadcast an event to all subscribers of that event * @param event - Event to broadcast. */ - broadcast(event: Event, filter?: 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}`); @@ -133,24 +286,17 @@ export class SubscriptionController { if (!eventData) { throw new Error(`Event ${entry.name} is not a registered as subscribable`); } - const toRemove: Subscriber[] = []; - const filtersToMatch = SubscriptionController.normalizeFilters(filter); - for (const subscriber of eventData.subscriptions) { + const broadcastFilters = SubscriptionFilters.fromShorthand(filters ?? event.filters); + for (const [addressIndex, subscriber] of eventData.subscriptions) { if ((subscriber.link.connector as WebSocketBaseConnector).closing) { - toRemove.push(subscriber); + eventData.subscriptions.delete(addressIndex); continue; } - - if ( - filtersToMatch === undefined - || subscriber.filters === undefined - || subscriber.filters.some(f => filtersToMatch.includes(f)) - ) { + if (subscriber.filters.intersects(broadcastFilters)) { subscriber.link.sendTo(subscriber.dst, event); } } - eventData.subscriptions = eventData.subscriptions.filter(subscriber => !toRemove.includes(subscriber)); } /** @@ -159,10 +305,9 @@ export class SubscriptionController { * @param address - Address to stop sending events to. */ unsubscribeAddress(address: Address) { - for (let eventData of this._events.values()) { - eventData.subscriptions = eventData.subscriptions.filter( - subscriber => !subscriber.dst.addressedTo(address) - ); + const addressIndex = address.addressIndex(); + for (const eventData of this._events.values()) { + eventData.subscriptions.delete(addressIndex); } } @@ -172,69 +317,76 @@ export class SubscriptionController { * @param link - Link to stop sending events to. */ unsubscribeLink(link: Link) { - for (let eventData of this._events.values()) { - eventData.subscriptions = eventData.subscriptions.filter( - subscriber => subscriber.link !== link - ); + for (const eventData of this._events.values()) { + for (const [addressIndex, subscriber] of eventData.subscriptions) { + if (subscriber.link === link) { + eventData.subscriptions.delete(addressIndex); + } + } } } /** * Handle incoming subscription requests on a link * @param link - Link message was received on - * @param event - incomming event. + * @param request - incoming request. * @param src - Source address of incomming request. * @param dst - destination address of incomming request. */ - async handleRequest(link: Link, event: SubscriptionRequest, src: Address, dst: Address) { - if (!Link._eventsByName.has(event.eventName)) { - throw new Error(`Event ${event.eventName} is not a registered event`); + async handleRequest(link: Link, request: SubscriptionRequest, src: Address, dst: Address) { + if (!Link._eventsByName.has(request.eventName)) { + throw new Error(`Event ${request.eventName} is not a registered event`); } - const eventData = this._events.get(event.eventName); + const eventData = this._events.get(request.eventName); if (!eventData) { - throw new Error(`Event ${event.eventName} is not a registered as subscribable`); + throw new Error(`Event ${request.eventName} is not a registered as subscribable`); } - if (event.subscribe === false) { - // Unsubscribe logic - const subscriber = eventData.subscriptions.find(sub => sub.dst.addressedTo(src)); - if (!subscriber) { - return false; - } - if (event.filters === undefined || subscriber.filters === undefined) { - // Remove entire subscription for this address - eventData.subscriptions = eventData.subscriptions.filter(sub => !sub.dst.addressedTo(src)); - } else { - const filtersToRemove = Array.isArray(event.filters) ? event.filters : [event.filters]; - subscriber.filters = subscriber.filters.filter(f => !filtersToRemove.includes(f)); - if (subscriber.filters.length === 0) { - // No filters left, remove subscription - eventData.subscriptions = eventData.subscriptions.filter(sub => sub !== subscriber); + const addressIndex = src.addressIndex(); + const subscriber = eventData.subscriptions.get(addressIndex); + switch (request.action) { + case "unsubscribe": + if (!subscriber) { + return false; } - } - } else { - // Subscribe logic - let subscriber = eventData.subscriptions.find(sub => sub.dst.addressedTo(src)); - const newFilters = SubscriptionController.normalizeFilters(event.filters); - if (!subscriber) { - eventData.subscriptions.push({ link: link, dst: src, filters: newFilters }); - } else if (subscriber.filters === undefined || newFilters === undefined) { - // Either was already subscribed to all or new request wants all - subscriber.filters = undefined; - } else { - for (const f of newFilters) { - if (!subscriber.filters.includes(f)) { - subscriber.filters.push(f); + subscriber.filters.subtract(request.filters); + if (subscriber.filters.isEmpty()) { + eventData.subscriptions.delete(addressIndex); + } + break; + + case "subscribe": + if (!subscriber) { + eventData.subscriptions.set(addressIndex, { link: link, dst: src, filters: request.filters }); + } else { + subscriber.filters.union(request.filters); + } + break; + + case "replace": + if (request.filters.isEmpty()) { + if (!subscriber) { + return false; } + eventData.subscriptions.delete(addressIndex); + } else if (!subscriber) { + eventData.subscriptions.set(addressIndex, { link: link, dst: src, filters: request.filters }); + } else { + subscriber.filters = request.filters; } - } - if (eventData.subscriptionUpdate) { - const eventReplay = await eventData.subscriptionUpdate(event, src, dst); - if (eventReplay) { - link.sendTo(src, eventReplay); - return true; - } + break; + + default: + throw new Error(`unreachable case: ${String(request.action)}`); + } + + if (eventData.subscriptionUpdate) { + const eventReplay = await eventData.subscriptionUpdate(request, src, dst); + if (eventReplay) { + link.sendTo(src, eventReplay); + return true; } } + return false; } } @@ -245,36 +397,27 @@ export interface SubscribableValue { isDeleted: boolean, } -export type EventSubscribable = 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; /** 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] = [this.makeSnapshot(), 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>(); + /** Filters applied to the filtered handler, if empty the filtered handler will not be called */ + private _filters = SubscriptionFilters.empty(); constructor( - private Event: EventClass, + 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 @@ -294,12 +437,10 @@ export class EventSubscriber< */ handleConnectionEvent(event: "connect" | "drop" | "resume" | "close") { if (event === "connect" || event === "resume") { - this._updateSubscription(); + this._sendRequest("replace", this._callbacks.length > 0 ? SubscriptionFilters.all() : this._filters); } else if (this.synced) { this.synced = false; - for (let callback of this._callbacks) { - callback(null, false); - } + this._notify(null); } } @@ -308,33 +449,35 @@ export class EventSubscriber< * @param event - event from subscribed resource * @internal */ - async _handleEvent(event: T) { - // We support the automatic maintaining of a map for events with the updates property - const updates = event.updates ?? []; - for (const value of updates) { - this.lastResponseTimeMs = Math.max(this.lastResponseTimeMs, value.updatedAtMs); - // Warn about updates which changes content but don't update updatedAtMs - const existing = this.values.get(value.id as K); - if (existing && existing.updatedAtMs === value.updatedAtMs && !isDeepStrictEqual(existing, value)) { - logger.warn( - `${this.Event.name} contains update for value with id ${value.id} that has an identical ` + - "updatedAtMs timestamp but differing content" - ); - } - if (value.isDeleted) { - this.values.delete(value.id as K); - } else { - this.values.set(value.id as K, value); - } - } - // Event handling logic - if (this._syncing && !this.synced) { - this._snapshotLastUpdatedMs = -1; - this.synced = true; + private async _handleEvent(event: E) { + const eventTimeMs = this.getLastUpdatedTimeMs(event); + if (eventTimeMs > this.lastUpdatedMs) { + this.lastUpdatedMs = eventTimeMs; } + this.synced = this._hasSubscriptions(); + this.processEvent(event); + this._notify(event); + } + + private _notify(event: E | null) { for (const callback of this._callbacks) { callback(event, this.synced); } + if (this.filteredHandler && !this._filters.isEmpty()) { + this.filteredHandler(event, this.synced); + } + } + + protected getLastUpdatedTimeMs(event: E) { + return Date.now(); + } + + protected processEvent(event: E) { + // Noop for EventSubscriber + } + + protected makeSnapshot(): S { + return null as S; } /** @@ -344,10 +487,10 @@ export class EventSubscriber< * synced property changes, in which case the event will be null. * @returns function that will unsubscribe from notifications */ - subscribe(handler: EventSubscriberCallback) { + 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 @@ -368,52 +511,145 @@ export class EventSubscriber< } this._callbacks.splice(index, 1); if (this._callbacks.length === 0) { - this._updateSubscription(); + this._sendRequest("replace", this._filters); } }); }; } + /** Returns true if all filters are present */ + hasFilters(filters: string | string[] | SubscriptionFilters) { + if (!(filters instanceof SubscriptionFilters)) { + filters = SubscriptionFilters.fromShorthand(filters); + } + return this._filters.extends(filters); + } + + /** Add filters to the filtered handler */ + addFilters(filters: string | string[] | SubscriptionFilters) { + if (!(filters instanceof SubscriptionFilters)) { + filters = SubscriptionFilters.fromShorthand(filters); + } + this._filters.union(filters); + this._sendRequest("subscribe", filters); + } + + /** Remove filters from the filtered handler */ + removeFilters(filters: string | string[] | SubscriptionFilters) { + if (!(filters instanceof SubscriptionFilters)) { + filters = SubscriptionFilters.fromShorthand(filters); + } + this._filters.subtract(filters); + this._sendRequest("unsubscribe", filters); + } + + /** Clear all filters from the filtered handler, it will no longer be called */ + clearFilters() { + this._filters = SubscriptionFilters.empty(); + if (!this._hasSubscriptions()) { + this._sendRequest("unsubscribe", SubscriptionFilters.all()); + } + } + /** * Obtain a snapshot of the current state of the tracked resource * @returns tuple of values map snapshot and synced property. */ - getSnapshot() { - if (this._snapshotLastUpdatedMs !== this.lastResponseTimeMs) { - this._snapshotLastUpdatedMs = this.lastResponseTimeMs; - this._snapshot = [new Map(this.values), this.synced]; + getSnapshot(): readonly [S, boolean] { + if (this._snapshotLastUpdatedMs !== this.lastUpdatedMs || this._snapshot[1] !== this.synced) { + this._snapshotLastUpdatedMs = this.lastUpdatedMs; + this._snapshot = [this.makeSnapshot(), this.synced]; } - return this._snapshot!; + return this._snapshot; + } + + /** Returns true if there are any active subscription callbacks */ + private _hasSubscriptions() { + return this._callbacks.length > 0 || !this._filters.isEmpty(); } /** * Update the subscription with the controller based on current handler counts */ - async _updateSubscription() { + private async _sendRequest(action: SubscriptionAction, filters: SubscriptionFilters) { if (!this.control.connector.valid) { return; } - const entry = Link._eventsByClass.get(this.Event)!; + const entry = Link._eventsByClass.get(this.Event)!; try { - this._syncing = this._callbacks.length > 0; const updatesSent = await this.control.sendTo("controller", new SubscriptionRequest( - entry.name, - this._syncing, - this.lastResponseTimeMs + entry.name, action, this.lastUpdatedMs, filters )); if (!updatesSent) { - this.synced = this._syncing; - for (const callback of this._callbacks) { - callback(null, this.synced); - } + this.synced = this._hasSubscriptions(); + this._notify(null); } } catch (err: any) { if (!(err instanceof RequestError) || err.code !== "SessionLost") { - logger.error( - `Unexpected error updating ${entry.name} subscription:\n${err.stack}` - ); + logger.error(`Unexpected error updating ${entry.name} subscription:\n${err.stack}`); } } } } + +export class ValueSubscriber< + E extends { value: V }, + V extends SubscribableValue = E["value"], +> extends EventSubscriber | 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)); 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.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..93f330b7 --- /dev/null +++ b/test/lib/subscriptions.test.js @@ -0,0 +1,1314 @@ +"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(".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(); + 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(RegisteredEvent.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 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]; + + 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); + }); + + 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, eventSubscriber; + + beforeEach(function() { + mockControl = new MockControl(new MockConnector( + addr({ controlId: 0 }), + addr("controller"), + )); + eventSubscriber = 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() { + eventSubscriber.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 = eventSubscriber.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 = eventSubscriber.subscribe((e, s) => { called1 = [e, s]; }); + eventSubscriber.subscribe((e, s) => { called2 = [e, s]; }); + + await eventSubscriber._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 = eventSubscriber.subscribe((e, s) => { called1 = [e, s]; }); + eventSubscriber.subscribe((e, s) => { called2 = [e, s]; }); + + unsubscribe(); + await new Promise(r => setImmediate(r)); + 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 = eventSubscriber.subscribe(() => {}); + unsubscribe(); + await onceConnectorSend(); + unsubscribe(); + }); + }); + + describe("filters", function() { + it("should add filters and send subscribe", async function() { + eventSubscriber.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() { + eventSubscriber.addFilters("foo"); + await onceConnectorSend(); + + eventSubscriber.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() { + eventSubscriber.addFilters("foo"); + await onceConnectorSend(); + + eventSubscriber.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() { + eventSubscriber.subscribe(() => {}); + await onceConnectorSend(); + + eventSubscriber.clearFilters(); + await new Promise(r => setImmediate(r)); + + // No new message + assert.equal(mockControl.connector.sentMessages.length, 1); + }); + + it("should correctly report hasFilters", function() { + 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; + eventSubscriber.filteredHandler = () => { called = true; }; + + await eventSubscriber._handleEvent(new RegisteredEvent()); + assert.equal(called, false); + + eventSubscriber.addFilters("foo"); + await onceConnectorSend(); + + await eventSubscriber._handleEvent(new RegisteredEvent()); + assert.equal(called, true); + }); + }); + + describe("getSnapshot()", function() { + it("should cache snapshot until state changes", async function() { + const snap1 = eventSubscriber.getSnapshot(); + const snap2 = eventSubscriber.getSnapshot(); + assert.strictEqual(snap1, snap2); + + await eventSubscriber._handleEvent(new RegisteredEvent()); + const snap3 = eventSubscriber.getSnapshot(); + assert.notStrictEqual(snap1, snap3); + + 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() { + eventSubscriber.subscribe(() => {}); + await onceConnectorSend(); + + eventSubscriber.control.connector.emit("connect"); + await onceConnectorSend(); + assertLastMessage(new lib.SubscriptionRequest( + RegisteredEvent.name, "replace", -1, lib.SubscriptionFilters.all() + )); + + mockControl.connector.sentMessages.pop(); + eventSubscriber.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() { + eventSubscriber.addFilters("foo"); + await onceConnectorSend(); + + eventSubscriber.control.connector.emit("connect"); + await onceConnectorSend(); + assertLastMessage(new lib.SubscriptionRequest( + RegisteredEvent.name, "replace", -1, lib.SubscriptionFilters.fromShorthand("foo") + )); + + mockControl.connector.sentMessages.pop(); + eventSubscriber.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() { + eventSubscriber.synced = true; + eventSubscriber.control.connector.emit("close"); + assert.equal(eventSubscriber.synced, false); + + eventSubscriber.synced = true; + eventSubscriber.handleConnectionEvent("drop"); + assert.equal(eventSubscriber.synced, false); + + eventSubscriber.synced = true; + eventSubscriber.handleConnectionEvent("close"); + assert.equal(eventSubscriber.synced, false); + }); + + it("should invoke callbacks on close", async function() { + let calledWith; + 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(eventSubscriber.synced, true); + calledWith = undefined; + + eventSubscriber.control.connector.emit("close"); + assert.equal(eventSubscriber.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; + eventSubscriber.subscribe(() => {}); + await new Promise(r => setImmediate(r)); + assertNoMessageSent(); + }); + + it("should set synced when no updates returned", async function() { + let calledWith; + eventSubscriber.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(eventSubscriber.synced, true); + }); + + it("should set synced when updates returned via _handleEvent", async function() { + let calledWith; + eventSubscriber.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(eventSubscriber.synced, false); + + const eventUpdate = new RegisteredEvent(); + eventSubscriber._handleEvent(eventUpdate); + + assert.deepEqual(calledWith, [eventUpdate, 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); + }); + }); + }); + + 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 logged = false; + const originalWarn = lib.logger.warn; + lib.logger.warn = msg => { logged = msg; }; + + 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" }, + ])); + } finally { + lib.logger.warn = originalWarn; + } + assert.notEqual(logged, undefined); + }); + }); +});