Class BunWebSocketTransport

src/bun.ts:58–135

}

export class BunWebSocketTransport<T = undefined> implements RpcTransport {
  constructor (ws: ServerWebSocket<T>) {
    this.#ws = ws;
  }

  #ws: ServerWebSocket<T>;
  #receiveResolver?: (message: string) => void;
  #receiveRejecter?: (err: any) => void;
  #receiveQueue: string[] = [];
  #error?: any;

  async send(message: string): Promise<void> {
    this.#ws.send(message);
  }

  async receive(): Promise<string> {
    if (this.#receiveQueue.length > 0) {
      return this.#receiveQueue.shift()!;
    } else if (this.#error) {
      throw this.#error;
    } else {
      return new Promise<string>((resolve, reject) => {
        this.#receiveResolver = resolve;
        this.#receiveRejecter = reject;
      });
    }
  }

  abort?(reason: any): void {
    let message: string;
    if (reason instanceof Error) {
      message = reason.message;
    } else {
      message = `${reason}`;
    }
    this.#ws.close(3000, message);

    if (!this.#error) {
      this.#error = reason;
      // No need to call receiveRejecter(); RPC implementation will stop listening anyway.
    }
  }

  dispatchMessage(data: string | Buffer): void {
    if (this.#error) {
      return;
    }

    let strData = typeof data === "string" ? data : data.toString("utf-8");

    if (this.#receiveResolver) {
      this.#receiveResolver(strData);
      this.#receiveResolver = undefined;
      this.#receiveRejecter = undefined;
    } else {
      this.#receiveQueue.push(strData);
    }
  }