Class WsNamespace

web/crux/src/websockets/namespace.ts:18–204

} from './common'

export default class WsNamespace implements WsSubscription {
  private readonly logger: Logger

  private readonly clients: Map<string, ClientResources> = new Map()

  private readonly path: string

  readonly params: Record<string, string>

  constructor(match: WsRouteMatch) {
    this.path = match.path
    this.params = match.params

    this.logger = new Logger(`${WsNamespace.name} ${this.path}`)
  }

  close() {
    this.clients.forEach(it => it.client.subscriptions.delete(this.path))
    this.clients.clear()

    this.logger.verbose('Closed')
  }

  getCompleter(clientToken: string): Observable<undefined> {
    const resources = this.clients.get(clientToken)
    return resources.completer as Observable<undefined>
  }

  getParameter(name: string): string | null {
    return this.params[name] ?? null
  }

  onSubscribe(
    client: WsClient,
    callbacks: WsClientCallbacks,
    message: WsMessage<SubscriptionMessage>,
  ): Observable<WsMessage> {
    if (this.clients.has(client.token)) {
      return EMPTY
    }

    const { subscribe, handlers, transform, unsubscribe } = callbacks

    const resources: ClientResources = {
      client,
      handlers,
      transform,
      unsubscribe,
      completer: new Subject(),
    }
    this.clients.set(client.token, resources)

    client.subscriptions.set(this.path, this)

    this.logger.verbose(`${client.token} subscribed`)

    const res: WsMessage<SubscriptionMessage> = {
      type: WS_TYPE_SUBBED,