'Unshift' in transform stream

[HungryHippo]

I'm having a problem with streams2 transform. I have a stream which reads from a file and transforms the csv data to json. the transform works fine but at present if I have an incomplete chunk of data my 'unshift' command moves the data to my transform output rather than back to my transform input ! 

Transform code :

 

var Transform = require('stream').Transform,

    util = require('util')

module.exports = function (columns, opts, cb) { return new CSVStream(columns, opts, cb) }

module.exports.CSVStream = CSVStream

function CSVStream (columns, opts, cb) {

  opts = opts || {}

  Transform.call(this, columns, opts)

  this.delimiter = opts.delimiter || ','

  this.newline = opts.newline || '\n'

  this.columns = columns;

}

util.inherits(CSVStream, Transform)

CSVStream.prototype._transform = function (chunk, encoding, done) 

{

  try {

    this._parse(chunk)

    done()

  } catch (err) {

    done(err)

  }

}

CSVStream.prototype._parse = function (data) 

{  

  console.log('_parse start');

  var i=0,j=0;

  var fieldsArray;

  var output={};

  var rowsArray = data.toString().split(this.newline);//split entire buffer into rows

  for (i=0;i<rowsArray.length;i++){

    fieldsArray = rowsArray[i].toString().split(this.delimiter);//split each row into fields

    //console.log('row:'+rowsArray[i]);

    if (fieldsArray.length<this.columns.length){//check we have an entire row

      console.log('unshift:'+ rowsArray[i]);

      this.unshift(rowsArray[i]);//push incomplete data back onto the beginning of the next buffer to arrive

    }

    else{

      for (j=0;j<this.columns.length;j++){

        key = this.columns[j];

        output[key] = fieldsArray[j];

      }

      //console.log('push');

      //console.log(JSON.stringify(output));

      this.push(JSON.stringify(output));

    }    

  }

  console.log('_parse end');

}

CSVStream.prototype.end = function (buf) {

  Transform.prototype.end.call(this, buf)

}

Stream/pipe code

var options = {};

options.delimiter = '\t';

 

var columns = ['geonameid','name','asciiname','alternatenames','latitude','longitude','feature class','feature    code','country code','cc2','admin1 code','admin2 code','admin3 code','admin4 code','population','elevation','demmeters','timezone','modification date'];

var fstream = fs.createReadStream(source),

                   parser = csv(columns, options);

        parser.on('readable', function () {
            var line = parser.read();
            console.log('line:'+line.toString());
        })

        parser.on('end', function () {            
            console.log('end');
        })
        // now pump some data into it (and pipe it somewhere else)

        fstream.pipe(parser);