line-stream.js

'use strict';

var Transform = require('stream').Transform,
    inherits = require('util').inherits;

var noBuf = new Buffer(0);

module.exports = LineStream;

function LineStream(opts) {
    opts = opts || { };
    
    if (!(this instanceof LineStream))
        return new LineStream(opts);
    
    this.scrap = noBuf;
    Transform.call(this, opts);
}
inherits(LineStream, Transform);

LineStream.prototype._transform = function (chunk, encoding, cb) {
    var start = 0, end = 0, line = noBuf;
    for (var i = 0, x = chunk.length; i < x; i++) {
        if (chunk[i] === 10 || chunk[i] === 13) {
            if (start <= end) {
                line = chunk.slice(start, end);
                if (this.scrap !== noBuf) {
                    console.log('outputting mixed segment');
                    console.log(Buffer.concat([this.scrap, line]));
                    this.push(Buffer.concat([this.scrap, line]));
                    this.scrap = noBuf;
                } else {
                    console.log('outputting pure segment');
                    console.log(line);
                    this.push(line);
                }
            } else if (this.scrap !== noBuf) {
                console.log('outputting stored segment');
                console.log(this.scrap);
                this.push(this.scrap);
                this.scrap = noBuf;
            }
            start = i + 1;
        } else {
            end = i + 1;
        }
    }
    if (start <= end) {
        console.log('storing remaining segment');
        this.scrap = chunk.slice(start, end);
    }
    cb(null);
};

LineStream.prototype._flush = function (cb) {
    this.push(this.scrap);
    this.scrap = noBuf;
    cb(null);
};