diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index 0be6fd0b..f8f64c80 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -39,30 +39,45 @@ export class SubscriptionRequest { public eventName: string, public subscribe: boolean, public lastRequestTimeMs: number = 0, + public filters: string | string[] | undefined = undefined, ) { if (!Link._eventsByName.has(eventName)) { throw new Error(`Unregistered Event class ${eventName}`); } } - static jsonSchema = Type.Tuple([ - Type.String(), - Type.Boolean(), - Type.Number(), + static jsonSchema = Type.Union([ + Type.Tuple([ + Type.String(), + Type.Boolean(), + Type.Number(), + ]), + Type.Tuple([ + Type.String(), + Type.Boolean(), + Type.Number(), + Type.Union([Type.String(), Type.Array(Type.String())]), + ]), ]); toJSON() { + if (this.filters !== undefined) { + return [this.eventName, this.subscribe, this.lastRequestTimeMs, this.filters]; + } return [this.eventName, this.subscribe, this.lastRequestTimeMs]; } static fromJSON(json: Static): SubscriptionRequest { - return new this(...json); + const [eventName, subscribe, lastRequestTimeMs, filters] = json; + return new this(eventName, subscribe, lastRequestTimeMs, filters); } } type Subscriber = { link: Link, dst: Address, + /** Optional filters this subscriber is interested in. Undefined means all events */ + filters?: string[], } type EventData = { @@ -77,6 +92,13 @@ type EventData = { export class SubscriptionController { _events = new Map(); + private static normalizeFilters(value?: string | string[]): string[] | undefined { + if (value === undefined) { + return undefined; + } + return Array.isArray(value) ? value : [value]; + } + /** * 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 @@ -102,7 +124,7 @@ export class SubscriptionController { * Broadcast an event to all subscribers of that event * @param event - Event to broadcast. */ - broadcast(event: Event) { + broadcast(event: Event, filter?: string | string[]) { const entry = Link._eventsByClass.get(event.constructor); if (!entry) { throw new Error(`Unregistered Event class ${event.constructor.name}`); @@ -112,10 +134,19 @@ export class SubscriptionController { throw new Error(`Event ${entry.name} is not a registered as subscribable`); } const toRemove: Subscriber[] = []; + const filtersToMatch = SubscriptionController.normalizeFilters(filter); + for (const subscriber of eventData.subscriptions) { if ((subscriber.link.connector as WebSocketBaseConnector).closing) { toRemove.push(subscriber); - } else { + continue; + } + + if ( + filtersToMatch === undefined + || subscriber.filters === undefined + || subscriber.filters.some(f => filtersToMatch.includes(f)) + ) { subscriber.link.sendTo(subscriber.dst, event); } } @@ -164,13 +195,42 @@ export class SubscriptionController { throw new Error(`Event ${event.eventName} is not a registered as subscribable`); } if (event.subscribe === false) { - const index = eventData.subscriptions.findLastIndex(subscriber => subscriber.dst.addressedTo(src)); - if (index >= 0) { - eventData.subscriptions.splice(index, 1); + // Unsubscribe logic + const subscriber = eventData.subscriptions.find(sub => sub.dst.addressedTo(src)); + if (!subscriber) { + return false; + } + if (event.filters === undefined) { + // Remove entire subscription for this address + eventData.subscriptions = eventData.subscriptions.filter(sub => !sub.dst.addressedTo(src)); + } else { + const filtersToRemove = Array.isArray(event.filters) ? event.filters : [event.filters]; + if (subscriber.filters === undefined) { + // Was subscribed to all, remove subscription entirely + eventData.subscriptions = eventData.subscriptions.filter(sub => !sub.dst.addressedTo(src)); + } else { + subscriber.filters = subscriber.filters.filter(f => !filtersToRemove.includes(f)); + if (subscriber.filters.length === 0) { + // No filters left, remove subscription + eventData.subscriptions = eventData.subscriptions.filter(sub => sub !== subscriber); + } + } } } else { - if (!eventData.subscriptions.some(subscriber => subscriber.dst.addressedTo(src))) { - eventData.subscriptions.push({ link: link, dst: src }); + // Subscribe logic + let subscriber = eventData.subscriptions.find(sub => sub.dst.addressedTo(src)); + const newFilters = SubscriptionController.normalizeFilters(event.filters); + if (!subscriber) { + eventData.subscriptions.push({ link: link, dst: src, filters: newFilters }); + } else if (subscriber.filters === undefined || newFilters === undefined) { + // Either was already subscribed to all or new request wants all + subscriber.filters = undefined; + } else { + for (const f of newFilters) { + if (!subscriber.filters.includes(f)) { + subscriber.filters.push(f); + } + } } if (eventData.subscriptionUpdate) { const eventReplay = await eventData.subscriptionUpdate(event, src, dst);