How to pipe awaited async values through streams in Node.js?

Question

[Disclaimer: This is my very first shot at Node (I am mostly a Clojure guy)]

I am parsing and transforming a CSV file with node-csv. The transform happens with IO over the wire and I wrapped the callback with a async/await construct.

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)

It appears that the stream ends before the values are piped out of the transformer.

How do you deal with delayed values in streams in Node.js? I am obviously getting it wrong…

(I can provide a dummy CSV file if it helps)

Answer 1

The issue is your transform function

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

What you are doing here is assuming the async can be just used without any implications. The issue is because you don't actually return anything from your async function in the expected callback, what gets passed to later functions is nothing

The fix is simple, pass the data back in the callback function

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

See the url below

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

The result

$ 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"