Method #spawn

node/src/transport.ts:144–172

()

  #startReader(stdout: Readable): void {
    void (async () => {
      try {
        for await (const envelope of sizeDelimitedDecodeStream(EnvelopeSchema, stdout)) {
          this.#dispatch(envelope);
        }
        this.#failAll(new ConnectionError("cmdop-core stdout closed", "transport"));
      } catch (error) {
        this.#failAll(
          new ConnectionError(
            `core read failed: ${error instanceof Error ? error.message : String(error)}`,
            "transport",
          ),
        );
      }
    })();
  }

  #dispatch(envelope: Envelope): void {
    const pending = this.#pending.get(envelope.id);
    if (!pending) return;
    if (pending.kind === "unary") {
      if (envelope.kind === Envelope_Kind.RESPONSE) {
        this.#pending.delete(envelope.id);
        clearTimeout(pending.timer);
        pending.resolve(envelope);
      } else if (envelope.kind === Envelope_Kind.ERROR && envelope.payload.case === "error") {
        this.#pending.delete(envelope.id);
        clearTimeout(pending.timer);
        pending.reject(mapCoreError(envelope.payload.value));
      }
      return;
    }
    if (envelope.kind === Envelope_Kind.EVENT) {