diff --git a/packages/controller/src/BaseConnection.ts b/packages/controller/src/BaseConnection.ts index 980b0a14..493ebbde 100644 --- a/packages/controller/src/BaseConnection.ts +++ b/packages/controller/src/BaseConnection.ts @@ -34,6 +34,7 @@ export default class BaseConnection extends lib.Link { this.handle(lib.ModPackGetDefaultRequest, this.handleModPackGetDefaultRequest.bind(this)); this.handle(lib.ModDownloadRequest, this.handleModDownloadRequest.bind(this)); + this.handle(lib.SubscriptionRequest, this.handleSubscriptionRequest.bind(this)); this.connector.on("close", () => { this._controller.subscriptions.unsubscribe(this); }); @@ -102,4 +103,8 @@ export default class BaseConnection extends lib.Link { return stream.id; } + + async handleSubscriptionRequest(request: lib.SubscriptionRequest, src: lib.Address, dst: lib.Address) { + return await this._controller.subscriptions.handleRequest(this, request, src, dst); + } } diff --git a/packages/controller/src/Controller.ts b/packages/controller/src/Controller.ts index 289acf89..7efe16be 100644 --- a/packages/controller/src/Controller.ts +++ b/packages/controller/src/Controller.ts @@ -76,7 +76,7 @@ export default class Controller { private _events: events.EventEmitter = new events.EventEmitter(); /** Event subscription controller */ - subscriptions: lib.SubscriptionController; + subscriptions = new lib.SubscriptionController(); // Possible states are new, starting, running, stopping, stopped private _state: string = "new"; @@ -148,7 +148,6 @@ export default class Controller { this.app.locals.streams = new Map(); this.wsServer = new WsServer(this); - this.subscriptions = new lib.SubscriptionController(this); this.modStore.on("change", mod => { this.modsUpdated([mod]); diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index 411449f4..03cdeb86 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -111,12 +111,6 @@ type EventData = { export class SubscriptionController { _events = new Map(); - 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 a client subscribes @@ -172,14 +166,14 @@ export class SubscriptionController { } /** - * Handle incoming event subscription requests + * Handle incoming subscription requests on a link + * @param link - Link message was received on * @param event - incomming event. * @param src - Source address of incomming request. * @param dst - destination address of incomming request. * @returns Response to subscription request. - * @internal */ - async _handleRequest(event: SubscriptionRequest, src: Address, dst: Address) { + async handleRequest(link: Link, event: SubscriptionRequest, src: Address, dst: Address) { if (!Link._eventsByName.has(event.eventName)) { throw new Error(`Event ${event.eventName} is not a registered event`); } @@ -188,7 +182,6 @@ export class SubscriptionController { throw new Error(`Event ${event.eventName} is not a registered as subscribable`); } let eventReplay: Event | null = null; - const link: Link = this.controller.wsServer.controlConnections.get(src.id); if (event.subscribe === false) { eventData.subscriptions.delete(link); } else { diff --git a/test/lib/subscriptions.js b/test/lib/subscriptions.js index 14abba32..947dad46 100644 --- a/test/lib/subscriptions.js +++ b/test/lib/subscriptions.js @@ -147,20 +147,22 @@ describe("lib/subscriptions", function() { dst: addr("controller"), }]; + function getLink(id) { + return mockController.wsServer.controlConnections.get(id); + } function onceConnectorSend(id) { - const link = mockController.wsServer.controlConnections.get(id); - return events.once(link.connector, "send"); + return events.once(getLink(id).connector, "send"); } function assertLastEvent(connectorId, Event) { - const link = mockController.wsServer.controlConnections.get(connectorId); + const link = getLink(connectorId); const lastMessage = link.connector.sentMessages.at(-1); const name = Event.plugin ? `${Event.plugin}:${Event.name}` : Event.name; assert.equal(lastMessage.name, name); } function assertNoEvent(connectorId) { - const link = mockController.wsServer.controlConnections.get(connectorId); + const link = getLink(connectorId); assert.equal(link.connector.sentMessages.length, 0); } @@ -174,10 +176,6 @@ describe("lib/subscriptions", function() { 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); @@ -201,7 +199,9 @@ describe("lib/subscriptions", function() { beforeEach(function() { subscriptions.handle(RegisteredEvent); for (let connectorData of connectorSetupDate) { - subscriptions._handleRequest(connectorData.request, connectorData.src, connectorData.dst); + subscriptions.handleRequest( + getLink(connectorData.id), connectorData.request, connectorData.src, connectorData.dst + ); } }); @@ -226,15 +226,17 @@ describe("lib/subscriptions", function() { it("should not notify a link who unsubscribed", async function() { const connectorData = connectorSetupDate[0]; const request = new lib.SubscriptionRequest(RegisteredEvent.name, false); - await subscriptions._handleRequest(request, connectorData.src, connectorData.dst); - mockController.wsServer.controlConnections.get(0).connector.sentMessages.pop(); // Response to request + await subscriptions.handleRequest( + getLink(connectorData.id), request, connectorData.src, connectorData.dst, + ); + getLink(0).connector.sentMessages.pop(); // Response to request subscriptions.broadcast(new RegisteredEvent()); await onceConnectorSend(1); assertNoEvent(0); assertLastEvent(1, RegisteredEvent); }); it("should not notify links which are closed or closing", async function() { - const link = mockController.wsServer.controlConnections.get(0); + const link = getLink(0); link.connector.closing = true; subscriptions.broadcast(new RegisteredEvent()); await onceConnectorSend(1); @@ -247,12 +249,14 @@ describe("lib/subscriptions", function() { beforeEach(function() { subscriptions.handle(RegisteredEvent); for (let connectorData of connectorSetupDate) { - subscriptions._handleRequest(connectorData.request, connectorData.src, connectorData.dst); + subscriptions.handleRequest( + getLink(connectorData.id), connectorData.request, connectorData.src, connectorData.dst + ); } }); it("should remove an active subscription", async function() { - const link = mockController.wsServer.controlConnections.get(0); + const link = getLink(0); subscriptions.unsubscribe(link); subscriptions.broadcast(new RegisteredEvent()); await onceConnectorSend(1); @@ -261,7 +265,7 @@ describe("lib/subscriptions", function() { }); }); - describe("_handleRequest()", function() { + describe("handleRequest()", function() { beforeEach(function() { subscriptions.handle(RegisteredEvent); }); @@ -270,7 +274,7 @@ describe("lib/subscriptions", function() { const request = new lib.SubscriptionRequest(RegisteredEvent.name, true); request.eventName = UnregisteredEvent.name; assert.rejects( - subscriptions._handleRequest(request, addr({ controlId: 0 }), addr("controller")), + subscriptions.handleRequest(getLink(0), request, addr({ controlId: 0 }), addr("controller")), new Error(`Unregistered Event class ${UnregisteredEvent.name}`) ); }); @@ -278,29 +282,35 @@ describe("lib/subscriptions", function() { const request = new lib.SubscriptionRequest(RegisteredEvent.name, true); request.eventName = StringPermissionEvent.name; assert.rejects( - subscriptions._handleRequest(request, addr({ controlId: 0 }), addr("controller")), + subscriptions.handleRequest(getLink(0), request, addr({ controlId: 0 }), addr("controller")), new Error(`Event ${StringPermissionEvent.eventName} is not a registered as subscribable`) ); }); it("should accept a subscription", async function() { const request = new lib.SubscriptionRequest(RegisteredEvent.name, true); - await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst); + await subscriptions.handleRequest( + getLink(0), request, connectorSetupDate[0].src, connectorSetupDate[0].dst + ); subscriptions.broadcast(new RegisteredEvent()); await onceConnectorSend(0); assertLastEvent(0, RegisteredEvent); }); it("should accept a unsubscription", async function() { const request = new lib.SubscriptionRequest(RegisteredEvent.name, true); - await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst); + await subscriptions.handleRequest( + getLink(0), request, connectorSetupDate[0].src, connectorSetupDate[0].dst + ); subscriptions.broadcast(new RegisteredEvent()); await onceConnectorSend(0); assertLastEvent(0, RegisteredEvent); - const link = mockController.wsServer.controlConnections.get(0); + const link = getLink(0); link.connector.sentMessages.pop(); const unsubRequest = new lib.SubscriptionRequest(RegisteredEvent.name, false); - await subscriptions._handleRequest(unsubRequest, connectorSetupDate[0].src, connectorSetupDate[0].dst); + await subscriptions.handleRequest( + getLink(0), unsubRequest, connectorSetupDate[0].src, connectorSetupDate[0].dst + ); subscriptions.broadcast(new RegisteredEvent()); assertNoEvent(0); }); @@ -310,12 +320,14 @@ describe("lib/subscriptions", function() { }); const request = new lib.SubscriptionRequest(StringPermissionEvent.name, true); - await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst); + await subscriptions.handleRequest( + getLink(0), 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 link = getLink(0); const lastMessage = link.connector.sentMessages.pop(); const eventEntry = lib.Link._eventsByClass.get(StringPermissionEvent); const response = eventEntry.eventFromJSON(lastMessage.data);