如何使用 fetch 处理流数据？

Question

我在使用 node.js 处理来自进程的输出流方面使用 async 取得了巨大成功，但我正在努力获得一些我希望可以“使用”浏览器fetchAPI 的东西。

这非常适合异步处理来自进程的输出流：

for await (const out of proc.child.stdout) {
  ...
}

（当然在异步函数上下文中）

我试图在浏览器中做类似的事情，我想在数据从服务器发送给我的同时访问数据。

for await (const chunk of (await fetch('/data.jsonl')).body) {
  console.log('got', chunk);
}

这在 Chrome ( Uncaught TypeError: (intermediate value).body is not async iterable) 中不起作用。

对于我的用例，这不是必需的，所以我现在只是let data = await (await fetch(datapath)).text();在我的客户端代码中使用。这类似于在等待的获取上使用.json()而不是的典型用法.text()，因此在浏览器接收到整个响应之前不能开始处理。由于显而易见的原因，这并不理想。

我在看Oboe.js（我认为相关的 impl 就在这里附近），它几乎处理了这个问题，但它的内部结构相当丑陋，所以看起来这可能是目前唯一的方法？

如果没有实现异步迭代（意味着 async for 还不能使用）是不是还有另一种以实用方式使用 ReadableStream 的方法？

Answer 1

不幸的是，异步可迭代支持尚未实现，尽管在规范中。相反，您可以手动迭代，如规范中的此示例所示。（我将在此答案中将示例转换为异步/等待。）

const reader = response.body.getReader();
const { value, done } = await reader.read();

if (done) {
  console.log("The stream was already closed!");
} else {
  console.log(value);
}

您可以使用递归或循环来重复执行此操作，如另一个示例所示：

async function readAllChunks(readableStream) {
  const reader = readableStream.getReader();
  const chunks = [];
  
  let done, value;
  while (!done) {
    ({ value, done } = await reader.read());
    if (done) {
      return chunks;
    }
    chunks.push(value);
  }
}

console.log(await readAllChunks(response.body));

Answer 2

根据规范，ReadableStream诸如 fetch-API 之类的方法Response.body确实有一个getIteratormethod。由于某种原因，它本身不是异步可迭代的，您必须显式调用该方法：

const response = await fetch('/data.json');
if (!response.ok)
    throw new Error(await response.text());
for await (const chunk of response.body.getIterator()) {
    console.log('got', chunk);
}