在没有回调的情况下使用 Node.js 流

Posted 2023-03-07

【中文标题】在没有回调的情况下使用 Node.js 流

【英文标题】：Working with Node.js streams without callbacks

【发布时间】：2021-09-13 00:53:49

【问题描述】：

要将 PDF 文件从 Node.js 服务器发送到客户端，我使用以下代码：

const pdf = printer.createPdfKitDocument(docDefinition);

const chunks = [];

pdf.on("data", (chunk) => 
    chunks.push(chunk);
);

pdf.on("end", () => 
    const pdfBuffered = `data:application/pdf;base64, $Buffer.concat(chunks).toString("base64")`;
    res.setHeader("Content-Type", "application/pdf");
    res.setHeader("Content-Length", pdfBuffered.length);
    res.send(pdfBuffered);
);

pdf.end();

一切正常，唯一的问题是这里的流使用的是回调方法而不是async/await。

我找到了possible solution：

const  pipeline  = require("stream/promises");

async function run() 
    await pipeline(
        fs.createReadStream('archive.tar'),
        zlib.createGzip(),
        fs.createWriteStream('archive.tar.gz')
    );

    console.log('Pipeline succeeded.');


run().catch(console.error);

但我不知道如何将初始代码采用stream/promises 的代码。

【参考方案1】：

您可以像这样手动将 PDF 代码包装在一个 Promise 中，然后将其用作返回 Promise 的函数：

function sendPDF(docDefinition) 
    return new Promise((resolve, reject) => 
        const pdf = printer.createPdfKitDocument(docDefinition);

        const chunks = [];

        pdf.on("data", (chunk) => 
            chunks.push(chunk);
        );

        pdf.on("end", () => 
            const pdfBuffered =
                `data:application/pdf;base64, $Buffer.concat(chunks).toString("base64")`;
            resolve(pdfBuffered);
        );

        pdf.on("error", reject);

        pdf.end();
    );


sendPDF(docDefinition).then(pdfBuffer => 
    res.setHeader("Content-Type", "application/pdf");
    res.setHeader("Content-Length", pdfBuffer.length);
    res.send(pdfBuffer);
).catch(err => 
    console.log(err);
    res.sendStatus(500);
);

因为有很多data 事件，你不能只承诺数据部分。您仍然需要监听每个 data 事件并收集数据。

【参考方案2】：

如果回调打算只执行一次，则只能将回调 API 转换为 async/await。

您在网上找到的那个有效，因为您只是在等待整个流完成，然后回调运行一次。你得到的是在每个传入的数据块上执行多次的回调。

您可以查看其他资源以使流更易于使用，例如 RXJS 或 this 即将提出的向语言添加可观察对象的 ECMAScript 提案。这两者都旨在处理回调可以执行多次的情况——这是async/await 无法做到的。

【讨论】：

感谢您的参考，我会检查它！