import { ChildProcess, spawn } from "node:child_process"; import { once } from "node:events"; import { resolve } from "node:path"; import { createInterface } from "node:readline"; import { json } from "../test/fixtures"; import WebSocket from "ws"; import { decodeNotification, WEBSOCKET_SUBPROTOCOL } from "../src/protocol"; export async function startImage(): Promise<{ address: string; stop: () => Promise; diagnostics: () => string; }> { const executable = resolve( __dirname, "../../../target/debug", process.platform === "win32" ? "runa.exe" : "runa", ); const child = spawn(executable, ["--listen", "127.0.0.1:0"], { stdio: ["ignore", "ignore", "pipe"], }); let diagnostics = ""; child.stderr.on("data", (chunk: Buffer) => { diagnostics = (diagnostics + chunk.toString()).slice(-8192); }); const stop = () => stopImage(child); let address: string; try { address = await new Promise((resolve, reject) => { const lines = createInterface({ input: child.stderr }); let diagnostics = ""; const cleanup = () => { clearTimeout(timer); child.off("error", fail); child.off("exit", exited); lines.close(); child.stderr.resume(); }; const fail = (error: Error) => { cleanup(); reject(error); }; const exited = () => fail(new Error(`Image exited during startup: ${diagnostics}`)); const timer = setTimeout( () => fail(new Error(`Image startup timed out: ${diagnostics}`)), 10000, ); child.once("error", fail); child.once("exit", exited); lines.on("line", (line) => { diagnostics = (diagnostics + line + "\n").slice(-8192); const address = line.match(/^runa: listening on (.+)$/)?.[1]; if (address) { cleanup(); resolve(address); } }); }); } catch (error) { await stop(); throw error; } return { address, stop, diagnostics: () => diagnostics }; } async function stopImage(child: ChildProcess): Promise { if (child.exitCode !== null || child.signalCode !== null || child.pid === undefined) return; const exited = once(child, "exit"); child.kill(); await exited; } export async function control(address: string, command: unknown): Promise { const socket = new WebSocket(`ws://${address}/`, WEBSOCKET_SUBPROTOCOL); const timer = setTimeout(() => socket.terminate(), 5000); try { await once(socket, "open"); const response = new Promise((resolve, reject) => { socket.on("message", (data, binary) => { try { if (binary) throw new Error("Unexpected binary control reply"); const text = data.toString(); if (decodeNotification(text) !== undefined) return; const decoded: unknown = JSON.parse(text); resolve(decoded); } catch (error) { reject(error); } }); socket.once("close", () => reject(new Error("Control connection closed"))); socket.once("error", reject); }); socket.send(json({ version: 5, request_id: "integration-control", command })); return await response; } finally { clearTimeout(timer); socket.close(); } } export async function spawnCounter(address: string): Promise { const reply = await control(address, { operation: "spawn", parameters: { path: "actors/counter", state: { type: "integer", data: 9223372036854775807n }, mailbox_capacity: 4, }, }); const ref = json(reply).match(/"actor_ref":"([A-Za-z0-9_-]{21})"/)?.[1]; if (!ref) throw new Error(`Spawn failed: ${json(reply)}`); return ref; }