import { getLogger } from "@scramjet/logger" ; import { Duplex , Writable } from "stream" ; import { IncomingHttpHeaders , IncomingMessage , ServerResponse } from "http" ; const BPMux = require ( "bpmux" ) . BPMux ; /**
 * VerserConnection class.
 *
 * Provides methods for handling connection to Verser server and streams in connection socket.
 */ export class VerserConnection { private logger = getLogger ( "VerserConnection" ) ; private request : IncomingMessage ; private bpmux : any ; private _socket : Duplex ; get socket ( ) : Duplex { return this . _socket ; } constructor ( request , socket ) { this . request = request ; this . _socket = socket ; this . request . on ( "error" , ( error ) => { this . logger . error ( "Request error:" , error ) ; } ) ; } param { string } name returns { string | string [ ] | undefined } getHeader ( name ) { return this . request . headers [ name ] ; } returns getHeaders ( ) { return this . request . headers ; } param { number } httpStatusCode respond ( httpStatusCode ) { this . _socket . write ( `HTTP/1.1 ${ httpStatusCode } \r\n\r\n` ) ; } param { number } httpStatusCode end ( httpStatusCode ) { this . _socket . end ( `HTTP/1.1 ${ httpStatusCode } \r\n\r\n` ) ; } param req { Incomingmessage } param res { ServerResponse } async forward ( req , res ) { const channel = this . bpmux . multiplex ( ) as Duplex ; channel . on ( "error" , ( error ) => { this . logger . error ( "Channel error, id:" , error . message ) ; } ) . on ( "end" , ( ) => { this . logger . log ( "Request ended." , req . method , req . url ) ; } ) ; this . logger . debug ( "Forwarding request" , req . method , req . url ) ; channel . write ( `${ req . method } ${ req . url } HTTP/1.1\r\n` + Object . keys ( req . headers ) . filter ( ( header ) => header !== "host" ) . map ( ( header ) => `${ header }: ${ req . headers [ header ] }` ) . join ( "\r\n" ) + "\r\n\r\n" ) ; const whenWrote = ( chunk , encoding , stream ) => new Promise < void > ( ( resolve ) => { if ( stream . write ( chunk , encoding ) === false ) { stream . once ( "drain" , resolve ) ; } else { resolve ( ) ; } } ) ; channel . pipe ( res . socket as Writable ) ; if ( req . headers [ "transfer-encoding" ] !== "chunked" ) { req . pipe ( channel ) ; return ; } try { for await ( const chunk of req ) { const eol = Buffer . from ( "\r\n" ) ; const chunked = Buffer . concat ( [ Buffer . from ( chunk . length . toString ( 16 ) ) , eol , chunk , eol ] ) ; await whenWrote ( chunked , "binary" , channel ) ; } await whenWrote ( "0\r\n\r\n" , "utf-8" , channel ) ; } catch ( err ) { this . logger . error ( "Error while forwarding request " , err ) ; } } param id { number } returns createChannel ( id ) { return this . bpmux . multiplex ( { channel : id } ) ; } reconnect ( ) { this . bpmux = new BPMux ( this . socket ) . on ( "error" , ( error ) => { this . logger . error ( error . message ) ; } ) ; } returns async close ( ) { this . logger . log ( "Closing VerserConnection..." ) ; return new Promise < void > ( res => { this . socket . end ( ( ) => { this . logger . log ( "VerserConnection closed." ) ; res ( ) ; } ) ; } ) ; } }