import { transports } from "./transports/index.js" ; import { installTimerFunctions } from "./util.js" ; import parseqs from "parseqs" ; import parseuri from "parseuri" ; import debugModule from "debug" ; import { Emitter } from "@socket.io/component-emitter" ; import { protocol } from "engine.io-parser" ; const debug = debugModule ( "engine.io-client:socket" ) ; export interface SocketOptions { /**
     * The host that we're connecting to. Set from the URI passed when connecting
     */ host : string ; /**
     * The hostname for our connection. Set from the URI passed when connecting
     */ hostname : string ; /**
     * If this is a secure connection. Set from the URI passed when connecting
     */ secure : boolean ; /**
     * The port for our connection. Set from the URI passed when connecting
     */ port : string | number ; /**
     * Any query parameters in our uri. Set from the URI passed when connecting
     */ query : { [ key ] : any ; } ; /**
     * `http.Agent` to use, defaults to `false` (NodeJS only)
     */ agent : string | boolean ; default upgrade : boolean ; /**
     * Forces base 64 encoding for polling transport even when XHR2
     * responseType is available and WebSocket even if the used standard
     * supports binary.
     */ forceBase64 : boolean ; default timestampParam : string ; default timestampRequests : boolean ; default transports : string [ ] ; default policyPost : number ; default rememberUpgrade : boolean ; /**
     * Are we only interested in transports that support binary?
     */ onlyBinaryUpgrades : boolean ; /**
     * Timeout for xhr-polling requests in milliseconds (0) (only for polling transport)
     */ requestTimeout : number ; /**
     * Transport options for Node.js client (headers etc)
     */ transportOptions : Object ; /**
     * (SSL) Certificate, Private key and CA certificates to use for SSL.
     * Can be used in Node.js client environment to manually specify
     * certificate information.
     */ pfx : string ; /**
     * (SSL) Private key to use for SSL. Can be used in Node.js client
     * environment to manually specify certificate information.
     */ key : string ; /**
     * (SSL) A string or passphrase for the private key or pfx. Can be
     * used in Node.js client environment to manually specify certificate
     * information.
     */ passphrase : string ; /**
     * (SSL) Public x509 certificate to use. Can be used in Node.js client
     * environment to manually specify certificate information.
     */ cert : string ; /**
     * (SSL) An authority certificate or array of authority certificates to
     * check the remote host against.. Can be used in Node.js client
     * environment to manually specify certificate information.
     */ ca : string | string [ ] ; /**
     * (SSL) A string describing the ciphers to use or exclude. Consult the
     * [cipher format list]
     * (http://www.openssl.org/docs/apps/ciphers.html#CIPHER_LIST_FORMAT) for
     * details on the format.. Can be used in Node.js client environment to
     * manually specify certificate information.
     */ ciphers : string ; /**
     * (SSL) If true, the server certificate is verified against the list of
     * supplied CAs. An 'error' event is emitted if verification fails.
     * Verification happens at the connection level, before the HTTP request
     * is sent. Can be used in Node.js client environment to manually specify
     * certificate information.
     */ rejectUnauthorized : boolean ; /**
     * Headers that will be passed for each request to the server (via xhr-polling and via websockets).
     * These values then can be used during handshake or for special proxies.
     */ extraHeaders ? : { [ header ] : string ; } ; default withCredentials : boolean ; default closeOnBeforeunload : boolean ; default useNativeTimers : boolean ; default autoUnref : boolean ; default perMessageDeflate : { threshold : number ; } ; default path : string ; default protocols : string | string [ ] ; } interface SocketReservedEvents { open : ( ) => void ; handshake : ( data ) => void ; packet : ( packet ) => void ; packetCreate : ( packet ) => void ; data : ( data ) => void ; message : ( data ) => void ; drain : ( ) => void ; flush : ( ) => void ; heartbeat : ( ) => void ; ping : ( ) => void ; pong : ( ) => void ; error : ( err : string | Error ) => void ; upgrading : ( transport ) => void ; upgrade : ( transport ) => void ; upgradeError : ( err : Error ) => void ; close : ( reason : string , desc ? : Error ) => void ; } export class Socket extends Emitter < { } , { } , SocketReservedEvents > { public id : string ; public transport : any ; public binaryType : string ; private readyState : string ; private writeBuffer ; private prevBufferLen : number ; private upgrades ; private pingInterval : number ; private pingTimeout : number ; private pingTimeoutTimer : NodeJS . Timer ; private setTimeoutFn : typeof setTimeout ; private clearTimeoutFn : typeof clearTimeout ; private offlineEventListener ; private upgrading : boolean ; private readonly opts : Partial < SocketOptions > ; private readonly secure : boolean ; private readonly hostname : string ; private readonly port : string | number ; private readonly transports : string [ ] ; static priorWebsocketSuccess : boolean ; static protocol = protocol ; param { String | Object } uri param { Object } opts api constructor ( uri , opts = { } ) { super ( ) ; if ( uri && "object" === typeof uri ) { opts = uri ; uri = null ; } if ( uri ) { uri = parseuri ( uri ) ; opts . hostname = uri . host ; opts . secure = uri . protocol === "https" || uri . protocol === "wss" ; opts . port = uri . port ; if ( uri . query ) opts . query = uri . query ; } else if ( opts . host ) { opts . hostname = parseuri ( opts . host ) . host ; } installTimerFunctions ( this , opts ) ; this . secure = null != opts . secure ? opts . secure : typeof location !== "undefined" && "https:" === location . protocol ; if ( opts . hostname && ! opts . port ) { opts . port = this . secure ? "443" : "80" ; } this . hostname = opts . hostname || ( typeof location !== "undefined" ? location . hostname : "localhost" ) ; this . port = opts . port || ( typeof location !== "undefined" && location . port ? location . port : this . secure ? "443" : "80" ) ; this . transports = opts . transports || [ "polling" , "websocket" ] ; this . readyState = "" ; this . writeBuffer = [ ] ; this . prevBufferLen = 0 ; this . opts = Object . assign ( { path : "/engine.io" , agent : false , withCredentials : false , upgrade : true , timestampParam : "t" , rememberUpgrade : false , rejectUnauthorized : true , perMessageDeflate : { threshold : 1024 } , transportOptions : { } , closeOnBeforeunload : true } , opts ) ; this . opts . path = this . opts . path . replace ( /\/$/ , "" ) + "/" ; if ( typeof this . opts . query === "string" ) { this . opts . query = parseqs . decode ( this . opts . query ) ; } this . id = null ; this . upgrades = null ; this . pingInterval = null ; this . pingTimeout = null ; this . pingTimeoutTimer = null ; if ( typeof addEventListener === "function" ) { if ( this . opts . closeOnBeforeunload ) { addEventListener ( "beforeunload" , ( ) => { if ( this . transport ) { this . transport . removeAllListeners ( ) ; this . transport . close ( ) ; } } , false ) ; } if ( this . hostname !== "localhost" ) { this . offlineEventListener = ( ) => { this . onClose ( "transport close" ) ; } ; addEventListener ( "offline" , this . offlineEventListener , false ) ; } } this . open ( ) ; } param { String } transport return { Transport } api private createTransport ( name ) { debug ( 'creating transport "%s"' , name ) ; const query = clone ( this . opts . query ) ; query . EIO = protocol ; query . transport = name ; if ( this . id ) query . sid = this . id ; const opts = Object . assign ( { } , this . opts . transportOptions [ name ] , this . opts , { query , socket : this , hostname : this . hostname , secure : this . secure , port : this . port } ) ; debug ( "options: %j" , opts ) ; return new transports [ name ] ( opts ) ; } api private open ( ) { let transport ; if ( this . opts . rememberUpgrade && Socket . priorWebsocketSuccess && this . transports . indexOf ( "websocket" ) !== - 1 ) { transport = "websocket" ; } else if ( 0 === this . transports . length ) { this . setTimeoutFn ( ( ) => { this . emitReserved ( "error" , "No transports available" ) ; } , 0 ) ; return ; } else { transport = this . transports [ 0 ] ; } this . readyState = "opening" ; try { transport = this . createTransport ( transport ) ; } catch ( e ) { debug ( "error while creating transport: %s" , e ) ; this . transports . shift ( ) ; this . open ( ) ; return ; } transport . open ( ) ; this . setTransport ( transport ) ; } api private setTransport ( transport ) { debug ( "setting transport %s" , transport . name ) ; if ( this . transport ) { debug ( "clearing existing transport %s" , this . transport . name ) ; this . transport . removeAllListeners ( ) ; } this . transport = transport ; transport . on ( "drain" , this . onDrain . bind ( this ) ) . on ( "packet" , this . onPacket . bind ( this ) ) . on ( "error" , this . onError . bind ( this ) ) . on ( "close" , ( ) => { this . onClose ( "transport close" ) ; } ) ; } param { String } transport api private probe ( name ) { debug ( 'probing transport "%s"' , name ) ; let transport = this . createTransport ( name ) ; let failed = false ; Socket . priorWebsocketSuccess = false ; const onTransportOpen = ( ) => { if ( failed ) return ; debug ( 'probe transport "%s" opened' , name ) ; transport . send ( [ { type : "ping" , data : "probe" } ] ) ; transport . once ( "packet" , msg => { if ( failed ) return ; if ( "pong" === msg . type && "probe" === msg . data ) { debug ( 'probe transport "%s" pong' , name ) ; this . upgrading = true ; this . emitReserved ( "upgrading" , transport ) ; if ( ! transport ) return ; Socket . priorWebsocketSuccess = "websocket" === transport . name ; debug ( 'pausing current transport "%s"' , this . transport . name ) ; this . transport . pause ( ( ) => { if ( failed ) return ; if ( "closed" === this . readyState ) return ; debug ( "changing transport and sending upgrade packet" ) ; cleanup ( ) ; this . setTransport ( transport ) ; transport . send ( [ { type : "upgrade" } ] ) ; this . emitReserved ( "upgrade" , transport ) ; transport = null ; this . upgrading = false ; this . flush ( ) ; } ) ; } else { debug ( 'probe transport "%s" failed' , name ) ; const err = new Error ( "probe error" ) ; err . transport = transport . name ; this . emitReserved ( "upgradeError" , err ) ; } } ) ; } ; function freezeTransport ( ) { if ( failed ) return ; failed = true ; cleanup ( ) ; transport . close ( ) ; transport = null ; } const onerror = err => { const error = new Error ( "probe error: " + err ) ; error . transport = transport . name ; freezeTransport ( ) ; debug ( 'probe transport "%s" failed because of error: %s' , name , err ) ; this . emitReserved ( "upgradeError" , error ) ; } ; function onTransportClose ( ) { onerror ( "transport closed" ) ; } function onclose ( ) { onerror ( "socket closed" ) ; } function onupgrade ( to ) { if ( transport && to . name !== transport . name ) { debug ( '"%s" works - aborting "%s"' , to . name , transport . name ) ; freezeTransport ( ) ; } } const cleanup = ( ) => { transport . removeListener ( "open" , onTransportOpen ) ; transport . removeListener ( "error" , onerror ) ; transport . removeListener ( "close" , onTransportClose ) ; this . off ( "close" , onclose ) ; this . off ( "upgrading" , onupgrade ) ; } ; transport . once ( "open" , onTransportOpen ) ; transport . once ( "error" , onerror ) ; transport . once ( "close" , onTransportClose ) ; this . once ( "close" , onclose ) ; this . once ( "upgrading" , onupgrade ) ; transport . open ( ) ; } api private onOpen ( ) { debug ( "socket open" ) ; this . readyState = "open" ; Socket . priorWebsocketSuccess = "websocket" === this . transport . name ; this . emitReserved ( "open" ) ; this . flush ( ) ; if ( "open" === this . readyState && this . opts . upgrade && this . transport . pause ) { debug ( "starting upgrade probes" ) ; let i = 0 ; const l = this . upgrades . length ; for ( ; i < l ; i ++ ) { this . probe ( this . upgrades [ i ] ) ; } } } api private onPacket ( packet ) { if ( "opening" === this . readyState || "open" === this . readyState || "closing" === this . readyState ) { debug ( 'socket receive: type "%s", data "%s"' , packet . type , packet . data ) ; this . emitReserved ( "packet" , packet ) ; this . emitReserved ( "heartbeat" ) ; switch ( packet . type ) { case "open" : this . onHandshake ( JSON . parse ( packet . data ) ) ; break ; case "ping" : this . resetPingTimeout ( ) ; this . sendPacket ( "pong" ) ; this . emitReserved ( "ping" ) ; this . emitReserved ( "pong" ) ; break ; case "error" : const err = new Error ( "server error" ) ; err . code = packet . data ; this . onError ( err ) ; break ; case "message" : this . emitReserved ( "data" , packet . data ) ; this . emitReserved ( "message" , packet . data ) ; break ; } } else { debug ( 'packet received with socket readyState "%s"' , this . readyState ) ; } } param { Object } data api private onHandshake ( data ) { this . emitReserved ( "handshake" , data ) ; this . id = data . sid ; this . transport . query . sid = data . sid ; this . upgrades = this . filterUpgrades ( data . upgrades ) ; this . pingInterval = data . pingInterval ; this . pingTimeout = data . pingTimeout ; this . onOpen ( ) ; if ( "closed" === this . readyState ) return ; this . resetPingTimeout ( ) ; } api private resetPingTimeout ( ) { this . clearTimeoutFn ( this . pingTimeoutTimer ) ; this . pingTimeoutTimer = this . setTimeoutFn ( ( ) => { this . onClose ( "ping timeout" ) ; } , this . pingInterval + this . pingTimeout ) ; if ( this . opts . autoUnref ) { this . pingTimeoutTimer . unref ( ) ; } } api private onDrain ( ) { this . writeBuffer . splice ( 0 , this . prevBufferLen ) ; this . prevBufferLen = 0 ; if ( 0 === this . writeBuffer . length ) { this . emitReserved ( "drain" ) ; } else { this . flush ( ) ; } } api private flush ( ) { if ( "closed" !== this . readyState && this . transport . writable && ! this . upgrading && this . writeBuffer . length ) { debug ( "flushing %d packets in socket" , this . writeBuffer . length ) ; this . transport . send ( this . writeBuffer ) ; this . prevBufferLen = this . writeBuffer . length ; this . emitReserved ( "flush" ) ; } } param { String } message . param { Function } callback param { Object } options . return { Socket } api public write ( msg , options , fn ? ) { this . sendPacket ( "message" , msg , options , fn ) ; return this ; } public send ( msg , options , fn ? ) { this . sendPacket ( "message" , msg , options , fn ) ; return this ; } param { String } packet param { String } data . param { Object } options . param { Function } callback api private sendPacket ( type , data ? , options ? , fn ? ) { if ( "function" === typeof data ) { fn = data ; data = undefined ; } if ( "function" === typeof options ) { fn = options ; options = null ; } if ( "closing" === this . readyState || "closed" === this . readyState ) { return ; } options = options || { } ; options . compress = false !== options . compress ; const packet = { type : type , data : data , options : options } ; this . emitReserved ( "packetCreate" , packet ) ; this . writeBuffer . push ( packet ) ; if ( fn ) this . once ( "flush" , fn ) ; this . flush ( ) ; } api public close ( ) { const close = ( ) => { this . onClose ( "forced close" ) ; debug ( "socket closing - telling transport to close" ) ; this . transport . close ( ) ; } ; const cleanupAndClose = ( ) => { this . off ( "upgrade" , cleanupAndClose ) ; this . off ( "upgradeError" , cleanupAndClose ) ; close ( ) ; } ; const waitForUpgrade = ( ) => { this . once ( "upgrade" , cleanupAndClose ) ; this . once ( "upgradeError" , cleanupAndClose ) ; } ; if ( "opening" === this . readyState || "open" === this . readyState ) { this . readyState = "closing" ; if ( this . writeBuffer . length ) { this . once ( "drain" , ( ) => { if ( this . upgrading ) { waitForUpgrade ( ) ; } else { close ( ) ; } } ) ; } else if ( this . upgrading ) { waitForUpgrade ( ) ; } else { close ( ) ; } } return this ; } api private onError ( err ) { debug ( "socket error %j" , err ) ; Socket . priorWebsocketSuccess = false ; this . emitReserved ( "error" , err ) ; this . onClose ( "transport error" , err ) ; } api private onClose ( reason , desc ? ) { if ( "opening" === this . readyState || "open" === this . readyState || "closing" === this . readyState ) { debug ( 'socket close with reason: "%s"' , reason ) ; this . clearTimeoutFn ( this . pingTimeoutTimer ) ; this . transport . removeAllListeners ( "close" ) ; this . transport . close ( ) ; this . transport . removeAllListeners ( ) ; if ( typeof removeEventListener === "function" ) { removeEventListener ( "offline" , this . offlineEventListener , false ) ; } this . readyState = "closed" ; this . id = null ; this . emitReserved ( "close" , reason , desc ) ; this . writeBuffer = [ ] ; this . prevBufferLen = 0 ; } } param { Array } server api private filterUpgrades ( upgrades ) { const filteredUpgrades = [ ] ; let i = 0 ; const j = upgrades . length ; for ( ; i < j ; i ++ ) { if ( ~ this . transports . indexOf ( upgrades [ i ] ) ) filteredUpgrades . push ( upgrades [ i ] ) ; } return filteredUpgrades ; } } function clone ( obj ) { const o = { } ; for ( let i in obj ) { if ( obj . hasOwnProperty ( i ) ) { o [ i ] = obj [ i ] ; } } return o ; }