// Forked from https://github.com/stardazed/sd-streams/tree/8928cf04b035fd02fb1340b7eb541c76be37e546 // Copyright (c) 2018-Present by Arthur Langereis - @zenmumbler MIT /* 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.d.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 = ( controller: SDReadableStreamControllerBase ) => void | PromiseLike; export type StartAlgorithm = () => Promise | void; export type PullFunction = ( controller: SDReadableStreamControllerBase ) => void | PromiseLike; export type PullAlgorithm = ( controller: SDReadableStreamControllerBase ) => PromiseLike; export type CancelAlgorithm = (reason?: shared.ErrorResult) => Promise; // ---- export interface SDReadableStreamControllerBase { readonly desiredSize: number | null; close(): void; error(e?: shared.ErrorResult): void; [cancelSteps_](reason: shared.ErrorResult): Promise; [pullSteps_](forAuthorCode: boolean): Promise>; } 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, 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; // 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; // 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 extends SDReadableStreamControllerBase, q.QueueContainer { enqueue(chunk?: OutputType): void; [controlledReadableStream_]: SDReadableStream; [pullAlgorithm_]: PullAlgorithm; [cancelAlgorithm_]: CancelAlgorithm; [strategySizeAlgorithm_]: QueuingStrategySizeCallback; [strategyHWM_]: number; [started_]: boolean; [closeRequested_]: boolean; [pullAgain_]: boolean; [pulling_]: boolean; } // ---- export interface SDReadableStreamReader { readonly closed: Promise; cancel(reason: shared.ErrorResult): Promise; releaseLock(): void; [ownerReadableStream_]: SDReadableStream | undefined; [closedPromise_]: shared.ControlledPromise; } export interface ReadRequest extends shared.ControlledPromise { forAuthorCode: boolean; } export declare class SDReadableStreamDefaultReader implements SDReadableStreamReader { constructor(stream: SDReadableStream); readonly closed: Promise; cancel(reason: shared.ErrorResult): Promise; releaseLock(): void; read(): Promise>; [ownerReadableStream_]: SDReadableStream | undefined; [closedPromise_]: shared.ControlledPromise; [readRequests_]: Array>>; } export declare class SDReadableStreamBYOBReader implements SDReadableStreamReader { constructor(stream: SDReadableStream); readonly closed: Promise; cancel(reason: shared.ErrorResult): Promise; releaseLock(): void; read(view: ArrayBufferView): Promise>; [ownerReadableStream_]: SDReadableStream | undefined; [closedPromise_]: shared.ControlledPromise; [readIntoRequests_]: Array>>; } /* TODO reenable this when we add WritableStreams and Transforms export interface GenericTransformStream { readable: SDReadableStream; writable: ws.WritableStream; } */ export type ReadableStreamState = "readable" | "closed" | "errored"; export declare class SDReadableStream { constructor( underlyingSource: UnderlyingByteSource, strategy?: { highWaterMark?: number; size?: undefined } ); constructor( underlyingSource?: UnderlyingSource, strategy?: QueuingStrategy ); readonly locked: boolean; cancel(reason?: shared.ErrorResult): Promise; getReader(): SDReadableStreamReader; getReader(options: { mode: "byob" }): SDReadableStreamBYOBReader; tee(): Array>; /* TODO reenable these methods when we bring in writableStreams and transport types pipeThrough( transform: GenericTransformStream, options?: PipeOptions ): SDReadableStream; pipeTo( dest: ws.WritableStream, options?: PipeOptions ): Promise; */ [shared.state_]: ReadableStreamState; [shared.storedError_]: shared.ErrorResult; [reader_]: SDReadableStreamReader | undefined; [readableStreamController_]: SDReadableStreamControllerBase; } // ---- Stream export function initializeReadableStream( stream: SDReadableStream ): 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 { if (typeof value !== "object" || value === null) { return false; } return readableStreamController_ in value; } export function isReadableStreamLocked( stream: SDReadableStream ): boolean { return stream[reader_] !== undefined; } export function readableStreamGetNumReadIntoRequests( stream: SDReadableStream ): number { // TODO remove the "as unknown" cast // This is in to workaround a compiler error // error TS2352: Conversion of type 'SDReadableStreamReader' 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' 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( stream: SDReadableStream ): number { const reader = stream[reader_] as SDReadableStreamDefaultReader; if (reader === undefined) { return 0; } return reader[readRequests_].length; } export function readableStreamCreateReadResult( value: T, done: boolean, forAuthorCode: boolean ): IteratorResult { const prototype = forAuthorCode ? Object.prototype : null; const result = Object.create(prototype); result.value = value; result.done = done; return result; } export function readableStreamAddReadIntoRequest( stream: SDReadableStream, forAuthorCode: boolean ): Promise> { // Assert: ! IsReadableStreamBYOBReader(stream.[[reader]]) is true. // Assert: stream.[[state]] is "readable" or "closed". const reader = stream[reader_] as SDReadableStreamBYOBReader; const conProm = shared.createControlledPromise< IteratorResult >() as ReadRequest>; conProm.forAuthorCode = forAuthorCode; reader[readIntoRequests_].push(conProm); return conProm.promise; } export function readableStreamAddReadRequest( stream: SDReadableStream, forAuthorCode: boolean ): Promise> { // Assert: ! IsReadableStreamDefaultReader(stream.[[reader]]) is true. // Assert: stream.[[state]] is "readable". const reader = stream[reader_] as SDReadableStreamDefaultReader; const conProm = shared.createControlledPromise< IteratorResult >() as ReadRequest>; conProm.forAuthorCode = forAuthorCode; reader[readRequests_].push(conProm); return conProm.promise; } export function readableStreamHasBYOBReader( stream: SDReadableStream ): boolean { const reader = stream[reader_]; return isReadableStreamBYOBReader(reader); } export function readableStreamHasDefaultReader( stream: SDReadableStream ): boolean { const reader = stream[reader_]; return isReadableStreamDefaultReader(reader); } export function readableStreamCancel( stream: SDReadableStream, reason: shared.ErrorResult ): Promise { 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( stream: SDReadableStream ): 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( stream: SDReadableStream, 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 { 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( reader: SDReadableStreamReader, stream: SDReadableStream ): void { reader[ownerReadableStream_] = stream; stream[reader_] = reader; const streamState = stream[shared.state_]; reader[closedPromise_] = shared.createControlledPromise(); 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( reader: SDReadableStreamReader ): 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(); } 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> { 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( reader: SDReadableStreamDefaultReader, forAuthorCode = false ): Promise> { 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( stream: SDReadableStream, 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( stream: SDReadableStream, chunk: OutputType, done: boolean ): void { const reader = stream[reader_] as SDReadableStreamDefaultReader; const readRequest = reader[readRequests_].shift()!; // <-- length check done in caller readRequest.resolve( readableStreamCreateReadResult(chunk, done, readRequest.forAuthorCode) ); } // ---- DefaultController export function setUpReadableStreamDefaultController( stream: SDReadableStream, controller: SDReadableStreamDefaultController, startAlgorithm: StartAlgorithm, pullAlgorithm: PullAlgorithm, cancelAlgorithm: CancelAlgorithm, highWaterMark: number, sizeAlgorithm: QueuingStrategySizeCallback ): 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 { if (typeof value !== "object" || value === null) { return false; } return controlledReadableStream_ in value; } export function readableStreamDefaultControllerHasBackpressure( controller: SDReadableStreamDefaultController ): boolean { return !readableStreamDefaultControllerShouldCallPull(controller); } export function readableStreamDefaultControllerCanCloseOrEnqueue( controller: SDReadableStreamDefaultController ): boolean { const state = controller[controlledReadableStream_][shared.state_]; return controller[closeRequested_] === false && state === "readable"; } export function readableStreamDefaultControllerGetDesiredSize( controller: SDReadableStreamDefaultController ): 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( controller: SDReadableStreamDefaultController ): 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( controller: SDReadableStreamDefaultController, 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( controller: SDReadableStreamDefaultController, 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( controller: SDReadableStreamDefaultController ): 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( controller: SDReadableStreamDefaultController ): 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( controller: SDReadableStreamDefaultController ): void { controller[pullAlgorithm_] = undefined!; controller[cancelAlgorithm_] = undefined!; controller[strategySizeAlgorithm_] = undefined!; } // ---- BYOBController export function setUpReadableByteStreamController( stream: SDReadableStream, controller: SDReadableByteStreamController, startAlgorithm: StartAlgorithm, pullAlgorithm: PullAlgorithm, 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, 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> { 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; }