如何在NodeJS中将ReadStream转换为ReadableStream？

这个问题与将 ReadableStream 转换为 ReadStream 完全相反。

• ReadStream

  是Node.js中使用的结构

• ReadableStream

  是网络平台带来的结构

随着非 Node.js 运行时（例如 Deno 或 Next.js 中的“Edge 运行时”）的出现，将 Node.js 特定的

ReadStream

ReadableStream

这对于从 Next.js 路由处理程序发送文件很有用，请参阅 Next.js GitHub 上的讨论。

我起草了一段代码，如下所示：

const downloadStream = fs.createReadStream(zipFilePath);
    const readStream = new ReadableStream({
      start(controller) {
        return pump();
        function pump() {
          return downloadStream.read().then(({ done, value }) => {
            // When no more data needs to be consumed, close the stream
            if (done) {
              controller.close();
              return;
            }
            // Enqueue the next data chunk into our target stream
            controller.enqueue(value);
            return pump();
          });
        }
      },
    });

我正在测试它。

编辑：第一稿的问题是

stream.Readable

read()

这是第二次尝试：

    const downloadStream = fs.createReadStream(zipFilePath);
    const readStream = new ReadableStream({
      start(controller) {
        return pump();
        function pump() {
          const buf = downloadStream.read() as Buffer
          if (buf === null) {
            controller.close();
            return;
          }
          controller.enqueue(buf.toString());
          return pump();
        }
      },
    });

尽管文件大小为 344 字节，但它立即为我提供了一个空缓冲区。当我拨打

isPaused()

pause()

read()

我还从 Next.js 收到一个奇怪的错误：

- error Error: aborted
    at connResetException (node:internal/errors:711:14)
    at Socket.socketCloseListener (node:_http_client:454:19)
    at Socket.emit (node:events:525:35)
    at TCP.<anonymous> (node:net:313:12) {
  code: 'ECONNRESET'
}

有更简单的语法解决方案吗？

设法找到了有效的语法，但仍然缺乏一些细节。

1. 我们希望能够使用命令式语法读取文件，而不是依赖传统的“数据”事件。

/**
 * From https://github/MattMorgis/async-stream-generator
 */
async function* nodeStreamToIterator(stream) {
    for await (const chunk of stream) {
        yield chunk;
    }
}

我对发电机不熟悉，所以我不确定：

• 为什么我们可以将“of”运算符应用于 Node.js

  ReadableStream

  ？
• 这里

  for await

  是什么意思？

但至少这个语法让我们可以循环使用流。

2. 现在我们想将迭代器转换为Web平台

   ReadStream

   。

/**
 * Taken from Next.js doc
 * https://nextjs/docs/app/building-your-application/routing/router-handlers#streaming
 * Itself taken from mozilla doc
 * https://developer.mozilla/en-US/docs/Web/API/ReadableStream#convert_async_iterator_to_stream
 * @param {*} iterator 
 * @returns {ReadableStream}
 */
function iteratorToStream(iterator) {
    return new ReadableStream({
        async pull(controller) {
            const { value, done } = await iterator.next()

            if (done) {
                controller.close()
            } else {

                controller.enqueue(new Uint8Array(value))
            }
        },
    })
}

注意“Uint8Array”：这似乎并不是所有场景都需要，但在某些平台上可能需要编码，我在 Next.js 中需要这种转换。请参阅 Next.js github 上的讨论。

最后我们可以在

Response

// highWaterMark affects the chunk size, here I use a small size to simulate many chunks
const nodeStream = fs.createReadStream("./.gitignore", { highWaterMark: 8 })
const iterator = nodeStreamToIterator(nodeStream)
const webStream = iteratorToStream(iterator)

const res = new Response(webStream)
const blob = await res.blob()
console.log(await blob.text())