如何将 ES8 async/await 与流一起使用?
How to use ES8 async/await with streams?
在 中有一个示例,说明如何使用内置加密库和流计算文件的 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 而不是使用上面看到的回调,同时仍然保持使用流的效率?
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);
}());
如果您使用的节点版本 >= v10.0.0 那么您可以使用 stream.pipeline and 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);
像这样的东西有效:
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();
});
});
}
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);
我们都应该感谢它在这里所做的工作:
- 捕获所有流中的错误。
- 出现错误时销毁未完成的流。
- 只有 return 当最后一个可写流 完成 .
这个管道是 Node.JS 最强大的功能之一。使其完全异步并不容易。现在我们有了。
我会评论,但没有足够的声誉。
注意事项:
如果您有一个正在传递流并执行 async/await 的应用程序,请在等待之前非常小心地连接所有管道。您最终可能会得到不包含您所认为的内容的流。这是最小的例子
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();
2021 年更新:
Node 文档中的新示例:
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_readable_symbol_asynciterator
在 中有一个示例,说明如何使用内置加密库和流计算文件的 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 而不是使用上面看到的回调,同时仍然保持使用流的效率?
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);
}());
如果您使用的节点版本 >= v10.0.0 那么您可以使用 stream.pipeline and 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);
像这样的东西有效:
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();
});
});
}
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);
我们都应该感谢它在这里所做的工作:
- 捕获所有流中的错误。
- 出现错误时销毁未完成的流。
- 只有 return 当最后一个可写流 完成 .
这个管道是 Node.JS 最强大的功能之一。使其完全异步并不容易。现在我们有了。
我会评论,但没有足够的声誉。
注意事项: 如果您有一个正在传递流并执行 async/await 的应用程序,请在等待之前非常小心地连接所有管道。您最终可能会得到不包含您所认为的内容的流。这是最小的例子
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();
2021 年更新:
Node 文档中的新示例:
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_readable_symbol_asynciterator