<commit_before>var util = require('util'),
    Stream          = require('stream').Stream,
    libs            = require('./libs.js'),
    slice           = libs.slice,
    inherit         = libs.inherit,
    mixin           = libs.mixin,
    StreamBuffer    = require('./streambuffer.js').StreamBuffer,
    snake           = require('./snake.js').snake,
    actions         = {
        flush   : function (source, options) {
            var dest = this;
            
            //we need a buffer to hold the outstanding streams
            if (!Array.isArray(dest._buffer)) {
                dest._buffer = [];
            }
            dest._pipeCount  = dest._pipeCount || 0;
            dest._pipeCount += 1;
            
            options             = options || {};
            
            /**
             *hellper function to pull the next source Stream or end dest Stream
             */
            function nextOrEnd() {
                var next = dest._buffer.shift();
                dest._pipeCount -= 1;
                
                if (typeof next === 'function') {
                    next();
                } else if (dest._pipeCount <= 0) {
                    dest.end();
                }
            }
            
            var buff = new StreamBuffer(dest, source, {end : false});
            function next () {
                if (buff.hasEnded()) {
                    buff.
                        once('end', nextOrEnd).
                        drain();
                } else {
                    dest._buffer.unshift(next);
                    dest._pipeCount += 1;
                }
            }
            dest._buffer.push(next);
            
            if (dest._pipeCount === 1) {
                buff.once('full', nextOrEnd);
            } else if (!options.serial) {
                buff.once('full', function () {
                    var i = dest._buffer.indexOf(next);
                    if (i >= 0) {
                        dest._buffer.shift(
                            dest._buffer.splice(i,1).shift()
                        );
                    }
                });
            }
            
            return dest;
        },
        funnel      : function (source, options) {
            var dest = this;
            
            //initial conditions/
            if (!Array.isArray(dest._buffer)) {
                dest._buffer = [];
            }
            dest._pipeCount     = dest._pipeCount || 0;
            dest._pipeCount    += 1;
            
            options             = options || {};
            options.buffer      = (options.buffer === false ? false : true);
            
            function nextOrEnd () {
                var startNext   = dest._buffer.shift();
                dest._pipeCount -= 1;
                
                if (typeof startNext === 'function') {
                    startNext();
                } else if (dest._pipeCount <= 0) {
                    dest.end();
                }
            }
            
           function next () {
                if (source instanceof StreamBuffer) {
                    if (source.hasEnded()) {
                        source.once('end', nextOrEnd).
                            drain();
                    } else {
                        source.drain().
                            once('end', nextOrEnd);
                    }
                } else if (source instanceof Stream) {
                    source.once('end', nextOrEnd);
                    source.pipe(dest, {end : false});
                    source.resume();
                } else {
                    nextOrEnd();
                }
            }
            
            if (dest._pipeCount === 1) {
                source.once('end', nextOrEnd);
                source.pipe(dest, {end : false});
                return dest;
            } else {
                if (options.buffer) {
                    source = new StreamBuffer(dest, source, {end : false});
                    if (!options.serial) {
                        source.once('full', function () {
                            var i = dest._buffer.indexOf(next);
                            if (i > 0) {
                                dest._buffer.unshift(
                                    dest._buffer.splice(i,1).shift()
                                );
                            }
                        });
                    }
                } else {
                    source.pause();
                }
            }
            
            dest._buffer.push(next);
            return dest;
        }
    },
    SerialPump      = inherit(
        Stream,
        actions
    );
    
SerialPump.prototype.end = function () {
    this.emit('end');
};

mixin(exports, {
    SerialPump      : SerialPump,
    StreamBuffer    : StreamBuffer,<commit_msg>format, comments and a _pipeCount problem<commit_after>var util = require('util'),
    Stream          = require('stream').Stream,
    libs            = require('./libs.js'),
    slice           = libs.slice,
    inherit         = libs.inherit,
    mixin           = libs.mixin,
    StreamBuffer    = require('./streambuffer.js').StreamBuffer,
    snake           = require('./snake.js').snake,
    actions         = {
        flush   : function (source, options) {
            var dest = this;
            
            //we need a buffer to hold the outstanding streams
            if (!Array.isArray(dest._buffer)) {
                dest._buffer    = [];
            }
            //Stream::pipe already gives us a var for this, why make my own?
            dest._pipeCount     = dest._pipeCount || 0;
            dest._pipeCount     += 1;
            
            options             = options || {};
            
            
             //hellper function to pull the next source or end the dest
            function nextOrEnd() {
                var next = dest._buffer.shift();
                dest._pipeCount -= 1;
                
                if (typeof next === 'function') {
                    next();
                } else if (dest._pipeCount <= 0) {
                    dest.end();
                }
            }
            
            //flush should result in ONLY one data event to the dest stream
            //so all source streams are buffered.
            var buff = new StreamBuffer(dest, source, {end : false});
            
            //hellper function to do the work for this source
            //this way flush and funnel can share the same buffer
            function next () {
                if (buff.hasEnded()) {
                    buff.
                        once('end', nextOrEnd).
                        drain();
                } else {
                    //every time we go through nextOrEnd _pipeCount goes down
                    //for this Stream we are going to go to nextOrEnd
                    //when we are full AND when the Stream is drained
                    //There is probobly a better way to do this...
                    dest._pipeCount += 2;
                    buff.once('full', nextOrEnd);
                    dest._buffer.unshift(next);
                }
            }
            
            //push onto the stack to proccess
            dest._buffer.push(next);
            
            if (dest._pipeCount === 1) {
                //this is the first, just start it
                buff.once('full', nextOrEnd);
            } else if (!options.serial) {
                //once it's full move it to the top of the stack
                buff.once('full', function () {
                    var i = dest._buffer.indexOf(next);
                    if (i >= 0) {
                        dest._buffer.shift(
                            dest._buffer.splice(i,1).shift()
                        );
                    }
                });
            }
            
            //let's do that again!
            return dest;
        },
        funnel      : function (source, options) {
            var dest = this;
            
            //we need a buffer to hold the outstanding streams
            if (!Array.isArray(dest._buffer)) {
                dest._buffer = [];
            }
            //Stream::pipe already gives us a var for this, why make my own?
            dest._pipeCount     = dest._pipeCount || 0;
            dest._pipeCount    += 1;
            
            options             = options || {};
            //if you really don't want to buffer somthing you need to tell me.
            options.buffer      = (options.buffer === false ? false : true);
            
            //hellper function to pull the next source or end the dest
            function nextOrEnd () {
                var next   = dest._buffer.shift();
                dest._pipeCount -= 1;
                
                console.log(dest._pipeCount);
                if (typeof next === 'function') {
                    next();
                } else if (dest._pipeCount <= 0) {
                    dest.end();
                }
            }
            
            //hellper function to do the work for this source
            //this way flush and funnel can share the same buffer
           function next () {
                if (source instanceof StreamBuffer) {
                    if (source.hasEnded()) {
                        source.once('end', nextOrEnd).
                            drain();
                    } else {
                        source.drain().
                            once('end', nextOrEnd);
                    }
                } else if (source instanceof Stream) {
                    source.once('end', nextOrEnd);
                    source.pipe(dest, {end : false});
                    source.resume();
                } else {
                    nextOrEnd();
                }
            }
            
            if (dest._pipeCount === 1) {
                //this is the first, just start it
                source.once('end', nextOrEnd);
                source.pipe(dest, {end : false});
                return dest;
            } else {
                if (options.buffer) {
                    //buffer the source
                    source = new StreamBuffer(dest, source, {end : false});
                    if (!options.serial) {
                        //once it's full move it to the top of the stack
                        source.once('full', function () {
                            var i = dest._buffer.indexOf(next);
                            if (i > 0) {
                                dest._buffer.unshift(
                                    dest._buffer.splice(i,1).shift()
                                );
                            }
                        });
                    }
                } else {
                    //no buffering, we are not ready for you yet.
                    source.pause();
                }
            }
            //push onto the stack to proccess
            dest._buffer.push(next);
            
            //let's do that again!
            return dest;
        }
    },
    SerialPump      = inherit(
        Stream,
        actions
    );
    
SerialPump.prototype.end = function () {
    this.emit('end');
};

mixin(exports, {
    SerialPump      : SerialPump,
    StreamBuffer    : StreamBuffer,
{'fraction_non_alphanumeric': 0.09849352479957713, 'fraction_numerical': 0.0026429389481102987, 'mean_word_length': 0.887281795511222, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5060274', 'n_tokens_mistral': 2697, 'n_tokens_neox': 2538, 'n_words': 683}