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); });