clusterflux-public/scripts/dap-client.js

const cp = require("child_process");

class DapClient {
  constructor({
    cwd = process.cwd(),
    env = process.env,
    command = "cargo",
    args = [
      "run",
      "-q",
      "-p",
      "clusterflux-dap",
      "--bin",
      "clusterflux-debug-dap",
    ],
  } = {}) {
    this.child = cp.spawn(
      command,
      args,
      { cwd, env }
    );
    this.seq = 1;
    this.buffer = Buffer.alloc(0);
    this.messages = [];
    this.waiters = [];
    this.stderr = "";

    this.child.stdout.on("data", (chunk) => {
      this.buffer = Buffer.concat([this.buffer, chunk]);
      this.parse();
    });
    this.child.stderr.on("data", (chunk) => {
      this.stderr += chunk.toString();
    });
    this.child.on("exit", () => this.flushWaiters());
  }

  send(command, args = {}) {
    const seq = this.seq++;
    const message = { seq, type: "request", command, arguments: args };
    const payload = Buffer.from(JSON.stringify(message));
    this.child.stdin.write(`Content-Length: ${payload.length}\r\n\r\n`);
    this.child.stdin.write(payload);
    return seq;
  }

  async response(seq, command) {
    const message = await this.waitFor(
      (item) =>
        item.type === "response" &&
        item.request_seq === seq &&
        item.command === command
    );
    if (!message.success) {
      throw new Error(
        `DAP ${command} failed: ${message.message || JSON.stringify(message)}\nAdapter stderr:\n${this.stderr}`
      );
    }
    return message;
  }

  async failure(seq, command) {
    const message = await this.waitFor(
      (item) =>
        item.type === "response" &&
        item.request_seq === seq &&
        item.command === command
    );
    if (message.success) {
      throw new Error(`DAP ${command} unexpectedly succeeded`);
    }
    return message;
  }

  waitFor(predicate, timeoutMs = 120000) {
    const existing = this.messages.find(predicate);
    if (existing) return Promise.resolve(existing);

    return new Promise((resolve, reject) => {
      const timer = setTimeout(() => {
        this.child.kill("SIGKILL");
        const recent = this.messages
          .slice(-10)
          .map((message) => JSON.stringify(message))
          .join("\n");
        reject(
          new Error(
            `timed out waiting for DAP message\nRecent DAP messages:\n${recent}\nAdapter stderr:\n${this.stderr}`
          )
        );
      }, timeoutMs);
      this.waiters.push({ predicate, resolve, timer });
    });
  }

  parse() {
    while (true) {
      const headerEnd = this.buffer.indexOf("\r\n\r\n");
      if (headerEnd < 0) return;
      const header = this.buffer.slice(0, headerEnd).toString();
      const match = header.match(/Content-Length: (\d+)/i);
      if (!match) throw new Error(`bad DAP header: ${header}`);
      const length = Number(match[1]);
      const start = headerEnd + 4;
      const end = start + length;
      if (this.buffer.length < end) return;
      const payload = this.buffer.slice(start, end).toString();
      this.buffer = this.buffer.slice(end);
      this.messages.push(JSON.parse(payload));
      this.flushWaiters();
    }
  }

  flushWaiters() {
    for (const waiter of [...this.waiters]) {
      const message = this.messages.find(waiter.predicate);
      if (!message) continue;
      clearTimeout(waiter.timer);
      this.waiters.splice(this.waiters.indexOf(waiter), 1);
      waiter.resolve(message);
    }
  }

  async close() {
    if (this.child.exitCode !== null) return;
    const seq = this.send("disconnect");
    await this.response(seq, "disconnect");
    this.child.stdin.end();
  }
}

module.exports = { DapClient };