Function newReadableStreamFromStreamReadable

lib/internal/webstreams/adapters.js:470–562

* @typedef {import('./queuingstrategies').QueuingStrategy} QueuingStrategy * @param {Readable} streamReadable * @param {{ * strategy? : QueuingStrategy * type? : 'bytes', * }} [options] * @returns {ReadableStream}

(streamReadable, options = kEmptyObject)

 * @returns {ReadableStream}
 */
function newReadableStreamFromStreamReadable(streamReadable, options = kEmptyObject) {
  // Not using the internal/streams/utils isReadableNodeStream utility
  // here because it will return false if streamReadable is a Duplex
  // whose readable option is false. For a Duplex that is not readable,
  // we want it to pass this check but return a closed ReadableStream.
  if (typeof streamReadable?._readableState !== 'object') {
    throw new ERR_INVALID_ARG_TYPE(
      'streamReadable',
      'stream.Readable',
      streamReadable);
  }
  validateObject(options, 'options');
  if (options.type !== undefined) {
    validateOneOf(options.type, 'options.type', ['bytes', undefined]);
  }

  const isBYOB = options.type === 'bytes';
  let controller;
  let wasCanceled = false;
  let strategy;

  /** @type {UnderlyingSource} */
  const underlyingSource = {
    __proto__: null,
    type: isBYOB ? 'bytes' : undefined,
    start(c) { controller = c; },
    cancel(reason) {
      wasCanceled = true;
      destroy(streamReadable, reason);
    },
  };

  const readable = isReadable(streamReadable);
  const objectMode = streamReadable.readableObjectMode;
  if (readable) {
    underlyingSource.pull = function pull() {
      streamReadable.resume();
    };

    const highWaterMark = streamReadable.readableHighWaterMark;
    strategy = isBYOB ? { highWaterMark } :
      options.strategy ?? new (objectMode ? CountQueuingStrategy : ByteLengthQueuingStrategy)({ highWaterMark });
  }
  const readableStream = new ReadableStream(underlyingSource, strategy);

  // When adapting a Duplex as a ReadableStream, readable completion should not
  // wait for a half-open writable side to finish as well.
  let cleanup = noop;
  cleanup = eos(streamReadable, {
    __proto__: null,
    writable: false,
    [kEosNodeSynchronousCallback]: true,
  }, (error) => {
    error = handleKnownInternalErrors(error);

    // If eos calls the callback synchronously, cleanup is still a no-op here.
    cleanup();