Class ObjectTestTransport

__tests__/index.test.ts:479–533

}

class ObjectTestTransport implements RpcTransportWithCustomEncoding {
  constructor(
      private partner?: ObjectTestTransport,
      readonly encodingLevel: CustomEncodingLevel = "jsonCompatible") {
    if (partner) {
      partner.partner = this;
    }
  }

  private queue: unknown[] = [];
  private waiter?: () => void;
  private aborter?: (err: any) => void;
  private fenced = false;

  send(message: unknown): void {
    let cloned = this.encodingLevel === "jsonCompatible" ? JSON.parse(JSON.stringify(message))
                                                         : structuredClone(message);
    this.partner!.queue.push(cloned);
    if (this.partner!.waiter && !this.partner!.fenced) {
      this.partner!.waiter();
      this.partner!.waiter = undefined;
      this.partner!.aborter = undefined;
    }
  }

  async receive(): Promise<unknown> {
    while (this.queue.length == 0 || this.fenced) {
      await new Promise<void>((resolve, reject) => {
        this.waiter = resolve;
        this.aborter = reject;
      });
    }

    return this.queue.shift()!;
  }

  fence() {
    this.fenced = true;
  }

  releaseFence() {
    this.fenced = false;
    if (this.queue.length > 0 && this.waiter) {
      this.waiter();
      this.waiter = undefined;
      this.aborter = undefined;
    }
  }

  abort(reason: any): void {
    this.aborter?.(reason);
    this.waiter = undefined;
    this.aborter = undefined;
  }
}

// Spin the microtask queue a bit to give messages time to be delivered and handled.
async function pumpMicrotasks() {