From 62e69e3dbf164cd34de328f785408cfc9cd7cffa Mon Sep 17 00:00:00 2001 From: Cooldude2606 <25043174+Cooldude2606@users.noreply.github.com> Date: Mon, 4 Sep 2023 21:46:28 +0100 Subject: [PATCH 01/15] Add SubscriptionController and EventSubscriber --- packages/lib/src/subscriptions.ts | 273 ++++++++++++++++++++++++++++++ 1 file changed, 273 insertions(+) create mode 100644 packages/lib/src/subscriptions.ts diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts new file mode 100644 index 00000000..a8be14b6 --- /dev/null +++ b/packages/lib/src/subscriptions.ts @@ -0,0 +1,273 @@ +import { Type, Static } from "@sinclair/typebox"; +import { Link, Event, EventClass, RequestHandler, WebSocketClientConnector } from "./link"; +import { Address, MessageRequest } from "./data"; +import { User } from "./users"; + +export type SubscriptionChannelCategoriser = (event: Event) => string | number; +export type SubscriptionRequestHandler = RequestHandler | null>; +export type EventSubscriberCallback = (event: Event) => Promise; + +/** + * Response received by the subscriber after a request + * It can contain an eventReplay value if the event sender implements a subscription handler + */ +export class SubscriptionResponse { + constructor( + public readonly eventReplay: Event | null = null, + ) { + if (eventReplay && !Link._eventsByClass.has(eventReplay.constructor)) { + throw new Error(`Unregistered Event class ${eventReplay.constructor.name}`); + } + } + + static jsonSchema = Type.Union([ + Type.Tuple([ + Type.String(), + Type.Unknown(), + ]), + Type.Null() + ]) + + toJSON() { + if (this.eventReplay) { + const entry = Link._eventsByClass.get(this.eventReplay.constructor)!; + return [entry.name, this.eventReplay]; + } else { + return null; + } + } + + static fromJSON(json: Static): SubscriptionResponse { + if (json) { + const entry = Link._eventsByName.get(json[0]); + if (!entry) { + throw new Error(`Unregistered Event class ${json[0]}`); + } else { + return new SubscriptionResponse(entry.eventFromJSON(json[1])); + } + } else { + return new SubscriptionResponse(); + } + } +} + +/** + * A subscription request sent by a subscriber, this updates what events the subscriber will be sent + * allChannels: false, channels: [] will unsubscribe the subscriber from being sent any events + */ +export class SubscriptionRequest { + declare ["constructor"]: typeof SubscriptionRequest; + static type = "request" as const; + static src = ["control", "instance"] as const; + static dst = "controller" as const; + static plugin = "exp_commands" as const; + static Response = SubscriptionResponse; + static permission(user: User, message: MessageRequest) { + if (typeof message.data === "object" && message.data !== null) { + const data = message.data as Static; + const entry = Link._eventsByName.get(data[0]); + if (entry && entry.Event.permission) { + if (typeof entry.Event.permission === "string") { + user.checkPermission(entry.Event.permission); + } else { + entry.Event.permission(user, message); + } + } + } + } + + constructor( + public eventName: string, + public allChannels: boolean, + public channels: Array = [], + public lastRequestTime: number = 0, + ) { + } + + static jsonSchema = Type.Tuple([ + Type.String(), + Type.Boolean(), + Type.Array(Type.Union([Type.String(), Type.Number()])), + Type.Number(), + ]) + + toJSON() { + return [this.eventName, this.allChannels, this.channels, this.lastRequestTime]; + } + + static fromJSON(json: Static): SubscriptionRequest { + return new this(...json); + } +} + +/** + * A class component to handle incoming subscription requests and offers a method to broadcast to subscribers + * After creation, no other handler can be registered for SubscriptionRequest + */ +export class SubscriptionController { + _events = new Map, + channelCategoriser?: SubscriptionChannelCategoriser, + subscriptions: Map }>, + }>(); + + constructor( + private controller: any, // Controller | Link + ) { + this.controller.handle(SubscriptionRequest, this._handleEvent.bind(this)); + } + + handle(Event: EventClass, subscriptionUpdate?: SubscriptionRequestHandler, channelCategoriser?: SubscriptionChannelCategoriser): void; + handle( + Event: EventClass, + subscriptionUpdate?: SubscriptionRequestHandler, + channelCategoriser?: SubscriptionChannelCategoriser, + ) { + const entry = Link._eventsByClass.get(Event); + if (!entry) { + throw new Error(`Unregistered Event class ${Event.name}`); + } + if (this._events.has(entry.name)) { + throw new Error(`Event ${entry.name} is already registered`); + } + this._events.set(entry.name, { + subscriptionUpdate: subscriptionUpdate, + channelCategoriser: channelCategoriser, + subscriptions: new Map(), + }); + } + + broadcast(event: Event): void; + broadcast(event: Event) { + const entry = Link._eventsByClass.get(event.constructor); + if (!entry) { + throw new Error(`Unregistered Event class ${Event.name}`); + } + const eventData = this._events.get(entry.name); + if (!eventData) { + throw new Error(`Event ${entry.name} is not a registered as subscribable`); + } + const channel = eventData.channelCategoriser ? eventData.channelCategoriser(event) : null; + for (let [link, subscription] of eventData.subscriptions.entries()) { + if (subscription.all || (channel && subscription.channels.includes(channel))) { + link.send(event); + } + } + } + + async _handleEvent(event: SubscriptionRequest, src: Address, dst: Address) { + if (!Link._eventsByName.has(event.eventName)) { + throw new Error(`Event ${event.eventName} is not a registered event`); + } + const eventData = this._events.get(event.eventName); + if (!eventData) { + throw new Error(`Event ${event.eventName} is not a registered as subscribable`); + } + const eventReplay = eventData.subscriptionUpdate ? await eventData.subscriptionUpdate(event, src, dst) : null; + const link: Link = this.controller.wsServer.controlConnections.get(src.id); + if (event.allChannels === false && event.channels.length === 0) { + eventData.subscriptions.delete(link); + return new SubscriptionResponse(eventReplay); + } else { + eventData.subscriptions.set(link, { all: event.allChannels, channels: event.channels }); + return new SubscriptionResponse(eventReplay); + } + } +} + +/** + * A class component to handle incoming subscription requests and offers a method to broadcast to subscribers + * After creation, no other handler can be registered for SubscriptionRequest + */ +export class EventSubscriber { + _callbacks = new Array>() + _channelCallbacks = new Map>>() + lastResponse?: Event = undefined; + lastResponseTime = 0; + + constructor( + private event: EventClass, + private channelCategoriser?: SubscriptionChannelCategoriser, + private control?: Link + ) { + const entry = Link._eventsByClass.get(this.event); + if (!entry) { + throw new Error(`Unregistered Event class ${this.event.name}`); + } + if (this.control) { + this.control.handle(this.event, this._handle.bind(this)); + } + } + + async _handle(response: Event) { + this.lastResponse = response; + this.lastResponseTime = Date.now(); + for (let callback of this._callbacks) { + callback(response); + } + } + + connectControl(control: Link) { + if (this.control === control) return; + this.control = control; + this.control.handle(this.event, this._handle.bind(this)); + } + + subscribeAll(handler: EventSubscriberCallback) { + this._callbacks.push(handler); + this._updateSubscription(); + } + + subscribe(channel: string | number, handler: EventSubscriberCallback) { + if (!this._channelCallbacks.get(channel)) { + this._channelCallbacks.set(channel, []); + } + this._channelCallbacks.get(channel)!.push(handler); + this._updateSubscription(); + } + + unsubscribeAll(handler: EventSubscriberCallback) { + let index = this._callbacks.lastIndexOf(handler); + if (index === -1) { + throw new Error("handler is not registered"); + } + + this._callbacks.splice(index, 1); + this._updateSubscription(); + } + + unsubscribe(channel: string | number, handler: EventSubscriberCallback) { + const channelCallbacks = this._channelCallbacks.get(channel); + if (!channelCallbacks) { + throw new Error("handler is not registered"); + } + + let index = channelCallbacks.lastIndexOf(handler); + if (index === -1) { + throw new Error("handler is not registered"); + } + + if (channelCallbacks.length === 1) { + this._channelCallbacks.delete(channel); + } else { + channelCallbacks.splice(index, 1); + } + this._updateSubscription(); + } + + async _updateSubscription() { + if (!this.control || !(this.control.connector as WebSocketClientConnector).connected) return; + const entry = Link._eventsByClass.get(this.event)!; + + const response = await this.control.send(new SubscriptionRequest( + entry.name, + this._callbacks.length > 0, + [...this._channelCallbacks.keys()], + this.lastResponseTime + )); + + if (response.eventReplay) { + this._handle(response.eventReplay); + } + } +} \ No newline at end of file From 17943c48c3d4e61a9cd1ab67bad126d2d113db4e Mon Sep 17 00:00:00 2001 From: Cooldude2606 <25043174+Cooldude2606@users.noreply.github.com> Date: Mon, 4 Sep 2023 22:26:18 +0100 Subject: [PATCH 02/15] A few renames and comments --- packages/lib/src/subscriptions.ts | 31 ++++++++++++++++++++++--------- 1 file changed, 22 insertions(+), 9 deletions(-) diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index a8be14b6..32cb28f4 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -1,5 +1,5 @@ import { Type, Static } from "@sinclair/typebox"; -import { Link, Event, EventClass, RequestHandler, WebSocketClientConnector } from "./link"; +import { Link, Event, EventClass, RequestHandler, WebSocketClientConnector, WebSocketBaseConnector } from "./link"; import { Address, MessageRequest } from "./data"; import { User } from "./users"; @@ -117,6 +117,10 @@ export class SubscriptionController { this.controller.handle(SubscriptionRequest, this._handleEvent.bind(this)); } + /** + * 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 any client updates their subscription + */ handle(Event: EventClass, subscriptionUpdate?: SubscriptionRequestHandler, channelCategoriser?: SubscriptionChannelCategoriser): void; handle( Event: EventClass, @@ -137,6 +141,9 @@ export class SubscriptionController { }); } + /** + * Broadcast an event to all subscribers of that event, will be filtered by channels when relevant + */ broadcast(event: Event): void; broadcast(event: Event) { const entry = Link._eventsByClass.get(event.constructor); @@ -148,8 +155,10 @@ export class SubscriptionController { throw new Error(`Event ${entry.name} is not a registered as subscribable`); } const channel = eventData.channelCategoriser ? eventData.channelCategoriser(event) : null; - for (let [link, subscription] of eventData.subscriptions.entries()) { - if (subscription.all || (channel && subscription.channels.includes(channel))) { + for (let [link, subscription] of eventData.subscriptions) { + if ((link.connector as WebSocketBaseConnector).closing) { + eventData.subscriptions.delete(link); + } else if ((subscription.all || (channel && subscription.channels.includes(channel)))) { link.send(event); } } @@ -169,6 +178,11 @@ export class SubscriptionController { eventData.subscriptions.delete(link); return new SubscriptionResponse(eventReplay); } else { + if (!eventData.subscriptions.has(link)) { + link.connector.once("close", () => { + eventData.subscriptions.delete(link); + }) + } eventData.subscriptions.set(link, { all: event.allChannels, channels: event.channels }); return new SubscriptionResponse(eventReplay); } @@ -182,12 +196,11 @@ export class SubscriptionController { export class EventSubscriber { _callbacks = new Array>() _channelCallbacks = new Map>>() - lastResponse?: Event = undefined; + lastResponse: Event | null = null; lastResponseTime = 0; constructor( private event: EventClass, - private channelCategoriser?: SubscriptionChannelCategoriser, private control?: Link ) { const entry = Link._eventsByClass.get(this.event); @@ -213,12 +226,12 @@ export class EventSubscriber { this.control.handle(this.event, this._handle.bind(this)); } - subscribeAll(handler: EventSubscriberCallback) { + subscribe(handler: EventSubscriberCallback) { this._callbacks.push(handler); this._updateSubscription(); } - subscribe(channel: string | number, handler: EventSubscriberCallback) { + subscribeToChannel(channel: string | number, handler: EventSubscriberCallback) { if (!this._channelCallbacks.get(channel)) { this._channelCallbacks.set(channel, []); } @@ -226,7 +239,7 @@ export class EventSubscriber { this._updateSubscription(); } - unsubscribeAll(handler: EventSubscriberCallback) { + unsubscribe(handler: EventSubscriberCallback) { let index = this._callbacks.lastIndexOf(handler); if (index === -1) { throw new Error("handler is not registered"); @@ -236,7 +249,7 @@ export class EventSubscriber { this._updateSubscription(); } - unsubscribe(channel: string | number, handler: EventSubscriberCallback) { + unsubscribeFromChannel(channel: string | number, handler: EventSubscriberCallback) { const channelCallbacks = this._channelCallbacks.get(channel); if (!channelCallbacks) { throw new Error("handler is not registered"); From da25196830c98362c2e7fe8a160647274c20f188 Mon Sep 17 00:00:00 2001 From: Cooldude2606 <25043174+Cooldude2606@users.noreply.github.com> Date: Tue, 5 Sep 2023 01:46:45 +0100 Subject: [PATCH 03/15] Refactor to use subsribers --- packages/controller/src/ControlConnection.ts | 128 ------- packages/controller/src/Controller.ts | 90 +++-- packages/lib/index.ts | 1 + packages/lib/src/data/messages_host.ts | 24 +- packages/lib/src/data/messages_instance.ts | 46 +-- packages/lib/src/data/messages_mod.ts | 46 +-- packages/lib/src/data/messages_user.ts | 23 +- packages/lib/src/link/messages.ts | 9 +- packages/lib/src/subscriptions.ts | 41 +- packages/web_ui/src/model/host.ts | 8 +- packages/web_ui/src/model/instance.ts | 8 +- packages/web_ui/src/model/mod_pack.ts | 8 +- packages/web_ui/src/model/mods.ts | 4 +- packages/web_ui/src/model/saves.ts | 4 +- packages/web_ui/src/model/user.tsx | 8 +- packages/web_ui/src/util/websocket.ts | 379 +------------------ 16 files changed, 120 insertions(+), 707 deletions(-) diff --git a/packages/controller/src/ControlConnection.ts b/packages/controller/src/ControlConnection.ts index b87e0255..b02e6f03 100644 --- a/packages/controller/src/ControlConnection.ts +++ b/packages/controller/src/ControlConnection.ts @@ -1,5 +1,4 @@ import type WsServerConnector from "./WsServerConnector"; -import type { HostInfo } from "./HostConnection"; import events from "events"; import fs from "fs-extra"; @@ -11,8 +10,6 @@ const { logFilter, logger } = lib; import BaseConnection from "./BaseConnection"; import * as routes from "./routes"; import Controller from "./Controller"; -import { promises } from "dns"; -import InstanceInfo from "./InstanceInfo"; const strcmp = new Intl.Collator(undefined, { numeric: true, sensitivity: "base" }).compare; @@ -30,12 +27,6 @@ const queryLogTime = new lib.Summary( export default class ControlConnection extends BaseConnection { private _agent: string; private _version: string; - hostSubscriptions: { all:boolean, hostIds:number[] } = { all:false, hostIds:[] }; - instanceSubscriptions: { all:boolean, instanceIds:number[] } = { all:false, instanceIds:[] }; - saveListSubscriptions: { all:boolean, instanceIds:number[] } = { all:false, instanceIds:[] }; - modPackSubscriptions: { all:boolean, modPackIds:number[] } = { all:false, modPackIds:[] }; - modSubscriptions: { all:boolean, modNames:string[] } = { all:false, modNames:[] }; - userSubscriptions: { all:boolean, names:string[]} = { all:false, names:[] }; logTransport: lib.LinkTransport | null = null; logSubscriptions: { all: boolean, @@ -88,33 +79,28 @@ export default class ControlConnection extends BaseConnection { this.handle(lib.ControllerConfigSetFieldRequest, this.handleControllerConfigSetFieldRequest.bind(this)); this.handle(lib.ControllerConfigSetPropRequest, this.handleControllerConfigSetPropRequest.bind(this)); this.handle(lib.HostListRequest, this.handleHostListRequest.bind(this)); - this.handle(lib.HostSetSubscriptionsRequest, this.handleHostSetSubscriptionsRequest.bind(this)); this.handle(lib.HostGenerateTokenRequest, this.handleHostGenerateTokenRequest.bind(this)); this.handle(lib.HostConfigCreateRequest, this.handleHostConfigCreateRequest.bind(this)); this.handle(lib.InstanceDetailsGetRequest, this.handleInstanceDetailsGetRequest.bind(this)); this.handle(lib.InstanceDetailsListRequest, this.handleInstanceDetailsListRequest.bind(this)); - this.handle(lib.InstanceDetailsSetSubscriptionsRequest, this.handleInstanceDetailsSetSubscriptionsRequest.bind(this)); this.handle(lib.InstanceCreateRequest, this.handleInstanceCreateRequest.bind(this)); this.handle(lib.InstanceDeleteRequest, this.handleInstanceDeleteRequest.bind(this)); this.handle(lib.InstanceConfigGetRequest, this.handleInstanceConfigGetRequest.bind(this)); this.handle(lib.InstanceConfigSetFieldRequest, this.handleInstanceConfigSetFieldRequest.bind(this)); this.handle(lib.InstanceConfigSetPropRequest, this.handleInstanceConfigSetPropRequest.bind(this)); this.handle(lib.InstanceAssignRequest, this.handleInstanceAssignRequest.bind(this)); - this.handle(lib.InstanceSetSaveListSubscriptionsRequest, this.handleInstanceSetSaveListSubscriptionsRequest.bind(this)); this.handle(lib.InstanceRenameSaveRequest, this._controller.sendToHostByInstanceId.bind(this._controller)); this.handle(lib.InstanceCopySaveRequest, this._controller.sendToHostByInstanceId.bind(this._controller)); this.handle(lib.InstanceDeleteSaveRequest, this._controller.sendToHostByInstanceId.bind(this._controller)); this.handle(lib.InstanceDownloadSaveRequest, this.handleInstanceDownloadSaveRequest.bind(this)); this.handle(lib.InstanceTransferSaveRequest, this.handleInstanceTransferSaveRequest.bind(this)); this.handle(lib.ModPackListRequest, this.handleModPackListRequest.bind(this)); - this.handle(lib.ModPackSetSubscriptionsRequest, this.handleModPackSetSubscriptionsRequest.bind(this)); this.handle(lib.ModPackCreateRequest, this.handleModPackCreateRequest.bind(this)); this.handle(lib.ModPackUpdateRequest, this.handleModPackUpdateRequest.bind(this)); this.handle(lib.ModPackDeleteRequest, this.handleModPackDeleteRequest.bind(this)); this.handle(lib.ModGetRequest, this.handleModGetRequest.bind(this)); this.handle(lib.ModListRequest, this.handleModListRequest.bind(this)); this.handle(lib.ModSearchRequest, this.handleModSearchRequest.bind(this)); - this.handle(lib.ModSetSubscriptionsRequest, this.handleModSetSubscriptionsRequest.bind(this)); this.handle(lib.ModDownloadRequest, this.handleModDownloadRequest.bind(this)); this.handle(lib.ModDeleteRequest, this.handleModDeleteRequest.bind(this)); this.handle(lib.LogSetSubscriptionsRequest, this.handleLogSetSubscriptionsRequest.bind(this)); @@ -127,7 +113,6 @@ export default class ControlConnection extends BaseConnection { this.handle(lib.RoleDeleteRequest, this.handleRoleDeleteRequest.bind(this)); this.handle(lib.UserGetRequest, this.handleUserGetRequest.bind(this)); this.handle(lib.UserListRequest, this.handleUserListRequest.bind(this)); - this.handle(lib.UserSetSubscriptionsRequest, this.handleUserSetSubscriptionsRequest.bind(this)); this.handle(lib.UserCreateRequest, this.handleUserCreateRequest.bind(this)); this.handle(lib.UserRevokeTokenRequest, this.handleUserRevokeTokenRequest.bind(this)); this.handle(lib.UserUpdateRolesRequest, this.handleUserUpdateRolesRequest.bind(this)); @@ -216,22 +201,6 @@ export default class ControlConnection extends BaseConnection { return list; } - async handleHostSetSubscriptionsRequest(request: lib.HostSetSubscriptionsRequest) { - this.hostSubscriptions = { ...request }; - } - - hostUpdated( - host: HostInfo, - update: lib.HostDetails - ) { - if ( - this.hostSubscriptions.all - || this.hostSubscriptions.hostIds.includes(host.id) - ) { - this.send(new lib.HostUpdateEvent(update)); - } - } - async handleHostGenerateTokenRequest(message: lib.HostGenerateTokenRequest) { let hostId = message.hostId; if (hostId === undefined) { @@ -304,38 +273,6 @@ export default class ControlConnection extends BaseConnection { return list; } - async handleInstanceDetailsSetSubscriptionsRequest(request: lib.InstanceDetailsSetSubscriptionsRequest) { - this.instanceSubscriptions = { ...request }; - } - - instanceUpdated(instance: InstanceInfo) { - if ( - this.instanceSubscriptions.all - || this.instanceSubscriptions.instanceIds.includes(instance.id) - ) { - let assigned_host: number|null|undefined = instance.config.get("instance.assigned_host"); - if (assigned_host === null) { - assigned_host = undefined; - } - - let game_port: number|null|undefined = instance.game_port - if (game_port === null) { - game_port = undefined; - } - - - this.send(new lib.InstanceDetailsUpdateEvent( - new lib.InstanceDetails( - instance.config.get("instance.name"), - instance.id, - assigned_host, - game_port, - instance.status, - ) - )); - } - } - // XXX should probably add a hook for host reuqests? async handleInstanceCreateRequest(request: lib.InstanceCreateRequest) { let instanceConfig = new lib.InstanceConfig("controller"); @@ -378,19 +315,6 @@ export default class ControlConnection extends BaseConnection { await this._controller.instanceAssign(request.instanceId, request.hostId); } - async handleInstanceSetSaveListSubscriptionsRequest(request: lib.InstanceSetSaveListSubscriptionsRequest) { - this.saveListSubscriptions = { ...request }; - } - - saveListUpdate(instanceId: number, saves: lib.SaveDetails[]) { - if ( - this.saveListSubscriptions.all - || this.saveListSubscriptions.instanceIds.includes(instanceId) - ) { - this.send(new lib.InstanceSaveListUpdateEvent(instanceId, saves)); - } - } - async handleInstanceDownloadSaveRequest(request: lib.InstanceDownloadSaveRequest) { let { instanceId, name } = request; let stream = await routes.createProxyStream(this._controller.app); @@ -488,10 +412,6 @@ export default class ControlConnection extends BaseConnection { return [...this._controller.modPacks!.values()]; } - async handleModPackSetSubscriptionsRequest(request: lib.ModPackSetSubscriptionsRequest) { - this.modPackSubscriptions = { ...request }; - } - async handleModPackCreateRequest(request: lib.ModPackCreateRequest) { let modPack = request.modPack; if (modPack.id === undefined) { @@ -524,15 +444,6 @@ export default class ControlConnection extends BaseConnection { this._controller.modPackUpdated(modPack); } - modPackUpdated(modPack: lib.ModPack) { - if ( - this.modPackSubscriptions.all - || this.modPackSubscriptions.modPackIds.includes(modPack.id!) - ) { - this.send(new lib.ModPackUpdateEvent(modPack)); - } - } - async handleModGetRequest(request: lib.ModGetRequest) { let { name, version } = request; let filename = `${name}_${version}.zip`; @@ -636,10 +547,6 @@ export default class ControlConnection extends BaseConnection { }; } - async handleModSetSubscriptionsRequest(request: lib.ModSetSubscriptionsRequest) { - this.modSubscriptions = { ...request }; - } - async handleModDownloadRequest(request: lib.ModDownloadRequest) { let { name, version } = request; let filename = `${name}_${version}.zip`; @@ -672,15 +579,6 @@ export default class ControlConnection extends BaseConnection { this.updateLogSubscriptions(); } - modUpdated(mod: lib.ModInfo) { - if ( - this.modSubscriptions.all - || this.modSubscriptions.modNames.includes(mod.name) - ) { - this.send(new lib.ModUpdateEvent(mod)); - } - } - updateLogSubscriptions() { let { all, controller, hostIds, instanceIds } = this.logSubscriptions; if (all || controller || hostIds.length || instanceIds.length) { @@ -832,32 +730,6 @@ export default class ControlConnection extends BaseConnection { return list; } - async handleUserSetSubscriptionsRequest(request: lib.UserSetSubscriptionsRequest) { - this.userSubscriptions = { ...request }; - } - - userUpdated(user: any) { - if ( - this.userSubscriptions.all - || this.userSubscriptions.names.includes(user.name) - ) { - this.send(new lib.UserUpdateEvent( - new lib.RawUser( - user.name, - [...user.roles].map(role => role.id), - [...user.instances], - user.isAdmin, - user.isBanned, - user.isWhitelisted, - user.banReason, - user.isDeleted, - user.playerStats, - user.instanceStats, - ) - )); - } - } - async handleUserCreateRequest(request: lib.UserCreateRequest) { let user = this._controller.userManager.createUser(request.name); this._controller.userUpdated(user); diff --git a/packages/controller/src/Controller.ts b/packages/controller/src/Controller.ts index f69b102b..9092a744 100644 --- a/packages/controller/src/Controller.ts +++ b/packages/controller/src/Controller.ts @@ -74,6 +74,9 @@ export default class Controller { wsServer: WsServer; debugEvents: events.EventEmitter = new events.EventEmitter(); private _events: events.EventEmitter = new events.EventEmitter(); + + /** Event subscription controller */ + subscriptions: lib.SubscriptionController; // Possible states are new, starting, running, stopping, stopped private _state: string = "new"; @@ -106,6 +109,7 @@ export default class Controller { this.wsServer = new WsServer(this); this.userManager = new UserManager(this.config); + this.subscriptions = new lib.SubscriptionController(this); } async start(args: ControllerArgs) { @@ -236,6 +240,14 @@ export default class Controller { this.app.locals.mainBundle = manifest["main.js"] || "no_web_build"; } + // Handle subscriptions for all internal properties + this.subscriptions.handle(lib.HostUpdateEvent, undefined, event => event.update.id); + this.subscriptions.handle(lib.InstanceDetailsUpdateEvent, undefined, event => event.details.id); + this.subscriptions.handle(lib.InstanceSaveListUpdateEvent, undefined, event => event.instanceId); + this.subscriptions.handle(lib.ModPackUpdateEvent, undefined, event => event.modPack.id!); + this.subscriptions.handle(lib.ModUpdateEvent); + this.subscriptions.handle(lib.UserUpdateEvent, undefined, event => event.user.name); + // Load plugins await this.loadPlugins(); @@ -575,13 +587,7 @@ export default class Controller { host.public_address === null ? undefined : host.public_address, ); - for (let controlConnection of this.wsServer.controlConnections.values()) { - if (controlConnection.connector.closing) { - continue; - } - - controlConnection.hostUpdated(host, update); - } + this.subscriptions.broadcast(new lib.HostUpdateEvent(update)); } /** @@ -761,56 +767,56 @@ export default class Controller { } instanceUpdated(instance: InstanceInfo) { - for (let controlConnection of this.wsServer.controlConnections.values()) { - if (controlConnection.connector.closing) { - continue; - } - - controlConnection.instanceUpdated(instance); + let assigned_host: number|null|undefined = instance.config.get("instance.assigned_host"); + if (assigned_host === null) { + assigned_host = undefined; } + + let game_port: number|null|undefined = instance.game_port + if (game_port === null) { + game_port = undefined; + } + + this.subscriptions.broadcast(new lib.InstanceDetailsUpdateEvent( + new lib.InstanceDetails( + instance.config.get("instance.name"), + instance.id, + assigned_host, + game_port, + instance.status, + ) + )); } saveListUpdate(instanceId: number, saves: lib.SaveDetails[]) { - for (let controlConnection of this.wsServer.controlConnections.values()) { - if (controlConnection.connector.closing) { - continue; - } - - controlConnection.saveListUpdate(instanceId, saves); - } + this.subscriptions.broadcast(new lib.InstanceSaveListUpdateEvent(instanceId, saves)); } modPackUpdated(modPack: lib.ModPack) { - for (let controlConnection of this.wsServer.controlConnections.values()) { - if (controlConnection.connector.closing) { - continue; - } - controlConnection.modPackUpdated(modPack); - } - + this.subscriptions.broadcast(new lib.ModPackUpdateEvent(modPack)); lib.invokeHook(this.plugins, "onModPackUpdated", modPack); } modUpdated(mod: lib.ModInfo) { - for (let controlConnection of this.wsServer.controlConnections.values()) { - if (controlConnection.connector.closing) { - continue; - } - - controlConnection.modUpdated(mod); - } - + this.subscriptions.broadcast(new lib.ModUpdateEvent(mod)); lib.invokeHook(this.plugins, "onModUpdated", mod); } userUpdated(user: lib.User) { - for (let controlConnection of this.wsServer.controlConnections.values()) { - if (controlConnection.connector.closing) { - continue; - } - - controlConnection.userUpdated(user); - } + this.subscriptions.broadcast(new lib.UserUpdateEvent( + new lib.RawUser( + user.name, + [...user.roles].map(role => role.id), + [...user.instances], + user.isAdmin, + user.isBanned, + user.isWhitelisted, + user.banReason, + user.isDeleted, + user.playerStats, + user.instanceStats, + ) + )); } /** diff --git a/packages/lib/index.ts b/packages/lib/index.ts index 14b567e0..19583966 100644 --- a/packages/lib/index.ts +++ b/packages/lib/index.ts @@ -25,6 +25,7 @@ export * from "./src/shared_commands"; export * from "./src/stream"; export * from "./src/users"; export * from "./src/zip_ops"; +export * from "./src/subscriptions" export { default as ExponentialBackoff } from "./src/ExponentialBackoff"; export { default as PlayerStats } from "./src/PlayerStats"; diff --git a/packages/lib/src/data/messages_host.ts b/packages/lib/src/data/messages_host.ts index c271a3b9..b5540ff0 100644 --- a/packages/lib/src/data/messages_host.ts +++ b/packages/lib/src/data/messages_host.ts @@ -35,34 +35,12 @@ export class HostListRequest { static Response = jsonArray(HostDetails); } -export class HostSetSubscriptionsRequest { - declare ["constructor"]: typeof HostSetSubscriptionsRequest; - static type = "request" as const; - static src = "control" as const; - static dst = "controller" as const; - static permission = "core.host.subscribe" as const; - - constructor( - public all: boolean, - public hostIds: number[], - ) { } - - static jsonSchema = Type.Object({ - "all": Type.Boolean(), - "hostIds": Type.Array(Type.Integer()), - }); - - static fromJSON(json: Static) { - return new this(json.all, json.hostIds); - } -} - - export class HostUpdateEvent { declare ["constructor"]: typeof HostUpdateEvent; static type = "event" as const; static src = "controller" as const; static dst = "control" as const; + static permission = "core.host.subscribe" as const; constructor( public update: HostDetails, diff --git a/packages/lib/src/data/messages_instance.ts b/packages/lib/src/data/messages_instance.ts index 42376ae6..8df53810 100644 --- a/packages/lib/src/data/messages_instance.ts +++ b/packages/lib/src/data/messages_instance.ts @@ -67,33 +67,12 @@ export class InstanceDetailsListRequest { static Response = jsonArray(InstanceDetails); }; -export class InstanceDetailsSetSubscriptionsRequest { - declare ["constructor"]: typeof InstanceDetailsSetSubscriptionsRequest; - static type = "request" as const; - static src = "control" as const; - static dst = "controller" as const; - static permission = "core.instance.subscribe" as const; - - constructor( - public all: boolean = false, - public instanceIds: number[] = [], - ) { } - - static jsonSchema = Type.Object({ - "all": Type.Boolean(), - "instanceIds": Type.Array(Type.Integer()), - }); - - static fromJSON(json: Static) { - return new this(json.all, json.instanceIds); - } -}; - export class InstanceDetailsUpdateEvent { declare ["constructor"]: typeof InstanceDetailsUpdateEvent; static type = "event" as const; static src = "controller" as const; static dst = "control" as const; + static permission = "core.instance.subscribe" as const; constructor( public details: InstanceDetails, @@ -299,33 +278,12 @@ export class InstanceListSavesRequest { static Response = jsonArray(SaveDetails); } -export class InstanceSetSaveListSubscriptionsRequest { - declare ["constructor"]: typeof InstanceSetSaveListSubscriptionsRequest; - static type = "request" as const; - static src = "control" as const; - static dst = "controller" as const; - static permission = "core.instance.save.list_subscribe" as const; - - constructor( - public all: boolean = false, - public instanceIds: number[] = [], - ) { } - - static jsonSchema = Type.Object({ - "all": Type.Boolean(), - "instanceIds": Type.Array(Type.Integer()), - }); - - static fromJSON(json: Static) { - return new this(json.all, json.instanceIds); - } -}; - export class InstanceSaveListUpdateEvent { declare ["constructor"]: typeof InstanceSaveListUpdateEvent; static type = "event" as const; static src = ["instance", "host", "controller"] as const; static dst = ["controller", "control"] as const; + static permission = "core.instance.save.list_subscribe" as const; constructor( public instanceId: number, diff --git a/packages/lib/src/data/messages_mod.ts b/packages/lib/src/data/messages_mod.ts index 0c4fe7ce..07d2b999 100644 --- a/packages/lib/src/data/messages_mod.ts +++ b/packages/lib/src/data/messages_mod.ts @@ -104,28 +104,6 @@ export class ModPackDeleteRequest { } } -export class ModPackSetSubscriptionsRequest { - declare ["constructor"]: typeof ModPackSetSubscriptionsRequest; - static type = "request" as const; - static src = "control" as const; - static dst = "controller" as const; - static permission = "core.mod_pack.subscribe" as const; - - constructor( - public all: boolean, - public modPackIds: number[], - ) { } - - static jsonSchema = Type.Object({ - "all": Type.Boolean(), - "modPackIds": Type.Array(Type.Integer()), - }); - - static fromJSON(json: Static) { - return new this(json.all, json.modPackIds); - } -} - export class ModGetRequest { declare ["constructor"]: typeof ModGetRequest; static type = "request" as const; @@ -217,28 +195,6 @@ export class ModSearchRequest { }; } -export class ModSetSubscriptionsRequest { - declare ["constructor"]: typeof ModSetSubscriptionsRequest; - static type = "request" as const; - static src = "control" as const; - static dst = "controller" as const; - static permission = "core.mod.subscribe" as const; - - constructor( - public all: boolean, - public modNames: string[], - ) { } - - static jsonSchema = Type.Object({ - "all": Type.Boolean(), - "modNames": Type.Array(Type.String()), - }); - - static fromJSON(json: Static) { - return new this(json.all, json.modNames); - } -} - export class ModDownloadRequest { declare ["constructor"]: typeof ModDownloadRequest; static type = "request" as const; @@ -290,6 +246,7 @@ export class ModPackUpdateEvent { static type = "event" as const; static src = "controller" as const; static dst = "control" as const; + static permission = "core.mod_pack.subscribe" as const; constructor( public modPack: ModPack, @@ -309,6 +266,7 @@ export class ModUpdateEvent { static type = "event" as const; static src = "controller" as const; static dst = "control" as const; + static permission = "core.mod.subscribe" as const; constructor( public mod: ModInfo, diff --git a/packages/lib/src/data/messages_user.ts b/packages/lib/src/data/messages_user.ts index 210254dd..c7e0a62d 100644 --- a/packages/lib/src/data/messages_user.ts +++ b/packages/lib/src/data/messages_user.ts @@ -251,28 +251,6 @@ export class UserListRequest { static Response = jsonArray(RawUser); } -export class UserSetSubscriptionsRequest { - declare ["constructor"]: typeof UserSetSubscriptionsRequest; - static type = "request" as const; - static src = "control" as const; - static dst = "controller" as const; - static permission = "core.user.subscribe" as const; - - constructor( - public all: boolean, - public names: string[], - ) { } - - static jsonSchema = Type.Object({ - "all": Type.Boolean(), - "names": Type.Array(Type.String()), - }); - - static fromJSON(json: Static) { - return new this(json.all, json.names); - } -} - export class UserCreateRequest { declare ["constructor"]: typeof UserCreateRequest; static type = "request" as const; @@ -460,6 +438,7 @@ export class UserUpdateEvent { static type = "event" as const; static src = "controller" as const; static dst = "control" as const; + static permission = "core.user.subscribe" as const; constructor( public user: RawUser, diff --git a/packages/lib/src/link/messages.ts b/packages/lib/src/link/messages.ts index b6161e10..37e2c249 100644 --- a/packages/lib/src/link/messages.ts +++ b/packages/lib/src/link/messages.ts @@ -5,6 +5,7 @@ import * as host from "../data/messages_host"; import * as instance from "../data/messages_instance"; import * as mod from "../data/messages_mod"; import * as user from "../data/messages_user"; +import * as subscriptions from "../subscriptions" import type { RequestClass, EventClass } from "./link"; /** @@ -27,8 +28,9 @@ export const dataClasses: (RequestClass | EventClass) controller.DebugDumpWsRequest, controller.DebugWsMessageEvent, + subscriptions.SubscriptionRequest, + host.HostListRequest, - host.HostSetSubscriptionsRequest, host.HostUpdateEvent, host.HostMetricsRequest, host.ControllerConnectionEvent, @@ -37,7 +39,6 @@ export const dataClasses: (RequestClass | EventClass) instance.InstanceDetailsGetRequest, instance.InstanceDetailsListRequest, - instance.InstanceDetailsSetSubscriptionsRequest, instance.InstanceDetailsUpdateEvent, instance.InstanceCreateRequest, instance.InstanceConfigGetRequest, @@ -47,7 +48,6 @@ export const dataClasses: (RequestClass | EventClass) instance.InstanceMetricsRequest, instance.InstanceStartRequest, instance.InstanceListSavesRequest, - instance.InstanceSetSaveListSubscriptionsRequest, instance.InstanceSaveListUpdateEvent, instance.InstanceCreateSaveRequest, instance.InstanceRenameSaveRequest, @@ -82,11 +82,9 @@ export const dataClasses: (RequestClass | EventClass) mod.ModPackCreateRequest, mod.ModPackUpdateRequest, mod.ModPackDeleteRequest, - mod.ModPackSetSubscriptionsRequest, mod.ModGetRequest, mod.ModListRequest, mod.ModSearchRequest, - mod.ModSetSubscriptionsRequest, mod.ModDownloadRequest, mod.ModDeleteRequest, mod.ModPackUpdateEvent, @@ -100,7 +98,6 @@ export const dataClasses: (RequestClass | EventClass) user.RoleDeleteRequest, user.UserGetRequest, user.UserListRequest, - user.UserSetSubscriptionsRequest, user.UserCreateRequest, user.UserRevokeTokenRequest, user.UserUpdateRolesRequest, diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index 32cb28f4..e8dff008 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -3,9 +3,9 @@ import { Link, Event, EventClass, RequestHandler, WebSocketClientConnector, WebS import { Address, MessageRequest } from "./data"; import { User } from "./users"; -export type SubscriptionChannelCategoriser = (event: Event) => string | number; +export type SubscriptionChannelCategoriser = (event: T) => string | number; export type SubscriptionRequestHandler = RequestHandler | null>; -export type EventSubscriberCallback = (event: Event) => Promise; +export type EventSubscriberCallback = (value: T) => void; /** * Response received by the subscriber after a request @@ -192,15 +192,17 @@ export class SubscriptionController { /** * A class component to handle incoming subscription requests and offers a method to broadcast to subscribers * After creation, no other handler can be registered for SubscriptionRequest + * An optional prehandler can be given to extract data from the event to be used by the subscribed handlers */ -export class EventSubscriber { - _callbacks = new Array>() - _channelCallbacks = new Map>>() - lastResponse: Event | null = null; +export class EventSubscriber { + _callbacks = new Array>() + _channelCallbacks = new Map>>() + lastResponse: T | null = null; // Does not work with Event lastResponseTime = 0; constructor( private event: EventClass, + private prehandler?: (event: T) => V, private control?: Link ) { const entry = Link._eventsByClass.get(this.event); @@ -212,26 +214,33 @@ export class EventSubscriber { } } - async _handle(response: Event) { + async _handle(response: T) { this.lastResponse = response; this.lastResponseTime = Date.now(); - for (let callback of this._callbacks) { - callback(response); - } + if (this.prehandler) { + for (let callback of this._callbacks) { + callback(this.prehandler(response)); + } + } else { + for (let callback of this._callbacks) { + callback(response as any); // Default behaviour is to pass the event unchanged + } + } } connectControl(control: Link) { - if (this.control === control) return; + if (this.control === control) return Promise.resolve(); this.control = control; this.control.handle(this.event, this._handle.bind(this)); + return this._updateSubscription(); } - subscribe(handler: EventSubscriberCallback) { + subscribe(handler: EventSubscriberCallback) { this._callbacks.push(handler); this._updateSubscription(); } - subscribeToChannel(channel: string | number, handler: EventSubscriberCallback) { + subscribeToChannel(channel: string | number, handler: EventSubscriberCallback) { if (!this._channelCallbacks.get(channel)) { this._channelCallbacks.set(channel, []); } @@ -239,7 +248,7 @@ export class EventSubscriber { this._updateSubscription(); } - unsubscribe(handler: EventSubscriberCallback) { + unsubscribe(handler: EventSubscriberCallback) { let index = this._callbacks.lastIndexOf(handler); if (index === -1) { throw new Error("handler is not registered"); @@ -249,7 +258,7 @@ export class EventSubscriber { this._updateSubscription(); } - unsubscribeFromChannel(channel: string | number, handler: EventSubscriberCallback) { + unsubscribeFromChannel(channel: string | number, handler: EventSubscriberCallback) { const channelCallbacks = this._channelCallbacks.get(channel); if (!channelCallbacks) { throw new Error("handler is not registered"); @@ -280,7 +289,7 @@ export class EventSubscriber { )); if (response.eventReplay) { - this._handle(response.eventReplay); + this._handle(response.eventReplay as T); } } } \ No newline at end of file diff --git a/packages/web_ui/src/model/host.ts b/packages/web_ui/src/model/host.ts index 7338572b..84dbe7e2 100644 --- a/packages/web_ui/src/model/host.ts +++ b/packages/web_ui/src/model/host.ts @@ -38,9 +38,9 @@ export function useHost(id: number): [ HostState, () => void ] { setHost({ ...newHost, loading:false, missing:false, present:true }); } - control.onHostUpdate(id, updateHandler); + control.hostUpdate.subscribeToChannel(id, updateHandler); return () => { - control.offHostUpdate(id, updateHandler); + control.hostUpdate.unsubscribeFromChannel(id, updateHandler); }; }, [id]); @@ -74,9 +74,9 @@ export function useHostList() { }); } - control.onHostUpdate(null, updateHandler); + control.hostUpdate.subscribe(updateHandler); return () => { - control.offHostUpdate(null, updateHandler); + control.hostUpdate.unsubscribe(updateHandler); }; }, []); diff --git a/packages/web_ui/src/model/instance.ts b/packages/web_ui/src/model/instance.ts index 60093453..c262c2b9 100644 --- a/packages/web_ui/src/model/instance.ts +++ b/packages/web_ui/src/model/instance.ts @@ -35,9 +35,9 @@ export function useInstance(id: number): [InstanceState, ()=>void] { setInstance({ ...newInstance, present: true }); } - control.onInstanceUpdate(id, updateHandler); + control.instanceUpdate.subscribeToChannel(id, updateHandler); return () => { - control.offInstanceUpdate(id, updateHandler); + control.instanceUpdate.unsubscribeFromChannel(id, updateHandler); }; }, [id]); @@ -76,9 +76,9 @@ export function useInstanceList() { }); } - control.onInstanceUpdate(null, updateHandler); + control.instanceUpdate.subscribe(updateHandler); return () => { - control.offInstanceUpdate(null, updateHandler); + control.instanceUpdate.unsubscribe(updateHandler); }; }, []); diff --git a/packages/web_ui/src/model/mod_pack.ts b/packages/web_ui/src/model/mod_pack.ts index f30ce27e..0d610f29 100644 --- a/packages/web_ui/src/model/mod_pack.ts +++ b/packages/web_ui/src/model/mod_pack.ts @@ -33,9 +33,9 @@ export function useModPack(id: number) { } updateModPack(); - control.onModPackUpdate(id, setModPack); + control.modPackUpdate.subscribeToChannel(id, setModPack); return () => { - control.offModPackUpdate(id, setModPack); + control.modPackUpdate.unsubscribeFromChannel(id, setModPack); }; }, [id]); @@ -74,9 +74,9 @@ export function useModPackList() { }); } - control.onModPackUpdate(null, updateHandler); + control.modPackUpdate.subscribe(updateHandler); return () => { - control.offModPackUpdate(null, updateHandler); + control.modPackUpdate.unsubscribe(updateHandler); }; }, []); diff --git a/packages/web_ui/src/model/mods.ts b/packages/web_ui/src/model/mods.ts index 6c0497b4..0d131fae 100644 --- a/packages/web_ui/src/model/mods.ts +++ b/packages/web_ui/src/model/mods.ts @@ -38,9 +38,9 @@ export function useModList() { }); } - control.onModUpdate(null, updateHandler); + control.modUpdate.subscribe(updateHandler); return () => { - control.offModUpdate(null, updateHandler); + control.modUpdate.unsubscribe(updateHandler); }; }, []); diff --git a/packages/web_ui/src/model/saves.ts b/packages/web_ui/src/model/saves.ts index 49d6d79b..fdb2c88e 100644 --- a/packages/web_ui/src/model/saves.ts +++ b/packages/web_ui/src/model/saves.ts @@ -29,9 +29,9 @@ export function useSaves(instanceId?: number): lib.SaveDetails[] { setSaves(data.saves); } - control.onSaveListUpdate(instanceId!, updateHandler); + control.saveListUpdate.subscribeToChannel(instanceId!, updateHandler); return () => { - control.offSaveListUpdate(instanceId!, updateHandler); + control.saveListUpdate.unsubscribeFromChannel(instanceId!, updateHandler); }; }, [instanceId]); diff --git a/packages/web_ui/src/model/user.tsx b/packages/web_ui/src/model/user.tsx index 792f8113..3a7cccfb 100644 --- a/packages/web_ui/src/model/user.tsx +++ b/packages/web_ui/src/model/user.tsx @@ -82,9 +82,9 @@ export function useUser(name: string): [RawUserState, () => void] { setUser({ ...newUser, present: true }); } - control.onUserUpdate(name, updateHandler); + control.userUpdate.subscribeToChannel(name, updateHandler); return () => { - control.offUserUpdate(name, updateHandler); + control.userUpdate.unsubscribeFromChannel(name, updateHandler); }; }, [name]); @@ -123,9 +123,9 @@ export function useUserList() { }); } - control.onUserUpdate(null, updateHandler); + control.userUpdate.subscribe(updateHandler); return () => { - control.offUserUpdate(null, updateHandler); + control.userUpdate.unsubscribe(updateHandler); }; }, []); diff --git a/packages/web_ui/src/util/websocket.ts b/packages/web_ui/src/util/websocket.ts index 86602cd8..f503e6e0 100644 --- a/packages/web_ui/src/util/websocket.ts +++ b/packages/web_ui/src/util/websocket.ts @@ -50,15 +50,19 @@ export class Control extends lib.Link { accountName: string | null = null; /** Roles of the account this control link is connected as. */ accountRoles: lib.AccountRole[] | null = null; + + /** Updates not handled by the subscription service */ accountUpdateHandlers: Function[] = []; - hostUpdateHandlers: Map = new Map(); - instanceUpdateHandlers: Map = new Map(); - saveListUpdateHandlers: Map = new Map(); - modPackUpdateHandlers: Map = new Map(); - modUpdateHandlers: Map = new Map(); - userUpdateHandlers: Map = new Map(); logHandlers: Map = new Map(); + /** Updates handled by the subscription service */ + hostUpdate = new lib.EventSubscriber(lib.HostUpdateEvent, event => event.update); + instanceUpdate = new lib.EventSubscriber(lib.InstanceDetailsUpdateEvent, event => event.details); + saveListUpdate = new lib.EventSubscriber(lib.InstanceSaveListUpdateEvent); + modPackUpdate = new lib.EventSubscriber(lib.InstanceSaveListUpdateEvent, event => event.modPack); + modUpdate = new lib.EventSubscriber(lib.InstanceSaveListUpdateEvent, event => event.mod); + userUpdate = new lib.EventSubscriber(lib.InstanceSaveListUpdateEvent, event => event.user); + declare connector: ControlConnector; constructor( @@ -75,22 +79,22 @@ export class Control extends lib.Link { this.accountName = data.account.name; this.accountRoles = data.account.roles; this.emitAccountUpdate(); - this.updateHostSubscriptions().catch(err => logger.error( + this.hostUpdate.connectControl(this).catch(err => logger.error( `Unexpected error updating host subscriptions:\n${err.stack}` - )); - this.updateInstanceSubscriptions().catch(err => logger.error( + ));; + this.instanceUpdate.connectControl(this).catch(err => logger.error( `Unexpected error updating instance subscriptions:\n${err.stack}` )); - this.updateSaveListSubscriptions().catch(err => logger.error( + this.saveListUpdate.connectControl(this).catch(err => logger.error( `Unexpected error updating save list subscriptions:\n${err.stack}` )); - this.updateModPackSubscriptions().catch(err => logger.error( + this.modPackUpdate.connectControl(this).catch(err => logger.error( `Unexpected error updating mod pack subscriptions:\n${err.stack}` )); - this.updateModSubscriptions().catch(err => logger.error( + this.modUpdate.connectControl(this).catch(err => logger.error( `Unexpected error updating mod subscriptions:\n${err.stack}` )); - this.updateUserSubscriptions().catch(err => logger.error( + this.userUpdate.connectControl(this).catch(err => logger.error( `Unexpected error updating user subscriptions:\n${err.stack}` )); this.updateLogSubscriptions().catch(err => logger.error( @@ -113,12 +117,6 @@ export class Control extends lib.Link { } this.handle(lib.AccountUpdateEvent, this.handleAccountUpdateEvent.bind(this)); - this.handle(lib.HostUpdateEvent, this.handleHostUpdateEvent.bind(this)); - this.handle(lib.InstanceDetailsUpdateEvent, this.handleInstanceDetailsUpdateEvent.bind(this)); - this.handle(lib.InstanceSaveListUpdateEvent, this.handleInstanceSaveListUpdateEvent.bind(this)); - this.handle(lib.ModPackUpdateEvent, this.handleModPackUpdateEvent.bind(this)); - this.handle(lib.ModUpdateEvent, this.handleModUpdateEvent.bind(this)); - this.handle(lib.UserUpdateEvent, this.handleUserUpdateEvent.bind(this)); this.handle(lib.LogMessageEvent, this.handleLogMessageEvent.bind(this)); this.handle(lib.DebugWsMessageEvent, this.handleDebugWsMessageEvent.bind(this)); } @@ -149,349 +147,6 @@ export class Control extends lib.Link { this.accountUpdateHandlers.splice(index, 1); } - async handleHostUpdateEvent(event: lib.HostUpdateEvent) { - let handlers: Function[] = ([] as Function[]).concat( - this.hostUpdateHandlers.get(null) || [], - this.hostUpdateHandlers.get(event.update.id) || [], - ); - for (let handler of handlers) { - handler(event.update); - } - } - - async onHostUpdate(id: number|null, handler: hostHandler) { - if (id !== null && !Number.isInteger(id)) { - throw new Error("Invalid host id"); - } - - let handlers = this.hostUpdateHandlers.get(id); - if (!handlers) { - handlers = []; - this.hostUpdateHandlers.set(id, handlers); - } - - handlers.push(handler); - - if (handlers.length === 1) { - await this.updateHostSubscriptions(); - } - } - - async offHostUpdate(id: number|null, handler: hostHandler) { - let handlers = this.hostUpdateHandlers.get(id); - if (!handlers || !handlers.length) { - throw new Error(`No handlers for host ${id} exists`); - } - - let index = handlers.lastIndexOf(handler); - if (index === -1) { - throw new Error(`Given handler is not registered for host ${id}`); - } - - handlers.splice(index, 1); - if (!handlers.length) { - this.hostUpdateHandlers.delete(id); - await this.updateHostSubscriptions(); - } - } - - async updateHostSubscriptions() { - if (!this.connector.connected) { - return; - } - - await this.send(new lib.HostSetSubscriptionsRequest( - this.hostUpdateHandlers.has(null), - [...this.hostUpdateHandlers.keys()].filter(e => e !== null) as number[], - )); - } - - async handleInstanceDetailsUpdateEvent(event: lib.InstanceDetailsUpdateEvent) { - let handlers = ([] as Function[]).concat( - this.instanceUpdateHandlers.get(null) || [], - this.instanceUpdateHandlers.get(event.details.id) || [], - ); - for (let handler of handlers) { - handler(event.details); - } - }; - - async onInstanceUpdate(id: number|null, handler: instanceHandler) { - if (id !== null && !Number.isInteger(id)) { - throw new Error("Invalid instance id"); - } - - let handlers = this.instanceUpdateHandlers.get(id); - if (!handlers) { - handlers = []; - this.instanceUpdateHandlers.set(id, handlers); - } - - handlers.push(handler); - - if (handlers.length === 1) { - await this.updateInstanceSubscriptions(); - } - } - - async offInstanceUpdate(id: number|null, handler: instanceHandler) { - let handlers = this.instanceUpdateHandlers.get(id); - if (!handlers || !handlers.length) { - throw new Error(`No handlers for instance ${id} exist`); - } - - let index = handlers.lastIndexOf(handler); - if (index === -1) { - throw new Error(`Given handler is not registered for instance ${id}`); - } - - handlers.splice(index, 1); - if (!handlers.length) { - this.instanceUpdateHandlers.delete(id); - await this.updateInstanceSubscriptions(); - } - } - - async updateInstanceSubscriptions() { - if (!this.connector.connected) { - return; - } - - await this.send(new lib.InstanceDetailsSetSubscriptionsRequest( - this.instanceUpdateHandlers.has(null), - [...this.instanceUpdateHandlers.keys()].filter(e => e !== null) as number[], - )); - } - - async handleInstanceSaveListUpdateEvent(event: lib.InstanceSaveListUpdateEvent) { - let handlers = this.saveListUpdateHandlers.get(event.instanceId); - for (let handler of handlers || []) { - handler(event); - } - }; - - async onSaveListUpdate(id: number, handler: saveListHandler) { - if (!Number.isInteger(id)) { - throw new Error("Invalid instance id"); - } - - let handlers = this.saveListUpdateHandlers.get(id); - if (!handlers) { - handlers = []; - this.saveListUpdateHandlers.set(id, handlers); - } - - handlers.push(handler); - - if (handlers.length === 1) { - await this.updateSaveListSubscriptions(); - } - } - - async offSaveListUpdate(id: number, handler: saveListHandler) { - let handlers = this.saveListUpdateHandlers.get(id); - if (!handlers) { - throw new Error(`No handlers for instance ${id} exists`); - } - - let index = handlers.lastIndexOf(handler); - if (index === -1) { - throw new Error(`Given handler is not registered for instance ${id}`); - } - - handlers.splice(index, 1); - if (!handlers.length) { - this.saveListUpdateHandlers.delete(id); - await this.updateSaveListSubscriptions(); - } - } - - async updateSaveListSubscriptions() { - if (!this.connector.connected) { - return; - } - - await this.send(new lib.InstanceSetSaveListSubscriptionsRequest( - false, - [...this.saveListUpdateHandlers.keys()].filter(e => e !== null) as number[], - )); - } - - async handleModPackUpdateEvent(event: lib.ModPackUpdateEvent) { - let modPack = event.modPack; - - let handlers = ([] as modPackHandler[]).concat( - this.modPackUpdateHandlers.get(null) || [], - this.modPackUpdateHandlers.get(modPack.id??null) || [], - ); - for (let handler of handlers) { - handler(modPack); - } - } - - async onModPackUpdate(id: number|null, handler: modPackHandler) { - if (id !== null && typeof id !== "number") { - throw new Error("Invalid mod pack id"); - } - - let handlers = this.modPackUpdateHandlers.get(id); - if (!handlers) { - handlers = []; - this.modPackUpdateHandlers.set(id, handlers); - } - - handlers.push(handler); - - if (handlers.length === 1) { - await this.updateModPackSubscriptions(); - } - } - - async offModPackUpdate(id: number|null, handler: modPackHandler) { - let handlers = this.modPackUpdateHandlers.get(id); - if (!handlers || !handlers.length) { - throw new Error(`No handlers for mod pack ${id} exist`); - } - - let index = handlers.lastIndexOf(handler); - if (index === -1) { - throw new Error(`Given handler is not registered for mod pack ${id}`); - } - - handlers.splice(index, 1); - if (!handlers.length) { - this.modPackUpdateHandlers.delete(id); - await this.updateModPackSubscriptions(); - } - } - - async updateModPackSubscriptions() { - if (!this.connector.connected) { - return; - } - - await this.send(new lib.ModPackSetSubscriptionsRequest( - this.modPackUpdateHandlers.has(null), - [...this.modPackUpdateHandlers.keys()] - .filter(k => k !== null && k !== undefined) as number[], - )); - } - - async handleModUpdateEvent(event: lib.ModUpdateEvent) { - let mod = event.mod; - let handlers = ([] as modInfoHandler[]).concat( - this.modUpdateHandlers.get(null) || [], - this.modUpdateHandlers.get(mod.name) || [] - ); - for (let handler of handlers) { - handler(mod); - } - } - - async onModUpdate(name: string|null, handler: modInfoHandler) { - if (name !== null && typeof name !== "string") { - throw new Error("Invalid mod name"); - } - - let handlers = this.modUpdateHandlers.get(name); - if (!handlers) { - handlers = []; - this.modUpdateHandlers.set(name, handlers); - } - - handlers.push(handler); - - if (handlers.length === 1) { - await this.updateModSubscriptions(); - } - } - - async offModUpdate(name: string|null, handler: modInfoHandler) { - let handlers = this.modUpdateHandlers.get(name); - if (!handlers || !handlers.length) { - throw new Error(`No handlers for mod ${name} exist`); - } - - let index = handlers.lastIndexOf(handler); - if (index === -1) { - throw new Error(`Given handler is not registered for mod ${name}`); - } - - handlers.splice(index, 1); - if (!handlers.length) { - this.modUpdateHandlers.delete(name); - await this.updateModSubscriptions(); - } - } - - async updateModSubscriptions() { - if (!this.connector.connected) { - return; - } - - await this.send(new lib.ModSetSubscriptionsRequest( - this.modUpdateHandlers.has(null), - [...this.modUpdateHandlers.keys()].filter(k => k !== null) as string[], - )); - } - - async handleUserUpdateEvent(event: lib.UserUpdateEvent) { - let handlers = ([] as userHandler[]).concat( - this.userUpdateHandlers.get(null) || [], - this.userUpdateHandlers.get(event.user.name) || [], - ); - for (let handler of handlers) { - handler(event.user); - } - }; - - async onUserUpdate(name: string|null, handler: userHandler) { - if (name !== null && typeof name !== "string") { - throw new Error("Invalid user name"); - } - - let handlers = this.userUpdateHandlers.get(name); - if (!handlers) { - handlers = []; - this.userUpdateHandlers.set(name, handlers); - } - - handlers.push(handler); - - if (handlers.length === 1) { - await this.updateUserSubscriptions(); - } - } - - async offUserUpdate(name: string|null, handler: userHandler) { - let handlers = this.userUpdateHandlers.get(name); - if (!handlers || !handlers.length) { - throw new Error(`No handlers for user ${name} exist`); - } - - let index = handlers.lastIndexOf(handler); - if (index === -1) { - throw new Error(`Given handler is not registered for user ${name}`); - } - - handlers.splice(index, 1); - if (!handlers.length) { - this.userUpdateHandlers.delete(name); - await this.updateUserSubscriptions(); - } - } - - async updateUserSubscriptions() { - if (!this.connector.connected) { - return; - } - - await this.send(new lib.UserSetSubscriptionsRequest( - this.userUpdateHandlers.has(null), - [...this.userUpdateHandlers.keys()].filter(e => e !== null) as string[], - )); - } - async handleLogMessageEvent(event: lib.LogMessageEvent) { let info = event.info; From 71166af75be62cc03af19a6cdfb00c2ba0f654ec Mon Sep 17 00:00:00 2001 From: Cooldude2606 <25043174+Cooldude2606@users.noreply.github.com> Date: Wed, 6 Sep 2023 01:09:02 +0100 Subject: [PATCH 04/15] Fixed TestControl subscription requests --- packages/controller/src/Controller.ts | 10 +++++----- packages/lib/src/subscriptions.ts | 4 ++-- test/integration/index.js | 13 ++++++++----- 3 files changed, 15 insertions(+), 12 deletions(-) diff --git a/packages/controller/src/Controller.ts b/packages/controller/src/Controller.ts index 9092a744..c4b5fb12 100644 --- a/packages/controller/src/Controller.ts +++ b/packages/controller/src/Controller.ts @@ -241,12 +241,12 @@ export default class Controller { } // Handle subscriptions for all internal properties - this.subscriptions.handle(lib.HostUpdateEvent, undefined, event => event.update.id); - this.subscriptions.handle(lib.InstanceDetailsUpdateEvent, undefined, event => event.details.id); - this.subscriptions.handle(lib.InstanceSaveListUpdateEvent, undefined, event => event.instanceId); - this.subscriptions.handle(lib.ModPackUpdateEvent, undefined, event => event.modPack.id!); + this.subscriptions.handle(lib.HostUpdateEvent, event => event.update.id); + this.subscriptions.handle(lib.InstanceDetailsUpdateEvent, event => event.details.id); + this.subscriptions.handle(lib.InstanceSaveListUpdateEvent, event => event.instanceId); + this.subscriptions.handle(lib.ModPackUpdateEvent, event => event.modPack.id!); this.subscriptions.handle(lib.ModUpdateEvent); - this.subscriptions.handle(lib.UserUpdateEvent, undefined, event => event.user.name); + this.subscriptions.handle(lib.UserUpdateEvent, event => event.user.name); // Load plugins await this.loadPlugins(); diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index e8dff008..1127be05 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -121,11 +121,11 @@ export class SubscriptionController { * 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 any client updates their subscription */ - handle(Event: EventClass, subscriptionUpdate?: SubscriptionRequestHandler, channelCategoriser?: SubscriptionChannelCategoriser): void; + handle(Event: EventClass, channelCategoriser?: SubscriptionChannelCategoriser, subscriptionUpdate?: SubscriptionRequestHandler,): void; handle( Event: EventClass, - subscriptionUpdate?: SubscriptionRequestHandler, channelCategoriser?: SubscriptionChannelCategoriser, + subscriptionUpdate?: SubscriptionRequestHandler, ) { const entry = Link._eventsByClass.get(Event); if (!entry) { diff --git a/test/integration/index.js b/test/integration/index.js index e9274609..360f0af9 100644 --- a/test/integration/index.js +++ b/test/integration/index.js @@ -34,19 +34,22 @@ class TestControl extends lib.Link { return; } this.send( - new lib.HostSetSubscriptionsRequest(true, []) + new lib.SubscriptionRequest(lib.HostUpdateEvent.name, true) ).catch(err => logger.error(`Error setting host subscriptions:\n${err.stack}`)); this.send( - new lib.InstanceDetailsSetSubscriptionsRequest(true, []) + new lib.SubscriptionRequest(lib.InstanceDetailsUpdateEvent.name, true) ).catch(err => logger.error(`Error setting instance subscriptions:\n${err.stack}`)); this.send( - new lib.InstanceSetSaveListSubscriptionsRequest(true, []) + new lib.SubscriptionRequest(lib.InstanceSaveListUpdateEvent.name, true) ).catch(err => logger.error(`Error setting save list subscriptions:\n${err.stack}`)); this.send( - new lib.ModSetSubscriptionsRequest(true, []) + new lib.SubscriptionRequest(lib.ModUpdateEvent.name, true) ).catch(err => logger.error(`Error setting mod subscriptions:\n${err.stack}`)); this.send( - new lib.UserSetSubscriptionsRequest(true, []) + new lib.SubscriptionRequest(lib.ModPackUpdateEvent.name, true) + ).catch(err => logger.error(`Error setting mod pack subscriptions:\n${err.stack}`)); + this.send( + new lib.SubscriptionRequest(lib.UserUpdateEvent.name, true) ).catch(err => logger.error(`Error setting user subscriptions:\n${err.stack}`)); }); From cccb27c7330b97a397efd281ee0e6627d9b46382 Mon Sep 17 00:00:00 2001 From: Cooldude2606 <25043174+Cooldude2606@users.noreply.github.com> Date: Wed, 6 Sep 2023 21:16:31 +0100 Subject: [PATCH 05/15] Add test skeleton --- test/lib/subscriptions.js | 232 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 232 insertions(+) create mode 100644 test/lib/subscriptions.js diff --git a/test/lib/subscriptions.js b/test/lib/subscriptions.js new file mode 100644 index 00000000..0fb8212f --- /dev/null +++ b/test/lib/subscriptions.js @@ -0,0 +1,232 @@ +"use strict"; +const assert = require("assert").strict; +const lib = require("@clusterio/lib"); + +describe("lib/subscriptions", function() { + describe("class SubscriptionResponse", function() { + describe("constructor()", function() { + it("should be constructable without an event replay", function() { + assert(false); + }); + it("should be constructable with an event replay", function() { + assert(false); + }); + }); + + describe("toJSON()", function() { + it("should be serialisable to json without an event replay", function() { + assert(false); + }); + it("should be serialisable to json with an event replay", function() { + assert(false); + }); + }); + + describe("fromJSON()", function() { + it("should be deserialisable from json without an event replay", function() { + assert(false); + }); + it("should be deserialisable from json with an event replay", function() { + assert(false); + }); + }); + + it("should be round trip json serialisable without an event replay", function() { + assert(false); + }); + + it("should be round trip json serialisable with an event replay", function() { + assert(false); + }); + }); + + describe("class SubscriptionRequest", function() { + describe("constructor()", function() { + it("should be constructable with only an allChannels argument", function() { + assert(false); + }); + it("should be constructable with without a lastRequestTime", function() { + assert(false); + }); + it("should be constructable with with a lastRequestTime", function() { + assert(false); + }); + }); + + describe("permission()", function() { + it("should do nothing when the event has no permission property", function() { + assert(false); + }); + it("should check user permission when the permission property is a string", function() { + assert(false); + }); + it("should call the permission property when it is a function", function() { + assert(false); + }); + }); + + it("should be serialisable to json", function() { + assert(false); + }); + + it("should be deserialisable from json", function() { + assert(false); + }); + + it("should be round trip json serialisable", function() { + assert(false); + }); + }); + + describe("class SubscriptionController", function() { + it("should handle the SubscriptionRequest event", function() { + assert(false); + }); + + describe("handle()", function() { + it("should accept registered events", function() { + assert(false); + }); + it("should not accept unregistered events", function() { + assert(false); + }); + it("should not accept events already handled by the class", function() { + assert(false); + }); + }); + + describe("broadcast()", function() { + it("should not accept unregistered events", function() { + assert(false); + }); + it("should not accept events not handled by the class", function() { + assert(false); + }); + it("should notify all links who subscribed all notifications, when channels are disabled", function() { + assert(false); + }); + it("should notify all links who subscribed all notifications, when channels are enabled", function() { + assert(false); + }); + it("should not notify a link who unsubscribed from all notifications", function() { + assert(false); + }); + it("should notify all links who subscribed the specific channel", function() { + assert(false); + }); + it("should not notify a link who did subscribe the specific channel", function() { + assert(false); + }); + it("should not notify a link who unsubscribed from the specific channel", function() { + assert(false); + }); + it("should not notify links which are closed or closing", function() { + assert(false); + }); + }); + + describe("_handleEvent()", function() { + it("should not accept subscriptions to unregistered events", function() { + assert(false); + }); + it("should not accept subscriptions to events not handled by the class", function() { + assert(false); + }); + it("should accept a subscription to all notifications for an event", function() { + assert(false); + }); + it("should accept a unsubscription from all notifications for an event", function() { + assert(false); + }); + it("should accept a subscription to a specific channel for an event", function() { + assert(false); + }); + it("should accept a unsubscription from a specific channel for an event", function() { + assert(false); + }); + it("should accept a respond with an event replay when returned by the handler", function() { + assert(false); + }); + }); + }); + + describe("class EventSubscriber", function() { + describe("constructor()", function() { + it("should not accept unregistered events", function() { + assert(false); + }); + it("should handle the provided event if a control link is given", function() { + assert(false); + }); + it("should call and use the return of a pre-handler if provided", function() { + assert(false); + }); + }); + + describe("connectControl()", function() { + it("should handle the provided event", function() { + assert(false); + }); + it("should do nothing if the control is already connected", function() { + assert(false); + }); + }); + + describe("subscribe()", function() { + it("should allow subscriptions to an event", function() { + assert(false); + }); + it("should call all handlers who subscripted, when channels are disabled", function() { + assert(false); + }); + it("should call all handlers who subscripted, when channels are enabled", function() { + assert(false); + }); + }); + + describe("subscribeToChannel()", function() { + it("should allow subscriptions to a channel for an event", function() { + assert(false); + }); + it("should call all handlers who subscripted to the specific channel", function() { + assert(false); + }); + it("should not call a handler who did not subscribe to the specific channel", function() { + assert(false); + }); + }); + + describe("unsubscribe()", function() { + it("should allow unsubscribing from an event", function() { + assert(false); + }); + it("should throw an error if the handler was not subscribed", function() { + assert(false); + }); + }); + + describe("unsubscribeFromChannel()", function() { + it("should allow unsubscribing from a channel for an event", function() { + assert(false); + }); + it("should throw an error if the handler was not subscribed", function() { + assert(false); + }); + }); + + describe("_updateSubscription()", function() { + it("should correctly request a subscription for all channels", function() { + assert(false); + }); + it("should correctly request a subscription for some channels", function() { + assert(false); + }); + it("should correctly request a subscription for no channels", function() { + assert(false); + }); + it("should call handlers when a replay event is returned", function() { + assert(false); + }); + }); + }); +}); From 2c9050824e426c6ccca0a26a222b4bbfe7c29c6a Mon Sep 17 00:00:00 2001 From: Cooldude2606 <25043174+Cooldude2606@users.noreply.github.com> Date: Thu, 7 Sep 2023 03:36:38 +0100 Subject: [PATCH 06/15] Test suite mostly ready --- packages/lib/src/subscriptions.ts | 24 +- test/lib/subscriptions.js | 577 +++++++++++++++++++++++------- test/mock.js | 4 +- 3 files changed, 473 insertions(+), 132 deletions(-) diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index 1127be05..c641923c 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -82,6 +82,9 @@ export class SubscriptionRequest { public channels: Array = [], public lastRequestTime: number = 0, ) { + if (!Link._eventsByName.has(eventName)) { + throw new Error(`Unregistered Event class ${eventName}`); + } } static jsonSchema = Type.Tuple([ @@ -114,7 +117,7 @@ export class SubscriptionController { constructor( private controller: any, // Controller | Link ) { - this.controller.handle(SubscriptionRequest, this._handleEvent.bind(this)); + this.controller.handle(SubscriptionRequest, this._handleRequest.bind(this)); } /** @@ -148,7 +151,7 @@ export class SubscriptionController { broadcast(event: Event) { const entry = Link._eventsByClass.get(event.constructor); if (!entry) { - throw new Error(`Unregistered Event class ${Event.name}`); + throw new Error(`Unregistered Event class ${event.constructor.name}`); } const eventData = this._events.get(entry.name); if (!eventData) { @@ -164,7 +167,7 @@ export class SubscriptionController { } } - async _handleEvent(event: SubscriptionRequest, src: Address, dst: Address) { + async _handleRequest(event: SubscriptionRequest, src: Address, dst: Address) { if (!Link._eventsByName.has(event.eventName)) { throw new Error(`Event ${event.eventName} is not a registered event`); } @@ -199,18 +202,19 @@ export class EventSubscriber { _channelCallbacks = new Map>>() lastResponse: T | null = null; // Does not work with Event lastResponseTime = 0; + control?: Link constructor( private event: EventClass, private prehandler?: (event: T) => V, - private control?: Link + control?: Link ) { const entry = Link._eventsByClass.get(this.event); if (!entry) { throw new Error(`Unregistered Event class ${this.event.name}`); } - if (this.control) { - this.control.handle(this.event, this._handle.bind(this)); + if (control) { + this.connectControl(control); } } @@ -237,7 +241,7 @@ export class EventSubscriber { subscribe(handler: EventSubscriberCallback) { this._callbacks.push(handler); - this._updateSubscription(); + return this._updateSubscription(); } subscribeToChannel(channel: string | number, handler: EventSubscriberCallback) { @@ -245,7 +249,7 @@ export class EventSubscriber { this._channelCallbacks.set(channel, []); } this._channelCallbacks.get(channel)!.push(handler); - this._updateSubscription(); + return this._updateSubscription(); } unsubscribe(handler: EventSubscriberCallback) { @@ -255,7 +259,7 @@ export class EventSubscriber { } this._callbacks.splice(index, 1); - this._updateSubscription(); + return this._updateSubscription(); } unsubscribeFromChannel(channel: string | number, handler: EventSubscriberCallback) { @@ -274,7 +278,7 @@ export class EventSubscriber { } else { channelCallbacks.splice(index, 1); } - this._updateSubscription(); + return this._updateSubscription(); } async _updateSubscription() { diff --git a/test/lib/subscriptions.js b/test/lib/subscriptions.js index 0fb8212f..586dba2a 100644 --- a/test/lib/subscriptions.js +++ b/test/lib/subscriptions.js @@ -1,231 +1,566 @@ "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; + } + lib.Link.register(RegisteredEvent); + + class ChannelEvent { + static type = "event"; + static src = ["controller", "control"]; + static dst = ["controller", "control"]; + static permission = null; + } + lib.Link.register(ChannelEvent); + + 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 SubscriptionResponse", function() { - describe("constructor()", function() { - it("should be constructable without an event replay", function() { - assert(false); - }); - it("should be constructable with an event replay", function() { - assert(false); - }); - }); - - describe("toJSON()", function() { - it("should be serialisable to json without an event replay", function() { - assert(false); - }); - it("should be serialisable to json with an event replay", function() { - assert(false); - }); - }); - - describe("fromJSON()", function() { - it("should be deserialisable from json without an event replay", function() { - assert(false); - }); - it("should be deserialisable from json with an event replay", function() { - assert(false); - }); - }); - it("should be round trip json serialisable without an event replay", function() { - assert(false); + const response = new lib.SubscriptionResponse(); + const json = JSON.stringify(response); + assert.equal(typeof json, "string"); + const reconstructed = lib.SubscriptionResponse.fromJSON(JSON.parse(json)); + assert.deepEqual(reconstructed, response); }); it("should be round trip json serialisable with an event replay", function() { - assert(false); + const response = new lib.SubscriptionResponse(new RegisteredEvent()); + const json = JSON.stringify(response); + assert.equal(typeof json, "string"); + const reconstructed = lib.SubscriptionResponse.fromJSON(JSON.parse(json)); + assert.deepEqual(reconstructed, response); + }); + + it("should be round trip json serialisable with a null event replay", function() { + const response = new lib.SubscriptionResponse(null); + const json = JSON.stringify(response); + assert.equal(typeof json, "string"); + const reconstructed = lib.SubscriptionResponse.fromJSON(JSON.parse(json)); + assert.deepEqual(reconstructed, response); + }); + + it("should be throw when given an unregistered event", function() { + assert.throws( + () => new lib.SubscriptionResponse(new UnregisteredEvent()), + new Error(`Unregistered Event class ${UnregisteredEvent.name}`) + ); }); }); describe("class SubscriptionRequest", function() { - describe("constructor()", function() { - it("should be constructable with only an allChannels argument", function() { - assert(false); - }); - it("should be constructable with without a lastRequestTime", function() { - assert(false); - }); - it("should be constructable with with a lastRequestTime", function() { - assert(false); - }); - }); - describe("permission()", function() { it("should do nothing when the event has no permission property", function() { - assert(false); + const mockUser = new MockUser(); + const request = new lib.SubscriptionRequest(RegisteredEvent.name, true); + lib.SubscriptionRequest.permission(mockUser, new lib.MessageRequest( + 0, + addr({ controlId: 0 }), + addr("controller"), + lib.SubscriptionRequest.name, + JSON.stringify(request) + )); + assert.equal(mockUser.lastPermissionCheck, undefined); }); it("should check user permission when the permission property is a string", function() { - assert(false); + const mockUser = new MockUser(); + const request = new lib.SubscriptionRequest(StringPermissionEvent.name, true); + lib.SubscriptionRequest.permission(mockUser, new lib.MessageRequest( + 0, + addr({ controlId: 0 }), + addr("controller"), + lib.SubscriptionRequest.name, + request.toJSON() + )); + assert.equal(mockUser.lastPermissionCheck, "StringPermission"); }); it("should call the permission property when it is a function", function() { - assert(false); + const mockUser = new MockUser(); + const request = new lib.SubscriptionRequest(FunctionPermissionEvent.name, true); + lib.SubscriptionRequest.permission(mockUser, new lib.MessageRequest( + 0, + addr({ controlId: 0 }), + addr("controller"), + lib.SubscriptionRequest.name, + request.toJSON() + )); + assert.equal(mockUser.lastPermissionCheck, "FunctionPermission"); }); }); - it("should be serialisable to json", function() { - assert(false); - }); - - it("should be deserialisable from json", function() { - assert(false); - }); - it("should be round trip json serialisable", function() { - assert(false); + const request = new lib.SubscriptionRequest(RegisteredEvent.name, true, ["channelOne"], 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 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: 1, + request: new lib.SubscriptionRequest(RegisteredEvent.name, true), + src: addr({ controlId: 1 }), + dst: addr("controller"), + }, { + id: 2, + request: new lib.SubscriptionRequest(RegisteredEvent.name, true), + src: addr({ controlId: 2 }), + dst: addr("controller"), + }, { + id: 3, + request: new lib.SubscriptionRequest(ChannelEvent.name, true), + src: addr({ controlId: 3 }), + dst: addr("controller"), + }, { + id: 4, + request: new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelOne"]), + src: addr({ controlId: 4 }), + dst: addr("controller"), + }, { + id: 5, + request: new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelTwo"]), + src: addr({ controlId: 5 }), + dst: addr("controller"), + }]; + + function awaitMessages(ids) { + return Promise.all(ids.map(function(id) { + const link = mockController.wsServer.controlConnections.get(id); + return events.once(link.connector, "send"); + })); + } + + function assertLastEvent(connectorId, Event) { + const link = mockController.wsServer.controlConnections.get(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 = mockController.wsServer.controlConnections.get(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); + }); + it("should handle the SubscriptionRequest event", function() { - assert(false); + assert.equal(mockController.handles.has(lib.SubscriptionRequest), true); }); describe("handle()", function() { it("should accept registered events", function() { - assert(false); + subscriptions.handle(RegisteredEvent); }); it("should not accept unregistered events", function() { - assert(false); + assert.throws( + () => subscriptions.handle(UnregisteredEvent), + new Error(`Unregistered Event class ${UnregisteredEvent.name}`) + ); }); it("should not accept events already handled by the class", function() { - assert(false); + subscriptions.handle(RegisteredEvent); + assert.throws( + () => subscriptions.handle(RegisteredEvent), + new Error(`Event ${RegisteredEvent.name} is already registered`) + ); }); }); describe("broadcast()", function() { + beforeEach(function() { + subscriptions.handle(RegisteredEvent); + subscriptions.handle(ChannelEvent, () => "channelOne"); + for (let connectorData of connectorSetupDate) { + subscriptions._handleRequest(connectorData.request, connectorData.src, connectorData.dst); + } + }); + it("should not accept unregistered events", function() { - assert(false); + 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(false); + assert.throws( + () => subscriptions.broadcast(new StringPermissionEvent()), + new Error(`Event ${StringPermissionEvent.name} is not a registered as subscribable`) + ); }); - it("should notify all links who subscribed all notifications, when channels are disabled", function() { - assert(false); + it("should notify all links who subscribed to all channels", async function() { + const messages = awaitMessages([1, 2]); + subscriptions.broadcast(new RegisteredEvent()); + await messages; + assertLastEvent(1, RegisteredEvent); // RegisteredEvent: All + assertLastEvent(2, RegisteredEvent); // RegisteredEvent: All + assertNoEvent(3); // ChannelEvent: All + assertNoEvent(4); // ChannelEvent: channelOne + assertNoEvent(5); // ChannelEvent: channelTwo }); - it("should notify all links who subscribed all notifications, when channels are enabled", function() { - assert(false); + it("should not notify a link who unsubscribed from all all channels", async function() { + const messages = awaitMessages([2]); + const connectorData = connectorSetupDate[1]; + const request = new lib.SubscriptionRequest(RegisteredEvent.name, false); + await subscriptions._handleRequest(request, connectorData.src, connectorData.dst); + subscriptions.broadcast(new RegisteredEvent()); + await messages; + assertNoEvent(1); + assertLastEvent(2, RegisteredEvent); + assertNoEvent(3); + assertNoEvent(4); + assertNoEvent(5); }); - it("should not notify a link who unsubscribed from all notifications", function() { - assert(false); + it("should notify all links who subscribed the specific channel", async function() { + const messages = awaitMessages([3, 4]); + subscriptions.broadcast(new ChannelEvent()); + await messages; + assertNoEvent(1); + assertNoEvent(2); + assertLastEvent(3, ChannelEvent); + assertNoEvent(4, ChannelEvent); + assertNoEvent(5); }); - it("should notify all links who subscribed the specific channel", function() { - assert(false); + it("should not notify a link who unsubscribed from the specific channel", async function() { + const messages = awaitMessages([3]); + const connectorData = connectorSetupDate[4]; + const request = new lib.SubscriptionRequest(ChannelEvent.name, false, []); + await subscriptions._handleRequest(request, connectorData.src, connectorData.dst); + subscriptions.broadcast(new ChannelEvent()); + await messages; + assertNoEvent(1); + assertNoEvent(2); + assertLastEvent(3, ChannelEvent); + assertNoEvent(4); + assertNoEvent(5); }); - it("should not notify a link who did subscribe the specific channel", function() { - assert(false); - }); - it("should not notify a link who unsubscribed from the specific channel", function() { - assert(false); - }); - it("should not notify links which are closed or closing", function() { - assert(false); + it("should not notify links which are closed or closing", async function() { + const messages = awaitMessages([2]); + const link = mockController.wsServer.controlConnections.get(1); + link.connector.closing = true; + subscriptions.broadcast(new RegisteredEvent()); + await messages; + assertNoEvent(1); + assertLastEvent(2, RegisteredEvent); + assertNoEvent(3); + assertNoEvent(4); + assertNoEvent(5); }); }); - describe("_handleEvent()", function() { + describe("_handleRequest()", function() { + beforeEach(function() { + subscriptions.handle(RegisteredEvent); + subscriptions.handle(ChannelEvent, () => "channelOne"); + }); + it("should not accept subscriptions to unregistered events", function() { - assert(false); + const request = new lib.SubscriptionRequest(RegisteredEvent.name, true); + request.eventName = UnregisteredEvent.name; + assert.throws( + () => subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst), + new Error(`Unregistered Event class ${UnregisteredEvent.name}`) + ); }); it("should not accept subscriptions to events not handled by the class", function() { - assert(false); + const request = new lib.SubscriptionRequest(RegisteredEvent.name, true); + request.eventName = StringPermissionEvent.name; + assert.throws( + () => subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst), + new Error(`Unregistered Event class ${StringPermissionEvent.name}`) + ); }); - it("should accept a subscription to all notifications for an event", function() { - assert(false); + it("should accept a subscription to all channels", async function() { + const messages = awaitMessages([1]); + const request = new lib.SubscriptionRequest(RegisteredEvent.name, true); + await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst); + subscriptions.broadcast(new RegisteredEvent()); + await messages; + assertLastEvent(1, RegisteredEvent); }); - it("should accept a unsubscription from all notifications for an event", function() { - assert(false); + it("should accept a unsubscription from all channels", async function() { + const messages = awaitMessages([1]); + const request = new lib.SubscriptionRequest(RegisteredEvent.name, true); + await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst); + subscriptions.broadcast(new RegisteredEvent()); + await messages; + assertLastEvent(1, RegisteredEvent); + + const link = mockController.wsServer.controlConnections.get(1); + link.connector.sentMessages.pop(); + + const unsubRequest = new lib.SubscriptionRequest(RegisteredEvent.name, false); + await subscriptions._handleRequest(unsubRequest, connectorSetupDate[0].src, connectorSetupDate[0].dst); + subscriptions.broadcast(new RegisteredEvent()); + assertNoEvent(1); }); - it("should accept a subscription to a specific channel for an event", function() { - assert(false); + it("should accept a subscription to a specific channel", async function() { + const messages = awaitMessages([1]); + const request = new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelOne"]); + await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst); + subscriptions.broadcast(new ChannelEvent()); + await messages; + assertLastEvent(1, ChannelEvent); }); - it("should accept a unsubscription from a specific channel for an event", function() { - assert(false); + it("should accept a unsubscription from a specific channel", async function() { + const messages = awaitMessages([1]); + const request = new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelOne"]); + await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst); + subscriptions.broadcast(new ChannelEvent()); + await messages; + assertLastEvent(1, ChannelEvent); + + const link = mockController.wsServer.controlConnections.get(1); + link.connector.sentMessages.pop(); + + const unsubRequest = new lib.SubscriptionRequest(ChannelEvent.name, false, []); + await subscriptions._handleRequest(unsubRequest, connectorSetupDate[0].src, connectorSetupDate[0].dst); + subscriptions.broadcast(new ChannelEvent()); + assertNoEvent(1); }); - it("should accept a respond with an event replay when returned by the handler", function() { - assert(false); + it("should accept a respond with an event replay when returned by the handler", async function() { + const messages = awaitMessages([1]); + subscriptions.handle(StringPermissionEvent, undefined, async function() { + return new StringPermissionEvent(); + }); + + const request = new lib.SubscriptionRequest(StringPermissionEvent.name, true); + await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst); + subscriptions.broadcast(new StringPermissionEvent()); + await messages; + assertLastEvent(1, StringPermissionEvent); + + const link = mockController.wsServer.controlConnections.get(1); + const lastMessage = link.connector.sentMessages.pop(); + const response = lib.SubscriptionResponse.fromJSON(lastMessage.data); + assert.deepEqual(response, new lib.SubscriptionResponse(new StringPermissionEvent())); }); }); }); describe("class EventSubscriber", function() { + let channelEvent, mockControl, registeredEvent; + beforeEach(function() { + mockControl = new MockControl(new MockConnector( + addr({ controlId: 1 }), + addr("controller"), + )); + registeredEvent = new lib.EventSubscriber(RegisteredEvent, undefined, mockControl); + channelEvent = new lib.EventSubscriber(ChannelEvent, undefined, mockControl); + }); + + function assertLastRequest(request) { + const Request = request.constructor; + const lastMessage = mockControl.connector.sentMessages.at(-1); + const name = Request.plugin ? `${Request.plugin}:${Request.name}` : Request.name; + assert.equal(lastMessage.name, name); + assert.deepEqual(lastMessage.data, request); + } + describe("constructor()", function() { it("should not accept unregistered events", function() { - assert(false); + assert.throws( + () => new lib.EventSubscriber(UnregisteredEvent), + new Error(`Unregistered Event class ${UnregisteredEvent.name}`) + ); }); - it("should handle the provided event if a control link is given", function() { - assert(false); + it("should handle the provided event, if a control link is provided", function() { + assert.equal(mockControl._eventHandlers.has(RegisteredEvent), true); + assert.equal(mockControl._eventHandlers.has(ChannelEvent), true); }); - it("should call and use the return of a pre-handler if provided", function() { - assert(false); + it("should call and use the return of a pre-handler, if provided", function() { + let calledWith = null; + function prehandler() { return "PreHandlerReturn"; } + const eventSubscriber = new lib.EventSubscriber(StringPermissionEvent, prehandler, mockControl); + eventSubscriber.subscribe(function(value) { calledWith = value; }); + eventSubscriber._handle(new StringPermissionEvent()); + assert.equal(calledWith, "PreHandlerReturn"); }); }); describe("connectControl()", function() { it("should handle the provided event", function() { - assert(false); + const control = new MockControl(new MockConnector( + addr({ controlId: 1 }), + addr("controller"), + )); + registeredEvent.connectControl(control); + assert.equal(control._eventHandlers.has(RegisteredEvent), true); }); it("should do nothing if the control is already connected", function() { - assert(false); + assert.equal(mockControl._eventHandlers.has(RegisteredEvent), true); + registeredEvent.connectControl(mockControl); + assert.equal(mockControl._eventHandlers.has(RegisteredEvent), true); }); }); describe("subscribe()", function() { it("should allow subscriptions to an event", function() { - assert(false); - }); - it("should call all handlers who subscripted, when channels are disabled", function() { - assert(false); - }); - it("should call all handlers who subscripted, when channels are enabled", function() { - assert(false); + let calledWith = null; + let calledWithTwo = null; + const event = new RegisteredEvent(); + registeredEvent.subscribe(function(value) { calledWith = value; }); + registeredEvent.subscribe(function(value) { calledWithTwo = value; }); + registeredEvent._handle(event); + assert.deepEqual(calledWith, event); + assert.deepEqual(calledWithTwo, event); }); }); describe("subscribeToChannel()", function() { it("should allow subscriptions to a channel for an event", function() { - assert(false); - }); - it("should call all handlers who subscripted to the specific channel", function() { - assert(false); - }); - it("should not call a handler who did not subscribe to the specific channel", function() { - assert(false); + let calledWith = null; + let calledWithTwo = null; + const event = new ChannelEvent(); + channelEvent.subscribeToChannel("channelOne", function(value) { calledWith = value; }); + channelEvent.subscribeToChannel("channelTwo", function(value) { calledWithTwo = value; }); + channelEvent._handle(event); + assert.deepEqual(calledWith, event); + assert.deepEqual(calledWithTwo, null); }); }); describe("unsubscribe()", function() { it("should allow unsubscribing from an event", function() { - assert(false); + let calledWith = null; + let event = new RegisteredEvent(); + function callback(value) { calledWith = value; } + registeredEvent.subscribe(callback); + registeredEvent._handle(event); + assert.deepEqual(calledWith, event); + + calledWith = null; + event = new RegisteredEvent(); + registeredEvent.unsubscribe(callback); + registeredEvent._handle(event); + assert.equal(calledWith, null); }); it("should throw an error if the handler was not subscribed", function() { - assert(false); + function callback(value) { calledWith = value; } + assert.throws( + () => registeredEvent.unsubscribe(callback), + new Error("handler is not registered") + ); }); }); describe("unsubscribeFromChannel()", function() { it("should allow unsubscribing from a channel for an event", function() { - assert(false); + let calledWith = null; + let event = new ChannelEvent(); + function callback(value) { calledWith = value; } + channelEvent.subscribeToChannel("channelOne", callback); + channelEvent._handle(event); + assert.deepEqual(calledWith, event); + + calledWith = null; + event = new ChannelEvent(); + channelEvent.unsubscribeFromChannel("channelOne", callback); + channelEvent._handle(event); + assert.equal(calledWith, null); }); it("should throw an error if the handler was not subscribed", function() { - assert(false); + function callback(value) { calledWith = value; } + assert.throws( + () => channelEvent.unsubscribeFromChannel("channelOne", callback), + new Error("handler is not registered") + ); }); }); describe("_updateSubscription()", function() { - it("should correctly request a subscription for all channels", function() { - assert(false); + it("should correctly request a subscription for all channels", async function() { + const expected = new lib.SubscriptionRequest(RegisteredEvent.name, true); + registeredEvent.subscribe(() => true); + await events.once(mockControl.connector, "send"); + assertLastRequest(expected); }); - it("should correctly request a subscription for some channels", function() { - assert(false); + it("should correctly request a subscription for some channels", async function() { + const expected = new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelOne"]); + channelEvent.subscribeToChannel("channelOne", () => true); + await events.once(mockControl.connector, "send"); + assertLastRequest(expected); }); - it("should correctly request a subscription for no channels", function() { - assert(false); + it("should correctly request a subscription for no channels", async function() { + const expected = new lib.SubscriptionRequest(RegisteredEvent.name, false); + function callback() {} + registeredEvent.subscribe(callback); + await events.once(mockControl.connector, "send"); + registeredEvent.unsubscribe(callback); + await events.once(mockControl.connector, "send"); + assertLastRequest(expected); }); - it("should call handlers when a replay event is returned", function() { - assert(false); + it("should call handlers when a replay event is returned", async function() { + let calledWith = null; + const event = new RegisteredEvent(); + const request = registeredEvent.subscribe(function(value) { calledWith = value; }); + await events.once(mockControl.connector, "send"); + + const responseMessage = new lib.MessageResponse( + 1, + mockControl.connector.dst, + mockControl.connector.src, + new lib.SubscriptionResponse(event) + ); + mockControl.connector.emit("message", responseMessage); + assert.deepEqual(await request, event); }); }); }); diff --git a/test/mock.js b/test/mock.js index 8ad5d751..1958ecfd 100644 --- a/test/mock.js +++ b/test/mock.js @@ -178,9 +178,11 @@ class MockController { ]); this.instances = new Map(); this.hosts = new Map(); + this.handles = new Map(); } - handle() { + handle(eventClass, handler) { + this.handles.set(eventClass, handler); } getControllerUrl() { From 2afddafe4d6014232c53581885279f535c369767 Mon Sep 17 00:00:00 2001 From: Cooldude2606 <25043174+Cooldude2606@users.noreply.github.com> Date: Thu, 7 Sep 2023 20:54:21 +0100 Subject: [PATCH 07/15] All tests passing --- packages/controller/src/Controller.ts | 12 +- packages/lib/src/data/messages_host.ts | 4 + packages/lib/src/data/messages_instance.ts | 8 ++ packages/lib/src/data/messages_mod.ts | 8 ++ packages/lib/src/data/messages_user.ts | 4 + packages/lib/src/subscriptions.ts | 49 ++++--- test/lib/subscriptions.js | 149 +++++++++++---------- 7 files changed, 138 insertions(+), 96 deletions(-) diff --git a/packages/controller/src/Controller.ts b/packages/controller/src/Controller.ts index c4b5fb12..58ace486 100644 --- a/packages/controller/src/Controller.ts +++ b/packages/controller/src/Controller.ts @@ -241,12 +241,12 @@ export default class Controller { } // Handle subscriptions for all internal properties - this.subscriptions.handle(lib.HostUpdateEvent, event => event.update.id); - this.subscriptions.handle(lib.InstanceDetailsUpdateEvent, event => event.details.id); - this.subscriptions.handle(lib.InstanceSaveListUpdateEvent, event => event.instanceId); - this.subscriptions.handle(lib.ModPackUpdateEvent, event => event.modPack.id!); - this.subscriptions.handle(lib.ModUpdateEvent); - this.subscriptions.handle(lib.UserUpdateEvent, event => event.user.name); + this.subscriptions.handle(lib.HostUpdateEvent); + this.subscriptions.handle(lib.InstanceDetailsUpdateEvent); + this.subscriptions.handle(lib.InstanceSaveListUpdateEvent); + this.subscriptions.handle(lib.ModPackUpdateEvent); + this.subscriptions.handle(lib.ModUpdateEvent); + this.subscriptions.handle(lib.UserUpdateEvent); // Load plugins await this.loadPlugins(); diff --git a/packages/lib/src/data/messages_host.ts b/packages/lib/src/data/messages_host.ts index b5540ff0..88838539 100644 --- a/packages/lib/src/data/messages_host.ts +++ b/packages/lib/src/data/messages_host.ts @@ -53,6 +53,10 @@ export class HostUpdateEvent { static fromJSON(json: Static) { return new this(HostDetails.fromJSON(json.update)); } + + get subscriptionChannel() { + return this.update.id; + } } export class HostMetricsRequest { diff --git a/packages/lib/src/data/messages_instance.ts b/packages/lib/src/data/messages_instance.ts index 8df53810..32cdab2b 100644 --- a/packages/lib/src/data/messages_instance.ts +++ b/packages/lib/src/data/messages_instance.ts @@ -87,6 +87,10 @@ export class InstanceDetailsUpdateEvent { static fromJSON(json: Static) { return new this(InstanceDetails.fromJSON(json)); } + + get subscriptionChannel() { + return this.details.id; + } }; export class InstanceCreateRequest { @@ -298,6 +302,10 @@ export class InstanceSaveListUpdateEvent { static fromJSON(json: Static) { return new this(json.instanceId, json.saves.map(i => SaveDetails.fromJSON(i))); } + + get subscriptionChannel() { + return this.instanceId; + } } export class InstanceCreateSaveRequest { diff --git a/packages/lib/src/data/messages_mod.ts b/packages/lib/src/data/messages_mod.ts index 07d2b999..d9225205 100644 --- a/packages/lib/src/data/messages_mod.ts +++ b/packages/lib/src/data/messages_mod.ts @@ -259,6 +259,10 @@ export class ModPackUpdateEvent { static fromJSON(json: Static) { return new this(ModPack.fromJSON(json.modPack)); } + + get subscriptionChannel() { + return this.modPack.id; + } } export class ModUpdateEvent { @@ -279,4 +283,8 @@ export class ModUpdateEvent { static fromJSON(json: Static) { return new this(ModInfo.fromJSON(json.mod)); } + + get subscriptionChannel() { + return this.mod.name; + } } diff --git a/packages/lib/src/data/messages_user.ts b/packages/lib/src/data/messages_user.ts index c7e0a62d..be7c1c9d 100644 --- a/packages/lib/src/data/messages_user.ts +++ b/packages/lib/src/data/messages_user.ts @@ -451,4 +451,8 @@ export class UserUpdateEvent { static fromJSON(json: Static) { return new this(RawUser.fromJSON(json.user)); } + + get subscriptionChannel() { + return this.user.name; + } } diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index c641923c..092d61a0 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -3,10 +3,17 @@ import { Link, Event, EventClass, RequestHandler, WebSocketClientConnector, WebS import { Address, MessageRequest } from "./data"; import { User } from "./users"; -export type SubscriptionChannelCategoriser = (event: T) => string | number; export type SubscriptionRequestHandler = RequestHandler | null>; export type EventSubscriberCallback = (value: T) => void; +export type ChannelEvent = Event & { + get subscriptionChannel(): number | string; +} + +export type ChannelEventClass = EventClass & { + new(...args: any): ChannelEvent +} + /** * Response received by the subscriber after a request * It can contain an eventReplay value if the event sender implements a subscription handler @@ -60,7 +67,6 @@ export class SubscriptionRequest { static type = "request" as const; static src = ["control", "instance"] as const; static dst = "controller" as const; - static plugin = "exp_commands" as const; static Response = SubscriptionResponse; static permission(user: User, message: MessageRequest) { if (typeof message.data === "object" && message.data !== null) { @@ -110,7 +116,6 @@ export class SubscriptionRequest { export class SubscriptionController { _events = new Map, - channelCategoriser?: SubscriptionChannelCategoriser, subscriptions: Map }>, }>(); @@ -124,10 +129,9 @@ export class SubscriptionController { * 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 any client updates their subscription */ - handle(Event: EventClass, channelCategoriser?: SubscriptionChannelCategoriser, subscriptionUpdate?: SubscriptionRequestHandler,): void; + handle(Event: EventClass, subscriptionUpdate?: SubscriptionRequestHandler,): void; handle( Event: EventClass, - channelCategoriser?: SubscriptionChannelCategoriser, subscriptionUpdate?: SubscriptionRequestHandler, ) { const entry = Link._eventsByClass.get(Event); @@ -139,7 +143,6 @@ export class SubscriptionController { } this._events.set(entry.name, { subscriptionUpdate: subscriptionUpdate, - channelCategoriser: channelCategoriser, subscriptions: new Map(), }); } @@ -147,8 +150,8 @@ export class SubscriptionController { /** * Broadcast an event to all subscribers of that event, will be filtered by channels when relevant */ - broadcast(event: Event): void; - broadcast(event: Event) { + broadcast(event: Event | ChannelEvent): void; + broadcast(event: Event | ChannelEvent) { const entry = Link._eventsByClass.get(event.constructor); if (!entry) { throw new Error(`Unregistered Event class ${event.constructor.name}`); @@ -157,7 +160,7 @@ export class SubscriptionController { if (!eventData) { throw new Error(`Event ${entry.name} is not a registered as subscribable`); } - const channel = eventData.channelCategoriser ? eventData.channelCategoriser(event) : null; + const channel = "subscriptionChannel" in event ? event.subscriptionChannel : null; for (let [link, subscription] of eventData.subscriptions) { if ((link.connector as WebSocketBaseConnector).closing) { eventData.subscriptions.delete(link); @@ -205,8 +208,8 @@ export class EventSubscriber { control?: Link constructor( - private event: EventClass, - private prehandler?: (event: T) => V, + private event: EventClass | ChannelEventClass, + private prehandler?: (event: Event) => V, control?: Link ) { const entry = Link._eventsByClass.get(this.event); @@ -218,16 +221,20 @@ export class EventSubscriber { } } - async _handle(response: T) { - this.lastResponse = response; + async _handle(response: Event | ChannelEvent) { + this.lastResponse = response as T; this.lastResponseTime = Date.now(); - if (this.prehandler) { - for (let callback of this._callbacks) { - callback(this.prehandler(response)); - } - } else { - for (let callback of this._callbacks) { - callback(response as any); // Default behaviour is to pass the event unchanged + const value = this.prehandler ? this.prehandler(response) : response as any; + for (let callback of this._callbacks) { + callback(value); + } + const channel = "subscriptionChannel" in response ? response.subscriptionChannel : null; + if (channel) { + const callbacks = this._channelCallbacks.get(channel); + if (callbacks) { + for (let callback of callbacks) { + callback(value); + } } } } @@ -293,7 +300,7 @@ export class EventSubscriber { )); if (response.eventReplay) { - this._handle(response.eventReplay as T); + await this._handle(response.eventReplay); } } } \ No newline at end of file diff --git a/test/lib/subscriptions.js b/test/lib/subscriptions.js index 586dba2a..1efd054a 100644 --- a/test/lib/subscriptions.js +++ b/test/lib/subscriptions.js @@ -27,6 +27,10 @@ describe("lib/subscriptions", function() { static src = ["controller", "control"]; static dst = ["controller", "control"]; static permission = null; + + get subscriptionChannel() { + return "channelOne"; + } } lib.Link.register(ChannelEvent); @@ -146,29 +150,29 @@ describe("lib/subscriptions", function() { describe("class SubscriptionController", function() { let mockController, subscriptions; const connectorSetupDate = [{ + id: 0, + request: new lib.SubscriptionRequest(RegisteredEvent.name, true), + src: addr({ controlId: 0 }), + dst: addr("controller"), + }, { id: 1, request: new lib.SubscriptionRequest(RegisteredEvent.name, true), src: addr({ controlId: 1 }), dst: addr("controller"), }, { id: 2, - request: new lib.SubscriptionRequest(RegisteredEvent.name, true), + request: new lib.SubscriptionRequest(ChannelEvent.name, true), src: addr({ controlId: 2 }), dst: addr("controller"), }, { id: 3, - request: new lib.SubscriptionRequest(ChannelEvent.name, true), + request: new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelOne"]), src: addr({ controlId: 3 }), dst: addr("controller"), }, { id: 4, - request: new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelOne"]), - src: addr({ controlId: 4 }), - dst: addr("controller"), - }, { - id: 5, request: new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelTwo"]), - src: addr({ controlId: 5 }), + src: addr({ controlId: 4 }), dst: addr("controller"), }]; @@ -227,7 +231,7 @@ describe("lib/subscriptions", function() { describe("broadcast()", function() { beforeEach(function() { subscriptions.handle(RegisteredEvent); - subscriptions.handle(ChannelEvent, () => "channelOne"); + subscriptions.handle(ChannelEvent); for (let connectorData of connectorSetupDate) { subscriptions._handleRequest(connectorData.request, connectorData.src, connectorData.dst); } @@ -246,137 +250,141 @@ describe("lib/subscriptions", function() { ); }); it("should notify all links who subscribed to all channels", async function() { - const messages = awaitMessages([1, 2]); + const messages = awaitMessages([0, 1]); subscriptions.broadcast(new RegisteredEvent()); await messages; + assertLastEvent(0, RegisteredEvent); // RegisteredEvent: All assertLastEvent(1, RegisteredEvent); // RegisteredEvent: All - assertLastEvent(2, RegisteredEvent); // RegisteredEvent: All - assertNoEvent(3); // ChannelEvent: All - assertNoEvent(4); // ChannelEvent: channelOne - assertNoEvent(5); // ChannelEvent: channelTwo + assertNoEvent(2); // ChannelEvent: All + assertNoEvent(3); // ChannelEvent: channelOne + assertNoEvent(4); // ChannelEvent: channelTwo }); it("should not notify a link who unsubscribed from all all channels", async function() { - const messages = awaitMessages([2]); - const connectorData = connectorSetupDate[1]; + const messages = awaitMessages([1]); + const connectorData = connectorSetupDate[0]; const request = new lib.SubscriptionRequest(RegisteredEvent.name, false); await subscriptions._handleRequest(request, connectorData.src, connectorData.dst); + mockController.wsServer.controlConnections.get(0).connector.sentMessages.pop(); // Response to request subscriptions.broadcast(new RegisteredEvent()); await messages; - assertNoEvent(1); - assertLastEvent(2, RegisteredEvent); + assertNoEvent(0); + assertLastEvent(1, RegisteredEvent); + assertNoEvent(2); assertNoEvent(3); assertNoEvent(4); - assertNoEvent(5); }); it("should notify all links who subscribed the specific channel", async function() { - const messages = awaitMessages([3, 4]); + const messages = awaitMessages([2, 3]); subscriptions.broadcast(new ChannelEvent()); await messages; + assertNoEvent(0); assertNoEvent(1); - assertNoEvent(2); - assertLastEvent(3, ChannelEvent); - assertNoEvent(4, ChannelEvent); - assertNoEvent(5); - }); - it("should not notify a link who unsubscribed from the specific channel", async function() { - const messages = awaitMessages([3]); - const connectorData = connectorSetupDate[4]; - const request = new lib.SubscriptionRequest(ChannelEvent.name, false, []); - await subscriptions._handleRequest(request, connectorData.src, connectorData.dst); - subscriptions.broadcast(new ChannelEvent()); - await messages; - assertNoEvent(1); - assertNoEvent(2); + assertLastEvent(2, ChannelEvent); assertLastEvent(3, ChannelEvent); assertNoEvent(4); - assertNoEvent(5); + }); + it("should not notify a link who unsubscribed from the specific channel", async function() { + const messages = awaitMessages([2]); + const connectorData = connectorSetupDate[3]; + const request = new lib.SubscriptionRequest(ChannelEvent.name, false, []); + await subscriptions._handleRequest(request, connectorData.src, connectorData.dst); + mockController.wsServer.controlConnections.get(3).connector.sentMessages.pop(); // Response to request + subscriptions.broadcast(new ChannelEvent()); + await messages; + assertNoEvent(0); + assertNoEvent(1); + assertLastEvent(2, ChannelEvent); + assertNoEvent(3); + assertNoEvent(4); }); it("should not notify links which are closed or closing", async function() { - const messages = awaitMessages([2]); - const link = mockController.wsServer.controlConnections.get(1); + const messages = awaitMessages([1]); + const link = mockController.wsServer.controlConnections.get(0); link.connector.closing = true; subscriptions.broadcast(new RegisteredEvent()); await messages; - assertNoEvent(1); - assertLastEvent(2, RegisteredEvent); + assertNoEvent(0); + assertLastEvent(1, RegisteredEvent); + assertNoEvent(2); assertNoEvent(3); assertNoEvent(4); - assertNoEvent(5); }); }); describe("_handleRequest()", function() { beforeEach(function() { subscriptions.handle(RegisteredEvent); - subscriptions.handle(ChannelEvent, () => "channelOne"); + subscriptions.handle(ChannelEvent); }); it("should not accept subscriptions to unregistered events", function() { const request = new lib.SubscriptionRequest(RegisteredEvent.name, true); request.eventName = UnregisteredEvent.name; - assert.throws( - () => subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst), + const connectorData = connectorSetupDate[0]; + assert.rejects( + subscriptions._handleRequest(request, connectorData.src, connectorData.dst), new Error(`Unregistered Event class ${UnregisteredEvent.name}`) ); }); it("should not accept subscriptions to events not handled by the class", function() { const request = new lib.SubscriptionRequest(RegisteredEvent.name, true); request.eventName = StringPermissionEvent.name; - assert.throws( - () => subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst), - new Error(`Unregistered Event class ${StringPermissionEvent.name}`) + const connectorData = connectorSetupDate[0]; + assert.rejects( + subscriptions._handleRequest(request, connectorData.src, connectorData.dst), + new Error(`Event ${StringPermissionEvent.eventName} is not a registered as subscribable`) ); }); it("should accept a subscription to all channels", async function() { - const messages = awaitMessages([1]); + const messages = awaitMessages([0]); const request = new lib.SubscriptionRequest(RegisteredEvent.name, true); await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst); subscriptions.broadcast(new RegisteredEvent()); await messages; - assertLastEvent(1, RegisteredEvent); + assertLastEvent(0, RegisteredEvent); }); it("should accept a unsubscription from all channels", async function() { - const messages = awaitMessages([1]); + const messages = awaitMessages([0]); const request = new lib.SubscriptionRequest(RegisteredEvent.name, true); await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst); subscriptions.broadcast(new RegisteredEvent()); await messages; - assertLastEvent(1, RegisteredEvent); + assertLastEvent(0, RegisteredEvent); - const link = mockController.wsServer.controlConnections.get(1); + const link = mockController.wsServer.controlConnections.get(0); link.connector.sentMessages.pop(); const unsubRequest = new lib.SubscriptionRequest(RegisteredEvent.name, false); await subscriptions._handleRequest(unsubRequest, connectorSetupDate[0].src, connectorSetupDate[0].dst); subscriptions.broadcast(new RegisteredEvent()); - assertNoEvent(1); + assertNoEvent(0); }); it("should accept a subscription to a specific channel", async function() { - const messages = awaitMessages([1]); + const messages = awaitMessages([0]); const request = new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelOne"]); await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst); subscriptions.broadcast(new ChannelEvent()); await messages; - assertLastEvent(1, ChannelEvent); + assertLastEvent(0, ChannelEvent); }); it("should accept a unsubscription from a specific channel", async function() { - const messages = awaitMessages([1]); + const messages = awaitMessages([0]); const request = new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelOne"]); await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst); subscriptions.broadcast(new ChannelEvent()); await messages; - assertLastEvent(1, ChannelEvent); + assertLastEvent(0, ChannelEvent); - const link = mockController.wsServer.controlConnections.get(1); + const link = mockController.wsServer.controlConnections.get(0); link.connector.sentMessages.pop(); const unsubRequest = new lib.SubscriptionRequest(ChannelEvent.name, false, []); await subscriptions._handleRequest(unsubRequest, connectorSetupDate[0].src, connectorSetupDate[0].dst); subscriptions.broadcast(new ChannelEvent()); - assertNoEvent(1); + assertNoEvent(0); }); it("should accept a respond with an event replay when returned by the handler", async function() { - const messages = awaitMessages([1]); + const messages = awaitMessages([0]); subscriptions.handle(StringPermissionEvent, undefined, async function() { return new StringPermissionEvent(); }); @@ -385,12 +393,13 @@ describe("lib/subscriptions", function() { await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst); subscriptions.broadcast(new StringPermissionEvent()); await messages; - assertLastEvent(1, StringPermissionEvent); + assertLastEvent(0, StringPermissionEvent); - const link = mockController.wsServer.controlConnections.get(1); + const link = mockController.wsServer.controlConnections.get(0); const lastMessage = link.connector.sentMessages.pop(); - const response = lib.SubscriptionResponse.fromJSON(lastMessage.data); - assert.deepEqual(response, new lib.SubscriptionResponse(new StringPermissionEvent())); + const eventEntry = lib.Link._eventsByClass.get(StringPermissionEvent); + const response = eventEntry.eventFromJSON(lastMessage.data); + assert.deepEqual(response, new StringPermissionEvent()); }); }); }); @@ -399,7 +408,7 @@ describe("lib/subscriptions", function() { let channelEvent, mockControl, registeredEvent; beforeEach(function() { mockControl = new MockControl(new MockConnector( - addr({ controlId: 1 }), + addr({ controlId: 0 }), addr("controller"), )); registeredEvent = new lib.EventSubscriber(RegisteredEvent, undefined, mockControl); @@ -438,7 +447,7 @@ describe("lib/subscriptions", function() { describe("connectControl()", function() { it("should handle the provided event", function() { const control = new MockControl(new MockConnector( - addr({ controlId: 1 }), + addr({ controlId: 0 }), addr("controller"), )); registeredEvent.connectControl(control); @@ -550,17 +559,19 @@ describe("lib/subscriptions", function() { it("should call handlers when a replay event is returned", async function() { let calledWith = null; const event = new RegisteredEvent(); + const message = events.once(mockControl.connector, "send"); const request = registeredEvent.subscribe(function(value) { calledWith = value; }); - await events.once(mockControl.connector, "send"); + await message; const responseMessage = new lib.MessageResponse( 1, mockControl.connector.dst, - mockControl.connector.src, - new lib.SubscriptionResponse(event) + new lib.Address(lib.Address.control, 0, mockControl._nextRequestId - 1), + new lib.SubscriptionResponse(event).toJSON() ); mockControl.connector.emit("message", responseMessage); - assert.deepEqual(await request, event); + await request; + assert.deepEqual(calledWith, event); }); }); }); From 621f8508048902573f0b7b1178cdbe7d30703fcd Mon Sep 17 00:00:00 2001 From: Cooldude2606 <25043174+Cooldude2606@users.noreply.github.com> Date: Thu, 7 Sep 2023 21:36:27 +0100 Subject: [PATCH 08/15] Some cleanup --- packages/lib/src/subscriptions.ts | 49 +++++++++++++---- test/lib/subscriptions.js | 91 +++++++++++-------------------- 2 files changed, 70 insertions(+), 70 deletions(-) diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index 092d61a0..9d0b0ed2 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -17,6 +17,7 @@ export type ChannelEventClass = EventClass & { /** * Response received by the subscriber after a request * It can contain an eventReplay value if the event sender implements a subscription handler + * This replay will be send to the handlers as if an update had just occurred */ export class SubscriptionResponse { constructor( @@ -60,7 +61,8 @@ export class SubscriptionResponse { /** * A subscription request sent by a subscriber, this updates what events the subscriber will be sent - * allChannels: false, channels: [] will unsubscribe the subscriber from being sent any events + * The permission for this request copies the permission from the event being subscribed to + * allChannels: false, channels: [] will unsubscribe the subscriber from all notifications */ export class SubscriptionRequest { declare ["constructor"]: typeof SubscriptionRequest; @@ -110,8 +112,8 @@ export class SubscriptionRequest { } /** - * A class component to handle incoming subscription requests and offers a method to broadcast to subscribers - * After creation, no other handler can be registered for SubscriptionRequest + * A class component to handle incoming subscription requests and offers a method to broadcast events to subscribers + * After creation, no other handler can be registered for SubscriptionRequest on the controller */ export class SubscriptionController { _events = new Map(event: Event | ChannelEvent): void; broadcast(event: Event | ChannelEvent) { @@ -170,6 +172,9 @@ export class SubscriptionController { } } + /** + * Handle incoming event subscription requests + */ async _handleRequest(event: SubscriptionRequest, src: Address, dst: Address) { if (!Link._eventsByName.has(event.eventName)) { throw new Error(`Event ${event.eventName} is not a registered event`); @@ -182,7 +187,6 @@ export class SubscriptionController { const link: Link = this.controller.wsServer.controlConnections.get(src.id); if (event.allChannels === false && event.channels.length === 0) { eventData.subscriptions.delete(link); - return new SubscriptionResponse(eventReplay); } else { if (!eventData.subscriptions.has(link)) { link.connector.once("close", () => { @@ -190,20 +194,20 @@ export class SubscriptionController { }) } eventData.subscriptions.set(link, { all: event.allChannels, channels: event.channels }); - return new SubscriptionResponse(eventReplay); } + return new SubscriptionResponse(eventReplay); } } /** - * A class component to handle incoming subscription requests and offers a method to broadcast to subscribers - * After creation, no other handler can be registered for SubscriptionRequest - * An optional prehandler can be given to extract data from the event to be used by the subscribed handlers + * A class component to allow subscribing and unsubscribing to/from an event and its channels + * Channels are an optional feature which can be implemented to only get notifications of reliant updates + * Multiple handlers can be subscribed at the same time, on the same or different channels */ export class EventSubscriber { _callbacks = new Array>() _channelCallbacks = new Map>>() - lastResponse: T | null = null; // Does not work with Event + lastResponse: Event | null = null; lastResponseTime = 0; control?: Link @@ -221,8 +225,11 @@ export class EventSubscriber { } } + /** + * Handle incoming events and distribute it to the correct callbacks + */ async _handle(response: Event | ChannelEvent) { - this.lastResponse = response as T; + this.lastResponse = response; this.lastResponseTime = Date.now(); const value = this.prehandler ? this.prehandler(response) : response as any; for (let callback of this._callbacks) { @@ -239,6 +246,9 @@ export class EventSubscriber { } } + /** + * If a control link was not connected at creation, it can be connected here (normally during onControllerConnectionEvent) + */ connectControl(control: Link) { if (this.control === control) return Promise.resolve(); this.control = control; @@ -246,11 +256,17 @@ export class EventSubscriber { return this._updateSubscription(); } + /** + * Subscribe to receive all event notifications + */ subscribe(handler: EventSubscriberCallback) { this._callbacks.push(handler); return this._updateSubscription(); } + /** + * Subscribe to receive event notifications for the given channel + */ subscribeToChannel(channel: string | number, handler: EventSubscriberCallback) { if (!this._channelCallbacks.get(channel)) { this._channelCallbacks.set(channel, []); @@ -259,6 +275,10 @@ export class EventSubscriber { return this._updateSubscription(); } + /** + * Unsubscribe from receiving all event notifications, does not effect the subscriptions of other handlers + * Will throw an error if your handler is not subscribed to all channels, does not accept anonymous functions + */ unsubscribe(handler: EventSubscriberCallback) { let index = this._callbacks.lastIndexOf(handler); if (index === -1) { @@ -269,6 +289,10 @@ export class EventSubscriber { return this._updateSubscription(); } + /** + * Unsubscribe from receiving event notifications for a given channel, does not effect the subscriptions of other handlers + * Will throw an error if your handler is not subscribed to the channel, does not accept anonymous functions + */ unsubscribeFromChannel(channel: string | number, handler: EventSubscriberCallback) { const channelCallbacks = this._channelCallbacks.get(channel); if (!channelCallbacks) { @@ -288,6 +312,9 @@ export class EventSubscriber { return this._updateSubscription(); } + /** + * Update the subscription with the controller based on current handler counts + */ async _updateSubscription() { if (!this.control || !(this.control.connector as WebSocketClientConnector).connected) return; const entry = Link._eventsByClass.get(this.event)!; diff --git a/test/lib/subscriptions.js b/test/lib/subscriptions.js index 1efd054a..acc4dd86 100644 --- a/test/lib/subscriptions.js +++ b/test/lib/subscriptions.js @@ -93,40 +93,31 @@ describe("lib/subscriptions", function() { describe("class SubscriptionRequest", function() { describe("permission()", function() { - 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, new lib.MessageRequest( + function newSubscriptionRequestMessage(request) { + return new lib.MessageRequest( 0, addr({ controlId: 0 }), addr("controller"), lib.SubscriptionRequest.name, - JSON.stringify(request) - )); + 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, new lib.MessageRequest( - 0, - addr({ controlId: 0 }), - addr("controller"), - lib.SubscriptionRequest.name, - request.toJSON() - )); + 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, new lib.MessageRequest( - 0, - addr({ controlId: 0 }), - addr("controller"), - lib.SubscriptionRequest.name, - request.toJSON() - )); + lib.SubscriptionRequest.permission(mockUser, newSubscriptionRequestMessage(request)); assert.equal(mockUser.lastPermissionCheck, "FunctionPermission"); }); }); @@ -176,11 +167,9 @@ describe("lib/subscriptions", function() { dst: addr("controller"), }]; - function awaitMessages(ids) { - return Promise.all(ids.map(function(id) { - const link = mockController.wsServer.controlConnections.get(id); - return events.once(link.connector, "send"); - })); + function onceConnectorSend(id) { + const link = mockController.wsServer.controlConnections.get(id); + return events.once(link.connector, "send"); } function assertLastEvent(connectorId, Event) { @@ -250,9 +239,8 @@ describe("lib/subscriptions", function() { ); }); it("should notify all links who subscribed to all channels", async function() { - const messages = awaitMessages([0, 1]); subscriptions.broadcast(new RegisteredEvent()); - await messages; + await Promise.all([onceConnectorSend(0), onceConnectorSend(1)]); assertLastEvent(0, RegisteredEvent); // RegisteredEvent: All assertLastEvent(1, RegisteredEvent); // RegisteredEvent: All assertNoEvent(2); // ChannelEvent: All @@ -260,13 +248,12 @@ describe("lib/subscriptions", function() { assertNoEvent(4); // ChannelEvent: channelTwo }); it("should not notify a link who unsubscribed from all all channels", async function() { - const messages = awaitMessages([1]); const connectorData = connectorSetupDate[0]; const request = new lib.SubscriptionRequest(RegisteredEvent.name, false); await subscriptions._handleRequest(request, connectorData.src, connectorData.dst); mockController.wsServer.controlConnections.get(0).connector.sentMessages.pop(); // Response to request subscriptions.broadcast(new RegisteredEvent()); - await messages; + await onceConnectorSend(1); assertNoEvent(0); assertLastEvent(1, RegisteredEvent); assertNoEvent(2); @@ -274,9 +261,8 @@ describe("lib/subscriptions", function() { assertNoEvent(4); }); it("should notify all links who subscribed the specific channel", async function() { - const messages = awaitMessages([2, 3]); subscriptions.broadcast(new ChannelEvent()); - await messages; + await Promise.all([onceConnectorSend(2), onceConnectorSend(3)]); assertNoEvent(0); assertNoEvent(1); assertLastEvent(2, ChannelEvent); @@ -284,13 +270,12 @@ describe("lib/subscriptions", function() { assertNoEvent(4); }); it("should not notify a link who unsubscribed from the specific channel", async function() { - const messages = awaitMessages([2]); const connectorData = connectorSetupDate[3]; const request = new lib.SubscriptionRequest(ChannelEvent.name, false, []); await subscriptions._handleRequest(request, connectorData.src, connectorData.dst); mockController.wsServer.controlConnections.get(3).connector.sentMessages.pop(); // Response to request subscriptions.broadcast(new ChannelEvent()); - await messages; + await onceConnectorSend(2); assertNoEvent(0); assertNoEvent(1); assertLastEvent(2, ChannelEvent); @@ -298,11 +283,10 @@ describe("lib/subscriptions", function() { assertNoEvent(4); }); it("should not notify links which are closed or closing", async function() { - const messages = awaitMessages([1]); const link = mockController.wsServer.controlConnections.get(0); link.connector.closing = true; subscriptions.broadcast(new RegisteredEvent()); - await messages; + await onceConnectorSend(1); assertNoEvent(0); assertLastEvent(1, RegisteredEvent); assertNoEvent(2); @@ -320,35 +304,31 @@ describe("lib/subscriptions", function() { it("should not accept subscriptions to unregistered events", function() { const request = new lib.SubscriptionRequest(RegisteredEvent.name, true); request.eventName = UnregisteredEvent.name; - const connectorData = connectorSetupDate[0]; assert.rejects( - subscriptions._handleRequest(request, connectorData.src, connectorData.dst), + subscriptions._handleRequest(request, addr({ controlId: 0 }), addr("controller")), new Error(`Unregistered Event class ${UnregisteredEvent.name}`) ); }); it("should not accept subscriptions to events not handled by the class", function() { const request = new lib.SubscriptionRequest(RegisteredEvent.name, true); request.eventName = StringPermissionEvent.name; - const connectorData = connectorSetupDate[0]; assert.rejects( - subscriptions._handleRequest(request, connectorData.src, connectorData.dst), + subscriptions._handleRequest(request, addr({ controlId: 0 }), addr("controller")), new Error(`Event ${StringPermissionEvent.eventName} is not a registered as subscribable`) ); }); it("should accept a subscription to all channels", async function() { - const messages = awaitMessages([0]); const request = new lib.SubscriptionRequest(RegisteredEvent.name, true); await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst); subscriptions.broadcast(new RegisteredEvent()); - await messages; + await onceConnectorSend(0); assertLastEvent(0, RegisteredEvent); }); it("should accept a unsubscription from all channels", async function() { - const messages = awaitMessages([0]); const request = new lib.SubscriptionRequest(RegisteredEvent.name, true); await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst); subscriptions.broadcast(new RegisteredEvent()); - await messages; + await onceConnectorSend(0); assertLastEvent(0, RegisteredEvent); const link = mockController.wsServer.controlConnections.get(0); @@ -360,19 +340,17 @@ describe("lib/subscriptions", function() { assertNoEvent(0); }); it("should accept a subscription to a specific channel", async function() { - const messages = awaitMessages([0]); const request = new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelOne"]); await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst); subscriptions.broadcast(new ChannelEvent()); - await messages; + await onceConnectorSend(0); assertLastEvent(0, ChannelEvent); }); it("should accept a unsubscription from a specific channel", async function() { - const messages = awaitMessages([0]); const request = new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelOne"]); await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst); subscriptions.broadcast(new ChannelEvent()); - await messages; + await onceConnectorSend(0); assertLastEvent(0, ChannelEvent); const link = mockController.wsServer.controlConnections.get(0); @@ -384,7 +362,6 @@ describe("lib/subscriptions", function() { assertNoEvent(0); }); it("should accept a respond with an event replay when returned by the handler", async function() { - const messages = awaitMessages([0]); subscriptions.handle(StringPermissionEvent, undefined, async function() { return new StringPermissionEvent(); }); @@ -392,7 +369,7 @@ describe("lib/subscriptions", function() { const request = new lib.SubscriptionRequest(StringPermissionEvent.name, true); await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst); subscriptions.broadcast(new StringPermissionEvent()); - await messages; + await onceConnectorSend(0); assertLastEvent(0, StringPermissionEvent); const link = mockController.wsServer.controlConnections.get(0); @@ -446,10 +423,7 @@ describe("lib/subscriptions", function() { describe("connectControl()", function() { it("should handle the provided event", function() { - const control = new MockControl(new MockConnector( - addr({ controlId: 0 }), - addr("controller"), - )); + const control = new MockControl(new MockConnector(addr({ controlId: 0 }), addr("controller"))); registeredEvent.connectControl(control); assert.equal(control._eventHandlers.has(RegisteredEvent), true); }); @@ -559,17 +533,16 @@ describe("lib/subscriptions", function() { it("should call handlers when a replay event is returned", async function() { let calledWith = null; const event = new RegisteredEvent(); - const message = events.once(mockControl.connector, "send"); const request = registeredEvent.subscribe(function(value) { calledWith = value; }); - await message; + await events.once(mockControl.connector, "send"); - const responseMessage = new lib.MessageResponse( - 1, + mockControl.connector.emit("message", new lib.MessageResponse( + 0, mockControl.connector.dst, new lib.Address(lib.Address.control, 0, mockControl._nextRequestId - 1), new lib.SubscriptionResponse(event).toJSON() - ); - mockControl.connector.emit("message", responseMessage); + )); + await request; assert.deepEqual(calledWith, event); }); From 4986a06cdfbaf6b402446f5bde557ff0175c8898 Mon Sep 17 00:00:00 2001 From: Cooldude2606 <25043174+Cooldude2606@users.noreply.github.com> Date: Thu, 7 Sep 2023 21:58:10 +0100 Subject: [PATCH 09/15] Fixed web_ui websocket type error --- packages/lib/src/subscriptions.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index 9d0b0ed2..7e6ef1d0 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -213,7 +213,7 @@ export class EventSubscriber { constructor( private event: EventClass | ChannelEventClass, - private prehandler?: (event: Event) => V, + private prehandler?: (event: T) => V, control?: Link ) { const entry = Link._eventsByClass.get(this.event); @@ -231,7 +231,7 @@ export class EventSubscriber { async _handle(response: Event | ChannelEvent) { this.lastResponse = response; this.lastResponseTime = Date.now(); - const value = this.prehandler ? this.prehandler(response) : response as any; + const value = this.prehandler ? this.prehandler(response as T) : response as any; for (let callback of this._callbacks) { callback(value); } From 11d820b085c712adbcfc2772d2e5dc250b213531 Mon Sep 17 00:00:00 2001 From: Cooldude2606 <25043174+Cooldude2606@users.noreply.github.com> Date: Thu, 7 Sep 2023 22:31:09 +0100 Subject: [PATCH 10/15] Increase patch coverage --- test/lib/subscriptions.js | 36 +++++++++++++++++++++++++++++++++++- 1 file changed, 35 insertions(+), 1 deletion(-) diff --git a/test/lib/subscriptions.js b/test/lib/subscriptions.js index acc4dd86..bdc1b8a2 100644 --- a/test/lib/subscriptions.js +++ b/test/lib/subscriptions.js @@ -83,6 +83,13 @@ describe("lib/subscriptions", function() { assert.deepEqual(reconstructed, response); }); + it("should be throw when given an unregistered event in a json", function() { + assert.throws( + () => lib.SubscriptionResponse.fromJSON([UnregisteredEvent.name, {}]), + new Error(`Unregistered Event class ${UnregisteredEvent.name}`) + ); + }); + it("should be throw when given an unregistered event", function() { assert.throws( () => new lib.SubscriptionResponse(new UnregisteredEvent()), @@ -498,13 +505,40 @@ describe("lib/subscriptions", function() { channelEvent.unsubscribeFromChannel("channelOne", callback); channelEvent._handle(event); assert.equal(calledWith, null); + + let calledWithTwo = null; + function callbackTwo(value) { calledWithTwo = value; } + channelEvent.subscribeToChannel("channelOne", callback); + channelEvent.subscribeToChannel("channelOne", callbackTwo); + channelEvent._handle(event); + assert.equal(calledWith, event); + assert.equal(calledWithTwo, event); + + calledWith = null; + calledWithTwo = null; + event = new ChannelEvent(); + channelEvent.unsubscribeFromChannel("channelOne", callback); + channelEvent._handle(event); + assert.equal(calledWith, null); + assert.equal(calledWithTwo, event); }); it("should throw an error if the handler was not subscribed", function() { - function callback(value) { calledWith = value; } + function callback() {} assert.throws( () => channelEvent.unsubscribeFromChannel("channelOne", callback), new Error("handler is not registered") ); + + let calledWith = null; + function callbackTwo(value) { calledWith = value; } + channelEvent.subscribeToChannel("channelOne", callbackTwo); + assert.throws( + () => channelEvent.unsubscribeFromChannel("channelOne", callback), + new Error("handler is not registered") + ); + const event = new ChannelEvent(); + channelEvent._handle(event); + assert.equal(calledWith, event); }); }); From b15b1a60074825a5087dc1c9c5fd99f6c0af3e64 Mon Sep 17 00:00:00 2001 From: Cooldude2606 <25043174+Cooldude2606@users.noreply.github.com> Date: Fri, 6 Oct 2023 18:30:15 +0100 Subject: [PATCH 11/15] Fixed bugs found during pre-review --- packages/lib/browser.js | 1 + packages/lib/src/subscriptions.ts | 42 +++++++++++++++------------ packages/web_ui/src/util/websocket.ts | 6 ++-- 3 files changed, 27 insertions(+), 22 deletions(-) diff --git a/packages/lib/browser.js b/packages/lib/browser.js index 1f2d0428..ad4ccaf1 100644 --- a/packages/lib/browser.js +++ b/packages/lib/browser.js @@ -9,6 +9,7 @@ export * from "./dist/src/logging"; export * from "./dist/src/plugin"; export * from "./dist/src/schema"; export * from "./dist/src/users"; +export * from "./dist/src/subscriptions"; export { default as ExponentialBackoff } from "./dist/src/ExponentialBackoff"; export { default as PlayerStats } from "./dist/src/PlayerStats"; diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index 7e6ef1d0..082a5a67 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -118,7 +118,7 @@ export class SubscriptionRequest { export class SubscriptionController { _events = new Map, - subscriptions: Map }>, + subscriptions: Map, onceClose: () => void }>, }>(); constructor( @@ -186,14 +186,18 @@ export class SubscriptionController { const eventReplay = eventData.subscriptionUpdate ? await eventData.subscriptionUpdate(event, src, dst) : null; const link: Link = this.controller.wsServer.controlConnections.get(src.id); if (event.allChannels === false && event.channels.length === 0) { - eventData.subscriptions.delete(link); - } else { - if (!eventData.subscriptions.has(link)) { - link.connector.once("close", () => { - eventData.subscriptions.delete(link); - }) + let onceClose = eventData.subscriptions.get(link)?.onceClose; + if (onceClose) { + link.connector.off("close", onceClose); + eventData.subscriptions.delete(link); } - eventData.subscriptions.set(link, { all: event.allChannels, channels: event.channels }); + } else { + let onceClose = eventData.subscriptions.get(link)?.onceClose; + if (!onceClose) { + onceClose = () => eventData.subscriptions.delete(link); + link.connector.once("close", onceClose); + } + eventData.subscriptions.set(link, { all: event.allChannels, channels: event.channels, onceClose: onceClose }); } return new SubscriptionResponse(eventReplay); } @@ -249,51 +253,51 @@ export class EventSubscriber { /** * If a control link was not connected at creation, it can be connected here (normally during onControllerConnectionEvent) */ - connectControl(control: Link) { - if (this.control === control) return Promise.resolve(); + async connectControl(control: Link) { + if (this.control === control) return; this.control = control; this.control.handle(this.event, this._handle.bind(this)); - return this._updateSubscription(); + await this._updateSubscription(); } /** * Subscribe to receive all event notifications */ - subscribe(handler: EventSubscriberCallback) { + async subscribe(handler: EventSubscriberCallback) { this._callbacks.push(handler); - return this._updateSubscription(); + await this._updateSubscription(); } /** * Subscribe to receive event notifications for the given channel */ - subscribeToChannel(channel: string | number, handler: EventSubscriberCallback) { + async subscribeToChannel(channel: string | number, handler: EventSubscriberCallback) { if (!this._channelCallbacks.get(channel)) { this._channelCallbacks.set(channel, []); } this._channelCallbacks.get(channel)!.push(handler); - return this._updateSubscription(); + await this._updateSubscription(); } /** * Unsubscribe from receiving all event notifications, does not effect the subscriptions of other handlers * Will throw an error if your handler is not subscribed to all channels, does not accept anonymous functions */ - unsubscribe(handler: EventSubscriberCallback) { + async unsubscribe(handler: EventSubscriberCallback) { let index = this._callbacks.lastIndexOf(handler); if (index === -1) { throw new Error("handler is not registered"); } this._callbacks.splice(index, 1); - return this._updateSubscription(); + await this._updateSubscription(); } /** * Unsubscribe from receiving event notifications for a given channel, does not effect the subscriptions of other handlers * Will throw an error if your handler is not subscribed to the channel, does not accept anonymous functions */ - unsubscribeFromChannel(channel: string | number, handler: EventSubscriberCallback) { + async unsubscribeFromChannel(channel: string | number, handler: EventSubscriberCallback) { const channelCallbacks = this._channelCallbacks.get(channel); if (!channelCallbacks) { throw new Error("handler is not registered"); @@ -309,7 +313,7 @@ export class EventSubscriber { } else { channelCallbacks.splice(index, 1); } - return this._updateSubscription(); + await this._updateSubscription(); } /** diff --git a/packages/web_ui/src/util/websocket.ts b/packages/web_ui/src/util/websocket.ts index f503e6e0..31368b8b 100644 --- a/packages/web_ui/src/util/websocket.ts +++ b/packages/web_ui/src/util/websocket.ts @@ -59,9 +59,9 @@ export class Control extends lib.Link { hostUpdate = new lib.EventSubscriber(lib.HostUpdateEvent, event => event.update); instanceUpdate = new lib.EventSubscriber(lib.InstanceDetailsUpdateEvent, event => event.details); saveListUpdate = new lib.EventSubscriber(lib.InstanceSaveListUpdateEvent); - modPackUpdate = new lib.EventSubscriber(lib.InstanceSaveListUpdateEvent, event => event.modPack); - modUpdate = new lib.EventSubscriber(lib.InstanceSaveListUpdateEvent, event => event.mod); - userUpdate = new lib.EventSubscriber(lib.InstanceSaveListUpdateEvent, event => event.user); + modPackUpdate = new lib.EventSubscriber(lib.ModPackUpdateEvent, event => event.modPack); + modUpdate = new lib.EventSubscriber(lib.ModUpdateEvent, event => event.mod); + userUpdate = new lib.EventSubscriber(lib.UserUpdateEvent, event => event.user); declare connector: ControlConnector; From 23ba5d612a291bf6b69ee1e8c5392ffb65a0f562 Mon Sep 17 00:00:00 2001 From: Cooldude2606 <25043174+Cooldude2606@users.noreply.github.com> Date: Fri, 6 Oct 2023 18:56:27 +0100 Subject: [PATCH 12/15] Fixed failing tests for unsubscribe --- test/lib/subscriptions.js | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/test/lib/subscriptions.js b/test/lib/subscriptions.js index bdc1b8a2..5656717b 100644 --- a/test/lib/subscriptions.js +++ b/test/lib/subscriptions.js @@ -484,7 +484,7 @@ describe("lib/subscriptions", function() { }); it("should throw an error if the handler was not subscribed", function() { function callback(value) { calledWith = value; } - assert.throws( + assert.rejects( () => registeredEvent.unsubscribe(callback), new Error("handler is not registered") ); @@ -524,7 +524,7 @@ describe("lib/subscriptions", function() { }); it("should throw an error if the handler was not subscribed", function() { function callback() {} - assert.throws( + assert.rejects( () => channelEvent.unsubscribeFromChannel("channelOne", callback), new Error("handler is not registered") ); @@ -532,7 +532,7 @@ describe("lib/subscriptions", function() { let calledWith = null; function callbackTwo(value) { calledWith = value; } channelEvent.subscribeToChannel("channelOne", callbackTwo); - assert.throws( + assert.rejects( () => channelEvent.unsubscribeFromChannel("channelOne", callback), new Error("handler is not registered") ); From d84445baa14c7067540f6bb356b8e281faedb820 Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Thu, 12 Oct 2023 17:07:07 +0200 Subject: [PATCH 13/15] Fix line endings and indents --- packages/lib/src/subscriptions.ts | 674 +++++++++++++++--------------- 1 file changed, 337 insertions(+), 337 deletions(-) diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index 082a5a67..0b186504 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -1,337 +1,337 @@ -import { Type, Static } from "@sinclair/typebox"; -import { Link, Event, EventClass, RequestHandler, WebSocketClientConnector, WebSocketBaseConnector } from "./link"; -import { Address, MessageRequest } from "./data"; -import { User } from "./users"; - -export type SubscriptionRequestHandler = RequestHandler | null>; -export type EventSubscriberCallback = (value: T) => void; - -export type ChannelEvent = Event & { - get subscriptionChannel(): number | string; -} - -export type ChannelEventClass = EventClass & { - new(...args: any): ChannelEvent -} - -/** - * Response received by the subscriber after a request - * It can contain an eventReplay value if the event sender implements a subscription handler - * This replay will be send to the handlers as if an update had just occurred - */ -export class SubscriptionResponse { - constructor( - public readonly eventReplay: Event | null = null, - ) { - if (eventReplay && !Link._eventsByClass.has(eventReplay.constructor)) { - throw new Error(`Unregistered Event class ${eventReplay.constructor.name}`); - } - } - - static jsonSchema = Type.Union([ - Type.Tuple([ - Type.String(), - Type.Unknown(), - ]), - Type.Null() - ]) - - toJSON() { - if (this.eventReplay) { - const entry = Link._eventsByClass.get(this.eventReplay.constructor)!; - return [entry.name, this.eventReplay]; - } else { - return null; - } - } - - static fromJSON(json: Static): SubscriptionResponse { - if (json) { - const entry = Link._eventsByName.get(json[0]); - if (!entry) { - throw new Error(`Unregistered Event class ${json[0]}`); - } else { - return new SubscriptionResponse(entry.eventFromJSON(json[1])); - } - } else { - return new SubscriptionResponse(); - } - } -} - -/** - * 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 - * allChannels: false, channels: [] will unsubscribe the subscriber from all notifications - */ -export class SubscriptionRequest { - declare ["constructor"]: typeof SubscriptionRequest; - static type = "request" as const; - static src = ["control", "instance"] as const; - static dst = "controller" as const; - static Response = SubscriptionResponse; - static permission(user: User, message: MessageRequest) { - if (typeof message.data === "object" && message.data !== null) { - const data = message.data as Static; - const entry = Link._eventsByName.get(data[0]); - if (entry && entry.Event.permission) { - if (typeof entry.Event.permission === "string") { - user.checkPermission(entry.Event.permission); - } else { - entry.Event.permission(user, message); - } - } - } - } - - constructor( - public eventName: string, - public allChannels: boolean, - public channels: Array = [], - public lastRequestTime: number = 0, - ) { - if (!Link._eventsByName.has(eventName)) { - throw new Error(`Unregistered Event class ${eventName}`); - } - } - - static jsonSchema = Type.Tuple([ - Type.String(), - Type.Boolean(), - Type.Array(Type.Union([Type.String(), Type.Number()])), - Type.Number(), - ]) - - toJSON() { - return [this.eventName, this.allChannels, this.channels, this.lastRequestTime]; - } - - static fromJSON(json: Static): SubscriptionRequest { - return new this(...json); - } -} - -/** - * A class component to handle incoming subscription requests and offers a method to broadcast events to subscribers - * After creation, no other handler can be registered for SubscriptionRequest on the controller - */ -export class SubscriptionController { - _events = new Map, - subscriptions: Map, onceClose: () => void }>, - }>(); - - constructor( - private controller: any, // Controller | Link - ) { - this.controller.handle(SubscriptionRequest, this._handleRequest.bind(this)); - } - - /** - * 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 any client updates their subscription - */ - handle(Event: EventClass, subscriptionUpdate?: SubscriptionRequestHandler,): void; - handle( - Event: EventClass, - subscriptionUpdate?: SubscriptionRequestHandler, - ) { - const entry = Link._eventsByClass.get(Event); - if (!entry) { - throw new Error(`Unregistered Event class ${Event.name}`); - } - if (this._events.has(entry.name)) { - throw new Error(`Event ${entry.name} is already registered`); - } - this._events.set(entry.name, { - subscriptionUpdate: subscriptionUpdate, - subscriptions: new Map(), - }); - } - - /** - * Broadcast an event to all subscribers of that event, will be filtered by channel if a ChannelEvent is provided - */ - broadcast(event: Event | ChannelEvent): void; - broadcast(event: Event | ChannelEvent) { - const entry = Link._eventsByClass.get(event.constructor); - if (!entry) { - throw new Error(`Unregistered Event class ${event.constructor.name}`); - } - const eventData = this._events.get(entry.name); - if (!eventData) { - throw new Error(`Event ${entry.name} is not a registered as subscribable`); - } - const channel = "subscriptionChannel" in event ? event.subscriptionChannel : null; - for (let [link, subscription] of eventData.subscriptions) { - if ((link.connector as WebSocketBaseConnector).closing) { - eventData.subscriptions.delete(link); - } else if ((subscription.all || (channel && subscription.channels.includes(channel)))) { - link.send(event); - } - } - } - - /** - * Handle incoming event subscription requests - */ - async _handleRequest(event: SubscriptionRequest, src: Address, dst: Address) { - if (!Link._eventsByName.has(event.eventName)) { - throw new Error(`Event ${event.eventName} is not a registered event`); - } - const eventData = this._events.get(event.eventName); - if (!eventData) { - throw new Error(`Event ${event.eventName} is not a registered as subscribable`); - } - const eventReplay = eventData.subscriptionUpdate ? await eventData.subscriptionUpdate(event, src, dst) : null; - const link: Link = this.controller.wsServer.controlConnections.get(src.id); - if (event.allChannels === false && event.channels.length === 0) { - let onceClose = eventData.subscriptions.get(link)?.onceClose; - if (onceClose) { - link.connector.off("close", onceClose); - eventData.subscriptions.delete(link); - } - } else { - let onceClose = eventData.subscriptions.get(link)?.onceClose; - if (!onceClose) { - onceClose = () => eventData.subscriptions.delete(link); - link.connector.once("close", onceClose); - } - eventData.subscriptions.set(link, { all: event.allChannels, channels: event.channels, onceClose: onceClose }); - } - return new SubscriptionResponse(eventReplay); - } -} - -/** - * A class component to allow subscribing and unsubscribing to/from an event and its channels - * Channels are an optional feature which can be implemented to only get notifications of reliant updates - * Multiple handlers can be subscribed at the same time, on the same or different channels - */ -export class EventSubscriber { - _callbacks = new Array>() - _channelCallbacks = new Map>>() - lastResponse: Event | null = null; - lastResponseTime = 0; - control?: Link - - constructor( - private event: EventClass | ChannelEventClass, - private prehandler?: (event: T) => V, - control?: Link - ) { - const entry = Link._eventsByClass.get(this.event); - if (!entry) { - throw new Error(`Unregistered Event class ${this.event.name}`); - } - if (control) { - this.connectControl(control); - } - } - - /** - * Handle incoming events and distribute it to the correct callbacks - */ - async _handle(response: Event | ChannelEvent) { - this.lastResponse = response; - this.lastResponseTime = Date.now(); - const value = this.prehandler ? this.prehandler(response as T) : response as any; - for (let callback of this._callbacks) { - callback(value); - } - const channel = "subscriptionChannel" in response ? response.subscriptionChannel : null; - if (channel) { - const callbacks = this._channelCallbacks.get(channel); - if (callbacks) { - for (let callback of callbacks) { - callback(value); - } - } - } - } - - /** - * If a control link was not connected at creation, it can be connected here (normally during onControllerConnectionEvent) - */ - async connectControl(control: Link) { - if (this.control === control) return; - this.control = control; - this.control.handle(this.event, this._handle.bind(this)); - await this._updateSubscription(); - } - - /** - * Subscribe to receive all event notifications - */ - async subscribe(handler: EventSubscriberCallback) { - this._callbacks.push(handler); - await this._updateSubscription(); - } - - /** - * Subscribe to receive event notifications for the given channel - */ - async subscribeToChannel(channel: string | number, handler: EventSubscriberCallback) { - if (!this._channelCallbacks.get(channel)) { - this._channelCallbacks.set(channel, []); - } - this._channelCallbacks.get(channel)!.push(handler); - await this._updateSubscription(); - } - - /** - * Unsubscribe from receiving all event notifications, does not effect the subscriptions of other handlers - * Will throw an error if your handler is not subscribed to all channels, does not accept anonymous functions - */ - async unsubscribe(handler: EventSubscriberCallback) { - let index = this._callbacks.lastIndexOf(handler); - if (index === -1) { - throw new Error("handler is not registered"); - } - - this._callbacks.splice(index, 1); - await this._updateSubscription(); - } - - /** - * Unsubscribe from receiving event notifications for a given channel, does not effect the subscriptions of other handlers - * Will throw an error if your handler is not subscribed to the channel, does not accept anonymous functions - */ - async unsubscribeFromChannel(channel: string | number, handler: EventSubscriberCallback) { - const channelCallbacks = this._channelCallbacks.get(channel); - if (!channelCallbacks) { - throw new Error("handler is not registered"); - } - - let index = channelCallbacks.lastIndexOf(handler); - if (index === -1) { - throw new Error("handler is not registered"); - } - - if (channelCallbacks.length === 1) { - this._channelCallbacks.delete(channel); - } else { - channelCallbacks.splice(index, 1); - } - await this._updateSubscription(); - } - - /** - * Update the subscription with the controller based on current handler counts - */ - async _updateSubscription() { - if (!this.control || !(this.control.connector as WebSocketClientConnector).connected) return; - const entry = Link._eventsByClass.get(this.event)!; - - const response = await this.control.send(new SubscriptionRequest( - entry.name, - this._callbacks.length > 0, - [...this._channelCallbacks.keys()], - this.lastResponseTime - )); - - if (response.eventReplay) { - await this._handle(response.eventReplay); - } - } -} \ No newline at end of file +import { Type, Static } from "@sinclair/typebox"; +import { Link, Event, EventClass, RequestHandler, WebSocketClientConnector, WebSocketBaseConnector } from "./link"; +import { Address, MessageRequest } from "./data"; +import { User } from "./users"; + +export type SubscriptionRequestHandler = RequestHandler | null>; +export type EventSubscriberCallback = (value: T) => void; + +export type ChannelEvent = Event & { + get subscriptionChannel(): number | string; +} + +export type ChannelEventClass = EventClass & { + new(...args: any): ChannelEvent +} + +/** + * Response received by the subscriber after a request + * It can contain an eventReplay value if the event sender implements a subscription handler + * This replay will be send to the handlers as if an update had just occurred + */ +export class SubscriptionResponse { + constructor( + public readonly eventReplay: Event | null = null, + ) { + if (eventReplay && !Link._eventsByClass.has(eventReplay.constructor)) { + throw new Error(`Unregistered Event class ${eventReplay.constructor.name}`); + } + } + + static jsonSchema = Type.Union([ + Type.Tuple([ + Type.String(), + Type.Unknown(), + ]), + Type.Null() + ]) + + toJSON() { + if (this.eventReplay) { + const entry = Link._eventsByClass.get(this.eventReplay.constructor)!; + return [entry.name, this.eventReplay]; + } else { + return null; + } + } + + static fromJSON(json: Static): SubscriptionResponse { + if (json) { + const entry = Link._eventsByName.get(json[0]); + if (!entry) { + throw new Error(`Unregistered Event class ${json[0]}`); + } else { + return new SubscriptionResponse(entry.eventFromJSON(json[1])); + } + } else { + return new SubscriptionResponse(); + } + } +} + +/** + * 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 + * allChannels: false, channels: [] will unsubscribe the subscriber from all notifications + */ +export class SubscriptionRequest { + declare ["constructor"]: typeof SubscriptionRequest; + static type = "request" as const; + static src = ["control", "instance"] as const; + static dst = "controller" as const; + static Response = SubscriptionResponse; + static permission(user: User, message: MessageRequest) { + if (typeof message.data === "object" && message.data !== null) { + const data = message.data as Static; + const entry = Link._eventsByName.get(data[0]); + if (entry && entry.Event.permission) { + if (typeof entry.Event.permission === "string") { + user.checkPermission(entry.Event.permission); + } else { + entry.Event.permission(user, message); + } + } + } + } + + constructor( + public eventName: string, + public allChannels: boolean, + public channels: Array = [], + public lastRequestTime: number = 0, + ) { + if (!Link._eventsByName.has(eventName)) { + throw new Error(`Unregistered Event class ${eventName}`); + } + } + + static jsonSchema = Type.Tuple([ + Type.String(), + Type.Boolean(), + Type.Array(Type.Union([Type.String(), Type.Number()])), + Type.Number(), + ]) + + toJSON() { + return [this.eventName, this.allChannels, this.channels, this.lastRequestTime]; + } + + static fromJSON(json: Static): SubscriptionRequest { + return new this(...json); + } +} + +/** + * A class component to handle incoming subscription requests and offers a method to broadcast events to subscribers + * After creation, no other handler can be registered for SubscriptionRequest on the controller + */ +export class SubscriptionController { + _events = new Map, + subscriptions: Map, onceClose: () => void }>, + }>(); + + constructor( + private controller: any, // Controller | Link + ) { + this.controller.handle(SubscriptionRequest, this._handleRequest.bind(this)); + } + + /** + * 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 any client updates their subscription + */ + handle(Event: EventClass, subscriptionUpdate?: SubscriptionRequestHandler,): void; + handle( + Event: EventClass, + subscriptionUpdate?: SubscriptionRequestHandler, + ) { + const entry = Link._eventsByClass.get(Event); + if (!entry) { + throw new Error(`Unregistered Event class ${Event.name}`); + } + if (this._events.has(entry.name)) { + throw new Error(`Event ${entry.name} is already registered`); + } + this._events.set(entry.name, { + subscriptionUpdate: subscriptionUpdate, + subscriptions: new Map(), + }); + } + + /** + * Broadcast an event to all subscribers of that event, will be filtered by channel if a ChannelEvent is provided + */ + broadcast(event: Event | ChannelEvent): void; + broadcast(event: Event | ChannelEvent) { + const entry = Link._eventsByClass.get(event.constructor); + if (!entry) { + throw new Error(`Unregistered Event class ${event.constructor.name}`); + } + const eventData = this._events.get(entry.name); + if (!eventData) { + throw new Error(`Event ${entry.name} is not a registered as subscribable`); + } + const channel = "subscriptionChannel" in event ? event.subscriptionChannel : null; + for (let [link, subscription] of eventData.subscriptions) { + if ((link.connector as WebSocketBaseConnector).closing) { + eventData.subscriptions.delete(link); + } else if ((subscription.all || (channel && subscription.channels.includes(channel)))) { + link.send(event); + } + } + } + + /** + * Handle incoming event subscription requests + */ + async _handleRequest(event: SubscriptionRequest, src: Address, dst: Address) { + if (!Link._eventsByName.has(event.eventName)) { + throw new Error(`Event ${event.eventName} is not a registered event`); + } + const eventData = this._events.get(event.eventName); + if (!eventData) { + throw new Error(`Event ${event.eventName} is not a registered as subscribable`); + } + const eventReplay = eventData.subscriptionUpdate ? await eventData.subscriptionUpdate(event, src, dst) : null; + const link: Link = this.controller.wsServer.controlConnections.get(src.id); + if (event.allChannels === false && event.channels.length === 0) { + let onceClose = eventData.subscriptions.get(link)?.onceClose; + if (onceClose) { + link.connector.off("close", onceClose); + eventData.subscriptions.delete(link); + } + } else { + let onceClose = eventData.subscriptions.get(link)?.onceClose; + if (!onceClose) { + onceClose = () => eventData.subscriptions.delete(link); + link.connector.once("close", onceClose); + } + eventData.subscriptions.set(link, { all: event.allChannels, channels: event.channels, onceClose: onceClose }); + } + return new SubscriptionResponse(eventReplay); + } +} + +/** + * A class component to allow subscribing and unsubscribing to/from an event and its channels + * Channels are an optional feature which can be implemented to only get notifications of reliant updates + * Multiple handlers can be subscribed at the same time, on the same or different channels + */ +export class EventSubscriber { + _callbacks = new Array>() + _channelCallbacks = new Map>>() + lastResponse: Event | null = null; + lastResponseTime = 0; + control?: Link + + constructor( + private event: EventClass | ChannelEventClass, + private prehandler?: (event: T) => V, + control?: Link + ) { + const entry = Link._eventsByClass.get(this.event); + if (!entry) { + throw new Error(`Unregistered Event class ${this.event.name}`); + } + if (control) { + this.connectControl(control); + } + } + + /** + * Handle incoming events and distribute it to the correct callbacks + */ + async _handle(response: Event | ChannelEvent) { + this.lastResponse = response; + this.lastResponseTime = Date.now(); + const value = this.prehandler ? this.prehandler(response as T) : response as any; + for (let callback of this._callbacks) { + callback(value); + } + const channel = "subscriptionChannel" in response ? response.subscriptionChannel : null; + if (channel) { + const callbacks = this._channelCallbacks.get(channel); + if (callbacks) { + for (let callback of callbacks) { + callback(value); + } + } + } + } + + /** + * If a control link was not connected at creation, it can be connected here (normally during onControllerConnectionEvent) + */ + async connectControl(control: Link) { + if (this.control === control) return; + this.control = control; + this.control.handle(this.event, this._handle.bind(this)); + await this._updateSubscription(); + } + + /** + * Subscribe to receive all event notifications + */ + async subscribe(handler: EventSubscriberCallback) { + this._callbacks.push(handler); + await this._updateSubscription(); + } + + /** + * Subscribe to receive event notifications for the given channel + */ + async subscribeToChannel(channel: string | number, handler: EventSubscriberCallback) { + if (!this._channelCallbacks.get(channel)) { + this._channelCallbacks.set(channel, []); + } + this._channelCallbacks.get(channel)!.push(handler); + await this._updateSubscription(); + } + + /** + * Unsubscribe from receiving all event notifications, does not effect the subscriptions of other handlers + * Will throw an error if your handler is not subscribed to all channels, does not accept anonymous functions + */ + async unsubscribe(handler: EventSubscriberCallback) { + let index = this._callbacks.lastIndexOf(handler); + if (index === -1) { + throw new Error("handler is not registered"); + } + + this._callbacks.splice(index, 1); + await this._updateSubscription(); + } + + /** + * Unsubscribe from receiving event notifications for a given channel, does not effect the subscriptions of other handlers + * Will throw an error if your handler is not subscribed to the channel, does not accept anonymous functions + */ + async unsubscribeFromChannel(channel: string | number, handler: EventSubscriberCallback) { + const channelCallbacks = this._channelCallbacks.get(channel); + if (!channelCallbacks) { + throw new Error("handler is not registered"); + } + + let index = channelCallbacks.lastIndexOf(handler); + if (index === -1) { + throw new Error("handler is not registered"); + } + + if (channelCallbacks.length === 1) { + this._channelCallbacks.delete(channel); + } else { + channelCallbacks.splice(index, 1); + } + await this._updateSubscription(); + } + + /** + * Update the subscription with the controller based on current handler counts + */ + async _updateSubscription() { + if (!this.control || !(this.control.connector as WebSocketClientConnector).connected) return; + const entry = Link._eventsByClass.get(this.event)!; + + const response = await this.control.send(new SubscriptionRequest( + entry.name, + this._callbacks.length > 0, + [...this._channelCallbacks.keys()], + this.lastResponseTime + )); + + if (response.eventReplay) { + await this._handle(response.eventReplay); + } + } +} From d58602f1a15e042999d015d694a31923576cf122 Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Thu, 12 Oct 2023 17:21:56 +0200 Subject: [PATCH 14/15] Fix merge conflicts --- packages/controller/src/ControlConnection.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/packages/controller/src/ControlConnection.ts b/packages/controller/src/ControlConnection.ts index b02e6f03..85a26170 100644 --- a/packages/controller/src/ControlConnection.ts +++ b/packages/controller/src/ControlConnection.ts @@ -1,4 +1,5 @@ import type WsServerConnector from "./WsServerConnector"; +import type { HostInfo } from "./HostConnection"; import events from "events"; import fs from "fs-extra"; From 54cb483a315b853ee530781af51cb58c3a7c78d0 Mon Sep 17 00:00:00 2001 From: Cooldude2606 <25043174+Cooldude2606@users.noreply.github.com> Date: Fri, 13 Oct 2023 11:21:55 +0100 Subject: [PATCH 15/15] Convert overloads to generic function Co-authored-by: Hornwitser --- packages/lib/src/subscriptions.ts | 9 ++------- 1 file changed, 2 insertions(+), 7 deletions(-) diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index 0b186504..ae346b65 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -131,11 +131,7 @@ export class SubscriptionController { * 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 any client updates their subscription */ - handle(Event: EventClass, subscriptionUpdate?: SubscriptionRequestHandler,): void; - handle( - Event: EventClass, - subscriptionUpdate?: SubscriptionRequestHandler, - ) { + handle(Event: EventClass, subscriptionUpdate?: SubscriptionRequestHandler) { const entry = Link._eventsByClass.get(Event); if (!entry) { throw new Error(`Unregistered Event class ${Event.name}`); @@ -152,8 +148,7 @@ export class SubscriptionController { /** * Broadcast an event to all subscribers of that event, will be filtered by channel if a ChannelEvent is provided */ - broadcast(event: Event | ChannelEvent): void; - broadcast(event: Event | ChannelEvent) { + broadcast(event: Event | ChannelEvent) { const entry = Link._eventsByClass.get(event.constructor); if (!entry) { throw new Error(`Unregistered Event class ${event.constructor.name}`);