import { IsReadableStreamDefaultReader , ReadableStreamDefaultReader , ReadableStreamDefaultReaderRead , ReadableStreamReader , ReadableStreamReaderGenericRelease } from "./readable_stream_reader.ts" ; import { Assert , isAbortSignal } from "./util.ts" ; import { ReadableStreamController , ReadableStreamDefaultController , ReadableStreamDefaultControllerClose , ReadableStreamDefaultControllerEnqueue , ReadableStreamDefaultControllerError , SetUpReadableStreamDefaultController , SetUpReadableStreamDefaultControllerFromUnderlyingSource } from "./readable_stream_controller.ts" ; import { IsNonNegativeNumber , MakeSizeAlgorithmFromSizeFunction , ValidateAndNormalizeHighWaterMark } from "./misc.ts" ; import { defer } from "./defer.ts" ; import { AcquireWritableStreamDefaultWriter , IsWritableStream , IsWritableStreamLocked , WritableStream , WritableStreamAbort , WritableStreamCloseQueuedOrInFlight } from "./writable_stream.ts" ; import { WritableStreamDefaultWriterCloseWithErrorPropagation , WritableStreamDefaultWriterGetDesiredSize , WritableStreamDefaultWriterRelease } from "./writable_stream_writer.ts" ; import { QueuingStrategy } from "./strategy.ts" ; import { IsReadableByteStreamController , SetUpReadableByteStreamController , SetUpReadableByteStreamControllerFromUnderlyingSource , ReadableByteStreamController } from "./readable_byte_stream_controller.ts" ; import { IsReadableStreamBYOBReader , ReadableStreamBYOBReader } from "./readable_stream_byob_reader.ts" ; export type UnderlyingSource < T = any > = { type ? : "bytes" ; autoAllocateChunkSize ? : number ; start ? : ( controller : ReadableStreamController < T > ) => any ; pull ? : ( controller : ReadableStreamController < T > ) => any ; cancel ? : CancelAlgorithm ; } ; export type StartAlgorithm = ( ) => any ; export type PullAlgorithm = ( ) => Promise < any > ; export type CancelAlgorithm = ( reason ) => Promise < any > ; export type SizeAlgorithm = ( chunk ) => number ; export type ReadableStreamReadResult < T > = { value : T ; done : boolean ; } ; export class ReadableStream < T = any > { disturbed : boolean ; readableStreamController : ReadableByteStreamController | ReadableStreamDefaultController < T > ; reader : ReadableStreamDefaultReader | ReadableStreamBYOBReader ; state : "readable" | "closed" | "errored" ; storedError : Error ; constructor ( underlyingSource = { } , strategy = { } ) { InitializeReadableStream ( this ) ; let { highWaterMark , size } = strategy ; const { type } = underlyingSource ; if ( type === "bytes" ) { if ( size !== void 0 ) { throw new RangeError ( ) ; } if ( highWaterMark === void 0 ) { highWaterMark = 0 ; } highWaterMark = ValidateAndNormalizeHighWaterMark ( highWaterMark ) ; SetUpReadableByteStreamControllerFromUnderlyingSource ( this , underlyingSource , highWaterMark ) ; } else if ( type === void 0 ) { const sizeAlgorithm = MakeSizeAlgorithmFromSizeFunction ( size ) ; if ( highWaterMark === void 0 ) { highWaterMark = 0 ; } highWaterMark = ValidateAndNormalizeHighWaterMark ( highWaterMark ) ; SetUpReadableStreamDefaultControllerFromUnderlyingSource ( { stream : this , underlyingSource , highWaterMark , sizeAlgorithm } ) ; } else { throw new RangeError ( ) ; } } get locked ( ) : boolean { if ( ! IsReadableStream ( this ) ) { throw new TypeError ( ) ; } return IsReadableStreamLocked ( this ) ; } cancel ( reason ? ) { if ( IsReadableStream ( this ) ) { return Promise . reject ( new TypeError ( ) ) ; } if ( IsReadableStreamLocked ( this ) ) { return Promise . reject ( new TypeError ( ) ) ; } return ReadableStreamCancel ( this , reason ) ; } getReader ( params = { } ) { if ( ! IsReadableStream ( this ) ) { throw new TypeError ( ) ; } if ( params . mode === void 0 ) { return AcquireReadableStreamDefaultReader ( this ) ; } if ( params . mode === "byob" ) { return AcquireReadableStreamBYOBReader ( this ) ; } throw new RangeError ( ) ; } pipeThrough < T > ( { writable , readable } , { preventClose , preventAbort , preventCancel , signal } = { } ) { if ( ! IsReadableStream ( this ) ) { throw new TypeError ( "this is not ReadableStream" ) ; } if ( ! IsWritableStream ( writable ) ) { throw new TypeError ( "writable is not WritableStream" ) ; } if ( ! IsReadableStream ( readable ) ) { throw new TypeError ( "readable is not ReadableStream" ) ; } preventClose = ! ! preventClose ; preventAbort = ! ! preventAbort ; preventCancel = ! ! preventCancel ; if ( signal !== void 0 && ! isAbortSignal ( signal ) ) { throw new TypeError ( "signal is not instance of AbortSignal" ) ; } if ( IsReadableStreamLocked ( this ) ) { throw new TypeError ( "this stream is locked" ) ; } if ( IsWritableStreamLocked ( writable ) ) { throw new TypeError ( "writable is locked" ) ; } ReadableStreamPipeTo ( this , writable , preventClose , preventAbort , preventCancel , signal ) ; return readable ; } async pipeTo ( dest , { preventClose , preventAbort , preventCancel , signal } = { } ) { if ( ! IsReadableStream ( this ) ) { throw new TypeError ( "this is not ReadableStream" ) ; } if ( ! IsWritableStream ( dest ) ) { throw new TypeError ( "dest is not WritableStream" ) ; } preventClose = ! ! preventClose ; preventAbort = ! ! preventAbort ; preventCancel = ! ! preventCancel ; if ( signal !== void 0 && ! isAbortSignal ( signal ) ) { throw new TypeError ( "signal is not instance of AbortSignal" ) ; } if ( IsReadableStreamLocked ( this ) ) { throw new TypeError ( "this stream is locked" ) ; } if ( IsWritableStreamLocked ( dest ) ) { throw new TypeError ( "writable is locked" ) ; } return ReadableStreamPipeTo ( this , dest , preventClose , preventCancel , preventAbort , signal ) ; } tee ( ) { if ( ! IsReadableStream ( this ) ) { throw new TypeError ( ) ; } return ReadableStreamTee ( this , false ) ; } } function AcquireReadableStreamBYOBReader ( stream ) { return new ReadableStreamBYOBReader ( stream ) ; } function AcquireReadableStreamDefaultReader ( stream ) { return new ReadableStreamDefaultReader ( stream ) ; } export function CreateReadableStream ( startAlgorithm , pullAlgorithm , cancelAlgorithm , highWaterMark = 1 , sizeAlgorithm = ( ) => 1 ) { Assert ( IsNonNegativeNumber ( highWaterMark ) ) ; const stream = Object . create ( ReadableStream . prototype ) ; InitializeReadableStream ( stream ) ; const controller = Object . create ( ReadableStreamDefaultController . prototype ) ; SetUpReadableStreamDefaultController ( { stream , controller , startAlgorithm , pullAlgorithm , cancelAlgorithm , highWaterMark , sizeAlgorithm } ) ; return stream ; } export function CreateReadableByteStream ( startAlgorithm , pullAlgorithm , cancelAlgorithm , highWaterMark = 1 , autoAllocateChunkSize ? ) { Assert ( IsNonNegativeNumber ( highWaterMark ) ) ; if ( autoAllocateChunkSize !== void 0 ) { Assert ( Number . isInteger ( autoAllocateChunkSize ) ) ; Assert ( autoAllocateChunkSize > 0 ) ; } const stream = Object . create ( ReadableStream . prototype ) ; InitializeReadableStream ( stream ) ; const controller = Object . create ( ReadableByteStreamController . prototype ) ; SetUpReadableByteStreamController ( stream , controller , startAlgorithm , pullAlgorithm , cancelAlgorithm , highWaterMark , autoAllocateChunkSize ) ; return stream ; } export function InitializeReadableStream ( stream ) { stream . state = "readable" ; stream . reader = void 0 ; stream . storedError = void 0 ; stream . disturbed = false ; } export function IsReadableStream ( x ) { return typeof x === "object" && x . hasOwnProperty ( "readableStreamController" ) ; } export function IsReadableStreamDisturbed ( stream ) { Assert ( IsReadableStream ( stream ) ) ; return stream . disturbed ; } export function IsReadableStreamLocked ( stream ) { Assert ( IsReadableStream ( stream ) ) ; return stream . reader !== void 0 ; } export function ReadableStreamTee ( stream , cloneForBranch2 ) { Assert ( IsReadableStream ( stream ) ) ; Assert ( typeof cloneForBranch2 === "boolean" ) ; const reader = AcquireReadableStreamDefaultReader ( stream ) ; let closedOrErrored = false ; let canceled1 = false ; let canceled2 = false ; let reason1 = void 0 ; let reason2 = void 0 ; let branch1 = void 0 ; let branch2 = void 0 ; let cancelPromise = defer ( ) ; const pullAlgorithm = ( ) => { return ReadableStreamDefaultReaderRead ( reader ) . then ( ( result ) => { Assert ( typeof result === "object" ) ; const { value , done } = result ; Assert ( typeof done === "boolean" ) ; if ( done && ! closedOrErrored ) { if ( ! canceled1 ) { ReadableStreamDefaultControllerClose ( branch1 . readableStreamController ) ; } if ( ! canceled2 ) { ReadableStreamDefaultControllerClose ( branch2 . readableStreamController ) ; } } if ( closedOrErrored ) { return ; } let [ value1 , value2 ] = [ value , value ] ; if ( ! canceled2 && cloneForBranch2 ) { } if ( ! canceled1 ) { ReadableStreamDefaultControllerEnqueue ( branch1 . readableStreamController , value1 ) ; } if ( ! canceled2 ) { ReadableStreamDefaultControllerEnqueue ( branch1 . readableStreamController , value2 ) ; } } ) ; } ; const cancel1Algorithm = reason => { canceled1 = true ; reason1 = reason ; if ( canceled2 ) { const compositeReason = [ reason1 , reason2 ] ; const cancelResult = ReadableStreamCancel ( stream , compositeReason ) ; cancelPromise . resolve ( cancelResult ) ; } return cancelPromise ; } ; const cancel2Algorithm = reason => { canceled2 = true ; reason2 = reason ; if ( canceled1 ) { const compositeReason = [ reason1 , reason2 ] ; const cancelResult = ReadableStreamCancel ( stream , compositeReason ) ; cancelPromise . resolve ( cancelResult ) ; } return cancelPromise ; } ; const startAlgorithm = ( ) => void 0 ; branch1 = CreateReadableStream ( startAlgorithm , pullAlgorithm , cancel1Algorithm ) ; branch2 = CreateReadableStream ( startAlgorithm , pullAlgorithm , cancel2Algorithm ) ; reader . closedPromise . catch ( r => { if ( ! closedOrErrored ) { ReadableStreamDefaultControllerError ( branch1 . readableStreamController , r ) ; ReadableStreamDefaultControllerError ( branch2 . readableStreamController , r ) ; closedOrErrored = true ; } } ) ; return [ branch1 , branch2 ] ; } export async function ReadableStreamPipeTo ( source , dest , preventClose , preventAbort , preventCancel , signal ? ) { Assert ( IsReadableStream ( source ) ) ; Assert ( IsWritableStream ( dest ) ) ; Assert ( typeof preventCancel === "boolean" ) ; Assert ( typeof preventAbort === "boolean" ) ; Assert ( typeof preventClose === "boolean" ) ; Assert ( signal === void 0 || isAbortSignal ( signal ) ) ; Assert ( ! IsReadableStreamLocked ( source ) ) ; Assert ( ! IsWritableStreamLocked ( dest ) ) ; let reader ; if ( IsReadableByteStreamController ( source . readableStreamController ) ) { reader = AcquireReadableStreamBYOBReader ( source ) ; } else { reader = AcquireReadableStreamDefaultReader ( source ) ; } const writer = AcquireWritableStreamDefaultWriter ( dest ) ; let shutingDown = false ; const promsie = defer ( ) ; let abortAlgorithm ; if ( ! signal ) { abortAlgorithm = ( ) => { let error = new Error ( "aborted" ) ; const actions = [ ] ; if ( ! preventAbort ) { actions . push ( async ( ) => { if ( dest . state === "writable" ) { return WritableStreamAbort ( dest , error ) ; } } ) ; } if ( ! preventCancel ) { actions . push ( async ( ) => { if ( source . state === "readable" ) { return ReadableStreamCancel ( source , error ) ; } } ) ; } shutdown ( error , ( ) => Promise . all ( actions . map ( p => p ( ) ) ) ) ; if ( signal . aborted ) { abortAlgorithm ( ) ; return promsie ; } signal . addEventListener ( "onabort" , abortAlgorithm ) ; } ; } const finalize = ( error ? ) => { WritableStreamDefaultWriterRelease ( writer ) ; ReadableStreamReaderGenericRelease ( reader ) ; if ( signal ) { signal . removeEventListener ( "onabort" , abortAlgorithm ) ; } if ( error ) { promsie . reject ( error ) ; } else { promsie . resolve ( ) ; } } ; const shutdown = ( err ? , action ? ) => { if ( shutingDown ) { return ; } shutingDown = true ; if ( dest . state === "writable" || ! WritableStreamCloseQueuedOrInFlight ( dest ) ) { } if ( ! action ) { finalize ( err ) ; return ; } action ( ) . then ( ( ) => finalize ( err ) ) . catch ( finalize ) ; } ; ( async ( ) => { while ( true ) { const desiredSize = WritableStreamDefaultWriterGetDesiredSize ( writer ) ; if ( desiredSize === null || desiredSize <= 0 ) { return ; } if ( source . state === "errored" ) { if ( ! preventAbort ) { shutdown ( source . storedError , ( ) => { return WritableStreamAbort ( dest , source . storedError ) ; } ) ; } else { shutdown ( source . storedError ) ; } } else if ( dest . state === "errored" ) { if ( ! preventCancel ) { shutdown ( dest . storedError , ( ) => { return ReadableStreamCancel ( source , dest . storedError ) ; } ) ; } else { shutdown ( dest . storedError ) ; } } else if ( source . state === "closed" ) { if ( ! preventClose ) { shutdown ( void 0 , ( ) => { return WritableStreamDefaultWriterCloseWithErrorPropagation ( writer ) ; } ) ; } else { shutdown ( ) ; } } else if ( WritableStreamCloseQueuedOrInFlight ( dest ) || dest . state === "closed" ) { const destClosed = new TypeError ( ) ; if ( ! preventCancel ) { shutdown ( destClosed , ( ) => { return ReadableStreamCancel ( source , destClosed ) ; } ) ; } else { shutdown ( destClosed ) ; } } if ( IsReadableStreamBYOBReader ( reader ) ) { let view = new Uint8Array ( desiredSize ) ; const { done } = await reader . read ( view ) ; if ( done ) break ; await writer . write ( view ) ; } else { const { value , done } = await reader . read ( ) ; if ( done ) break ; await writer . write ( value ) ; } } } ) ( ) ; return promsie ; } export function ReadableStreamAddReadIntoRequest ( stream , forAuthorCode ) { Assert ( IsReadableStreamBYOBReader ( stream . reader ) ) ; const reader = stream . reader as ReadableStreamBYOBReader ; Assert ( stream . state === "readable" || stream . state === "closed" ) ; const promise = defer ( ) ; const readIntoRequest = { promise , forAuthorCode } ; reader . readIntoRequests . push ( readIntoRequest ) ; return promise ; } export function ReadableStreamAddReadRequest ( stream , forAuthorCode ) { Assert ( IsReadableStreamDefaultReader ( stream . reader ) ) ; const reader = stream . reader as ReadableStreamDefaultReader ; Assert ( stream . state === "readable" || stream . state === "closed" ) ; const promise = defer < { value ; done : boolean ; } > ( ) ; const readIntoRequest = { promise , forAuthorCode } ; reader . readRequests . push ( readIntoRequest ) ; return promise ; } export function ReadableStreamCancel ( stream , reason ) { stream . disturbed = true ; if ( stream . state === "closed" ) { return Promise . reject ( void 0 ) ; } if ( stream . state === "errored" ) { return Promise . reject ( stream . storedError ) ; } ReadableStreamClose ( stream ) ; const sourceCancelPromise = stream . readableStreamController . cancelAlgorithm ( reason ) ; return sourceCancelPromise . then ( ( ) => void 0 ) ; } export function ReadableStreamClose ( stream ) { Assert ( stream . state === "readable" ) ; stream . state = "closed" ; const reader = stream . reader ; if ( reader === void 0 ) { return ; } if ( IsReadableStreamDefaultReader ( reader ) ) { for ( let req of reader . readRequests ) { const resolved = ReadableStreamCreateReadResult ( void 0 , true , req . forAuthorCode ) ; req . promise . resolve ( resolved ) ; } reader . readRequests = [ ] ; } reader . closedPromise . resolve ( void 0 ) ; } export function ReadableStreamCreateReadResult < T > ( value , done , forAuthorCode ) { const ret = forAuthorCode ? Object . create ( { } ) : Object . create ( null ) ; ret [ "value" ] = value as T ; ret [ "done" ] = done ; return { value , done } ; } export function ReadableStreamError ( stream , e ) { Assert ( IsReadableStream ( stream ) ) ; Assert ( stream . state === "readable" ) ; stream . state = "errored" ; stream . storedError = e ; const reader = stream . reader ; if ( stream . reader === void 0 ) { return ; } if ( IsReadableStreamDefaultReader ( reader ) ) { for ( const req of reader . readRequests ) { req . promise . reject ( e ) ; } reader . readRequests = [ ] ; } else if ( IsReadableStreamBYOBReader ( reader ) ) { for ( const req of reader . readIntoRequests ) { req . promise . reject ( e ) ; } reader . readIntoRequests = [ ] ; } reader . closedPromise . reject ( e ) ; } export function ReadableStreamFulfillReadIntoRequest ( stream , chunk , done ) { const reader = stream . reader ; const req = ( < ReadableStreamBYOBReader > reader ) . readIntoRequests . shift ( ) ; req . promise . resolve ( ReadableStreamCreateReadResult ( chunk , done , req . forAuthorCode ) ) ; } export function ReadableStreamFulfillReadRequest < T > ( stream , chunk , done ) { const reader = stream . reader ; const req = ( < ReadableStreamDefaultReader < T > > reader ) . readRequests . shift ( ) ; req . promise . resolve ( ReadableStreamCreateReadResult ( chunk , done , req . forAuthorCode ) ) ; } export function ReadableStreamGetNumReadIntoRequests ( stream ) { return ( < ReadableStreamBYOBReader > stream . reader ) . readIntoRequests . length ; } export function ReadableStreamGetNumReadRequests < T > ( stream ) { return ( < ReadableStreamDefaultReader < T > > stream . reader ) . readRequests . length ; } export function ReadableStreamHasBYOBReader ( stream ) { const reader = stream . reader ; if ( reader === void 0 ) { return false ; } return IsReadableStreamBYOBReader ( reader ) ; } export function ReadableStreamHasDefaultReader ( stream ) { const reader = stream . reader ; if ( reader === void 0 ) { return false ; } return IsReadableStreamDefaultReader ( reader ) ; }