import { OutgoingHttpHeaders , request , Agent } from "http" ; import { merge , TypedEmitter } from "@scramjet/utility" ; import { VerserClientOptions , VerserClientConnection , RegisteredChannels , RegisteredChannelCallback } from "../types" ; import { Duplex } from "stream" ; import { Socket } from "net" ; import { getLogger } from "@scramjet/logger" ; import { defaultVerserClientOptions } from "./verser-client-default-config" ; import { Logger } from "@scramjet/types" ; import { URL } from "url" ; const BPMux = require ( "bpmux" ) . BPMux ; type Events = { error : ( err : Error ) => void ; } ; /**
 * VerserClient class.
 *
 * Provides methods for connecting to Verser server and handling of incoming muxed duplex streams.
 */ export class VerserClient extends TypedEmitter < Events > { type { VerserClientOptions } private opts : VerserClientOptions ; type { http . Agent } see https private agent : Agent ; type { Socket } private socket ? : Socket ; /**
     * Map of registered channels.
     */ private registeredChannels : RegisteredChannels = new Map < number , RegisteredChannelCallback > ( ) ; type { Logger } public logger : Logger = getLogger ( this ) ; constructor ( opts = defaultVerserClientOptions ) { super ( ) ; this . opts = opts ; this . agent = new Agent ( { keepAlive : true } ) ; } returns { Promise < VerserClientConnection > } public async connect ( ) { return new Promise ( ( resolve , reject ) => { const { hostname , port , pathname } = this . opts . verserUrl instanceof URL ? this . opts . verserUrl : new URL ( this . opts . verserUrl ) ; const connectRequest = request ( { agent : this . agent , headers : this . opts . headers , hostname , method : "CONNECT" , pathname , port } ) ; connectRequest . on ( "error" , ( err ) => { this . logger . log ( "Connect error" , err ) ; reject ( err ) ; } ) ; connectRequest . on ( "connect" , ( _req , socket ) => { this . socket = socket ; this . mux ( ) ; resolve ( { req : _req , socket } ) ; } ) ; connectRequest . end ( ) ; } ) ; } param { VerserClientConnection } connection private mux ( ) { new BPMux ( this . socket ) . on ( "handshake" , async ( mSocket ) => { const registeredChannelCallback = this . registeredChannels . get ( mSocket . _chan ) ; if ( registeredChannelCallback ) { await registeredChannelCallback ( mSocket ) ; } else { this . opts . server ?. emit ( "connection" , mSocket ) ; } } ) . on ( "error" , ( err ) => { this . emit ( "error" , err ) ; } ) ; } param headers { IncomingHttpHeaders } updateHeaders ( headers ) { merge ( this . opts . headers , headers ) ; } param channelId { number } param data { RegisteredChannelCallback } registerChannel ( channelId , data ) { this . registeredChannels . set ( channelId , data ) ; } }