src/index.js
function createReadStreamBackpressureManager(readableStream) {
  const manager = {
    waitPush: true,
    programmedPushs: [],
    programPush: function programPush(chunk, encoding, done) {
      // Store the current write
      manager.programmedPushs.push([chunk, encoding, done]);
      // Need to be async to avoid nested push attempts
      // Programm a push attempt
      setImmediate(manager.attemptPush);
      // Let's say we're ready for a read
      readableStream.emit('readable');
      readableStream.emit('drain');
    },
    attemptPush: function attemptPush() {
      let nextPush;

      if (manager.waitPush) {
        if (manager.programmedPushs.length) {
          nextPush = manager.programmedPushs.shift();
          manager.waitPush = readableStream.push(nextPush[0], nextPush[1]);
          nextPush[2]();
        }
      } else {
        setImmediate(() => {
          // Need to be async to avoid nested push attempts
          readableStream.emit('readable');
        });
      }
    },
  };

  // Patch the readable stream to manage reads
  readableStream._read = function streamFilterRestoreRead() {
    manager.waitPush = true;
    // Need to be async to avoid nested push attempts
    setImmediate(manager.attemptPush);
  };

  return manager;
}
function createReadStreamBackpressureManager(readableStream) {
  const manager = {
    waitPush: true,
    programmedPushs: [],
    programPush: function programPush(chunk, encoding, done) {
      // Store the current write
      manager.programmedPushs.push([chunk, encoding, done]);
      // Need to be async to avoid nested push attempts
      // Programm a push attempt
      setImmediate(manager.attemptPush);
      // Let's say we're ready for a read
      readableStream.emit('readable');
      readableStream.emit('drain');
    },
    attemptPush: function attemptPush() {
      let nextPush;

      if (manager.waitPush) {
        if (manager.programmedPushs.length) {
          nextPush = manager.programmedPushs.shift();
          manager.waitPush = readableStream.push(nextPush[0], nextPush[1]);
          nextPush[2]();
        }
      } else {
        setImmediate(() => {
          // Need to be async to avoid nested push attempts
          readableStream.emit('readable');
        });
      }
    },
  };

  // Patch the readable stream to manage reads
  readableStream._read = function streamFilterRestoreRead() {
    manager.waitPush = true;
    // Need to be async to avoid nested push attempts
    setImmediate(manager.attemptPush);
  };

  return manager;
}
function createReadStreamBackpressureManager ( readableStream ) { const manager = { waitPush : true , programmedPushs : [ ] , programPush : function programPush ( chunk , encoding , done ) { // Store the current write manager . programmedPushs . push ( [ chunk , encoding , done ] ) ; // Need to be async to avoid nested push attempts // Programm a push attempt setImmediate ( manager . attemptPush ) ; // Let's say we're ready for a read readableStream . emit ( 'readable' ) ; readableStream . emit ( 'drain' ) ; } , attemptPush : function attemptPush ( ) { let nextPush ; if ( manager . waitPush ) { if ( manager . programmedPushs . length ) { nextPush = manager . programmedPushs . shift ( ) ; manager . waitPush = readableStream . push ( nextPush [ 0 ] , nextPush [ 1 ] ) ; nextPush [ 2 ] ( ) ; } } else { setImmediate ( ( ) => { // Need to be async to avoid nested push attempts readableStream . emit ( 'readable' ) ; } ) ; } } , } ; // Patch the readable stream to manage reads readableStream . _read = function streamFilterRestoreRead ( ) { manager . waitPush = true ; // Need to be async to avoid nested push attempts setImmediate ( manager . attemptPush ) ; } ; return manager ; }
Utils to manage readable stream backpressure
Utils to manage readable stream backpressure