estoy usandofast-csv iterar sobre un archivo CSV usando unstream. Para cada fila del archivo CSV, quiero crear un trabajo en redis, para lo cual usokue. Analizar una fila es una función síncrona. Todo se ve así:

var csvStream = fastCsv(_config.csvOptions)
  .on('data', function(data) {
    var stream = this;
    stream.pause();

    var payload = parseRow(data, _config);
    console.log(payload); // the payload is always printed to the console

    var job = kue.create('csv-row', {
        payload: payload
      })
      .save(function(err) {
        if (!err) console.log('Enqueued at ' + job.id);
        else console.log('Redis error ' + JSON.stringify(err));

        stream.resume();
      });
  })
  .on('end', function() {
    callback(); // this ends my parsing
  });

Lo simpleconsole.log(payload); se muestra para cada fila de mi archivo CSV, sin embargo, el trabajo no se crea. Es decir, ninguna de las salidas en la devolución de llamada desave se imprimen y el trabajo no está en mi redis.

Asumo, porque es la última fila del archivo CSV, la transmisión ya emiteend y así el últimokue.create() no se puede ejecutar antes de que finalice el proceso?

¿Hay alguna manera de detener la corrienteend hasta que termine kue?