import { ReadableStreamDefaultControllerCanCloseOrEnqueue , ReadableStreamDefaultControllerClose , ReadableStreamDefaultControllerEnqueue , ReadableStreamDefaultControllerGetDesiredSize , ReadableStreamDefaultControllerHasBackpressure } from "./readable_stream_controller" ; import { FlushAlgorithm , IsTransformStream , TransformAlgorithm , Transformer , TransformStream , TransformStreamError , TransformStreamErrorWritableAndUnblockWrite , TransformStreamSetBackpressure } from "./transform_stream" ; import { Assert } from "./util" ; import { CreateAlgorithmFromUnderlyingMethod , PromiseCall } from "./misc" ; import { read , write } from "deno" ; export interface TransformStreamController < T > { readonly desiredSize : number ; enqueue ( chunk ) ; error ( reason ) ; terminate ( ) ; } export class TransformStreamDefaultController < T = any > implements TransformStreamController < T > { constructor ( ) { throw new TypeError ( "TransformStreamDefaultController is not constructable" ) ; } get desiredSize ( ) { if ( ! IsTransformStreamDefaultController ( this ) ) { throw new TypeError ( "this is not TransformStreamDefaultController" ) ; } return ReadableStreamDefaultControllerGetDesiredSize ( this . controlledTransformStream . readable . readableStreamController ) ; } enqueue ( chunk ) { if ( ! IsTransformStreamDefaultController ( this ) ) { throw new TypeError ( "this is not TransformStreamDefaultController" ) ; } TransformStreamDefaultControllerEnqueue ( this , chunk ) ; } error ( reason ) { if ( ! IsTransformStreamDefaultController ( this ) ) { throw new TypeError ( "this is not TransformStreamDefaultController" ) ; } TransformStreamDefaultControllerError ( this , reason ) ; } terminate ( ) { if ( ! IsTransformStreamDefaultController ( this ) ) { throw new TypeError ( "this is not TransformStreamDefaultController" ) ; } TransformStreamDefaultControllerTerminate ( this ) ; } controlledTransformStream : TransformStream < T > ; flushAlgorithm : FlushAlgorithm ; transformAlgorithm : TransformAlgorithm < T > ; } export function IsTransformStreamDefaultController ( x ) { return typeof x === "object" && x . hasOwnProperty ( "controlledTransformStream" ) ; } export function SetUpTransformStreamDefaultController < T > ( stream , controller , transformAlgorithm , flushAlgorithm ) { Assert ( IsTransformStream ( stream ) ) ; Assert ( stream . transformStreamController === void 0 ) ; controller . controlledTransformStream = stream ; stream . transformStreamController = controller ; controller . transformAlgorithm = transformAlgorithm ; controller . flushAlgorithm = flushAlgorithm ; } export function SetUpTransformStreamDefaultControllerFromTransformer < T > ( stream , transformer ) { Assert ( transformer !== void 0 ) ; const controller = Object . create ( TransformStreamDefaultController . prototype ) ; let transformAlgorithm = async ( chunk ) => { try { TransformStreamDefaultControllerEnqueue ( controller , chunk ) ; } catch ( e ) { throw void 0 ; } return ; } ; const method = transformer . transform ; if ( method !== void 0 ) { if ( typeof method . call !== "function" ) { throw new TypeError ( "transformer.transform is not callable" ) ; } transformAlgorithm = async ( chunk ) => PromiseCall ( method , transformer , chunk , controller ) ; } const flushAlgorithm = CreateAlgorithmFromUnderlyingMethod ( transformer , "flush" , 0 , controller ) ; SetUpTransformStreamDefaultController < T > ( stream , controller , transformAlgorithm , flushAlgorithm ) ; } export function TransformStreamDefaultControllerClearAlgorithms ( controller ) { controller . transformAlgorithm = void 0 ; controller . flushAlgorithm = void 0 ; } export function TransformStreamDefaultControllerEnqueue ( controller , chunk ) { const stream = controller . controlledTransformStream ; const readableController = stream . readable . readableStreamController ; if ( ! ReadableStreamDefaultControllerCanCloseOrEnqueue ( readableController ) ) { throw new TypeError ( "readable stream controller cannot close on enqueue" ) ; } try { ReadableStreamDefaultControllerEnqueue ( readableController , chunk ) ; } catch ( e ) { TransformStreamErrorWritableAndUnblockWrite ( stream , e ) ; throw stream . readable . storedError ; } const backpressure = ReadableStreamDefaultControllerHasBackpressure ( readableController ) ; if ( backpressure !== stream . backpressure ) { Assert ( backpressure ) ; TransformStreamSetBackpressure ( stream , true ) ; } } export function TransformStreamDefaultControllerError ( controller , e ) { TransformStreamError ( controller . controlledTransformStream , e ) ; } export function TransformStreamDefaultControllerPerformTransform ( controller , chunk ) { controller . transformAlgorithm ( chunk ) . catch ( r => { TransformStreamError ( controller . controlledTransformStream , r ) ; throw r ; } ) ; } export function TransformStreamDefaultControllerTerminate ( controller ) { const stream = controller . controlledTransformStream ; const readableController = stream . readable . readableStreamController ; if ( ReadableStreamDefaultControllerCanCloseOrEnqueue ( readableController ) ) { ReadableStreamDefaultControllerClose ( readableController ) ; } const error = new TypeError ( "stream ended" ) ; TransformStreamErrorWritableAndUnblockWrite ( stream , error ) ; } export function TransformStreamDefaultSinkWriteAlgorithm ( stream , chunk ) { Assert ( stream . writable . state === "writable" ) ; const controller = stream . transformStreamController ; if ( stream . backpressure ) { const p = stream . backpressureChangePromise ; Assert ( p !== void 0 ) ; return p . then ( ( ) => { const writable = stream . writable ; const { state } = writable ; if ( state === "erroring" ) { throw writable . storedError ; } Assert ( state === "writable" ) ; return TransformStreamDefaultControllerPerformTransform ( controller , chunk ) ; } ) ; } return TransformStreamDefaultControllerPerformTransform ( controller , chunk ) ; } export async function TransformStreamDefaultSinkAbortAlgorithm ( stream , reason ) { TransformStreamError ( stream , reason ) ; } export function TransformStreamDefaultSinkCloseAlgorithm ( stream ) { const { readable } = stream ; const controller = stream . transformStreamController ; const flushPromise = controller . flushAlgorithm ( ) ; TransformStreamDefaultControllerClearAlgorithms ( controller ) ; return flushPromise . then ( ( ) => { if ( readable . state === "errored" ) { throw readable . storedError ; } const readableController = readable . readableStreamController ; if ( ReadableStreamDefaultControllerCanCloseOrEnqueue ( readableController ) ) { ReadableStreamDefaultControllerClose ( readableController ) ; } } ) . catch ( r => { TransformStreamError ( stream , r ) ; throw readable . storedError ; } ) ; } export function TransformStreamDefaultSourcePullAlgorithm ( stream ) { Assert ( stream . backpressure ) ; Assert ( stream . backpressureChangePromise !== void 0 ) ; TransformStreamSetBackpressure ( stream , false ) ; return stream . backpressureChangePromise ; }