在读取所有数据之前异步/等待 createReadStream 结束：Node.js

Question

我正在使用fs.createReadStream 读取上传的 csv 文件数据。当收到数据行时，我正在验证这些数据并将验证数据推送到数组中。

问题是，它在调用end 后收到最后一行。

我已使用四行的 csv 进行波纹管测试。所有四条记录均通过验证。

const stream = fs.createReadStream(filePath)
    .pipe(csv.parse({ headers: true }))
    .on("error", (error) => {
        throw error.message;
    })
    .on("data", async (row) => {
        try {
            stream.pause();
            const mobile = Helper.validateMobile(row.Telephone)

            const validationerrors = await Helper.ValidateCustomer(null, mobile, row.code, "BULK");

            console.log(`validationerrors ---> ${validationerrors} Telephone--> ${row.Telephone}`)

            if (validationerrors.length) {
                throw validationerrors.message

            } else {
                csvData.push(row);
            }
        } finally {
            stream.resume();
        }
    })
    .on("end", async () => {
        console.log("ENDDD ---->", csvData)
    });

Console.log 输出是这样的。最后一行（777223478 行）在on("end") 调用后接收。

    validationerrors --->  Telephone--> 778786516
    validationerrors --->  Telephone--> 718254596 
    validationerrors --->  Telephone--> 712760763 
    
    ENDDD ----> [ 
       { 
        Code: 'CTLD000323',',   
        Telephone: '778786516' }, 
       { 
        Code: 'CTLD000324',  
        Telephone: '718254596' }, 
       { 
        Code: 'CTLD000376',    
        Telephone: '712760763' } 
      ] 
    
    validationerrors --->  Telephone--> 777223478

Answer 1

因此，CSV 库似乎不遵守 .pause() 方法，如果它有多行排队，它会继续并为它们触发 data 事件，即使流已暂停。我知道的两种解决方法是修改 csv 模块以尊重流暂停或在它们进入时排队，这样当你已经在处理某些东西时，你可以手动暂停它们的处理。而且，您还必须排队 end 事件，因为流将在您完成处理之前发送它。我通过合成我自己的finalEnd 事件来处理结束事件。这是一个队列的粗略实现，它缓冲在流暂停时到达的data 事件。

let paused = false;
const queue = [];
let end = false;

const stream = fs.createReadStream(filePath)
    .pipe(csv.parse({ headers: true }))
    .on("error", (error) => {
        throw error.message;
    })
    .on("data", async (row) => {
        queue.push(row);
        if (!paused) {
            stream.pause();
            paused = true;
            while (queue.length) {
                try {
                    await processRow(queue.shift());
                } catch (e) {
                    // decide what to do here if you get an error processing a row
                    console.log(e);
                }
            }
            paused = false;
            stream.resume();
            if (end) {
                stream.emit("finalEnd");
            }
        }

        async function processRow(row) {
            const mobile = Helper.validateMobile(row.Telephone)

            const validationerrors = await Helper.ValidateCustomer(null, mobile, row.code, "BULK");

            console.log(`validationerrors ---> ${validationerrors} Telephone--> ${row.Telephone}`)

            if (validationerrors.length) {
                throw validationerrors.message

            } else {
                csvData.push(row);
            }
        }

    })
    .on("end", async () => {
        end = true;
        if (!queue.length && !paused) {
            stream.emit("finalEnd");
        }
    }).on("finalEnd", () => {
        console.log("ENDDD ---->", csvData)
    });