在写入文件之前完成承诺

Question

我对 NodeJS 和 promises 非常陌生，在 google 搜索找到下面的代码并将其变为使用 fast-csv 读取 CSV 文件的 promise 之后，基于 packNr 创建一个数据数组。下面的代码按预期拆分文件，但我的问题是将数组写回新的 CSV 文件。不知何故，我需要在下一个开始之前完成每个 packNr 的编写。使用下面的代码，我只在代码完成之前获取每个文件的标题。

var readCSV = function(path, options) {
const datas = {}; // data['123'] = CSV data filtered for id = 123
return new Promise(function(resolve, reject) {

fastCsv
.parseFile(path, options)
.on('data', d => {
    if (!datas[d.packNr]) datas[d.packNr] = [];
    datas[d.packNr].push(d)        
})
.on('end', () => {

    Object.keys(data).forEach(packNr => {
        // For each ID, write a new CSV file
        fastCsv
        .write(datas[packNr],options)
        .pipe(fs.createWriteStream(`.\data-id-${packNr}.csv`))           
        })

  resolve();     
})
}) 

Answer 1

fastCsv 是一个流。 Stream 有自己的流程（如果我可以这么说的话），它们继承自 EventEmitters 构造函数，因此它们是事件发射器。这意味着将他们的流程包含在您的文件流中，您应该使用他们的事件。

当流连接关闭时，它会发出finish 和close 事件。所以你可以使用它来解决你的 Promise，就像这样

.on('end', () => {

    Object.keys(data).forEach(packNr => {
        // For each ID, write a new CSV file
        fastCsv
        .write(datas[packNr],options)
        .pipe(fs.createWriteStream(`.\data-id-${packNr}.csv`))        
   })
    
    fastCsv.on('finish', () => resolve())
    // fastCsv.on('close', () => resolve()) that should work as well 
       
})

另一个解决方案，实际上更好，因为它将处理流连接结束的所有事件（结束，完成，关闭，错误）是使用流模块中的管道方法

   // on the top of your file
   const { pipeline } = require('stream')

     .on('end', () => {
    
        Object.keys(data).forEach(packNr => {
            // For each ID, write a new CSV file
            pipeline(
               fastCsv.write(datas[packNr],options),
               fs.createWriteStream(`.\data-id-${packNr}.csv`),
               e => {
                    if(e) reject(e)
                    else resolve() 
               }           
            )
        })
     })
   

=====（编辑）最终解决方案=======

在我的逻辑中仍然是一个问题，因为在每次迭代 (forEach()) 时它都会创建一个新流。承诺将解决第一个结束流的结束，但不是在最后一个结束时。要修复它，您可以这样做

       const { pipeline } = require('stream')
    
         .on('end', () => {
        
            const dataObj = Object.keys(data)
            const lg = dataObj.length
            let i = 0
            dataObj.forEach(packNr => {
                // For each ID, write a new CSV file
                pipeline(
                   fastCsv.write(datas[packNr],options),
                   fs.createWriteStream(`.\data-id-${packNr}.csv`),
                   e => {
                        if(e) reject(e)
                        i++
                        if(i === lg) resolve()
                   }           
                )
            })
         })

Answer 2

我认为问题在于 fastCsv.write 本身是异步的，所以当然 resolve() 在完成之前正在执行。我相信答案是等到fastCsv.write() 完成后再解决。