more streams2

underscore.stream.coffee

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

# apply() for constructors
construct = (constructor, args) ->
  F = -> constructor.apply this, args
  F.prototype = constructor.prototype
  new F

# chainable pipe()
class UnderscoreStream
  constructor: (@read_stream) ->
  @mixin: (ReadableStreamKlass) ->
    UnderscoreStream::[ReadableStreamKlass.name] = () ->
      instance = construct ReadableStreamKlass, arguments
      @read_stream.pipe instance if @read_stream
      @read_stream = instance
      @

# add the ability to chain a csv parser
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 (@options.name or 'CSVP'), chunk if @options.debug
    @csv.write chunk
    done()

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

# add some builtin readable streams
UnderscoreStream.mixin fs.ReadStream

# add custom streams
UnderscoreStream.mixin CSVParser
UnderscoreStream.mixin Dump

# add it to underscore
_.mixin stream: (read_stream) -> new UnderscoreStream read_stream

# all together now
_.stream()
.ReadStream("#{__dirname}/test1.csv")
.CSVParser({ from: { header: true, columns: true }})
.Dump()