code: 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;
  }
docs: Utils to manage readable stream backpressure