From e49b82842d8847e58432d3e1d87d14f51ce77d70 Mon Sep 17 00:00:00 2001 From: Cooldude2606 <25043174+Cooldude2606@users.noreply.github.com> Date: Thu, 28 May 2026 01:12:05 +0100 Subject: [PATCH] Add `ValueSubscriber` and `MapSubscriber` --- packages/lib/src/subscriptions.ts | 148 +++++++++++++++++--------- packages/web_ui/src/util/websocket.ts | 16 +-- 2 files changed, 107 insertions(+), 57 deletions(-) diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index 7b53289e..14f6067c 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -384,35 +384,25 @@ export interface SubscribableValue { isDeleted: boolean, } -export type EventSubscribable = Event & Partial<{ - updates: V[], -}> - /** * Component for subscribing to and tracking updates of a remote resource * Multiple handlers can be subscribed at the same time */ -export class EventSubscriber< - T extends EventSubscribable, - K extends string | number = NonNullable[number]["id"], - V extends SubscribableValue = NonNullable[number], -> { +export class EventSubscriber { /** The time at which an event was last received, is less than 0 when there have been no events */ - lastResponseTimeMs = -1; - /** Values of the subscribed resource */ - values = new Map(); + lastUpdatedMs = -1; /** True if this subscriber is currently synced with the source */ synced = false; /** True if this subscriber is expecting to receive updates from the source */ - _syncing = false; + private _syncing = false; /** Repeat calls to getSnapshot will return the same readonly copy unless values has updated */ - _snapshot: readonly [ReadonlyMap>, boolean] = [new Map(), false]; - _snapshotLastUpdatedMs = -1; + private _snapshot: readonly [S, boolean] = [null as S, false]; + private _snapshotLastUpdatedMs = -1; /** Callbacks will be called when an event is received or the synced state changes */ - _callbacks = new Array>(); + private _callbacks = new Array>(); constructor( - private Event: EventClass, + protected Event: EventClass, public control: Link, ) { control.handle(Event, this._handleEvent.bind(this)); @@ -436,9 +426,7 @@ export class EventSubscriber< this._updateSubscription(); } else if (this.synced) { this.synced = false; - for (let callback of this._callbacks) { - callback(null, false); - } + this.notify(null); } } @@ -447,35 +435,38 @@ export class EventSubscriber< * @param event - event from subscribed resource * @internal */ - async _handleEvent(event: T) { - // We support the automatic maintaining of a map for events with the updates property - const updates = event.updates ?? []; - for (const value of updates) { - this.lastResponseTimeMs = Math.max(this.lastResponseTimeMs, value.updatedAtMs); - // Warn about updates which changes content but don't update updatedAtMs - const existing = this.values.get(value.id as K); - if (existing && existing.updatedAtMs === value.updatedAtMs && !isDeepStrictEqual(existing, value)) { - logger.warn( - `${this.Event.name} contains update for value with id ${value.id} that has an identical ` + - "updatedAtMs timestamp but differing content" - ); - } - if (value.isDeleted) { - this.values.delete(value.id as K); - } else { - this.values.set(value.id as K, value); - } - } + private async _handleEvent(event: E) { // Event handling logic if (this._syncing && !this.synced) { this._snapshotLastUpdatedMs = -1; this.synced = true; } + const eventTimeMs = this.getLastUpdatedTimeMs(event); + if (eventTimeMs > this.lastUpdatedMs) { + this.lastUpdatedMs = eventTimeMs; + } + this.processEvent(event); + this.notify(event); + } + + private notify(event: E | null) { for (const callback of this._callbacks) { callback(event, this.synced); } } + protected getLastUpdatedTimeMs(event: E) { + return Date.now(); + } + + protected processEvent(event: E) { + // Noop for EventSubscriber + } + + protected makeSnapshot(): S { + return null as S; + } + /** * Subscribe to receive all event notifications * @param handler - @@ -483,7 +474,7 @@ export class EventSubscriber< * synced property changes, in which case the event will be null. * @returns function that will unsubscribe from notifications */ - subscribe(handler: EventSubscriberCallback) { + subscribe(handler: EventSubscriberCallback) { this._callbacks.push(handler); if (this._callbacks.length === 1) { this._updateSubscription(); @@ -517,12 +508,12 @@ export class EventSubscriber< * Obtain a snapshot of the current state of the tracked resource * @returns tuple of values map snapshot and synced property. */ - getSnapshot() { - if (this._snapshotLastUpdatedMs !== this.lastResponseTimeMs) { - this._snapshotLastUpdatedMs = this.lastResponseTimeMs; - this._snapshot = [new Map(this.values), this.synced]; + getSnapshot(): readonly [S, boolean] { + if (this._snapshotLastUpdatedMs !== this.lastUpdatedMs || this._snapshot[1] !== this.synced) { + this._snapshotLastUpdatedMs = this.lastUpdatedMs; + this._snapshot = [this.makeSnapshot(), this.synced]; } - return this._snapshot!; + return this._snapshot; } /** @@ -539,13 +530,11 @@ export class EventSubscriber< const updatesSent = await this.control.sendTo("controller", new SubscriptionRequest( entry.name, this._syncing ? "subscribe" : "unsubscribe", - this.lastResponseTimeMs + this.lastUpdatedMs )); if (!updatesSent) { this.synced = this._syncing; - for (const callback of this._callbacks) { - callback(null, this.synced); - } + this.notify(null); } } catch (err: any) { if (!(err instanceof RequestError) || err.code !== "SessionLost") { @@ -556,3 +545,64 @@ export class EventSubscriber< } } } + +export class ValueSubscriber< + E extends { value: V }, + V extends SubscribableValue = E["value"], +> extends EventSubscriber | null> { + /** Current value of the subscribed resource */ + value: V | null = null; + + protected override getLastUpdatedTimeMs(event: E) { + return event.value.updatedAtMs; + } + + protected override processEvent(event: E) { + this.value = event.value; + } + + protected override makeSnapshot() { + return this.value; + } +} + +export class MapSubscriber< + E extends { updates: V[] }, + K extends string | number = E["updates"][number]["id"], + V extends SubscribableValue = E["updates"][number], +> extends EventSubscriber>> { + /** Current value of the subscribed resource */ + values = new Map(); + + protected override getLastUpdatedTimeMs(event: E) { + let maxTime = this.lastUpdatedMs; + for (const update of event.updates) { + if (update.updatedAtMs > maxTime) { + maxTime = update.updatedAtMs; + } + } + return maxTime; + } + + protected override processEvent(event: E) { + for (const value of event.updates) { + // Warn about updates which changes content but don't update updatedAtMs + const existing = this.values.get(value.id as K); + if (existing && existing.updatedAtMs === value.updatedAtMs && !isDeepStrictEqual(existing, value)) { + logger.warn( + `${this.Event.name} contains update for value with id ${value.id} that has an identical ` + + "updatedAtMs timestamp but differing content" + ); + } + if (value.isDeleted) { + this.values.delete(value.id as K); + } else { + this.values.set(value.id as K, value); + } + } + } + + protected override makeSnapshot() { + return new Map(this.values); + } +} diff --git a/packages/web_ui/src/util/websocket.ts b/packages/web_ui/src/util/websocket.ts index 8d2d657c..05747892 100644 --- a/packages/web_ui/src/util/websocket.ts +++ b/packages/web_ui/src/util/websocket.ts @@ -60,14 +60,14 @@ export class Control extends lib.Link { logHandlers: Map = new Map(); /** Updates handled by the subscription service */ - systems = new lib.EventSubscriber(lib.SystemInfoUpdateEvent, this); - hosts = new lib.EventSubscriber(lib.HostUpdatesEvent, this); - instances = new lib.EventSubscriber(lib.InstanceDetailsUpdatesEvent, this); - saves = new lib.EventSubscriber(lib.InstanceSaveDetailsUpdatesEvent, this); - modPacks = new lib.EventSubscriber(lib.ModPackUpdatesEvent, this); - mods = new lib.EventSubscriber(lib.ModUpdatesEvent, this); - users = new lib.EventSubscriber(lib.UserUpdatesEvent, this); - roles = new lib.EventSubscriber(lib.RoleUpdatesEvent, this); + systems = new lib.MapSubscriber(lib.SystemInfoUpdateEvent, this); + hosts = new lib.MapSubscriber(lib.HostUpdatesEvent, this); + instances = new lib.MapSubscriber(lib.InstanceDetailsUpdatesEvent, this); + saves = new lib.MapSubscriber(lib.InstanceSaveDetailsUpdatesEvent, this); + modPacks = new lib.MapSubscriber(lib.ModPackUpdatesEvent, this); + mods = new lib.MapSubscriber(lib.ModUpdatesEvent, this); + users = new lib.MapSubscriber(lib.UserUpdatesEvent, this); + roles = new lib.MapSubscriber(lib.RoleUpdatesEvent, this); /** Cache for factorio versions to avoid repeat calls to the controller */ factorioVersions = new lib.ValueCache(this.requestFactorioVersions.bind(this));