webtransport.ts

import { Transport } from "../transport.js";
import { nextTick } from "./websocket-constructor.js";
import {
  Packet,
  createPacketDecoderStream,
  createPacketEncoderStream,
} from "engine.io-parser";
import debugModule from "debug"; // debug()

const debug = debugModule("engine.io-client:webtransport"); // debug()

export class WT extends Transport {
  private transport: any;
  private writer: any;

  get name() {
    return "webtransport";
  }

  protected doOpen() {
    // @ts-ignore
    if (typeof WebTransport !== "function") {
      return;
    }
    // @ts-ignore
    this.transport = new WebTransport(
      this.createUri("https"),
      this.opts.transportOptions[this.name]
    );

    this.transport.closed
      .then(() => {
        debug("transport closed gracefully");
        this.onClose();
      })
      .catch((err) => {
        debug("transport closed due to %s", err);
        this.onError("webtransport error", err);
      });

    // note: we could have used async/await, but that would require some additional polyfills
    this.transport.ready.then(() => {
      this.transport.createBidirectionalStream().then((stream) => {
        const decoderStream = createPacketDecoderStream(
          Number.MAX_SAFE_INTEGER,
          this.socket.binaryType
        );
        const reader = stream.readable.pipeThrough(decoderStream).getReader();

        const encoderStream = createPacketEncoderStream();
        encoderStream.readable.pipeTo(stream.writable);
        this.writer = encoderStream.writable.getWriter();

        const read = () => {
          reader
            .read()
            .then(({ done, value }) => {
              if (done) {
                debug("session is closed");
                return;
              }
              debug("received chunk: %o", value);
              this.onPacket(value);
              read();
            })
            .catch((err) => {
              debug("an error occurred while reading: %s", err);
            });
        };

        read();

        const packet: Packet = { type: "open" };
        if (this.query.sid) {
          packet.data = `{"sid":"${this.query.sid}"}`;
        }
        this.writer.write(packet).then(() => this.onOpen());
      });
    });
  }

  protected write(packets: Packet[]) {
    this.writable = false;

    for (let i = 0; i < packets.length; i++) {
      const packet = packets[i];
      const lastPacket = i === packets.length - 1;

      this.writer.write(packet).then(() => {
        if (lastPacket) {
          nextTick(() => {
            this.writable = true;
            this.emitReserved("drain");
          }, this.setTimeoutFn);
        }
      });
    }
  }

  protected doClose() {
    this.transport?.close();
  }
}