Class FrameQueue

node/src/transport.ts:53–104

* A bounded push→pull async-iterator: bridges the demux's push of frames into a * `for await` consumer (the canonical "queueable" pattern, report 11 §1.4).

  }

  end(): void {
    this.#done = true;
    this.#wakeDone();
  }

  fail(error: Error): void {
    this.#error = error;
    this.#done = true;
    this.#wakeDone();
  }

  #wakeDone(): void {
    if (this.#wait) {
      const waiter = this.#wait;
      this.#wait = undefined;
      waiter({ value: undefined as never, done: true });
    }
  }

  async *iterate(): AsyncGenerator<Envelope> {
    for (;;) {
      if (this.#error) throw this.#error;
      if (this.#frames.length) {
        yield this.#frames.shift()!;
        continue;
      }
      if (this.#done) return;
      const result = await new Promise<IteratorResult<Envelope>>((resolve) => {
        this.#wait = resolve;
      });
      if (this.#error) throw this.#error;
      if (result.done) return;
      yield result.value;
    }
  }
}

export class Transport {
  #child?: ChildProcessWithoutNullStreams;
  #stdin?: Writable;
  #pending = new Map<bigint, UnaryPending | StreamPending>();
  #nextId = 1n;
  #closed = false;
  #starting?: Promise<void>;

  constructor(
    private readonly config: ClientConfig,
    private readonly binaryPath: string,
  ) {}

  /** @internal test seam */
  attachStreamsForTest(stdin: Writable, stdout: Readable): void {
    this.#stdin = stdin;
    this.#child = { stdin } as unknown as ChildProcessWithoutNullStreams;
    this.#startReader(stdout);