import { EOF , Reader , Writer } from "./deno" ; import { TextDecoder , TextEncoder } from "./encoding" ; interface Deferred < T > extends Promise < T > { status ( ) : "resolved" | "rejected" | undefined ; resolve ( t ) ; reject ( e ) ; } export function deferred < T = void > ( ) { let resolve , reject ; let st ; const p = new Promise < T > ( ( a , b ) => { resolve = ( ... args ) => { try { a ( ... args ) ; } finally { st = "resolved" ; } } ; reject = ( ... args ) => { try { b ( ... args ) ; } finally { st = "rejected" ; } } ; } ) ; return Object . assign ( p , { resolve , reject , status ( ) { return st ; } } ) ; } export function concatBytes ( ... chunks ) { const total = chunks . reduce ( ( sum , i ) => sum + i . byteLength , 0 ) ; const ret = new Uint8Array ( total ) ; let done = 0 ; for ( const chunk of chunks ) { ret . set ( chunk , done ) ; done += chunk . byteLength ; } return ret ; } export function streamToReader ( stream ) { let ended = false ; let err ; let buf = new Uint8Array ( ) ; let currDeferred = deferred ( ) ; stream . on ( "data" , chunk => { buf = concatBytes ( buf , chunk ) ; currDeferred . resolve ( ) ; currDeferred = deferred ( ) ; } ) . on ( "end" , ( ) => { ended = true ; } ) . on ( "error" , e => { err = e ; } ) ; async function read ( p ) { if ( ended ) { return EOF ; } else if ( err ) { throw err ; } const rem = Math . min ( p . byteLength , buf . byteLength ) ; if ( rem > 0 ) { p . set ( buf . subarray ( 0 , rem ) ) ; buf = buf . subarray ( rem ) ; return rem ; } else { await currDeferred ; return read ( p ) ; } } return { read } ; } export function streamToWriter ( stream ) { async function write ( p ) { return new Promise < number > ( ( resolve , reject ) => { stream . write ( p , err => { err ? reject ( err ) : resolve ( p . byteLength ) ; } ) ; } ) ; } return { write } ; } const encoder = new TextEncoder ( ) ; export function encode ( s ) { return encoder . encode ( s ) ; } const decoder = new TextDecoder ( ) ; export function decode ( p ) { return decoder . decode ( p ) ; }