diff --git a/packages/controller/src/ControlConnection.ts b/packages/controller/src/ControlConnection.ts index 487c4c54..3ace64f8 100644 --- a/packages/controller/src/ControlConnection.ts +++ b/packages/controller/src/ControlConnection.ts @@ -11,8 +11,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 +28,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, @@ -89,33 +81,28 @@ export default class ControlConnection extends BaseConnection { this.handle(lib.ControllerConfigSetPropRequest, this.handleControllerConfigSetPropRequest.bind(this)); this.handle(lib.HostRevokeTokensRequest, this.handleHostRevokeTokensRequest.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)); @@ -128,7 +115,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)); @@ -234,22 +220,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) { @@ -322,38 +292,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"); @@ -396,19 +334,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); @@ -506,10 +431,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) { @@ -542,15 +463,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`; @@ -654,10 +566,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`; @@ -690,15 +598,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) { @@ -850,32 +749,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 b6493704..e355b836 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); + 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(); @@ -576,13 +588,7 @@ export default class Controller { host.token_valid_after, ); - for (let controlConnection of this.wsServer.controlConnections.values()) { - if (controlConnection.connector.closing) { - continue; - } - - controlConnection.hostUpdated(host, update); - } + this.subscriptions.broadcast(new lib.HostUpdateEvent(update)); } /** @@ -762,56 +768,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/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/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 8f34ae35..5389a7c2 100644 --- a/packages/lib/src/data/messages_host.ts +++ b/packages/lib/src/data/messages_host.ts @@ -36,34 +36,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, @@ -76,6 +54,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 42376ae6..32cdab2b 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, @@ -108,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 { @@ -299,33 +282,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, @@ -340,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 0c4fe7ce..d9225205 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, @@ -302,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 { @@ -309,6 +270,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, @@ -321,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 210254dd..be7c1c9d 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, @@ -472,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/link/messages.ts b/packages/lib/src/link/messages.ts index 3c3400d6..0f21bbd0 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, @@ -38,7 +40,6 @@ export const dataClasses: (RequestClass | EventClass) instance.InstanceDetailsGetRequest, instance.InstanceDetailsListRequest, - instance.InstanceDetailsSetSubscriptionsRequest, instance.InstanceDetailsUpdateEvent, instance.InstanceCreateRequest, instance.InstanceConfigGetRequest, @@ -48,7 +49,6 @@ export const dataClasses: (RequestClass | EventClass) instance.InstanceMetricsRequest, instance.InstanceStartRequest, instance.InstanceListSavesRequest, - instance.InstanceSetSaveListSubscriptionsRequest, instance.InstanceSaveListUpdateEvent, instance.InstanceCreateSaveRequest, instance.InstanceRenameSaveRequest, @@ -83,11 +83,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, @@ -101,7 +99,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 new file mode 100644 index 00000000..ae346b65 --- /dev/null +++ b/packages/lib/src/subscriptions.ts @@ -0,0 +1,332 @@ +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) { + 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) { + 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); + } + } +} 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..31368b8b 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.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; 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; 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}`)); }); diff --git a/test/lib/subscriptions.js b/test/lib/subscriptions.js new file mode 100644 index 00000000..5656717b --- /dev/null +++ b/test/lib/subscriptions.js @@ -0,0 +1,585 @@ +"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; + + get subscriptionChannel() { + return "channelOne"; + } + } + 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() { + it("should be round trip json serialisable without an event replay", function() { + 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() { + 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 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()), + new Error(`Unregistered Event class ${UnregisteredEvent.name}`) + ); + }); + }); + + describe("class SubscriptionRequest", function() { + describe("permission()", function() { + function newSubscriptionRequestMessage(request) { + return new lib.MessageRequest( + 0, + addr({ controlId: 0 }), + addr("controller"), + lib.SubscriptionRequest.name, + request.toJSON() + ); + } + it("should do nothing when the event has no permission property", function() { + const mockUser = new MockUser(); + const request = new lib.SubscriptionRequest(RegisteredEvent.name, true); + lib.SubscriptionRequest.permission(mockUser, newSubscriptionRequestMessage(request)); + assert.equal(mockUser.lastPermissionCheck, undefined); + }); + it("should check user permission when the permission property is a string", function() { + const mockUser = new MockUser(); + const request = new lib.SubscriptionRequest(StringPermissionEvent.name, true); + lib.SubscriptionRequest.permission(mockUser, newSubscriptionRequestMessage(request)); + assert.equal(mockUser.lastPermissionCheck, "StringPermission"); + }); + it("should call the permission property when it is a function", function() { + const mockUser = new MockUser(); + const request = new lib.SubscriptionRequest(FunctionPermissionEvent.name, true); + lib.SubscriptionRequest.permission(mockUser, newSubscriptionRequestMessage(request)); + assert.equal(mockUser.lastPermissionCheck, "FunctionPermission"); + }); + }); + + it("should be round trip json serialisable", function() { + const request = new lib.SubscriptionRequest(RegisteredEvent.name, true, ["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: 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(ChannelEvent.name, true), + src: addr({ controlId: 2 }), + dst: addr("controller"), + }, { + id: 3, + request: new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelOne"]), + src: addr({ controlId: 3 }), + dst: addr("controller"), + }, { + id: 4, + request: new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelTwo"]), + src: addr({ controlId: 4 }), + dst: addr("controller"), + }]; + + function onceConnectorSend(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.equal(mockController.handles.has(lib.SubscriptionRequest), true); + }); + + describe("handle()", function() { + it("should accept registered events", function() { + subscriptions.handle(RegisteredEvent); + }); + it("should not accept unregistered events", function() { + assert.throws( + () => subscriptions.handle(UnregisteredEvent), + new Error(`Unregistered Event class ${UnregisteredEvent.name}`) + ); + }); + it("should not accept events already handled by the class", function() { + subscriptions.handle(RegisteredEvent); + assert.throws( + () => subscriptions.handle(RegisteredEvent), + new Error(`Event ${RegisteredEvent.name} is already registered`) + ); + }); + }); + + describe("broadcast()", function() { + beforeEach(function() { + subscriptions.handle(RegisteredEvent); + subscriptions.handle(ChannelEvent); + for (let connectorData of connectorSetupDate) { + subscriptions._handleRequest(connectorData.request, connectorData.src, connectorData.dst); + } + }); + + it("should not accept unregistered events", function() { + assert.throws( + () => subscriptions.broadcast(new UnregisteredEvent()), + new Error(`Unregistered Event class ${UnregisteredEvent.name}`) + ); + }); + it("should not accept events not handled by the class", function() { + assert.throws( + () => subscriptions.broadcast(new StringPermissionEvent()), + new Error(`Event ${StringPermissionEvent.name} is not a registered as subscribable`) + ); + }); + it("should notify all links who subscribed to all channels", async function() { + subscriptions.broadcast(new RegisteredEvent()); + await Promise.all([onceConnectorSend(0), onceConnectorSend(1)]); + assertLastEvent(0, RegisteredEvent); // RegisteredEvent: All + assertLastEvent(1, RegisteredEvent); // RegisteredEvent: All + 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 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 onceConnectorSend(1); + assertNoEvent(0); + assertLastEvent(1, RegisteredEvent); + assertNoEvent(2); + assertNoEvent(3); + assertNoEvent(4); + }); + it("should notify all links who subscribed the specific channel", async function() { + subscriptions.broadcast(new ChannelEvent()); + await Promise.all([onceConnectorSend(2), onceConnectorSend(3)]); + assertNoEvent(0); + assertNoEvent(1); + assertLastEvent(2, ChannelEvent); + assertLastEvent(3, ChannelEvent); + assertNoEvent(4); + }); + it("should not notify a link who unsubscribed from the specific channel", async function() { + 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 onceConnectorSend(2); + assertNoEvent(0); + assertNoEvent(1); + assertLastEvent(2, ChannelEvent); + assertNoEvent(3); + assertNoEvent(4); + }); + it("should not notify links which are closed or closing", async function() { + const link = mockController.wsServer.controlConnections.get(0); + link.connector.closing = true; + subscriptions.broadcast(new RegisteredEvent()); + await onceConnectorSend(1); + assertNoEvent(0); + assertLastEvent(1, RegisteredEvent); + assertNoEvent(2); + assertNoEvent(3); + assertNoEvent(4); + }); + }); + + describe("_handleRequest()", function() { + beforeEach(function() { + subscriptions.handle(RegisteredEvent); + 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.rejects( + 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; + assert.rejects( + 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 request = new lib.SubscriptionRequest(RegisteredEvent.name, true); + await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst); + subscriptions.broadcast(new RegisteredEvent()); + await onceConnectorSend(0); + assertLastEvent(0, RegisteredEvent); + }); + it("should accept a unsubscription from all channels", async function() { + const request = new lib.SubscriptionRequest(RegisteredEvent.name, true); + await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst); + subscriptions.broadcast(new RegisteredEvent()); + await onceConnectorSend(0); + assertLastEvent(0, RegisteredEvent); + + 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(0); + }); + it("should accept a subscription to a specific channel", async function() { + const request = new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelOne"]); + await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst); + subscriptions.broadcast(new ChannelEvent()); + await onceConnectorSend(0); + assertLastEvent(0, ChannelEvent); + }); + it("should accept a unsubscription from a specific channel", async function() { + const request = new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelOne"]); + await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst); + subscriptions.broadcast(new ChannelEvent()); + await onceConnectorSend(0); + assertLastEvent(0, ChannelEvent); + + 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(0); + }); + it("should accept a respond with an event replay when returned by the handler", async function() { + 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 onceConnectorSend(0); + assertLastEvent(0, StringPermissionEvent); + + const link = mockController.wsServer.controlConnections.get(0); + const lastMessage = link.connector.sentMessages.pop(); + const eventEntry = lib.Link._eventsByClass.get(StringPermissionEvent); + const response = eventEntry.eventFromJSON(lastMessage.data); + assert.deepEqual(response, new StringPermissionEvent()); + }); + }); + }); + + describe("class EventSubscriber", function() { + let channelEvent, mockControl, registeredEvent; + beforeEach(function() { + mockControl = new MockControl(new MockConnector( + addr({ controlId: 0 }), + 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.throws( + () => new lib.EventSubscriber(UnregisteredEvent), + new Error(`Unregistered Event class ${UnregisteredEvent.name}`) + ); + }); + 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() { + 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() { + const control = new MockControl(new MockConnector(addr({ controlId: 0 }), addr("controller"))); + registeredEvent.connectControl(control); + assert.equal(control._eventHandlers.has(RegisteredEvent), true); + }); + it("should do nothing if the control is already connected", function() { + 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() { + 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() { + 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() { + 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() { + function callback(value) { calledWith = value; } + assert.rejects( + () => registeredEvent.unsubscribe(callback), + new Error("handler is not registered") + ); + }); + }); + + describe("unsubscribeFromChannel()", function() { + it("should allow unsubscribing from a channel for an event", function() { + 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); + + 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() {} + assert.rejects( + () => channelEvent.unsubscribeFromChannel("channelOne", callback), + new Error("handler is not registered") + ); + + let calledWith = null; + function callbackTwo(value) { calledWith = value; } + channelEvent.subscribeToChannel("channelOne", callbackTwo); + assert.rejects( + () => channelEvent.unsubscribeFromChannel("channelOne", callback), + new Error("handler is not registered") + ); + const event = new ChannelEvent(); + channelEvent._handle(event); + assert.equal(calledWith, event); + }); + }); + + describe("_updateSubscription()", function() { + 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", 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", 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", async function() { + let calledWith = null; + const event = new RegisteredEvent(); + const request = registeredEvent.subscribe(function(value) { calledWith = value; }); + await events.once(mockControl.connector, "send"); + + 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() + )); + + await request; + assert.deepEqual(calledWith, 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() {