denoland-deno/cli/js/streams/readable-internals.ts

// Forked from https://github.com/stardazed/sd-streams/tree/8928cf04b035fd02fb1340b7eb541c76be37e546
// Copyright (c) 2018-Present by Arthur Langereis - @zenmumbler MIT

/**
 * streams/readable-internals - internal types and functions for readable streams
 * Part of Stardazed
 * (c) 2018-Present by Arthur Langereis - @zenmumbler
 * https://github.com/stardazed/sd-streams
 */

/* eslint-disable @typescript-eslint/no-explicit-any */
// TODO reenable this lint here

import * as shared from "./shared-internals.ts";
import * as q from "./queue-mixin.ts";
import {
  QueuingStrategy,
  QueuingStrategySizeCallback,
  UnderlyingSource,
  UnderlyingByteSource
} from "../dom_types.ts";

// ReadableStreamDefaultController
export const controlledReadableStream_ = Symbol("controlledReadableStream_");
export const pullAlgorithm_ = Symbol("pullAlgorithm_");
export const cancelAlgorithm_ = Symbol("cancelAlgorithm_");
export const strategySizeAlgorithm_ = Symbol("strategySizeAlgorithm_");
export const strategyHWM_ = Symbol("strategyHWM_");
export const started_ = Symbol("started_");
export const closeRequested_ = Symbol("closeRequested_");
export const pullAgain_ = Symbol("pullAgain_");
export const pulling_ = Symbol("pulling_");
export const cancelSteps_ = Symbol("cancelSteps_");
export const pullSteps_ = Symbol("pullSteps_");

// ReadableByteStreamController
export const autoAllocateChunkSize_ = Symbol("autoAllocateChunkSize_");
export const byobRequest_ = Symbol("byobRequest_");
export const controlledReadableByteStream_ = Symbol(
  "controlledReadableByteStream_"
);
export const pendingPullIntos_ = Symbol("pendingPullIntos_");

// ReadableStreamDefaultReader
export const closedPromise_ = Symbol("closedPromise_");
export const ownerReadableStream_ = Symbol("ownerReadableStream_");
export const readRequests_ = Symbol("readRequests_");
export const readIntoRequests_ = Symbol("readIntoRequests_");

// ReadableStreamBYOBRequest
export const associatedReadableByteStreamController_ = Symbol(
  "associatedReadableByteStreamController_"
);
export const view_ = Symbol("view_");

// ReadableStreamBYOBReader

// ReadableStream
export const reader_ = Symbol("reader_");
export const readableStreamController_ = Symbol("readableStreamController_");

export type StartFunction<OutputType> = (
  controller: SDReadableStreamControllerBase<OutputType>
) => void | PromiseLike<void>;
export type StartAlgorithm = () => Promise<void> | void;
export type PullFunction<OutputType> = (
  controller: SDReadableStreamControllerBase<OutputType>
) => void | PromiseLike<void>;
export type PullAlgorithm<OutputType> = (
  controller: SDReadableStreamControllerBase<OutputType>
) => PromiseLike<void>;
export type CancelAlgorithm = (reason?: shared.ErrorResult) => Promise<void>;

// ----

export interface SDReadableStreamControllerBase<OutputType> {
  readonly desiredSize: number | null;
  close(): void;
  error(e?: shared.ErrorResult): void;

  [cancelSteps_](reason: shared.ErrorResult): Promise<void>;
  [pullSteps_](forAuthorCode: boolean): Promise<IteratorResult<OutputType>>;
}

export interface SDReadableStreamBYOBRequest {
  readonly view: ArrayBufferView;
  respond(bytesWritten: number): void;
  respondWithNewView(view: ArrayBufferView): void;

  [associatedReadableByteStreamController_]:
    | SDReadableByteStreamController
    | undefined;
  [view_]: ArrayBufferView | undefined;
}

interface ArrayBufferViewCtor {
  new (
    buffer: ArrayBufferLike,
    byteOffset?: number,
    byteLength?: number
  ): ArrayBufferView;
}

export interface PullIntoDescriptor {
  readerType: "default" | "byob";
  ctor: ArrayBufferViewCtor;
  buffer: ArrayBufferLike;
  byteOffset: number;
  byteLength: number;
  bytesFilled: number;
  elementSize: number;
}

export interface SDReadableByteStreamController
  extends SDReadableStreamControllerBase<ArrayBufferView>,
    q.ByteQueueContainer {
  readonly byobRequest: SDReadableStreamBYOBRequest | undefined;
  enqueue(chunk: ArrayBufferView): void;

  [autoAllocateChunkSize_]: number | undefined; // A positive integer, when the automatic buffer allocation feature is enabled. In that case, this value specifies the size of buffer to allocate. It is undefined otherwise.
  [byobRequest_]: SDReadableStreamBYOBRequest | undefined; // A ReadableStreamBYOBRequest instance representing the current BYOB pull request
  [cancelAlgorithm_]: CancelAlgorithm; // A promise-returning algorithm, taking one argument (the cancel reason), which communicates a requested cancelation to the underlying source
  [closeRequested_]: boolean; // A boolean flag indicating whether the stream has been closed by its underlying byte source, but still has chunks in its internal queue that have not yet been read
  [controlledReadableByteStream_]: SDReadableStream<ArrayBufferView>; // The ReadableStream instance controlled
  [pullAgain_]: boolean; // A boolean flag set to true if the stream’s mechanisms requested a call to the underlying byte source’s pull() method to pull more data, but the pull could not yet be done since a previous call is still executing
  [pullAlgorithm_]: PullAlgorithm<ArrayBufferView>; // A promise-returning algorithm that pulls data from the underlying source
  [pulling_]: boolean; // A boolean flag set to true while the underlying byte source’s pull() method is executing and has not yet fulfilled, used to prevent reentrant calls
  [pendingPullIntos_]: PullIntoDescriptor[]; // A List of descriptors representing pending BYOB pull requests
  [started_]: boolean; // A boolean flag indicating whether the underlying source has finished starting
  [strategyHWM_]: number; // A number supplied to the constructor as part of the stream’s queuing strategy, indicating the point at which the stream will apply backpressure to its underlying byte source
}

export interface SDReadableStreamDefaultController<OutputType>
  extends SDReadableStreamControllerBase<OutputType>,
    q.QueueContainer<OutputType> {
  enqueue(chunk?: OutputType): void;

  [controlledReadableStream_]: SDReadableStream<OutputType>;
  [pullAlgorithm_]: PullAlgorithm<OutputType>;
  [cancelAlgorithm_]: CancelAlgorithm;
  [strategySizeAlgorithm_]: QueuingStrategySizeCallback<OutputType>;
  [strategyHWM_]: number;

  [started_]: boolean;
  [closeRequested_]: boolean;
  [pullAgain_]: boolean;
  [pulling_]: boolean;
}

// ----

export interface SDReadableStreamReader<OutputType> {
  readonly closed: Promise<void>;
  cancel(reason: shared.ErrorResult): Promise<void>;
  releaseLock(): void;

  [ownerReadableStream_]: SDReadableStream<OutputType> | undefined;
  [closedPromise_]: shared.ControlledPromise<void>;
}

export interface ReadRequest<V> extends shared.ControlledPromise<V> {
  forAuthorCode: boolean;
}

export declare class SDReadableStreamDefaultReader<OutputType>
  implements SDReadableStreamReader<OutputType> {
  constructor(stream: SDReadableStream<OutputType>);

  readonly closed: Promise<void>;
  cancel(reason: shared.ErrorResult): Promise<void>;
  releaseLock(): void;
  read(): Promise<IteratorResult<OutputType | undefined>>;

  [ownerReadableStream_]: SDReadableStream<OutputType> | undefined;
  [closedPromise_]: shared.ControlledPromise<void>;
  [readRequests_]: Array<ReadRequest<IteratorResult<OutputType>>>;
}

export declare class SDReadableStreamBYOBReader
  implements SDReadableStreamReader<ArrayBufferView> {
  constructor(stream: SDReadableStream<ArrayBufferView>);

  readonly closed: Promise<void>;
  cancel(reason: shared.ErrorResult): Promise<void>;
  releaseLock(): void;
  read(view: ArrayBufferView): Promise<IteratorResult<ArrayBufferView>>;

  [ownerReadableStream_]: SDReadableStream<ArrayBufferView> | undefined;
  [closedPromise_]: shared.ControlledPromise<void>;
  [readIntoRequests_]: Array<ReadRequest<IteratorResult<ArrayBufferView>>>;
}

/* TODO reenable this when we add WritableStreams and Transforms
export interface GenericTransformStream<InputType, OutputType> {
  readable: SDReadableStream<OutputType>;
  writable: ws.WritableStream<InputType>;
}
*/

export type ReadableStreamState = "readable" | "closed" | "errored";

export declare class SDReadableStream<OutputType> {
  constructor(
    underlyingSource: UnderlyingByteSource,
    strategy?: { highWaterMark?: number; size?: undefined }
  );
  constructor(
    underlyingSource?: UnderlyingSource<OutputType>,
    strategy?: QueuingStrategy<OutputType>
  );

  readonly locked: boolean;
  cancel(reason?: shared.ErrorResult): Promise<void>;
  getReader(): SDReadableStreamReader<OutputType>;
  getReader(options: { mode: "byob" }): SDReadableStreamBYOBReader;
  tee(): Array<SDReadableStream<OutputType>>;

  /* TODO reenable these methods when we bring in writableStreams and transport types
  pipeThrough<ResultType>(
    transform: GenericTransformStream<OutputType, ResultType>,
    options?: PipeOptions
  ): SDReadableStream<ResultType>;
  pipeTo(
    dest: ws.WritableStream<OutputType>,
    options?: PipeOptions
  ): Promise<void>;
  */
  [shared.state_]: ReadableStreamState;
  [shared.storedError_]: shared.ErrorResult;
  [reader_]: SDReadableStreamReader<OutputType> | undefined;
  [readableStreamController_]: SDReadableStreamControllerBase<OutputType>;
}

// ---- Stream

export function initializeReadableStream<OutputType>(
  stream: SDReadableStream<OutputType>
): void {
  stream[shared.state_] = "readable";
  stream[reader_] = undefined;
  stream[shared.storedError_] = undefined;
  stream[readableStreamController_] = undefined!; // mark slot as used for brand check
}

export function isReadableStream(
  value: unknown
): value is SDReadableStream<any> {
  if (typeof value !== "object" || value === null) {
    return false;
  }
  return readableStreamController_ in value;
}

export function isReadableStreamLocked<OutputType>(
  stream: SDReadableStream<OutputType>
): boolean {
  return stream[reader_] !== undefined;
}

export function readableStreamGetNumReadIntoRequests<OutputType>(
  stream: SDReadableStream<OutputType>
): number {
  // TODO remove the "as unknown" cast
  // This is in to workaround a compiler error
  // error TS2352: Conversion of type 'SDReadableStreamReader<OutputType>' to type 'SDReadableStreamBYOBReader' may be a mistake because neither type sufficiently overlaps with the other. If this was intentional, convert the expression to 'unknown' first.
  // Type 'SDReadableStreamReader<OutputType>' is missing the following properties from type 'SDReadableStreamBYOBReader': read, [readIntoRequests_]
  const reader = (stream[reader_] as unknown) as SDReadableStreamBYOBReader;
  if (reader === undefined) {
    return 0;
  }
  return reader[readIntoRequests_].length;
}

export function readableStreamGetNumReadRequests<OutputType>(
  stream: SDReadableStream<OutputType>
): number {
  const reader = stream[reader_] as SDReadableStreamDefaultReader<OutputType>;
  if (reader === undefined) {
    return 0;
  }
  return reader[readRequests_].length;
}

export function readableStreamCreateReadResult<T>(
  value: T,
  done: boolean,
  forAuthorCode: boolean
): IteratorResult<T> {
  const prototype = forAuthorCode ? Object.prototype : null;
  const result = Object.create(prototype);
  result.value = value;
  result.done = done;
  return result;
}

export function readableStreamAddReadIntoRequest(
  stream: SDReadableStream<ArrayBufferView>,
  forAuthorCode: boolean
): Promise<IteratorResult<ArrayBufferView, any>> {
  // Assert: ! IsReadableStreamBYOBReader(stream.[[reader]]) is true.
  // Assert: stream.[[state]] is "readable" or "closed".
  const reader = stream[reader_] as SDReadableStreamBYOBReader;
  const conProm = shared.createControlledPromise<
    IteratorResult<ArrayBufferView>
  >() as ReadRequest<IteratorResult<ArrayBufferView>>;
  conProm.forAuthorCode = forAuthorCode;
  reader[readIntoRequests_].push(conProm);
  return conProm.promise;
}

export function readableStreamAddReadRequest<OutputType>(
  stream: SDReadableStream<OutputType>,
  forAuthorCode: boolean
): Promise<IteratorResult<OutputType, any>> {
  // Assert: ! IsReadableStreamDefaultReader(stream.[[reader]]) is true.
  // Assert: stream.[[state]] is "readable".
  const reader = stream[reader_] as SDReadableStreamDefaultReader<OutputType>;
  const conProm = shared.createControlledPromise<
    IteratorResult<OutputType>
  >() as ReadRequest<IteratorResult<OutputType>>;
  conProm.forAuthorCode = forAuthorCode;
  reader[readRequests_].push(conProm);
  return conProm.promise;
}

export function readableStreamHasBYOBReader<OutputType>(
  stream: SDReadableStream<OutputType>
): boolean {
  const reader = stream[reader_];
  return isReadableStreamBYOBReader(reader);
}

export function readableStreamHasDefaultReader<OutputType>(
  stream: SDReadableStream<OutputType>
): boolean {
  const reader = stream[reader_];
  return isReadableStreamDefaultReader(reader);
}

export function readableStreamCancel<OutputType>(
  stream: SDReadableStream<OutputType>,
  reason: shared.ErrorResult
): Promise<undefined> {
  if (stream[shared.state_] === "closed") {
    return Promise.resolve(undefined);
  }
  if (stream[shared.state_] === "errored") {
    return Promise.reject(stream[shared.storedError_]);
  }
  readableStreamClose(stream);

  const sourceCancelPromise = stream[readableStreamController_][cancelSteps_](
    reason
  );
  return sourceCancelPromise.then(_ => undefined);
}

export function readableStreamClose<OutputType>(
  stream: SDReadableStream<OutputType>
): void {
  // Assert: stream.[[state]] is "readable".
  stream[shared.state_] = "closed";
  const reader = stream[reader_];
  if (reader === undefined) {
    return;
  }

  if (isReadableStreamDefaultReader(reader)) {
    for (const readRequest of reader[readRequests_]) {
      readRequest.resolve(
        readableStreamCreateReadResult(
          undefined,
          true,
          readRequest.forAuthorCode
        )
      );
    }
    reader[readRequests_] = [];
  }
  reader[closedPromise_].resolve();
  reader[closedPromise_].promise.catch(() => {});
}

export function readableStreamError<OutputType>(
  stream: SDReadableStream<OutputType>,
  error: shared.ErrorResult
): void {
  if (stream[shared.state_] !== "readable") {
    throw new RangeError("Stream is in an invalid state");
  }
  stream[shared.state_] = "errored";
  stream[shared.storedError_] = error;

  const reader = stream[reader_];
  if (reader === undefined) {
    return;
  }
  if (isReadableStreamDefaultReader(reader)) {
    for (const readRequest of reader[readRequests_]) {
      readRequest.reject(error);
    }
    reader[readRequests_] = [];
  } else {
    // Assert: IsReadableStreamBYOBReader(reader).
    // TODO remove the "as unknown" cast
    const readIntoRequests = ((reader as unknown) as SDReadableStreamBYOBReader)[
      readIntoRequests_
    ];
    for (const readIntoRequest of readIntoRequests) {
      readIntoRequest.reject(error);
    }
    // TODO remove the "as unknown" cast
    ((reader as unknown) as SDReadableStreamBYOBReader)[readIntoRequests_] = [];
  }

  reader[closedPromise_].reject(error);
}

// ---- Readers

export function isReadableStreamDefaultReader(
  reader: unknown
): reader is SDReadableStreamDefaultReader<any> {
  if (typeof reader !== "object" || reader === null) {
    return false;
  }
  return readRequests_ in reader;
}

export function isReadableStreamBYOBReader(
  reader: unknown
): reader is SDReadableStreamBYOBReader {
  if (typeof reader !== "object" || reader === null) {
    return false;
  }
  return readIntoRequests_ in reader;
}

export function readableStreamReaderGenericInitialize<OutputType>(
  reader: SDReadableStreamReader<OutputType>,
  stream: SDReadableStream<OutputType>
): void {
  reader[ownerReadableStream_] = stream;
  stream[reader_] = reader;
  const streamState = stream[shared.state_];

  reader[closedPromise_] = shared.createControlledPromise<void>();
  if (streamState === "readable") {
    // leave as is
  } else if (streamState === "closed") {
    reader[closedPromise_].resolve(undefined);
  } else {
    reader[closedPromise_].reject(stream[shared.storedError_]);
    reader[closedPromise_].promise.catch(() => {});
  }
}

export function readableStreamReaderGenericRelease<OutputType>(
  reader: SDReadableStreamReader<OutputType>
): void {
  // Assert: reader.[[ownerReadableStream]] is not undefined.
  // Assert: reader.[[ownerReadableStream]].[[reader]] is reader.
  const stream = reader[ownerReadableStream_];
  if (stream === undefined) {
    throw new TypeError("Reader is in an inconsistent state");
  }

  if (stream[shared.state_] === "readable") {
    // code moved out
  } else {
    reader[closedPromise_] = shared.createControlledPromise<void>();
  }
  reader[closedPromise_].reject(new TypeError());
  reader[closedPromise_].promise.catch(() => {});

  stream[reader_] = undefined;
  reader[ownerReadableStream_] = undefined;
}

export function readableStreamBYOBReaderRead(
  reader: SDReadableStreamBYOBReader,
  view: ArrayBufferView,
  forAuthorCode = false
): Promise<IteratorResult<ArrayBufferView, any>> {
  const stream = reader[ownerReadableStream_]!;
  // Assert: stream is not undefined.

  if (stream[shared.state_] === "errored") {
    return Promise.reject(stream[shared.storedError_]);
  }
  return readableByteStreamControllerPullInto(
    stream[readableStreamController_] as SDReadableByteStreamController,
    view,
    forAuthorCode
  );
}

export function readableStreamDefaultReaderRead<OutputType>(
  reader: SDReadableStreamDefaultReader<OutputType>,
  forAuthorCode = false
): Promise<IteratorResult<OutputType | undefined>> {
  const stream = reader[ownerReadableStream_]!;
  // Assert: stream is not undefined.

  if (stream[shared.state_] === "closed") {
    return Promise.resolve(
      readableStreamCreateReadResult(undefined, true, forAuthorCode)
    );
  }
  if (stream[shared.state_] === "errored") {
    return Promise.reject(stream[shared.storedError_]);
  }
  // Assert: stream.[[state]] is "readable".
  return stream[readableStreamController_][pullSteps_](forAuthorCode);
}

export function readableStreamFulfillReadIntoRequest<OutputType>(
  stream: SDReadableStream<OutputType>,
  chunk: ArrayBufferView,
  done: boolean
): void {
  // TODO remove the "as unknown" cast
  const reader = (stream[reader_] as unknown) as SDReadableStreamBYOBReader;
  const readIntoRequest = reader[readIntoRequests_].shift()!; // <-- length check done in caller
  readIntoRequest.resolve(
    readableStreamCreateReadResult(chunk, done, readIntoRequest.forAuthorCode)
  );
}

export function readableStreamFulfillReadRequest<OutputType>(
  stream: SDReadableStream<OutputType>,
  chunk: OutputType,
  done: boolean
): void {
  const reader = stream[reader_] as SDReadableStreamDefaultReader<OutputType>;
  const readRequest = reader[readRequests_].shift()!; // <-- length check done in caller
  readRequest.resolve(
    readableStreamCreateReadResult(chunk, done, readRequest.forAuthorCode)
  );
}

// ---- DefaultController

export function setUpReadableStreamDefaultController<OutputType>(
  stream: SDReadableStream<OutputType>,
  controller: SDReadableStreamDefaultController<OutputType>,
  startAlgorithm: StartAlgorithm,
  pullAlgorithm: PullAlgorithm<OutputType>,
  cancelAlgorithm: CancelAlgorithm,
  highWaterMark: number,
  sizeAlgorithm: QueuingStrategySizeCallback<OutputType>
): void {
  // Assert: stream.[[readableStreamController]] is undefined.
  controller[controlledReadableStream_] = stream;
  q.resetQueue(controller);
  controller[started_] = false;
  controller[closeRequested_] = false;
  controller[pullAgain_] = false;
  controller[pulling_] = false;
  controller[strategySizeAlgorithm_] = sizeAlgorithm;
  controller[strategyHWM_] = highWaterMark;
  controller[pullAlgorithm_] = pullAlgorithm;
  controller[cancelAlgorithm_] = cancelAlgorithm;
  stream[readableStreamController_] = controller;

  const startResult = startAlgorithm();
  Promise.resolve(startResult).then(
    _ => {
      controller[started_] = true;
      // Assert: controller.[[pulling]] is false.
      // Assert: controller.[[pullAgain]] is false.
      readableStreamDefaultControllerCallPullIfNeeded(controller);
    },
    error => {
      readableStreamDefaultControllerError(controller, error);
    }
  );
}

export function isReadableStreamDefaultController(
  value: unknown
): value is SDReadableStreamDefaultController<any> {
  if (typeof value !== "object" || value === null) {
    return false;
  }
  return controlledReadableStream_ in value;
}

export function readableStreamDefaultControllerHasBackpressure<OutputType>(
  controller: SDReadableStreamDefaultController<OutputType>
): boolean {
  return !readableStreamDefaultControllerShouldCallPull(controller);
}

export function readableStreamDefaultControllerCanCloseOrEnqueue<OutputType>(
  controller: SDReadableStreamDefaultController<OutputType>
): boolean {
  const state = controller[controlledReadableStream_][shared.state_];
  return controller[closeRequested_] === false && state === "readable";
}

export function readableStreamDefaultControllerGetDesiredSize<OutputType>(
  controller: SDReadableStreamDefaultController<OutputType>
): number | null {
  const state = controller[controlledReadableStream_][shared.state_];
  if (state === "errored") {
    return null;
  }
  if (state === "closed") {
    return 0;
  }
  return controller[strategyHWM_] - controller[q.queueTotalSize_];
}

export function readableStreamDefaultControllerClose<OutputType>(
  controller: SDReadableStreamDefaultController<OutputType>
): void {
  // Assert: !ReadableStreamDefaultControllerCanCloseOrEnqueue(controller) is true.
  controller[closeRequested_] = true;
  const stream = controller[controlledReadableStream_];
  if (controller[q.queue_].length === 0) {
    readableStreamDefaultControllerClearAlgorithms(controller);
    readableStreamClose(stream);
  }
}

export function readableStreamDefaultControllerEnqueue<OutputType>(
  controller: SDReadableStreamDefaultController<OutputType>,
  chunk: OutputType
): void {
  const stream = controller[controlledReadableStream_];
  // Assert: !ReadableStreamDefaultControllerCanCloseOrEnqueue(controller) is true.
  if (
    isReadableStreamLocked(stream) &&
    readableStreamGetNumReadRequests(stream) > 0
  ) {
    readableStreamFulfillReadRequest(stream, chunk, false);
  } else {
    // Let result be the result of performing controller.[[strategySizeAlgorithm]], passing in chunk,
    // and interpreting the result as an ECMAScript completion value.
    // impl note: assuming that in JS land this just means try/catch with rethrow
    let chunkSize: number;
    try {
      chunkSize = controller[strategySizeAlgorithm_](chunk);
    } catch (error) {
      readableStreamDefaultControllerError(controller, error);
      throw error;
    }
    try {
      q.enqueueValueWithSize(controller, chunk, chunkSize);
    } catch (error) {
      readableStreamDefaultControllerError(controller, error);
      throw error;
    }
  }
  readableStreamDefaultControllerCallPullIfNeeded(controller);
}

export function readableStreamDefaultControllerError<OutputType>(
  controller: SDReadableStreamDefaultController<OutputType>,
  error: shared.ErrorResult
): void {
  const stream = controller[controlledReadableStream_];
  if (stream[shared.state_] !== "readable") {
    return;
  }
  q.resetQueue(controller);
  readableStreamDefaultControllerClearAlgorithms(controller);
  readableStreamError(stream, error);
}

export function readableStreamDefaultControllerCallPullIfNeeded<OutputType>(
  controller: SDReadableStreamDefaultController<OutputType>
): void {
  if (!readableStreamDefaultControllerShouldCallPull(controller)) {
    return;
  }
  if (controller[pulling_]) {
    controller[pullAgain_] = true;
    return;
  }
  if (controller[pullAgain_]) {
    throw new RangeError("Stream controller is in an invalid state.");
  }

  controller[pulling_] = true;
  controller[pullAlgorithm_](controller).then(
    _ => {
      controller[pulling_] = false;
      if (controller[pullAgain_]) {
        controller[pullAgain_] = false;
        readableStreamDefaultControllerCallPullIfNeeded(controller);
      }
    },
    error => {
      readableStreamDefaultControllerError(controller, error);
    }
  );
}

export function readableStreamDefaultControllerShouldCallPull<OutputType>(
  controller: SDReadableStreamDefaultController<OutputType>
): boolean {
  const stream = controller[controlledReadableStream_];
  if (!readableStreamDefaultControllerCanCloseOrEnqueue(controller)) {
    return false;
  }
  if (controller[started_] === false) {
    return false;
  }
  if (
    isReadableStreamLocked(stream) &&
    readableStreamGetNumReadRequests(stream) > 0
  ) {
    return true;
  }
  const desiredSize = readableStreamDefaultControllerGetDesiredSize(controller);
  if (desiredSize === null) {
    throw new RangeError("Stream is in an invalid state.");
  }
  return desiredSize > 0;
}

export function readableStreamDefaultControllerClearAlgorithms<OutputType>(
  controller: SDReadableStreamDefaultController<OutputType>
): void {
  controller[pullAlgorithm_] = undefined!;
  controller[cancelAlgorithm_] = undefined!;
  controller[strategySizeAlgorithm_] = undefined!;
}

// ---- BYOBController

export function setUpReadableByteStreamController(
  stream: SDReadableStream<ArrayBufferView>,
  controller: SDReadableByteStreamController,
  startAlgorithm: StartAlgorithm,
  pullAlgorithm: PullAlgorithm<ArrayBufferView>,
  cancelAlgorithm: CancelAlgorithm,
  highWaterMark: number,
  autoAllocateChunkSize: number | undefined
): void {
  // Assert: stream.[[readableStreamController]] is undefined.
  if (stream[readableStreamController_] !== undefined) {
    throw new TypeError("Cannot reuse streams");
  }
  if (autoAllocateChunkSize !== undefined) {
    if (
      !shared.isInteger(autoAllocateChunkSize) ||
      autoAllocateChunkSize <= 0
    ) {
      throw new RangeError(
        "autoAllocateChunkSize must be a positive, finite integer"
      );
    }
  }
  // Set controller.[[controlledReadableByteStream]] to stream.
  controller[controlledReadableByteStream_] = stream;
  // Set controller.[[pullAgain]] and controller.[[pulling]] to false.
  controller[pullAgain_] = false;
  controller[pulling_] = false;
  readableByteStreamControllerClearPendingPullIntos(controller);
  q.resetQueue(controller);
  controller[closeRequested_] = false;
  controller[started_] = false;
  controller[strategyHWM_] = shared.validateAndNormalizeHighWaterMark(
    highWaterMark
  );
  controller[pullAlgorithm_] = pullAlgorithm;
  controller[cancelAlgorithm_] = cancelAlgorithm;
  controller[autoAllocateChunkSize_] = autoAllocateChunkSize;
  controller[pendingPullIntos_] = [];
  stream[readableStreamController_] = controller;

  // Let startResult be the result of performing startAlgorithm.
  const startResult = startAlgorithm();
  Promise.resolve(startResult).then(
    _ => {
      controller[started_] = true;
      // Assert: controller.[[pulling]] is false.
      // Assert: controller.[[pullAgain]] is false.
      readableByteStreamControllerCallPullIfNeeded(controller);
    },
    error => {
      readableByteStreamControllerError(controller, error);
    }
  );
}

export function isReadableStreamBYOBRequest(
  value: unknown
): value is SDReadableStreamBYOBRequest {
  if (typeof value !== "object" || value === null) {
    return false;
  }
  return associatedReadableByteStreamController_ in value;
}

export function isReadableByteStreamController(
  value: unknown
): value is SDReadableByteStreamController {
  if (typeof value !== "object" || value === null) {
    return false;
  }
  return controlledReadableByteStream_ in value;
}

export function readableByteStreamControllerCallPullIfNeeded(
  controller: SDReadableByteStreamController
): void {
  if (!readableByteStreamControllerShouldCallPull(controller)) {
    return;
  }
  if (controller[pulling_]) {
    controller[pullAgain_] = true;
    return;
  }
  // Assert: controller.[[pullAgain]] is false.
  controller[pulling_] = true;
  controller[pullAlgorithm_](controller).then(
    _ => {
      controller[pulling_] = false;
      if (controller[pullAgain_]) {
        controller[pullAgain_] = false;
        readableByteStreamControllerCallPullIfNeeded(controller);
      }
    },
    error => {
      readableByteStreamControllerError(controller, error);
    }
  );
}

export function readableByteStreamControllerClearAlgorithms(
  controller: SDReadableByteStreamController
): void {
  controller[pullAlgorithm_] = undefined!;
  controller[cancelAlgorithm_] = undefined!;
}

export function readableByteStreamControllerClearPendingPullIntos(
  controller: SDReadableByteStreamController
): void {
  readableByteStreamControllerInvalidateBYOBRequest(controller);
  controller[pendingPullIntos_] = [];
}

export function readableByteStreamControllerClose(
  controller: SDReadableByteStreamController
): void {
  const stream = controller[controlledReadableByteStream_];
  // Assert: controller.[[closeRequested]] is false.
  // Assert: stream.[[state]] is "readable".
  if (controller[q.queueTotalSize_] > 0) {
    controller[closeRequested_] = true;
    return;
  }
  if (controller[pendingPullIntos_].length > 0) {
    const firstPendingPullInto = controller[pendingPullIntos_][0];
    if (firstPendingPullInto.bytesFilled > 0) {
      const error = new TypeError();
      readableByteStreamControllerError(controller, error);
      throw error;
    }
  }
  readableByteStreamControllerClearAlgorithms(controller);
  readableStreamClose(stream);
}

export function readableByteStreamControllerCommitPullIntoDescriptor(
  stream: SDReadableStream<ArrayBufferView>,
  pullIntoDescriptor: PullIntoDescriptor
): void {
  // Assert: stream.[[state]] is not "errored".
  let done = false;
  if (stream[shared.state_] === "closed") {
    // Assert: pullIntoDescriptor.[[bytesFilled]] is 0.
    done = true;
  }
  const filledView = readableByteStreamControllerConvertPullIntoDescriptor(
    pullIntoDescriptor
  );
  if (pullIntoDescriptor.readerType === "default") {
    readableStreamFulfillReadRequest(stream, filledView, done);
  } else {
    // Assert: pullIntoDescriptor.[[readerType]] is "byob".
    readableStreamFulfillReadIntoRequest(stream, filledView, done);
  }
}

export function readableByteStreamControllerConvertPullIntoDescriptor(
  pullIntoDescriptor: PullIntoDescriptor
): ArrayBufferView {
  const { bytesFilled, elementSize } = pullIntoDescriptor;
  // Assert: bytesFilled <= pullIntoDescriptor.byteLength
  // Assert: bytesFilled mod elementSize is 0
  return new pullIntoDescriptor.ctor(
    pullIntoDescriptor.buffer,
    pullIntoDescriptor.byteOffset,
    bytesFilled / elementSize
  );
}

export function readableByteStreamControllerEnqueue(
  controller: SDReadableByteStreamController,
  chunk: ArrayBufferView
): void {
  const stream = controller[controlledReadableByteStream_];
  // Assert: controller.[[closeRequested]] is false.
  // Assert: stream.[[state]] is "readable".
  const { buffer, byteOffset, byteLength } = chunk;

  const transferredBuffer = shared.transferArrayBuffer(buffer);

  if (readableStreamHasDefaultReader(stream)) {
    if (readableStreamGetNumReadRequests(stream) === 0) {
      readableByteStreamControllerEnqueueChunkToQueue(
        controller,
        transferredBuffer,
        byteOffset,
        byteLength
      );
    } else {
      // Assert: controller.[[queue]] is empty.
      const transferredView = new Uint8Array(
        transferredBuffer,
        byteOffset,
        byteLength
      );
      readableStreamFulfillReadRequest(stream, transferredView, false);
    }
  } else if (readableStreamHasBYOBReader(stream)) {
    readableByteStreamControllerEnqueueChunkToQueue(
      controller,
      transferredBuffer,
      byteOffset,
      byteLength
    );
    readableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(
      controller
    );
  } else {
    // Assert: !IsReadableStreamLocked(stream) is false.
    readableByteStreamControllerEnqueueChunkToQueue(
      controller,
      transferredBuffer,
      byteOffset,
      byteLength
    );
  }
  readableByteStreamControllerCallPullIfNeeded(controller);
}

export function readableByteStreamControllerEnqueueChunkToQueue(
  controller: SDReadableByteStreamController,
  buffer: ArrayBufferLike,
  byteOffset: number,
  byteLength: number
): void {
  controller[q.queue_].push({ buffer, byteOffset, byteLength });
  controller[q.queueTotalSize_] += byteLength;
}

export function readableByteStreamControllerError(
  controller: SDReadableByteStreamController,
  error: shared.ErrorResult
): void {
  const stream = controller[controlledReadableByteStream_];
  if (stream[shared.state_] !== "readable") {
    return;
  }
  readableByteStreamControllerClearPendingPullIntos(controller);
  q.resetQueue(controller);
  readableByteStreamControllerClearAlgorithms(controller);
  readableStreamError(stream, error);
}

export function readableByteStreamControllerFillHeadPullIntoDescriptor(
  controller: SDReadableByteStreamController,
  size: number,
  pullIntoDescriptor: PullIntoDescriptor
): void {
  // Assert: either controller.[[pendingPullIntos]] is empty, or the first element of controller.[[pendingPullIntos]] is pullIntoDescriptor.
  readableByteStreamControllerInvalidateBYOBRequest(controller);
  pullIntoDescriptor.bytesFilled += size;
}

export function readableByteStreamControllerFillPullIntoDescriptorFromQueue(
  controller: SDReadableByteStreamController,
  pullIntoDescriptor: PullIntoDescriptor
): boolean {
  const elementSize = pullIntoDescriptor.elementSize;
  const currentAlignedBytes =
    pullIntoDescriptor.bytesFilled -
    (pullIntoDescriptor.bytesFilled % elementSize);
  const maxBytesToCopy = Math.min(
    controller[q.queueTotalSize_],
    pullIntoDescriptor.byteLength - pullIntoDescriptor.bytesFilled
  );
  const maxBytesFilled = pullIntoDescriptor.bytesFilled + maxBytesToCopy;
  const maxAlignedBytes = maxBytesFilled - (maxBytesFilled % elementSize);
  let totalBytesToCopyRemaining = maxBytesToCopy;
  let ready = false;

  if (maxAlignedBytes > currentAlignedBytes) {
    totalBytesToCopyRemaining =
      maxAlignedBytes - pullIntoDescriptor.bytesFilled;
    ready = true;
  }
  const queue = controller[q.queue_];

  while (totalBytesToCopyRemaining > 0) {
    const headOfQueue = queue.front()!;
    const bytesToCopy = Math.min(
      totalBytesToCopyRemaining,
      headOfQueue.byteLength
    );
    const destStart =
      pullIntoDescriptor.byteOffset + pullIntoDescriptor.bytesFilled;
    shared.copyDataBlockBytes(
      pullIntoDescriptor.buffer,
      destStart,
      headOfQueue.buffer,
      headOfQueue.byteOffset,
      bytesToCopy
    );
    if (headOfQueue.byteLength === bytesToCopy) {
      queue.shift();
    } else {
      headOfQueue.byteOffset += bytesToCopy;
      headOfQueue.byteLength -= bytesToCopy;
    }
    controller[q.queueTotalSize_] -= bytesToCopy;
    readableByteStreamControllerFillHeadPullIntoDescriptor(
      controller,
      bytesToCopy,
      pullIntoDescriptor
    );
    totalBytesToCopyRemaining -= bytesToCopy;
  }
  if (!ready) {
    // Assert: controller[queueTotalSize_] === 0
    // Assert: pullIntoDescriptor.bytesFilled > 0
    // Assert: pullIntoDescriptor.bytesFilled < pullIntoDescriptor.elementSize
  }
  return ready;
}

export function readableByteStreamControllerGetDesiredSize(
  controller: SDReadableByteStreamController
): number | null {
  const stream = controller[controlledReadableByteStream_];
  const state = stream[shared.state_];
  if (state === "errored") {
    return null;
  }
  if (state === "closed") {
    return 0;
  }
  return controller[strategyHWM_] - controller[q.queueTotalSize_];
}

export function readableByteStreamControllerHandleQueueDrain(
  controller: SDReadableByteStreamController
): void {
  // Assert: controller.[[controlledReadableByteStream]].[[state]] is "readable".
  if (controller[q.queueTotalSize_] === 0 && controller[closeRequested_]) {
    readableByteStreamControllerClearAlgorithms(controller);
    readableStreamClose(controller[controlledReadableByteStream_]);
  } else {
    readableByteStreamControllerCallPullIfNeeded(controller);
  }
}

export function readableByteStreamControllerInvalidateBYOBRequest(
  controller: SDReadableByteStreamController
): void {
  const byobRequest = controller[byobRequest_];
  if (byobRequest === undefined) {
    return;
  }
  byobRequest[associatedReadableByteStreamController_] = undefined;
  byobRequest[view_] = undefined;
  controller[byobRequest_] = undefined;
}

export function readableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(
  controller: SDReadableByteStreamController
): void {
  // Assert: controller.[[closeRequested]] is false.
  const pendingPullIntos = controller[pendingPullIntos_];
  while (pendingPullIntos.length > 0) {
    if (controller[q.queueTotalSize_] === 0) {
      return;
    }
    const pullIntoDescriptor = pendingPullIntos[0];
    if (
      readableByteStreamControllerFillPullIntoDescriptorFromQueue(
        controller,
        pullIntoDescriptor
      )
    ) {
      readableByteStreamControllerShiftPendingPullInto(controller);
      readableByteStreamControllerCommitPullIntoDescriptor(
        controller[controlledReadableByteStream_],
        pullIntoDescriptor
      );
    }
  }
}

export function readableByteStreamControllerPullInto(
  controller: SDReadableByteStreamController,
  view: ArrayBufferView,
  forAuthorCode: boolean
): Promise<IteratorResult<ArrayBufferView, any>> {
  const stream = controller[controlledReadableByteStream_];

  const elementSize = (view as Uint8Array).BYTES_PER_ELEMENT || 1; // DataView exposes this in Webkit as 1, is not present in FF or Blink
  const ctor = view.constructor as Uint8ArrayConstructor; // the typecast here is just for TS typing, it does not influence buffer creation

  const byteOffset = view.byteOffset;
  const byteLength = view.byteLength;
  const buffer = shared.transferArrayBuffer(view.buffer);
  const pullIntoDescriptor: PullIntoDescriptor = {
    buffer,
    byteOffset,
    byteLength,
    bytesFilled: 0,
    elementSize,
    ctor,
    readerType: "byob"
  };

  if (controller[pendingPullIntos_].length > 0) {
    controller[pendingPullIntos_].push(pullIntoDescriptor);
    return readableStreamAddReadIntoRequest(stream, forAuthorCode);
  }
  if (stream[shared.state_] === "closed") {
    const emptyView = new ctor(
      pullIntoDescriptor.buffer,
      pullIntoDescriptor.byteOffset,
      0
    );
    return Promise.resolve(
      readableStreamCreateReadResult(emptyView, true, forAuthorCode)
    );
  }

  if (controller[q.queueTotalSize_] > 0) {
    if (
      readableByteStreamControllerFillPullIntoDescriptorFromQueue(
        controller,
        pullIntoDescriptor
      )
    ) {
      const filledView = readableByteStreamControllerConvertPullIntoDescriptor(
        pullIntoDescriptor
      );
      readableByteStreamControllerHandleQueueDrain(controller);
      return Promise.resolve(
        readableStreamCreateReadResult(filledView, false, forAuthorCode)
      );
    }
    if (controller[closeRequested_]) {
      const error = new TypeError();
      readableByteStreamControllerError(controller, error);
      return Promise.reject(error);
    }
  }

  controller[pendingPullIntos_].push(pullIntoDescriptor);
  const promise = readableStreamAddReadIntoRequest(stream, forAuthorCode);
  readableByteStreamControllerCallPullIfNeeded(controller);
  return promise;
}

export function readableByteStreamControllerRespond(
  controller: SDReadableByteStreamController,
  bytesWritten: number
): void {
  bytesWritten = Number(bytesWritten);
  if (!shared.isFiniteNonNegativeNumber(bytesWritten)) {
    throw new RangeError("bytesWritten must be a finite, non-negative number");
  }
  // Assert: controller.[[pendingPullIntos]] is not empty.
  readableByteStreamControllerRespondInternal(controller, bytesWritten);
}

export function readableByteStreamControllerRespondInClosedState(
  controller: SDReadableByteStreamController,
  firstDescriptor: PullIntoDescriptor
): void {
  firstDescriptor.buffer = shared.transferArrayBuffer(firstDescriptor.buffer);
  // Assert: firstDescriptor.[[bytesFilled]] is 0.
  const stream = controller[controlledReadableByteStream_];
  if (readableStreamHasBYOBReader(stream)) {
    while (readableStreamGetNumReadIntoRequests(stream) > 0) {
      const pullIntoDescriptor = readableByteStreamControllerShiftPendingPullInto(
        controller
      )!;
      readableByteStreamControllerCommitPullIntoDescriptor(
        stream,
        pullIntoDescriptor
      );
    }
  }
}

export function readableByteStreamControllerRespondInReadableState(
  controller: SDReadableByteStreamController,
  bytesWritten: number,
  pullIntoDescriptor: PullIntoDescriptor
): void {
  if (
    pullIntoDescriptor.bytesFilled + bytesWritten >
    pullIntoDescriptor.byteLength
  ) {
    throw new RangeError();
  }
  readableByteStreamControllerFillHeadPullIntoDescriptor(
    controller,
    bytesWritten,
    pullIntoDescriptor
  );
  if (pullIntoDescriptor.bytesFilled < pullIntoDescriptor.elementSize) {
    return;
  }
  readableByteStreamControllerShiftPendingPullInto(controller);
  const remainderSize =
    pullIntoDescriptor.bytesFilled % pullIntoDescriptor.elementSize;
  if (remainderSize > 0) {
    const end = pullIntoDescriptor.byteOffset + pullIntoDescriptor.bytesFilled;
    const remainder = shared.cloneArrayBuffer(
      pullIntoDescriptor.buffer,
      end - remainderSize,
      remainderSize,
      ArrayBuffer
    );
    readableByteStreamControllerEnqueueChunkToQueue(
      controller,
      remainder,
      0,
      remainder.byteLength
    );
  }
  pullIntoDescriptor.buffer = shared.transferArrayBuffer(
    pullIntoDescriptor.buffer
  );
  pullIntoDescriptor.bytesFilled =
    pullIntoDescriptor.bytesFilled - remainderSize;
  readableByteStreamControllerCommitPullIntoDescriptor(
    controller[controlledReadableByteStream_],
    pullIntoDescriptor
  );
  readableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(controller);
}

export function readableByteStreamControllerRespondInternal(
  controller: SDReadableByteStreamController,
  bytesWritten: number
): void {
  const firstDescriptor = controller[pendingPullIntos_][0];
  const stream = controller[controlledReadableByteStream_];
  if (stream[shared.state_] === "closed") {
    if (bytesWritten !== 0) {
      throw new TypeError();
    }
    readableByteStreamControllerRespondInClosedState(
      controller,
      firstDescriptor
    );
  } else {
    // Assert: stream.[[state]] is "readable".
    readableByteStreamControllerRespondInReadableState(
      controller,
      bytesWritten,
      firstDescriptor
    );
  }
  readableByteStreamControllerCallPullIfNeeded(controller);
}

export function readableByteStreamControllerRespondWithNewView(
  controller: SDReadableByteStreamController,
  view: ArrayBufferView
): void {
  // Assert: controller.[[pendingPullIntos]] is not empty.
  const firstDescriptor = controller[pendingPullIntos_][0];
  if (
    firstDescriptor.byteOffset + firstDescriptor.bytesFilled !==
    view.byteOffset
  ) {
    throw new RangeError();
  }
  if (firstDescriptor.byteLength !== view.byteLength) {
    throw new RangeError();
  }
  firstDescriptor.buffer = view.buffer;
  readableByteStreamControllerRespondInternal(controller, view.byteLength);
}

export function readableByteStreamControllerShiftPendingPullInto(
  controller: SDReadableByteStreamController
): PullIntoDescriptor | undefined {
  const descriptor = controller[pendingPullIntos_].shift();
  readableByteStreamControllerInvalidateBYOBRequest(controller);
  return descriptor;
}

export function readableByteStreamControllerShouldCallPull(
  controller: SDReadableByteStreamController
): boolean {
  // Let stream be controller.[[controlledReadableByteStream]].
  const stream = controller[controlledReadableByteStream_];
  if (stream[shared.state_] !== "readable") {
    return false;
  }
  if (controller[closeRequested_]) {
    return false;
  }
  if (!controller[started_]) {
    return false;
  }
  if (
    readableStreamHasDefaultReader(stream) &&
    readableStreamGetNumReadRequests(stream) > 0
  ) {
    return true;
  }
  if (
    readableStreamHasBYOBReader(stream) &&
    readableStreamGetNumReadIntoRequests(stream) > 0
  ) {
    return true;
  }
  const desiredSize = readableByteStreamControllerGetDesiredSize(controller);
  // Assert: desiredSize is not null.
  return desiredSize! > 0;
}

export function setUpReadableStreamBYOBRequest(
  request: SDReadableStreamBYOBRequest,
  controller: SDReadableByteStreamController,
  view: ArrayBufferView
): void {
  if (!isReadableByteStreamController(controller)) {
    throw new TypeError();
  }
  if (!ArrayBuffer.isView(view)) {
    throw new TypeError();
  }
  // Assert: !IsDetachedBuffer(view.[[ViewedArrayBuffer]]) is false.

  request[associatedReadableByteStreamController_] = controller;
  request[view_] = view;
}