Function toReadable

lib/internal/streams/iter/classic.js:263–339

* Create a byte-mode Readable from an AsyncIterable . * The source must yield Uint8Array[] batches (the stream/iter native * format). Each Uint8Array in a batch is pushed as a separate chunk. * @param {AsyncIterable } source * @param {object} [options] * @param {number

(source, options = kNullPrototype)

 * @returns {stream.Readable}
 */
function toReadable(source, options = kNullPrototype) {
  if (typeof source?.[SymbolAsyncIterator] !== 'function') {
    throw new ERR_INVALID_ARG_TYPE('source', 'AsyncIterable', source);
  }

  validateObject(options, 'options');
  const {
    highWaterMark = 64 * 1024,
    signal,
  } = options;
  validateInteger(highWaterMark, 'options.highWaterMark', 0);

  const ReadableCtor = lazyReadable();
  const iterator = source[SymbolAsyncIterator]();
  let backpressure;
  let pumping = false;
  let done = false;

  const readable = new ReadableCtor({
    __proto__: null,
    highWaterMark,
    read() {
      if (backpressure) {
        const { resolve } = backpressure;
        backpressure = null;
        resolve();
      } else if (!pumping && !done) {
        pumping = true;
        pump();
      }
    },
    destroy(err, cb) {
      done = true;
      // Wake up the pump if it's waiting on backpressure so it
      // can see done === true and exit cleanly.
      if (backpressure) {
        backpressure.resolve();
        backpressure = null;
      }
      if (typeof iterator.return === 'function') {
        PromisePrototypeThen(iterator.return(),
                             () => cb(err), (e) => cb(e || err));
      } else {
        cb(err);
      }
    },
  });

  if (signal) {
    addAbortSignalNoValidate(signal, readable);
  }

  async function pump() {
    try {
      while (!done) {
        const { value: batch, done: iterDone } = await iterator.next();
        if (iterDone) {
          done = true;