import { constants , ClientHttp2Session } from 'http2' ; const { HTTP2_HEADER_PATH , HTTP2_HEADER_STATUS , HTTP2_HEADER_CONTENT_TYPE , HTTP2_HEADER_METHOD , } = constants ; type DdlArgs = { ksql : string ; } ; export const DELIMITED_CONTENT_TYPE = 'application/vnd.ksqlapi.delimited.v1' ; class Ddl { constructor ( protected session , protected ddlArgs ) { this . session = session ; this . ddlArgs = ddlArgs ; } headers ( ) { return { [ HTTP2_HEADER_PATH ] : '/ksql' , [ HTTP2_HEADER_METHOD ] : 'POST' , [ HTTP2_HEADER_CONTENT_TYPE ] : DELIMITED_CONTENT_TYPE , } ; } init = ( ) : Promise < any > => { return new Promise ( resolve => { try { const stream = this . session . request ( this . headers ( ) ) ; const reqPayload = Buffer . from ( JSON . stringify ( this . ddlArgs ) ) ; stream . on ( 'error' , ( e ) => { resolve ( e ) ; } ) ; stream . on ( 'connect' , ( ) => { resolve ( stream ) ; } ) ; stream . end ( reqPayload ) ; return stream ; } catch ( e ) { resolve ( e ) ; } } ) ; } ; } type DdlRepsonse = { message : any ; statusCode : string ; error ? : boolean ; } ; export const ddl = ( session , ddlArgs ) : Promise < void | DdlRepsonse > => { return new Promise < void | DdlRepsonse > ( async ( resolve ) => { const ddl = new Ddl ( session , ddlArgs ) ; const stream = await ddl . init ( ) ; const destroyStream = ( message , statusCode ) : DdlRepsonse => { stream . close ( ) ; return { message , statusCode , } ; } ; let data = '' ; if ( stream instanceof Error ) { const errors = stream as NodeJS . ErrnoException ; resolve ( { message : errors . stack , statusCode : errors . code , error : true } ) ; return ; } stream . on ( 'response' , headers => { const status = headers [ HTTP2_HEADER_STATUS ] ; if ( Number ( status ) > 302 || isNaN ( status ) ) { resolve ( destroyStream ( `ddl failed. Status ${ headers [ HTTP2_HEADER_STATUS ] }` , status ) ) ; } stream . on ( 'data' , chunk => ( data += chunk ) ) . on ( 'error' , error => resolve ( destroyStream ( error , status ) ) ) . on ( 'abort' , ( ) => resolve ( destroyStream ( new Error ( 'abort' ) , status ) ) ) . on ( 'timeout' , ( ) => resolve ( destroyStream ( new Error ( 'timeout' ) , status ) ) ) . on ( 'end' , ( ) => { resolve ( destroyStream ( data , status ) ) ; } ) ; } ) ; } ) . catch ( e => { console . error ( e ) ; Promise . resolve ( e ) ; } ) ; } ;