code: function iterateStream(readStream, task, pipeline, done) {
    let tail = '';
  
    readStream.on('data', function (chunk) {
      const str = tail + chunk;
      let lines = str.split(/\r\n|\r|\n/);
      tail = lines.pop();
      for (let t = 0; t < pipeline.length; t++)
        lines = pipeline[t].transform(lines, task);
    });
  
    readStream.once('end', function () {
      if (tail) {
        let buffer = [tail];
        for (let t = 0; t < pipeline.length; t++)
          buffer = pipeline[t].transform(buffer, task);
      }
      done();
    });
  
    readStream.on('error', function (err) {
      log('iterateStream stream error:', err);
    });
  }
docs: Complete download from azure to local, then process. /*
  function azureDownload(az, container, name, task, pipeline, done) {
  var filename = task.basedir + 'import/' + name.replace(/\//g, '-');
  var gz = filename.slice(-3) === '.gz';
  var delay = 1000, retry = 5;
  task.log('getBlob', name);
  
  function getBlob() {
  az.getBlobToLocalFile(container, name, filename, function (err) {
  task.log('getBlob', name, 'error:', err);
  if (err) {
  delay *= 2;
  if (!retry) throw new Error(err);
  task.log('retry getBlob', name, 'in', delay, 'ms');
  return setTimeout(getBlob, delay);
  }
  var stream = task.lib.fs.createReadStream(filename);
  if (gz) stream = stream.pipe(task.lib.zlib.createGunzip({chunkSize: 65536}));
  iterateStream(stream, task, pipeline, done);
  });
  }
  getBlob();
  }