流式命令输出

在命令运行时，于浏览器中显示命令输出。Durable Object 会将标准输出和标准错误作为服务器发送事件 ↗发送，报告退出代码，并在客户端断开连接时通知命令停止。

前提条件

带有 Durable Object 的 Worker，该 Durable Object 使用 Durable Object 调度策略启动 container。要创建一个，请参阅运行 Linux 命令。

流式运行命令

向 Durable Object 添加方法，运行命令并将其输出作为事件流返回：

src/index.tsts

export class MyContainer extends DurableObject<Env> {
    // ...

    async stream(argv: string[]): Promise<Response> {
        const container = this.ctx.container;

        if (!container) {
            throw new Error("No container is configured");
        }

        if (!container.running) {
            container.start({
                image: "cloudflare/debian-trixie",
                entrypoint: ["sleep", "infinity"],
                enableInternet: false,
            });
        }

        const process = await container.exec(argv);
        const { readable, writable } = new TransformStream<Uint8Array, Uint8Array>();
        const writer = writable.getWriter();
        const encoder = new TextEncoder();

        const send = (event: string, data: unknown) =>
            writer.write(
                encoder.encode(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`),
            );

        const forward = async (
            stream: ReadableStream<Uint8Array>,
            event: "stdout" | "stderr",
        ) => {
            for await (const text of stream.pipeThrough(new TextDecoderStream())) {
                await send(event, text);
            }
        };

        let exited = false;
        const markExited = () => {
            exited = true;
        };
        process.exitCode.then(markExited, markExited);

        // Signaling a process that has exited records an internal error.
        const stop = () => {
            if (!exited) process.kill();
        };

        // Signal the process when the client disconnects.
        writer.closed.catch(stop);

        // Detect a disconnected client while the process writes nothing.
        const heartbeat = setInterval(() => {
            writer.write(encoder.encode(": keep-alive\n\n")).catch(() => {});
        }, 5_000);

        const pump = async () => {
            try {
                await Promise.all([
                    forward(process.stdout!, "stdout"),
                    forward(process.stderr!, "stderr"),
                ]);
                await send("exit", { exitCode: await process.exitCode });
                await writer.close();
            } catch (error) {
                stop();
                // End the response, so a connected client does not wait forever.
                await writer.abort(error).catch(() => {});
            } finally {
                clearInterval(heartbeat);
            }
        };

        // Keep streaming after this method returns the response.
        void pump();

        return new Response(readable, {
            headers: {
                "Content-Type": "text/event-stream",
                "Cache-Control": "no-cache",
            },
        });
    }
}

该方法会在命令完成前返回响应。随后 pump() 会在进程产生数据时写入每个块。pump() 同时读取标准输出和标准错误，因为无人读取的流可能会阻塞持续写入它的进程。TextDecoderStream 会保持跨越两个块的字符完整。

客户端断开连接时，下次写入会失败，随后该方法调用 kill()，向命令发送 SIGTERM。命令可能会忽略 SIGTERM；如需强制执行时间限制，请参阅在时间限制后停止命令。每 5 秒发送一次的保持活动注释，可确保命令没有输出时也会发生写入。

向 Worker 添加路由，流式运行你的应用选择的命令：

src/index.jsjs

const build = [
	"sh",
	"-c",
	'for i in 1 2 3; do echo "step $i"; sleep 1; done; echo "warning: almost done" >&2; exit 3',
];

export default {
	async fetch(request, env) {
		const url = new URL(request.url);
		const match =
			/^\/sandboxes\/([a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?)\/build$/.exec(
				url.pathname,
			);

		if (!match || request.method !== "GET") {
			return new Response("Not found", { status: 404 });
		}

		const sandbox = env.MY_CONTAINER.getByName(match[1]);
		return sandbox.stream(build);
	},
};

src/index.tsts

const build = [
    "sh",
    "-c",
    'for i in 1 2 3; do echo "step $i"; sleep 1; done; echo "warning: almost done" >&2; exit 3',
];

export default {
    async fetch(request: Request, env: Env): Promise<Response> {
        const url = new URL(request.url);
        const match = /^\/sandboxes\/([a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?)\/build$/.exec(url.pathname);

        if (!match || request.method !== "GET") {
            return new Response("Not found", { status: 404 });
        }

        const sandbox = env.MY_CONTAINER.getByName(match[1]);
        return sandbox.stream(build);
    },
} satisfies ExportedHandler<Env>;

示例命令每秒写入一行，向标准错误写入警告，并以代码 3 退出。请将其替换为你的构建或测试命令。Worker 选择命令，而不是从请求中读取命令。更多信息请参阅 Sandbox 安全。

部署你的 Worker：
npx wrangler deploy
yarn wrangler deploy
pnpm wrangler deploy

在名为 ada 的 sandbox 中流式运行命令。将示例 hostname 替换为 Wrangler 打印的 workers.dev URL：

curl https://<YOUR_WORKER>.<YOUR_SUBDOMAIN>.workers.dev/sandboxes/ada/build --no-buffer

事件大约每隔一秒到达：

event: stdout
data: "step 1\n"

event: stdout
data: "step 2\n"

event: stdout
data: "step 3\n"

event: stderr
data: "warning: almost done\n"

event: exit
data: {"exitCode":3}

在浏览器中读取流

每个事件都有类型和 JSON 编码的 data 值。stdout 或 stderr 事件可能包含一行的一部分或若干行，因此应追加文本，而不是将每个事件当作一行。exit 事件最后到达。以 : 开头的行是注释，客户端会忽略它们。

该路由响应 GET 请求，因此页面可以用 EventSource 读取它：

const output = document.querySelector("pre");
const events = new EventSource("/sandboxes/ada/build");

for (const type of ["stdout", "stderr"]) {
    events.addEventListener(type, (event) => {
        output.textContent += JSON.parse(event.data);
    });
}

events.addEventListener("exit", (event) => {
    output.textContent += `\nExit code ${JSON.parse(event.data).exitCode}\n`;
    events.close();
});

流结束时 EventSource 会重连，这会再次运行命令。页面在 exit 事件上调用 close() 以防止这种情况。

在时限后停止命令

断开连接只会通知由 exec() 启动的进程。命令启动的进程（例如 sh -c 脚本中的命令）会继续保持输出流打开，直到它们退出。如需在时间限制后停止命令及其启动的进程，请使用 GNU coreutils 的 timeout 运行命令：

const build = ["timeout", "--kill-after=5", "60", "sh", "-c", "npm test"];

运行超过 60 秒时，命令会以退出代码 124 结束；如果 timeout 也发送了 SIGKILL，则以 137 结束。转移到自己进程组的进程（例如通过 setsid 启动的进程）会继续运行。更多信息请参阅停止命令启动的进程。

流式命令输出

前提条件

流式运行命令

在浏览器中读取流

在时限后停止命令

相关资源