/**
 * @license
 * Copyright The Closure Library Authors.
 * SPDX-License-Identifier: Apache-2.0
 */

/**
 * @fileoverview A lite polyfill of the ReadableStream native API with a subset
 * of methods supported.
 */
goog.module('goog.streams.liteImpl');

const NativeResolver = goog.require('goog.promise.NativeResolver');
const liteTypes = goog.require('goog.streams.liteTypes');
const {assert, assertFunction} = goog.require('goog.asserts');

/**
 * The lite implementation of ReadableStream.
 *
 * Supports the getReader() method and locked property.
 *
 * The only method of underlying sources that is supported is enqueueing,
 * closing, and erroring.
 *
 * Pulling (including backpressure and sizes) and cancellation are not
 * supported.
 * @template T
 * @implements {liteTypes.ReadableStream<T>}
 */
class ReadableStream {
  /** @package */
  constructor() {
    /** @package {!ReadableStream.State} */
    this.state = ReadableStream.State.READABLE;

    /**  @package {!ReadableStreamDefaultReader|undefined} */
    this.reader = undefined;

    /** @type {*} */
    this.storedError = undefined;

    /** @package {!ReadableStreamDefaultController} */
    this.readableStreamController;
  }

  /**
   * Returns true if the ReadableStream has been locked to a reader.
   * https://streams.spec.whatwg.org/#rs-locked
   * @return {boolean}
   * @override
   */
  get locked() {
    return this.reader !== undefined;
  }

  /**
   * Returns a ReadableStreamDefaultReader that enables reading chunks from
   * the source.
   * https://streams.spec.whatwg.org/#rs-get-reader
   * @return {!ReadableStreamDefaultReader<T>}
   * @override
   */
  getReader() {
    return this.reader = new ReadableStreamDefaultReader(this);
  }

  /**
   * @return {!Promise<!IIterableResult<T>>}
   * @package
   */
  addReadRequest() {
    const request = new NativeResolver();
    this.reader.readRequests.push(request);
    return request.promise;
  }

  /** @package */
  close() {
    this.state = ReadableStream.State.CLOSED;
    if (!this.reader) {
      return;
    }
    for (const readRequest of this.reader.readRequests) {
      readRequest.resolve({value: undefined, done: true});
    }
    this.reader.readRequests = [];
    this.reader.closedResolver.resolve();
  }

  /**
   * @param {*} e
   * @package
   */
  error(e) {
    this.state = ReadableStream.State.ERRORED;
    this.storedError = e;
    if (!this.reader) {
      return;
    }
    for (const readRequest of this.reader.readRequests) {
      readRequest.reject(e);
    }
    this.reader.readRequests = [];
    this.reader.closedResolver.promise.catch(() => {});
    this.reader.closedResolver.reject(e);
  }

  /**
   * @param {T} chunk
   * @param {boolean} done
   * @package
   */
  fulfillReadRequest(chunk, done) {
    const readRequest = assert(this.reader).readRequests.shift();
    readRequest.resolve({value: chunk, done});
  }

  /**
   * @return {number}
   * @package
   */
  getNumReadRequests() {
    return assert(this.reader).readRequests.length;
  }

  /**
   * @return {boolean}
   * @package
   */
  hasDefaultReader() {
    return this.reader !== undefined;
  }
}

/** @package @enum {number} */
ReadableStream.State = {
  READABLE: 1,
  CLOSED: 2,
  ERRORED: 3,
};

/**
 * Creates and returns a new ReadableStream.
 *
 * The underlying source should only have a start() method, and no other
 * properties.
 * @param {!liteTypes.ReadableStreamUnderlyingSource<T>} underlyingSource
 * @return {!ReadableStream<T>}
 * @suppress {strictMissingProperties}
 * @template T
 */
function newReadableStream(underlyingSource) {
  assertFunction(
      underlyingSource.start,
      `'start' property must be a function on an underlying source for a ` +
          'lite ReadableStream');
  const verifyObject =
      /** @type {!Object} */ (underlyingSource);
  assert(
      !(verifyObject.pull),
      `'pull' property not allowed on an underlying source for a ` +
          'lite ReadableStream');
  assert(
      !(verifyObject.cancel),
      `'cancel' property not allowed on an underlying source for a ` +
          'lite ReadableStream');
  assert(
      !(verifyObject.type),
      `'type' property not allowed on an underlying source for a ` +
          'lite ReadableStream');
  assert(
      !(verifyObject.autoAllocateChunkSize),
      `'autoAllocateChunkSize' property not allowed on an underlying ` +
          'source for a lite ReadableStream');
  const startAlgorithm = (controller) => underlyingSource.start(controller);
  const stream = new ReadableStream();
  const controller = new ReadableStreamDefaultController(stream);
  stream.readableStreamController = controller;
  controller.start(startAlgorithm);
  return stream;
}

/**
 * A reader for a lite ReadableStream.
 *
 * Supports the read() and releaseLock() methods, along with the closed
 * property.
 * @template T
 * @implements {liteTypes.ReadableStreamDefaultReader<T>}
 */
class ReadableStreamDefaultReader {
  /**
   * @param {!ReadableStream} stream
   * @package
   */
  constructor(stream) {
    if (stream.reader) {
      throw new TypeError(
          'ReadableStreamReader constructor can only accept readable streams ' +
          'that are not yet locked to a reader');
    }
    /** @package {!ReadableStream|undefined} */
    this.ownerReadableStream = stream;

    /** @package {!NativeResolver<undefined>} */
    this.closedResolver = new NativeResolver();

    /** @package {!Array<!NativeResolver<!IIterableResult<T>>>} */
    this.readRequests = [];

    if (stream.state === ReadableStream.State.CLOSED) {
      this.closedResolver.resolve();
    } else if (stream.state === ReadableStream.State.ERRORED) {
      this.closedResolver.promise.catch(() => {});
      this.closedResolver.reject(stream.storedError);
    }
  }

  /**
   * Returns a Promise that resolves when the Stream closes or is errored, or if
   * the reader releases its lock.
   * https://streams.spec.whatwg.org/#default-reader-closed
   * @return {!Promise<undefined>}
   * @override
   */
  get closed() {
    return this.closedResolver.promise;
  }

  /**
   * Returns a Promise that resolves with an IIterableResult providing the next
   * chunk or that the stream is closed. The Promise may reject if the stream
   * is errored.
   * https://streams.spec.whatwg.org/#default-reader-read
   * @return {!Promise<!IIterableResult<T>>}
   * @override
   */
  read() {
    if (!this.ownerReadableStream) {
      throw new TypeError(
          'This readable stream reader has been released and cannot be used ' +
          'to read from its previous owner stream');
    }
    return this.readInternal();
  }

  /**
   * Release the lock on the stream. Any further calls to read() will error,
   * and the stream can create another reader.
   * https://streams.spec.whatwg.org/#default-reader-release-lock
   * @return {void}
   * @override
   */
  releaseLock() {
    if (!this.ownerReadableStream) {
      return;
    }
    if (this.readRequests.length) {
      throw new TypeError(
          'Cannot release a readable stream reader when it still has ' +
          'outstanding read() calls that have not yet settled');
    }
    this.release();
  }

  /** @package */
  release() {
    const stream = assert(this.ownerReadableStream);
    const e = new TypeError(
        'This readable stream reader has been released and cannot be used ' +
        `to monitor the stream's state`);
    if (stream.state === ReadableStream.State.READABLE) {
      this.closedResolver.promise.catch(() => {});
      this.closedResolver.reject(e);
    } else {
      this.closedResolver = new NativeResolver();
      this.closedResolver.promise.catch(() => {});
      this.closedResolver.reject(e);
    }
    stream.reader = undefined;
    this.ownerReadableStream = undefined;
  }

  /**
   * @return {!Promise<!IIterableResult<T>>}
   * @package
   */
  readInternal() {
    const stream = assert(this.ownerReadableStream);
    if (stream.state === ReadableStream.State.CLOSED) {
      return Promise.resolve({value: undefined, done: true});
    }
    if (stream.state === ReadableStream.State.ERRORED) {
      return Promise.reject(stream.storedError);
    }
    return stream.readableStreamController.pullSteps();
  }
}

/**
 * A controller for a lite ReadableStream.
 *
 * Provides the enqueue(), error(), and close() methods.
 * @template T
 * @implements {liteTypes.ReadableStreamDefaultController<T>}
 */
class ReadableStreamDefaultController {
  /**
   * @param {!ReadableStream} stream
   * @package
   */
  constructor(stream) {
    /** @package @const {!ReadableStream} */
    this.controlledReadableStream = stream;

    /** @package @const {!Queue} */
    this.queue = new Queue();

    /** @package {boolean} */
    this.closeRequested = false;
  }

  /**
   * Signals that the ReadableStream should close. The ReadableStream will
   * actually close once all of its chunks have been read.
   * https://streams.spec.whatwg.org/#rs-default-controller-close
   * @return {void}
   * @override
   */
  close() {
    if (!this.canCloseOrEnqueue()) {
      throw new TypeError(
          'Cannot close a readable stream that has already been requested to ' +
          'be closed');
    }
    this.closeInternal();
  }

  /**
   * Enqueues a new chunk into the stream that can be read.
   * https://streams.spec.whatwg.org/#rs-default-controller-enqueue
   * @param {T} chunk
   * @override
   */
  enqueue(chunk) {
    if (!this.canCloseOrEnqueue()) {
      throw new TypeError(
          'Cannot enqueue a readable stream that has already been requested ' +
          'to be closed');
    }
    this.enqueueInternal(chunk);
  }

  /**
   * Closes the stream with an error. Any future interactions with the
   * controller will throw an error.
   * https://streams.spec.whatwg.org/#rs-default-controller-error
   * @param {*} e
   * @override
   */
  error(e) {
    this.errorInternal(e);
  }

  /**
   * @param {function(!ReadableStreamDefaultController):
   *     (!Promise<undefined>|undefined)} startAlgorithm
   * @package
   */
  start(startAlgorithm) {
    Promise.resolve(startAlgorithm(this))
        .then(
            () => {
              this.started();
            },
            (e) => {
              this.errorInternal(e);
            });
  }

  /**
   * @return {!Promise<!IIterableResult<T>>}
   * @package
   */
  pullSteps() {
    if (!this.queue.empty()) {
      const chunk = this.dequeueFromQueue();
      if (this.closeRequested && this.queue.empty()) {
        this.clearAlgorithms();
        this.controlledReadableStream.close();
      } else {
        this.callPullIfNeeded();
      }
      return Promise.resolve({value: chunk, done: false});
    }
    const promise = this.controlledReadableStream.addReadRequest();
    this.callPullIfNeeded();
    return promise;
  }

  /** @package */
  started() {}

  /** @package */
  callPullIfNeeded() {}

  /** @package */
  clearAlgorithms() {}

  /**
   * @package
   */
  closeInternal() {
    this.closeRequested = true;
    if (this.queue.empty()) {
      this.clearAlgorithms();
      this.controlledReadableStream.close();
    }
  }

  /**
   * @param {T} chunk
   * @package
   */
  enqueueInternal(chunk) {
    if (this.controlledReadableStream.locked &&
        this.controlledReadableStream.getNumReadRequests() > 0) {
      this.controlledReadableStream.fulfillReadRequest(
          chunk, /* done= */ false);
      return;
    }
    this.enqueueIntoQueue(chunk);
  }

  /**
   * @param {*} e
   * @package
   */
  errorInternal(e) {
    if (this.controlledReadableStream.state !== ReadableStream.State.READABLE) {
      return;
    }
    this.resetQueue();
    this.clearAlgorithms();
    this.controlledReadableStream.error(e);
  }

  /**
   * @return {boolean}
   * @package
   */
  canCloseOrEnqueue() {
    return !this.closeRequested &&
        this.controlledReadableStream.state === ReadableStream.State.READABLE;
  }

  /**
   * @param {T} chunk
   * @protected
   */
  enqueueIntoQueue(chunk) {
    this.queue.enqueueValue(chunk);
  }

  /**
   * @return {T}
   * @protected
   */
  dequeueFromQueue() {
    return this.queue.dequeueValue();
  }

  /**
   * @protected
   */
  resetQueue() {
    this.queue.resetQueue();
  }
}

/**
 * An internal Queue representation. This simple Queue just wraps an Array.
 * Other implementations may also have a size associated with each element.
 * @template T
 * @package
 */
class Queue {
  constructor() {
    /** @private {!Array<T>} */
    this.queue_ = [];
  }

  /**
   * @return {boolean}
   */
  empty() {
    return this.queue_.length === 0;
  }

  /**
   * @param {T} value
   */
  enqueueValue(value) {
    this.queue_.push(value);
  }

  /**
   * @return {T}
   */
  dequeueValue() {
    return this.queue_.shift();
  }

  /**
   * @return {void}
   */
  resetQueue() {
    this.queue_ = [];
  }
}

exports = {
  Queue,
  ReadableStream,
  ReadableStreamDefaultController,
  ReadableStreamDefaultReader,
  newReadableStream,
};
