Line-by-line protocol for node.js

LineStream.js

var net = require("net");
var events = require("events");
var sys = require("sys");
var fs = require("fs");

function _cutFromRight(str) {
	if (str.charAt(str.length-1)=="\n") {
		return [str.substr(0,str.length-1), ''];
	}
	var li = str.lastIndexOf("\n");
	var str1 = str.substr(0,li);
	var str2 = str.substr(li+1, str.length);
	return [str1, str2];
}

exports._cutFromRight=_cutFromRight;

LineStream = function() {
	net.Stream.call(this);
	var self=this;
	var previousBuffer="";

	this.addListener('data', function(data) {
		data = (previousBuffer+data).replace(/\r/g, "");
		var somethingLeft = _cutFromRight(data);
		previousBuffer=somethingLeft[1];
		
		if (somethingLeft[0]=='') {
			return;
		}
		
		self.emit('lines', somethingLeft[0].split("\n"));
	});
}

sys.inherits(LineStream, net.Stream);
exports.LineStream=LineStream;

LineStream.prototype.puts = function(line) {
	return this.write(line+"\r\n");
}

FileLineStream = function(path) {
	events.EventEmitter.call(this);	
	var self=this;
	var previousBuffer="";
	
	this.stream = fs.createReadStream(path, {bufferSize: 1024*1024*10});
	this.stream.addListener('data', function(data) {
		data = (previousBuffer+data).replace(/\r/g, "");
		var somethingLeft = _cutFromRight(data);
		previousBuffer=somethingLeft[1];
		
		if (somethingLeft[0]=='') {
			return;
		}
		
		self.emit('lines', somethingLeft[0].split("\n"));
	})
	.addListener('close', function() {self.emit('close')});
}
sys.inherits(FileLineStream, events.EventEmitter);


FileLineStream.prototype.pause = function() {
	this.stream.pause();
}

FileLineStream.prototype.resume = function() {
	this.stream.resume();
}

FileLineStream.prototype.destroy = function() {
	this.stream.destroy();
}

exports.FileLineStream=FileLineStream;