如何无需中间缓冲区将Busboy文件流附加到请求体？

问题 #

我有一个使用busboy读取表单数据的Fastify插件，相关代码如下：

const bus = busboy({ headers: message.headers, limits: options });
bus.on("file", (fieldName: string, file: Readable, fileInfo: FileInfo) => {
    const chunks: Array<Uint8Array> = [];
    const fileObject = new File(fileInfo);
    fileObject.fieldName = fieldName;
    file.on("data", chunk => chunks.push(chunk));
    file.on("close", () => {
        fileObject.data = Readable.from(Buffer.concat(chunks));
        fileList.push(fileObject);
        formData[fieldName] = JSON.stringify(fileInfo);
    });
});

请问能否跳过file.on("data", chunk => chunks.push(chunk));和Readable.from(Buffer.concat(chunks))步骤，直接将file对象赋值给fileObject.data？我尝试过使用pipe等方法，但除非监听data事件，否则close事件不会触发，恳请解答！

解答 #

可以直接将file对象赋值给fileObject.data，核心问题是要让Busboy的文件流从暂停状态转为流动状态，这样才能触发close事件。以下是两种可行的实现方式：

方式1：调用resume()触发流流动 #

直接复用原文件流，调用resume()让流自动消费数据，即可正常触发close事件：

bus.on("file", (fieldName: string, file: Readable, fileInfo: FileInfo) => {
    const fileObject = new File(fileInfo);
    fileObject.fieldName = fieldName;
    // 直接赋值原可读流
    fileObject.data = file;
    // 启动流的流动，触发后续close事件
    file.resume();
    file.on("close", () => {
        fileList.push(fileObject);
        formData[fieldName] = JSON.stringify(fileInfo);
    });
});

注意：原可读流只能被消费一次，如果后续需要多次读取该流，你还是需要缓存chunk（即你最初的实现），或者使用stream-cache这类库给流添加缓存层。

方式2：通过空可写流消费数据 #

如果担心resume()可能导致数据丢失，也可以将流pipe到一个空的可写流中，同样能让流流动并触发close：

import { Writable } from 'stream';

bus.on("file", (fieldName: string, file: Readable, fileInfo: FileInfo) => {
    const fileObject = new File(fileInfo);
    fileObject.fieldName = fieldName;
    fileObject.data = file;
    // pipe到空可写流，消费流数据
    file.pipe(Writable({ write: () => {} }));
    file.on("close", () => {
        fileList.push(fileObject);
        formData[fieldName] = JSON.stringify(fileInfo);
    });
});

核心原理 #

Busboy生成的文件可读流默认处于暂停状态，只有当满足以下任一条件时，流才会开始推送数据：

• 绑定了data事件监听
• 调用了resume()方法
• 被pipe到某个可写流
  只有流完成数据推送后，才会触发close事件。你之前直接赋值但没触发close，就是因为流一直处于暂停状态，没有被消费。

内容的提问来源于stack exchange，提问作者Mr. X