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;
}
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
read file by specified amount of lines parse head line perfect for CSV - like data files