import { Assert } from "./util.ts" ; import { ReadableStreamBYOBRequest } from "./readable_stream_request.ts" ; import { CancelAlgorithm , IsReadableStreamLocked , PullAlgorithm , ReadableStream , ReadableStreamAddReadRequest , ReadableStreamClose , ReadableStreamCreateReadResult , ReadableStreamError , ReadableStreamFulfillReadRequest , ReadableStreamGetNumReadRequests , SizeAlgorithm , StartAlgorithm , UnderlyingSource } from "./readable_stream.ts" ; import { CreateAlgorithmFromUnderlyingMethod , DequeueValue , EnqueueValueWithSize , InvokeOrNoop , ResetQueue } from "./misc.ts" ; export type PullIntoDescriptor = { buffer : ArrayBuffer ; byteOffset : number ; bytesFilled : number ; byteLength : number ; elementSize : number ; ctor : any ; readerType : string ; } ; export interface ReadableStreamController < T > { readonly byobRequest ? : ReadableStreamBYOBRequest ; readonly desiredSize : number ; close ( ) : void ; enqueue ( chunk ) : void ; error ( e ) : void ; } export abstract class ReadableStreamControllerBase { autoAllocateChunkSize : number ; cancelAlgorithm : CancelAlgorithm ; closeRequested : boolean ; pullAgain : boolean ; pullAlgorithm : PullAlgorithm ; pulling : boolean ; pendingPullIntos : PullIntoDescriptor [ ] ; queue : { buffer : ArrayBuffer ; byteLength : number ; byteOffset : number ; } [ ] ; queueTotalSize ; started : boolean ; strategyHWM : number ; } export class ReadableStreamDefaultController < T > extends ReadableStreamControllerBase implements ReadableStreamController < T > { constructor ( ) { super ( ) ; throw new TypeError ( ) ; } controlledReadableStream : ReadableStream ; strategySizeAlgorithm : ( chunk ) => number ; get desiredSize ( ) : number { if ( ! IsReadableStreamDefaultController ( this ) ) { throw new TypeError ( ) ; } return ReadableStreamDefaultControllerGetDesiredSize ( this ) ; } close ( ) { if ( ! IsReadableStreamDefaultController ( this ) ) { throw new TypeError ( ) ; } if ( ! ReadableStreamDefaultControllerCanCloseOrEnqueue ( this ) ) { throw new TypeError ( ) ; } ReadableStreamDefaultControllerClose ( this ) ; } enqueue ( chunk ) { if ( ! IsReadableStreamDefaultController ( this ) ) { throw new TypeError ( ) ; } if ( ! ReadableStreamDefaultControllerCanCloseOrEnqueue ( this ) ) { throw new TypeError ( ) ; } return ReadableStreamDefaultControllerEnqueue ( this , chunk ) ; } error ( e ) { if ( ! IsReadableStreamDefaultController ( this ) ) { throw new TypeError ( ) ; } ReadableStreamDefaultControllerError ( this , e ) ; } CancelSteps ( reason ) { ResetQueue ( this ) ; const result = this . cancelAlgorithm ( reason ) ; ReadableStreamDefaultControllerClearAlgorithms ( this ) ; return result ; } PullSteps ( forAuthorCode ? ) { const stream = this . controlledReadableStream ; if ( this . queue . length > 0 ) { const chunk = DequeueValue ( this ) ; if ( this . closeRequested && this . queue . length === 0 ) { ReadableStreamDefaultControllerClearAlgorithms ( this ) ; ReadableStreamClose ( stream ) ; } else { ReadableStreamDefaultControllerCallPullIfNeeded ( this ) ; } return Promise . resolve ( ReadableStreamCreateReadResult ( chunk , false , forAuthorCode ) ) ; } const pendingPromise = ReadableStreamAddReadRequest ( stream , forAuthorCode ) ; ReadableStreamDefaultControllerCallPullIfNeeded ( this ) ; return pendingPromise ; } } export function IsReadableStreamDefaultController < T > ( x ) { return typeof x === "object" && x . hasOwnProperty ( "controlledReadableStream" ) ; } export function ReadableStreamDefaultControllerCallPullIfNeeded < T > ( controller ) { const shouldPull = ReadableStreamDefaultControllerShouldCallPull ( controller ) ; if ( ! shouldPull ) { return ; } if ( controller . pulling ) { controller . pullAgain = true ; return ; } Assert ( ! controller . pullAgain ) ; controller . pulling = true ; controller . pullAlgorithm ( ) . then ( ( ) => { controller . pulling = false ; if ( controller . pullAgain ) { controller . pullAgain = false ; ReadableStreamDefaultControllerCallPullIfNeeded ( controller ) ; } } ) . catch ( r => { ReadableStreamDefaultControllerError ( controller , r ) ; } ) ; } export function ReadableStreamDefaultControllerShouldCallPull < T > ( controller ) { const stream = controller . controlledReadableStream ; if ( ! ReadableStreamDefaultControllerCanCloseOrEnqueue ( controller ) ) { return false ; } if ( ! controller . started ) { return false ; } if ( IsReadableStreamLocked ( stream ) && ReadableStreamGetNumReadRequests ( stream ) > 0 ) { return true ; } const desiredSize = ReadableStreamDefaultControllerGetDesiredSize ( controller ) ; Assert ( desiredSize !== null ) ; return desiredSize > 0 ; } export function ReadableStreamDefaultControllerClearAlgorithms < T > ( controller ) { controller . pullAlgorithm = void 0 ; controller . cancelAlgorithm = void 0 ; controller . strategySizeAlgorithm = void 0 ; } export function ReadableStreamDefaultControllerClose < T > ( controller ) { const stream = controller . controlledReadableStream ; Assert ( ReadableStreamDefaultControllerCanCloseOrEnqueue ( controller ) ) ; controller . closeRequested = true ; if ( controller . queue . length === 0 ) { ReadableStreamDefaultControllerClearAlgorithms ( controller ) ; ReadableStreamClose ( stream ) ; } } export function ReadableStreamDefaultControllerEnqueue < T > ( controller , chunk ) { if ( IsReadableStreamDefaultController ( controller ) ) { const stream = controller . controlledReadableStream ; Assert ( ReadableStreamDefaultControllerCanCloseOrEnqueue ( controller ) ) ; if ( IsReadableStreamLocked ( stream ) && ReadableStreamGetNumReadRequests ( stream ) > 0 ) { ReadableStreamFulfillReadRequest ( stream , chunk , false ) ; } else { let result ; try { result = controller . strategySizeAlgorithm ( chunk ) ; } catch ( e ) { ReadableStreamDefaultControllerError ( controller , e ) ; return e ; } const chunkSize = result ; try { EnqueueValueWithSize ( controller , chunk , chunkSize ) ; } catch ( e ) { ReadableStreamDefaultControllerError ( controller , e ) ; return e ; } ReadableStreamDefaultControllerCallPullIfNeeded ( controller ) ; } } } export function ReadableStreamDefaultControllerError < T > ( controller , e ) { if ( IsReadableStreamDefaultController ( controller ) ) { const stream = controller . controlledReadableStream ; if ( stream . state !== "readable" ) { return ; } ResetQueue ( controller ) ; ReadableStreamDefaultControllerClearAlgorithms ( controller ) ; ReadableStreamError ( stream , e ) ; } } export function ReadableStreamDefaultControllerGetDesiredSize < T > ( controller ) { const stream = controller . controlledReadableStream ; const state = stream . state ; if ( state === "errored" ) { return null ; } if ( state === "closed" ) { return 0 ; } return controller . strategyHWM - controller . queueTotalSize ; } export function ReadableStreamDefaultControllerHasBackpressure < T > ( controller ) { return ! ReadableStreamDefaultControllerShouldCallPull ( controller ) ; } export function ReadableStreamDefaultControllerCanCloseOrEnqueue < T > ( controller ) { const state = controller . controlledReadableStream . state ; return ! controller . closeRequested && state === "readable" ; } export function SetUpReadableStreamDefaultController < T > ( params ) { const { stream , controller , startAlgorithm , pullAlgorithm , cancelAlgorithm } = params ; let { highWaterMark , sizeAlgorithm } = params ; Assert ( stream . readableStreamController === void 0 ) ; controller . controlledReadableStream = stream ; controller . queue = void 0 ; controller . queueTotalSize = void 0 ; 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 ; Promise . resolve ( startAlgorithm ( ) ) . then ( ( ) => { controller . started = true ; Assert ( controller . pulling == false ) ; Assert ( controller . pullAgain == false ) ; ReadableStreamDefaultControllerCallPullIfNeeded ( controller ) ; } ) . catch ( r => { ReadableStreamDefaultControllerError ( controller , r ) ; } ) ; } export function SetUpReadableStreamDefaultControllerFromUnderlyingSource ( params ) { const { stream , underlyingSource , highWaterMark , sizeAlgorithm } = params ; Assert ( underlyingSource !== void 0 ) ; const controller = Object . create ( ReadableStreamDefaultController . prototype ) ; const startAlgorithm = ( ) => InvokeOrNoop ( underlyingSource , "start" , controller ) ; const pullAlgorithm = CreateAlgorithmFromUnderlyingMethod ( underlyingSource , "pull" , 0 , controller ) ; const cancelAlgorithm = CreateAlgorithmFromUnderlyingMethod ( underlyingSource , "cancel" , 1 ) ; SetUpReadableStreamDefaultController ( { stream , controller , startAlgorithm , pullAlgorithm , cancelAlgorithm , highWaterMark , sizeAlgorithm } ) ; }