Function newReadableStreamFromStreamBase

lib/internal/webstreams/adapters.js:1035–1101

* @param {StreamBase} streamBase * @param {QueuingStrategy} strategy * @returns {ReadableStream}

(streamBase, strategy, options = kEmptyObject)

 * @returns {ReadableStream}
 */
function newReadableStreamFromStreamBase(streamBase, strategy, options = kEmptyObject) {
  validateObject(streamBase, 'streamBase');
  validateObject(options, 'options');

  const {
    ondone = () => {},
  } = options;

  if (typeof streamBase.onread === 'function')
    throw new ERR_INVALID_STATE('StreamBase already has a consumer');

  validateFunction(ondone, 'options.ondone');

  let controller;

  streamBase.onread = (arrayBuffer) => {
    const nread = streamBaseState[kReadBytesOrError];

    if (nread === 0)
      return;

    try {
      if (nread === UV_EOF) {
        controller.close();
        streamBase.readStop();
        try {
          ondone();
        } catch (error) {
          controller.error(error);
        }
        return;
      }

      controller.enqueue(arrayBuffer);

      if (controller.desiredSize <= 0)
        streamBase.readStop();
    } catch (error) {
      controller.error(error);
      streamBase.readStop();
    }
  };

  return new ReadableStream({
    start(c) { controller = c; },

    pull() {
      streamBase.readStart();
    },

    cancel() {
      const promise = PromiseWithResolvers();
      try {
        ondone();
      } catch (error) {
        promise.reject(error);
        return promise.promise;
      }