Add ValueSubscriber and MapSubscriber

packages/lib/src/subscriptions.ts

@@ -384,35 +384,25 @@ export interface SubscribableValue {
 	isDeleted: boolean,
 }
 
-export type EventSubscribable<T, V extends SubscribableValue> = Event<T> & Partial<{
-	updates: V[],
-}>
-
 /**
  * Component for subscribing to and tracking updates of a remote resource
  * Multiple handlers can be subscribed at the same time
  */
-export class EventSubscriber<
+export class EventSubscriber<E, S = null> {
-	T extends EventSubscribable<T, V>,
-	K extends string | number = NonNullable<T["updates"]>[number]["id"],
-	V extends SubscribableValue = NonNullable<T["updates"]>[number],
-> {
 	/** The time at which an event was last received, is less than 0 when there have been no events */
-	lastResponseTimeMs = -1;
+	lastUpdatedMs = -1;
-	/** Values of the subscribed resource */
-	values = new Map<K, V>();
 	/** 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<K, Readonly<V>>, boolean] = [new Map<K, V>(), false];
+	private _snapshot: readonly [S, boolean] = [null as S, false];
-	_snapshotLastUpdatedMs = -1;
+	private _snapshotLastUpdatedMs = -1;
 	/** Callbacks will be called when an event is received or the synced state changes */
-	_callbacks = new Array<EventSubscriberCallback<T>>();
+	private _callbacks = new Array<EventSubscriberCallback<E>>();
 
 	constructor(
-		private Event: EventClass<T>,
+		protected Event: EventClass<E>,
 		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) {
+			this.notify(null);
-				callback(null, false);
-			}
 		}
 	}
 
@@ -447,35 +435,38 @@ export class EventSubscriber<
 	 * @param event - event from subscribed resource
 	 * @internal
 	 */
-	async _handleEvent(event: T) {
+	private async _handleEvent(event: E) {
-		// 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;
 		}
+		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<T>) {
+	subscribe(handler: EventSubscriberCallback<E>) {
 		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() {
+	getSnapshot(): readonly [S, boolean] {
-		if (this._snapshotLastUpdatedMs !== this.lastResponseTimeMs) {
+		if (this._snapshotLastUpdatedMs !== this.lastUpdatedMs || this._snapshot[1] !== this.synced) {
-			this._snapshotLastUpdatedMs = this.lastResponseTimeMs;
+			this._snapshotLastUpdatedMs = this.lastUpdatedMs;
-			this._snapshot = [new Map(this.values), this.synced];
+			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) {
+				this.notify(null);
-					callback(null, this.synced);
-				}
 			}
 		} 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<E, Readonly<V> | null> {
+	/** Current value of the subscribed resource */
+	value: V | null = null;
+
+	protected override getLastUpdatedTimeMs(event: E) {
+		return event.value.updatedAtMs;
+	}
+
+	protected override processEvent(event: E) {
+		this.value = event.value;
+	}
+
+	protected override makeSnapshot() {
+		return this.value;
+	}
+}
+
+export class MapSubscriber<
+	E extends { updates: V[] },
+	K extends string | number = E["updates"][number]["id"],
+	V extends SubscribableValue = E["updates"][number],
+> extends EventSubscriber<E, ReadonlyMap<K, Readonly<V>>> {
+	/** Current value of the subscribed resource */
+	values = new Map<K, V>();
+
+	protected override getLastUpdatedTimeMs(event: E) {
+		let maxTime = this.lastUpdatedMs;
+		for (const update of event.updates) {
+			if (update.updatedAtMs > maxTime) {
+				maxTime = update.updatedAtMs;
+			}
+		}
+		return maxTime;
+	}
+
+	protected override processEvent(event: E) {
+		for (const value of event.updates) {
+			// Warn about updates which changes content but don't update updatedAtMs
+			const existing = this.values.get(value.id as K);
+			if (existing && existing.updatedAtMs === value.updatedAtMs && !isDeepStrictEqual(existing, value)) {
+				logger.warn(
+					`${this.Event.name} contains update for value with id ${value.id} that has an identical ` +
+					"updatedAtMs timestamp but differing content"
+				);
+			}
+			if (value.isDeleted) {
+				this.values.delete(value.id as K);
+			} else {
+				this.values.set(value.id as K, value);
+			}
+		}
+	}
+
+	protected override makeSnapshot() {
+		return new Map(this.values);
+	}
+}

packages/web_ui/src/util/websocket.ts

@@ -60,14 +60,14 @@ export class Control extends lib.Link {
 	logHandlers: Map<lib.LogFilter, logHandler[]> = new Map();
 
 	/** Updates handled by the subscription service */
-	systems = new lib.EventSubscriber(lib.SystemInfoUpdateEvent, this);
+	systems = new lib.MapSubscriber(lib.SystemInfoUpdateEvent, this);
-	hosts = new lib.EventSubscriber(lib.HostUpdatesEvent, this);
+	hosts = new lib.MapSubscriber(lib.HostUpdatesEvent, this);
-	instances = new lib.EventSubscriber(lib.InstanceDetailsUpdatesEvent, this);
+	instances = new lib.MapSubscriber(lib.InstanceDetailsUpdatesEvent, this);
-	saves = new lib.EventSubscriber(lib.InstanceSaveDetailsUpdatesEvent, this);
+	saves = new lib.MapSubscriber(lib.InstanceSaveDetailsUpdatesEvent, this);
-	modPacks = new lib.EventSubscriber(lib.ModPackUpdatesEvent, this);
+	modPacks = new lib.MapSubscriber(lib.ModPackUpdatesEvent, this);
-	mods = new lib.EventSubscriber(lib.ModUpdatesEvent, this);
+	mods = new lib.MapSubscriber(lib.ModUpdatesEvent, this);
-	users = new lib.EventSubscriber(lib.UserUpdatesEvent, this);
+	users = new lib.MapSubscriber(lib.UserUpdatesEvent, this);
-	roles = new lib.EventSubscriber(lib.RoleUpdatesEvent, 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));