node.js - 上传数据时暂停 Node csv-parser

Question

我正在使用 Node csv-parser我的读取流来自 std in。我将其通过管道传输到解析器。在解析器可读函数内，我正在进行异步 HTTP 调用来上传此数据。在此期间，我希望解析器暂停读取。直到异步调用完成执行。

var parse = require('csv-parse');
  var output = [];
// Create the parser
  var parser = parse({delimiter: ',', columns: true, trim: true});
  
  parser.on('readable', function () {
    while (record = parser.read()) {
      console.log('Still reading');
      var jsonRec = convertIpToInt(record);
      var jsonData = JSON.stringify(jsonRec);
      output.push(new Buffer(jsonData + "\n"));
      //parser.pause();
      var assertPromise = uploadBatches1(indexName, function() {
          //parser.resume();
          console.log('Returned');
      });
     }
  });

  function uploadBatches1(indexToAssert, cb) {
    //uploads data got form parser
  }

Answer 1

由于 csv-parse 公开了可读流，因此您可以使用 parser.pause() 暂停流。

同样，更新完成后，您可以调用parser.resume() 。您还可以使用parser.isPaused()检查是否需要提前恢复。 .

您可以在此处阅读有关可读流以及暂停方法的作用的更多信息: https://nodejs.org/api/stream.html#stream_readable_pause

根据上述内容，您应该实现 parser.on('data')可读流事件，而不是 while (record = parser.read())环形。如果您自己主动调用 read 方法，则暂停流不会执行任何操作。

考虑像这样重组代码，然后使用文件系统中的 parser.pipe 读取:

var parse = require('csv-parse');
var output = [];
// Create the parser
var parser = parse({delimiter: ',', columns: true, trim: true});

parser.on('data', function () {
  console.log('Still reading');
  var jsonRec = convertIpToInt(record);
  var jsonData = JSON.stringify(jsonRec);
  output.push(new Buffer(jsonData + "\n"));
  parser.pause();
  var assertPromise = uploadBatches1(indexName, function() {
      parser.resume();
      console.log('Returned');
  });
});

function uploadBatches1(indexToAssert, cb) {
  //uploads data got form parser
}