node.js - Node.js Stream API 的意外行为

Question

我尝试通过解析器通过管道传输 CSV 文件，删除给定 ID 的行，将新数组通过管道传输到字符串生成器，最后将结果输出到新的 CSV 中。

问题是新数组没有正确返回。我的意思是，我设法删除了该行，但没有返回新的 CSV。

这是CSV file

这是代码:

'use strict';


const Fs    = require('fs');
const Csv   = require('csv');


let input  = 'data_stack.csv';
let output = 'data_output.csv';


let readStream  = Fs.createReadStream(input);
let writeStream = Fs.createWriteStream(output);

let opt  = {delimiter: ',', quote: '"', escape: '"', relax: true, skip_empty_lines: true};

function deleteRow (id) {

    return Csv.parse(opt, (err, data) => {
        if (err) throw err;

        let delId = [];
        // The row corresponding to the id is pushed to an array to console.log it. The other rows are returned 
        let newData = data.filter(line => {
            if (line[0] === id) {
                delId.push(line);
            }
            else {
                return line;
            }
        });

        if (delId !== 0) {
            console.log("Deleted line\n");
            console.log(delId);
        }

        // Testing if the row was deleted
        let result = [];
        // If the row is not deleted, push it into the result array
        newData.forEach(line => {
            if (line[0] === id) {
                result.push(line);
            }
        });
        // "Not found !" is console logged. The row has been deleted 
        if (result.length !== 0) {
            console.log(result);
        }
        else {
            console.log("Not found !");
        }

        return newData;
    });
}

readStream.pipe(deleteRow('566150121ae803f53751c3f2'))

此代码有效 但是，如果我选择返回 newData 数组以将其传递给字符串化器，然后将结果输出到 CSV 文件，则会出现 2 个意外行为:

1. delId 数组现在为空，我不知道为什么
2. 输出文件已创建，但之前删除的行仍然存在于输出文件中

这是代码:

'use strict';


const Fs    = require('fs');
const Csv   = require('csv');


let input  = 'data_stack.csv';
let output = 'data_output.csv';


let readStream  = Fs.createReadStream(input);
let writeStream = Fs.createWriteStream(output);

let opt  = {delimiter: ',', quote: '"', escape: '"', relax: true, skip_empty_lines: true};

function deleteRow (id) {

    return Csv.parse(opt, (err, data) => {
        if (err) throw err;

        let delId = [];
        // The row corresponding to the id is pushed to an array to console.log it. The other rows are returned 
        let newData = data.filter(line => {
            if (line[0] === id) {
                delId.push(line);
            }
            else {
                return line;
            }
        });

        if (delId !== 0) {
            console.log("Deleted line\n");
            console.log(delId);
        }

        // Testing if the row was deleted
        let result = [];
        // If the row is not deleted, push it into the result array
        newData.forEach(line => {
            if (line[0] === id) {
                result.push(line);
            }
        });
        // "Not found !" is console logged. The row has been deleted 
        if (result.length !== 0) {
            console.log(result);
        }
        else {
            console.log("Not found !");
        }

        return newData;
    });
}

let stringifier = Csv.stringify({quoted: true});

readStream.pipe(deleteRow('566150121ae803f53751c3f2')).pipe(stringifier).pipe(writeStream);

我不太明白通过管道传输来自deleteRow 函数的结果如何影响其行为。

有什么建议吗？

Answer 1

csv.parse 有 2 APIs :

1. 基于回调(缓冲区)；
2. 基于流。

如果您传递回调函数，就像您在示例中所做的那样，您将订阅回调 API，这意味着您可以通过 data 参数获得完整的解析数据。

但是，该回调是一个流“接收器”，这意味着对数据的任何修改都不会写入您的 writeStream。

您需要调整代码以使用回调 API，如下所示:

function deleteRow(readStream, id, done) {
  readStream.pipe(Csv.parse(opt, (err, data) => {
    if (err) return done(err);

    // Wrire new data to file
    Csv.stringify(data.filter(line => line[0] !== id))
      .pipe(writeStream)
      .on('end', done);
  }));
}

deleteRow(readStream, '5660d7e11ae803f53751aeb0', function (err) {
  console.log('done!');
});

或者，像这样使用流 API: