script for adapting node-csv to allow .pipe()

node-csv-parser-pipeable.js

var streams = require('stream');
var events = require('events');
var util = require('util');
var csv = require('csv');

var InputStream = function () {};
util.inherits(InputStream, streams.Stream);

var OutputStream = function () {};
util.inherits(OutputStream, events.EventEmitter);
OutputStream.prototype.write = function (data) {};
OutputStream.prototype.end = function () {};
OutputStream.prototype.destroy = function () {};

var MiddleStream = function () {
  this.writable = true;
  this.readable = true;
};
util.inherits(MiddleStream, streams.Stream);
MiddleStream.prototype.write = function (data) {};
MiddleStream.prototype.end = function () {};
MiddleStream.prototype.destroy = function () {};


//main function
createCsvParser = function (opts) {
  var input = new InputStream(); // Readable Stream
  var output = new OutputStream(); // Writable Stream
  var middle = new MiddleStream(); // Readable & Writable Stream, 

  // middle.write, triggers input.on('data')
  middle.write = function (data) {
    input.emit('data', data);
    MiddleStream.prototype.write.call(this, data);
  };
  middle.end = function () {
    input.emit('end');
    MiddleStream.prototype.end.call(this);
  };
  middle.destroy = function () {
    input.emit('close');
    MiddleStream.prototype.destroy.call(this);
  };

var stream = csv().fromStream(input, opts)
  .transform( function (data, index) {
    return opts.transform(data,index);
  })
  .on('data', function (data) {
    middle.emit('data', data);
  })
  .on('error', function (error) {
    middle.emit('error', error);
  })
  .on('end', function () {
    middle.emit('end');
  });

  return middle;
};

test.js

//test
var request = require('request');
request('http://domain.tld/file.csv').pipe(createCsvParser({
  'trim': true,
  'columns': true,
  'delimitator': '|',
  'transform': function (data, index) {
    return data;
  }
}).pipe(process.stdout);