lib/stream-util.js
function multiread(stream, get) {
    if (stream._multiread === undefined) {
        stream._multiread = stream._multiread || [];
        stream._multireading = stream._multireading || 0;
        stream._multireadmax = stream._multireadmax || concurrency;
    }
    if (stream._multiread.length) {
        push();
    } else {
        stream.once('_push', push);
    }
    while ((stream._multiread.length + stream._multireading) < stream._multireadmax) {
        stream._multireading++;
        if (!get(done)) break;
    }

    function done(item) {
        stream._multireading--;
        if (item === null && !stream._multireading) {
            stream.emit('_push');
        } else if (item) {
            stream._multiread.push(item);
            stream.emit('_push');
        }
    }

    function push() {
        if (stream._multiread.length) {
            stream.push(stream._multiread.shift());
        } else {
            stream.push(null);
        }
    }
}
function multiread(stream, get) {
    if (stream._multiread === undefined) {
        stream._multiread = stream._multiread || [];
        stream._multireading = stream._multireading || 0;
        stream._multireadmax = stream._multireadmax || concurrency;
    }
    if (stream._multiread.length) {
        push();
    } else {
        stream.once('_push', push);
    }
    while ((stream._multiread.length + stream._multireading) < stream._multireadmax) {
        stream._multireading++;
        if (!get(done)) break;
    }

    function done(item) {
        stream._multireading--;
        if (item === null && !stream._multireading) {
            stream.emit('_push');
        } else if (item) {
            stream._multiread.push(item);
            stream.emit('_push');
        }
    }

    function push() {
        if (stream._multiread.length) {
            stream.push(stream._multiread.shift());
        } else {
            stream.push(null);
        }
    }
}
function multiread ( stream , get ) { if ( stream . _multiread === undefined ) { stream . _multiread = stream . _multiread || [ ] ; stream . _multireading = stream . _multireading || 0 ; stream . _multireadmax = stream . _multireadmax || concurrency ; } if ( stream . _multiread . length ) { push ( ) ; } else { stream . once ( '_push' , push ) ; } while ( ( stream . _multiread . length + stream . _multireading ) < stream . _multireadmax ) { stream . _multireading ++ ; if ( ! get ( done ) ) break ; } function done ( item ) { stream . _multireading -- ; if ( item === null && ! stream . _multireading ) { stream . emit ( '_push' ) ; } else if ( item ) { stream . _multiread . push ( item ) ; stream . emit ( '_push' ) ; } } function push ( ) { if ( stream . _multiread . length ) { stream . push ( stream . _multiread . shift ( ) ) ; } else { stream . push ( null ) ; } } }
Helper utility for _read functions. Pass your stream object and a getter that can be called concurrently and it will concurrently call your getter behind the scenes and manage the endstate of the stream.
Helper utility for _read functions . Pass your stream object and a getter that can be called concurrently and it will concurrently call your getter behind the scenes and manage the endstate of the stream .