如何使用ES8异步/等待流？

Question

在/sf/answers/1306102941/中是如何使用内置加密库和流计算文件的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而不是如上所述使用回调,但是仍然保持使用流的效率吗？

Answer 1

async/ await仅适用于承诺,而不适用于流.有一些想法可以创建一个额外的流式数据类型,它可以获得自己的语法,但是如果有的话,这些都是高度实验的,我不会详细介绍.

无论如何,你的回调只等待流的结束,这非常适合承诺.你只需要包装流:

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

现在你可以等待这个承诺:

(async function() {
    let sha1sum = await end;
    console.log(sha1sum);
}());

Answer 2

2021 年更新：

节点文档中的新示例：

async function print(readable) {
  readable.setEncoding('utf8');
  let data = '';
  for await (const chunk of readable) {
    data += chunk;
  }
  console.log(data);
}

请参阅https://nodejs.org/api/stream.html#stream_read_symbol_asynciterator

Answer 3

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 最强大的特性之一。使其完全异步并不容易。现在我们有了。

Answer 4

如果使用的节点版本> = v10.0.0，则可以使用stream.pipeline和util.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);

Answer 5

像这样的工作：

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

Answer 6

我相信这对某人会有帮助：

async function readFile(filename) {
    let records = []
    return new Promise(resolve => {
        fs.createReadStream(filename)
            .on("data", (data) => {
                records.push(data);
            })
            .on("end", () => {
                resolve(records)
            });
    })
}