node streams v2

output.txt

CSVP <Buffer 61 2c 62 2c 63 2c 64 0a 31 2c 32 2c 33 2c 34 0a 35 2c 36 2c 37 2c 38 0a>
DUMP { a: '1', b: '2', c: '3', d: '4' }

test.coffee

_      = require 'underscore'
stream = require 'stream'
fs     = require 'fs'
csv    = require 'csv'

# takes in a csv file stream, emits json objects
class CSVParser extends stream.Transform
  constructor: (@options) ->
    super @options
    @csv = csv().from.options @options.from
    @csv.on 'record', (record) => @push record
    @csv.on 'end',    (count) =>  @end()
    @csv.on 'error',  (error) =>  @end(error)
  _transform: (chunk, output, done) =>
    console.log 'CSVP', chunk if @options.debug
    @csv.write chunk
    done()

# passthrough that dumps to stdout
class Dump extends stream.Transform
  _transform: (chunk, output, done) =>
    console.log 'DUMP', chunk
    output chunk
    done()

# _.filter
class Filter extends stream.Transform
  constructor: (@options) ->
    @options = { test: @options } if _(@options).isFunction()
    super @options
  _transform: (chunk, output, done) =>
    output chunk if @options.test chunk
    done()

filestream = fs.createReadStream "#{__dirname}/test1.csv"
csvparser  = new CSVParser {from: { header: true, columns: true}, debug: true}
filter     = new Filter (obj) -> obj.a < 5
dump       = new Dump()
filestream.pipe(csvparser).pipe(filter).pipe(dump)

test1.csv