From d84445baa14c7067540f6bb356b8e281faedb820 Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Thu, 12 Oct 2023 17:07:07 +0200 Subject: [PATCH] Fix line endings and indents --- packages/lib/src/subscriptions.ts | 674 +++++++++++++++--------------- 1 file changed, 337 insertions(+), 337 deletions(-) diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index 082a5a67..0b186504 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -1,337 +1,337 @@ -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,): void; - 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): void; - 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); - } - } -} \ No newline at end of file +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,): void; + 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): void; + 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); + } + } +}