lib/fsExt.js
function readFileLine(filepath, opts, dataCb, completeCb){
    if(typeof arguments[0] !== 'string'){        
        completeCb = arguments[2];
        dataCb = arguments[1];
        opts = arguments[0];
        filepath = opts.file || opts.filename || opts.filepath;
    }
    
    filepath = resolve(filepath).fullPath;

    opts = opts || {};
    var chunkSize = opts.chunkSize || 1;
    var hasHeader = opts.header || opts.headers || opts.hasHeader;
    var headLine;
    var instream = fs.createReadStream(filepath, opts);
    var outstream = new Stream();
    var rl = readline.createInterface(instream, outstream);
    var onLines = dataCb || opts.onLine || opts.onLines || opts.onData;
    var onClose = completeCb || opts.onClose || opts.onComplete;

    var done = dataCb || completeCb;
    instream.on('error', function(err) {
        rl.closed = true;
        rl.close();
        if(done) done(err);
        else throw err;
    });
    
    opts.separator = opts.separator || opts.delimiter;
    
    var lineParser = opts.lineParser || function(line){
        if(!opts.separator) return line;
        return line.split(opts.separator);
    };

    var nextChunk = chunkSize;
    var chunkData = [];
    var lineError;
    var index = -1;

    // callback queue, because pause stream does not guarante that it will pause on current line
    var queue = [];

    function syncExecQueue(){
        if(!queue.length || queue.processing) return;

        var chunkData = queue.shift();
        queue.processing = true;

        onLines.call(rl, chunkSize === 1 ? chunkData[0] : chunkData, function(err){
            if(err) {
                lineError = err;
                queue = [];
                chunkData = [];
                if(!chunkData.last) rl.close();
            }
            else {
                queue.processing = false;
                if(queue.length) syncExecQueue();
                else if(!chunkData.last) rl.resume();
                else if(chunkData.last && onClose) onClose(lineError, index);
            }
        });

    }

    rl.on('line', function(line) {
        index++;
        if(!onLines) return;

        line = lineParser(line);

        if(headLine){
            var lineWithHeader = {};
            for(var i=0;i<headLine.length;i++){
                lineWithHeader[ headLine[i] ] = line[i];
            }
            line = lineWithHeader;
        }
        else if(index===0 && hasHeader){
            headLine = line;
            chunkData = [];
            index--;
            return;
        }

        chunkData.push(line);

        if(nextChunk <= index+1){
            rl.pause();
            nextChunk += chunkSize;
            if(onLines) {
                queue.push(chunkData);
                queue[ queue.length-1 ].index = index+0;
                chunkData = [];
                syncExecQueue();
            }
        }
    });

    rl.on('close', function() {
        if(chunkData.length && !lineError){
            queue.push(chunkData);
            queue[ queue.length-1 ].index = index+0;
            queue[ queue.length-1 ].last = true;
            syncExecQueue();
        }
        else if(onClose) onClose(lineError, index);
    });

    return rl;
}
function readFileLine(filepath, opts, dataCb, completeCb){
    if(typeof arguments[0] !== 'string'){        
        completeCb = arguments[2];
        dataCb = arguments[1];
        opts = arguments[0];
        filepath = opts.file || opts.filename || opts.filepath;
    }
    
    filepath = resolve(filepath).fullPath;

    opts = opts || {};
    var chunkSize = opts.chunkSize || 1;
    var hasHeader = opts.header || opts.headers || opts.hasHeader;
    var headLine;
    var instream = fs.createReadStream(filepath, opts);
    var outstream = new Stream();
    var rl = readline.createInterface(instream, outstream);
    var onLines = dataCb || opts.onLine || opts.onLines || opts.onData;
    var onClose = completeCb || opts.onClose || opts.onComplete;

    var done = dataCb || completeCb;
    instream.on('error', function(err) {
        rl.closed = true;
        rl.close();
        if(done) done(err);
        else throw err;
    });
    
    opts.separator = opts.separator || opts.delimiter;
    
    var lineParser = opts.lineParser || function(line){
        if(!opts.separator) return line;
        return line.split(opts.separator);
    };

    var nextChunk = chunkSize;
    var chunkData = [];
    var lineError;
    var index = -1;

    // callback queue, because pause stream does not guarante that it will pause on current line
    var queue = [];

    function syncExecQueue(){
        if(!queue.length || queue.processing) return;

        var chunkData = queue.shift();
        queue.processing = true;

        onLines.call(rl, chunkSize === 1 ? chunkData[0] : chunkData, function(err){
            if(err) {
                lineError = err;
                queue = [];
                chunkData = [];
                if(!chunkData.last) rl.close();
            }
            else {
                queue.processing = false;
                if(queue.length) syncExecQueue();
                else if(!chunkData.last) rl.resume();
                else if(chunkData.last && onClose) onClose(lineError, index);
            }
        });

    }

    rl.on('line', function(line) {
        index++;
        if(!onLines) return;

        line = lineParser(line);

        if(headLine){
            var lineWithHeader = {};
            for(var i=0;i<headLine.length;i++){
                lineWithHeader[ headLine[i] ] = line[i];
            }
            line = lineWithHeader;
        }
        else if(index===0 && hasHeader){
            headLine = line;
            chunkData = [];
            index--;
            return;
        }

        chunkData.push(line);

        if(nextChunk <= index+1){
            rl.pause();
            nextChunk += chunkSize;
            if(onLines) {
                queue.push(chunkData);
                queue[ queue.length-1 ].index = index+0;
                chunkData = [];
                syncExecQueue();
            }
        }
    });

    rl.on('close', function() {
        if(chunkData.length && !lineError){
            queue.push(chunkData);
            queue[ queue.length-1 ].index = index+0;
            queue[ queue.length-1 ].last = true;
            syncExecQueue();
        }
        else if(onClose) onClose(lineError, index);
    });

    return rl;
}
read file by specified amount of lines, parse head line, perfect for CSV-like data files
@param   {String} file path
@param   {Object} opts read options and callbacks
@param   {Function} dataCb
@param   {Function} completeCb
@returns {Readline} nodejs readline interface