diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index a8be14b6..32cb28f4 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -1,5 +1,5 @@ import { Type, Static } from "@sinclair/typebox"; -import { Link, Event, EventClass, RequestHandler, WebSocketClientConnector } from "./link"; +import { Link, Event, EventClass, RequestHandler, WebSocketClientConnector, WebSocketBaseConnector } from "./link"; import { Address, MessageRequest } from "./data"; import { User } from "./users"; @@ -117,6 +117,10 @@ export class SubscriptionController { this.controller.handle(SubscriptionRequest, this._handleEvent.bind(this)); } + /** + * Allow clients to subscribe to an event by telling the subscription controller to accept them + * Has an optional subscription update handler which is called when any client updates their subscription + */ handle(Event: EventClass, subscriptionUpdate?: SubscriptionRequestHandler, channelCategoriser?: SubscriptionChannelCategoriser): void; handle( Event: EventClass, @@ -137,6 +141,9 @@ export class SubscriptionController { }); } + /** + * Broadcast an event to all subscribers of that event, will be filtered by channels when relevant + */ broadcast(event: Event): void; broadcast(event: Event) { const entry = Link._eventsByClass.get(event.constructor); @@ -148,8 +155,10 @@ export class SubscriptionController { throw new Error(`Event ${entry.name} is not a registered as subscribable`); } const channel = eventData.channelCategoriser ? eventData.channelCategoriser(event) : null; - for (let [link, subscription] of eventData.subscriptions.entries()) { - if (subscription.all || (channel && subscription.channels.includes(channel))) { + for (let [link, subscription] of eventData.subscriptions) { + if ((link.connector as WebSocketBaseConnector).closing) { + eventData.subscriptions.delete(link); + } else if ((subscription.all || (channel && subscription.channels.includes(channel)))) { link.send(event); } } @@ -169,6 +178,11 @@ export class SubscriptionController { eventData.subscriptions.delete(link); return new SubscriptionResponse(eventReplay); } else { + if (!eventData.subscriptions.has(link)) { + link.connector.once("close", () => { + eventData.subscriptions.delete(link); + }) + } eventData.subscriptions.set(link, { all: event.allChannels, channels: event.channels }); return new SubscriptionResponse(eventReplay); } @@ -182,12 +196,11 @@ export class SubscriptionController { export class EventSubscriber { _callbacks = new Array>() _channelCallbacks = new Map>>() - lastResponse?: Event = undefined; + lastResponse: Event | null = null; lastResponseTime = 0; constructor( private event: EventClass, - private channelCategoriser?: SubscriptionChannelCategoriser, private control?: Link ) { const entry = Link._eventsByClass.get(this.event); @@ -213,12 +226,12 @@ export class EventSubscriber { this.control.handle(this.event, this._handle.bind(this)); } - subscribeAll(handler: EventSubscriberCallback) { + subscribe(handler: EventSubscriberCallback) { this._callbacks.push(handler); this._updateSubscription(); } - subscribe(channel: string | number, handler: EventSubscriberCallback) { + subscribeToChannel(channel: string | number, handler: EventSubscriberCallback) { if (!this._channelCallbacks.get(channel)) { this._channelCallbacks.set(channel, []); } @@ -226,7 +239,7 @@ export class EventSubscriber { this._updateSubscription(); } - unsubscribeAll(handler: EventSubscriberCallback) { + unsubscribe(handler: EventSubscriberCallback) { let index = this._callbacks.lastIndexOf(handler); if (index === -1) { throw new Error("handler is not registered"); @@ -236,7 +249,7 @@ export class EventSubscriber { this._updateSubscription(); } - unsubscribe(channel: string | number, handler: EventSubscriberCallback) { + unsubscribeFromChannel(channel: string | number, handler: EventSubscriberCallback) { const channelCallbacks = this._channelCallbacks.get(channel); if (!channelCallbacks) { throw new Error("handler is not registered");