使用busboy+AWS上传文件时遇req.pipe is not a function错误求助

错误原因 #

你遇到的req.pipe is not a function错误，核心原因有两点：

• 请求已被文件解析中间件（比如express-fileupload）处理过，此时req不再是原始可读流对象，原生的pipe方法已被移除。
• 代码逻辑冲突：既试图用busboy处理原始请求流，又依赖已被解析的req.files，这两种处理方式无法同时生效。

修复步骤 #

1. 禁用路由级文件解析中间件：确保该上传路由未被express-fileupload、multer这类提前解析文件的中间件拦截，让req保持原始可读流状态。
2. 重构上传逻辑：直接用原始req与busboy交互，在busboy的事件回调里处理文件流和表单字段，不再依赖req.files、req.body。
3. 用Promise包装busboy流程：busboy基于事件驱动，需要将其包装为Promise才能配合async/await语法完成异步流程控制。

修改后的完整代码 #

const uploadDeArquivos = async (req) => {
    return new Promise((resolve, reject) => {
        const bb = busboy({ headers: req.headers });
        const formData = {};
        let arquivos = [];
        let errors = [];

        // 收集表单非文件字段
        bb.on('field', (name, value) => {
            formData[name] = value;
        });

        // 处理每个上传的文件
        bb.on('file', async (name, file, info) => {
            try {
                const { filename } = info;
                const splitName = filename.split('.');
                const fileName = `${new Date().getTime()}.${splitName.at(-1)}`;
                const path = `servicos/${formData.salaoId}/${fileName}`;

                // 直接将文件流上传至S3
                const response = await aws.uploadToS3(file, path);
                response.error 
                    ? errors.push({ error: true, message: response.message })
                    : arquivos.push(path);
            } catch (err) {
                errors.push({ error: true, message: err.message });
            }
        });

        // 所有文件处理完成后的逻辑
        bb.on('finish', async () => {
            try {
                if (errors.length > 0) throw errorThrow(500, errors[0]);

                // 创建Servico记录
                const jsonServico = JSON.parse(formData.servico);
                const novoServico = await Servico(jsonServico).save();

                // 创建Arquivo关联记录
                const arquivosParaSalvar = arquivos.map(arquivo => ({
                    referenciaId: novoServico._id,
                    model: 'Servico',
                    caminho: arquivo,
                }));
                await Arquivo.insertMany(arquivosParaSalvar);

                resolve({ servico: novoServico, arquivos });
            } catch (err) {
                reject(err);
            }
        });

        // 捕获busboy处理过程中的错误
        bb.on('error', (err) => {
            reject(err);
        });

        // 将原始请求流导向busboy
        req.pipe(bb);
    });
};

const newServico = async (req, res, next) => {
    try{
        const resultado = await uploadDeArquivos(req);
        return res.status(201).json(resultado);
    } catch(error){
        next(error);
    }
};

// 确保该路由未被文件解析中间件处理
router.post('/', newServico);

关键说明 #

• 用Promise包裹busboy的事件流程，保证async/await能正确等待整个上传+数据存储流程完成。
• 通过bb.on('field')获取表单中的文本字段，替代原代码中直接传入的req.body。
• 通过bb.on('file')直接获取文件流并上传至S3，避免依赖已被解析的req.files。

内容的提问来源于stack exchange，提问作者Lucas Junqueira