1-file.txt

foo bar
baz
qux

last line (there may or may not be a trailing newline after this line)

2-output

start
0 foo bar
1 baz
2 qux
3
4 last line (there may or may not be a trailing newline after this line)
end

line-parser.js

var stream = require("stream");
var util = require("util");

function LineParser() {
  stream.Transform.apply(this, arguments);

  // start off our stream with our "start" and some default "buffer"
  this.push("start\n");
  this._line = "";
  this._count = 0;
}
util.inherits(LineParser, stream.Transform);


// emit that line with the counter and the newline added again
LineParser.prototype._sendLine = function(line) {
  this.push(this._count++ + " " + line + "\n");
}

// _transform gets called with every incoming bit of data, done() when done
// in case you are async...
LineParser.prototype._transform = function(data, encoding, done) {
  var lines = (this._line + (data.toString())).split('\n');
  // take the last line (which didn't have a trailing newline) for the next
  // buffer chunk
  this._line = lines.pop();
  lines.forEach(this._sendLine, this);
  done();
};

// _flush gets called when there will be no more _transform's called
// after the "readable" ends, you get _flush, then your stream ends
LineParser.prototype._flush = function(done) {
  if (this._line) {
    this._sendLine(this._line);
  }
  this.push("end\n");
  done();
}

module.exports = LineParser;

nodejs-stream.js

#!/usr/bin/env node

// Streaming IO is great, but far more complex than Synchronous or
// Asynchronous IO

// you should probably abstract it into a reusable thing and make a module

// ... unless you use a pre-built third party library. you can write your own streams

var fs = require("fs");
var LineParser = require("./line-parser");

fs.createReadStream("1-file.txt")
  .pipe(new LineParser)
  .pipe(process.stdout);