Node CSV pull parser

I solved this myself by creating my own Writable and piping the CSV parser to it. My write method does its stuff and wraps a promise to the node callback passed to _write() (here implemented using Q.nodeify):

class CsvConsumer extends stream.Writable {
    _write(data, encoding, cb) {
        console.log('Got data: ', data);

        Q.delay(1000).then(() => {
            console.log('Waited 1 s');
        }).nodeify(cb);
    }
}

csvtojson()
    .fromStream(is)
    .pipe(new CsvConsumer())
    .on('finish', err => {
        if (err) {
            console.log('Error!');
        } else {
            console.log('Done!');
        }
    });

This will process lines one by one:

Got data: {"a": "1"}
Waited 1 s
Got data: {"a": "2"}
Waited 1 s
Got data: {"a": "3"}
Waited 1 s
Done!

If you want to process each line asynchronously you can do that with node's native LineReader.

const lineStream = readline.createInterface({
  input: fs.createReadStream('data/test.csv'),
});

lineStream.on('line', (eachLine) =>{
    //process each line
});

If you want to do the same in synchronous fashion you can use line-by-line. It doesn't buffer the entire file into memory. It provides event handlers to pause and resume the 'line' emit event.

lr.on('line', function (line) {
    // pause emitting of lines...
    lr.pause();

    // ...do your asynchronous line processing..
    setTimeout(function () {

        // ...and continue emitting lines. (1 sec delay)
        lr.resume();
    }, 1000);
});