如何通过 Node.js 中的流管道等待异步值？

Question

[免责声明：这是我对 Node 的第一次尝试（我主要是一个 Clojure 人）]

我正在使用node-csv解析和转换 CSV 文件。转换发生在线路上的 IO 上，我用 async/await 构造包装了回调。

const in_stream  = fs.createReadStream('test-data')
const out_stream = fs.createWriteStream('test-output')

const parse = csv.parse({ delimiter: "\t", quote: false })

const transform = csv.transform(async (row) => {
    await translate(row[1], { from: 'en', to: 'fr' })
        .then(res => { row[1] = res.text})
    console.log(row) // this shows that I succesfully wait for and get the values
    return row
})

const stringify = csv.stringify({
    delimiter: ';',
    quoted: true 
})

in_stream.pipe(parse).pipe(transform).pipe(stringify).pipe(out_stream)

看起来流在值从转换器中输出之前就结束了。

你如何处理 Node.js 中流中的延迟值？我显然弄错了……

（如果有帮助，我可以提供一个虚拟的 CSV 文件）

Answer 1

问题是你的transform功能

const transform = csv.transform(async (row) => {
    await translate(row[1], { from: 'en', to: 'fr' })
        .then(res => { row[1] = res.text})
    console.log(row) // this shows that I succesfully wait for and get the values
    return row
})

您在这里所做的是假设async可以直接使用而没有任何影响。问题是因为您实际上并未在预期的回调中从异步函数返回任何内容，传递给后续函数的内容什么都不是

修复很简单，在回调函数中传回数据

const transform = csv.transform(async (row, done) => {
    await translate(row[1], { from: 'en', to: 'fr' })
        .then(res => { row[1] = res.text})
    console.log(row) // this shows that I succesfully wait for and get the values
    done(null, row)
})

看下面的网址

https://csv.js.org/transform/options/

结果

$ node index.js && cat test-output
[ '7228', 'Pot de café en acier inoxydable', '17.26' ]
[ '5010',
  'Set de 4 bidons avec couvercle PS (acier inoxydable)',
  '19.92' ]
[ '7229', 'Cafetière pour 6 tasses (acier inoxydable)', '19.07' ]
"7228";"Pot de café en acier inoxydable";"17.26"
"5010";"Set de 4 bidons avec couvercle PS (acier inoxydable)";"19.92"
"7229";"Cafetière pour 6 tasses (acier inoxydable)";"19.07"