如何在流中使用 ES8 async/await?

IT技术 javascript node.js async-await ecmascript-2017
2021-02-13 06:48:30

https://stackoverflow.com/a/18658613/779159 中是如何使用内置加密库和流计算文件 md5 的示例。

var fs = require('fs');
var crypto = require('crypto');

// the file you want to get the hash    
var fd = fs.createReadStream('/some/file/name.txt');
var hash = crypto.createHash('sha1');
hash.setEncoding('hex');

fd.on('end', function() {
    hash.end();
    console.log(hash.read()); // the desired sha1sum
});

// read all file and pipe it (write it) to the hash object
fd.pipe(hash);

但是是否可以将其转换为使用 ES8 async/await 而不是使用上面看到的回调,同时仍然保持使用流的效率?

6个回答

async/await仅适用于Promise,不适用于流。有一些想法可以创建一个额外的类似流的数据类型,以获得自己的语法,但这些都是高度实验性的,我不会详细介绍。

无论如何,您的回调只是在等待流结束,这非常适合Promise。你只需要包装流:

var fd = fs.createReadStream('/some/file/name.txt');
var hash = crypto.createHash('sha1');
hash.setEncoding('hex');
// read all file and pipe it (write it) to the hash object
fd.pipe(hash);

var end = new Promise(function(resolve, reject) {
    hash.on('end', () => resolve(hash.read()));
    fd.on('error', reject); // or something like that. might need to close `hash`
});

现在你可以等待那个Promise:

(async function() {
    let sha1sum = await end;
    console.log(sha1sum);
}());
@MiF 所以你的意思是答案是错误的,因为它不是“高度实验性的”而只是“实验性的”?:-D
2021-03-15 06:48:30
JS 转向在所有异步情况下使用 promises 和 async/await。太奇怪了,这个功能没有在生产中发布。我认为,它现在是一个很好的 sintacsys,不会改变。
2021-03-22 06:48:30
恭喜!现在这不是一个正确的答案,现在您可以将异步循环与流2ality.com/2018/04/async-iter-nodejs.html 一起使用不幸的是,它现在是一个实验性功能。
2021-03-24 06:48:30
你为什么要等待 ReadStream 到end在散列函数有机会对数据执行之前,Promise 不会解析吗?或者,至少,它不会造成竞争条件吗?我假设您希望将 的值存储fd.pipe(hash)在变量中并侦听finish事件,因为这将表明 WriteableStream(散列)已完成。
2021-04-07 06:48:30
@adam-beck 听起来你是对的。为什么我这样做?因为 OP 在他们的问题中也这样做了。随意提出编辑建议。
2021-04-09 06:48:30

如果您使用的节点版本 >= v10.0.0,那么您可以使用stream.pipelineutil.promisify

const fs = require('fs');
const crypto = require('crypto');
const util = require('util');
const stream = require('stream');

const pipeline = util.promisify(stream.pipeline);

const hash = crypto.createHash('sha1');
hash.setEncoding('hex');

async function run() {
  await pipeline(
    fs.createReadStream('/some/file/name.txt'),
    hash
  );
  console.log('Pipeline succeeded');
}

run().catch(console.error);
也应该在某处调用 unpipe 吗?
2021-03-17 06:48:30
但它在文档中。nodejs.org/api/…
2021-03-19 06:48:30
你是救命恩人
2021-03-25 06:48:30
我不得不挖掘几个小时,直到找到这个答案。为什么这不包含在 Node Streams 文档中作为示例......?
2021-04-09 06:48:30
这是一个更好的答案
2021-04-09 06:48:30

Node V15 现在在stream/promises 中有一个 promisfiy 管道这是最干净、最正式的方式。

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);

我们都应该感谢它在这里完成了多少工作:

  • 捕获所有流中的错误。
  • 出现错误时销毁未完成的流。
  • 仅在最后一个可写流完成时返回

这个管道是 Node.JS 最强大的特性之一。使其完全异步并不容易。现在我们有了。

像这样的工作:

for (var res of fetchResponses){ //node-fetch package responses
    const dest = fs.createWriteStream(filePath,{flags:'a'});
    totalBytes += Number(res.headers.get('content-length'));
    await new Promise((resolve, reject) => {
        res.body.pipe(dest);
        res.body.on("error", (err) => {
            reject(err);
        });
        dest.on("finish", function() {
            resolve();
        });
    });         
}
很好的解释,我使用这个Promise来包装我使用 https module中内置的 NodeJS 发出的 https 请求。谢谢!
2021-04-11 06:48:30

我会发表评论,但没有足够的声誉。

警告:如果您的应用程序正在传递流并执行异步/等待,请在等待之前非常小心地连接所有管道。您最终可能会得到不包含您认为它们所做的事情的流。这是最小的例子

const { PassThrough } = require('stream');

async function main() {
    const initialStream = new PassThrough();

    const otherStream = new PassThrough();
    const data = [];
    otherStream.on('data', dat => data.push(dat));
    const resultOtherStreamPromise = new Promise(resolve => otherStream.on('end', () => { resolve(Buffer.concat(data)) }));

    const yetAnotherStream = new PassThrough();
    const data2 = [];
    yetAnotherStream.on('data', dat => data2.push(dat));
    const resultYetAnotherStreamPromise = new Promise(resolve => yetAnotherStream.on('end', () => { resolve(Buffer.concat(data2)) }));

    initialStream.pipe(otherStream);
    initialStream.write('some ');

    await Promise.resolve(); // Completely unrelated await

    initialStream.pipe(yetAnotherStream);
    initialStream.end('data');
    const [resultOtherStream, resultYetAnotherStream] = await Promise.all([
        resultOtherStreamPromise,
        resultYetAnotherStreamPromise,
    ]);

    console.log('other stream:', resultOtherStream.toString()); // other stream: some data
    console.log('yet another stream:', resultYetAnotherStream.toString()); // yet another stream: data
}
main();